Function run_direct

core/peer_shell.py:329–372

Run a peer CLI directly in the real terminal (no nested PTY). Used for CLIs that bundle their own native PTY module and crash when spawned inside pexpect's PTY. Output is captured via the `script` utility so Codey can read it. The prepared prompt is shown above the CLI so you c

(cli_name: str, cmd: str, prompt_text: str = "")

def run_direct(cli_name: str, cmd: str, prompt_text: str = "") -> str:
    """
    Run a peer CLI directly in the real terminal (no nested PTY).
    Used for CLIs that bundle their own native PTY module
    and crash when spawned inside pexpect's PTY.

    Output is captured via the `script` utility so Codey can read it.
    The prepared prompt is shown above the CLI so you can paste it in.
    """
    import tempfile
    width   = _terminal_width()
    border  = "─" * width
    outfile = tempfile.mktemp(prefix="codey_peer_", suffix=".txt")

    print(f"\n{_CYAN}{_header(cli_name.upper() + ' CLI', width)}{_RESET}")
    info(f"Opening {cli_name} in your terminal.")

    if prompt_text:
        print(f"\n{_DIM}── Paste this into {cli_name} ──────────────────────────{_RESET}")
        print(f"{prompt_text}")
        print(f"{_DIM}{border}{_RESET}\n")

    info(f"Close {cli_name} normally when done (/exit, Ctrl+D, /quit…)")
    print(f"{_DIM}Ctrl+B → return to Codey at any point{_RESET}\n")

    # `script -q -c <cmd> <file>` records the session to outfile
    # while letting the CLI run in the real terminal with full PTY support.
    # -q = quiet (no "Script started/done" lines)
    os.system(f'script -q -c "{cmd}" "{outfile}"')

    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…")

    try:
        raw = Path(outfile).read_bytes()
        Path(outfile).unlink(missing_ok=True)
        # Strip ANSI escape codes from the recorded script output
        text = raw.decode("utf-8", errors="replace")
        text = re.sub(r"\x1b\[[0-9;]*[a-zA-Z]", "", text)
        text = re.sub(r"\x1b\][^\x07]*\x07", "", text)
        return text
    except Exception as e:
        return f"[Could not read output: {e}]"


def _wait_for_gemini(child, pexpect_mod) -> None: