Spaces:
Running
Running
| """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 | |