Source code for smarter.apps.mcpserver.logs.tools

"""
The operations tools of the platform's pod logs: ``list_log_sources``, ``search_logs`` and ``summarize_log_errors``.

They read :data:`smarter.apps.infrastructure.services.infrastructure` ``.logs``: the last 72 hours
of the app (``smarter``), the Celery workers (``smarter-worker``, ``smarter-worker-infrastructure``),
beat (``smarter-beat``) and MariaDB (``mariadb``). Each query is structured, not LogQL, so the LLM
cannot inject a query of its own.

They are tools of the staff-only Smarter Operations MCP server, built on
:class:`~smarter.apps.mcpserver.protocol.operations_tools.SmarterOperationsTool`, which records every
call in the audit trail. A result that quotes log lines starts with
:data:`~smarter.apps.mcpserver.protocol.operations_tools.UNTRUSTED_DATA_NOTICE`, because a log line
can carry text that a user wrote.

**Side effects:** each call makes one or two HTTP GETs to Loki (in the tests, to the in-memory log
service), and the audit trail's database write. Nothing is cached.

.. note::

    **Experimental.** The MCPServer was designed and coded by Claude Code (Anthropic's
    Claude Opus 5.5), with Lawrence McDaniel as co-author. It is experimental, and will
    be documented.
"""

from datetime import datetime, timedelta, timezone
from typing import Literal, Optional

from pydantic import Field, ValidationError

from smarter.apps.infrastructure.services import infrastructure
from smarter.apps.infrastructure.services.logs import (
    LOG_RETENTION,
    LogLevel,
    LogQuery,
    at_least,
)
from smarter.apps.mcpserver.exceptions import SmarterMCPServerToolError
from smarter.apps.mcpserver.logs.summary import error_groups, format_line
from smarter.apps.mcpserver.protocol.operations_tools import (
    UNTRUSTED_DATA_NOTICE,
    OperationsToolArguments,
    SmarterOperationsTool,
)

COMPONENTS = ("smarter", "smarter-worker", "smarter-worker-infrastructure", "smarter-beat", "mariadb")
"""The components whose logs are kept: the ``app.kubernetes.io/name`` of their pods."""

Component = Literal["smarter", "smarter-worker", "smarter-worker-infrastructure", "smarter-beat", "mariadb"]
"""
A component's name, as a tool argument: one of :data:`COMPONENTS`.

A ``Literal`` puts the names in the tool's ``inputSchema`` as an ``enum``, so that the LLM sees
them, and Pydantic refuses any other.
"""

MAX_MINUTES = int(LOG_RETENTION.total_seconds() // 60)
"""The longest period that a tool reads: the log store's retention."""

SUMMARY_LINES = 5000
"""The most warning and error lines that ``summarize_log_errors`` groups."""

COMPONENTS_DESCRIPTION = (
    "The components to read: smarter (the web app and api), smarter-worker (Celery tasks, e.g. prompts' "
    "background work), smarter-worker-infrastructure (deploys, DNS, certificates, Kubernetes), smarter-beat "
    "(the scheduler) and mariadb (the database). Omit it to read all of them."
)
"""The description of a tool's ``components`` argument, which says what each component does, for the LLM."""


def log_query(**kwargs) -> LogQuery:
    """
    Return a log query, with a validation error as a tool error that the LLM can correct.

    :class:`~smarter.apps.infrastructure.services.logs.LogQuery` validates what the tool's arguments
    model cannot, e.g. that a ``pattern`` compiles, or that a ``pod`` is a Kubernetes name.

    :param kwargs: The query's fields, for
        :meth:`~smarter.apps.infrastructure.services.logs.LogQuery.last`, starting with ``period``.
    :returns: The query, of the most recent period.
    :raises SmarterMCPServerToolError: If the query is not valid.
    """
    try:
        query = LogQuery.last(**kwargs)
    except ValidationError as e:
        messages = [str(problem["msg"]) for problem in e.errors()]
        problems = "; ".join(messages)
        raise SmarterMCPServerToolError(f"invalid log query: {problems}") from e
    return query


class ListLogSourcesArguments(OperationsToolArguments):
    """The arguments of ``list_log_sources``: the ``ticket``, and how far back to look."""

    minutes: int = Field(default=60, ge=1, le=MAX_MINUTES, description="How far back to look, in minutes.")
    """How far back to look, in minutes, from 1 to :data:`MAX_MINUTES`.

    Defaults to an hour.
    """




class ListLogSourcesTool(SmarterOperationsTool):
    """
    The ``list_log_sources`` tool: the components and pods that logged recently.

    A component that is missing from the list wrote nothing in the period, which usually means
    that its pods are down. The pods' names are what ``search_logs``' ``pod`` argument takes.
    """

    name = "list_log_sources"
    title = "List log sources"
    description = (
        "List the components and pods that wrote to the logs in the last minutes, from the log store, which "
        "keeps 72 hours. A component that is missing has written nothing, e.g. because its pods are down. "
        "Pod names change on every deploy and restart: use them with search_logs' pod argument."
    )
    arguments_model = ListLogSourcesArguments



    def run(self, arguments: ListLogSourcesArguments) -> str:  # type: ignore[override]
        """
        List the components and pods that wrote to the logs in the period.

        **Side effects:** two HTTP GETs to Loki, for the values of the ``component`` and ``pod``
        labels.

        :param arguments: The validated arguments.
        :returns: Markdown: the components, those of :data:`COMPONENTS` that wrote nothing, and the pods.
        :raises LogServiceError: If Loki could not be reached.
        :raises InfrastructureNotReadyError: If there is no log store.
        """
        end = datetime.now(timezone.utc)
        start = end - timedelta(minutes=arguments.minutes)
        components = infrastructure.logs.label_values("component", start, end)
        pods = infrastructure.logs.label_values("pod", start, end)
        missing = [component for component in COMPONENTS if component not in components]

        lines = [f"## Log sources in the last {arguments.minutes} minutes", ""]
        if components:
            component_list = ", ".join(components)
        else:
            component_list = "none"
        lines.append("Components: " + component_list)
        if missing:
            lines.append("Components that wrote nothing: " + ", ".join(missing))
        lines.append("")
        lines.append("Pods:")
        if pods:
            for pod in pods:
                lines.append(f"- {pod}")
        else:
            lines.append("- none")
        text = "\n".join(lines)
        return text




class SearchLogsArguments(OperationsToolArguments):
    """The arguments of ``search_logs``: the ``ticket``, and the filters of the search, which all apply at once."""

    components: list[Component] = Field(default_factory=list, description=COMPONENTS_DESCRIPTION)
    """The components to read, or an empty list for all of them."""

    pod: Optional[str] = Field(default=None, description="One pod's name, from list_log_sources.")
    """One pod's name, from ``list_log_sources``, or None for every pod of the components."""

    contains: list[str] = Field(
        default_factory=list,
        max_length=5,
        description="Text that every line must contain, case insensitively, e.g. ['timeout', 'openai'].",
    )
    """Up to five texts, each of which every line must contain, case insensitively."""

    pattern: Optional[str] = Field(
        default=None, description="A regular expression, in RE2 syntax, that every line must match."
    )
    """A regular expression, in RE2 syntax, that every line must match, or None."""

    min_level: Optional[LogLevel] = Field(
        default=None,
        description="The least severe level to return, e.g. 'warning' for warnings, errors and critical lines.",
    )
    """The least severe level to return, or None for lines of every level, and lines without one."""

    minutes: int = Field(default=60, ge=1, le=MAX_MINUTES, description="How far back to search, in minutes.")
    """How far back to search, in minutes, from 1 to :data:`MAX_MINUTES`.

    Defaults to an hour.
    """

    limit: int = Field(default=50, ge=1, le=200, description="The most lines to return.")
    """The most lines to return, from 1 to 200.

    Defaults to 50.
    """

    oldest_first: bool = Field(
        default=False, description="Return the oldest lines of the period first, rather than the newest."
    )
    """Whether to return the oldest lines of the period first.

    By default the newest are, like ``tail``.
    """




class SearchLogsTool(SmarterOperationsTool):
    """
    The ``search_logs`` tool: the lines of the pod logs that match a set of filters.

    The filters become a :class:`~smarter.apps.infrastructure.services.logs.LogQuery`, which the log
    service turns into escaped LogQL, so the LLM's text is never run as a query.
    """

    name = "search_logs"
    title = "Search pod logs"
    description = (
        "Search the logs of the platform's pods, from the log store, which keeps 72 hours. Filter by component, "
        "pod, text, a regular expression and the least severe level. Returns the newest matching lines first, "
        "each with its time (UTC), component and pod. With no filters, it returns the latest lines, like tail. "
        "To find what went wrong, start with summarize_log_errors, then search for a signature it returns."
    )
    arguments_model = SearchLogsArguments



    def run(self, arguments: SearchLogsArguments) -> str:  # type: ignore[override]
        """
        Search the pod logs.

        **Side effects:** one HTTP GET to Loki's ``query_range``.

        :param arguments: The validated arguments.
        :returns: Markdown: a heading with the number of lines, then the lines in a code block, after
            :data:`~smarter.apps.mcpserver.protocol.operations_tools.UNTRUSTED_DATA_NOTICE`, or a
            hint to widen the search if no line matches.
        :raises SmarterMCPServerToolError: If the query is not valid, e.g. its pattern does not compile.
        :raises LogServiceError: If Loki could not be reached, or rejected the query.
        :raises InfrastructureNotReadyError: If there is no log store.
        """
        if arguments.oldest_first:
            direction = "forward"
            order = "oldest first"
        else:
            direction = "backward"
            order = "newest first"
        query = log_query(
            period=timedelta(minutes=arguments.minutes),
            components=tuple(arguments.components),
            pod=arguments.pod,
            contains=tuple(arguments.contains),
            pattern=arguments.pattern,
            min_level=arguments.min_level,
            limit=arguments.limit,
            direction=direction,
        )
        lines = infrastructure.logs.query(query)
        if arguments.min_level:
            # Loki matches the level anywhere in the line; keep the lines whose own level it is.
            lines = [line for line in lines if at_least(line.level, arguments.min_level)]

        heading = f"## {len(lines)} log lines in the last {arguments.minutes} minutes"
        if not lines:
            return f"{heading}\n\nNo line matches. Widen the period, or remove a filter."
        if len(lines) == arguments.limit:
            heading += f" (the limit, {order}: there may be more)"
        else:
            heading += f" ({order})"
        formatted_lines = [format_line(line) for line in lines]
        body = "\n".join(formatted_lines)
        text = f"{UNTRUSTED_DATA_NOTICE}{heading}\n\n```text\n{body}\n```"
        return text




class SummarizeLogErrorsArguments(OperationsToolArguments):
    """The arguments of ``summarize_log_errors``: the ``ticket``, the components, the period, the level and the number of groups."""

    components: list[Component] = Field(default_factory=list, description=COMPONENTS_DESCRIPTION)
    """The components to read, or an empty list for all of them."""

    minutes: int = Field(default=60, ge=1, le=MAX_MINUTES, description="How far back to look, in minutes.")
    """How far back to look, in minutes, from 1 to :data:`MAX_MINUTES`.

    Defaults to an hour.
    """

    min_level: Literal["warning", "error"] = Field(
        default="error", description="'error' for errors and critical lines, 'warning' to include warnings."
    )
    """``error`` for errors and critical lines, the default, or ``warning`` to include warnings."""

    groups: int = Field(default=15, ge=1, le=50, description="The most groups to return.")
    """The most groups to show, from 1 to 50.

    Defaults to 15. The result says how many more there are.
    """




class SummarizeLogErrorsTool(SmarterOperationsTool):
    """
    The ``summarize_log_errors`` tool: the errors of the pod logs, grouped by their signature.

    It reads up to :data:`SUMMARY_LINES` warning and error lines, and groups them with
    :func:`~smarter.apps.mcpserver.logs.summary.error_groups`, so that the LLM sees each problem
    once, with how often it happened, rather than thousands of lines.
    """

    name = "summarize_log_errors"
    title = "Summarize log errors"
    description = (
        "Summarize the errors, and optionally the warnings, in the platform's pod logs over the last minutes: "
        "the lines are grouped by component, level and signature (the message without its numbers, ids and "
        "quoted values), and each group has its count, its pods, when it was first and last seen, and its "
        "latest line. Call it first when a user reports errors, slowness or an outage, then use search_logs "
        "for the lines around a group."
    )
    arguments_model = SummarizeLogErrorsArguments



    def run(self, arguments: SummarizeLogErrorsArguments) -> str:  # type: ignore[override]
        """
        Group the errors, and optionally the warnings, of the period.

        **Side effects:** one HTTP GET to Loki's ``query_range``.

        :param arguments: The validated arguments.
        :returns: Markdown: a heading with the number of lines, then one section for each group, the
            most frequent first, after
            :data:`~smarter.apps.mcpserver.protocol.operations_tools.UNTRUSTED_DATA_NOTICE`, or a
            heading that says there were none.
        :raises SmarterMCPServerToolError: If the query is not valid.
        :raises LogServiceError: If Loki could not be reached, or rejected the query.
        :raises InfrastructureNotReadyError: If there is no log store.
        """
        min_level = LogLevel(arguments.min_level)
        query = log_query(
            period=timedelta(minutes=arguments.minutes),
            components=tuple(arguments.components),
            min_level=min_level,
            limit=SUMMARY_LINES,
        )
        matching_lines = infrastructure.logs.query(query)
        # Loki matches the level anywhere in the line; keep the lines whose own level it is.
        lines = [line for line in matching_lines if at_least(line.level, min_level)]
        if min_level == LogLevel.ERROR:
            levels = "errors"
        else:
            levels = "warnings and errors"

        heading = f"## {len(lines)} {levels} in the last {arguments.minutes} minutes"
        if not lines:
            return f"{heading}\n\nNone. The logs show no {levels} in this period."
        if len(lines) >= SUMMARY_LINES:
            heading += f" (only the latest {SUMMARY_LINES} were grouped)"

        groups = error_groups(lines)
        shown_groups = groups[: arguments.groups]
        out = [UNTRUSTED_DATA_NOTICE + heading, ""]
        for group in shown_groups:
            seen = f"{group.first_seen:%H:%M:%S} to {group.last_seen:%H:%M:%S} UTC"
            if group.level:
                level = group.level
            else:
                level = "unknown"
            pod_count = len(group.pods)
            out.append(f"### {group.count} × {level} in {group.component} ({pod_count} pods, {seen})")
            out.append(f"Signature: `{group.signature}`")
            out.append("```text")
            out.append(group.example)
            out.append("```")
            out.append("")
        hidden_groups = len(groups) - arguments.groups
        if hidden_groups > 0:
            out.append(f"_{hidden_groups} more groups are not shown._")
        text = "\n".join(out)
        return text




__all__ = [
    "COMPONENTS",
    "ListLogSourcesTool",
    "SearchLogsTool",
    "SummarizeLogErrorsTool",
]