mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
229 lines
9.2 KiB
Python
229 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
# ==============================================================================
|
|
# COPYRIGHT NO ONE EVERYWHERE LLC (WYOMING HOLDING COMPANY)
|
|
# PROJECT: SOVEREIGN STACK
|
|
# This artifact is entirely proprietary and cryptographically proven.
|
|
# Open-Source usage requires explicit permission from Brandon Scott Schneider.
|
|
# ==============================================================================
|
|
"""GraphVM Canal Router — Structural classification and routing.
|
|
|
|
This module implements the Cognitive Triage and Dynamic Canal Routing
|
|
equations to evaluate the "action-worthiness" and "thermodynamic cost"
|
|
(KOT) of interacting with a contract.
|
|
|
|
Equations:
|
|
- Triage: T(x,t) = [wU*U + wS*S + wR*R + wH*H + wT*T - wB*B] * Pi
|
|
- Canal: d_canal(x->y,t) = ||Phi(y)-Phi(x)|| * eta * (1 + torsion) * (1 + accumulation)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
try:
|
|
from scripts.evm_bytecode_waveprobe import EVMBytecodeWaveprobe, EVMWaveprobeResult
|
|
except ModuleNotFoundError:
|
|
from evm_bytecode_waveprobe import EVMBytecodeWaveprobe, EVMWaveprobeResult # type: ignore
|
|
|
|
# ── Feature dimension ────────────────────────────────────────────────────────
|
|
FEATURE_DIM = 8
|
|
|
|
# ── DEFAULT TOLERANCES ───────────────────────────────────────────────────────
|
|
DEFAULT_TOLERANCES = {
|
|
"graphvm": {
|
|
"weights": {
|
|
"urgency": 0.4,
|
|
"salience": 0.2,
|
|
"surprise": 0.1,
|
|
"heat": 0.3,
|
|
"torsion": 0.2,
|
|
"background": 0.1
|
|
},
|
|
"thresholds": {
|
|
"theta_heat": 0.30,
|
|
"theta_canal": 1000.0,
|
|
"theta_working_set": 0.10,
|
|
"theta_immediate": 0.80
|
|
}
|
|
}
|
|
}
|
|
|
|
@dataclass
|
|
class CanalRoutingReport:
|
|
"""Formal routing decision based on structural grain."""
|
|
contract_sha256: str
|
|
overall_heat: float
|
|
overall_torsion: float
|
|
overall_anisotropy: float
|
|
triage_score: float
|
|
canal_cost_kot: float
|
|
valve_phi: int # 0 or 1
|
|
routing_decision: str # immediate | review | freeze
|
|
reason: str
|
|
applied_tolerances: Dict[str, Any] # Snapshot of thresholds/weights used
|
|
|
|
def deep_merge_dicts(base: dict, override: dict) -> dict:
|
|
"""Recursively merge two dictionaries."""
|
|
res = base.copy()
|
|
for k, v in override.items():
|
|
if k in res and isinstance(res[k], dict) and isinstance(v, dict):
|
|
res[k] = deep_merge_dicts(res[k], v)
|
|
else:
|
|
res[k] = v
|
|
return res
|
|
|
|
class GraphVMCanalRouter:
|
|
"""Integrates waveprobe metrics into a routing decision."""
|
|
|
|
def __init__(
|
|
self,
|
|
target_bytecode_hex: str,
|
|
baseline_drift_vec: Optional[List[float]] = None,
|
|
tolerances: Optional[dict] = None,
|
|
jurisdiction: Optional[str] = None,
|
|
):
|
|
self.probe = EVMBytecodeWaveprobe(bytecode_hex=target_bytecode_hex)
|
|
|
|
# Load Soliton Baseline if not provided
|
|
if baseline_drift_vec is None:
|
|
baseline_path = "4-Infrastructure/config/soliton_baseline.json"
|
|
if os.path.exists(baseline_path):
|
|
try:
|
|
with open(baseline_path, "r", encoding="utf-8") as f_in:
|
|
data = json.load(f_in)
|
|
self.baseline_drift = data.get("soliton_baseline_vector", [0.0] * FEATURE_DIM)
|
|
except (json.JSONDecodeError, IOError):
|
|
self.baseline_drift = [0.0] * FEATURE_DIM
|
|
else:
|
|
self.baseline_drift = [0.0] * FEATURE_DIM
|
|
else:
|
|
self.baseline_drift = baseline_drift_vec
|
|
|
|
# 1. Start with global defaults
|
|
full_tols = tolerances or DEFAULT_TOLERANCES
|
|
|
|
# 2. Overlay jurisdiction-specific overrides if present
|
|
if jurisdiction and "jurisdictions" in full_tols:
|
|
overrides = full_tols["jurisdictions"].get(jurisdiction, {})
|
|
if overrides:
|
|
full_tols = deep_merge_dicts(full_tols, overrides)
|
|
|
|
self.tols = full_tols.get("graphvm", {})
|
|
self.weights = self.tols.get("weights", {})
|
|
self.thresholds = self.tols.get("thresholds", {})
|
|
|
|
# Pull specific thresholds for fast access
|
|
self.theta_heat = self.thresholds.get("theta_heat", 0.30)
|
|
self.theta_canal = self.thresholds.get("theta_canal", 1000.0)
|
|
|
|
def _cosine_similarity(self, v1: List[float], v2: List[float]) -> float:
|
|
dot = sum(a * b for a, b in zip(v1, v2))
|
|
norm1 = math.sqrt(sum(a * a for a in v1))
|
|
norm2 = math.sqrt(sum(a * a for a in v2))
|
|
if norm1 < 1e-9 or norm2 < 1e-9:
|
|
return 0.0
|
|
return dot / (norm1 * norm2)
|
|
|
|
def route(self, urgency: float = 0.0, salience: float = 0.5) -> CanalRoutingReport:
|
|
"""Run waveprobe and compute triage + canal metrics."""
|
|
result: EVMWaveprobeResult = self.probe.analyze()
|
|
|
|
# 1. Aggregate Torsion across chunks
|
|
torsion_scores: list[float] = []
|
|
for chunk in result.chunks:
|
|
# Torsion = 1 - cos(base_features, baseline_drift)
|
|
sim = self._cosine_similarity(chunk.base_features, self.baseline_drift)
|
|
torsion_scores.append(1.0 - sim)
|
|
|
|
overall_torsion = sum(torsion_scores) / max(1, len(torsion_scores))
|
|
|
|
# 2. Raw Surprise (R) derived from overall heat/torsion
|
|
surprise = (result.overall_heat + overall_torsion) / 2.0
|
|
|
|
# 3. Admissibility Valve (Pi)
|
|
# Fail-closed if overall heat exceeds threshold
|
|
valve_phi = 1 if result.overall_heat < self.theta_heat else 0
|
|
|
|
# 4. Cognitive Triage (T)
|
|
w = self.weights
|
|
triage = (
|
|
w.get("urgency", 0.4) * urgency +
|
|
w.get("salience", 0.2) * salience +
|
|
w.get("surprise", 0.1) * surprise +
|
|
w.get("heat", 0.3) * result.overall_heat +
|
|
w.get("torsion", 0.2) * overall_torsion
|
|
) * valve_phi
|
|
|
|
# 5. Dynamic Canal Cost (d_canal) in KOT
|
|
# We model "moving through" this bytecode as traversing the Phi-space length
|
|
# weighted by structural heat and torsion.
|
|
phi_mag = result.overall_heat # Proxy for average feature movement magnitude
|
|
eta = 1.0 # Gas resistance placeholder
|
|
canal_cost = (phi_mag * 100) * eta * (1 + overall_torsion)
|
|
|
|
# 6. Routing Decision
|
|
if valve_phi == 0:
|
|
decision = "freeze"
|
|
reason = f"Heat valve failure (H={result.overall_heat:.3f} > {self.theta_heat})"
|
|
elif canal_cost > self.theta_canal:
|
|
decision = "review"
|
|
reason = f"Canal cost ceiling (KOT={canal_cost:.1f} > {self.theta_canal})"
|
|
elif triage > self.thresholds.get("theta_immediate", 0.8):
|
|
decision = "immediate"
|
|
reason = f"High triage score (T={triage:.3f})"
|
|
else:
|
|
decision = "review"
|
|
reason = f"Medium triage score (T={triage:.3f})"
|
|
|
|
return CanalRoutingReport(
|
|
contract_sha256=result.bytecode_sha256,
|
|
overall_heat=result.overall_heat,
|
|
overall_torsion=overall_torsion,
|
|
overall_anisotropy=sum(c.anisotropy for c in result.chunks) / max(1, len(result.chunks)),
|
|
triage_score=triage,
|
|
canal_cost_kot=canal_cost,
|
|
valve_phi=valve_phi,
|
|
routing_decision=decision,
|
|
reason=reason,
|
|
applied_tolerances={
|
|
"thresholds": self.thresholds,
|
|
"weights": self.weights
|
|
}
|
|
)
|
|
|
|
# ── CLI ──────────────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="GraphVM Canal Router")
|
|
parser.add_argument("bytecode", help="Hex-encoded bytecode")
|
|
parser.add_argument("--urgency", type=float, default=0.0)
|
|
parser.add_argument("--config", help="Path to compliance_tolerances.json")
|
|
parser.add_argument("--jurisdiction", help="Candidate jurisdiction (e.g. 'United States')")
|
|
args = parser.parse_args()
|
|
|
|
tols = None
|
|
if args.config:
|
|
with open(args.config, "r", encoding="utf-8") as f:
|
|
tols = json.load(f)
|
|
|
|
router = GraphVMCanalRouter(args.bytecode, tolerances=tols, jurisdiction=args.jurisdiction)
|
|
report = router.route(urgency=args.urgency)
|
|
|
|
print(json.dumps({
|
|
"contract": report.contract_sha256[:16],
|
|
"metrics": {
|
|
"heat": round(report.overall_heat, 6),
|
|
"torsion": round(report.overall_torsion, 6),
|
|
"anisotropy": round(report.overall_anisotropy, 6),
|
|
"triage": round(report.triage_score, 6),
|
|
"canal_cost_kot": round(report.canal_cost_kot, 2)
|
|
},
|
|
"valve": report.valve_phi,
|
|
"decision": report.routing_decision,
|
|
"reason": report.reason
|
|
}, indent=2))
|