# domain/telemetry.py - V7.2 Observability Layer """ BuddyMath V7.2 Runtime Telemetry Emits structured log-based metrics compatible with Datadog/Grafana log pipelines. All metrics are emitted as structured JSON log lines on the logger named "buddymath.metrics". DevOps should configure a log-based metric parser for lines containing "METRIC_EVENT". Metric naming convention: . """ import logging import json import time from datetime import datetime, timezone from typing import Optional metrics_logger = logging.getLogger("buddymath.metrics") # ==================== METRIC KEY CONSTANTS ==================== # Use these constants everywhere to prevent typos and enable grep-ability. class M: """V7.2 Metric Key Registry""" # Core runtime outcome (feeds the Pie Chart) RUNTIME_OUTCOME = "runtime.outcome" # values: success | hint_mode | planner_retry | leakage_fail | placeholder_fail | crs_block # Fail Closed Rate (derived from RUNTIME_OUTCOME != success) FAIL_CLOSED = "fail_closed" # emitted on any non-success outcome # Planner (LLM #1) PLANNER_RETRY = "planner.retry.count" # emitted on each retry attempt PLANNER_ENUM_VIOLATION = "planner.enum_violation" # emitted when ComputeAction enum is violated PLANNER_JSON_ERROR = "planner.json_error" # emitted on JSON parse / boundary failure PLANNER_STRATEGY_DIST = "planner.strategy_distribution" # which Enum actions are chosen (drift detection) # Renderer (LLM #2) RENDERER_LEAKAGE_FAIL = "renderer.leakage_fail" # Whitelist scan failed RENDERER_PLACEHOLDER_FAIL = "renderer.placeholder_violation" # Missing or invented placeholder RENDERER_LEAKAGE_CHARS = "renderer.leakage_chars" # Offending chars (for slicing) # Solver / Server Determinism SOLVER_EXECUTION_MS = "solver.execution_time_ms" # Math engine wall-clock time SIGNATURE_HASH_COLLISION = "signature.hash_collision" # MUST always be 0 # CRS Pre-Flight CRS_BLOCK = "crs.preflight_block" # Emitted when CRS > 0.7 blocks LLM call CRS_VALUE = "crs.value" # Raw CRS score for histogram # Security SUSPICIOUS_INPUT = "suspicious.input_pattern" # Potential prompt injection detected # Pedagogical Diversity (V7.2.5) PEDAGOGICAL_DRIFT = "pedagogical.drift_score" # Narrative laziness: > 0.35 = same template always chosen # ==================== EMITTER ==================== def emit(metric: str, value, tags: Optional[dict] = None): """ Emit a single metric event as a structured JSON log line. Datadog/Grafana log-based metric pipelines should parse lines with 'METRIC_EVENT'. Format: {"event": "METRIC_EVENT", "metric": "", "value": , "tags": {...}, "timestamp": "..."} """ payload = { "event": "METRIC_EVENT", "metric": metric, "value": value, "tags": tags or {}, "timestamp": datetime.now(timezone.utc).isoformat() } metrics_logger.info(json.dumps(payload, ensure_ascii=False)) def emit_runtime_outcome(outcome: str, problem_id: str = "unknown", grade: str = "unknown"): """Feeds the main Pie Chart and Fail Closed Rate gauge.""" emit(M.RUNTIME_OUTCOME, outcome, {"problem_id": problem_id, "grade": grade}) if outcome != "success": emit(M.FAIL_CLOSED, 1, {"reason": outcome}) def emit_planner_retry(attempt: int, reason: str): emit(M.PLANNER_RETRY, attempt, {"reason": reason}) def emit_planner_error(error_type: str, details: str = ""): """error_type: 'enum_violation' | 'json_error'""" key = M.PLANNER_ENUM_VIOLATION if error_type == "enum_violation" else M.PLANNER_JSON_ERROR emit(key, 1, {"details": details[:120]}) def emit_planner_strategy_distribution(actions: list): """ Phase 1 Live: Emit one metric event per chosen Enum action. Feeds the planner.strategy_distribution dashboard panel. A sudden shift in distribution signals Model Drift. Example: [SOLVE_EQUATION, SIMPLIFY, SOLVE_EQUATION] → 3 events """ for action in actions: emit(M.PLANNER_STRATEGY_DIST, 1, {"action": str(action)}) def emit_renderer_leakage(offending_chars: str): emit(M.RENDERER_LEAKAGE_FAIL, 1, {"offending_chars": offending_chars[:50]}) emit(M.RENDERER_LEAKAGE_CHARS, offending_chars[:50]) def emit_renderer_placeholder_violation(violation_type: str, missing_id: str = ""): """violation_type: 'missing' | 'invented'""" emit(M.RENDERER_PLACEHOLDER_FAIL, 1, {"type": violation_type, "id": missing_id}) def emit_solver_timing(start_time: float): """Call with time.time() snapshot taken BEFORE solver runs.""" elapsed_ms = round((time.time() - start_time) * 1000, 2) emit(M.SOLVER_EXECUTION_MS, elapsed_ms) return elapsed_ms def emit_hash_collision(step_id: str, problem_id: str): """ CRITICAL: This MUST never be emitted in a correct system. If it fires, it means two different expressions produced the same hash → P0 alert. """ metrics_logger.critical( json.dumps({ "event": "METRIC_EVENT", "metric": M.SIGNATURE_HASH_COLLISION, "value": 1, "tags": {"step_id": step_id, "problem_id": problem_id}, "timestamp": datetime.now(timezone.utc).isoformat(), "severity": "P0_CRITICAL" }) ) def emit_crs_block(crs_value: float, problem_id: str = "unknown"): emit(M.CRS_BLOCK, 1, {"crs": crs_value, "problem_id": problem_id}) emit(M.CRS_VALUE, crs_value) def emit_crs_value(crs_value: float): emit(M.CRS_VALUE, crs_value) def emit_suspicious_input(pattern: str, problem_id: str = "unknown"): emit(M.SUSPICIOUS_INPUT, 1, {"pattern": pattern[:80], "problem_id": problem_id}) def emit_pedagogical_drift(concept_tag: str, drift_score: float): """ V7.2.5: Emitted when DiversityEngine detects narrative laziness. drift_score > 0.35 means one template variant is dominating selection. Alert DevOps: Renderer is repeating the same pedagogical phrasing — student experience degrades. """ emit(M.PEDAGOGICAL_DRIFT, round(drift_score, 3), {"concept": concept_tag}) # ==================== TIMER CONTEXT MANAGER ==================== class SolverTimer: """ Context manager for measuring Math Engine execution time. Usage: with SolverTimer() as t: result = sympy_solve(...) print(t.elapsed_ms) """ def __enter__(self): self._start = time.time() return self def __exit__(self, *args): self.elapsed_ms = round((time.time() - self._start) * 1000, 2) emit(M.SOLVER_EXECUTION_MS, self.elapsed_ms)