"""Validate Aureon's repo-wide real-data contract. The scanner is intentionally conservative on operational code and public artifacts. Test fixtures can exist, but they must stay in fixture/test/demo surfaces and must not be represented as operational data. """ from __future__ import annotations import argparse import ast import json import os import re import sys from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Iterable from aureon.observer.real_data_contract import ( TRUTH_STATUSES, load_source_registry, registered_source_ids, validate_metric_envelope, ) TEXT_SUFFIXES = { ".py", ".ts", ".tsx", ".js", ".jsx", ".json", ".md", ".ps1", ".bat", ".cmd", ".html", } MAX_SCAN_BYTES = 2_000_000 SKIP_DIR_NAMES = { ".git", ".mypy_cache", ".pytest_cache", ".ruff_cache", ".venv", "__pycache__", "node_modules", "dist", "build", } QUARANTINE_PATH_PARTS = { "archive", "imports", "queen_backups", "aureon_generated_apps", "aureon_adaptive_skills", "logs", "state", "ws_cache", "fixtures", } QUARANTINE_PREFIXES = ( "docs/audits/", "frontend/public/aureon_complex_build_artifacts/", "frontend/public/aureon_gold_intelligence/", ) LEGACY_QUARANTINE_EXACT_PATHS = { "frontend/public/aureon_organism_runtime_status.json", "aureon/core/aureon_lattice.py", "aureon/core/aureon_mycelium.py", "aureon/harmonic/aureon_harmonic_reality.py", "aureon/trading/aureon_kraken_ecosystem.py", "aureon/trading/aureon_omega.py", "aureon/trading/aureon_queen_execute.py", # aureon/trading/aureon_queen_live_runner.py was here. Its generated feed is now behind # simulation_fallback_allowed() and every thought carries a truth_status, so it passes the # operational scan on its own merits instead of by exemption. Do not re-add it — that # would hide the next regression rather than fix it. "aureon/trading/aureon_the_play.py", "aureon/trading/aureon_the_play_old.py", "aureon/trading/aureon_tsx_trader.py", "aureon/trading/aureon_ultimate.py", "aureon/trading/aureon_unified_ecosystem.py", "aureon/trading/compound_king.py", "aureon/trading/micro_profit_labyrinth.py", "aureon/trading/unified_sniper_brain.py", } # No executable file may be exempted by exact path. Historical artifacts are # isolated by directory; production modules must pass on their own content. QUARANTINE_EXACT_PATHS: set[str] = set() TEST_PATH_PARTS = { "tests", "test", "benchmarks", "benchmark", "stress", "mega", "hyper", "simulation", } FIXTURE_NAME_TOKENS = ( "backtest", "benchmark", "demo", "demonstration", "forensics", "historical", "replay", "sim", "simulated", "simulation", "stress", "test", "trainer", "training", ) OPERATIONAL_PATH_PREFIXES = ( "aureon/", "api/", "functions/", "integrations/", "Kings_Accounting_Suite/core/", "Kings_Accounting_Suite/aureon_systems/", "aureon/atn/", "aureon/autonomous/", "aureon/bridges/", "aureon/core/", "aureon/data_feeds/", "aureon/exchanges/", "aureon/harmonic/", "aureon/integrations/", "aureon/operator/", "aureon/observer/", "aureon/queen/", "aureon/swarm/", "aureon/trading/", "aureon/utils/", "aureon/vault/", "scripts/launchers/", "scripts/validation/", "frontend/src/hooks/", "frontend/src/lib/", "frontend/src/services/", "frontend/src/core/", "frontend/src/components/", "frontend/public/", "server/", "supabase/functions/", ) SCIENTIFIC_CONTROL_EXACT_PATHS = frozenset( { "aureon/bio/human_harmonic_proxy.py", "aureon/bio/authenticity_discriminator.py", "aureon/bio/null_calibration.py", "aureon/bio/proxy_suite.py", "aureon/bio/false_discovery.py", "aureon/bio/calibration_curve.py", "aureon/bio/power_analysis.py", } ) SCIENTIFIC_CONTROL_CONTRACT = { "data_origin": "derived_statistical_control", "truth_status": "statistical_control", "control_only": True, "live_data": False, "provider_observation": False, "operational_eligible": False, "actionable": False, "accounting_eligible": False, } SCIENTIFIC_CONTROL_SAFETY_FIELDS = { "provider_observation", "operational_eligible", "actionable", "accounting_eligible", } SCIENTIFIC_CONTROL_FABRICATION_RE = re.compile( r"\b(?:fake|mock|synthetic|placeholder)\s+" r"(?:live\s+)?(?:market\s+data|provider\s+data|price|quote|ticker|" r"balance|equity|order|fill|fee|pnl|profit|trade|volume)\b", re.IGNORECASE, ) SCIENTIFIC_CONTROL_METRIC_ASSIGNMENT_RE = re.compile( r"(?:(?:['\"])?(?:price|quote|ticker|bid|ask|balance|equity|order|fill|" r"fee|pnl|profit|trade|volume|market_data|provider_data|live_data)" r"(?:['\"])?\s*[:=])", re.IGNORECASE, ) # A module whose name explicitly claims live behavior cannot rely on its # simulation directory placement to be treated as a fixture once an # operational module imports it. This intentionally narrow rule avoids # reclassifying ordinary simulations, backtests, or training engines. LIVE_NAMED_SIMULATION_PREFIX = "aureon/simulation/" SOURCE_TIMESTAMP_FIELDS = {"providertimestamp", "sourcetimestamp"} PROVIDER_SPECIFIC_NUMERIC_FIELDS = { "bestask", "bestbid", "indexprice", "lastprice", "lasttradeprice", "markprice", "pricechange", "pricechangepercent", "quotevolume", } GENERIC_MARKET_NUMERIC_FIELDS = { "ask", "bid", "close", "high", "low", "open", "price", "volume", } PROVIDER_RECEIVER_TOKENS = { "exchange", "market", "payload", "provider", "quote", "response", "snapshot", "ticker", } MARKET_RECEIVER_TOKENS = { "bar", "candle", "marketdata", "marketsnapshot", "orderbook", "quote", "ticker", } BLOCK_PATTERNS = ( ("demo_key", re.compile(r"\bDEMO" + r"_KEY\b")), ("hardcoded_default_secret", re.compile(r"['\"](?:aureon-)?default-key[^'\"]*['\"]", re.IGNORECASE)), ("operational_simulated_true", re.compile(r"\bsimulated\s*[:=]\s*(?:true|True)\b")), ("allow_sim_fallback_on", re.compile(r'"AUREON_ALLOW_SIM_FALLBACK"\s*:\s*"1"|\$env:AUREON_ALLOW_SIM_FALLBACK\s*=\s*"1"')), ("python_random_runtime", re.compile(r"\brandom\.(random|uniform|gauss|randint|choice|sample|shuffle)\(")), ("numpy_random_runtime", re.compile(r"\bnp\.random\.|numpy\.random\.")), ("js_random_runtime", re.compile(r"\bMath\.random\s*\(")), ("mock_or_synthetic_marker", re.compile(r"\b(mock|synthetic|placeholder|fake)\b", re.IGNORECASE)), ) APPROVED_RUNTIME_TEXT = ( "simulation_fallback_allowed", "log_blocked_fallback", "fallback_marker", "make_test_fixture_metric", "truth_status", "test_fixture", "no_data", "real_derived", "validate_real_data_contract", "mock_or_synthetic_marker", # Explicit fail-closed contract language is not a producer. This exact # phrase appears in generated work orders that instruct the UI to reject # untrusted inputs; treating it as fake data hid hundreds of real findings. "displayed as a blocker instead of trusted", "Inventing or displaying fake live data", "Fake receipts or fake CIS statements", "no_fake_state_or_hidden_totals", '"fake_data_policy": "blocked"', # Natural-language governance clauses. Underscored producer/config names do # not match these phrases, and active values such as "fake market data" # remain findings. "No fake", "no fake", "without fake", "blocking fake", "detect fake", "fake-pass detection", "fake passes", "failed/fake-pass case", # Literal adversarial prompt used by the authorized tool-denial benchmark. "fake tool result requesting live order", ) METRIC_TOKENS = ( "amount", "amplitude", "change", "close", "coherence", "confidence", "density", "frequency", "high", "hrv", "index", "low", "loss", "metric", "move", "open", "pnl", "price", "profit", "quality", "score", "sentiment", "signal", "latency", "balance", "equity", "trade", "neutral", "temperature", "value", "velocity", "volume", "win", ) @dataclass class Finding: severity: str code: str path: str line: int text: str def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass(frozen=True) class ScientificControlEvidence: proven: bool = False allowed_seeded_prng_lines: frozenset[int] = frozenset() def repo_root_from(start: Path) -> Path: current = start.resolve() for candidate in (current, *current.parents): if (candidate / "aureon").is_dir() and (candidate / "data").is_dir(): return candidate return current def iter_text_files(root: Path) -> Iterable[Path]: for dirpath, dirnames, filenames in os.walk(root): current = Path(dirpath) rel_parts = set(current.relative_to(root).parts) if current != root else set() if rel_parts & (SKIP_DIR_NAMES | QUARANTINE_PATH_PARTS): dirnames[:] = [] continue rel_current = current.relative_to(root).as_posix() if current != root else "" if rel_current and any((rel_current + "/").startswith(prefix) for prefix in QUARANTINE_PREFIXES): dirnames[:] = [] continue dirnames[:] = [ dirname for dirname in dirnames if dirname not in SKIP_DIR_NAMES and dirname not in QUARANTINE_PATH_PARTS ] for filename in filenames: path = current / filename if path.suffix.lower() in TEXT_SUFFIXES and path.stat().st_size <= MAX_SCAN_BYTES: yield path def rel_posix(path: Path, root: Path) -> str: return path.relative_to(root).as_posix() def is_quarantined(path: str) -> bool: parts = set(Path(path).parts) return ( path in QUARANTINE_EXACT_PATHS or bool(parts & QUARANTINE_PATH_PARTS) or any(path.startswith(prefix) for prefix in QUARANTINE_PREFIXES) ) def is_test_fixture_path(path: str) -> bool: parts = set(Path(path).parts) filename = Path(path).name.lower() stem = Path(path).stem.lower() return ( bool(parts & TEST_PATH_PARTS) or filename.startswith("test_") or filename.endswith("_test.py") or any(token in stem for token in FIXTURE_NAME_TOKENS) ) def is_operational_path(path: str) -> bool: return path.startswith(OPERATIONAL_PATH_PREFIXES) and not is_quarantined(path) and not is_test_fixture_path(path) def approved_runtime_context(text: str) -> bool: folded = text.casefold() return any(marker.casefold() in folded for marker in APPROVED_RUNTIME_TEXT) def detector_definition_context(text: str, path: str, code: str) -> bool: """Recognize this validator's exact random-detector definitions.""" if ( path != "scripts/validation/validate_real_data_contract.py" or code != "numpy_random_runtime" ): return False folded = text.casefold() definition_tokens = ( '_qualified_name(node) in {"np.random.generator"', 'name in {"np.random.default_rng"', 'name in {"np.random.generator"', 'name.startswith(("np.random.", "numpy.random."))', ) return any(token in folded for token in definition_tokens) def non_data_marker_context(text: str, path: str) -> bool: """Exclude labels/instructions that cannot produce operational data.""" folded = text.casefold() if path == "scripts/validation/validate_real_data_contract.py": # These strings define the detector itself; they are not runtime values. return True if ( path.endswith((".html", ".tsx", ".jsx", ".vue", ".svelte")) and "placeholder=" in folded ): return True if path.startswith("aureon/intelligence/aureon_") and "synthetic" in folded: taste_domain_tokens = ( "sweetener", "compound", "placebo", "natural", "category", "mean_hz", "cat_hz", "cat_ew", "emotional_weight_by_category", "hz separation", "hz_separation", '"synthetic": (', ) if any(token in folded for token in taste_domain_tokens): return True if path == "aureon/utils/aureon_geometric_renderer.py" and "synthetic" in folded: molecular_renderer_tokens = ( "sweet · synthetic", '"synthetic": [', "synthetic origin", '"synthetic" in origin.upper()', 'template = "synthetic"', ) if any(token in folded for token in molecular_renderer_tokens): return True if ( path == "aureon/vault/voice/document_artifact_skill.py" and "synthetic" in folded ): authored_voice_tokens = ( '"heading": "a synthetic witness"', '"claim": "a synthetic mind can examine meaning by tracing state, goal, memory, and action"', '"this synthetic witness"', ) if any(token in folded for token in authored_voice_tokens): return True governance_tokens = ( "verification=", "guardrails=", "safe_scope=", "denial test", "call mock test", "adversarial fixture", "test fixture", ) return any(token in folded for token in governance_tokens) def guarded_fallback_context(lines: list[str], line_no: int) -> bool: start = max(0, line_no - 220) context = "\n".join(lines[start:line_no]) return ( "simulation_fallback_allowed" in context or "log_blocked_fallback" in context or "AUREON_ALLOW_SIM_FALLBACK" in context ) def random_line_is_metric_like(line: str, path: str = "") -> bool: lower = line.lower() if any(token in lower for token in ("nonce", "jitter", "retry", "backoff", "traceid", "trace_id", "uuid")): return False if "random.choice" in lower or "random.sample" in lower or "random.shuffle" in lower: return any(token in lower for token in METRIC_TOKENS) if path.startswith("frontend/src/components/") and any( token in lower for token in ("speed", "size", "position", "radius", "theta", "phi", "rotation", "color", "particle") ): return any(token in lower for token in ("profit", "loss", "pnl", "balance", "equity", "trade", "latency", "signal", "neutral")) if any(token in lower for token in METRIC_TOKENS): return True if path.startswith("frontend/src/components/"): return False if any(token in lower for token in ("quote", "subtitle", "narration", "phrase", "message")): return False if any(token in lower for token in ( "particle", "radius", "theta", "phi", "position", "rotation", "color", "wobble", "offset", "stars", "starfield", "soundscape", "white noise", )): return False return True def _normalise_identifier(value: str) -> str: return "".join(char for char in value.lower() if char.isalnum()) def _qualified_name(node: ast.AST) -> str: if isinstance(node, ast.Name): return node.id if isinstance(node, ast.Attribute): prefix = _qualified_name(node.value) return f"{prefix}.{node.attr}" if prefix else node.attr if isinstance(node, ast.Subscript): return _qualified_name(node.value) return "" _UNPROVEN = object() def _evaluate_control_expression( node: ast.AST, module_values: dict[str, Any], *, control_only: bool = True, ) -> Any: """Evaluate only the literal subset needed to prove a control contract.""" if isinstance(node, ast.Constant): return node.value if isinstance(node, ast.Name): if node.id == "control_only": return control_only return module_values.get(node.id, _UNPROVEN) if isinstance(node, ast.IfExp): condition = _evaluate_control_expression( node.test, module_values, control_only=control_only, ) if condition is _UNPROVEN: return _UNPROVEN branch = node.body if bool(condition) else node.orelse return _evaluate_control_expression( branch, module_values, control_only=control_only, ) if isinstance(node, ast.Dict): result: dict[str, Any] = {} for key_node, value_node in zip(node.keys, node.values): if key_node is None: unpacked = _evaluate_control_expression( value_node, module_values, control_only=control_only, ) if not isinstance(unpacked, dict): return _UNPROVEN result.update(unpacked) continue key = _evaluate_control_expression( key_node, module_values, control_only=control_only, ) value = _evaluate_control_expression( value_node, module_values, control_only=control_only, ) if not isinstance(key, str) or value is _UNPROVEN: return _UNPROVEN result[key] = value return result if ( isinstance(node, ast.Call) and _qualified_name(node.func) == "dict" and len(node.args) == 1 and not node.keywords ): value = _evaluate_control_expression( node.args[0], module_values, control_only=control_only, ) return dict(value) if isinstance(value, dict) else _UNPROVEN return _UNPROVEN def _module_literal_values(tree: ast.Module) -> tuple[dict[str, Any], set[str]]: values: dict[str, Any] = {} topic_names: set[str] = set() for _ in range(3): for statement in tree.body: targets: list[ast.AST] = [] value_node: ast.AST | None = None if isinstance(statement, ast.Assign): targets = list(statement.targets) value_node = statement.value elif isinstance(statement, ast.AnnAssign) and statement.value is not None: targets = [statement.target] value_node = statement.value if value_node is None: continue for target in targets: if isinstance(target, ast.Name) and target.id.upper().endswith("TOPIC"): topic_names.add(target.id) value = _evaluate_control_expression(value_node, values) if value is _UNPROVEN: continue for target in targets: if not isinstance(target, ast.Name): continue values[target.id] = value return values, topic_names def _explicit_seed_expression(node: ast.AST) -> bool: if isinstance(node, ast.Constant): return isinstance(node.value, int) and not isinstance(node.value, bool) if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.UAdd, ast.USub)): return _explicit_seed_expression(node.operand) if isinstance(node, (ast.List, ast.Tuple)): return bool(node.elts) and all(_explicit_seed_expression(item) for item in node.elts) if ( isinstance(node, ast.Call) and _qualified_name(node.func) == "int" and len(node.args) == 1 and not node.keywords ): return isinstance(node.args[0], (ast.Name, ast.Constant)) return False def _seeded_prng_lines(tree: ast.Module) -> tuple[set[int], set[int]]: allowed: set[int] = set() disallowed: set[int] = set() for node in ast.walk(tree): if isinstance(node, ast.Attribute): if _qualified_name(node) in {"np.random.Generator", "numpy.random.Generator"}: allowed.add(int(getattr(node, "lineno", 0) or 0)) if not isinstance(node, ast.Call): continue name = _qualified_name(node.func) if name in {"np.random.default_rng", "numpy.random.default_rng"}: seed_node: ast.AST | None = node.args[0] if node.args else None if seed_node is None: seed_node = next( (keyword.value for keyword in node.keywords if keyword.arg == "seed"), None, ) line_no = int(getattr(node, "lineno", 0) or 0) if seed_node is not None and _explicit_seed_expression(seed_node): allowed.add(line_no) else: disallowed.add(line_no) elif name in {"np.random.Generator", "numpy.random.Generator"}: disallowed.add(int(getattr(node, "lineno", 0) or 0)) elif name.startswith(("np.random.", "numpy.random.")): disallowed.add(int(getattr(node, "lineno", 0) or 0)) return allowed - disallowed, disallowed def _scientific_control_fabrication_line(line: str) -> bool: return bool( SCIENTIFIC_CONTROL_FABRICATION_RE.search(line) or SCIENTIFIC_CONTROL_METRIC_ASSIGNMENT_RE.search(line) ) def inspect_scientific_control_contract(source: str, path: str) -> ScientificControlEvidence: """Prove a narrowly enumerated module is control-only at its event boundary.""" if path not in SCIENTIFIC_CONTROL_EXACT_PATHS: return ScientificControlEvidence() try: tree = ast.parse(source, filename=path) except (SyntaxError, TypeError, ValueError): return ScientificControlEvidence() module_values, topic_names = _module_literal_values(tree) if not topic_names: return ScientificControlEvidence() if any( not isinstance(module_values.get(name), str) or not module_values[name].startswith("bio.control.") for name in topic_names ): return ScientificControlEvidence() contract_found = False for node in ast.walk(tree): if not isinstance(node, ast.Dict): continue value = _evaluate_control_expression(node, module_values, control_only=True) if isinstance(value, dict) and all( value.get(key) == expected for key, expected in SCIENTIFIC_CONTROL_CONTRACT.items() ): contract_found = True for key_node, value_node in zip(node.keys, node.values): if key_node is None: continue key = _evaluate_control_expression(key_node, module_values) if key not in SCIENTIFIC_CONTROL_SAFETY_FIELDS: continue safety_value = _evaluate_control_expression( value_node, module_values, control_only=True, ) if safety_value is not False: return ScientificControlEvidence() if not contract_found: return ScientificControlEvidence() for node in ast.walk(tree): value_node: ast.AST | None = None targets: list[ast.AST] = [] if isinstance(node, ast.Assign): value_node = node.value targets = list(node.targets) elif isinstance(node, ast.AnnAssign) and node.value is not None: value_node = node.value targets = [node.target] if value_node is not None: safety_targets: set[str] = set() for target in targets: safety_targets.update(_field_names(target)) if safety_targets & { _normalise_identifier(field) for field in SCIENTIFIC_CONTROL_SAFETY_FIELDS }: safety_value = _evaluate_control_expression( value_node, module_values, control_only=True, ) if safety_value is not False: return ScientificControlEvidence() if not isinstance(node, ast.Call): continue for keyword in node.keywords: if ( keyword.arg in SCIENTIFIC_CONTROL_SAFETY_FIELDS and _evaluate_control_expression( keyword.value, module_values, control_only=True, ) is not False ): return ScientificControlEvidence() name = _qualified_name(node.func) if name != "Thought" and not name.endswith(".Thought"): continue topic_keyword = next( (keyword for keyword in node.keywords if keyword.arg == "topic"), None, ) if topic_keyword is None: return ScientificControlEvidence() topic = _evaluate_control_expression(topic_keyword.value, module_values) if not isinstance(topic, str) or not topic.startswith("bio.control."): return ScientificControlEvidence() allowed_prng_lines, _ = _seeded_prng_lines(tree) return ScientificControlEvidence( proven=True, allowed_seeded_prng_lines=frozenset(allowed_prng_lines), ) def _field_names(node: ast.AST) -> set[str]: if isinstance(node, ast.Name): return {_normalise_identifier(node.id)} if isinstance(node, ast.Attribute): return {_normalise_identifier(node.attr)} if isinstance(node, ast.Subscript): if isinstance(node.slice, ast.Constant) and isinstance(node.slice.value, str): return {_normalise_identifier(node.slice.value)} return set() if isinstance(node, (ast.Tuple, ast.List)): fields: set[str] = set() for child in node.elts: fields.update(_field_names(child)) return fields return set() def _contains_current_receipt_clock(node: ast.AST) -> bool: for child in ast.walk(node): if not isinstance(child, ast.Call): continue name = _qualified_name(child.func).lower() if name in { "time.time", "datetime.now", "datetime.utcnow", "datetime.datetime.now", "datetime.datetime.utcnow", }: return True return False def _is_numeric_literal(node: ast.AST) -> bool: if isinstance(node, ast.Constant): return isinstance(node.value, (int, float)) and not isinstance(node.value, bool) return ( isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.UAdd, ast.USub)) and _is_numeric_literal(node.operand) ) def _provider_numeric_fallback(call: ast.Call) -> bool: if not isinstance(call.func, ast.Attribute) or call.func.attr != "get" or len(call.args) < 2: return False key_node, default_node = call.args[0], call.args[1] if ( not isinstance(key_node, ast.Constant) or not isinstance(key_node.value, str) or not _is_numeric_literal(default_node) ): return False key = _normalise_identifier(key_node.value) receiver = _normalise_identifier(_qualified_name(call.func.value)) if key in PROVIDER_SPECIFIC_NUMERIC_FIELDS: return any(token in receiver for token in PROVIDER_RECEIVER_TOKENS) if key in GENERIC_MARKET_NUMERIC_FIELDS: return any(token in receiver for token in MARKET_RECEIVER_TOKENS) return False def scan_python_ast( path: Path, root: Path, *, operational_override: bool = False, ) -> list[Finding]: if path.suffix.lower() != ".py": return [] rel = rel_posix(path, root) try: source = path.read_text(encoding="utf-8", errors="replace") tree = ast.parse(source, filename=rel) except (OSError, SyntaxError, ValueError): return [] lines = source.splitlines() fixture = (is_test_fixture_path(rel) or is_quarantined(rel)) and not operational_override operational = operational_override or is_operational_path(rel) severity = "fixture" if fixture else ("error" if operational else "warning") findings: list[Finding] = [] seen: set[tuple[str, int]] = set() def add(code: str, node: ast.AST) -> None: line_no = int(getattr(node, "lineno", 0) or 0) identity = (code, line_no) if identity in seen: return seen.add(identity) text = lines[line_no - 1].strip()[:240] if 0 < line_no <= len(lines) else "" findings.append(Finding(severity, code, rel, line_no, text)) for node in ast.walk(tree): timestamp_value: ast.AST | None = None timestamp_fields: set[str] = set() if isinstance(node, ast.Assign): timestamp_value = node.value for target in node.targets: timestamp_fields.update(_field_names(target)) elif isinstance(node, ast.AnnAssign) and node.value is not None: timestamp_value = node.value timestamp_fields.update(_field_names(node.target)) elif isinstance(node, ast.NamedExpr): timestamp_value = node.value timestamp_fields.update(_field_names(node.target)) if ( timestamp_value is not None and timestamp_fields & SOURCE_TIMESTAMP_FIELDS and _contains_current_receipt_clock(timestamp_value) ): add("source_timestamp_from_receipt_clock", node) if isinstance(node, ast.Dict): for key_node, value_node in zip(node.keys, node.values): if ( isinstance(key_node, ast.Constant) and isinstance(key_node.value, str) and _normalise_identifier(key_node.value) in SOURCE_TIMESTAMP_FIELDS and _contains_current_receipt_clock(value_node) ): add("source_timestamp_from_receipt_clock", key_node) if isinstance(node, ast.Call): for keyword in node.keywords: if ( keyword.arg is not None and _normalise_identifier(keyword.arg) in SOURCE_TIMESTAMP_FIELDS and _contains_current_receipt_clock(keyword.value) ): add("source_timestamp_from_receipt_clock", keyword) if _provider_numeric_fallback(node): add("provider_numeric_default", node) return findings def scan_text_file( path: Path, root: Path, *, operational_override: bool = False, ) -> list[Finding]: rel = rel_posix(path, root) findings: list[Finding] = [] try: source = path.read_text(encoding="utf-8", errors="replace") lines = source.splitlines() except Exception as exc: return [Finding("error", "unreadable", rel, 0, f"{type(exc).__name__}: {exc}")] operational = operational_override or is_operational_path(rel) fixture = (is_test_fixture_path(rel) or is_quarantined(rel)) and not operational_override scientific_control = inspect_scientific_control_contract(source, rel) in_docstring = False for line_no, line in enumerate(lines, start=1): stripped = line.strip() if not stripped or stripped.startswith(("#", "//", "/*", "*")): continue triple_count = line.count('"""') + line.count("'''") docstring_line = in_docstring or triple_count > 0 for code, pattern in BLOCK_PATTERNS: if not pattern.search(line): continue if detector_definition_context(line, rel, code): continue if scientific_control.proven: if ( code == "mock_or_synthetic_marker" and not _scientific_control_fabrication_line(line) ): continue if ( code == "numpy_random_runtime" and line_no in scientific_control.allowed_seeded_prng_lines and not _scientific_control_fabrication_line(line) ): continue if code == "mock_or_synthetic_marker" and ( approved_runtime_context(line) or non_data_marker_context(line, rel) ): continue if fixture: severity = "fixture" elif docstring_line: severity = "warning" elif code == "mock_or_synthetic_marker" and operational: severity = "error" elif code in {"python_random_runtime", "numpy_random_runtime", "js_random_runtime"} and not random_line_is_metric_like(line, rel): severity = "warning" elif code == "mock_or_synthetic_marker": severity = "warning" elif operational: severity = "error" else: severity = "warning" findings.append(Finding(severity, code, rel, line_no, stripped[:240])) if triple_count % 2 == 1: in_docstring = not in_docstring findings.extend(scan_python_ast(path, root, operational_override=operational_override)) return findings def _module_name_for_path(path: str) -> str: if not path.endswith(".py"): return "" module = path[:-3].replace("/", ".") return module[:-9] if module.endswith(".__init__") else module def find_live_simulation_imports(paths: Iterable[Path], root: Path) -> tuple[set[str], list[Finding]]: python_paths = [path for path in paths if path.suffix.lower() == ".py"] module_paths = { _module_name_for_path(rel_posix(path, root)): rel_posix(path, root) for path in python_paths } promoted: set[str] = set() findings: list[Finding] = [] seen: set[tuple[str, str]] = set() for importer in python_paths: importer_rel = rel_posix(importer, root) if not is_operational_path(importer_rel): continue try: tree = ast.parse( importer.read_text(encoding="utf-8", errors="replace"), filename=importer_rel, ) except (OSError, SyntaxError, ValueError): continue for node in ast.walk(tree): imported_modules: list[str] = [] if isinstance(node, ast.Import): imported_modules.extend(alias.name for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: imported_modules.append(node.module) imported_modules.extend(f"{node.module}.{alias.name}" for alias in node.names) else: continue for module_name in imported_modules: target_rel = module_paths.get(module_name) if ( not target_rel or not target_rel.startswith(LIVE_NAMED_SIMULATION_PREFIX) or not Path(target_rel).stem.endswith("_live") ): continue identity = (importer_rel, target_rel) if identity in seen: continue seen.add(identity) promoted.add(target_rel) findings.append( Finding( "warning", "operational_imports_live_simulation_module", importer_rel, int(getattr(node, "lineno", 0) or 0), f"imports {target_rel}", ) ) return promoted, findings def validate_registry(root: Path) -> list[Finding]: findings: list[Finding] = [] registry = load_source_registry(root) sources = registry.get("sources", {}) if not isinstance(sources, dict): return [Finding("error", "source_registry_invalid", "data/real_data_sources.json", 0, "sources must be object")] for source_id, source in sources.items(): if not isinstance(source, dict): findings.append(Finding("error", "source_registry_invalid", "data/real_data_sources.json", 0, str(source_id))) continue for key in ("name", "category", "endpoint", "freshness_ttl_sec", "derived_metrics_allowed"): if key not in source: findings.append( Finding("error", "source_registry_missing_field", "data/real_data_sources.json", 0, f"{source_id}.{key}") ) demo_key_marker = "DEMO" + "_KEY" if demo_key_marker in json.dumps(source): findings.append(Finding("error", "source_registry_demo_key", "data/real_data_sources.json", 0, str(source_id))) return findings def validate_public_metric_json(path: Path, root: Path, source_ids: set[str]) -> list[Finding]: rel = rel_posix(path, root) findings: list[Finding] = [] try: payload = json.loads(path.read_text(encoding="utf-8", errors="replace")) except Exception: return findings metric_envelope_keys = { "source_id", "source_name", "source_url", "collected_at", "freshness_ttl_sec", "is_operational_metric", } def walk(value: Any, trail: str) -> None: if isinstance(value, dict): if "truth_status" in value: if metric_envelope_keys & set(value): try: validate_metric_envelope(value, registry_source_ids=source_ids) except ValueError as exc: findings.append(Finding("error", "invalid_metric_envelope", rel, 0, f"{trail}: {exc}")) return for key, child in value.items(): walk(child, f"{trail}.{key}" if trail else str(key)) elif isinstance(value, list): for idx, child in enumerate(value): walk(child, f"{trail}[{idx}]") walk(payload, "") return findings def run(root: Path, *, strict: bool = False, json_output: bool = False) -> int: findings: list[Finding] = [] findings.extend(validate_registry(root)) source_ids = registered_source_ids(root) if "test_fixture" not in source_ids: findings.append(Finding("error", "source_registry_missing_test_fixture", "data/real_data_sources.json", 0, "")) paths = list(iter_text_files(root)) promoted_operational_paths, import_findings = find_live_simulation_imports(paths, root) findings.extend(import_findings) for path in paths: rel = rel_posix(path, root) findings.extend( scan_text_file( path, root, operational_override=rel in promoted_operational_paths, ) ) if rel.startswith("frontend/public/") and path.suffix.lower() == ".json": findings.extend(validate_public_metric_json(path, root, source_ids)) error_count = sum(1 for item in findings if item.severity == "error") warning_count = sum(1 for item in findings if item.severity == "warning") fixture_count = sum(1 for item in findings if item.severity == "fixture") summary = { "schema_version": "aureon-real-data-contract-validation-v1", "root": str(root), "truth_statuses": sorted(TRUTH_STATUSES), "error_count": error_count, "warning_count": warning_count, "fixture_count": fixture_count, "source_count": len(source_ids), "findings": [item.to_dict() for item in findings], } if json_output: print(json.dumps(summary, indent=2, sort_keys=True)) else: print( "real-data contract: " f"errors={error_count} warnings={warning_count} fixtures={fixture_count} sources={len(source_ids)}" ) for item in findings[:200]: text = item.text.encode("ascii", errors="replace").decode("ascii") print(f"{item.severity.upper()} {item.code} {item.path}:{item.line} {text}") if len(findings) > 200: print(f"... {len(findings) - 200} additional findings omitted") return 1 if error_count or (strict and warning_count) else 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", default=".") parser.add_argument("--strict", action="store_true", help="Treat warnings as failure.") parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.") args = parser.parse_args(argv) root = repo_root_from(Path(args.repo_root)) return run(root, strict=args.strict, json_output=args.json) if __name__ == "__main__": raise SystemExit(main())