# domain/validator.py - V7.4 (DOMAIN-AWARE CONSISTENCY GATE) """ V7.4 ConsistencyGate: Domain-Aware Validation. Validates by StepType — NOT by string shape. ALGEBRAIC → sympy.sympify() (existing behaviour, unchanged) GEOMETRY → structural validation of payload (list[str]) NUMERIC → {future expansion point} CTO directive: - Geometry steps must NEVER reach sympify (Category Error prevention) - Geometry gets its own structural validator, not a "skip blind" - expression is display-only — validation uses payload for geometry """ import sympy import logging from typing import List, Tuple from domain.step_types import StepType, SignedStep logger = logging.getLogger(__name__) class ConsistencyGate: """ V7.4 Wall 3: Post-Execution Domain-Aware Consistency Gate. Receives the list of SignedStep objects from the Deterministic Solver and verifies structural and logical integrity before passing to the Renderer. Checks performed (all domains): 1. Non-empty: at least one signed step exists. 2. Hash integrity: every step has a non-empty SHA256 hash. 3a. ALGEBRAIC: expression is SymPy-parseable (sympy.sympify). 3b. GEOMETRY: payload is a non-empty list[str] (structural validation). """ @staticmethod def _validate_algebraic(step: SignedStep) -> Tuple[bool, str]: """ Validates an ALGEBRAIC step by attempting sympy.sympify on each sub-expression (split on " OR " for multi-solution results). """ expr_str = step.expression if not expr_str: logger.error(f"[CONSISTENCY_GATE] Step '{step.id}' has an empty expression.") return False, f"EMPTY_EXPRESSION:{step.id}" parts = [p.strip() for p in expr_str.split(" OR ")] for part in parts: try: sympy.sympify(part) except Exception as e: logger.error( f"[CONSISTENCY_GATE] Step '{step.id}' expression not parseable: " f"'{part}' — {e}" ) return False, f"UNPARSEABLE_EXPRESSION:{step.id}" return True, "" @staticmethod def _validate_geometry(step: SignedStep) -> Tuple[bool, str]: """ Validates a GEOMETRY step structurally via payload. Geometry output (points, distances, labels) is NOT SymPy-parseable. We verify: - payload is a non-empty list - every element is a string Hash integrity (checked earlier) is the cryptographic guarantee. """ payload = step.payload if not isinstance(payload, list) or len(payload) == 0: logger.error( f"[CONSISTENCY_GATE] Geometry step '{step.id}' has empty or non-list payload." ) return False, f"GEOMETRY_EMPTY_PAYLOAD:{step.id}" if not all(isinstance(p, str) for p in payload): logger.error( f"[CONSISTENCY_GATE] Geometry step '{step.id}' payload contains non-string items." ) return False, f"GEOMETRY_INVALID_PAYLOAD_TYPE:{step.id}" logger.info( f"[CONSISTENCY_GATE] Geometry step '{step.id}' passed structural validation " f"({len(payload)} point(s))." ) return True, "" @staticmethod def validate( signed_steps: List, ast_registry: dict, problem_id: str ) -> Tuple[bool, str]: """ Returns (True, "") if all checks pass. Returns (False, reason) if any check fails → caller must Fail Closed. Accepts both SignedStep objects and legacy dicts for backwards compatibility during any partial migration. """ # Check 1: Non-empty output if not signed_steps: logger.error("[CONSISTENCY_GATE] No signed steps produced by solver.") return False, "EMPTY_SOLVER_OUTPUT" for step in signed_steps: step_id = step.get("id", "?") if hasattr(step, "get") else step.get("id", "?") # Check 2: Hash presence h = step.get("hash", "") if hasattr(step, "get") else step.get("hash", "") if not h or len(h) < 10: logger.error(f"[CONSISTENCY_GATE] Step '{step_id}' is missing a valid SHA256 hash.") return False, f"MISSING_HASH:{step_id}" # Check 3: Domain-aware expression / payload validation step_type = step.get("step_type") if hasattr(step, "get") else None if step_type == StepType.GEOMETRY: ok, reason = ConsistencyGate._validate_geometry(step) if not ok: return False, reason elif step_type == StepType.ALGEBRAIC or step_type is None: # None = legacy dict without step_type → treat as ALGEBRAIC (backwards compat) expr_str = step.get("expression", "") if not expr_str: logger.error(f"[CONSISTENCY_GATE] Step '{step_id}' has an empty expression.") return False, f"EMPTY_EXPRESSION:{step_id}" parts = [p.strip() for p in expr_str.split(" OR ")] for part in parts: try: sympy.sympify(part) except Exception as e: logger.error( f"[CONSISTENCY_GATE] Step '{step_id}' expression not parseable: " f"'{part}' — {e}" ) return False, f"UNPARSEABLE_EXPRESSION:{step_id}" # StepType.NUMERIC → future expansion point # Other unknown types → pass through (hash check is the integrity guarantee) logger.info( f"[CONSISTENCY_GATE] ✅ {len(signed_steps)} signed steps passed all checks " f"for problem '{problem_id}'." ) return True, ""