cli.py

# cli.py
import os
import sys
from pathlib import Path

import click
from rich.console import Console
from rich.markdown import Markdown

from agent.core import TaskAgent, _ASSESSMENT_RE
from agent.escalator import Escalator
from agent.session import Session
from cli_layout import RichGroup, render_banner, render_landing
from plugins import list_domain_plugins, load_domain_plugin
from plugins.escalation.github.plugin import GitHubEscalationPlugin

console = Console()


def _build_escalator() -> Escalator:
    token = os.environ.get("GITHUB_TOKEN", "")
    repo = os.environ.get("GITHUB_REPO", "")
    if not token or not repo:
        console.print(
            "[bold red]Warning:[/bold red] [dim]GITHUB_TOKEN and GITHUB_REPO not set. "
            "Escalation will fail if triggered.[/dim]",
            stderr=True,
        )
    return Escalator(GitHubEscalationPlugin(token=token, repo=repo))


def _strip_assessment(text: str) -> str:
    """Remove the trailing self-assessment JSON block from a response string."""
    return _ASSESSMENT_RE.sub("", text).rstrip()


def _print_response(attempt) -> None:
    display_text = _strip_assessment(attempt.response)
    console.print("\n[bold white]Agent:[/bold white]")
    console.print(Markdown(display_text))
    sa = attempt.self_assessment
    converged = sa.get("converged", False)
    confidence = sa.get("confidence", 0.0)
    console.print(f"[dim][converged: {converged} | confidence: {confidence:.1f}][/dim]")


@click.group(
    cls=RichGroup,
    context_settings={"help_option_names": ["-h", "--help"]},
    invoke_without_command=True,
)
@click.version_option(version="0.1.0", prog_name="self-escalate")
@click.pass_context
def cli(ctx: click.Context) -> None:
    """Self-Escalate Agent — a minimal escalating CLI agent."""
    if ctx.invoked_subcommand is None:
        render_landing()
        raise SystemExit(0)


def _run_session(
    task: str,
    domain: str | None,
    task_agent: TaskAgent,
    escalator: Escalator,
    persist: bool,
    conversation_history: list[dict],
) -> None:
    """Run one agent turn, update conversation history, and print the response."""
    conversation_history.append({"role": "user", "content": task})

    session = Session(task=task, domain=domain)
    session.context = list(conversation_history)

    with console.status("[dim]Thinking…[/dim]"):
        attempt = task_agent.run(session)
    session.attempts.append(attempt)
    _print_response(attempt)

    display_text = _strip_assessment(attempt.response)
    conversation_history.append({"role": "assistant", "content": display_text})

    if attempt.needs_escalation:
        try:
            issue_url = escalator.trigger(session)
            console.print(
                f"[bold yellow]Escalated →[/bold yellow] [dim]{issue_url}[/dim]"
            )
        except Exception as exc:  # noqa: BLE001
            console.print(f"[bold red]Escalation failed:[/bold red] [dim]{exc}[/dim]")
        return

    if attempt.used_tools:
        session.status = "resolved"
        console.print("[dim cyan]resolved ✓[/dim cyan]")

    if persist and session.status == "resolved":
        path = session.save()
        console.print(f"[dim cyan]Session saved → {path}[/dim cyan]")


@cli.command()
@click.option("--domain", default=None, help="Domain plugin name (e.g. 'general')")
@click.option(
    "--persist", is_flag=True, help="Persist session JSON to ~/.self-escalate/sessions/"
)
def chat(domain: str | None, persist: bool):
    """Start an interactive chat session."""
    render_banner()
    domain_prompt = ""
    domain_tools: list[dict] = []

    if domain:
        try:
            plugin = load_domain_plugin(domain)
            domain_prompt = plugin.get_prompt()
            domain_tools = plugin.get_tools()
            console.print(f"[dim cyan]Domain plugin '{domain}' loaded[/dim cyan]")
        except ValueError as exc:
            console.print(f"[bold red]{exc}[/bold red]", stderr=True)
            sys.exit(1)

    task_agent = TaskAgent(domain_prompt=domain_prompt, domain_tools=domain_tools)
    escalator = _build_escalator()
    conversation_history: list[dict] = []

    first = True
    while True:
        if not first:
            console.print()
            console.rule(style="dim")
        first = False

        try:
            console.print("\n[bold white]You[/bold white]")
            task = click.prompt(
                click.style("›", fg="cyan", bold=True), prompt_suffix=" "
            )
        except (EOFError, KeyboardInterrupt):
            console.print("\n[dim]goodbye.[/dim]")
            break

        if task.strip().lower() in ("/exit", "/quit"):
            console.print("[dim]goodbye.[/dim]")
            break

        _run_session(task, domain, task_agent, escalator, persist, conversation_history)


@cli.group()
def plugin():
    """Manage domain plugins."""


@plugin.command("list")
def plugin_list():
    """List all installed domain plugins."""
    plugins = list_domain_plugins()
    if not plugins:
        console.print("[dim]No domain plugins installed.[/dim]")
    else:
        for name in plugins:
            console.print(f"[dim cyan]  • {name}[/dim cyan]")


@plugin.command("install")
@click.argument("source", type=click.Path(exists=True))
def plugin_install(source: str):
    """Install a domain plugin from a local directory."""
    import shutil

    src = Path(source).resolve()
    plugin_py = src / "plugin.py"
    if not plugin_py.exists():
        console.print(
            f"[bold red]Error: {src} must contain a plugin.py file.[/bold red]",
            stderr=True,
        )
        sys.exit(1)

    dest_dir = Path(__file__).parent / "plugins" / "domain" / src.name
    if dest_dir.exists():
        console.print(
            f"[bold red]Plugin '{src.name}' already installed. Remove {dest_dir} to reinstall.[/bold red]",
            stderr=True,
        )
        sys.exit(1)

    shutil.copytree(src, dest_dir)
    console.print(f"[dim cyan]Plugin '{src.name}' installed → {dest_dir}[/dim cyan]")


@cli.group()
def config():
    """Configure the agent (env-var based for POC)."""


@config.command("set")
@click.argument("key")
@click.argument("value")
def config_set(key: str, value: str):
    """Print the environment variable to set for a config key."""
    mapping = {
        "escalation.provider": "ESCALATION_PROVIDER",
        "escalation.github.token": "GITHUB_TOKEN",
        "escalation.github.repo": "GITHUB_REPO",
    }
    env_var = mapping.get(key)
    if not env_var:
        console.print(
            f"[bold red]Unknown config key: {key}. Known keys: {list(mapping)}[/bold red]",
            stderr=True,
        )
        sys.exit(1)
    console.print("[dim]Add this to your shell profile:[/dim]")
    console.print(f'[dim cyan]  export {env_var}="{value}"[/dim cyan]')


if __name__ == "__main__":
    cli()