Manages escalation to external AI coding CLIs.
| 88 | # ── Manager ─────────────────────────────────────────────────────────────────── |
| 89 | |
| 90 | class PeerCLIManager: |
| 91 | """Manages escalation to external AI coding CLIs.""" |
| 92 | |
| 93 | def __init__(self): |
| 94 | self._available: Optional[List[PeerCLI]] = None |
| 95 | |
| 96 | def available(self) -> List[PeerCLI]: |
| 97 | """Return cached list of installed peer CLIs.""" |
| 98 | if self._available is None: |
| 99 | self._available = [c for c in PEER_REGISTRY if self._is_installed(c)] |
| 100 | return self._available |
| 101 | |
| 102 | def _is_installed(self, cli: PeerCLI) -> bool: |
| 103 | # shutil.which is the most reliable check — works even if |
| 104 | # the CLI doesn't support --version or returns non-zero for it |
| 105 | base_cmd = cli.cmd.split()[0] |
| 106 | if not shutil.which(base_cmd): |
| 107 | return False |
| 108 | # For CLIs that bundle native node modules (e.g. node-pty), do a quick |
| 109 | # smoke-test to catch platforms where the binary exists but crashes on |
| 110 | # start (e.g. missing pty.node prebuilds on Android ARM64). |
| 111 | if cli.check_cmd: |
| 112 | try: |
| 113 | result = subprocess.run( |
| 114 | cli.check_cmd.split(), |
| 115 | capture_output=True, timeout=5 |
| 116 | ) |
| 117 | stderr = (result.stderr or b"").decode("utf-8", errors="replace") |
| 118 | # Detect node native-module crash signatures |
| 119 | native_crash = any(sig in stderr for sig in [ |
| 120 | "Failed to load native module", |
| 121 | "pty.node", |
| 122 | "prebuilds/", |
| 123 | "NODE_MODULE_VERSION", |
| 124 | ]) |
| 125 | return not native_crash |
| 126 | except FileNotFoundError: |
| 127 | return False |
| 128 | except subprocess.TimeoutExpired: |
| 129 | # Timed out but binary exists (shutil.which passed) — assume installed |
| 130 | return True |
| 131 | except Exception: |
| 132 | return False |
| 133 | return True |
| 134 | |
| 135 | def detect_task_type(self, user_message: str, errors: List[str]) -> str: |
| 136 | """Infer task type from the user message and accumulated error log.""" |
| 137 | msg = user_message.lower() |
| 138 | err_text = " ".join(errors).lower() |
| 139 | if any(k in msg for k in ["fix", "bug", "error", "broken", "crash", "fail", "debug"]): |
| 140 | return "debugging" |
| 141 | if any(k in msg for k in ["refactor", "rewrite", "restructure", "clean up"]): |
| 142 | return "refactor" |
| 143 | if any(k in msg for k in ["explain", "what does", "how does", "why does"]): |
| 144 | return "explain" |
| 145 | if any(k in msg for k in ["review", "audit", "analyze", "analyse", "check"]): |
| 146 | return "review" |
| 147 | if any(k in msg for k in ["create", "write", "build", "generate", "make", "implement"]): |
no outgoing calls
no test coverage detected