tracer-cloud--opensre/tests/benchmarks/cloudopsbench/scoring.py

from __future__ import annotations

import json
import re
from dataclasses import asdict, dataclass
from typing import Any

from tests.benchmarks.cloudopsbench.case_loader import CloudOpsCase
from tests.benchmarks.cloudopsbench.replay_backend import normalize_resource_type
from tests.benchmarks.cloudopsbench.taxonomy import (
    infer_fault_object as _infer_fault_object,
)
from tests.benchmarks.cloudopsbench.taxonomy import (
    taxonomy_for_root_cause as _taxonomy_for_root_cause,
)

TOOL_KEY_PARAMS: dict[str, list[str]] = {
    "GetResources": ["resource_type"],
    "DescribeResource": ["resource_type", "name"],
    "CheckNodeServiceStatus": ["node_name", "service_name"],
    "GetClusterConfiguration": [],
    "GetAlerts": [],
    "GetErrorLogs": ["service_name"],
    "GetRecentLogs": ["service_name"],
    "GetServiceDependencies": ["service_name"],
    "GetAppYAML": ["app_name"],
    "CheckServiceConnectivity": ["service_name", "port"],
}

TOOL_REQUIRED_PARAMS: dict[str, list[str]] = {
    "GetResources": ["resource_type"],
    "DescribeResource": ["resource_type", "name"],
    "CheckNodeServiceStatus": ["node_name", "service_name"],
    "GetClusterConfiguration": [],
    "GetAlerts": [],
    "GetErrorLogs": ["service_name"],
    "GetRecentLogs": ["service_name"],
    "GetServiceDependencies": ["service_name"],
    "GetAppYAML": ["app_name"],
    "CheckServiceConnectivity": ["service_name", "port"],
}


@dataclass(frozen=True)
class CloudOpsMetrics:
    a1: float
    a3: float
    partial_a1: float
    partial_a3: float
    # Localization-only accuracy: fault_object correct regardless of whether
    # the root_cause/taxonomy strings also matched. Isolates "did we find the
    # right thing" from "did we name the failure with the exact dataset token",
    # which the strict triple-match a1/partial_a1 conflate. object_a1 = rank-1
    # object correct; object_a3 = correct object anywhere in the top-3.
    object_a1: float
    object_a3: float
    # Investigation-native scoring: rebuild a single triple from opensre's
    # prose (report + root_cause + causal_chain + validated_claims) using the
    # deterministic keyword parser ``infer_final_answer_from_opensre_text``,
    # then score it the same way. This isolates opensre's investigation
    # quality from the LLM predictor that formalizes the prose into
    # ``top_3_predictions``: a lift on ``a1`` could come from a better
    # investigation OR a better predictor — ``investigation_a1`` only moves
    # when the investigation itself names the correct triple.
    #
    # ``investigation_a1`` is a CONSERVATIVE lower bound on investigation
    # quality: the keyword parser misses synonyms and freer phrasings.
    # ``translation_loss`` flips on cases where investigation_a1 is right but
    # ``a1`` is wrong — the formalization step lost what opensre found.
    # Read together they answer "is opensre getting better, or just the
    # wrapper around it?"
    investigation_a1: float
    investigation_partial_a1: float
    investigation_object_a1: float
    translation_loss: float
    tcr: float
    exact: float
    in_order: float
    any_order: float
    rel: float
    cov: float
    steps: float
    mtti: float
    iac: float
    rar: float
    ztdr: float


@dataclass(frozen=True)
class CloudOpsCaseScore:
    case_id: str
    ground_truth: dict[str, Any]
    top_3_predictions: list[dict[str, Any]]
    final_answer_source: str
    standardized_agent_steps: list[str]
    expert_steps: list[str]
    matched_path: str
    invalid_reasons: list[str]
    metrics: CloudOpsMetrics
    error: str = ""

    def to_dict(self) -> dict[str, Any]:
        payload = asdict(self)
        payload["metrics"] = asdict(self.metrics)
        return payload


def normalize_text(value: Any) -> str:
    if value is None:
        return ""
    return str(value).strip().lower()


def strip_pod_suffix(name: Any) -> str:
    if not isinstance(name, str):
        return str(name)
    patterns = [
        r"^([a-z0-9-]+)-[a-f0-9]{8,10}-[a-z0-9]{4,6}$",
        r"^([a-z0-9-]+)-[a-z0-9]{5}$",
    ]
    for pattern in patterns:
        match = re.match(pattern, name)
        if match:
            return match.group(1)
    return name


def parse_json_maybe(raw: Any) -> dict[str, Any] | None:
    if isinstance(raw, dict):
        return raw
    if raw is None:
        return None
    text = str(raw).strip()
    if not text:
        return None
    for pattern in (r"```json\s*(.*?)\s*```", r"```\s*(.*?)\s*```"):
        match = re.search(pattern, text, re.DOTALL)
        if match:
            text = match.group(1).strip()
            break
    try:
        parsed = json.loads(text)
    except json.JSONDecodeError:
        return None
    return parsed if isinstance(parsed, dict) else None


def extract_final_answer_payload(case_data: dict[str, Any]) -> tuple[dict[str, Any] | None, str]:
    candidates: list[tuple[str, Any]] = [
        ("top_level_final_answer", case_data.get("final_answer")),
        ("root_cause", case_data.get("root_cause")),
        ("report", case_data.get("report")),
    ]
    final_state = case_data.get("final_state")
    if isinstance(final_state, dict):
        candidates.extend(
            [
                ("final_state_final_answer", final_state.get("final_answer")),
                ("final_state_root_cause", final_state.get("root_cause")),
                ("final_state_report", final_state.get("report")),
            ]
        )
    for step in reversed(case_data.get("steps", [])):
        if not isinstance(step, dict):
            continue
        if step.get("final_answer"):
            candidates.append((f"step_{step.get('step_id', 'unknown')}_final_answer", step))
        if step.get("raw_model_output"):
            candidates.append((f"step_{step.get('step_id', 'unknown')}_raw_model_output", step))

    for source, candidate in candidates:
        parsed = parse_json_maybe(candidate)
        if parsed and isinstance(parsed.get("top_3_predictions"), list):
            return parsed, source

    inferred = infer_final_answer_from_opensre_text(case_data)
    if inferred is not None:
        return inferred, "inferred_from_opensre_text"
    return None, "unparsed"


def infer_final_answer_from_opensre_text(
    case_data: dict[str, Any],
    *,
    include_predictor_output: bool = True,
) -> dict[str, Any] | None:
    """Parse opensre's free-text RCA into a single-prediction paper triple.

    Set ``include_predictor_output=False`` for investigation-native scoring:
    by default this function also reads ``case_data["final_answer"]`` (the
    structured-predictor JSON, stringified) which would feed predictor
    signal back through the keyword parser and defeat the purpose of an
    "is opensre alone right?" metric. Existing callers (legacy fallback in
    ``extract_final_answer_payload``) keep the default behavior.
    """
    final_state = case_data.get("final_state")
    texts = [
        case_data.get("root_cause"),
        case_data.get("report"),
    ]
    if include_predictor_output:
        texts.append(case_data.get("final_answer"))
    if isinstance(final_state, dict):
        texts.extend(
            [
                final_state.get("root_cause"),
                final_state.get("report"),
                " ".join(str(item) for item in final_state.get("causal_chain", [])),
                " ".join(
                    str(claim.get("claim", ""))
                    for claim in final_state.get("validated_claims", [])
                    if isinstance(claim, dict)
                ),
            ]
        )
    text = " ".join(str(item or "") for item in texts).lower()
    if not text.strip():
        return None

    root_cause = _infer_root_cause(text)
    fault_object = _infer_fault_object(text)
    if not root_cause or not fault_object:
        return None

    return {
        "key_evidence_summary": "Inferred from OpenSRE RCA text for CloudOpsBench scoring.",
        "top_3_predictions": [
            {
                "rank": 1,
                "fault_taxonomy": _taxonomy_for_root_cause(root_cause),
                "fault_object": fault_object,
                "root_cause": root_cause,
            }
        ],
    }


def _infer_root_cause(text: str) -> str:
    checks = [
        (
            "service_env_var_address_mismatch",
            ("env", "environment", "address", "hostname", "redis-cart-invalid", "invalid"),
        ),
        ("service_dns_resolution_failure", ("dns", "resolution", "no such host")),
        ("service_selector_mismatch", ("selector", "endpoint", "no endpoints")),
        ("service_port_mapping_mismatch", ("port mapping", "targetport", "target port")),
        ("pod_cpu_overload", ("cpu", "overload", "saturation")),
        ("oom_killed", ("oom", "out of memory", "oomkilled")),
        ("incorrect_image_reference", ("imagepullbackoff", "image pull", "incorrect image")),
        ("missing_image_pull_secret", ("image pull secret", "pull secret")),
        ("deployment_zero_replicas", ("zero replicas", "replica count is 0")),
        ("db_connection_exhaustion", ("connection exhaustion", "too many connections")),
        ("mysql_invalid_credentials", ("mysql", "access denied", "invalid credentials")),
        ("mysql_invalid_port", ("mysql", "invalid port", "wrong port")),
        ("node_network_delay", ("node", "network delay")),
        ("node_network_packet_loss", ("packet loss",)),
        ("kubelet_unavailable", ("kubelet", "unavailable")),
        ("containerd_unavailable", ("containerd", "unavailable")),
        ("kube_proxy_unavailable", ("kube-proxy", "unavailable")),
        ("kube_scheduler_unavailable", ("kube-scheduler", "unavailable")),
    ]
    for root_cause, tokens in checks:
        if all(token in text for token in tokens):
            return root_cause
    return ""


def compare_prediction(
    prediction: dict[str, Any], ground_truth: dict[str, Any]
) -> tuple[bool, bool]:
    gt_tax = normalize_text(ground_truth.get("fault_taxonomy"))
    gt_obj = normalize_text(ground_truth.get("fault_object"))
    gt_root = normalize_text(ground_truth.get("root_cause"))
    pr_tax = normalize_text(prediction.get("fault_taxonomy"))
    pr_obj = normalize_text(prediction.get("fault_object"))
    pr_root = normalize_text(prediction.get("root_cause"))
    full_match = pr_tax == gt_tax and pr_obj == gt_obj and pr_root == gt_root
    partial_match = pr_obj == gt_obj and pr_root == gt_root
    return full_match, partial_match


def score_predictions(
    predictions: list[dict[str, Any]],
    ground_truth: dict[str, Any],
) -> dict[str, float]:
    a1 = 0.0
    a3 = 0.0
    partial_a1 = 0.0
    partial_a3 = 0.0
    object_a1 = 0.0
    object_a3 = 0.0
    gt_obj = normalize_text(ground_truth.get("fault_object"))
    for idx, prediction in enumerate(predictions[:3]):
        full_match, partial_match = compare_prediction(prediction, ground_truth)
        if full_match:
            if idx == 0:
                a1 = 1.0
            a3 = 1.0
        if partial_match:
            if idx == 0:
                partial_a1 = 1.0
            partial_a3 = 1.0
        if normalize_text(prediction.get("fault_object")) == gt_obj:
            if idx == 0:
                object_a1 = 1.0
            object_a3 = 1.0
    return {
        "a1": a1,
        "a3": a3,
        "partial_a1": partial_a1,
        "partial_a3": partial_a3,
        "object_a1": object_a1,
        "object_a3": object_a3,
    }


def standardize_tool_step(step: dict[str, Any]) -> tuple[str | None, str | None]:
    action_name = step.get("action_name")
    action_input = step.get("action_input")
    if not action_name or not isinstance(action_name, str):
        return None, "missing_action_name"
    if not isinstance(action_input, dict):
        return None, "invalid_action_input"

    required = TOOL_REQUIRED_PARAMS.get(action_name, [])
    for key in required:
        value = action_input.get(key)
        if value is None or str(value).strip() == "":
            return None, f"missing_required_param:{key}"

    if action_name == "GetResources":
        resource_type = normalize_resource_type(action_input.get("resource_type"))
        if not resource_type:
            return None, "missing_required_param:resource_type"
        return f"{action_name}::{resource_type}", None

    if action_name == "DescribeResource":
        resource_type = normalize_resource_type(action_input.get("resource_type"))
        name = action_input.get("name")
        if resource_type == "pods":
            name = strip_pod_suffix(name)
        elif name is not None:
            name = str(name).strip()
        if not resource_type or not name:
            return None, "missing_describe_resource_fields"
        return f"{action_name}::{resource_type}::{name}", None

    params = TOOL_KEY_PARAMS.get(action_name)
    if params is None:
        params = sorted(key for key in action_input if key != "namespace")

    parts = [action_name]
    for key in params:
        if key == "namespace":
            continue
        value = action_input.get(key)
        if value is None or str(value).strip() == "":
            continue
        parts.append(str(value).strip())
    if len(parts) == 1:
        parts.append("")
    return "::".join(parts), None


def standardize_agent_steps(case_data: dict[str, Any]) -> tuple[list[str], int, list[str]]:
    provided_steps = case_data.get("standardized_agent_steps")
    if isinstance(provided_steps, list) and all(isinstance(step, str) for step in provided_steps):
        return list(provided_steps), 0, []

    standardized: list[str] = []
    invalid_reasons: list[str] = []
    invalid_count = 0
    for step in case_data.get("steps", []):
        if not isinstance(step, dict) or step.get("action_type") != "tool":
            continue
        if step.get("error"):
            invalid_count += 1
            invalid_reasons.append(f"step_{step.get('step_id', 'unknown')}:error")
            continue
        standardized_step, reason = standardize_tool_step(step)
        if reason:
            invalid_count += 1
            invalid_reasons.append(f"step_{step.get('step_id', 'unknown')}:{reason}")
            continue
        if standardized_step is not None:
            standardized.append(standardized_step)
    return standardized, invalid_count, invalid_reasons


def precision_recall_f1(
    agent_steps: list[str], expert_steps: list[str]
) -> tuple[float, float, float]:
    if not agent_steps and not expert_steps:
        return 1.0, 1.0, 1.0
    if not agent_steps:
        return 0.0, 0.0, 0.0
    agent_set = set(agent_steps)
    expert_set = set(expert_steps)
    intersection = len(agent_set & expert_set)
    precision = intersection / len(agent_set) if agent_set else 0.0
    recall = intersection / len(expert_set) if expert_set else 0.0
    f1 = 2 * precision * recall / (precision + recall) if precision + recall > 0 else 0.0
    return precision, recall, f1


def in_order_match(expert_steps: list[str], agent_steps: list[str]) -> float:
    if not expert_steps:
        return 1.0
    idx = 0
    for step in agent_steps:
        if step == expert_steps[idx]:
            idx += 1
            if idx == len(expert_steps):
                return 1.0
    return 0.0


def any_order_match(expert_steps: list[str], agent_steps: list[str]) -> float:
    if not expert_steps:
        return 1.0
    return 1.0 if set(expert_steps).issubset(set(agent_steps)) else 0.0


def exact_match(expert_steps: list[str], agent_steps: list[str]) -> float:
    return 1.0 if expert_steps == agent_steps else 0.0


def choose_best_path(agent_steps: list[str], process: dict[str, list[str]]) -> dict[str, Any]:
    best: dict[str, Any] | None = None
    for path_name in ("path1", "path2"):
        expert_steps = process.get(path_name, [])
        precision, recall, f1 = precision_recall_f1(agent_steps, expert_steps)
        current = {
            "matched_path": path_name,
            "expert_steps": expert_steps,
            "rel": precision,
            "cov": recall,
            "f1": f1,
            "in_order": in_order_match(expert_steps, agent_steps),
            "exact": exact_match(expert_steps, agent_steps),
            "any_order": any_order_match(expert_steps, agent_steps),
        }
        if best is None or current["f1"] > best["f1"]:
            best = current
            continue
        if current["f1"] == best["f1"] and current["in_order"] > best["in_order"]:
            best = current
    return best or {
        "matched_path": "path1",
        "expert_steps": [],
        "rel": 0.0,
        "cov": 0.0,
        "f1": 0.0,
        "in_order": 0.0,
        "exact": 0.0,
        "any_order": 0.0,
    }


def calculate_rar(agent_steps: list[str]) -> float:
    total = len(agent_steps)
    if total == 0:
        return 0.0
    counts: dict[str, int] = {}
    for step in agent_steps:
        counts[step] = counts.get(step, 0) + 1
    redundant = sum(count - 1 for count in counts.values())
    return redundant / total


def calculate_total_latency(case_data: dict[str, Any]) -> float:
    """Mean-time-to-identify, in seconds: wall-clock from investigation start
    to the agent's diagnosis.

    The benchmark replays tool results deterministically, so per-step
    ``tool_latency`` is meaningless (~microseconds of dict lookup). The honest
    signal is the LLM-dominated wall-clock the runner already measures with a
    monotonic timer around ``run_investigation`` and stores on
    ``RunResult.latency_ms`` (the scoring-only predictor call runs *after* that
    stop-watch, so it isn't counted — exactly the paper's "time to identify").

    Priority:
      1. Real measured wall-clock — ``case_data["latency_ms"]`` (preferred).
      2. Sum of per-step ``model_latency``/``tool_latency`` — kept for any
         future per-step instrumentation and for callers that pass timed steps.

    Returns 0.0 only when neither source is present (e.g. a hand-built
    ``case_data`` in a unit test), so a missing measurement is visibly 0
    rather than a silently fabricated number.
    """
    latency_ms = case_data.get("latency_ms")
    if isinstance(latency_ms, (int, float)) and latency_ms > 0:
        return float(latency_ms) / 1000.0

    total = 0.0
    for step in case_data.get("steps", []):
        if not isinstance(step, dict):
            continue
        for key in ("model_latency", "tool_latency"):
            value = step.get(key)
            if isinstance(value, (int, float)):
                total += float(value)
    return total


def _score_investigation_native(
    case_data: dict[str, Any],
    ground_truth: dict[str, Any],
) -> dict[str, float]:
    """Score opensre's investigation prose directly, bypassing the predictor.

    Builds a single-prediction triple from opensre's text via the
    deterministic keyword parser ``infer_final_answer_from_opensre_text``,
    then runs it through ``score_predictions`` against ground truth. Returns
    the same key shape as ``score_predictions`` so the call site can mirror
    its handling. Returns all zeros when the parser cannot extract a triple
    (empty text or unmatched root_cause / fault_object), which is the honest
    floor — we have no evidence the investigation named the right answer.
    """
    inferred = infer_final_answer_from_opensre_text(case_data, include_predictor_output=False)
    if not inferred:
        return {"a1": 0.0, "partial_a1": 0.0, "object_a1": 0.0}
    predictions = inferred.get("top_3_predictions", []) or []
    if not isinstance(predictions, list) or not predictions:
        return {"a1": 0.0, "partial_a1": 0.0, "object_a1": 0.0}
    scored = score_predictions(predictions, ground_truth)
    return {
        "a1": scored["a1"],
        "partial_a1": scored["partial_a1"],
        "object_a1": scored["object_a1"],
    }


def score_case(case: CloudOpsCase, case_data: dict[str, Any]) -> CloudOpsCaseScore:
    ground_truth = {
        "fault_taxonomy": case.result.fault_taxonomy,
        "fault_object": case.result.fault_object,
        "root_cause": case.result.root_cause,
    }
    parsed_final_answer, final_answer_source = extract_final_answer_payload(case_data)
    predictions = parsed_final_answer.get("top_3_predictions", []) if parsed_final_answer else []
    if not isinstance(predictions, list):
        predictions = []
    predictions = [prediction for prediction in predictions if isinstance(prediction, dict)]
    outcome_scores = (
        score_predictions(predictions, ground_truth)
        if predictions
        else {
            "a1": 0.0,
            "a3": 0.0,
            "partial_a1": 0.0,
            "partial_a3": 0.0,
            "object_a1": 0.0,
            "object_a3": 0.0,
        }
    )

    investigation_scores = _score_investigation_native(case_data, ground_truth)
    translation_loss = (
        1.0 if investigation_scores["a1"] >= 1.0 and outcome_scores["a1"] < 1.0 else 0.0
    )

    agent_steps, invalid_count, invalid_reasons = standardize_agent_steps(case_data)
    best_path = choose_best_path(agent_steps, case.process)
    steps = len(agent_steps)
    ztdr = 1.0 if steps == 0 and predictions else 0.0
    metrics = CloudOpsMetrics(
        a1=outcome_scores["a1"],
        a3=outcome_scores["a3"],
        partial_a1=outcome_scores["partial_a1"],
        partial_a3=outcome_scores["partial_a3"],
        object_a1=outcome_scores["object_a1"],
        object_a3=outcome_scores["object_a3"],
        investigation_a1=investigation_scores["a1"],
        investigation_partial_a1=investigation_scores["partial_a1"],
        investigation_object_a1=investigation_scores["object_a1"],
        translation_loss=translation_loss,
        tcr=1.0 if predictions else 0.0,
        exact=best_path["exact"],
        in_order=best_path["in_order"],
        any_order=best_path["any_order"],
        rel=best_path["rel"],
        cov=best_path["cov"],
        steps=float(steps),
        mtti=calculate_total_latency(case_data),
        iac=float(invalid_count),
        rar=calculate_rar(agent_steps),
        ztdr=ztdr,
    )

    return CloudOpsCaseScore(
        case_id=case.case_id,
        ground_truth=ground_truth,
        top_3_predictions=predictions,
        final_answer_source=final_answer_source,
        standardized_agent_steps=agent_steps,
        expert_steps=list(best_path["expert_steps"]),
        matched_path=str(best_path["matched_path"]),
        invalid_reasons=invalid_reasons,
        metrics=metrics,
        error="" if parsed_final_answer else "unparsed_final_answer",
    )


def summarize_scores(scores: list[CloudOpsCaseScore]) -> dict[str, Any]:
    metric_names = [
        "a1",
        "a3",
        "partial_a1",
        "partial_a3",
        "object_a1",
        "object_a3",
        "investigation_a1",
        "investigation_partial_a1",
        "investigation_object_a1",
        "translation_loss",
        "tcr",
        "exact",
        "in_order",
        "any_order",
        "rel",
        "cov",
        "steps",
        "mtti",
        "iac",
        "rar",
        "ztdr",
    ]
    total = len(scores)
    sums = dict.fromkeys(metric_names, 0.0)
    parse_failures = 0
    for score in scores:
        if score.error == "unparsed_final_answer":
            parse_failures += 1
        metrics = asdict(score.metrics)
        for name in metric_names:
            sums[name] += float(metrics[name])

    averages = {name: round(sums[name] / total, 4) if total else 0.0 for name in metric_names}
    return {
        "counts": {
            "total_cases": total,
            "final_answer_parse_failures": parse_failures,
        },
        "metrics": {
            "Accuracy @1": averages["a1"],
            "Accuracy @3": averages["a3"],
            "Partial Accuracy @1": averages["partial_a1"],
            "Partial Accuracy @3": averages["partial_a3"],
            "Object Accuracy @1": averages["object_a1"],
            "Object Accuracy @3": averages["object_a3"],
            "Investigation Accuracy @1": averages["investigation_a1"],
            "Investigation Partial Accuracy @1": averages["investigation_partial_a1"],
            "Investigation Object Accuracy @1": averages["investigation_object_a1"],
            "Translation Loss Rate": averages["translation_loss"],
            "Task Completion Rate": averages["tcr"],
            "ExactMatch": averages["exact"],
            "InOrder": averages["in_order"],
            "AnyOrder": averages["any_order"],
            "Relevant": averages["rel"],
            "Coverage": averages["cov"],
            "Steps": round(averages["steps"], 2),
            "Mean Time to Identify": round(averages["mtti"], 4),
            "Invalid Action Count": round(averages["iac"], 2),
            "Redundant Action Rate": averages["rar"],
            "Zero-Tool Direct Resolution": averages["ztdr"],
        },
    }