tracer-cloud--opensre/surfaces/interactive_shell/grounding/cli_reference.py

"""Reference text for OpenSRE interactive-shell CLI answers."""

from __future__ import annotations

import logging
import time
from collections.abc import Callable, Mapping
from typing import Any

import click

from core.agent_harness.grounding.diagnostics import (
    GroundingSource,
    log_grounding_cache_diagnostics,
)
from core.agent_harness.grounding.models import CacheStats
from core.agent_harness.prompts.prompt_context import DefaultPromptContextProvider

_logger = logging.getLogger(__name__)

_MAX_REFERENCE_CHARS = 28_000

# Heuristic: truncated or failed reference output must not be cached or the
# assistant would keep an empty reference for the whole process.
_MIN_CACHEABLE_CLI_REFERENCE_CHARS = 80
_CLI_REFERENCE_SENTINEL = "=== opensre --help ==="
SlashCommandProvider = Callable[[], Mapping[str, Any]]
CommandGroupProvider = Callable[[], click.Command | None]


def _is_cacheable_cli_reference(text: str) -> bool:
    stripped = text.strip()
    if len(stripped) < _MIN_CACHEABLE_CLI_REFERENCE_CHARS:
        return False
    return _CLI_REFERENCE_SENTINEL in text


def _slash_command_names(provider: SlashCommandProvider | None) -> list[str]:
    if provider is None:
        return []
    return sorted(provider().keys())


def _current_cli_signature(
    cli_group: click.Command | None,
    slash_provider: SlashCommandProvider | None = None,
) -> str:
    """Stable signature of the CLI command surface and interactive slash commands.

    Bumps cache when subcommands change, slash-command metadata changes, or the
    installed package version changes.
    """
    from config.version import get_opensre_version

    if isinstance(cli_group, click.Group):
        cmd_names = ",".join(sorted(cli_group.commands.keys()))
    else:
        cmd_names = ""
    slash_names = ",".join(_slash_command_names(slash_provider))
    return f"opensre={get_opensre_version()}|commands={cmd_names}|slash={slash_names}"


def _format_param(param: click.Parameter) -> str:
    """Return a compact, side-effect-free description of a Click parameter."""
    if isinstance(param, click.Option):
        names = ", ".join((*param.opts, *param.secondary_opts))
        if not names:
            names = param.name or "(option)"
        value_hint = ""
        if not param.is_flag and not param.count:
            value_hint = " " + (param.metavar or param.name or "VALUE").upper()
        default = ""
        if param.show_default and param.default not in (None, "", ()):
            default = f" [default: {param.default}]"
        help_text = (param.help or "").strip()
        return f"  {names}{value_hint} - {help_text}{default}".rstrip()

    if isinstance(param, click.Argument):
        name = (param.name or "ARG").upper()
        required = "" if param.required else " (optional)"
        return f"  {name}{required}"

    return f"  {param.name or type(param).__name__}"


def _format_command_reference(
    command: click.Command,
    *,
    path: str,
    include_subcommands: bool = True,
) -> str:
    """Render Click command metadata without invoking help callbacks.

    ``click.Command.get_help()`` eventually calls ``format_help()``. The root
    OpenSRE group overrides that path to render via Rich's live console, which
    can leak into the interactive terminal when the assistant builds grounding
    context. This renderer inspects command objects directly instead.
    """
    lines = [f"Usage: {path} [OPTIONS]"]
    if isinstance(command, click.Group):
        lines[0] += " COMMAND [ARGS]..."
    elif command.params:
        arg_names = [
            (param.name or "ARG").upper()
            for param in command.params
            if isinstance(param, click.Argument)
        ]
        if arg_names:
            lines[0] += " " + " ".join(arg_names)

    help_text = (command.help or command.short_help or "").strip()
    if help_text:
        lines.extend(["", help_text])

    params = [param for param in command.params if not getattr(param, "hidden", False)]
    if params:
        lines.extend(["", "Options/Arguments:"])
        lines.extend(_format_param(param) for param in params)

    if include_subcommands and isinstance(command, click.Group):
        command_rows: list[tuple[str, str]] = []
        with click.Context(command, info_name=path.rsplit(" ", 1)[-1]) as ctx:
            for name in command.list_commands(ctx):
                subcommand = command.get_command(ctx, name)
                if subcommand is None or subcommand.hidden:
                    continue
                command_rows.append((name, subcommand.get_short_help_str(limit=160)))
        if command_rows:
            lines.extend(["", "Commands:"])
            lines.extend(f"  {name} - {summary}".rstrip() for name, summary in command_rows)

    return "\n".join(lines).rstrip() + "\n"


def _build_cli_reference_text_uncached(
    cli_group: click.Command | None,
    slash_provider: SlashCommandProvider | None = None,
) -> str:
    """Build a side-effect-free CLI reference without invoking Click commands."""
    parts: list[str] = []

    if cli_group is None:
        parts.append("=== opensre --help ===\n")
        parts.append(
            "The CLI command surface is not available in this runtime.\n"
            "Launch the OpenSRE CLI (`opensre`) for command-level help.\n"
        )
    else:
        parts.append("=== opensre --help ===\n")
        parts.append(_format_command_reference(cli_group, path="opensre"))

        if isinstance(cli_group, click.Group):
            with click.Context(cli_group, info_name="opensre") as ctx:
                for name in sorted(cli_group.commands.keys()):
                    command = cli_group.get_command(ctx, name)
                    if command is None or command.hidden:
                        continue
                    parts.append(f"\n=== opensre {name} --help ===\n")
                    parts.append(_format_command_reference(command, path=f"opensre {name}"))

    parts.append("\n=== Interactive-shell slash commands ===\n")
    parts.append(_interactive_shell_slash_hints(slash_provider))

    text = "".join(parts)
    if len(text) > _MAX_REFERENCE_CHARS:
        return text[:_MAX_REFERENCE_CHARS] + "\n\n[... reference truncated ...]\n"
    return text


def _interactive_shell_slash_hints(provider: SlashCommandProvider | None = None) -> str:
    lines = [
        "In the interactive shell, describe an incident or paste alert JSON to run "
        + "a investigation pipeline, or chat with the terminal assistant for CLI help.",
        "Alpha mode runs every shell command with no guardrails: plain commands are parsed to "
        + "argv and run without a shell, while pipes, redirects, command substitution, and a "
        + "leading ! all run through a full shell. There is no read-only allowlist or blocked "
        + "command list.",
        "Slash commands:",
        "",
    ]
    if provider is not None:
        for cmd in provider().values():
            name = getattr(cmd, "name", "")
            description = getattr(cmd, "description", "")
            if name:
                lines.append(f"  {name} - {description}".rstrip())
    else:
        lines.append("  (slash command registry is supplied by the interactive shell runtime)")
    lines.extend(
        [
            "",
            "Non-interactive investigation: `opensre investigate` with stdin, file, or flags.",
            "Launch the interactive shell: `opensre` (requires a TTY).",
        ]
    )
    return "\n".join(lines)


class CliReference:
    """Session-scoped cache for assembled CLI help reference text.

    Holds its cache as instance state so each REPL session (via
    :func:`session_cli_reference`) owns an isolated cache with no module-level
    mutable globals.
    """

    name = "cli"

    def __init__(self) -> None:
        self._signature: str | None = None
        self._text: str | None = None
        self._created_at_monotonic: float = 0.0
        self._hits: int = 0
        self._misses: int = 0
        self._slash_commands_provider: SlashCommandProvider | None = None
        self._command_group_provider: CommandGroupProvider | None = None

    def set_slash_commands_provider(self, provider: SlashCommandProvider | None) -> None:
        """Set the shell-owned slash command source used for grounding text."""
        self._slash_commands_provider = provider
        self.invalidate()

    def set_command_group_provider(self, provider: CommandGroupProvider | None) -> None:
        """Bind a surface-owned callable that returns the root Click command group."""
        self._command_group_provider = provider
        self.invalidate()

    def _resolve_command_group(self) -> click.Command | None:
        provider = self._command_group_provider
        if provider is None:
            return None
        try:
            return provider()
        except Exception:
            _logger.debug("Command group provider raised; skipping CLI reference", exc_info=True)
            return None

    def build_text(self) -> str:
        """Assemble ``opensre`` and subcommand ``--help`` output for LLM grounding.

        Cached on this instance while the command registry signature matches.
        """
        slash_provider = self._slash_commands_provider
        cli_group = self._resolve_command_group()
        sig = _current_cli_signature(cli_group, slash_provider)
        if self._text is not None and self._signature == sig:
            self._hits += 1
            return self._text

        self._misses += 1
        text = _build_cli_reference_text_uncached(cli_group, slash_provider)
        if cli_group is not None and _is_cacheable_cli_reference(text):
            self._signature = sig
            self._text = text
            self._created_at_monotonic = time.monotonic()
        else:
            self._signature = None
            self._text = None
            self._created_at_monotonic = 0.0
            _logger.warning(
                "CLI reference build produced non-cacheable output (%d chars); skipping cache",
                len(text),
            )
        return text

    def invalidate(self) -> None:
        """Drop cached CLI reference text (for tests or forced refresh)."""
        self._signature = None
        self._text = None
        self._created_at_monotonic = 0.0
        self._hits = 0
        self._misses = 0

    def stats(self) -> CacheStats:
        """Debug counters for grounding cache hit/miss and last signature."""
        return CacheStats(
            name=self.name,
            hits=self._hits,
            misses=self._misses,
            cached=self._text is not None,
            signature=self._signature,
            created_at_monotonic=self._created_at_monotonic,
        )

    def as_grounding_source(self) -> GroundingSource:
        return GroundingSource(name=self.name, stats_fn=self.stats)


def _slash_commands() -> Mapping[str, object]:
    from surfaces.interactive_shell.command_registry import SLASH_COMMANDS

    return SLASH_COMMANDS


def _cli_command_group() -> click.Command | None:
    from surfaces.cli.__main__ import cli

    return cli


_SESSION_CLI_REFERENCE_ATTR = "_shell_cli_reference"


def session_cli_reference(session: Any) -> CliReference:
    """Return the session-scoped CLI catalog cache, creating it on first use."""
    cached = getattr(session, _SESSION_CLI_REFERENCE_ATTR, None)
    if isinstance(cached, CliReference):
        return cached
    cli = CliReference()
    cli.set_command_group_provider(_cli_command_group)
    cli.set_slash_commands_provider(_slash_commands)
    setattr(session, _SESSION_CLI_REFERENCE_ATTR, cli)
    return cli


def shell_prompt_context_provider(session: Any) -> ShellPromptContextProvider:
    """Build a prompt provider that reuses the session's CLI catalog cache."""
    return ShellPromptContextProvider(session)


class ShellPromptContextProvider:
    """:class:`core.agent_harness.ports.PromptContextProvider` for the interactive shell.

    Owns CLI catalog assembly (``surfaces.cli`` + slash commands) and delegates
    repo-level grounding (AGENTS.md, environment block) to
    :class:`~core.agent_harness.prompts.prompt_context.DefaultPromptContextProvider`.
    """

    def __init__(self, session: Any) -> None:
        self._base = DefaultPromptContextProvider(session)
        self._cli = session_cli_reference(session)

    def cli_reference(self) -> str:
        return self._cli.build_text()

    def agents_md(self) -> str:
        return self._base.agents_md()

    def investigation_flow(self) -> str:
        return self._base.investigation_flow()

    def environment_block(self) -> str:
        return self._base.environment_block()

    def suggested_synthetic_prompt(self) -> str:
        return self._base.suggested_synthetic_prompt()

    def log_diagnostics(self, reason: str) -> None:
        self._base.log_diagnostics(reason)
        log_grounding_cache_diagnostics([self._cli.as_grounding_source()], reason)


__all__ = [
    "CliReference",
    "CommandGroupProvider",
    "ShellPromptContextProvider",
    "SlashCommandProvider",
    "session_cli_reference",
    "shell_prompt_context_provider",
]