Function run_prompted

core/peer_shell.py:225–283

Run a peer CLI in non-interactive mode by passing the prompt as a flag. e.g. claude -p "write a function that reverses a string" gemini --model gemini-2.0-flash -p "explain this" Streams output live to the terminal and captures it for Codey. No TUI, no trust dialogs, no

(cli_name: str, cmd: str, flag: str, prompt_text: str)

def run_prompted(cli_name: str, cmd: str, flag: str, prompt_text: str) -> str:
    """
    Run a peer CLI in non-interactive mode by passing the prompt as a flag.
    e.g.  claude -p "write a function that reverses a string"
          gemini --model gemini-2.0-flash -p "explain this"

    Streams output live to the terminal and captures it for Codey.
    No TUI, no trust dialogs, no pexpect needed.
    Returns "[PEER_ERROR: ...]" if the CLI fails so callers can handle it.
    """
    import shlex
    import subprocess
    width  = _terminal_width()
    border = "─" * width

    print(f"\n{_CYAN}{_header(cli_name.upper() + ' CLI  (direct)', width)}{_RESET}")
    info(f"Asking {cli_name}: {prompt_text[:100]}{'…' if len(prompt_text) > 100 else ''}")
    print(f"{_DIM}{border}{_RESET}\n")

    captured = []
    returncode = 0
    try:
        # shlex.split handles "gemini --model x" → ["gemini", "--model", "x"]
        argv = shlex.split(cmd) + [flag, prompt_text]
        proc = subprocess.Popen(
            argv,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
        )
        for line in proc.stdout:
            sys.stdout.write(line)
            sys.stdout.flush()
            captured.append(line)
        proc.wait()
        returncode = proc.returncode
    except FileNotFoundError:
        print(f"{_RED}[peer_shell] Command not found: {cmd}{_RESET}")
        returncode = 127
    except Exception as e:
        print(f"{_RED}[peer_shell] Error: {e}{_RESET}")
        returncode = 1

    print(f"\n{_DIM}{border}{_RESET}")
    print(f"{_CYAN}{_header(cli_name.upper() + ' DONE', width)}{_RESET}\n")
    info(f"Back in Codey — reading {cli_name} output…")

    output = "".join(captured)
    # Strip startup noise before error detection so noise lines don't
    # interfere, then return the clean output to the agent.
    if cli_name == "gemini":
        output = _strip_gemini_noise(output)
    reason = _detect_peer_error(output, returncode)
    if reason:
        msg = f"[PEER_ERROR: {cli_name} failed — {reason}]"
        warning(f"Peer {cli_name} failed: {reason}")
        return msg