debug_env/server/tools/curl_endpoint/tool.py

"""
curl_endpoint tool — send an HTTP request to the running buggy server.

Diagnostic tool (pass_rate=0.0): inspecting HTTP responses does not affect
the score but lets the agent probe endpoints and reproduce error conditions.
"""

import json
import logging
import urllib.error
import urllib.request
from typing import Optional

from debug_env.server.schemas import ToolResult

logger = logging.getLogger(__name__)


def curl_endpoint(
    port: int,
    method: str = "GET",
    path: str = "/",
    payload: str = "",
    headers: Optional[dict] = None,
) -> ToolResult:
    """
    Send an HTTP request to ``http://localhost:{port}{path}`` and return the response.

    Args:
        port:    Port the buggy server is listening on (injected by the environment).
        method:  HTTP method — ``"GET"``, ``"POST"``, etc. (case-insensitive).
        path:    Request path, e.g. ``"/health"`` or ``"/jobs"``.
        payload: Optional request body (UTF-8 string, typically JSON).
        headers: Extra request headers as a dict.  ``Content-Type`` defaults to
                 ``application/json`` when *payload* is non-empty.

    Returns:
        :class:`ToolResult` with ``pass_rate=0.0`` and the formatted response in ``logs``.
    """
    if port is None:
        return ToolResult(
            pass_rate=0.0,
            logs="No service is running. Reset with an incident task first.",
            success=False,
        )

    url = f"http://localhost:{port}{path}"
    method_upper = method.upper()
    data = payload.encode("utf-8") if payload else None

    req_headers: dict = {}
    if data:
        req_headers["Content-Type"] = "application/json"
    if headers:
        req_headers.update(headers)

    req = urllib.request.Request(url, data=data, headers=req_headers, method=method_upper)

    try:
        with urllib.request.urlopen(req, timeout=5) as resp:
            status = resp.status
            body = resp.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as exc:
        status = exc.code
        body = exc.read().decode("utf-8", errors="replace")
    except urllib.error.URLError as exc:
        return ToolResult(
            pass_rate=0.0,
            logs=f"Connection failed: {exc.reason}",
            success=False,
        )
    except Exception as exc:
        return ToolResult(pass_rate=0.0, logs=f"Request error: {exc}", success=False)

    # Pretty-print JSON responses
    try:
        pretty_body = json.dumps(json.loads(body), indent=2)
    except (json.JSONDecodeError, ValueError):
        pretty_body = body

    output = f"HTTP {status} {method_upper} {path}\n\n{pretty_body}"
    logger.info("curl_endpoint: %s %s → HTTP %s", method_upper, url, status)
    return ToolResult(pass_rate=0.0, logs=output, success=True)