tracer-cloud--opensre/surfaces/interactive_shell/runtime/integration_tool_gathering.py

"""Gather integration evidence for a conversational shell answer.

The bounded think -> call-tools -> observe loop lives in the decoupled
:func:`core.agent_harness.turns.evidence_driver.gather_tool_evidence`. This module is the terminal adapter:
it renders each gathering step to the console and persists the gathered tool
calls into the shell's session storage, then hands the collected observation back
to :func:`interactive_shell.runtime.answer_turn.answer_shell_question`.
"""

from __future__ import annotations

import contextlib
import json
from typing import Any

from rich.console import Console
from rich.markup import escape

from core.agent_harness.turns import evidence_driver
from core.agent_harness.turns.evidence_driver import GatherAgentFactory
from surfaces.interactive_shell.session import Session
from surfaces.interactive_shell.ui import DIM
from surfaces.interactive_shell.utils.error_handling.exception_reporting import report_exception
from surfaces.shared.tool_labels import tool_short_label, tool_source_label

# Cap so a chatty tool result can't blow up persistence writes.
_MAX_PER_TOOL_CHARS = 4_000

# Keys most likely to distinguish back-to-back calls to the same tool.
_GATHER_INPUT_HINT_KEYS: tuple[str, ...] = (
    "metric_name",
    "query",
    "search",
    "filter",
    "expression",
    "promql",
    "service_name",
    "owner",
    "repo",
    "log_group",
    "monitor_id",
    "alert_id",
    "issue_id",
    "trace_id",
    "span_id",
    "dashboard_uid",
    "panel_id",
    "from",
    "to",
    "time_range",
)


class _ShellGatherErrorReporter:
    """Minimal :class:`core.agent_harness.ports.ErrorReporter` over ``report_exception``."""

    def report(self, exc: BaseException, *, context: str, expected: bool = False) -> None:
        report_exception(exc, context=context, expected=expected)


def _truncate_hint(text: str, *, max_len: int = 48) -> str:
    compact = " ".join(text.split())
    if len(compact) <= max_len:
        return compact
    return f"{compact[: max_len - 1]}…"


def _tool_input_hint(tool_input: Any) -> str:
    if not isinstance(tool_input, dict):
        return ""
    hints: list[str] = []
    seen: set[str] = set()
    for key in _GATHER_INPUT_HINT_KEYS:
        value = tool_input.get(key)
        if value in (None, "", [], {}):
            continue
        rendered = _truncate_hint(str(value))
        if not rendered or rendered in seen:
            continue
        seen.add(rendered)
        hints.append(rendered)
        if len(hints) >= 2:
            break
    return " · ".join(hints)


def _format_gathering_progress_line(
    tool_name: str,
    tool_input: Any,
    *,
    repeat_index: int,
) -> str:
    source = tool_source_label(tool_name)
    label = tool_short_label(tool_name, source)
    call_display = f"{source} · {label}" if label else source
    if repeat_index > 1:
        call_display = f"{call_display} ({repeat_index})"
    safe_display = escape(call_display)
    hint = _tool_input_hint(tool_input)
    if hint:
        return f"· gathering via {safe_display} — {escape(hint)}…"
    return f"· gathering via {safe_display}…"


def _resolve_gather_integrations(
    session: Session,
    message: str,
    resolved_integrations: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Resolve gather integrations through the decoupled agent helper."""
    return evidence_driver._resolve_gather_integrations(  # noqa: SLF001
        session, message, resolved_integrations=resolved_integrations
    )


def _truncate(text: str, limit: int) -> str:
    if len(text) <= limit:
        return text
    return text[:limit] + f"\n…[truncated, {len(text)} chars total]"


def _persist_tool_calls(session: Session, executed: list[tuple[Any, Any]]) -> None:
    """Record each gathered tool-call result into the session log.

    Arguments and results are redacted and bounded before writing; failures are
    swallowed so logging never breaks the turn.
    """
    from core.agent_harness.session import default_session_storage
    from platform.observability.trace.redaction import redact_sensitive

    storage = default_session_storage()
    for tc, output in executed:
        with contextlib.suppress(Exception):
            body = (
                output
                if isinstance(output, str)
                else json.dumps(redact_sensitive(output), default=str)
            )
            arguments = (
                redact_sensitive(tc.input) if isinstance(tc.input, dict) else {"value": tc.input}
            )
            storage.append_tool_call(
                session.session_id,
                tool=str(tc.name),
                arguments=arguments,
                result=_truncate(body, _MAX_PER_TOOL_CHARS),
                ok=not (isinstance(output, dict) and "error" in output),
            )


def gather_integration_tool_evidence(
    message: str,
    session: Session,
    console: Console,
    *,
    is_tty: bool | None = None,
    agent_factory: GatherAgentFactory | None = None,
    resolved_integrations: dict[str, Any] | None = None,
) -> str | None:
    """Run a bounded tool-calling loop and return collected evidence, or None.

    Returns a formatted observation block when at least one tool was executed;
    otherwise ``None`` so the caller falls back to the normal text-only answer.
    ``resolved_integrations`` is the turn's resolved view; it is forwarded so the
    gather phase reuses it instead of resolving again.
    """
    tool_call_counts: dict[str, int] = {}

    def on_progress(kind: str, data: dict[str, Any]) -> None:
        if kind == "tool_start":
            name = str(data.get("name", "")).strip() or "tool"
            tool_call_counts[name] = tool_call_counts.get(name, 0) + 1
            line = _format_gathering_progress_line(
                name,
                data.get("input"),
                repeat_index=tool_call_counts[name],
            )
            console.print(f"[{DIM}]{line}[/]")
        elif kind == "gather_cancelled":
            console.print(f"[{DIM}]· gathering cancelled[/]")

    def persist(executed: list[tuple[Any, Any]]) -> None:
        _persist_tool_calls(session, executed)

    return evidence_driver.gather_tool_evidence(
        message,
        session,
        on_progress=on_progress,
        persist=persist,
        error_reporter=_ShellGatherErrorReporter(),
        is_tty=is_tty,
        agent_factory=agent_factory,
        resolved_integrations=resolved_integrations,
    )


__all__ = ["gather_integration_tool_evidence"]