debug_env/server/tools/query_logs/tool.py

"""
query_logs tool — tail and filter a log file in the task workdir.

Diagnostic tool (pass_rate=0.0): reading logs does not affect the score
but helps the agent locate error messages, stack traces, and crash signatures.
"""

import logging
from pathlib import Path

from debug_env.server.schemas import ToolResult

logger = logging.getLogger(__name__)


def query_logs(
    workdir: str,
    log_file_path: str = "app.log",
    filter_string: str = "",
    tail: int = 100,
) -> ToolResult:
    """
    Read the last *tail* lines of a log file, optionally filtered.

    Args:
        workdir:       Absolute path to the task working directory.
        log_file_path: Path to the log file, relative to workdir (default: ``"app.log"``).
        filter_string: Case-insensitive substring filter.  Only lines containing
                       this string are returned.  Empty string → no filter.
        tail:          Maximum number of lines to return (from the end of the file).

    Returns:
        :class:`ToolResult` with ``pass_rate=0.0`` and the log content in ``logs``.
    """
    # Validate path stays inside workdir
    workdir_path = Path(workdir).resolve()
    log_path = (workdir_path / log_file_path).resolve()

    if not str(log_path).startswith(str(workdir_path)):
        return ToolResult(
            pass_rate=0.0,
            logs=f"Access denied: '{log_file_path}' is outside the task workdir.",
            success=False,
        )

    if not log_path.exists():
        return ToolResult(
            pass_rate=0.0,
            logs=f"Log file '{log_file_path}' not found in workdir.",
            success=False,
        )

    try:
        content = log_path.read_text(encoding="utf-8", errors="replace")
    except OSError as exc:
        return ToolResult(pass_rate=0.0, logs=f"Failed to read log: {exc}", success=False)

    lines = content.splitlines()

    if filter_string:
        needle = filter_string.lower()
        lines = [line for line in lines if needle in line.lower()]

    lines = lines[-tail:]

    output = "\n".join(lines) if lines else "(no matching log entries)"
    logger.info(
        "query_logs: file=%s filter=%r tail=%d → %d lines",
        log_file_path,
        filter_string,
        tail,
        len(lines),
    )
    return ToolResult(
        pass_rate=0.0,
        logs=f"# {log_file_path} (last {len(lines)} matching lines)\n{output}",
        success=True,
    )