Top-level escalation entry point. Called by agent when retries are exhausted. Returns: - Summary string to inject into agent context (peer ran successfully) - "[redirect]: " (user wants Codey to try differently) - None
(
user_message: str,
errors: List[str],
files: List[str],
)
| 295 | |
| 296 | |
| 297 | def escalate( |
| 298 | user_message: str, |
| 299 | errors: List[str], |
| 300 | files: List[str], |
| 301 | ) -> Optional[str]: |
| 302 | """ |
| 303 | Top-level escalation entry point. Called by agent when retries are exhausted. |
| 304 | |
| 305 | Returns: |
| 306 | - Summary string to inject into agent context (peer ran successfully) |
| 307 | - "[redirect]: <instruction>" (user wants Codey to try differently) |
| 308 | - None (user skipped / no CLIs available) |
| 309 | """ |
| 310 | mgr = get_peer_cli_manager() |
| 311 | |
| 312 | if not mgr.available(): |
| 313 | warning( |
| 314 | "No peer CLIs found. Install claude / gemini / qwen " |
| 315 | "to enable escalation." |
| 316 | ) |
| 317 | return None |
| 318 | |
| 319 | task_type = mgr.detect_task_type(user_message, errors) |
| 320 | excluded: List[str] = [] |
| 321 | |
| 322 | while True: |
| 323 | cli = mgr.select_cli(task_type, exclude=excluded) |
| 324 | if not cli: |
| 325 | warning("No more peer CLIs available to try.") |
| 326 | return None |
| 327 | |
| 328 | result, payload = mgr.confirm(cli, task_type, user_message) |
| 329 | |
| 330 | if result is False: |
| 331 | info("Peer CLI escalation skipped.") |
| 332 | return None |
| 333 | |
| 334 | if result == "switch": |
| 335 | by_name = {c.name: c for c in mgr.available()} |
| 336 | cli = by_name.get(payload) |
| 337 | if not cli: |
| 338 | warning(f"CLI '{payload}' is not available.") |
| 339 | excluded.append(payload) |
| 340 | continue |
| 341 | result = True # fall through to call |
| 342 | |
| 343 | if result == "redirect": |
| 344 | return f"[redirect]: {payload}" |
| 345 | |
| 346 | if result is True: |
| 347 | prompt = mgr.build_prompt(user_message, errors, files) |
| 348 | output = mgr.call(cli, prompt) |
| 349 | summary = mgr.summarize_result(cli.name, output, user_message) |
| 350 | success(f"Peer CLI ({cli.name}) done — Codey is reading the result…") |
| 351 | return summary |
| 352 | |
| 353 | # Shouldn't reach here, but skip and try next |
| 354 | excluded.append(cli.name) |
no test coverage detected