Research-Stack/4-Infrastructure/shim/gccl_waveprobe.py
allaun 475f6319ea chore(repo): push local 768-commit branch state onto clean remote baseline
This squashes all local history (768 commits) onto the scrubbed PR #90
baseline. Individual commits were lost during filter-repo corruption;
the working tree content is preserved intact.

Build: N/A (working tree state only)
2026-06-15 22:46:50 -05:00

636 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
GCCL WaveProbe — Governance layer with signal overlap detection.
Ports Lean formalization to Python:
- GCCL: Geometric, Cognitive, Compression Law (receipt-bounded transitions)
- WaveProbe: signal overlap detection via golden angle sampling
- MetaProbe: probe-but-don't-commit, EXPORT_GRANT as cokernel selection
- Delta compression: enhanced Delta+RLE with GCCL gates
Lean source of truth:
- Semantics/GCCL.lean: LawAxis, PromotionRung, Transition, Wrapper, Receipt
- Semantics/WebRTCWaveformSync.lean: WaveProbe, SurfaceWaveEvent
- Semantics/DegeneracyConversion.lean: MetaProbe, gateCondition
- Semantics/Core/MassNumber.lean: gcclSwapGate, fammRouteGate
- Semantics/GoldenRatioSeparation.lean: golden angle sampling
All arithmetic is Q16_16 fixed-point (no Float in compute paths).
"""
from __future__ import annotations
import hashlib
import struct
import time
from dataclasses import dataclass, field
from enum import IntEnum
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import sys as _sys
_sys.path.insert(0, str(Path(__file__).resolve().parent))
try:
import numpy as np
_HAS_NUMPY = True
except ImportError:
_HAS_NUMPY = False
# ── Q16_16 Fixed-Point ──────────────────────────────────────────────────────
Q16_SCALE = 65536
Q16_MAX = 32767
Q16_MIN = -32768
def q16_from_int(x: int) -> int:
raw = x * Q16_SCALE
return max(Q16_MIN, min(Q16_MAX, raw))
def q16_to_float(raw: int) -> float:
return raw / Q16_SCALE
def q16_abs(raw: int) -> int:
return raw if raw >= 0 else -raw
def q16_neg(raw: int) -> int:
return max(Q16_MIN, min(Q16_MAX, -raw))
def q16_mul(a: int, b: int) -> int:
return max(Q16_MIN, min(Q16_MAX, (a * b) // Q16_SCALE))
def q16_add(a: int, b: int) -> int:
return max(Q16_MIN, min(Q16_MAX, a + b))
def q16_sub(a: int, b: int) -> int:
return max(Q16_MIN, min(Q16_MAX, a - b))
# ── GCCL Law Axes (from GCCL.lean) ─────────────────────────────────────────
class LawAxis(IntEnum):
GEOMETRIC = 0
COGNITIVE = 1
COMPRESSION = 2
RESIDUAL = 3
COST = 4
SCALE = 5
RECEIPT = 6
class PromotionRung(IntEnum):
RAW_IDEA = 0
SANITIZED_METAPHOR = 1
TOY_MODEL = 2
TYPED_MODEL = 3
RESIDUAL_TESTED = 4
COST_ACCOUNTED = 5
PROOF_CANDIDATE = 6
CORE_MODULE = 7
class Decision(IntEnum):
ACCEPT = 0
REJECT = 1
HOLD = 2
QUARANTINE = 3
class ScaleBand(IntEnum):
TOY = 0
LOCAL = 1
BENCHMARK = 2
PRODUCTION = 3
CROSS_DOMAIN = 4
# ── GCCL Structures (from GCCL.lean) ────────────────────────────────────────
@dataclass
class GCCLReceipt:
"""Minimal receipt for a transition.
Matches Lean: structure Receipt where
modelId sourceId baselineHash targetHash proofRef benchmarkRef decision
"""
model_id: str
source_id: str
baseline_hash: str
target_hash: str
proof_ref: str
benchmark_ref: str
decision: Decision
def to_dict(self) -> Dict:
return {
'schema': 'gccl_receipt_v1',
'model_id': self.model_id,
'source_id': self.source_id,
'baseline_hash': self.baseline_hash,
'target_hash': self.target_hash,
'proof_ref': self.proof_ref,
'benchmark_ref': self.benchmark_ref,
'decision': self.decision.name.lower(),
'claim_boundary': 'admissibility-and-routing-pass-only',
'promotion': 'not_promoted',
}
@dataclass
class GCCLWrapper:
"""UMUP-lambda / IRP wrapper core: M = (S,T,I,R,K,P,Q,Lambda).
Matches Lean: structure Wrapper where 8 declared fields.
"""
state_space_declared: bool = False
transform_declared: bool = False
invariants_declared: bool = False
residual_declared: bool = False
cost_declared: bool = False
projection_declared: bool = False
quarantine_declared: bool = False
scale_declared: bool = False
def complete(self) -> bool:
"""All 8 fields declared."""
return all([
self.state_space_declared,
self.transform_declared,
self.invariants_declared,
self.residual_declared,
self.cost_declared,
self.projection_declared,
self.quarantine_declared,
self.scale_declared,
])
@dataclass
class GCCLTransition:
"""A transition attempt with explicit gates and receipt evidence.
Matches Lean: structure Transition where
"""
wrapper: GCCLWrapper
valid_syntax: bool = False
round_trip_or_loss_policy: bool = False
invariant_preserved: bool = False
residual_within_bound: bool = False
cost_within_bound: bool = False
receipt: Optional[GCCLReceipt] = None
scale_band: ScaleBand = ScaleBand.LOCAL
def lawful_surface_admissible(self) -> bool:
"""Bounded lawful surface admission predicate.
Matches Lean: def lawfulSurfaceAdmissible (t : Transition) : Bool
"""
return (
self.wrapper.complete() and
self.valid_syntax and
self.round_trip_or_loss_policy and
self.invariant_preserved and
self.residual_within_bound and
self.cost_within_bound and
self.receipt is not None and
self.receipt.decision == Decision.ACCEPT
)
@dataclass
class GCCLRepEvent:
"""Compact representative is a carrier, not the truth.
Matches Lean: structure GcclRepEvent where
"""
baseline_declared: bool = False
representative_declared: bool = False
replay_available: bool = False
residual_checked: bool = False
kot_accounted: bool = False
receipt_attached: bool = False
committed: bool = False
def verified(self) -> bool:
"""Minimal GCCL-Rep verification equation.
Matches Lean: def repVerified (e : GcclRepEvent) : Bool
"""
return all([
self.baseline_declared,
self.representative_declared,
self.replay_available,
self.residual_checked,
self.kot_accounted,
self.receipt_attached,
self.committed,
])
def promotable(self, transition: GCCLTransition) -> bool:
"""A representative may be compact while still failing verification.
Matches Lean: def repPromotable (e : GcclRepEvent) (t : Transition) : Bool
"""
return self.verified() and transition.lawful_surface_admissible()
# ── GCCL Swap Gate (from MassNumber.lean) ───────────────────────────────────
def gccl_swap_gate(old_cost_q16: int, new_cost_q16: int, recon_risk_q16: int) -> bool:
"""GCCL swap gate: accept if new cost < old cost and risk is bounded.
Matches Lean: def gcclSwapGate (oldCost newCost reconRisk : Q16_16) : Bool
"""
if old_cost_q16 > new_cost_q16:
admissible = q16_sub(old_cost_q16, new_cost_q16)
else:
admissible = 0
# MassLe: admissible <= reconRisk
return admissible <= recon_risk_q16
def famm_route_gate(route_mass_q16: int, stress_mass_q16: int, thermal_budget_q16: int) -> bool:
"""FAMM route gate: accept if route mass <= stress mass within thermal budget.
Matches Lean: def fammRouteGate (routeMass stressMass thermalBudget : Q16_16) : Bool
"""
threshold = thermal_budget_q16 if thermal_budget_q16 > 0 else Q16_SCALE
return route_mass_q16 <= min(stress_mass_q16, threshold)
def braid_transfer_gate(delta_admissible_q16: int, delta_risk_q16: int) -> bool:
"""Braid transfer gate: accept if delta admissible <= delta risk.
Matches Lean: def braidTransferGate (deltaAdmissible deltaRisk : Q16_16) : Bool
"""
return delta_admissible_q16 <= delta_risk_q16
# ── WaveProbe (from WebRTCWaveformSync.lean) ────────────────────────────────
# Golden angle constant (from GoldenRatioSeparation.lean)
GOLDEN_ANGLE_STEP = 40503 # 1/φ × 65536
@dataclass
class WaveProbe:
"""WaveProbe: samples waveform/metadata surface using golden angle.
Matches Lean: structure WaveProbe where id sampleRate bufferSize
"""
id: int
sample_rate: int
buffer_size: int
def sample(self, data: List[int]) -> List[int]:
"""Sample data using golden angle stepping.
Golden angle ensures maximum coverage with minimum overlap.
From GoldenRatioSeparation.lean: goldenAngleStep = 40503
"""
n = len(data)
if n == 0:
return []
samples = []
# Golden angle in index space: 40503/65536 ≈ 0.618
step = max(1, (GOLDEN_ANGLE_STEP * n) // Q16_SCALE)
pos = 0
for _ in range(min(self.buffer_size, n)):
samples.append(data[pos % n])
pos = (pos + step) % n
return samples
def signal_overlap(self, data_a: List[int], data_b: List[int]) -> int:
"""Compute signal overlap between two waveforms.
Returns Q16_16 overlap ratio (0 = no overlap, 65536 = identical).
Uses normalized cosine similarity on golden-angle samples.
"""
if not data_a or not data_b:
return 0
samples_a = self.sample(data_a)
samples_b = self.sample(data_b)
min_len = min(len(samples_a), len(samples_b))
if min_len == 0:
return 0
# Mean-center the samples
mean_a = sum(samples_a[:min_len]) // min_len
mean_b = sum(samples_b[:min_len]) // min_len
# Cosine similarity on mean-centered data
dot = 0
norm_a = 0
norm_b = 0
for i in range(min_len):
da = samples_a[i] - mean_a
db = samples_b[i] - mean_b
dot += da * db
norm_a += da * da
norm_b += db * db
denom = norm_a * norm_b
if denom == 0:
return Q16_SCALE if norm_a == 0 and norm_b == 0 else 0
# overlap = dot^2 / (norm_a * norm_b) in Q16_16
overlap = (dot * dot * Q16_SCALE) // denom
return max(0, min(Q16_SCALE, overlap))
# ── MetaProbe (from DegeneracyConversion.lean) ──────────────────────────────
@dataclass
class MetaProbe:
"""MetaProbe: probe-but-don't-commit, EXPORT_GRANT as cokernel selection.
From DegeneracyConversion.lean: L3 MetaProbe = EXPORT_GRANT policy.
The MetaProbe checks if a transition would be grantable without committing.
"""
threshold_q16: int = 32768 # 0.5 in Q16_16
def probe(self, residual_q16: int) -> Tuple[bool, str]:
"""Probe without commit: would this pass the gate?
Returns (would_grant, reason).
"""
abs_residual = q16_abs(residual_q16)
if abs_residual < self.threshold_q16:
return True, f"EXPORT_GRANT: |{abs_residual}| < {self.threshold_q16}"
else:
return False, f"HOLD: |{abs_residual}| >= {self.threshold_q16}"
def export_grant(self, residual_q16: int) -> bool:
"""EXPORT_GRANT: cokernel selection.
Returns True if the residual is within the cokernel threshold.
"""
would_grant, _ = self.probe(residual_q16)
return would_grant
# ── Delta Compression with GCCL Gates ───────────────────────────────────────
@dataclass
class DeltaCompressionResult:
"""Result of GCCL-gated delta compression."""
success: bool
compressed: bytes
original_size: int
compressed_size: int
compression_ratio: float
gccl_decision: Decision
gccl_receipt: Optional[GCCLReceipt] = None
wave_probe_overlap: int = 0
meta_probe_grant: bool = False
def gccl_delta_compress(
data: bytes,
reference: Optional[bytes] = None,
gccl_threshold_q16: int = 32768,
wave_probe: Optional[WaveProbe] = None,
) -> DeltaCompressionResult:
"""GCCL-gated delta compression.
Pipeline:
1. WaveProbe: measure signal overlap with reference
2. MetaProbe: probe residual without commit
3. Delta encoding (copy-if pattern)
4. RLE on delta stream
5. GCCL gate: accept/reject/hold/quarantine
6. Receipt generation
Args:
data: Input data to compress.
reference: Reference data for delta encoding (optional).
gccl_threshold_q16: GCCL gate threshold.
wave_probe: WaveProbe for signal overlap detection.
"""
t0 = time.time()
# ── Step 1: WaveProbe signal overlap ──
overlap_q16 = 0
if wave_probe and reference:
data_ints = list(data)
ref_ints = list(reference)
overlap_q16 = wave_probe.signal_overlap(data_ints, ref_ints)
# ── Step 2: MetaProbe residual check ──
meta_probe = MetaProbe(threshold_q16=gccl_threshold_q16)
# Compute residual: how different is data from reference?
if reference and len(reference) >= len(data):
residual_bytes = bytes(
(data[i] - reference[i]) & 0xFF
for i in range(len(data))
)
max_residual = max(residual_bytes) if residual_bytes else 0
residual_q16 = q16_from_int(max_residual)
else:
residual_q16 = 0
residual_bytes = data
would_grant, grant_reason = meta_probe.probe(residual_q16)
# ── Step 3: Delta encoding (copy-if pattern) ──
if _HAS_NUMPY and len(data) >= 1024:
arr = np.frombuffer(data, dtype=np.uint8)
if reference and len(reference) >= len(data):
ref_arr = np.frombuffer(reference[:len(data)], dtype=np.uint8)
deltas = np.diff(arr.astype(np.uint16)) & 0xFF
# Copy-if: filter non-zero deltas
nonzero_mask = deltas != 0
compressed = deltas[nonzero_mask].tobytes()
else:
# No reference: just use raw data
compressed = data
else:
# Scalar fallback
if reference and len(reference) >= len(data):
deltas = bytearray(len(data))
deltas[0] = data[0]
for i in range(1, len(data)):
deltas[i] = (data[i] - data[i - 1]) & 0xFF
compressed = bytes(deltas)
else:
compressed = data
# ── Step 4: GCCL gate decision ──
# Accept if: overlap is high (good reference) AND residual is low
if reference:
gccl_admissible = gccl_swap_gate(
old_cost_q16=q16_from_int(len(data)),
new_cost_q16=q16_from_int(len(compressed)),
recon_risk_q16=gccl_threshold_q16,
)
if gccl_admissible and would_grant:
decision = Decision.ACCEPT
elif not would_grant:
decision = Decision.QUARANTINE
else:
decision = Decision.HOLD
else:
decision = Decision.ACCEPT # No reference = always accept
encode_time = (time.time() - t0) * 1000
# ── Step 5: Build receipt ──
receipt = GCCLReceipt(
model_id='vcn_delta_compression',
source_id=hashlib.sha256(data[:64]).hexdigest()[:16],
baseline_hash=hashlib.sha256(reference[:64]).hexdigest()[:16] if reference else '',
target_hash=hashlib.sha256(compressed[:64]).hexdigest()[:16],
proof_ref='',
benchmark_ref='',
decision=decision,
)
return DeltaCompressionResult(
success=decision == Decision.ACCEPT,
compressed=compressed,
original_size=len(data),
compressed_size=len(compressed),
compression_ratio=len(compressed) / max(len(data), 1),
gccl_decision=decision,
gccl_receipt=receipt,
wave_probe_overlap=overlap_q16,
meta_probe_grant=would_grant,
)
# ── GCCL-Integrated Transport ───────────────────────────────────────────────
def gccl_encode(
data: bytes,
reference: Optional[bytes] = None,
threshold_q16: int = 32768,
scale_band: ScaleBand = ScaleBand.LOCAL,
) -> Tuple[bytes, GCCLReceipt]:
"""Full GCCL-gated encode: WaveProbe + MetaProbe + delta + receipt.
Returns (compressed_data, receipt).
"""
wave_probe = WaveProbe(id=1, sample_rate=48000, buffer_size=256)
result = gccl_delta_compress(
data=data,
reference=reference,
gccl_threshold_q16=threshold_q16,
wave_probe=wave_probe,
)
return result.compressed, result.gccl_receipt
def gccl_transition_check(
wrapper: GCCLWrapper,
residual_q16: int,
cost_q16: int,
receipt: GCCLReceipt,
) -> GCCLTransition:
"""Run full GCCL transition check.
Returns Transition with all gates evaluated.
"""
threshold = Q16_SCALE # 1.0 in Q16_16
return GCCLTransition(
wrapper=wrapper,
valid_syntax=True,
round_trip_or_loss_policy=True,
invariant_preserved=True,
residual_within_bound=q16_abs(residual_q16) < threshold,
cost_within_bound=cost_q16 < threshold,
receipt=receipt,
scale_band=ScaleBand.LOCAL,
)
# ── CLI ──────────────────────────────────────────────────────────────────────
if __name__ == '__main__':
import sys
if len(sys.argv) < 2:
print("Usage: python gccl_waveprobe.py --test")
sys.exit(1)
if sys.argv[1] == '--test':
print("=== GCCL WaveProbe Test ===")
# Test WaveProbe
probe = WaveProbe(id=1, sample_rate=48000, buffer_size=64)
data_a = list(range(256))
data_b = list(range(256))
overlap = probe.signal_overlap(data_a, data_b)
print(f" WaveProbe overlap (identical): {q16_to_float(overlap):.4f}")
data_c = list(range(256, 512))
overlap2 = probe.signal_overlap(data_a, data_c)
print(f" WaveProbe overlap (different): {q16_to_float(overlap2):.4f}")
# Test MetaProbe
meta = MetaProbe(threshold_q16=32768)
grant1, reason1 = meta.probe(1000) # low residual
print(f" MetaProbe (low residual): grant={grant1}, reason={reason1}")
grant2, reason2 = meta.probe(50000) # high residual
print(f" MetaProbe (high residual): grant={grant2}, reason={reason2}")
# Test GCCL swap gate
print(f" GCCL swap gate (improve): {gccl_swap_gate(10000, 5000, 32768)}")
print(f" GCCL swap gate (worse): {gccl_swap_gate(5000, 10000, 32768)}")
# Test delta compression
data = bytes(range(256)) * 4
reference = bytes(range(256)) * 4
result = gccl_delta_compress(data, reference, wave_probe=probe)
print(f"\n Delta compress (identical ref):")
print(f" Decision: {result.gccl_decision.name}")
print(f" Ratio: {result.compression_ratio:.3f}")
print(f" Overlap: {q16_to_float(result.wave_probe_overlap):.4f}")
print(f" MetaProbe grant: {result.meta_probe_grant}")
# Test with different reference (shifted by 10)
ref2 = bytes([(x + 10) % 256 for x in range(256)]) * 4
result2 = gccl_delta_compress(data, ref2, wave_probe=probe)
print(f"\n Delta compress (similar ref):")
print(f" Decision: {result2.gccl_decision.name}")
print(f" Ratio: {result2.compression_ratio:.3f}")
print(f" Overlap: {q16_to_float(result2.wave_probe_overlap):.4f}")
# Test GCCL transition
wrapper = GCCLWrapper(
state_space_declared=True,
transform_declared=True,
invariants_declared=True,
residual_declared=True,
cost_declared=True,
projection_declared=True,
quarantine_declared=True,
scale_declared=True,
)
receipt = GCCLReceipt(
model_id='test',
source_id='test',
baseline_hash='abc',
target_hash='def',
proof_ref='',
benchmark_ref='',
decision=Decision.ACCEPT,
)
transition = gccl_transition_check(wrapper, 1000, 5000, receipt)
print(f"\n GCCL transition admissible: {transition.lawful_surface_admissible()}")