diff --git a/app.py b/app.py index c9bae2c..f8e9f08 100644 --- a/app.py +++ b/app.py @@ -1222,6 +1222,7 @@ def _format_obliteration_metrics( coh = metrics.get("coherence") ref = metrics.get("refusal_rate") kl = metrics.get("kl_divergence") + spectral = metrics.get("spectral_certification") n_layers = len(getattr(pipeline, "_strong_layers", [])) parts = ["### Liberation Results\n"] @@ -1244,6 +1245,16 @@ def _format_obliteration_metrics( icon = "🟢" if kl_budget is not None and kl <= kl_budget else "🔴" budget_note = f" / {kl_budget:.4f}" if kl_budget is not None else "" parts.append(f"| Token KL / Budget | **{kl:.4f}{budget_note}** | {icon} |") + if spectral is not None: + spectral_icon = { + "GREEN": "🟢", + "YELLOW": "🟡", + "RED": "🔴", + "INCONCLUSIVE": "⚪", + }.get(spectral, "⚪") + reason = metrics.get("spectral_diagnostic_reason") + reason_note = f" ({reason})" if reason else "" + parts.append(f"| Spectral Diagnostic | **{spectral}{reason_note}** | {spectral_icon} |") if n_layers > 0: parts.append(f"| Layers Modified | **{n_layers}** | |") if load_settings is not None: diff --git a/obliteratus/abliterate.py b/obliteratus/abliterate.py index 1ba9736..b829645 100644 --- a/obliteratus/abliterate.py +++ b/obliteratus/abliterate.py @@ -7283,7 +7283,7 @@ class AbliterationPipeline: overall = "RED (incomplete)" overall_level = "RED" elif CertificationLevel.INCONCLUSIVE in levels: - overall = "INCONCLUSIVE (insufficient samples)" + overall = "INCONCLUSIVE (insufficient statistical evidence)" overall_level = "INCONCLUSIVE" elif CertificationLevel.YELLOW in levels: overall = "YELLOW (distributed refusal detected)" @@ -7293,13 +7293,29 @@ class AbliterationPipeline: overall_level = "GREEN" self._quality_metrics["spectral_certification"] = overall_level + inconclusive_reasons = sorted({ + c.diagnostic_reason + for c in cert_results + if c.diagnostic_reason is not None + }) + self._quality_metrics["spectral_diagnostic_reason"] = ( + ",".join(inconclusive_reasons) if inconclusive_reasons else None + ) + self._quality_metrics["spectral_effective_noise_rank"] = min( + c.effective_noise_rank for c in cert_results + ) + self._quality_metrics["spectral_samples_required"] = max( + c.n_samples_required for c in cert_results + ) self.log(f" Spectral certificate: {overall}") for c in cert_results: self.log( f" Layer {cert_layers[cert_results.index(c)]}: " f"{c.level.value} (leading_eig={c.leading_eigenvalue:.4f}, " - f"bbp_threshold={c.bbp_threshold:.4f}, " - f"margin={c.eigenvalue_margin:+.4f})" + f"bbp_threshold={c.bbp_threshold:.6g}, " + f"margin={c.eigenvalue_margin:+.4f}, " + f"noise_rank={c.effective_noise_rank}, " + f"reason={c.diagnostic_reason or 'none'})" ) if overall_level == "RED": n_above = max(c.n_eigenvalues_above_threshold for c in cert_results) diff --git a/obliteratus/analysis/spectral_certification.py b/obliteratus/analysis/spectral_certification.py index 1b010cb..56c017d 100644 --- a/obliteratus/analysis/spectral_certification.py +++ b/obliteratus/analysis/spectral_certification.py @@ -105,6 +105,8 @@ class SpectralCertificate: n_samples_used: int # samples used for this analysis n_samples_required: int # minimum samples for reliable detection is_sample_sufficient: bool # whether we have enough data + effective_noise_rank: int # non-zero rank of within-class spectrum + diagnostic_reason: str | None # machine-readable inconclusive reason # Recommendations recommendation: str # human-readable recommendation @@ -167,9 +169,20 @@ class SpectralCertifier: SpectralCertificate with a statistical classification, or an inconclusive result when the sample requirement is unmet. """ + if harmful_activations.ndim != 2 or harmless_activations.ndim != 2: + raise ValueError("spectral certification expects two 2-D activation matrices") + if harmful_activations.shape[1] != harmless_activations.shape[1]: + raise ValueError("harmful and harmless activations must share hidden width") n_h, d = harmful_activations.shape n_b = harmless_activations.shape[0] n = n_h + n_b + if n_h < 2 or n_b < 2 or d < 1: + raise ValueError("spectral certification requires at least two samples per class") + + finite_inputs = bool( + torch.isfinite(harmful_activations).all() + and torch.isfinite(harmless_activations).all() + ) # Step 1: Compute difference covariance matrix # Pooled covariance minus individual covariances @@ -183,14 +196,35 @@ class SpectralCertifier: harmful_centered = harmful_activations - harmful_mean harmless_centered = harmless_activations - harmless_mean - # Pooled within-class covariance (standard formula: sum of scatter - # matrices divided by total degrees of freedom) - cov_h = harmful_centered.T @ harmful_centered / max(n_h - 1, 1) - cov_b = harmless_centered.T @ harmless_centered / max(n_b - 1, 1) - pooled_cov = (cov_h * (n_h - 1) + cov_b * (n_b - 1)) / max(n - 2, 1) + # Work in sample space when n << d. The non-zero eigenvalues of + # X.T @ X and X @ X are identical, avoiding a 5120x5120 covariance + # whose median is structurally zero in the production regime. + pooled_centered = torch.cat((harmful_centered, harmless_centered), dim=0).float() + degrees_freedom = n - 2 + if finite_inputs: + singular_values = torch.linalg.svdvals(pooled_centered) + noise_spectrum = singular_values.square() / degrees_freedom + tolerance = ( + torch.finfo(noise_spectrum.dtype).eps + * max(pooled_centered.shape) + * noise_spectrum.max().item() + if noise_spectrum.numel() + else 0.0 + ) + positive_spectrum = noise_spectrum[noise_spectrum > tolerance] + else: + noise_spectrum = torch.empty(0) + positive_spectrum = torch.empty(0) + effective_noise_rank = int(positive_spectrum.numel()) - # Step 2: Estimate noise variance (median eigenvalue method) - noise_var = self._estimate_noise_variance(pooled_cov, n, d) + # trace(covariance) / d is an unbiased isotropic variance estimate and + # remains defined in dual space. A zero/non-finite estimate is evidence + # failure, never a tiny positive threshold. + noise_var = ( + noise_spectrum.sum().item() / d + if finite_inputs and noise_spectrum.numel() + else 0.0 + ) # Step 3: Compute BBP threshold gamma = d / max(n, 1) # aspect ratio @@ -199,7 +233,7 @@ class SpectralCertifier: isotropic_threshold = noise_var * (1 + math.sqrt(gamma)) ** 2 # Non-isotropic correction (OBLITERATUS heuristic extension) - kappa = self._estimate_condition_number(pooled_cov) + kappa = self._estimate_condition_number_from_spectrum(positive_spectrum) anisotropic_threshold = isotropic_threshold * math.sqrt(kappa) anisotropy_correction = math.sqrt(kappa) @@ -210,9 +244,15 @@ class SpectralCertifier: mp_lower = noise_var * max(0, (1 - math.sqrt(gamma)) ** 2) # Step 5: Eigenvalue analysis of between-class covariance - between_cov = torch.outer(diff, diff) # rank-1 between-class scatter - eigen_result = self._eigenvalue_analysis( - between_cov, bbp_threshold, mp_upper + leading = diff_norm ** 2 if math.isfinite(diff_norm) else 0.0 + eigenvalues = torch.tensor([leading]) + above = [0] if math.isfinite(leading) and leading > bbp_threshold else [] + eigen_result = EigenvalueAnalysis( + eigenvalues=eigenvalues, + eigenvectors=torch.empty(0), + above_threshold=above, + in_bulk=[0] if math.isfinite(leading) and mp_upper < leading <= bbp_threshold else [], + signal_subspace_dim=len(above), ) # Step 6: Classify certification level @@ -242,10 +282,27 @@ class SpectralCertifier: ) # Sample sufficiency check - # From BBP: need n > d / rho^2 where rho = signal_strength / noise_var - rho = diff_norm / max(math.sqrt(noise_var), 1e-10) - n_required = max(self.min_samples, int(d / max(rho ** 2, 0.01))) - is_sufficient = n >= n_required + # Observed signal strength cannot be allowed to waive identifiability: + # that made a random high-dimensional mean difference "prove" its own + # sufficiency. Retain the signal-based term, but impose a dimension-aware + # floor and require a usable within-class noise spectrum. + noise_valid = math.isfinite(noise_var) and noise_var > 0.0 + rho = diff_norm / math.sqrt(noise_var) if noise_valid else 0.0 + signal_requirement = int(d / max(rho ** 2, 0.01)) if noise_valid else d + n_required = max(self.min_samples, math.ceil(math.sqrt(d)), signal_requirement) + rank_valid = effective_noise_rank >= 2 + is_sufficient = n >= n_required and noise_valid and rank_valid and finite_inputs + + if not finite_inputs: + diagnostic_reason = "non_finite_activations" + elif not noise_valid: + diagnostic_reason = "zero_or_non_finite_noise_scale" + elif not rank_valid: + diagnostic_reason = "rank_deficient_noise_estimate" + elif n < n_required: + diagnostic_reason = "insufficient_samples" + else: + diagnostic_reason = None # Certification level. Do not emit a definitive traffic-light result # when the diagnostic's own sample requirement is unmet: doing so made @@ -266,9 +323,10 @@ class SpectralCertifier: # Recommendations if level == CertificationLevel.INCONCLUSIVE: recommendation = ( - f"Spectral diagnostic is inconclusive: only {n} samples were " - f"available, while {n_required} are recommended for reliable " - f"detection at this dimensionality." + f"Spectral diagnostic is inconclusive ({diagnostic_reason}): " + f"{n} samples and effective noise rank {effective_noise_rank} " + f"were available; at least {n_required} samples and a finite, " + "positive noise estimate are required." ) action = "more_samples" elif level == CertificationLevel.GREEN: @@ -318,6 +376,8 @@ class SpectralCertifier: n_samples_used=n, n_samples_required=n_required, is_sample_sufficient=is_sufficient, + effective_noise_rank=effective_noise_rank, + diagnostic_reason=diagnostic_reason, recommendation=recommendation, suggested_action=action, ) @@ -420,6 +480,17 @@ class SpectralCertifier: except Exception: return 1.0 + @staticmethod + def _estimate_condition_number_from_spectrum(spectrum: torch.Tensor) -> float: + """Return a bounded condition estimate from non-zero dual eigenvalues.""" + if spectrum.numel() < 2 or not torch.isfinite(spectrum).all(): + return 1.0 + minimum = spectrum.min().item() + maximum = spectrum.max().item() + if minimum <= 0 or not math.isfinite(minimum) or not math.isfinite(maximum): + return 1.0 + return max(1.0, min(maximum / minimum, 1e6)) + def _eigenvalue_analysis( self, between_cov: torch.Tensor, diff --git a/obliteratus/telemetry.py b/obliteratus/telemetry.py index 4852396..ecd5864 100644 --- a/obliteratus/telemetry.py +++ b/obliteratus/telemetry.py @@ -59,6 +59,8 @@ _PUBLIC_METRIC_RANGES: dict[str, tuple[float | None, float | None]] = { "kl_divergence": (0.0, None), "capability_score": (0.0, 1.0), "degenerate_count": (0.0, None), + "spectral_effective_noise_rank": (0.0, None), + "spectral_samples_required": (0.0, None), "time_seconds": (0.0, None), } @@ -116,12 +118,12 @@ def _normalize_quality_metrics( statuses: dict[str, str] = {} for name, raw_value in sorted((metrics or {}).items()): if name not in _PUBLIC_METRIC_RANGES and name not in { - "capability_results", "spectral_certification", + "capability_results", "spectral_certification", "spectral_diagnostic_reason", }: continue if name == "capability_results": value = _sanitize_public_value(raw_value) if isinstance(raw_value, dict) else None - elif name == "spectral_certification": + elif name in {"spectral_certification", "spectral_diagnostic_reason"}: value = _sanitize_public_text(raw_value, 40) if isinstance(raw_value, str) else None else: value = _metric_value(name, raw_value) diff --git a/tests/test_app_model_load_settings.py b/tests/test_app_model_load_settings.py index cae9203..9bcbbdc 100644 --- a/tests/test_app_model_load_settings.py +++ b/tests/test_app_model_load_settings.py @@ -95,3 +95,32 @@ assert "🔴" in card check=False, ) assert result.returncode == 0, result.stdout + result.stderr + + +@pytest.mark.operator_ui +def test_result_card_exposes_spectral_inconclusive_reason(): + script = r''' +from types import SimpleNamespace + +import app + +pipeline = SimpleNamespace( + _quality_metrics={ + "spectral_certification": "INCONCLUSIVE", + "spectral_diagnostic_reason": "insufficient_samples", + }, + _strong_layers=[1], +) +card = app._format_obliteration_metrics(pipeline, "advanced", "1s") +assert "Spectral Diagnostic" in card +assert "INCONCLUSIVE (insufficient_samples)" in card +assert "⚪" in card +''' + result = subprocess.run( + [sys.executable, "-c", script], + capture_output=True, + text=True, + timeout=60, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr diff --git a/tests/test_breakthrough_modules.py b/tests/test_breakthrough_modules.py index ae4bbdd..76bdc4b 100644 --- a/tests/test_breakthrough_modules.py +++ b/tests/test_breakthrough_modules.py @@ -10,6 +10,7 @@ from __future__ import annotations import math +import pytest import torch from obliteratus.analysis.riemannian_manifold import ( @@ -687,6 +688,57 @@ class TestSpectralCertification: assert result.confidence < 0.95 assert "inconclusive" in result.recommendation.lower() + def test_qwen_width_dual_space_case_is_inconclusive_not_zero_threshold_red(self): + """The production n=40, d=5120 regime cannot self-certify via signal size.""" + torch.manual_seed(42) + harmful = torch.randn(20, 5120) * 0.3 + harmless = torch.randn(20, 5120) * 0.3 + + result = SpectralCertifier().certify(harmful, harmless) + + assert result.level == CertificationLevel.INCONCLUSIVE + assert result.diagnostic_reason == "insufficient_samples" + assert result.effective_noise_rank > 1 + assert math.isfinite(result.bbp_threshold) + assert result.bbp_threshold > 0 + assert result.n_samples_required >= math.ceil(math.sqrt(5120)) + + def test_zero_noise_scale_is_machine_readable_inconclusive(self): + harmful = torch.ones(30, 32) + harmless = torch.ones(30, 32) + + result = SpectralCertifier().certify(harmful, harmless) + + assert result.level == CertificationLevel.INCONCLUSIVE + assert result.diagnostic_reason == "zero_or_non_finite_noise_scale" + assert result.effective_noise_rank == 0 + assert result.is_sample_sufficient is False + + def test_non_finite_activations_are_machine_readable_inconclusive(self): + harmful = torch.randn(30, 32) + harmless = torch.randn(30, 32) + harmful[0, 0] = float("nan") + + result = SpectralCertifier().certify(harmful, harmless) + + assert result.level == CertificationLevel.INCONCLUSIVE + assert result.diagnostic_reason == "non_finite_activations" + assert result.is_sample_sufficient is False + assert math.isfinite(result.bbp_threshold) + assert math.isfinite(result.leading_eigenvalue) + + @pytest.mark.parametrize( + ("harmful", "harmless", "message"), + [ + (torch.zeros(2, 2, 2), torch.zeros(2, 2), "2-D"), + (torch.zeros(2, 3), torch.zeros(2, 4), "hidden width"), + (torch.zeros(1, 3), torch.zeros(2, 3), "two samples"), + ], + ) + def test_invalid_activation_shapes_fail_closed(self, harmful, harmless, message): + with pytest.raises(ValueError, match=message): + SpectralCertifier().certify(harmful, harmless) + def test_overall_prefers_sufficient_red_over_inconclusive(self): """A reliable RED layer remains actionable in a mixed result set.""" torch.manual_seed(42) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 9f9fd06..6e617fe 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -231,6 +231,22 @@ class TestBuildReport: "refusal_rate": "measured", } + def test_spectral_inconclusive_reason_and_rank_are_public_metrics(self): + report = build_report(**self._base_kwargs(quality_metrics={ + "spectral_certification": "INCONCLUSIVE", + "spectral_diagnostic_reason": "insufficient_samples", + "spectral_effective_noise_rank": 38, + "spectral_samples_required": 72, + })) + + assert report["quality_metrics"] == { + "spectral_certification": "INCONCLUSIVE", + "spectral_diagnostic_reason": "insufficient_samples", + "spectral_effective_noise_rank": 38.0, + "spectral_samples_required": 72.0, + } + assert set(report["quality_metric_status"].values()) == {"measured"} + def test_public_payload_redacts_paths_tokens_and_secret_keys(self): report = build_report(**self._base_kwargs( architecture="/private/models/LlamaForCausalLM",