Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 19)
da0ce7b verified | """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", | |
| ) | |
| class Finding: | |
| severity: str | |
| code: str | |
| path: str | |
| line: int | |
| text: str | |
| def to_dict(self) -> dict[str, Any]: | |
| return asdict(self) | |
| 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()) | |