benchmarks/shadow_telemetry.py

"""Fail-open shadow telemetry for benchmark output validators.

Shadow mode records validator outcomes only. It never changes the answer, retry
budget, provider selection, or benchmark score. Raw model output is deliberately
not persisted; only length and normalized validator evidence are stored.
"""

from __future__ import annotations

from datetime import datetime, timezone
import json
import os
from pathlib import Path
import threading
from typing import Any, Mapping, Optional

from .validators import ValidationResult, validate_coding_output, validate_mmlu_output


_ENABLED_VALUES = frozenset({"1", "true", "yes", "on"})
_WRITE_LOCK = threading.Lock()


def shadow_enabled() -> bool:
    return os.getenv("BENCHMARK_SHADOW_MODE", "0").strip().lower() in _ENABLED_VALUES


def infer_benchmark_category(goal: Any) -> Optional[str]:
    """Infer only the two supported benchmark categories from explicit markers."""

    text = str(goal or "")
    lowered = text.lower()
    if "mmlu" in lowered or "scelta multipla" in lowered or "a/b/c/d" in lowered:
        return "mmlu"
    if "code_correct" in lowered or "typescript" in lowered or "```typescript" in lowered:
        return "coding"
    return None


def _safe_metadata(metadata: Optional[Mapping[str, Any]]) -> dict[str, Any]:
    allowed = {
        "provider",
        "model",
        "profile",
        "attempt",
        "latency_ms",
        "first_token_ms",
        "task_id",
        "source",
    }
    safe: dict[str, Any] = {}
    for key in allowed:
        value = (metadata or {}).get(key)
        if value is None:
            continue
        if isinstance(value, (str, int, float, bool)):
            safe[key] = value
        else:
            safe[key] = str(value)[:120]
    return safe


def _evidence_for_log(result: ValidationResult) -> dict[str, Any]:
    evidence: dict[str, Any] = {}
    for key, value in result.evidence.items():
        if key == "source_length":
            evidence[key] = value
        elif key in {"candidates", "distinct_candidates", "required_symbols", "missing_symbols", "declarations", "fence_count", "languages", "extraction", "significant_lines", "correct", "expected", "has_import_or_export", "has_syntax_tokens"}:
            evidence[key] = value
    return evidence


def _log_path() -> Path:
    return Path(os.getenv("BENCHMARK_SHADOW_LOG_PATH", "/tmp/baida98-benchmark-shadow.jsonl"))


def _append_event(event: dict[str, Any]) -> None:
    path = _log_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    with _WRITE_LOCK:
        with path.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n")


def validate_and_record_shadow(
    *,
    goal: Any,
    answer: Any,
    metadata: Optional[Mapping[str, Any]] = None,
) -> Optional[ValidationResult]:
    """Validate and record a supported benchmark response in fail-open shadow mode."""

    if not shadow_enabled():
        return None

    category = infer_benchmark_category(goal)
    if category is None:
        return None

    if category == "mmlu":
        result = validate_mmlu_output(answer)
        validator = "mmlu_v1"
    else:
        result = validate_coding_output(answer)
        validator = "coding_v1"

    text = answer if isinstance(answer, str) else str(answer or "")
    event = {
        "schema_version": 1,
        "event": "benchmark_shadow_validation",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "category": category,
        "validator": validator,
        "valid": result.valid,
        "failure_code": result.failure_code,
        "response_chars": len(text),
        "evidence": _evidence_for_log(result),
        "metadata": _safe_metadata(metadata),
    }
    try:
        _append_event(event)
    except Exception:
        # Shadow telemetry must never break the agent loop.
        return result
    return result