Timmy-time-dashboard/src/timmy/cli.py

"""Typer CLI entry point for the ``timmy`` command (chat, think, status)."""
import asyncio
import logging
import subprocess
import sys

import typer

from timmy.agent import create_timmy
from timmy.prompts import STATUS_PROMPT
from timmy.tool_safety import format_action_description, get_impact_level, is_allowlisted

logger = logging.getLogger(__name__)

app = typer.Typer(help="Timmy — sovereign AI agent")

# Stable session ID — Agno persists conversation history in SQLite keyed on this.
# Every `timmy chat` invocation reuses the same session so context carries over.
_CLI_SESSION_ID = "cli"

# Shared option definitions (reused across commands for consistency).
_BACKEND_OPTION = typer.Option(
    None,
    "--backend",
    "-b",
    help="Inference backend: 'ollama' (default) | 'grok' | 'claude'",
)
_MODEL_SIZE_OPTION = typer.Option(
    None,
    "--model-size",
    "-s",
    help="Model size (reserved for future use).",
)


def _is_interactive() -> bool:
    """Return True if stdin is a real terminal (human present)."""
    return hasattr(sys.stdin, "isatty") and sys.stdin.isatty()


def _read_message_input(message: list[str]) -> str:
    """Join CLI args into a message, reading from stdin when requested.

    Returns the final message string.  Raises ``typer.Exit(1)`` when
    stdin is explicitly requested (``-``) but empty.
    """
    message_str = " ".join(message)

    if message_str == "-" or not _is_interactive():
        try:
            stdin_content = sys.stdin.read().strip()
        except (KeyboardInterrupt, EOFError):
            stdin_content = ""
        if stdin_content:
            message_str = stdin_content
        elif message_str == "-":
            typer.echo("No input provided via stdin.", err=True)
            raise typer.Exit(1)

    return message_str


def _resolve_session_id(session_id: str | None, new_session: bool) -> str:
    """Return the effective session ID for a chat invocation."""
    import uuid

    if session_id is not None:
        return session_id
    if new_session:
        return str(uuid.uuid4())
    return _CLI_SESSION_ID


def _prompt_interactive(req, tool_name: str, tool_args: dict) -> None:
    """Display tool details and prompt the human for approval."""
    description = format_action_description(tool_name, tool_args)
    impact = get_impact_level(tool_name)

    typer.echo()
    typer.echo(typer.style("Tool confirmation required", bold=True))
    typer.echo(f"  Impact: {impact.upper()}")
    typer.echo(f"  {description}")
    typer.echo()

    if typer.confirm("Allow this action?", default=False):
        req.confirm()
        logger.info("CLI: approved %s", tool_name)
    else:
        req.reject(note="User rejected from CLI")
        logger.info("CLI: rejected %s", tool_name)


def _decide_autonomous(req, tool_name: str, tool_args: dict) -> None:
    """Auto-approve allowlisted tools; reject everything else."""
    if is_allowlisted(tool_name, tool_args):
        req.confirm()
        logger.info("AUTO-APPROVED (allowlist): %s", tool_name)
    else:
        req.reject(note="Auto-rejected: not in allowlist")
        logger.info("AUTO-REJECTED (not allowlisted): %s %s", tool_name, str(tool_args)[:100])


def _handle_tool_confirmation(agent, run_output, session_id: str, *, autonomous: bool = False):
    """Prompt user to approve/reject dangerous tool calls.

    When Agno pauses a run because a tool requires confirmation, this
    function displays the action, asks for approval via stdin, and
    resumes or rejects the run accordingly.

    When autonomous=True (or stdin is not a terminal), tool calls are
    checked against config/allowlist.yaml instead of prompting.
    Allowlisted calls are auto-approved; everything else is auto-rejected.

    Returns the final RunOutput after all confirmations are resolved.
    """
    interactive = _is_interactive() and not autonomous
    decide = _prompt_interactive if interactive else _decide_autonomous

    max_rounds = 10  # safety limit
    for _ in range(max_rounds):
        status = getattr(run_output, "status", None)
        is_paused = status == "PAUSED" or str(status) == "RunStatus.paused"
        if not is_paused:
            return run_output

        reqs = getattr(run_output, "active_requirements", None) or []
        if not reqs:
            return run_output

        for req in reqs:
            if not getattr(req, "needs_confirmation", False):
                continue
            te = req.tool_execution
            tool_name = getattr(te, "tool_name", "unknown")
            tool_args = getattr(te, "tool_args", {}) or {}
            decide(req, tool_name, tool_args)

        # Resume the run so the agent sees the confirmation result
        try:
            run_output = agent.continue_run(
                run_response=run_output, stream=False, session_id=session_id
            )
        except Exception as exc:
            logger.error("CLI: continue_run failed: %s", exc)
            typer.echo(f"Error resuming: {exc}", err=True)
            return run_output

    return run_output


@app.command()
def tick(
    prompt: str | None = typer.Argument(
        None, help="Optional journal prompt for Timmy to reflect on"
    ),
):
    """Run one thinking cycle. Pass a prompt to ask Timmy a specific question."""
    import asyncio

    from timmy.thinking import thinking_engine

    thought = asyncio.run(thinking_engine.think_once(prompt=prompt))
    if thought:
        typer.echo(f"[{thought.seed_type}] {thought.content}")
    else:
        typer.echo("No thought produced (thinking disabled or Ollama down).")


@app.command()
def think(
    topic: str = typer.Argument(..., help="Topic to reason about"),
    backend: str | None = _BACKEND_OPTION,
    model_size: str | None = _MODEL_SIZE_OPTION,
):
    """Ask Timmy to think carefully about a topic."""
    timmy = create_timmy(backend=backend, session_id=_CLI_SESSION_ID)
    timmy.print_response(f"Think carefully about: {topic}", stream=True, session_id=_CLI_SESSION_ID)


def _read_message_input(message: list[str]) -> str:
    """Join CLI arguments and read from stdin when appropriate."""
    message_str = " ".join(message)

    if message_str == "-" or not _is_interactive():
        try:
            stdin_content = sys.stdin.read().strip()
        except (KeyboardInterrupt, EOFError):
            stdin_content = ""
        if stdin_content:
            message_str = stdin_content
        elif message_str == "-":
            typer.echo("No input provided via stdin.", err=True)
            raise typer.Exit(1)

    return message_str


def _resolve_session_id(session_id: str | None, new_session: bool) -> str:
    """Return the effective session ID based on CLI flags."""
    import uuid

    if session_id is not None:
        return session_id
    if new_session:
        return str(uuid.uuid4())
    return _CLI_SESSION_ID


@app.command()
def chat(
    message: list[str] = typer.Argument(
        ..., help="Message to send (multiple words are joined automatically)"
    ),
    backend: str | None = _BACKEND_OPTION,
    model_size: str | None = _MODEL_SIZE_OPTION,
    new_session: bool = typer.Option(
        False,
        "--new",
        "-n",
        help="Start a fresh conversation (ignore prior context)",
    ),
    session_id: str | None = typer.Option(
        None,
        "--session-id",
        help="Use a specific session ID for this conversation",
    ),
    autonomous: bool = typer.Option(
        False,
        "--autonomous",
        "-a",
        help="Autonomous mode: auto-approve allowlisted tools, reject the rest (no stdin prompts)",
    ),
):
    """Send a message to Timmy.

    Conversation history persists across invocations. Use --new to start fresh,
    or --session-id to use a specific session.

    Use --autonomous for non-interactive contexts (scripts, dev loops). Tool
    calls are checked against config/allowlist.yaml — allowlisted operations
    execute automatically, everything else is safely rejected.

    Read from stdin by passing "-" as the message or piping input.
    """
    message_str = _read_message_input(message)
    session_id = _resolve_session_id(session_id, new_session)
    timmy = create_timmy(backend=backend, session_id=session_id)

    run_output = timmy.run(message_str, stream=False, session_id=session_id)
    run_output = _handle_tool_confirmation(timmy, run_output, session_id, autonomous=autonomous)

    content = run_output.content if hasattr(run_output, "content") else str(run_output)
    if content:
        from timmy.session import _clean_response

        typer.echo(_clean_response(content))


@app.command()
def repl(
    backend: str | None = _BACKEND_OPTION,
    model_size: str | None = _MODEL_SIZE_OPTION,
    session_id: str | None = typer.Option(
        None,
        "--session-id",
        help="Use a specific session ID for this conversation",
    ),
):
    """Start an interactive REPL session with Timmy.

    Keeps the agent warm between messages. Conversation history is persisted
    across invocations. Use Ctrl+C or Ctrl+D to exit gracefully.
    """
    from timmy.session import chat

    if session_id is None:
        session_id = _CLI_SESSION_ID

    typer.echo(typer.style("Timmy REPL", bold=True))
    typer.echo("Type your messages below. Use Ctrl+C or Ctrl+D to exit.")
    typer.echo()

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    try:
        while True:
            try:
                user_input = input("> ")
            except (KeyboardInterrupt, EOFError):
                typer.echo()
                typer.echo("Goodbye!")
                break

            user_input = user_input.strip()
            if not user_input:
                continue

            if user_input.lower() in ("exit", "quit", "q"):
                typer.echo("Goodbye!")
                break

            try:
                response = loop.run_until_complete(chat(user_input, session_id=session_id))
                if response:
                    typer.echo(response)
                    typer.echo()
            except Exception as exc:
                typer.echo(f"Error: {exc}", err=True)
    finally:
        loop.close()


@app.command()
def status(
    backend: str | None = _BACKEND_OPTION,
    model_size: str | None = _MODEL_SIZE_OPTION,
):
    """Print Timmy's operational status."""
    timmy = create_timmy(backend=backend, session_id=_CLI_SESSION_ID)
    timmy.print_response(STATUS_PROMPT, stream=False, session_id=_CLI_SESSION_ID)


@app.command()
def interview(
    backend: str | None = _BACKEND_OPTION,
    model_size: str | None = _MODEL_SIZE_OPTION,
):
    """Initialize Timmy and run a structured interview.

    Asks Timmy a series of questions about his identity, capabilities,
    values, and operation to verify he is working correctly.
    """
    import asyncio

    from timmy.interview import InterviewEntry, format_transcript, run_interview
    from timmy.session import chat

    typer.echo("Initializing Timmy for interview...\n")

    # Use a single persistent event loop for the entire interview.
    # Calling asyncio.run() per question kills the loop between calls,
    # orphaning MCP stdio transports and causing "Event loop is closed."
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    try:
        # Force agent creation by calling chat once with a warm-up prompt
        try:
            loop.run_until_complete(
                chat(
                    "Hello, Timmy. We're about to start your interview.",
                    session_id="interview",
                )
            )
        except Exception as exc:
            typer.echo(f"Warning: Initialization issue — {exc}", err=True)

        def _on_answer(entry: InterviewEntry) -> None:
            typer.echo(f"[{entry.category}]")
            typer.echo(f"  Q: {entry.question}")
            typer.echo(f"  A: {entry.answer}")
            typer.echo()

        typer.echo("Starting interview...\n")
        transcript = run_interview(
            chat_fn=lambda msg: loop.run_until_complete(chat(msg, session_id="interview")),
            on_answer=_on_answer,
        )

        # Print full transcript at the end
        typer.echo("\n" + format_transcript(transcript))
    finally:
        # Clean shutdown: close MCP sessions, then the loop
        try:
            from timmy.mcp_tools import close_mcp_sessions

            loop.run_until_complete(close_mcp_sessions())
        except Exception as exc:
            logger.warning("MCP session close failed: %s", exc)
            pass
        loop.close()


@app.command()
def up(
    dev: bool = typer.Option(False, "--dev", help="Enable hot-reload for development"),
    build: bool = typer.Option(True, "--build/--no-build", help="Rebuild images before starting"),
):
    """Start Timmy Time in Docker (dashboard + agents)."""
    cmd = ["docker", "compose"]
    if dev:
        cmd += ["-f", "docker-compose.yml", "-f", "docker-compose.dev.yml"]
    cmd += ["up", "-d"]
    if build:
        cmd.append("--build")

    mode = "dev mode (hot-reload active)" if dev else "production mode"
    typer.echo(f"Starting Timmy Time in {mode}...")
    result = subprocess.run(cmd)
    if result.returncode == 0:
        typer.echo(f"\n  Timmy Time running at http://localhost:8000  ({mode})\n")
    else:
        typer.echo("Failed to start. Is Docker running?", err=True)
        raise typer.Exit(1)


@app.command()
def down():
    """Stop all Timmy Time Docker containers."""
    subprocess.run(["docker", "compose", "down"], check=True)


@app.command()
def voice(
    whisper_model: str = typer.Option(
        "base.en",
        "--whisper",
        "-w",
        help="Whisper model: tiny.en, base.en, small.en, medium.en",
    ),
    use_say: bool = typer.Option(False, "--say", help="Use macOS `say` instead of Piper TTS"),
    threshold: float = typer.Option(
        0.015,
        "--threshold",
        "-t",
        help="Mic silence threshold (RMS). Lower = more sensitive.",
    ),
    silence: float = typer.Option(1.5, "--silence", help="Seconds of silence to end recording"),
    backend: str | None = _BACKEND_OPTION,
    model_size: str | None = _MODEL_SIZE_OPTION,
):
    """Start the sovereign voice loop — listen, think, speak.

    Everything runs locally: Whisper for STT, Ollama for LLM, Piper for TTS.
    No cloud, no network calls, no microphone data leaves your machine.
    """
    from timmy.voice_loop import VoiceConfig, VoiceLoop

    config = VoiceConfig(
        whisper_model=whisper_model,
        use_say_fallback=use_say,
        silence_threshold=threshold,
        silence_duration=silence,
        backend=backend,
        model_size=model_size,
    )
    loop = VoiceLoop(config=config)
    loop.run()


@app.command()
def route(
    message: list[str] = typer.Argument(..., help="Message to route"),
):
    """Show which agent would handle a message (debug routing)."""
    full_message = " ".join(message)
    from timmy.agents.loader import route_request_with_match

    agent_id, matched_pattern = route_request_with_match(full_message)
    if agent_id:
        typer.echo(f"→ {agent_id} (matched: {matched_pattern})")
    else:
        typer.echo("→ orchestrator (no pattern match)")


@app.command()
def focus(
    topic: str | None = typer.Argument(
        None,
        help='Topic to focus on (e.g. "three-phase loop"). Omit to show current focus.',
    ),
    clear: bool = typer.Option(False, "--clear", "-c", help="Clear focus and return to broad mode"),
):
    """Set deep-focus mode on a single problem.

    When focused, Timmy prioritizes the active topic in all responses
    and deprioritizes unrelated context. Focus persists across sessions.

    Examples:
        timmy focus "three-phase loop"   # activate deep focus
        timmy focus                       # show current focus
        timmy focus --clear               # return to broad mode
    """
    from timmy.focus import focus_manager

    if clear:
        focus_manager.clear()
        typer.echo("Focus cleared — back to broad mode.")
        return

    if topic:
        focus_manager.set_topic(topic)
        typer.echo(f'Deep focus activated: "{topic}"')
    else:
        # Show current focus status
        if focus_manager.is_focused():
            typer.echo(f'Deep focus: "{focus_manager.get_topic()}"')
        else:
            typer.echo("No active focus (broad mode).")


@app.command(name="healthcheck")
def healthcheck(
    json_output: bool = typer.Option(False, "--json", "-j", help="Output as JSON"),
    verbose: bool = typer.Option(
        False, "--verbose", "-v", help="Show verbose output including issue details"
    ),
    quiet: bool = typer.Option(False, "--quiet", "-q", help="Only show status line (no details)"),
):
    """Quick health snapshot before coding.

    Shows CI status, critical issues (P0/P1), test flakiness, and token economy.
    Fast execution (< 5 seconds) for pre-work checks.

    Refs: #710
    """
    import subprocess
    import sys
    from pathlib import Path

    script_path = (
        Path(__file__).resolve().parent.parent.parent
        / "timmy_automations"
        / "daily_run"
        / "health_snapshot.py"
    )

    cmd = [sys.executable, str(script_path)]
    if json_output:
        cmd.append("--json")
    if verbose:
        cmd.append("--verbose")
    if quiet:
        cmd.append("--quiet")

    result = subprocess.run(cmd)
    raise typer.Exit(result.returncode)


@app.command()
def learn(
    target: str | None = typer.Option(
        None,
        "--target",
        "-t",
        help="Module or file to optimise (e.g. 'src/timmy/agent.py')",
    ),
    metric: str = typer.Option(
        "unit_pass_rate",
        "--metric",
        "-m",
        help="Metric to track: unit_pass_rate | coverage | val_bpb | <custom>",
    ),
    budget: int = typer.Option(
        5,
        "--budget",
        help="Time limit per experiment in minutes",
    ),
    max_experiments: int = typer.Option(
        10,
        "--max-experiments",
        help="Cap on total experiments per run",
    ),
    dry_run: bool = typer.Option(
        False,
        "--dry-run",
        help="Show hypothesis without executing experiments",
    ),
    program_file: str | None = typer.Option(
        None,
        "--program",
        "-p",
        help="Path to research direction file (default: program.md in cwd)",
    ),
    tox_env: str = typer.Option(
        "unit",
        "--tox-env",
        help="Tox environment to run for each evaluation",
    ),
    model: str = typer.Option(
        "qwen3:30b",
        "--model",
        help="Ollama model forwarded to Aider for code edits",
    ),
):
    """Start an autonomous improvement loop (autoresearch).

    Reads program.md for research direction, then iterates:
    hypothesis → edit → tox → evaluate → commit/revert.

    Experiments continue until --max-experiments is reached or the loop is
    interrupted with Ctrl+C.  Use --dry-run to preview hypotheses without
    making any changes.

    Example:
        timmy learn --target src/timmy/agent.py --metric unit_pass_rate
    """
    from pathlib import Path

    from timmy.autoresearch import SystemExperiment

    repo_root = Path.cwd()
    program_path = Path(program_file) if program_file else repo_root / "program.md"

    if program_path.exists():
        program_content = program_path.read_text()
        typer.echo(f"Research direction: {program_path}")
    else:
        program_content = ""
        typer.echo(
            f"Note: {program_path} not found — proceeding without research direction.",
            err=True,
        )

    if target is None:
        typer.echo(
            "Error: --target is required. Specify the module or file to optimise.",
            err=True,
        )
        raise typer.Exit(1)

    experiment = SystemExperiment(
        target=target,
        metric=metric,
        budget_minutes=budget,
    )

    typer.echo()
    typer.echo(typer.style("Autoresearch", bold=True) + f" — {target}")
    typer.echo(f"  metric={metric}  budget={budget}min  max={max_experiments}  tox={tox_env}")
    if dry_run:
        typer.echo("  (dry-run — no changes will be made)")
    typer.echo()

    def _progress_callback(iteration: int, max_iter: int, message: str) -> None:
        """Print progress updates during experiment iterations."""
        if iteration > 0:
            prefix = typer.style(f"[{iteration}/{max_iter}]", bold=True)
            typer.echo(f"{prefix} {message}")

    try:
        # Run the full experiment loop via the SystemExperiment class
        result = experiment.run(
            tox_env=tox_env,
            model=model,
            program_content=program_content,
            max_iterations=max_experiments,
            dry_run=dry_run,
            create_branch=False,  # CLI mode: work on current branch
        )

        # Display results for each iteration
        for i, record in enumerate(experiment.results, 1):
            _progress_callback(i, max_experiments, record["hypothesis"])

            if dry_run:
                continue

            # Edit phase result
            typer.echo("  → editing …", nl=False)
            if record.get("edit_failed"):
                typer.echo(f" skipped ({record.get('edit_result', 'unknown')})")
            else:
                typer.echo(" done")

            # Evaluate phase result
            duration = record.get("duration_s", 0)
            typer.echo(f"  → running tox … {duration}s")

            # Assessment
            assessment = record.get("assessment", "No assessment")
            typer.echo(f"  → {assessment}")

            # Outcome
            if record.get("committed"):
                typer.echo("  → committed")
            elif record.get("reverted"):
                typer.echo("  → reverted (no improvement)")

            typer.echo()

    except KeyboardInterrupt:
        typer.echo("\nInterrupted.")
        raise typer.Exit(0) from None

    typer.echo(typer.style("Autoresearch complete.", bold=True))
    if result.get("baseline") is not None:
        typer.echo(f"Final {metric}: {result['baseline']:.4f}")


def main():
    app()