From 51529ad1d71709521c34d3efd728347f3fa154ce Mon Sep 17 00:00:00 2001 From: Joseph Magly <1159087+jmagly@users.noreply.github.com> Date: Sat, 15 Aug 2026 11:16:37 -0400 Subject: [PATCH] test: add Gate 3 numerical oracle contracts --- .aiwg/testing/gate3-increment-2-report.md | 61 ++ .aiwg/testing/master-test-plan.md | 6 +- .github/workflows/ci.yml | 32 +- ci/test-quality-policy.json | 2 +- ci/test-risk-map.json | 2 + obliteratus/abliterate.py | 175 +---- obliteratus/analysis/numerical_contracts.py | 195 +++++ obliteratus/cli.py | 43 +- obliteratus/prompts.py | 92 +++ pyproject.toml | 8 + scripts/check_mutation_score.py | 5 +- scripts/check_mutation_targets.py | 224 ++++++ scripts/check_quality_policy.py | 2 +- .../sitecustomize.py | 60 ++ scripts/prepare_mutation_coverage.py | 451 +++++++++++ scripts/run_prepared_mutmut.py | 43 ++ tests/test_cli.py | 200 +++++ tests/test_mutmut_coverage_sitecustomize.py | 110 +++ tests/test_numerical_contracts.py | 8 + tests/test_offline_integration.py | 85 ++- tests/test_projection_math_contracts.py | 711 ++++++++++++++++++ tests/test_prompt_sources.py | 110 +++ tests/test_quality_gate_scripts.py | 77 ++ tests/test_quality_policy.py | 539 ++++++++++++- tests/test_whitened_svd_oracles.py | 372 +++++++++ 25 files changed, 3443 insertions(+), 170 deletions(-) create mode 100644 .aiwg/testing/gate3-increment-2-report.md create mode 100644 scripts/check_mutation_targets.py create mode 100644 scripts/mutmut_coverage_sitecustomize/sitecustomize.py create mode 100644 scripts/prepare_mutation_coverage.py create mode 100644 scripts/run_prepared_mutmut.py create mode 100644 tests/test_mutmut_coverage_sitecustomize.py create mode 100644 tests/test_projection_math_contracts.py create mode 100644 tests/test_whitened_svd_oracles.py diff --git a/.aiwg/testing/gate3-increment-2-report.md b/.aiwg/testing/gate3-increment-2-report.md new file mode 100644 index 0000000..d9241f2 --- /dev/null +++ b/.aiwg/testing/gate3-increment-2-report.md @@ -0,0 +1,61 @@ +# Gate 3 increment 2 numerical-contract report + +Date: 2026-08-15 +Exact base: `369417fbb45d25e544a72246abb05e21d1ac00e0` +Evidence scope: the containing commit, resolved with `git rev-parse HEAD` +Canonical merge: pending +Status: PASS locally; hosted exact-head audit and merge pending + +## Results + +| Gate | Result | +|---|---| +| Mandatory CPU matrix | Python 3.10, 3.11, and 3.12: 1,789 passed and 6 conditional skips on each interpreter; 0 failures | +| Repository coverage | 76.48% statements; 63.00% branches | +| Mature CPU scope | 93.05% statements; 82.41% branches | +| Changed executable lines | 222/222 (100%) against the exact base | +| Touched production modules | 4/4 pass exact-base line and branch no-regression | +| Numerical seam | `numerical_contracts.py`: 100% statements and 100% branches | +| Whitened SVD | `whitened_svd.py`: 98.21% statements and 93.33% branches | +| Selective mutation | 1,325/1,538 killed (86.15%); 208 survived; 5 timed out; 0 no-test or segfault results | +| Mutation runtime | 1,296.49s for covered-line preparation, stats preparation, runaway-mutant proof, and full execution; 1,235,596 KiB maximum parent RSS | +| Installed offline vertical slice | Five real prompt pairs exercise installed CLI mutation, checkpoint metadata/state change, reload, installed reporting, and JSON/CSV output without network or accelerators | + +## Contracts added + +1. Independent float64 reference oracles cover row orthogonalization, harmless + principal-component removal, shield-atom residualization, coefficient + selection, and standard/transposed weight projection. +2. Metamorphic checks cover projection idempotence, direction-scale invariance, + orthogonality, norm behavior, unsupported layouts, non-finite coefficients, + zero/tiny boundaries, integer restoration, and half/bfloat stable compute. +3. Whitened SVD is checked against independent covariance/eigendecomposition/SVD + oracles, activation-shape validation, degenerate variance, permutation/sign + behavior, dtype promotion, layer ordering, and deterministic CPU behavior. +4. A bounded `--prompt-pairs-file` seam permits installed offline integration + without weakening the built-in 842-pair default. The loader rejects malformed, + oversized, non-UTF-8, NUL-containing, blank, unequal, and undersized inputs. +5. Required mutation targets fail closed when absent, stale, empty, or mapped to + no tests. No-test exit-code interpretation is pinned to mutmut 3.7.0 while + ordinary metadata validation remains portable to the Python 3.10 test job. +6. Mutation execution uses three fresh processes: covered-line preparation, + function/test stats preparation, then execution-only mutmut. Clean and + forced-fail preflights run in subprocesses so the fork parent never inherits + pytest/Torch teardown state. +7. The deep gate uses eight single-threaded workers, a 30-second minimum wall + allowance (`timeout_constant = 2.0`, mutmut multiplier 15), and a 45-minute + hosted hard cap. A known infinite-growth mutant was independently replayed + and bounded before the complete campaign. + +## Scope and release decision + +This increment implements Gate 3 bounded delivery item 2. It does not reopen +ordinary feature work: items 3–7, conditional-evidence reconciliation, and the +final canonical Gate 3 publication remain open. CUDA, MPS, MLX, network, +download, remote, and operator-UI mappings were not changed; their six local +skips remain governed by item 8 rather than being treated as CPU-gate failures. + +The mutation artifacts are preserved in +`/tmp/obliteratus-gate3-merge-proof-OyhvE4/quality-evidence/` for the local +audit. Hosted CI artifacts become authoritative after exact-head review and +merge. diff --git a/.aiwg/testing/master-test-plan.md b/.aiwg/testing/master-test-plan.md index 89a2350..ef685a5 100644 --- a/.aiwg/testing/master-test-plan.md +++ b/.aiwg/testing/master-test-plan.md @@ -362,8 +362,10 @@ single repository-wide coverage PR. measurements, and add duration-budget enforcement without changing production behavior. Exact canonical evidence: https://github.com/elder-plinius/OBLITERATUS/actions/runs/31874768085. -2. Add numerical/reference-oracle and metamorphic contracts; expand mutation to - the corresponding pure math. +2. **Implemented in Gate 3 increment 2; canonical merge pending.** Add + numerical/reference-oracle and metamorphic contracts; expand mutation to the + corresponding pure math. Local evidence is recorded in + `gate3-increment-2-report.md`. 3. Add loader, architecture, dtype, quantization, and shared-weight decision contracts with mutation evidence. 4. Add checkpoint failure injection, concurrency, atomicity, and retry contracts. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b616250..2ba76a6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -202,6 +202,10 @@ jobs: scripts/check_coverage_thresholds.py scripts/check_mutation_score.py scripts/check_quality_policy.py + scripts/check_mutation_targets.py + scripts/prepare_mutation_coverage.py + scripts/run_prepared_mutmut.py + scripts/mutmut_coverage_sitecustomize/sitecustomize.py scripts/check_conditional_policy.py scripts/check_test_risk_map.py scripts/conditional_gate_summary.py @@ -218,6 +222,10 @@ jobs: scripts/check_coverage_thresholds.py scripts/check_mutation_score.py scripts/check_quality_policy.py + scripts/check_mutation_targets.py + scripts/prepare_mutation_coverage.py + scripts/run_prepared_mutmut.py + scripts/mutmut_coverage_sitecustomize/sitecustomize.py scripts/check_conditional_policy.py scripts/check_test_risk_map.py scripts/conditional_gate_summary.py @@ -388,7 +396,7 @@ jobs: quality-depth: name: Quality depth runs-on: ubuntu-latest - timeout-minutes: 20 + timeout-minutes: 45 env: CUDA_VISIBLE_DEVICES: "" HF_DATASETS_OFFLINE: "1" @@ -424,18 +432,32 @@ jobs: --output quality-evidence/repeat-gate.json - name: Run bounded selective mutation gate + env: + BLIS_NUM_THREADS: "1" + MKL_NUM_THREADS: "1" + NUMEXPR_NUM_THREADS: "1" + OMP_THREAD_LIMIT: "1" + OMP_NUM_THREADS: "1" + OPENBLAS_NUM_THREADS: "1" + VECLIB_MAXIMUM_THREADS: "1" run: | + "$QUALITY_ENV/bin/python" scripts/check_mutation_targets.py prepare + # shellcheck disable=SC2016 /usr/bin/time -f 'elapsed_seconds=%e\nmax_rss_kb=%M' \ -o quality-evidence/mutation-time.txt \ - "$QUALITY_ENV/bin/python" -c \ - 'import torch, yaml; from mutmut.__main__ import cli; cli()' \ - run --max-children 4 + bash -euo pipefail -c ' + "$QUALITY_ENV/bin/python" scripts/prepare_mutation_coverage.py prepare-coverage --max-children 8 + "$QUALITY_ENV/bin/python" scripts/prepare_mutation_coverage.py prepare-stats --max-children 8 + PATH="$QUALITY_ENV/bin:$PATH" \ + "$QUALITY_ENV/bin/python" scripts/run_prepared_mutmut.py run --max-children 8 + ' + "$QUALITY_ENV/bin/python" scripts/check_mutation_targets.py check "$QUALITY_ENV/bin/mutmut" results > quality-evidence/mutation-survivors.txt "$QUALITY_ENV/bin/mutmut" export-cicd-stats cp mutants/mutmut-cicd-stats.json quality-evidence/mutation-stats.json cp mutants/mutmut-stats.json quality-evidence/mutation-test-selection.json "$QUALITY_ENV/bin/python" scripts/check_mutation_score.py \ - quality-evidence/mutation-stats.json --minimum 75 + quality-evidence/mutation-stats.json --minimum 85.0 - name: Write normalized quality trend evidence if: always() diff --git a/ci/test-quality-policy.json b/ci/test-quality-policy.json index 2ca1218..94cf419 100644 --- a/ci/test-quality-policy.json +++ b/ci/test-quality-policy.json @@ -6,7 +6,7 @@ "changed_line": 95.0, "mature_cpu_statement": 92.0, "mature_cpu_branch": 80.0, - "mutation_score": 75.0, + "mutation_score": 85.0, "warning_budget": 0 }, "critical_cpu_paths": [ diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json index 3c9e91d..91fc588 100644 --- a/ci/test-risk-map.json +++ b/ci/test-risk-map.json @@ -185,6 +185,8 @@ "tests/test_new_analysis_modules.py", "tests/test_numerical_contracts.py", "tests/test_novel_analysis.py", + "tests/test_projection_math_contracts.py", + "tests/test_whitened_svd_oracles.py", "tests/test_visualization.py" ] }, diff --git a/obliteratus/abliterate.py b/obliteratus/abliterate.py index 0817451..1bf833d 100644 --- a/obliteratus/abliterate.py +++ b/obliteratus/abliterate.py @@ -41,6 +41,13 @@ from obliteratus import device as dev # noqa: E402 — must import before CUDA dev.configure_cuda_alloc() from obliteratus.models.loader import ModelHandle, load_model # noqa: E402 +from obliteratus.analysis.numerical_contracts import ( # noqa: E402 + orthogonalize_subspace_rows, + project_weight_against_direction, + remove_harmless_principal_components, + residualize_against_shield_atoms, + select_projection_coefficients, +) from obliteratus.persistence_contracts import ( # noqa: E402 atomic_checkpoint_directory as _atomic_checkpoint_directory, ensure_checkpoint_capacity, @@ -2642,15 +2649,7 @@ class AbliterationPipeline: Returns: Orthonormalized subspace tensor with the same shape. """ - if sub.shape[0] <= 1: - return sub - # QR on the transpose: sub^T = Q @ R, then Q^T has orthonormal rows - Q, _ = torch.linalg.qr(sub.T) - result = Q[:, :sub.shape[0]].T # (k, hidden_dim) - # Ensure row 0 points in the same direction as original - if (result[0] @ sub[0]) < 0: - result[0] = -result[0] - return result + return orthogonalize_subspace_rows(sub) def _remove_harmless_principal_components( self, @@ -2659,32 +2658,7 @@ class AbliterationPipeline: pc_count: int, ) -> torch.Tensor: """Subtract dominant benign activation PCs from refusal directions.""" - if pc_count <= 0 or harmless_stack.shape[0] < 3 or subspace.numel() == 0: - return subspace - - centered = harmless_stack.float() - harmless_stack.float().mean(dim=0, keepdim=True) - try: - _, _, Vh = torch.linalg.svd(centered, full_matrices=False) - except Exception: - return subspace - - k = min(int(pc_count), Vh.shape[0], subspace.shape[1]) - if k <= 0: - return subspace - - original = subspace.float() - pcs = Vh[:k] - residual = original - (original @ pcs.T) @ pcs - row_norms = residual.norm(dim=-1, keepdim=True) - near_zero = row_norms.squeeze(-1) < 1e-8 - if near_zero.any(): - residual[near_zero] = original[near_zero] - row_norms = residual.norm(dim=-1, keepdim=True) - - residual = residual / row_norms.clamp(min=1e-8) - if residual.shape[0] > 1: - residual = self._orthogonalize_subspace(residual) - return residual.to(dtype=subspace.dtype, device=subspace.device) + return remove_harmless_principal_components(subspace, harmless_stack, pc_count) def _residualize_against_shield_atoms( self, @@ -2693,30 +2667,7 @@ class AbliterationPipeline: ridge: float, ) -> torch.Tensor: """Remove protected concept atoms with ridge-regularized projection.""" - if atoms.numel() == 0 or subspace.numel() == 0: - return subspace - - original = subspace.float() - A = atoms.float() - A = A / A.norm(dim=-1, keepdim=True).clamp(min=1e-8) - gram = A @ A.T - eye = torch.eye(gram.shape[0], dtype=gram.dtype, device=gram.device) - try: - coeff = torch.linalg.solve(gram + float(ridge) * eye, A @ original.T) - except Exception: - return subspace - - residual = original - coeff.T @ A - row_norms = residual.norm(dim=-1, keepdim=True) - near_zero = row_norms.squeeze(-1) < 1e-8 - if near_zero.any(): - residual[near_zero] = original[near_zero] - row_norms = residual.norm(dim=-1, keepdim=True) - - residual = residual / row_norms.clamp(min=1e-8) - if residual.shape[0] > 1: - residual = self._orthogonalize_subspace(residual) - return residual.to(dtype=subspace.dtype, device=subspace.device) + return residualize_against_shield_atoms(subspace, atoms, ridge) @staticmethod def _select_layers_knee(sorted_layers: list[tuple[int, float]]) -> list[int]: @@ -4855,25 +4806,7 @@ class AbliterationPipeline: projection_row_fraction: float, ) -> torch.Tensor: """Keep only the strongest projection coefficients when requested.""" - if not 0.0 < projection_row_fraction <= 1.0: - raise ValueError("projection_row_fraction must be in (0.0, 1.0]") - if projection_row_fraction >= 1.0: - return coeff - - flat = coeff.detach().abs().reshape(-1).float().cpu() - n_coeffs = flat.numel() - if n_coeffs == 0: - return coeff - - keep = max(1, min(n_coeffs, math.ceil(n_coeffs * projection_row_fraction))) - if keep >= n_coeffs: - return coeff - - idx = torch.topk(flat, keep, sorted=False).indices - mask = torch.zeros(n_coeffs, dtype=torch.bool) - mask[idx] = True - mask = mask.reshape(coeff.shape).to(device=coeff.device) - return coeff * mask.to(dtype=coeff.dtype) + return select_projection_coefficients(coeff, projection_row_fraction) @staticmethod def _project_out_advanced( @@ -4901,7 +4834,6 @@ class AbliterationPipeline: dequantizes before projection, re-quantizing afterward. Without this, in-place operations on packed NF4 storage are silent no-ops. """ - scale = 1.0 - regularization count = 0 for name in candidate_names: @@ -4910,81 +4842,22 @@ class AbliterationPipeline: continue W, is_quantized = AbliterationPipeline._dequantize_weight(proj) - d = direction.to(device=W.device, dtype=W.dtype) - - # Skip projection if weight or direction contains NaN/Inf - if not torch.isfinite(W).all() or not torch.isfinite(d).all(): + result = project_weight_against_direction( + W, + direction, + norm_preserve=norm_preserve, + regularization=regularization, + projection_row_fraction=projection_row_fraction, + max_norm_ratio=_MAX_NORM_RATIO, + ) + if not result.projected: continue - if W.shape[-1] == d.shape[0]: - # Standard Linear: W is (out_features, hidden_dim) - original_norm_sq = W.pow(2).sum().item() if norm_preserve else 0.0 + W.copy_(result.weight) + if is_quantized: + AbliterationPipeline._replace_quantized_weight(proj, W) - coeff = W @ d # (out_features, 1) - # Guard: if projection coefficient is NaN, skip this weight - if not torch.isfinite(coeff).all(): - del coeff - continue - coeff_to_remove = AbliterationPipeline._select_projection_coefficients( - coeff, projection_row_fraction, - ) - coeff_norm_sq = ( - coeff_to_remove.pow(2).sum().item() if norm_preserve else 0.0 - ) - W.sub_(d.T * (scale * coeff_to_remove)) # in-place rank-1 update - del coeff, coeff_to_remove - - # Analytical norm: ||W'||² = ||W||² - scale(2-scale)||coeff||² - if norm_preserve and original_norm_sq > 0: - new_norm_sq = max(0.0, original_norm_sq - scale * (2 - scale) * coeff_norm_sq) - if new_norm_sq > 0: - import math - ratio = math.sqrt(original_norm_sq / new_norm_sq) - # Cap amplification: uncapped rescaling compounds - # across layers and directions, destroying coherence. - # 1.10 keeps per-projection drift bounded while - # allowing legitimate norm preservation. - if ratio > _MAX_NORM_RATIO: - ratio = _MAX_NORM_RATIO - W.mul_(ratio) - - if is_quantized: - AbliterationPipeline._replace_quantized_weight(proj, W) - - count += 1 - - elif W.shape[0] == d.shape[0]: - # Transposed (e.g. GPT-2 Conv1D): W is (hidden_dim, out_features) - original_norm_sq = W.pow(2).sum().item() if norm_preserve else 0.0 - - coeff = d.T @ W # (1, out_features) - # Guard: if projection coefficient is NaN, skip this weight - if not torch.isfinite(coeff).all(): - del coeff - continue - coeff_to_remove = AbliterationPipeline._select_projection_coefficients( - coeff, projection_row_fraction, - ) - coeff_norm_sq = ( - coeff_to_remove.pow(2).sum().item() if norm_preserve else 0.0 - ) - W.sub_((scale * d) * coeff_to_remove) # in-place rank-1 update - del coeff, coeff_to_remove - - # Analytical norm: ||W'||² = ||W||² - scale(2-scale)||coeff||² - if norm_preserve and original_norm_sq > 0: - new_norm_sq = max(0.0, original_norm_sq - scale * (2 - scale) * coeff_norm_sq) - if new_norm_sq > 0: - import math - ratio = math.sqrt(original_norm_sq / new_norm_sq) - if ratio > _MAX_NORM_RATIO: - ratio = _MAX_NORM_RATIO - W.mul_(ratio) - - if is_quantized: - AbliterationPipeline._replace_quantized_weight(proj, W) - - count += 1 + count += 1 return count diff --git a/obliteratus/analysis/numerical_contracts.py b/obliteratus/analysis/numerical_contracts.py index cdc6f4a..a809da9 100644 --- a/obliteratus/analysis/numerical_contracts.py +++ b/obliteratus/analysis/numerical_contracts.py @@ -3,6 +3,30 @@ from __future__ import annotations import math +from dataclasses import dataclass + +import torch + + +@dataclass(frozen=True) +class ProjectionResult: + """Pure rank-1 projection result and metadata.""" + + weight: torch.Tensor + projected: bool + coefficient_norm_sq: float + layout: str | None + + +def _stable_float_dtype(*tensors: torch.Tensor) -> torch.dtype: + """Select a supported compute dtype without downcasting float64 inputs.""" + if any(tensor.dtype == torch.float64 for tensor in tensors): + return torch.float64 + if any(tensor.dtype in {torch.float16, torch.bfloat16} for tensor in tensors): + return torch.float32 + if all(tensor.is_floating_point() for tensor in tensors): + return tensors[0].dtype + return torch.float32 def validate_whitened_parameters( @@ -45,3 +69,174 @@ def validate_whitened_request( if n_directions <= 0: raise ValueError("n_directions must be a positive integer") return n_directions + + +def orthogonalize_subspace_rows(subspace: torch.Tensor) -> torch.Tensor: + """Orthogonalize rows of a subspace matrix with QR while preserving dtype/device.""" + if subspace.shape[0] <= 1 or subspace.numel() == 0: + return subspace + if not torch.isfinite(subspace).all(): + return subspace + compute_dtype = _stable_float_dtype(subspace) + work = subspace.to(dtype=compute_dtype) + if work.norm() < 1e-8: + return torch.zeros_like(subspace) + + q, _ = torch.linalg.qr(work.T) + result = q[:, : subspace.shape[0]].T + if (result[0] @ work[0]) < 0: + result[0] = -result[0] + return result.to(dtype=subspace.dtype, device=subspace.device) + + +def remove_harmless_principal_components( + subspace: torch.Tensor, + harmless_stack: torch.Tensor, + pc_count: int, +) -> torch.Tensor: + """Subtract dominant benign activation PCs from refusal directions.""" + if pc_count <= 0 or harmless_stack.shape[0] < 3 or subspace.numel() == 0: + return subspace + + compute_dtype = _stable_float_dtype(subspace, harmless_stack) + harmless_work = harmless_stack.to(dtype=compute_dtype) + centered = harmless_work - harmless_work.mean(dim=0, keepdim=True) + try: + _, _, vh = torch.linalg.svd(centered, full_matrices=False) + except Exception: + return subspace + + k = min(int(pc_count), vh.shape[0], subspace.shape[1]) + if k <= 0: + return subspace + + original = subspace.to(dtype=compute_dtype) + pcs = vh[:k] + residual = original - (original @ pcs.T) @ pcs + row_norms = residual.norm(dim=-1, keepdim=True) + near_zero = row_norms.squeeze(-1) < 1e-8 + if near_zero.any(): + residual[near_zero] = original[near_zero] + row_norms = residual.norm(dim=-1, keepdim=True) + + residual = residual / row_norms.clamp(min=1e-8) + if residual.shape[0] > 1: + residual = orthogonalize_subspace_rows(residual) + return residual.to(dtype=subspace.dtype, device=subspace.device) + + +def residualize_against_shield_atoms( + subspace: torch.Tensor, + atoms: torch.Tensor, + ridge: float, +) -> torch.Tensor: + """Remove protected concept atoms with ridge-regularized projection.""" + if atoms.numel() == 0 or subspace.numel() == 0: + return subspace + + compute_dtype = _stable_float_dtype(subspace, atoms) + original = subspace.to(dtype=compute_dtype) + normalized_atoms = atoms.to(dtype=compute_dtype) + normalized_atoms = normalized_atoms / normalized_atoms.norm(dim=-1, keepdim=True).clamp(min=1e-8) + gram = normalized_atoms @ normalized_atoms.T + eye = torch.eye(gram.shape[0], dtype=gram.dtype, device=gram.device) + try: + coeff = torch.linalg.solve(gram + float(ridge) * eye, normalized_atoms @ original.T) + except Exception: + return subspace + + residual = original - coeff.T @ normalized_atoms + row_norms = residual.norm(dim=-1, keepdim=True) + near_zero = row_norms.squeeze(-1) < 1e-8 + if near_zero.any(): + residual[near_zero] = original[near_zero] + row_norms = residual.norm(dim=-1, keepdim=True) + + residual = residual / row_norms.clamp(min=1e-8) + if residual.shape[0] > 1: + residual = orthogonalize_subspace_rows(residual) + return residual.to(dtype=subspace.dtype, device=subspace.device) + + +def select_projection_coefficients( + coeff: torch.Tensor, + projection_row_fraction: float, +) -> torch.Tensor: + """Keep only the strongest projection coefficients when selective projection is requested.""" + if not 0.0 < projection_row_fraction <= 1.0: + raise ValueError("projection_row_fraction must be in (0.0, 1.0]") + if projection_row_fraction >= 1.0: + return coeff + + flat = coeff.detach().abs().reshape(-1).float().cpu() + n_coeffs = flat.numel() + if n_coeffs == 0: + return coeff + + keep = max(1, min(n_coeffs, math.ceil(n_coeffs * projection_row_fraction))) + if keep >= n_coeffs: + return coeff + + idx = torch.topk(flat, keep, sorted=False).indices + mask = torch.zeros(n_coeffs, dtype=torch.bool) + mask[idx] = True + mask = mask.reshape(coeff.shape).to(device=coeff.device) + return coeff * mask.to(dtype=coeff.dtype) + + +def project_weight_against_direction( + weight: torch.Tensor, + direction: torch.Tensor, + *, + norm_preserve: bool = False, + regularization: float = 0.0, + projection_row_fraction: float = 1.0, + max_norm_ratio: float = 1.10, +) -> ProjectionResult: + """Return a pure rank-1 projection update for standard or transposed weight layouts.""" + compute_dtype = _stable_float_dtype(weight, direction) + work = weight.to(dtype=compute_dtype) + d = direction.to(device=weight.device, dtype=compute_dtype).reshape(-1, 1) + if not torch.isfinite(work).all() or not torch.isfinite(d).all(): + return ProjectionResult(weight=weight.clone(), projected=False, coefficient_norm_sq=0.0, layout=None) + d_norm = d.norm() + if d_norm < 1e-8: + return ProjectionResult(weight=weight.clone(), projected=False, coefficient_norm_sq=0.0, layout=None) + d = d / d_norm + + scale = 1.0 - regularization + original_norm_sq = work.pow(2).sum().item() if norm_preserve else 0.0 + + if work.shape[-1] == d.shape[0]: + layout = "standard" + coeff = work @ d + if not torch.isfinite(coeff).all(): + return ProjectionResult(weight=weight.clone(), projected=False, coefficient_norm_sq=0.0, layout=layout) + coeff_to_remove = select_projection_coefficients(coeff, projection_row_fraction) + coeff_norm_sq = coeff_to_remove.pow(2).sum().item() if norm_preserve else 0.0 + projected = work - d.T * (scale * coeff_to_remove) + elif work.shape[0] == d.shape[0]: + layout = "transposed" + coeff = d.T @ work + if not torch.isfinite(coeff).all(): + return ProjectionResult(weight=weight.clone(), projected=False, coefficient_norm_sq=0.0, layout=layout) + coeff_to_remove = select_projection_coefficients(coeff, projection_row_fraction) + coeff_norm_sq = coeff_to_remove.pow(2).sum().item() if norm_preserve else 0.0 + projected = work - (scale * d) * coeff_to_remove + else: + return ProjectionResult(weight=weight.clone(), projected=False, coefficient_norm_sq=0.0, layout=None) + + if norm_preserve and original_norm_sq > 0: + new_norm_sq = max(0.0, original_norm_sq - scale * (2 - scale) * coeff_norm_sq) + if new_norm_sq > 0: + ratio = math.sqrt(original_norm_sq / new_norm_sq) + if ratio > max_norm_ratio: + ratio = max_norm_ratio + projected = projected * ratio + + return ProjectionResult( + weight=projected.to(dtype=weight.dtype, device=weight.device), + projected=True, + coefficient_norm_sq=coeff_norm_sq, + layout=layout, + ) diff --git a/obliteratus/cli.py b/obliteratus/cli.py index ae1967e..333dbea 100644 --- a/obliteratus/cli.py +++ b/obliteratus/cli.py @@ -315,7 +315,20 @@ def main(argv: list[str] | None = None): "--dataset", type=str, default="builtin", help="Prompt dataset source for contrastive extraction when using residue mining (default: builtin).", ) - p.add_argument( + prompt_source_group = p.add_mutually_exclusive_group() + prompt_source_group.add_argument( + "--prompt-pairs-file", + "--prompt-pair-file", + dest="prompt_pairs_file", + type=str, + default=None, + metavar="PATH", + help=( + "Local UTF-8 JSON file with exactly 'harmful' and 'harmless' arrays " + "for contrastive extraction." + ), + ) + prompt_source_group.add_argument( "--residue-file", action="append", default=[], help="Refusal-audit/residue JSON to upweight as hard negatives. Can be passed multiple times.", ) @@ -482,6 +495,20 @@ def main(argv: list[str] | None = None): args = parser.parse_args(argv) + if getattr(args, "prompt_pairs_file", None): + residue_only_options = [] + if getattr(args, "dataset", "builtin") != "builtin": + residue_only_options.append("--dataset") + if getattr(args, "residue_weight", 5) != 5: + residue_only_options.append("--residue-weight") + if getattr(args, "residue_max", None) is not None: + residue_only_options.append("--residue-max") + if residue_only_options: + parser.error( + f"{', '.join(residue_only_options)} can only be used with --residue-file, " + "not --prompt-pairs-file" + ) + # Apply GPU selection early (before any CUDA init) _apply_gpu_selection(args) @@ -1095,7 +1122,19 @@ def _cmd_abliterate(args): live = None prompt_kwargs = {} residue_meta = None - if getattr(args, "residue_file", None): + if getattr(args, "prompt_pairs_file", None): + from obliteratus.prompts import load_prompt_pairs_file + + try: + harmful, harmless = load_prompt_pairs_file(args.prompt_pairs_file) + except ValueError as exc: + console.print(f"[red]Invalid --prompt-pairs-file:[/] {exc}") + raise SystemExit(2) from exc + prompt_kwargs = {"harmful_prompts": harmful, "harmless_prompts": harmless} + log_lines.append( + f"Loaded explicit prompt pairs: {len(harmful)} harmful + {len(harmless)} harmless." + ) + elif getattr(args, "residue_file", None): from obliteratus.hard_negative import build_weighted_prompt_pairs harmful, harmless, residue_meta = build_weighted_prompt_pairs( diff --git a/obliteratus/prompts.py b/obliteratus/prompts.py index f065e07..6681902 100644 --- a/obliteratus/prompts.py +++ b/obliteratus/prompts.py @@ -10,8 +10,12 @@ dropdown. External datasets are fetched on demand from HuggingFace Hub. from __future__ import annotations +import json import logging +import stat from dataclasses import dataclass +from pathlib import Path +from typing import Any from typing import Callable logger = logging.getLogger(__name__) @@ -23,6 +27,10 @@ logger = logging.getLogger(__name__) _dataset_cache: dict[str, tuple[list[str], list[str]]] = {} +MIN_PROMPT_PAIRS = 5 +MAX_PROMPT_PAIRS = 10_000 +MAX_PROMPT_PAIRS_FILE_BYTES = 1_048_576 + # ── Dataset source registry ───────────────────────────────────────────── @@ -448,6 +456,90 @@ def load_custom_prompts(harmful_text: str, harmless_text: str) -> tuple[list[str return harmful, harmless +def load_prompt_pairs_file(path: str | Path) -> tuple[list[str], list[str]]: + """Load explicit harmful/harmless prompt pairs from a bounded local JSON file. + + The schema is exactly ``{"harmful": [...], "harmless": [...]}``: no extra + top-level keys are accepted. The arrays must have equal length, contain at + least five pairs and at most ``MAX_PROMPT_PAIRS`` pairs, and every item must + be a nonblank UTF-8 string without NUL bytes. + """ + + prompt_path = Path(path).expanduser() + try: + file_stat = prompt_path.stat() + except FileNotFoundError as exc: + raise ValueError(f"Prompt pairs file does not exist: {prompt_path}") from exc + except OSError as exc: + raise ValueError(f"Cannot access prompt pairs file {prompt_path}: {exc}") from exc + + if not stat.S_ISREG(file_stat.st_mode): + raise ValueError(f"Prompt pairs file must be a regular file: {prompt_path}") + if file_stat.st_size > MAX_PROMPT_PAIRS_FILE_BYTES: + raise ValueError( + "Prompt pairs file is too large: " + f"{file_stat.st_size} bytes > {MAX_PROMPT_PAIRS_FILE_BYTES} bytes" + ) + + try: + text = prompt_path.read_text(encoding="utf-8", errors="strict") + except UnicodeDecodeError as exc: + raise ValueError(f"Prompt pairs file must be valid UTF-8: {prompt_path}") from exc + except OSError as exc: + raise ValueError(f"Cannot read prompt pairs file {prompt_path}: {exc}") from exc + + try: + raw = json.loads(text) + except json.JSONDecodeError as exc: + raise ValueError(f"Prompt pairs file is malformed JSON: {exc.msg}") from exc + + if not isinstance(raw, dict): + raise ValueError("Prompt pairs file must contain a JSON object") + if set(raw) != {"harmful", "harmless"}: + raise ValueError( + "Prompt pairs file object must contain exactly 'harmful' and 'harmless' arrays" + ) + + harmful = _validate_prompt_array(raw["harmful"], "harmful") + harmless = _validate_prompt_array(raw["harmless"], "harmless") + if len(harmful) != len(harmless): + raise ValueError( + "Prompt pairs file arrays must have equal length: " + f"harmful={len(harmful)}, harmless={len(harmless)}" + ) + if len(harmful) < MIN_PROMPT_PAIRS: + raise ValueError( + f"Prompt pairs file must contain at least {MIN_PROMPT_PAIRS} pairs, " + f"got {len(harmful)}" + ) + if len(harmful) > MAX_PROMPT_PAIRS: + raise ValueError( + f"Prompt pairs file must contain at most {MAX_PROMPT_PAIRS} pairs, " + f"got {len(harmful)}" + ) + return harmful, harmless + + +def _validate_prompt_array(value: Any, field: str) -> list[str]: + if not isinstance(value, list): + raise ValueError(f"Prompt pairs file field {field!r} must be an array") + prompts: list[str] = [] + for index, item in enumerate(value): + if not isinstance(item, str): + raise ValueError( + f"Prompt pairs file field {field!r} item {index} must be a string" + ) + prompt = item.strip() + if not prompt: + raise ValueError( + f"Prompt pairs file field {field!r} must contain only nonblank strings" + ) + if "\x00" in prompt: + raise ValueError(f"Prompt pairs file field {field!r} item {index} contains NUL") + prompts.append(prompt) + return prompts + + def get_source_choices() -> list[str]: """Return display labels for use in a Gradio dropdown.""" return [s.label for s in DATASET_SOURCES.values()] diff --git a/pyproject.toml b/pyproject.toml index 5d44cc0..7391e5e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -118,24 +118,32 @@ source_paths = ["obliteratus/", "scripts/"] only_mutate = [ "obliteratus/config.py", "obliteratus/analysis/numerical_contracts.py", + "obliteratus/analysis/whitened_svd.py", "obliteratus/runtime_contracts.py", "obliteratus/persistence_contracts.py", "obliteratus/remote_contracts.py", "obliteratus/evaluation/lm_eval_integration.py", "scripts/check_coverage_thresholds.py", ] +required_mutation_targets = [ + "obliteratus/analysis/numerical_contracts.py", + "obliteratus/analysis/whitened_svd.py", +] pytest_add_cli_args = ["--no-cov", "-q"] pytest_add_cli_args_test_selection = [ "tests/test_config.py", "tests/test_config_properties.py", "tests/test_coverage_thresholds.py", + "tests/test_projection_math_contracts.py", "tests/test_lm_eval_reporting_contracts.py", "tests/test_numerical_contracts.py", "tests/test_persistence_contracts.py", "tests/test_remote_contracts.py", "tests/test_runtime_contracts.py", + "tests/test_whitened_svd_oracles.py", ] mutate_only_covered_lines = true +timeout_constant = 2.0 on_dependency_change = "rerun" [tool.uv] diff --git a/scripts/check_mutation_score.py b/scripts/check_mutation_score.py index c36f2c2..2583cf0 100644 --- a/scripts/check_mutation_score.py +++ b/scripts/check_mutation_score.py @@ -10,6 +10,9 @@ from pathlib import Path from typing import Any +DEFAULT_MUTATION_SCORE_MINIMUM = 85.0 + + def mutation_score(stats: dict[str, Any]) -> tuple[int, int, float]: """Return killed, total, and percentage after validating mutmut statistics.""" killed = stats.get("killed") @@ -48,7 +51,7 @@ def validate_mutation_stats( def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("stats", type=Path, help="mutmut-cicd-stats.json path") - parser.add_argument("--minimum", type=float, required=True) + parser.add_argument("--minimum", type=float, default=DEFAULT_MUTATION_SCORE_MINIMUM) return parser diff --git a/scripts/check_mutation_targets.py b/scripts/check_mutation_targets.py new file mode 100644 index 0000000..03d0916 --- /dev/null +++ b/scripts/check_mutation_targets.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python3 +"""Fail closed when configured mutmut targets produce no mutants.""" + +from __future__ import annotations + +import argparse +from importlib import metadata +import json +from pathlib import Path +from typing import Any + +try: # Python 3.11+ + import tomllib +except ModuleNotFoundError: # pragma: no cover - exercised on Python 3.10 + import tomli as tomllib # type: ignore[no-redef] + + +GLOB_MARKERS = frozenset("*?[") +SUPPORTED_MUTMUT_VERSION = "3.7.0" +MUTMUT_NO_TEST_EXIT_CODES = frozenset({5, 33}) + + +def load_mutmut_config(pyproject: Path) -> dict[str, Any]: + """Return the `[tool.mutmut]` table from pyproject.toml.""" + data = tomllib.loads(pyproject.read_text(encoding="utf-8")) + mutmut_config = data.get("tool", {}).get("mutmut", {}) + if not isinstance(mutmut_config, dict): + raise ValueError("[tool.mutmut] must be a table") + return mutmut_config + + +def exact_python_targets(mutmut_config: dict[str, Any]) -> list[Path]: + """Return exact `.py` paths that must enumerate mutants.""" + raw_targets = mutmut_config.get("required_mutation_targets", mutmut_config.get("only_mutate", [])) + if not isinstance(raw_targets, list) or not all(isinstance(path, str) for path in raw_targets): + raise ValueError( + "[tool.mutmut].required_mutation_targets must be a list of strings", + ) + + targets: list[Path] = [] + strict_required = "required_mutation_targets" in mutmut_config + for raw_path in raw_targets: + path = Path(raw_path) + invalid = ( + path.is_absolute() + or ".." in path.parts + or not raw_path.endswith(".py") + or any(marker in raw_path for marker in GLOB_MARKERS) + ) + if invalid: + if strict_required: + raise ValueError(f"invalid required mutation target: {raw_path}") + continue + targets.append(path) + return sorted(targets, key=str) + + +def _owned_mutants_dir(project_root: Path, mutants_dir: Path) -> Path: + if mutants_dir != Path("mutants"): + raise ValueError( + "mutants directory must be project-owned as the literal project-owned mutants directory", + ) + root = project_root.resolve() + configured = root / "mutants" + if configured.is_symlink(): + raise ValueError( + "mutants directory must be project-owned as the literal project-owned mutants directory", + ) + return configured + + +def _owned_artifact(project_root: Path, mutants_dir: Path, target: Path, suffix: str) -> Path: + configured = _owned_mutants_dir(project_root, mutants_dir) + artifact = configured / f"{target}{suffix}" + resolved = artifact.resolve() if artifact.exists() else artifact.resolve(strict=False) + try: + resolved.relative_to(configured) + except ValueError as exc: + raise ValueError(f"mutation artifact escapes project-owned mutants/: {artifact}") from exc + return artifact + + +def _validate_owned_source(project_root: Path, target: Path) -> Path: + source_path = project_root / target + if not source_path.is_file(): + raise ValueError(f"configured mutation target does not exist: {target}") + try: + source_path.resolve().relative_to(project_root.resolve()) + except ValueError as exc: + raise ValueError(f"configured mutation target escapes project root: {target}") from exc + return source_path + + +def mutant_count(meta_path: Path) -> int: + """Return the number of generated mutants recorded in a mutmut metadata file.""" + try: + metadata = json.loads(meta_path.read_text(encoding="utf-8")) + except FileNotFoundError: + return 0 + except json.JSONDecodeError as exc: + raise ValueError(f"{meta_path} is not valid JSON: {exc}") from exc + + exit_code_by_key = metadata.get("exit_code_by_key") + if not isinstance(exit_code_by_key, dict): + raise ValueError(f"{meta_path} does not contain an exit_code_by_key object") + return len(exit_code_by_key) + + +def _installed_mutmut_version() -> str | None: + try: + return metadata.version("mutmut") + except metadata.PackageNotFoundError: + return None + + +def assert_supported_mutmut_no_test_codes() -> None: + version = _installed_mutmut_version() + if version != SUPPORTED_MUTMUT_VERSION: + raise ValueError( + f"unsupported mutmut version {version!r}; expected {SUPPORTED_MUTMUT_VERSION} " + "before interpreting no-test exit codes", + ) + + +def no_test_mutants(meta_path: Path) -> list[str]: + """Return required-target mutants that mutmut marked as having no covering tests.""" + try: + metadata = json.loads(meta_path.read_text(encoding="utf-8")) + except FileNotFoundError: + return [] + except json.JSONDecodeError as exc: + raise ValueError(f"{meta_path} is not valid JSON: {exc}") from exc + + exit_code_by_key = metadata.get("exit_code_by_key") + if not isinstance(exit_code_by_key, dict): + raise ValueError(f"{meta_path} does not contain an exit_code_by_key object") + no_tests = sorted( + key for key, exit_code in exit_code_by_key.items() + if exit_code in MUTMUT_NO_TEST_EXIT_CODES + ) + if no_tests: + assert_supported_mutmut_no_test_codes() + return no_tests + + +def stale_or_empty_targets( + targets: list[Path], *, project_root: Path = Path("."), mutants_dir: Path = Path("mutants"), +) -> list[Path]: + """Return configured targets whose mutant metadata is missing or empty.""" + stale: list[Path] = [] + _owned_mutants_dir(project_root, mutants_dir) + for target in targets: + _validate_owned_source(project_root, target) + if mutant_count(_owned_artifact(project_root, mutants_dir, target, ".meta")) == 0: + stale.append(target) + return stale + + +def prepare_required_targets( + pyproject: Path = Path("pyproject.toml"), *, mutants_dir: Path = Path("mutants"), +) -> list[Path]: + """Remove stale copied mutant files so mutmut regenerates required target metadata.""" + targets = exact_python_targets(load_mutmut_config(pyproject)) + stale = stale_or_empty_targets(targets, project_root=pyproject.parent, mutants_dir=mutants_dir) + for target in stale: + for suffix in ("", ".meta", ".spans"): + artifact = _owned_artifact(pyproject.parent, mutants_dir, target, suffix) + if artifact.exists(): + artifact.unlink() + return stale + + +def validate_required_targets( + pyproject: Path = Path("pyproject.toml"), *, mutants_dir: Path = Path("mutants"), +) -> list[str]: + """Return fail-closed messages for exact mutation targets with zero generated mutants.""" + targets = exact_python_targets(load_mutmut_config(pyproject)) + failures: list[str] = [] + for target in stale_or_empty_targets(targets, project_root=pyproject.parent, mutants_dir=mutants_dir): + failures.append(f"configured mutation target produced zero mutants: {target}") + for target in targets: + no_tests = no_test_mutants(_owned_artifact(pyproject.parent, mutants_dir, target, ".meta")) + if no_tests: + failures.append( + f"configured mutation target has {len(no_tests)} mutant(s) with no tests: {target}", + ) + return failures + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("command", choices=("prepare", "check")) + parser.add_argument("--pyproject", type=Path, default=Path("pyproject.toml")) + parser.add_argument("--mutants-dir", type=Path, default=Path("mutants")) + return parser + + +def main() -> int: + args = _parser().parse_args() + try: + if args.command == "prepare": + stale = prepare_required_targets(args.pyproject, mutants_dir=args.mutants_dir) + if stale: + joined = ", ".join(str(path) for path in stale) + print(f"mutation target guard invalidated stale metadata for: {joined}") + else: + print("mutation target guard found no stale required targets") + return 0 + + failures = validate_required_targets(args.pyproject, mutants_dir=args.mutants_dir) + except (OSError, ValueError) as exc: + print(f"mutation target guard failed: {exc}") + return 1 + + if failures: + for failure in failures: + print(f"mutation target guard failed: {failure}") + return 1 + print("mutation target guard passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/check_quality_policy.py b/scripts/check_quality_policy.py index 4a7dd57..1648f67 100644 --- a/scripts/check_quality_policy.py +++ b/scripts/check_quality_policy.py @@ -18,7 +18,7 @@ BASELINE_FLOORS = { "changed_line": 95.0, "mature_cpu_statement": 92.0, "mature_cpu_branch": 80.0, - "mutation_score": 75.0, + "mutation_score": 85.0, "warning_budget": 0.0, } diff --git a/scripts/mutmut_coverage_sitecustomize/sitecustomize.py b/scripts/mutmut_coverage_sitecustomize/sitecustomize.py new file mode 100644 index 0000000..5731c88 --- /dev/null +++ b/scripts/mutmut_coverage_sitecustomize/sitecustomize.py @@ -0,0 +1,60 @@ +"""Fail-closed mutmut prepared-artifact reuse hooks for CI.""" + +from __future__ import annotations + +import os + + +if ( + os.environ.get("OBLITERATUS_MUTMUT_REUSE_COVERAGE") == "1" + or os.environ.get("OBLITERATUS_MUTMUT_REUSE_STATS") == "1" + or os.environ.get("OBLITERATUS_MUTMUT_SUBPROCESS_PREFLIGHT") == "1" +): + from mutmut import __main__ as mutmut_main + + from scripts.prepare_mutation_coverage import validate_execution_manifest + from scripts.prepare_mutation_coverage import validate_manifest + + def store_lines_covered_by_tests_from_prepared_artifacts() -> None: + validate_manifest() + print("Reusing prepared mutmut covered-line artifacts") + + mutmut_main.store_lines_covered_by_tests = store_lines_covered_by_tests_from_prepared_artifacts + + if os.environ.get("OBLITERATUS_MUTMUT_REUSE_STATS") == "1": + + def collect_or_load_prepared_stats( + runner, + *, + mutants_caught_by_type_checker=None, + apply_config_invalidation=False, + invalidate_stale_callers=True, + ) -> None: + del runner, mutants_caught_by_type_checker + del apply_config_invalidation, invalidate_stale_callers + validate_execution_manifest() + if not mutmut_main.load_stats(): + raise RuntimeError("prepared mutmut stats failed to load") + print("Reusing prepared mutmut test-selection stats") + + mutmut_main.collect_or_load_stats = collect_or_load_prepared_stats + + if os.environ.get("OBLITERATUS_MUTMUT_SUBPROCESS_PREFLIGHT") == "1": + _original_execute_pytest = mutmut_main.PytestRunner.execute_pytest + + def execute_pytest_with_subprocess_preflight(self, params, **kwargs): + if kwargs: + return _original_execute_pytest(self, params, **kwargs) + if os.environ.get("MUTANT_UNDER_TEST") not in {"", "fail"}: + return _original_execute_pytest(self, params, **kwargs) + + import subprocess + import sys + + full_params = ["--rootdir=.", "--tb=native", *params, *self._pytest_add_cli_args] + result = subprocess.run([sys.executable, "-m", "pytest", *full_params], check=False) + if result.returncode == 4: + raise mutmut_main.BadTestExecutionCommandsException(full_params) + return int(result.returncode) + + mutmut_main.PytestRunner.execute_pytest = execute_pytest_with_subprocess_preflight diff --git a/scripts/prepare_mutation_coverage.py b/scripts/prepare_mutation_coverage.py new file mode 100644 index 0000000..eabf27d --- /dev/null +++ b/scripts/prepare_mutation_coverage.py @@ -0,0 +1,451 @@ +#!/usr/bin/env python3 +"""Prepare mutmut artifacts in throwaway processes before execution.""" + +from __future__ import annotations + +import argparse +import hashlib +import inspect +import json +import os +from pathlib import Path +import sys +from typing import Any + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.check_mutation_targets import exact_python_targets +from scripts.check_mutation_targets import load_mutmut_config +from scripts.check_mutation_targets import SUPPORTED_MUTMUT_VERSION + + +MANIFEST_PATH = Path("mutants/.covered-lines-prepass.json") +STATS_PATH = Path("mutants/mutmut-stats.json") +HOOK_HASH_PATHS = ( + Path("scripts/prepare_mutation_coverage.py"), + Path("scripts/run_prepared_mutmut.py"), + Path("scripts/mutmut_coverage_sitecustomize/sitecustomize.py"), +) +META_KEYS = { + "exit_code_by_key", + "hash_by_function_name", + "type_check_error_by_key", + "durations_by_key", + "estimated_durations_by_key", +} + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _validated_relative_file(raw_path: str, *, label: str) -> Path: + path = Path(raw_path) + if path.is_absolute() or ".." in path.parts: + raise RuntimeError(f"{label} must be a project-relative file: {raw_path}") + if not path.is_file(): + raise RuntimeError(f"{label} is missing: {raw_path}") + return path + + +def _hash_relative_files(raw_paths: list[str], *, label: str) -> dict[str, str]: + return { + str(path): _sha256(path) + for path in (_validated_relative_file(raw_path, label=label) for raw_path in raw_paths) + } + + +def _validated_mutation_path(path: Path) -> Path: + if path.is_absolute() or ".." in path.parts: + raise RuntimeError(f"mutation path must be project-relative: {path}") + return path + + +def _owned_mutants_root() -> Path: + root = Path.cwd().resolve() + expected = root / "mutants" + if expected.is_symlink(): + raise RuntimeError( + "mutants directory must be project-owned as the literal project-owned mutants directory", + ) + return expected + + +def _owned_mutation_artifact(path: Path, suffix: str) -> Path: + relative = _validated_mutation_path(path) + root = _owned_mutants_root() + artifact = Path("mutants") / f"{relative}{suffix}" + resolved = artifact.resolve() if artifact.exists() else artifact.resolve(strict=False) + try: + resolved.relative_to(root) + except ValueError as exc: + raise RuntimeError(f"mutation artifact escapes project-owned mutants/: {artifact}") from exc + return artifact + + +def _mutmut_version() -> str: + import mutmut + + return getattr(mutmut, "__version__", "") + + +def assert_supported_mutmut() -> None: + """Fail closed if the mutmut internals this script calls have changed.""" + from mutmut import __main__ as mutmut_main + + if _mutmut_version() != SUPPORTED_MUTMUT_VERSION: + raise RuntimeError( + f"unsupported mutmut version {_mutmut_version()!r}; expected {SUPPORTED_MUTMUT_VERSION}", + ) + + expected = { + "copy_src_dir": ("() -> 'None'",), + "copy_also_copy_files": ("() -> 'None'",), + "setup_source_paths": ("() -> 'None'",), + "create_mutants": ("(max_children: 'int') -> 'MutantGenerationStats'",), + "collect_or_load_stats": ( + "(runner: 'TestRunner', *, mutants_caught_by_type_checker: 'dict[str, Any] | None' = None, " + "apply_config_invalidation: 'bool' = False, invalidate_stale_callers: 'bool' = True) -> 'None'", + "(runner, *, mutants_caught_by_type_checker=None, apply_config_invalidation=False, " + "invalidate_stale_callers=True) -> 'None'", + ), + "load_stats": ("() -> 'bool'",), + } + for name, signatures in expected.items(): + actual = str(inspect.signature(getattr(mutmut_main, name))) + if actual not in signatures: + raise RuntimeError(f"mutmut internal {name} signature changed: {actual}") + + +def configured_paths() -> tuple[list[Path], list[Path]]: + """Return all mutatable paths and exact required paths from the loaded config.""" + from mutmut import __main__ as mutmut_main + from mutmut.configuration import Config + + Config.ensure_loaded() + if not Config.get().mutate_only_covered_lines: + raise RuntimeError("[tool.mutmut].mutate_only_covered_lines must be true") + + mutatable = sorted(mutmut_main.walk_mutatable_files(), key=str) + required = exact_python_targets(load_mutmut_config(Path("pyproject.toml"))) + missing = [path for path in required if path not in mutatable] + if missing: + joined = ", ".join(str(path) for path in missing) + raise RuntimeError(f"required mutation target is not mutatable by mutmut config: {joined}") + if not mutatable: + raise RuntimeError("mutmut configuration produced no mutatable files") + return mutatable, required + + +def remove_mutation_artifacts(paths: list[Path]) -> None: + """Remove generated files whose freshness matters for covered-line reuse.""" + for path in paths: + for suffix in ("", ".meta", ".spans"): + artifact = _owned_mutation_artifact(path, suffix) + if artifact.exists(): + artifact.unlink() + if MANIFEST_PATH.exists(): + MANIFEST_PATH.unlink() + + +def _read_meta(path: Path) -> dict[str, Any]: + meta_path = Path("mutants") / f"{path}.meta" + try: + metadata = json.loads(meta_path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise RuntimeError(f"missing mutmut metadata for {path}") from exc + except json.JSONDecodeError as exc: + raise RuntimeError(f"malformed mutmut metadata for {path}: {exc}") from exc + if not isinstance(metadata, dict) or set(metadata) != META_KEYS: + raise RuntimeError(f"unexpected mutmut metadata schema for {path}: {sorted(metadata)}") + if not isinstance(metadata["exit_code_by_key"], dict): + raise RuntimeError(f"mutmut metadata exit_code_by_key is not an object for {path}") + return metadata + + +def _validate_spans(path: Path, metadata: dict[str, Any]) -> None: + from mutmut.mutation.data import MutantLineSpans + + spans = MutantLineSpans.load(path) + if spans is None: + raise RuntimeError(f"missing or unsupported mutmut spans index for {path}") + generated_names = {key.rpartition(".")[2] for key in metadata["exit_code_by_key"]} + missing = sorted(generated_names - set(spans.span_by_function_name)) + if missing: + raise RuntimeError(f"mutmut spans index for {path} is missing mutants: {missing[:5]}") + + +def validate_prepared_artifacts( + *, + mutatable: list[Path], + required: list[Path], + covered_lines: dict[str, set[int]] | None = None, +) -> dict[str, Any]: + """Validate generated artifacts and return a fresh-source manifest.""" + from mutmut.configuration import Config + + source_hashes: dict[str, str] = {} + selected_test_hashes: dict[str, str] = {} + hook_hashes: dict[str, str] = {} + covered_line_counts: dict[str, int] = {} + mutant_counts: dict[str, int] = {} + + for path in mutatable: + source = _validated_mutation_path(Path(path)) + mutant = _owned_mutation_artifact(path, "") + if not source.is_file(): + raise RuntimeError(f"configured mutation source does not exist: {path}") + if not mutant.is_file() or source.stat().st_mtime >= mutant.stat().st_mtime: + raise RuntimeError(f"mutmut artifact for {path} is missing or stale") + metadata = _read_meta(path) + _validate_spans(path, metadata) + source_hashes[str(path)] = _sha256(source) + mutant_counts[str(path)] = len(metadata["exit_code_by_key"]) + if covered_lines is not None: + key = str((Path("mutants") / path).absolute()) + covered_line_counts[str(path)] = len(covered_lines.get(key, set())) + + for path in required: + if mutant_counts.get(str(path), 0) == 0: + raise RuntimeError(f"required mutation target produced zero covered mutants: {path}") + if covered_lines is not None and covered_line_counts.get(str(path), 0) == 0: + raise RuntimeError(f"required mutation target had no covered lines: {path}") + + config = Config.get() + selected_test_hashes = _hash_relative_files( + list(config.pytest_add_cli_args_test_selection), + label="selected mutation test file", + ) + hook_hashes = _hash_relative_files([str(path) for path in HOOK_HASH_PATHS], label="mutation hook") + return { + "version": 1, + "mutmut_version": _mutmut_version(), + "mutate_only_covered_lines": config.mutate_only_covered_lines, + "source_paths": [str(path) for path in config.source_paths], + "only_mutate": list(config.only_mutate), + "pytest_add_cli_args": list(config.pytest_add_cli_args), + "pytest_add_cli_args_test_selection": list(config.pytest_add_cli_args_test_selection), + "required_mutation_targets": [str(path) for path in required], + "mutatable_paths": [str(path) for path in mutatable], + "source_hashes": source_hashes, + "selected_test_hashes": selected_test_hashes, + "hook_hashes": hook_hashes, + "covered_line_counts": covered_line_counts, + "mutant_counts": mutant_counts, + } + + +def prepare(max_children: int) -> dict[str, Any]: + """Collect coverage, generate covered-only mutants, validate them, and write a manifest.""" + import mutmut + from mutmut import __main__ as mutmut_main + from mutmut.code_coverage import gather_coverage + from mutmut.state import state + + assert_supported_mutmut() + mutatable, required = configured_paths() + Path("mutants").mkdir(exist_ok=True) + remove_mutation_artifacts(mutatable) + + mutmut_main.copy_src_dir() + mutmut_main.copy_also_copy_files() + mutmut_main.setup_source_paths() + source_files = list(mutmut_main.walk_source_files()) + covered_lines = gather_coverage(mutmut_main.PytestRunner(), source_files) + mutmut._covered_lines = covered_lines + stats = mutmut_main.create_mutants(max_children) + state().current_function_hashes.clear() + + manifest = validate_prepared_artifacts( + mutatable=mutatable, + required=required, + covered_lines=covered_lines, + ) + manifest["generation_stats"] = { + "mutated": stats.mutated, + "ignored": stats.ignored, + "unmodified": stats.unmodified, + } + MANIFEST_PATH.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return manifest + + +def validate_coverage_manifest() -> dict[str, Any]: + """Validate that a prior prepass still matches the current sources and config.""" + assert_supported_mutmut() + mutatable, required = configured_paths() + try: + manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise RuntimeError(f"missing covered-line prepass manifest: {MANIFEST_PATH}") from exc + except json.JSONDecodeError as exc: + raise RuntimeError(f"malformed covered-line prepass manifest: {exc}") from exc + + expected = validate_prepared_artifacts(mutatable=mutatable, required=required) + comparable_keys = ( + "version", + "mutmut_version", + "mutate_only_covered_lines", + "source_paths", + "only_mutate", + "pytest_add_cli_args", + "pytest_add_cli_args_test_selection", + "required_mutation_targets", + "mutatable_paths", + "source_hashes", + "selected_test_hashes", + "hook_hashes", + ) + for key in comparable_keys: + if manifest.get(key) != expected[key]: + raise RuntimeError(f"covered-line prepass manifest is stale for {key}") + return manifest + + +def validate_manifest() -> dict[str, Any]: + """Backward-compatible name for covered-line manifest validation.""" + return validate_coverage_manifest() + + +def _read_stats() -> dict[str, Any]: + try: + stats = json.loads(STATS_PATH.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise RuntimeError(f"missing prepared mutmut stats: {STATS_PATH}") from exc + except json.JSONDecodeError as exc: + raise RuntimeError(f"malformed prepared mutmut stats: {exc}") from exc + if not isinstance(stats, dict): + raise RuntimeError("prepared mutmut stats root must be an object") + return stats + + +def validate_stats_artifacts() -> dict[str, Any]: + """Validate prepared mutmut test-selection stats and return manifest fields.""" + stats = _read_stats() + expected_keys = { + "tests_by_mangled_function_name", + "duration_by_test", + "stats_time", + "function_hashes", + "function_dependencies", + "config_fingerprint", + "watched_file_hashes", + "git_commit", + } + if set(stats) != expected_keys: + raise RuntimeError(f"unexpected mutmut stats schema: {sorted(stats)}") + + tests_by_function = stats["tests_by_mangled_function_name"] + duration_by_test = stats["duration_by_test"] + function_hashes = stats["function_hashes"] + if not isinstance(tests_by_function, dict) or not tests_by_function: + raise RuntimeError("prepared mutmut stats have no test-to-function mapping") + if not isinstance(duration_by_test, dict) or not duration_by_test: + raise RuntimeError("prepared mutmut stats have no test duration mapping") + if not isinstance(function_hashes, dict) or not function_hashes: + raise RuntimeError("prepared mutmut stats have no function hash baseline") + associated_tests = { + test + for tests in tests_by_function.values() + if isinstance(tests, list) + for test in tests + } + if not associated_tests: + raise RuntimeError("prepared mutmut stats do not associate any tests with mutants") + missing_durations = sorted(associated_tests - set(duration_by_test)) + if missing_durations: + raise RuntimeError( + "prepared mutmut stats reference tests without durations: " + f"{missing_durations[:5]}", + ) + return { + "stats_hash": _sha256(STATS_PATH), + "stats_function_count": len(function_hashes), + "stats_test_count": len(duration_by_test), + "stats_association_count": sum( + len(tests) for tests in tests_by_function.values() if isinstance(tests, list) + ), + } + + +def prepare_stats(max_children: int) -> dict[str, Any]: + """Build mutmut test-selection stats in a fresh process, then exit.""" + from mutmut import __main__ as mutmut_main + from mutmut.state import state + + manifest = validate_coverage_manifest() + if STATS_PATH.exists(): + STATS_PATH.unlink() + + mutmut_main.copy_src_dir() + mutmut_main.copy_also_copy_files() + mutmut_main.setup_source_paths() + mutmut_main.create_mutants(max_children) + runner = mutmut_main.PytestRunner() + runner.prepare_main_test_run() + mutmut_main.collect_or_load_stats( + runner, + mutants_caught_by_type_checker={}, + apply_config_invalidation=False, + invalidate_stale_callers=False, + ) + if not state().current_function_hashes: + raise RuntimeError("mutmut stats prepass did not populate function hashes") + + manifest.update(validate_stats_artifacts()) + MANIFEST_PATH.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return manifest + + +def validate_execution_manifest() -> dict[str, Any]: + """Validate coverage artifacts plus prepared test-selection stats.""" + manifest = validate_coverage_manifest() + stats_fields = validate_stats_artifacts() + for key, value in stats_fields.items(): + if manifest.get(key) != value: + raise RuntimeError(f"prepared mutmut execution manifest is stale for {key}") + return manifest + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "command", + choices=("prepare", "prepare-coverage", "prepare-stats", "validate", "validate-execution"), + ) + parser.add_argument("--max-children", type=int, default=4) + return parser + + +def main() -> int: + args = _parser().parse_args() + try: + if args.command in {"prepare", "prepare-coverage"}: + manifest = prepare(args.max_children) + total = sum(manifest["mutant_counts"].values()) + print(f"prepared covered-line mutation artifacts for {total} mutants") + elif args.command == "prepare-stats": + manifest = prepare_stats(args.max_children) + print( + "prepared mutmut test-selection stats for " + f"{manifest['stats_association_count']} function/test associations", + ) + elif args.command == "validate-execution": + validate_execution_manifest() + print("prepared mutation execution artifacts are fresh") + else: + validate_coverage_manifest() + print("covered-line mutation artifacts are fresh") + except Exception as exc: + print(f"mutation coverage preparation failed: {exc}") + return 1 + return 0 + + +if __name__ == "__main__": + exit_code = main() + sys.stdout.flush() + sys.stderr.flush() + os._exit(exit_code) diff --git a/scripts/run_prepared_mutmut.py b/scripts/run_prepared_mutmut.py new file mode 100644 index 0000000..50d73fb --- /dev/null +++ b/scripts/run_prepared_mutmut.py @@ -0,0 +1,43 @@ +#!/usr/bin/env python3 +"""Run mutmut with prepared coverage and stats artifacts in a fresh interpreter.""" + +from __future__ import annotations + +import argparse +import os +from pathlib import Path +import shutil + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +SITECUSTOMIZE = PROJECT_ROOT / "scripts" / "mutmut_coverage_sitecustomize" + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("mutmut_args", nargs=argparse.REMAINDER) + return parser + + +def main() -> int: + args = _parser().parse_args() + mutmut = shutil.which("mutmut") + if mutmut is None: + print("prepared mutmut runner failed: mutmut executable is not on PATH") + return 1 + + pythonpath = os.environ.get("PYTHONPATH") + prefixes = [str(SITECUSTOMIZE), str(PROJECT_ROOT)] + os.environ["PYTHONPATH"] = os.pathsep.join( + [*prefixes, *([pythonpath] if pythonpath else [])], + ) + os.environ["OBLITERATUS_MUTMUT_REUSE_COVERAGE"] = "1" + os.environ["OBLITERATUS_MUTMUT_REUSE_STATS"] = "1" + os.environ["OBLITERATUS_MUTMUT_SUBPROCESS_PREFLIGHT"] = "1" + + command = [mutmut, *(args.mutmut_args or ["run"])] + os.execv(mutmut, command) + raise AssertionError("os.execv returned unexpectedly") + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_cli.py b/tests/test_cli.py index 8926c9b..2f15643 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -8,7 +8,10 @@ downloading real models or running any pipeline. They use from __future__ import annotations +import json from io import StringIO +from types import SimpleNamespace +from unittest.mock import MagicMock from unittest.mock import patch import pytest @@ -155,6 +158,203 @@ class TestCLIDispatch: assert args_passed.contribute is True assert args_passed.contribute_notes == "Testing contribution system" + @pytest.mark.parametrize("flag", ["--prompt-pairs-file", "--prompt-pair-file"]) + @pytest.mark.parametrize("command", ["obliterate", "abliterate"]) + def test_prompt_pairs_file_flag_is_available_on_obliterate_and_alias( + self, + command, + flag, + tmp_path, + ): + """Explicit prompt-pair files are parsed for both commands and flag spellings.""" + path = tmp_path / "pairs.json" + path.write_text("{}", encoding="utf-8") + with patch("obliteratus.cli._cmd_abliterate") as mock_cmd: + main([command, "fake/model", flag, str(path)]) + args_passed = mock_cmd.call_args[0][0] + assert args_passed.prompt_pairs_file == str(path) + + @pytest.mark.parametrize("flag", ["--prompt-pairs-file", "--prompt-pair-file"]) + @pytest.mark.parametrize("command", ["obliterate", "abliterate"]) + def test_prompt_pairs_file_is_mutually_exclusive_with_residue_files( + self, + command, + flag, + tmp_path, + ): + """Explicit prompt-pair files and mined residue construction cannot be mixed.""" + path = tmp_path / "pairs.json" + path.write_text("{}", encoding="utf-8") + stderr_text = _capture_exit( + [ + command, + "fake/model", + flag, + str(path), + "--residue-file", + "audit.json", + ], + expect_code=2, + ) + assert "not allowed with argument" in stderr_text.lower() + + @pytest.mark.parametrize( + ("option", "value"), + [ + ("--dataset", "custom"), + ("--residue-weight", "7"), + ("--residue-max", "3"), + ], + ) + @pytest.mark.parametrize("flag", ["--prompt-pairs-file", "--prompt-pair-file"]) + @pytest.mark.parametrize("command", ["obliterate", "abliterate"]) + def test_prompt_pairs_file_rejects_residue_only_options( + self, + command, + flag, + option, + value, + tmp_path, + ): + """Explicit prompt-pair files cannot be mixed with residue-only options.""" + path = tmp_path / "pairs.json" + path.write_text("{}", encoding="utf-8") + stderr_text = _capture_exit( + [command, "fake/model", flag, str(path), option, value], + expect_code=2, + ) + assert option in stderr_text + assert "can only be used with --residue-file" in stderr_text + + +def test_cmd_abliterate_wires_prompt_pairs_file_into_pipeline(tmp_path): + """Loaded prompt-pair files are passed directly to AbliterationPipeline.""" + path = tmp_path / "pairs.json" + path.write_text( + json.dumps( + { + "harmful": [f"harm {index}" for index in range(5)], + "harmless": [f"safe {index}" for index in range(5)], + } + ), + encoding="utf-8", + ) + result_path = tmp_path / "result" + result_path.mkdir() + pipeline = MagicMock() + pipeline.run.return_value = str(result_path) + + class FakeLive: + def __init__(self, *_args, **_kwargs): + self.update = MagicMock() + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + args = SimpleNamespace( + model="org/model", + output_dir=str(tmp_path / "out"), + device="cpu", + dtype="float32", + method="basic", + n_directions=1, + direction_method=None, + regularization=None, + refinement_passes=1, + min_layer_fraction=None, + max_layer_fraction=None, + harmless_pc_count=None, + shield_concept_count=None, + shield_ridge=None, + shield_residualize=None, + shield_layer_penalty=None, + projection_target=None, + projection_row_fraction=None, + quantization=None, + gpu_memory_utilization=None, + large_model=False, + verify_sample_size=1, + refusal_max_tokens=1, + residue_file=[], + dataset="builtin", + residue_weight=5, + residue_max=None, + prompt_pairs_file=str(path), + contribute=False, + contribute_notes="", + ) + with ( + patch("rich.live.Live", FakeLive), + patch("obliteratus.abliterate.AbliterationPipeline", return_value=pipeline) as factory, + patch("obliteratus.telemetry.maybe_send_pipeline_report"), + ): + from obliteratus import cli + + cli._cmd_abliterate(args) + + assert factory.call_args.kwargs["harmful_prompts"] == [f"harm {index}" for index in range(5)] + assert factory.call_args.kwargs["harmless_prompts"] == [f"safe {index}" for index in range(5)] + + args.prompt_pairs_file = None + pipeline.reset_mock() + pipeline.run.return_value = str(result_path) + with ( + patch("rich.live.Live", FakeLive), + patch("obliteratus.abliterate.AbliterationPipeline", return_value=pipeline) as factory, + patch("obliteratus.telemetry.maybe_send_pipeline_report"), + ): + from obliteratus import cli + + cli._cmd_abliterate(args) + + assert "harmful_prompts" not in factory.call_args.kwargs + assert "harmless_prompts" not in factory.call_args.kwargs + + +def test_cmd_abliterate_reports_invalid_prompt_pairs_file(tmp_path): + from obliteratus import cli + + args = SimpleNamespace( + model="org/model", + output_dir=str(tmp_path / "out"), + device="cpu", + dtype="float32", + method="basic", + n_directions=1, + direction_method=None, + regularization=None, + refinement_passes=1, + min_layer_fraction=None, + max_layer_fraction=None, + harmless_pc_count=None, + shield_concept_count=None, + shield_ridge=None, + shield_residualize=None, + shield_layer_penalty=None, + projection_target=None, + projection_row_fraction=None, + quantization=None, + gpu_memory_utilization=None, + large_model=False, + verify_sample_size=1, + refusal_max_tokens=1, + residue_file=[], + dataset="builtin", + residue_weight=5, + residue_max=None, + prompt_pairs_file=str(tmp_path / "missing.json"), + contribute=False, + contribute_notes="", + ) + + with pytest.raises(SystemExit) as exc: + cli._cmd_abliterate(args) + + assert exc.value.code == 2 + class _EncodingOnlyStdout: """Minimal stream stand-in for encoding-selection tests.""" diff --git a/tests/test_mutmut_coverage_sitecustomize.py b/tests/test_mutmut_coverage_sitecustomize.py new file mode 100644 index 0000000..cf0b044 --- /dev/null +++ b/tests/test_mutmut_coverage_sitecustomize.py @@ -0,0 +1,110 @@ +"""Isolated tests for the CI-only mutmut covered-line reuse hook.""" + +from __future__ import annotations + +import os +from pathlib import Path +import subprocess +import sys + + +ROOT = Path(__file__).parents[1] +SITECUSTOMIZE = ROOT / "scripts" / "mutmut_coverage_sitecustomize" + + +def _write_fake_modules(tmp_path: Path, *, validation: str = "return None") -> None: + mutmut_dir = tmp_path / "mutmut" + scripts_dir = tmp_path / "scripts" + mutmut_dir.mkdir() + scripts_dir.mkdir() + (mutmut_dir / "__init__.py").write_text("", encoding="utf-8") + (mutmut_dir / "__main__.py").write_text( + "def store_lines_covered_by_tests():\n" + " print('original coverage collector')\n", + encoding="utf-8", + ) + (scripts_dir / "__init__.py").write_text("", encoding="utf-8") + (scripts_dir / "prepare_mutation_coverage.py").write_text( + "def validate_manifest():\n" + f" {validation}\n" + "def validate_execution_manifest():\n" + f" {validation}\n", + encoding="utf-8", + ) + + +def _run_hook_probe(tmp_path: Path, *, env_flag: bool, validation: str = "return None"): + _write_fake_modules(tmp_path, validation=validation) + env = os.environ.copy() + env.pop("OBLITERATUS_MUTMUT_REUSE_COVERAGE", None) + env["PYTHONPATH"] = f"{SITECUSTOMIZE}:{tmp_path}" + if env_flag: + env["OBLITERATUS_MUTMUT_REUSE_COVERAGE"] = "1" + return subprocess.run( + [ + sys.executable, + "-c", + "from mutmut import __main__ as m; m.store_lines_covered_by_tests()", + ], + check=False, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + + +def test_sitecustomize_does_not_patch_without_reuse_env(tmp_path): + result = _run_hook_probe(tmp_path, env_flag=False) + + assert result.returncode == 0 + assert result.stdout.strip() == "original coverage collector" + + +def test_sitecustomize_patches_only_with_reuse_env_and_valid_manifest(tmp_path): + result = _run_hook_probe(tmp_path, env_flag=True, validation="print('validated')") + + assert result.returncode == 0 + assert "validated" in result.stdout + assert "Reusing prepared mutmut covered-line artifacts" in result.stdout + assert "original coverage collector" not in result.stdout + + +def test_sitecustomize_fails_closed_when_manifest_validation_fails(tmp_path): + result = _run_hook_probe( + tmp_path, + env_flag=True, + validation="raise RuntimeError('stale manifest')", + ) + + assert result.returncode != 0 + assert "stale manifest" in result.stderr + + +def test_sitecustomize_reuses_prepared_stats_when_enabled(tmp_path): + _write_fake_modules(tmp_path, validation="print('validated execution')") + env = os.environ.copy() + env.pop("OBLITERATUS_MUTMUT_REUSE_COVERAGE", None) + env["OBLITERATUS_MUTMUT_REUSE_STATS"] = "1" + env["PYTHONPATH"] = f"{SITECUSTOMIZE}:{tmp_path}" + + result = subprocess.run( + [ + sys.executable, + "-c", + ( + "from mutmut import __main__ as m\n" + "m.load_stats = lambda: True\n" + "m.collect_or_load_stats(object())\n" + ), + ], + check=False, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + + assert result.returncode == 0 + assert "validated execution" in result.stdout + assert "Reusing prepared mutmut test-selection stats" in result.stdout diff --git a/tests/test_numerical_contracts.py b/tests/test_numerical_contracts.py index 6216b06..2900ce2 100644 --- a/tests/test_numerical_contracts.py +++ b/tests/test_numerical_contracts.py @@ -58,6 +58,14 @@ def test_direction_count_must_be_a_positive_integer(n_directions): validate_whitened_request(2, 2, n_directions) +@pytest.mark.parametrize("n_directions", [-1, 0]) +def test_non_positive_direction_count_uses_public_error_message(n_directions): + with pytest.raises(ValueError) as excinfo: + validate_whitened_request(2, 2, n_directions) + + assert str(excinfo.value) == "n_directions must be a positive integer" + + @pytest.mark.parametrize("n_directions", [1, 2, 100]) def test_valid_direction_count_is_returned(n_directions): assert validate_whitened_request(2, 2, n_directions) == n_directions diff --git a/tests/test_offline_integration.py b/tests/test_offline_integration.py index 83238ff..c50acf4 100644 --- a/tests/test_offline_integration.py +++ b/tests/test_offline_integration.py @@ -23,20 +23,62 @@ from tests.fixtures.tiny_offline_model import build_tiny_offline_model pytestmark = [pytest.mark.cpu, pytest.mark.integration] REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +THREAD_BOUND_ENV = { + "BLIS_NUM_THREADS": "1", + "MKL_NUM_THREADS": "1", + "NUMEXPR_NUM_THREADS": "1", + "OMP_NUM_THREADS": "1", + "OMP_THREAD_LIMIT": "1", + "OPENBLAS_NUM_THREADS": "1", + "VECLIB_MAXIMUM_THREADS": "1", +} def _offline_cli_env(home: Path) -> dict[str, str]: return { **os.environ, + **THREAD_BOUND_ENV, + "CUDA_VISIBLE_DEVICES": "", "HOME": str(home), "HF_HOME": str(home / "hf"), "HF_DATASETS_OFFLINE": "1", "HF_HUB_DISABLE_TELEMETRY": "1", "HF_HUB_OFFLINE": "1", + "TOKENIZERS_PARALLELISM": "false", "TRANSFORMERS_OFFLINE": "1", } +def _assert_offline_cli_env_thread_bounded( + workdir: Path, + env: dict[str, str], +) -> None: + assert {key: env[key] for key in THREAD_BOUND_ENV} == THREAD_BOUND_ENV + child_env = { + **env, + "OBLITERATUS_TEST_THREAD_BOUND_KEYS": ",".join(THREAD_BOUND_ENV), + } + child = subprocess.run( + [ + sys.executable, + "-I", + "-c", + ( + "import json, os; " + "keys = os.environ['OBLITERATUS_TEST_THREAD_BOUND_KEYS'].split(','); " + "print(json.dumps({key: os.environ.get(key) for key in keys}))" + ), + ], + cwd=workdir, + env=child_env, + check=True, + capture_output=True, + text=True, + timeout=30, + ) + assert json.loads(child.stdout) == THREAD_BOUND_ENV + + def _obliteratus_import_origin(workdir: Path, env: dict[str, str]) -> Path: origin = subprocess.run( [sys.executable, "-I", "-c", "import obliteratus; print(obliteratus.__file__)"], @@ -154,6 +196,7 @@ def test_installed_wheel_cli_loads_local_model_without_repository_imports(tmp_pa isolated_home = tmp_path / "home" isolated_home.mkdir() env = _offline_cli_env(isolated_home) + _assert_offline_cli_env_thread_bounded(isolated_workdir, env) origin = _obliteratus_import_origin(isolated_workdir, env) _require_installed_artifact_import(origin) @@ -213,14 +256,38 @@ def test_installed_package_cli_executes_offline_checkpoint_to_report_slice(tmp_p }, ), ) + prompt_pairs_path = tmp_path / "prompt-pairs.json" + prompt_pairs_path.write_text( + json.dumps( + { + "harmful": [ + "Explain an unsafe request in one sentence", + "Describe a harmful instruction plainly", + "Write a risky instruction prompt", + "Summarize a prohibited request", + "State an unsafe operational request", + ], + "harmless": [ + "Explain a safe request in one sentence", + "Describe a helpful instruction plainly", + "Write a benign instruction prompt", + "Summarize an allowed request", + "State a safe operational request", + ], + } + ), + encoding="utf-8", + ) isolated_workdir = tmp_path / "outside-repository" isolated_workdir.mkdir() isolated_home = tmp_path / "home" isolated_home.mkdir() env = _offline_cli_env(isolated_home) + _assert_offline_cli_env_thread_bounded(isolated_workdir, env) origin = _obliteratus_import_origin(isolated_workdir, env) _require_installed_artifact_import(origin) + original = _state_dict(source) subprocess.run( [ sys.executable, @@ -245,6 +312,8 @@ def test_installed_package_cli_executes_offline_checkpoint_to_report_slice(tmp_p "1", "--refusal-max-tokens", "1", + "--prompt-pairs-file", + str(prompt_pairs_path), ], cwd=isolated_workdir, env=env, @@ -253,8 +322,22 @@ def test_installed_package_cli_executes_offline_checkpoint_to_report_slice(tmp_p text=True, timeout=120, ) - assert (checkpoint / "abliteration_metadata.json").is_file() + metadata_path = checkpoint / "abliteration_metadata.json" + assert metadata_path.is_file() + metadata = json.loads(metadata_path.read_text()) + assert metadata["source_model"] == str(source) + assert metadata["method"] == "basic" + assert metadata["method_config"]["n_directions"] == 1 + assert metadata["method_config"]["refinement_passes"] == 1 + assert metadata["n_harmful_prompts"] == 5 + assert metadata["n_harmless_prompts"] == 5 AutoModelForCausalLM.from_pretrained(checkpoint, local_files_only=True) + checkpoint_state = _state_dict(checkpoint) + assert original.keys() == checkpoint_state.keys() + assert any( + not torch.equal(original[name], tensor) + for name, tensor in checkpoint_state.items() + ) subprocess.run( [sys.executable, "-I", "-m", "obliteratus", "run", str(config_path)], diff --git a/tests/test_projection_math_contracts.py b/tests/test_projection_math_contracts.py new file mode 100644 index 0000000..c54169a --- /dev/null +++ b/tests/test_projection_math_contracts.py @@ -0,0 +1,711 @@ +"""Reference-oracle contracts for pure projection and orthogonalization math.""" + +from __future__ import annotations + +import math +from types import SimpleNamespace + +import pytest +import torch + +from obliteratus.abliterate import AbliterationPipeline +from obliteratus.analysis.numerical_contracts import ( + orthogonalize_subspace_rows, + project_weight_against_direction, + remove_harmless_principal_components, + residualize_against_shield_atoms, + select_projection_coefficients, +) + + +def _canonical_rows(rows: torch.Tensor) -> torch.Tensor: + result = rows.clone() + for idx in range(result.shape[0]): + pivot = result[idx].abs().argmax() + if result[idx, pivot] < 0: + result[idx] = -result[idx] + return result + + +def _reference_gram_schmidt(rows: torch.Tensor) -> torch.Tensor: + basis: list[torch.Tensor] = [] + for row in rows.to(dtype=torch.float64): + residual = row.clone() + for prev in basis: + residual = residual - (residual @ prev) * prev + norm = residual.norm() + if norm > 1e-8: + basis.append(residual / norm) + return torch.stack(basis)[: rows.shape[0]].to(dtype=rows.dtype, device=rows.device) + + +def _reference_projection(weight: torch.Tensor, direction: torch.Tensor, scale: float) -> torch.Tensor: + work = weight.to(dtype=torch.float64) + d = direction.reshape(-1, 1).to(dtype=torch.float64, device=weight.device) + d_norm = d.norm() + if d_norm < 1e-8: + return weight.clone() + d = d / d_norm + if weight.shape[-1] == d.shape[0]: + coeff = work @ d + return (work - d.T * (scale * coeff)).to(dtype=weight.dtype) + coeff = d.T @ work + return (work - (scale * d) * coeff).to(dtype=weight.dtype) + + +def test_orthogonalize_matches_reference_and_preserves_primary_orientation(): + subspace = torch.tensor( + [[2.0, 0.0, 0.0], [1.0, 3.0, 0.0], [1.0, 1.0, 4.0]], + dtype=torch.float64, + ) + + actual = orthogonalize_subspace_rows(subspace) + expected = _reference_gram_schmidt(subspace) + + assert actual.dtype == subspace.dtype + assert torch.allclose(actual @ actual.T, torch.eye(3, dtype=torch.float64), atol=1e-12) + assert torch.allclose(_canonical_rows(actual), _canonical_rows(expected), atol=1e-12) + assert actual[0] @ subspace[0] > 0 + + +def test_orthogonalize_returns_degenerate_inputs_without_allocating_new_tensor(): + single_row = torch.tensor([[1.0, 2.0, 3.0]], dtype=torch.float64) + empty = torch.empty((2, 0), dtype=torch.float64) + + assert orthogonalize_subspace_rows(single_row) is single_row + assert orthogonalize_subspace_rows(empty) is empty + + +def test_orthogonalize_non_finite_subspace_is_explicit_noop(): + subspace = torch.tensor([[1.0, 0.0], [float("nan"), 1.0]], dtype=torch.float64) + + actual = orthogonalize_subspace_rows(subspace) + + assert actual is subspace + assert torch.isnan(actual[1, 0]) + + +def test_orthogonalization_preserves_float64_precision_for_nearly_collinear_rows(): + subspace = torch.tensor( + [[1.0, 1e-6, 0.0], [1.0, 0.0, 1e-6], [0.0, 1.0, 1.0]], + dtype=torch.float64, + ) + + actual = orthogonalize_subspace_rows(subspace) + expected = _reference_gram_schmidt(subspace) + + assert actual.dtype == torch.float64 + assert torch.allclose(_canonical_rows(actual), _canonical_rows(expected), atol=1e-10) + + +def test_orthogonalize_two_half_precision_rows_uses_stable_compute_dtype(): + subspace = torch.tensor( + [[1.0, 1.0, 0.0], [1.0, 0.0, 1.0]], + dtype=torch.float16, + ) + + actual = orthogonalize_subspace_rows(subspace) + expected = _reference_gram_schmidt(subspace) + + assert actual is not subspace + assert actual.dtype == torch.float16 + assert torch.allclose(_canonical_rows(actual), _canonical_rows(expected), atol=1e-3) + gram = actual.float() @ actual.float().T + assert torch.allclose(gram, torch.eye(2), atol=1e-3) + + +def test_integer_projection_uses_supported_float_compute_then_restores_weight_dtype(): + weight = torch.tensor([[2, 0], [0, 2]], dtype=torch.int64) + direction = torch.tensor([1, 0], dtype=torch.int64) + + projected = project_weight_against_direction(weight, direction) + + assert projected.projected is True + assert projected.weight.dtype == torch.int64 + assert torch.equal(projected.weight, torch.tensor([[0, 0], [0, 2]], dtype=torch.int64)) + + +def test_projection_full_removal_is_idempotent_and_orthogonal_to_direction(): + weight = torch.tensor([[3.0, 4.0, 0.0], [1.0, -2.0, 2.0]], dtype=torch.float64) + direction = torch.tensor([0.6, 0.8, 0.0], dtype=torch.float64) + + first = project_weight_against_direction(weight, direction, regularization=0.0) + second = project_weight_against_direction(first.weight, direction, regularization=0.0) + + assert first.projected + assert torch.allclose(first.weight @ direction, torch.zeros(2, dtype=torch.float64), atol=1e-12) + assert torch.allclose(first.weight, second.weight, atol=1e-12) + assert torch.allclose(first.weight, _reference_projection(weight, direction, scale=1.0), atol=1e-12) + + +def test_projection_normalizes_non_unit_directions_before_applying_formula(): + weight = torch.tensor([[3.0, 4.0], [-5.0, 6.0]], dtype=torch.float64) + + unit = project_weight_against_direction(weight, torch.tensor([1.0, 0.0], dtype=torch.float64)) + non_unit = project_weight_against_direction(weight, torch.tensor([2.0, 0.0], dtype=torch.float64)) + + assert unit.projected + assert non_unit.projected + assert torch.allclose(non_unit.weight, unit.weight, atol=1e-12) + assert torch.allclose(non_unit.weight, _reference_projection(weight, torch.tensor([2.0, 0.0]), 1.0)) + + +def test_projection_zero_direction_is_deterministic_no_op(): + weight = torch.tensor([[3.0, 4.0], [-5.0, 6.0]], dtype=torch.float64) + direction = torch.zeros(2, dtype=torch.float64) + + first = project_weight_against_direction(weight, direction) + second = project_weight_against_direction(weight, direction) + + assert not first.projected + assert first.layout is None + assert torch.equal(first.weight, weight) + assert torch.equal(second.weight, first.weight) + assert first.coefficient_norm_sq == 0.0 + + +def test_projection_projects_at_exact_tiny_direction_threshold(): + weight = torch.tensor([[3.0, 4.0]], dtype=torch.float64) + direction = torch.tensor([1e-8, 0.0], dtype=torch.float64) + + projected = project_weight_against_direction(weight, direction) + + assert projected.projected is True + assert projected.layout == "standard" + assert torch.allclose(projected.weight, torch.tensor([[0.0, 4.0]], dtype=torch.float64), atol=1e-12) + + +def test_projection_zero_direction_metadata_uses_strict_false_flag(): + result = project_weight_against_direction( + torch.tensor([[3.0, 4.0]], dtype=torch.float64), + torch.zeros(2, dtype=torch.float64), + ) + + assert result.projected is False + assert result.weight is not None + assert result.coefficient_norm_sq == 0.0 + assert result.layout is None + + +@pytest.mark.parametrize("projection_row_fraction", [0.0, -0.1, 1.01]) +def test_select_projection_coefficients_rejects_invalid_fraction(projection_row_fraction): + coeff = torch.tensor([[1.0], [2.0]], dtype=torch.float64) + + with pytest.raises(ValueError, match=r"projection_row_fraction must be in"): + select_projection_coefficients(coeff, projection_row_fraction) + + +def test_select_projection_coefficients_empty_and_singleton_inputs_are_noops(): + empty = torch.empty((0, 1), dtype=torch.float64) + singleton = torch.tensor([[3.0]], dtype=torch.float64) + + assert select_projection_coefficients(empty, 0.5) is empty + assert select_projection_coefficients(singleton, 0.5) is singleton + + +def test_projection_unsupported_layout_returns_complete_noop_metadata_clone(): + weight = torch.tensor([[1.0, 2.0, 3.0]], dtype=torch.float64) + direction = torch.tensor([1.0, 0.0], dtype=torch.float64) + + projected = project_weight_against_direction(weight, direction) + + assert projected.projected is False + assert projected.layout is None + assert projected.coefficient_norm_sq == 0.0 + assert projected.weight is not weight + assert torch.equal(projected.weight, weight) + + +def test_projection_standard_layout_non_finite_coefficients_fail_closed(): + huge = torch.finfo(torch.float32).max + weight = torch.tensor([[huge, huge]], dtype=torch.float32) + direction = torch.tensor([1.0, 1.0], dtype=torch.float32) + + projected = project_weight_against_direction(weight, direction) + + assert projected.projected is False + assert projected.layout == "standard" + assert projected.coefficient_norm_sq == 0.0 + assert torch.equal(projected.weight, weight) + + +def test_projection_transposed_layout_non_finite_coefficients_fail_closed(): + huge = torch.finfo(torch.float32).max + weight = torch.tensor([[huge], [huge]], dtype=torch.float32) + direction = torch.tensor([1.0, 1.0], dtype=torch.float32) + + projected = project_weight_against_direction(weight, direction) + + assert projected.projected is False + assert projected.layout == "transposed" + assert projected.coefficient_norm_sq == 0.0 + assert torch.equal(projected.weight, weight) + + +def test_projection_preserves_orthogonal_coordinates_and_contracts_norm_without_restore(): + direction = torch.tensor([1.0, 0.0, 0.0]) + orthogonal_probe = torch.tensor([0.0, 2.0, -1.0]) + weight = torch.tensor([[5.0, 2.0, -1.0], [-3.0, 4.0, 7.0]]) + + projected = project_weight_against_direction(weight, direction, norm_preserve=False) + + assert torch.allclose(projected.weight @ direction, torch.zeros(2)) + assert torch.allclose(projected.weight @ orthogonal_probe, weight @ orthogonal_probe) + assert projected.weight.norm() <= weight.norm() + + +def test_norm_preservation_uses_cap_when_projection_would_amplify_too_much(): + weight = torch.tensor([[100.0, 1.0], [0.0, 0.0]]) + direction = torch.tensor([1.0, 0.0]) + + projected = project_weight_against_direction( + weight, + direction, + norm_preserve=True, + max_norm_ratio=1.10, + ) + + assert projected.projected + assert torch.isclose(projected.weight.norm(), torch.tensor(1.10), atol=1e-6) + + +def test_norm_preservation_keeps_zero_projection_without_restoration(): + weight = torch.tensor([[3.0, 0.0]], dtype=torch.float64) + direction = torch.tensor([1.0, 0.0], dtype=torch.float64) + + projected = project_weight_against_direction(weight, direction, norm_preserve=True) + + assert projected.projected is True + assert projected.layout == "standard" + assert projected.coefficient_norm_sq == 9.0 + assert torch.allclose(projected.weight, torch.zeros_like(weight), atol=0.0, rtol=0.0) + assert torch.isfinite(projected.weight).all() + + +def test_norm_preservation_at_max_ratio_boundary_preserves_original_norm(): + direction = torch.tensor([1.0, 0.0], dtype=torch.float64) + weight = torch.tensor([[math.sqrt(21.0), 10.0]], dtype=torch.float64) + + projected = project_weight_against_direction( + weight, + direction, + norm_preserve=True, + max_norm_ratio=1.10, + ) + + assert projected.projected + assert torch.allclose(projected.weight.norm(), weight.norm(), atol=1e-12) + + +def test_transposed_norm_preservation_reports_removed_coefficient_energy(): + direction = torch.tensor([1.0, 0.0], dtype=torch.float64) + weight = torch.tensor([[3.0, 4.0, 0.0], [10.0, 20.0, 30.0]], dtype=torch.float64) + + projected = project_weight_against_direction(weight, direction, norm_preserve=True) + + assert projected.projected is True + assert projected.layout == "transposed" + assert projected.coefficient_norm_sq == 25.0 + assert torch.allclose(projected.weight[0], torch.zeros(3, dtype=torch.float64), atol=1e-12) + + +def test_transposed_projection_without_norm_preservation_reports_zero_metadata_energy(): + direction = torch.tensor([1.0, 0.0], dtype=torch.float64) + weight = torch.tensor([[3.0, 4.0, 0.0], [10.0, 20.0, 30.0]], dtype=torch.float64) + + projected = project_weight_against_direction(weight, direction, norm_preserve=False) + + assert projected.projected is True + assert projected.layout == "transposed" + assert projected.coefficient_norm_sq == 0.0 + assert torch.allclose(projected.weight[0], torch.zeros(3, dtype=torch.float64), atol=1e-12) + + +def test_projection_supports_standard_and_transposed_layouts(): + direction = torch.tensor([1.0, 0.0]) + standard = torch.tensor([[3.0, 4.0], [5.0, 6.0], [7.0, 8.0]]) + transposed = standard.T.contiguous() + + standard_result = project_weight_against_direction(standard, direction) + transposed_result = project_weight_against_direction(transposed, direction) + + assert torch.allclose(standard_result.weight[:, 0], torch.zeros(3)) + assert torch.allclose(transposed_result.weight[0, :], torch.zeros(3)) + assert torch.allclose(standard_result.weight, _reference_projection(standard, direction, 1.0)) + assert torch.allclose(transposed_result.weight, _reference_projection(transposed, direction, 1.0)) + + +def test_projection_rejects_orthogonal_direction_magnitude_as_a_signal(): + weight = torch.tensor([[4.0, 3.0], [2.0, -1.0]], dtype=torch.float64) + unit = project_weight_against_direction(weight, torch.tensor([1.0, 0.0], dtype=torch.float64)) + scaled = project_weight_against_direction(weight, torch.tensor([5.0, 0.0], dtype=torch.float64)) + + assert torch.allclose(unit.weight, scaled.weight, atol=1e-12) + assert unit.layout == scaled.layout == "standard" + + +def test_row_fraction_selects_largest_coefficients_and_is_permutation_equivariant(): + coeff = torch.tensor([[0.5], [-3.0], [2.0], [0.1]]) + selected = select_projection_coefficients(coeff, 0.5) + + assert selected.tolist() == [[0.0], [-3.0], [2.0], [0.0]] + + permutation = torch.tensor([2, 0, 3, 1]) + permuted = select_projection_coefficients(coeff[permutation], 0.5) + assert torch.allclose(permuted, selected[permutation]) + + +def test_projection_row_fraction_removes_only_selected_rows(): + weight = torch.tensor([[10.0, 1.0], [1.0, 7.0], [-5.0, 2.0], [0.2, 9.0]]) + direction = torch.tensor([1.0, 0.0]) + + projected = project_weight_against_direction(weight, direction, projection_row_fraction=0.5) + + assert torch.allclose(projected.weight[:, 0], torch.tensor([0.0, 1.0, 0.0, 0.2])) + assert torch.allclose(projected.weight[:, 1], weight[:, 1]) + + +def test_projection_row_fraction_keeps_only_the_two_largest_magnitudes(): + coeff = torch.tensor([[0.5], [-3.0], [2.0], [0.1]], dtype=torch.float64) + selected = select_projection_coefficients(coeff, 0.5) + + assert torch.equal(selected != 0, torch.tensor([[False], [True], [True], [False]])) + assert torch.allclose(selected.abs().sum(), torch.tensor(5.0, dtype=torch.float64)) + + +@pytest.mark.parametrize("regularization", [-1.0, 0.25, 1.25]) +def test_finite_regularization_values_are_applied_without_unit_interval_clamping(regularization): + weight = torch.tensor([[4.0, 3.0]], dtype=torch.float64) + direction = torch.tensor([1.0, 0.0], dtype=torch.float64) + scale = 1.0 - regularization + + projected = project_weight_against_direction(weight, direction, regularization=regularization) + + assert projected.projected + assert torch.allclose(projected.weight, _reference_projection(weight, direction, scale), atol=1e-12) + + +def test_harmless_pc_removal_orthogonalizes_against_dominant_component(): + subspace = torch.tensor([[1.0, 1.0, 0.0], [0.5, 0.0, 1.0]], dtype=torch.float64) + harmless = torch.tensor( + [[-2.0, 0.0, 0.0], [0.0, 0.0, 0.0], [2.0, 0.0, 0.0], [4.0, 0.0, 0.0]], + dtype=torch.float64, + ) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert torch.allclose(residual[:, 0], torch.zeros(2, dtype=torch.float64), atol=1e-12) + assert torch.allclose(residual.norm(dim=-1), torch.ones(2, dtype=torch.float64), atol=1e-12) + assert torch.allclose(residual @ residual.T, torch.eye(2, dtype=torch.float64), atol=1e-12) + + +@pytest.mark.parametrize( + ("subspace", "harmless", "pc_count"), + [ + ( + torch.tensor([[1.0, 0.0]], dtype=torch.float64), + torch.eye(3, 2, dtype=torch.float64), + 0, + ), + ( + torch.tensor([[1.0, 0.0]], dtype=torch.float64), + torch.eye(2, dtype=torch.float64), + 1, + ), + ( + torch.empty((0, 2), dtype=torch.float64), + torch.eye(3, 2, dtype=torch.float64), + 1, + ), + ], +) +def test_harmless_pc_removal_noops_when_preconditions_are_not_met( + subspace, + harmless, + pc_count, +): + residual = remove_harmless_principal_components(subspace, harmless, pc_count) + + assert residual is subspace + + +def test_harmless_pc_removal_returns_subspace_when_svd_fails(monkeypatch): + subspace = torch.tensor([[1.0, 0.5]], dtype=torch.float64) + harmless = torch.tensor([[-2.0, 0.0], [0.0, 0.0], [2.0, 0.0]], dtype=torch.float64) + + def fail_svd(*_args, **_kwargs): + raise RuntimeError("svd fixture failure") + + monkeypatch.setattr(torch.linalg, "svd", fail_svd) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert residual is subspace + + +def test_harmless_pc_removal_noops_when_svd_returns_no_components(): + subspace = torch.tensor([[1.0]], dtype=torch.float64) + harmless = torch.empty((3, 0), dtype=torch.float64) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert residual is subspace + + +def test_harmless_pc_removal_subtracts_pc_component_before_row_normalization(): + subspace = torch.tensor([[1.0, 0.5]], dtype=torch.float64) + harmless = torch.tensor([[-2.0, 0.0], [0.0, 0.0], [2.0, 0.0]], dtype=torch.float64) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert torch.allclose(residual, torch.tensor([[0.0, 1.0]], dtype=torch.float64), atol=1e-12) + assert torch.allclose(residual.norm(dim=-1), torch.ones(1, dtype=torch.float64), atol=1e-12) + + +def test_harmless_pc_removal_treats_exact_epsilon_residual_as_usable_signal(): + subspace = torch.tensor([[1.0, 1e-8]], dtype=torch.float64) + harmless = torch.tensor([[-2.0, 0.0], [0.0, 0.0], [2.0, 0.0]], dtype=torch.float64) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert torch.allclose(residual, torch.tensor([[0.0, 1.0]], dtype=torch.float64), atol=1e-12) + + +def test_harmless_pc_removal_preserves_subunit_residuals_instead_of_restoring_original_row(): + subspace = torch.tensor([[1.0, 0.5]], dtype=torch.float64) + harmless = torch.tensor([[-2.0, 0.0], [0.0, 0.0], [2.0, 0.0]], dtype=torch.float64) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert torch.allclose(residual[:, 0], torch.zeros(1, dtype=torch.float64), atol=1e-12) + assert torch.allclose(residual[:, 1], torch.ones(1, dtype=torch.float64), atol=1e-12) + + +def test_harmless_pc_removal_restores_only_rows_with_near_zero_residuals(): + subspace = torch.tensor([[1.0, 0.0], [1.0, 0.5]], dtype=torch.float64) + harmless = torch.tensor([[-2.0, 0.0], [0.0, 0.0], [2.0, 0.0]], dtype=torch.float64) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert torch.allclose(residual @ residual.T, torch.eye(2, dtype=torch.float64), atol=1e-12) + assert torch.allclose(residual[0], torch.tensor([1.0, 0.0], dtype=torch.float64), atol=1e-12) + assert torch.allclose(residual[1], torch.tensor([0.0, 1.0], dtype=torch.float64), atol=1e-12) + + +def test_harmless_pc_removal_near_zero_fallback_is_per_row_before_qr(): + subspace = torch.tensor([[1.0, 1e-9, 0.0], [0.2, 0.5, 1.0]], dtype=torch.float64) + harmless = torch.tensor( + [[-2.0, 0.0, 0.0], [0.0, 0.0, 0.0], [2.0, 0.0, 0.0]], + dtype=torch.float64, + ) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert torch.allclose(residual[0], subspace[0], atol=1e-12) + assert torch.allclose(residual @ residual.T, torch.eye(2, dtype=torch.float64), atol=1e-8) + + +def test_harmless_pc_removal_replays_deterministically_for_singular_inputs(): + subspace = torch.tensor([[0.0, 1.0, 1.0], [0.0, 2.0, 2.0]], dtype=torch.float64) + harmless = torch.ones((4, 3), dtype=torch.float64) + + first = remove_harmless_principal_components(subspace, harmless, pc_count=2) + second = remove_harmless_principal_components(subspace, harmless, pc_count=2) + + assert torch.allclose(first, second, atol=0.0, rtol=0.0) + assert torch.isfinite(first).all() + + +def test_shield_atom_residualization_handles_rank_deficient_atoms(): + subspace = torch.tensor([[1.0, 1.0, 0.0], [1.0, 0.0, 1.0]], dtype=torch.float64) + atoms = torch.tensor([[1.0, 0.0, 0.0], [2.0, 0.0, 0.0]], dtype=torch.float64) + + residual = residualize_against_shield_atoms(subspace, atoms, ridge=1e-6) + + assert torch.allclose(residual[:, 0], torch.zeros(2, dtype=torch.float64), atol=2e-6) + assert torch.allclose(residual.norm(dim=-1), torch.ones(2, dtype=torch.float64), atol=1e-12) + assert torch.allclose(residual @ residual.T, torch.eye(2, dtype=torch.float64), atol=1e-12) + + +def test_shield_atom_residualization_noops_when_atoms_or_subspace_are_empty(): + subspace = torch.tensor([[1.0, 0.0]], dtype=torch.float64) + atoms = torch.empty((0, 2), dtype=torch.float64) + empty_subspace = torch.empty((0, 2), dtype=torch.float64) + + assert residualize_against_shield_atoms(subspace, atoms, ridge=1e-3) is subspace + assert residualize_against_shield_atoms(empty_subspace, torch.eye(2), ridge=1e-3) is empty_subspace + + +def test_shield_atom_residualization_returns_subspace_when_solve_fails(monkeypatch): + subspace = torch.tensor([[1.0, 1.0]], dtype=torch.float64) + atoms = torch.tensor([[1.0, 0.0]], dtype=torch.float64) + + def fail_solve(*_args, **_kwargs): + raise RuntimeError("solve fixture failure") + + monkeypatch.setattr(torch.linalg, "solve", fail_solve) + + residual = residualize_against_shield_atoms(subspace, atoms, ridge=1e-6) + + assert residual is subspace + + +def test_shield_atom_residualization_uses_mixed_dtype_compute_but_returns_subspace_dtype(): + subspace = torch.tensor([[1.0, 1.0, 0.0]], dtype=torch.float32) + atoms = torch.tensor([[1.0, 0.0, 0.0]], dtype=torch.float64) + + residual = residualize_against_shield_atoms(subspace, atoms, ridge=1e-6) + + assert residual.dtype == torch.float32 + assert residual.device == subspace.device + assert torch.allclose(residual, torch.tensor([[0.0, 1.0, 0.0]], dtype=torch.float32), atol=2e-6) + + +def test_shield_atom_residualization_upcasts_atoms_to_match_float64_subspace_compute(): + subspace = torch.tensor([[1.0, 1.0, 0.0]], dtype=torch.float64) + atoms = torch.tensor([[1.0, 0.0, 0.0]], dtype=torch.float32) + + residual = residualize_against_shield_atoms(subspace, atoms, ridge=1e-6) + + assert residual.dtype == torch.float64 + assert torch.allclose(residual, torch.tensor([[0.0, 1.0, 0.0]], dtype=torch.float64), atol=2e-6) + + +def test_shield_atom_residualization_keeps_float64_precision_with_float32_atoms(): + subspace = torch.tensor([[1.0, 1e-4, 1.0 - 1e-4]], dtype=torch.float64) + atoms = torch.tensor([[1.0, 1e-4, 0.0], [1e-4, 1.0, 1e-4]], dtype=torch.float32) + + residual = residualize_against_shield_atoms(subspace, atoms, ridge=1e-12) + + assert residual.dtype == torch.float64 + assert residual[0, 0] > 1e-8 + assert torch.allclose( + residual, + torch.tensor( + [[1.0000999534067183e-08, -9.9999997973787514e-05, 0.9999999950000001]], + dtype=torch.float64, + ), + atol=1e-15, + ) + + +def test_harmless_pc_removal_upcasts_half_precision_for_svd_then_returns_input_dtype(): + subspace = torch.tensor([[1.0, 0.5]], dtype=torch.float16) + harmless = torch.tensor([[-2.0, 0.0], [0.0, 0.0], [2.0, 0.0]], dtype=torch.float16) + + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert residual.dtype == torch.float16 + assert torch.allclose(residual.float(), torch.tensor([[0.0, 1.0]]), atol=1e-3) + + +def test_dtype_and_device_are_preserved_for_projection_and_residualization(): + weight = torch.tensor([[1.0, 2.0]], dtype=torch.float32) + direction = torch.tensor([1.0, 0.0], dtype=torch.float64) + subspace = torch.tensor([[1.0, 1.0]], dtype=torch.float32) + harmless = torch.tensor([[-1.0, 0.0], [0.0, 0.0], [1.0, 0.0]], dtype=torch.float64) + + projected = project_weight_against_direction(weight, direction) + residual = remove_harmless_principal_components(subspace, harmless, pc_count=1) + + assert projected.weight.dtype == torch.float32 + assert projected.weight.device == weight.device + assert residual.dtype == torch.float32 + assert residual.device == subspace.device + + +@pytest.mark.parametrize("bad_value", [float("nan"), float("inf")]) +def test_projection_non_finite_policy_is_skip_without_mutation(bad_value): + weight = torch.tensor([[1.0, 2.0], [bad_value, 4.0]]) + direction = torch.tensor([1.0, 0.0]) + + projected = project_weight_against_direction(weight, direction) + + assert not projected.projected + assert torch.allclose(projected.weight, weight, equal_nan=True) + + +def test_zero_inputs_follow_existing_fallback_policy_without_non_finite_output(): + zero_subspace = torch.zeros((2, 3), dtype=torch.float64) + harmless = torch.tensor([[-1.0, 0.0, 0.0], [0.0, 0.0, 0.0], [1.0, 0.0, 0.0]]) + atoms = torch.tensor([[1.0, 0.0, 0.0]]) + + pc_residual = remove_harmless_principal_components(zero_subspace, harmless, pc_count=1) + shield_residual = residualize_against_shield_atoms(zero_subspace, atoms, ridge=1e-3) + + assert torch.isfinite(pc_residual).all() + assert torch.isfinite(shield_residual).all() + assert torch.allclose(pc_residual, zero_subspace) + assert torch.allclose(shield_residual, zero_subspace) + + +def test_regularized_projection_replay_matches_closed_form_decay(): + weight = torch.tensor([[4.0, 3.0]], dtype=torch.float64) + direction = torch.tensor([1.0, 0.0], dtype=torch.float64) + regularization = 0.25 + scale = 1.0 - regularization + + first = project_weight_against_direction(weight, direction, regularization=regularization) + second = project_weight_against_direction(first.weight, direction, regularization=regularization) + + assert torch.allclose(first.weight[:, 0], weight[:, 0] * regularization) + assert torch.allclose(second.weight[:, 0], weight[:, 0] * math.pow(regularization, 2)) + assert torch.allclose(first.weight, _reference_projection(weight, direction, scale)) + + +def test_abliteration_pipeline_math_wrappers_delegate_to_contract_helpers(): + subspace = torch.tensor([[1.0, 0.0], [1.0, 1.0]], dtype=torch.float64) + harmless = torch.tensor([[-2.0, 0.0], [0.0, 0.0], [2.0, 0.0]], dtype=torch.float64) + atoms = torch.tensor([[1.0, 0.0]], dtype=torch.float64) + coeff = torch.tensor([[0.1], [3.0], [-2.0]], dtype=torch.float64) + + assert torch.allclose( + AbliterationPipeline._orthogonalize_subspace(subspace), + orthogonalize_subspace_rows(subspace), + ) + assert torch.allclose( + AbliterationPipeline(None)._remove_harmless_principal_components( + subspace, + harmless, + 1, + ), + remove_harmless_principal_components(subspace, harmless, 1), + ) + assert torch.allclose( + AbliterationPipeline(None)._residualize_against_shield_atoms(subspace, atoms, 1e-6), + residualize_against_shield_atoms(subspace, atoms, 1e-6), + ) + assert torch.equal( + AbliterationPipeline._select_projection_coefficients(coeff, 0.5), + select_projection_coefficients(coeff, 0.5), + ) + + +def test_project_out_advanced_replaces_quantized_weight_after_successful_projection(monkeypatch): + linear = torch.nn.Linear(2, 2, bias=False) + with torch.no_grad(): + linear.weight.copy_(torch.tensor([[2.0, 0.0], [0.0, 2.0]])) + module = SimpleNamespace(o_proj=linear) + replacement_calls = [] + + monkeypatch.setattr( + AbliterationPipeline, + "_dequantize_weight", + staticmethod(lambda proj: (proj.weight.data, True)), + ) + monkeypatch.setattr( + AbliterationPipeline, + "_replace_quantized_weight", + staticmethod(lambda proj, weight: replacement_calls.append((proj, weight.clone()))), + ) + + count = AbliterationPipeline._project_out_advanced( + module, + torch.tensor([1.0, 0.0]), + ["o_proj"], + ) + + assert count == 1 + assert len(replacement_calls) == 1 + assert replacement_calls[0][0] is linear + assert torch.allclose(replacement_calls[0][1][:, 0], torch.zeros(2)) diff --git a/tests/test_prompt_sources.py b/tests/test_prompt_sources.py index 4558f83..4a76342 100644 --- a/tests/test_prompt_sources.py +++ b/tests/test_prompt_sources.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import sys from types import SimpleNamespace @@ -122,6 +123,115 @@ def test_custom_prompt_validation_padding_and_registry_access(): assert prompts.get_valid_volumes("harmbench")[-1] == "all (use entire dataset)" +def test_prompt_pairs_file_loader_accepts_exact_schema(tmp_path): + path = tmp_path / "pairs.json" + path.write_text( + """{ + "harmful": ["harm 0", "harm 1", "harm 2", "harm 3", "harm 4"], + "harmless": ["safe 0", "safe 1", "safe 2", "safe 3", "safe 4"] + }""", + encoding="utf-8", + ) + + harmful, harmless = prompts.load_prompt_pairs_file(path) + + assert harmful == [f"harm {index}" for index in range(5)] + assert harmless == [f"safe {index}" for index in range(5)] + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + ("not json", "malformed JSON"), + ("[]", "JSON object"), + ('{"harmful": ["h"], "harmless": ["s"], "extra": []}', "exactly"), + ('{"harmful": "h", "harmless": ["s"]}', "must be an array"), + ( + '{"harmful": ["h0", "h1", "h2", "h3", 4], ' + '"harmless": ["s0", "s1", "s2", "s3", "s4"]}', + "must be a string", + ), + ('{"harmful": ["h"], "harmless": ["s"]}', "at least 5"), + ( + '{"harmful": ["h0", "h1", "h2", "h3", "h4"], ' + '"harmless": ["s0", "s1", "s2", "s3"]}', + "equal length", + ), + ( + '{"harmful": ["h0", "h1", "h2", "h3", "h4"], ' + '"harmless": ["s0", "s1", "s2", "s3", ""]}', + "nonblank strings", + ), + ( + '{"harmful": ["h0", "h1", "h2", "h3", "h4"], ' + '"harmless": ["s0", "s1", "s2", "s3", "s\\u0000"]}', + "NUL", + ), + ], +) +def test_prompt_pairs_file_loader_rejects_invalid_schema(tmp_path, payload, message): + path = tmp_path / "pairs.json" + path.write_text(payload, encoding="utf-8") + + with pytest.raises(ValueError, match=message): + prompts.load_prompt_pairs_file(path) + + +def test_prompt_pairs_file_loader_rejects_missing_non_regular_oversized_and_utf8(tmp_path): + with pytest.raises(ValueError, match="does not exist"): + prompts.load_prompt_pairs_file(tmp_path / "missing.json") + + with pytest.raises(ValueError, match="regular file"): + prompts.load_prompt_pairs_file(tmp_path) + + oversized = tmp_path / "oversized.json" + oversized.write_bytes(b" " * (prompts.MAX_PROMPT_PAIRS_FILE_BYTES + 1)) + with pytest.raises(ValueError, match="too large"): + prompts.load_prompt_pairs_file(oversized) + + invalid_utf8 = tmp_path / "invalid.json" + invalid_utf8.write_bytes(b"\xff") + with pytest.raises(ValueError, match="UTF-8"): + prompts.load_prompt_pairs_file(invalid_utf8) + + +def test_prompt_pairs_file_loader_rejects_filesystem_errors_and_pair_limit( + tmp_path, + monkeypatch, +): + path = tmp_path / "pairs.json" + path.write_text('{"harmful": [], "harmless": []}', encoding="utf-8") + + def stat_error(_self): + raise OSError("denied") + + with monkeypatch.context() as m: + m.setattr(type(path), "stat", stat_error) + with pytest.raises(ValueError, match="Cannot access"): + prompts.load_prompt_pairs_file(path) + + def read_error(*_args, **_kwargs): + raise OSError("denied") + + with monkeypatch.context() as m: + m.setattr(type(path), "read_text", read_error) + with pytest.raises(ValueError, match="Cannot read"): + prompts.load_prompt_pairs_file(path) + + monkeypatch.setattr(prompts, "MAX_PROMPT_PAIRS", 4) + path.write_text( + json.dumps( + { + "harmful": [f"harm {index}" for index in range(5)], + "harmless": [f"safe {index}" for index in range(5)], + } + ), + encoding="utf-8", + ) + with pytest.raises(ValueError, match="at most"): + prompts.load_prompt_pairs_file(path) + + def test_harmless_generator_cycles_deterministically(): count = len(prompts._HARMLESS_POOL) + 1 generated = prompts._generate_harmless_counterparts(count) diff --git a/tests/test_quality_gate_scripts.py b/tests/test_quality_gate_scripts.py index a973bfd..39d7e79 100644 --- a/tests/test_quality_gate_scripts.py +++ b/tests/test_quality_gate_scripts.py @@ -3,9 +3,11 @@ from __future__ import annotations import json +from pathlib import Path from types import SimpleNamespace from scripts import check_mutation_score +from scripts import check_mutation_targets from scripts import run_repeat_gate @@ -17,6 +19,13 @@ def test_mutation_score_accepts_exact_floor_and_rejects_regression(): ] +def test_mutation_score_validator_default_matches_immutable_policy_floor(): + parser = check_mutation_score._parser() + + assert check_mutation_score.DEFAULT_MUTATION_SCORE_MINIMUM == 85.0 + assert parser.parse_args(["stats.json"]).minimum == 85.0 + + def test_mutation_score_rejects_malformed_and_interrupted_runs(): assert check_mutation_score.validate_mutation_stats( {"killed": True, "total": 1}, minimum=70, @@ -26,6 +35,74 @@ def test_mutation_score_rejects_malformed_and_interrupted_runs(): ) == ["mutation run was interrupted"] +def test_mutation_target_guard_invalidates_stale_copied_targets(tmp_path): + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text( + """ +[tool.mutmut] +only_mutate = [ + "obliteratus/config.py", + "obliteratus/analysis/whitened_svd.py", +] +required_mutation_targets = [ + "obliteratus/analysis/whitened_svd.py", +] +""".lstrip(), + encoding="utf-8", + ) + for target in ( + "obliteratus/config.py", + "obliteratus/analysis/whitened_svd.py", + ): + source = tmp_path / target + source.parent.mkdir(parents=True, exist_ok=True) + source.write_text("pass\n", encoding="utf-8") + + valid_meta = tmp_path / "mutants/obliteratus/config.py.meta" + valid_meta.parent.mkdir(parents=True, exist_ok=True) + valid_meta.write_text('{"exit_code_by_key": {"obliteratus.config.x__mutmut_1": null}}') + + stale_copy = tmp_path / "mutants/obliteratus/analysis/whitened_svd.py" + stale_copy.parent.mkdir(parents=True, exist_ok=True) + stale_copy.write_text("pass\n", encoding="utf-8") + + stale = check_mutation_targets.prepare_required_targets(pyproject) + + assert stale == [Path("obliteratus/analysis/whitened_svd.py")] + assert not stale_copy.exists() + assert check_mutation_targets.validate_required_targets(pyproject) == [ + "configured mutation target produced zero mutants: obliteratus/analysis/whitened_svd.py", + ] + + +def test_mutation_target_guard_passes_when_each_exact_target_has_mutants(tmp_path): + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text( + """ +[tool.mutmut] +only_mutate = [ + "obliteratus/analysis/*.py", + "obliteratus/analysis/numerical_contracts.py", +] +required_mutation_targets = [ + "obliteratus/analysis/numerical_contracts.py", +] +""".lstrip(), + encoding="utf-8", + ) + source = tmp_path / "obliteratus/analysis/numerical_contracts.py" + source.parent.mkdir(parents=True, exist_ok=True) + source.write_text("pass\n", encoding="utf-8") + meta = tmp_path / "mutants/obliteratus/analysis/numerical_contracts.py.meta" + meta.parent.mkdir(parents=True, exist_ok=True) + meta.write_text( + '{"exit_code_by_key": {"obliteratus.analysis.numerical_contracts.x__mutmut_1": null}}', + encoding="utf-8", + ) + + assert check_mutation_targets.validate_required_targets(pyproject) == [] + + def test_repeat_orders_are_distinct_and_deterministic(): paths = ["a", "b", "c", "d"] assert run_repeat_gate.test_orders(paths) == [ diff --git a/tests/test_quality_policy.py b/tests/test_quality_policy.py index c902d54..31de0b8 100644 --- a/tests/test_quality_policy.py +++ b/tests/test_quality_policy.py @@ -9,21 +9,42 @@ from pathlib import Path import sys import pytest +import yaml +from scripts import check_mutation_targets +from scripts import prepare_mutation_coverage +from scripts import run_prepared_mutmut from scripts import check_quality_policy as quality -def test_mutation_campaign_preloads_native_modules_before_covered_line_discovery(): +def _workflow_step(name: str) -> dict: + workflow = yaml.safe_load(Path(".github/workflows/ci.yml").read_text()) + steps = workflow["jobs"]["quality-depth"]["steps"] + matches = [step for step in steps if step.get("name") == name] + assert len(matches) == 1 + return matches[0] + + +def test_mutation_campaign_uses_fork_safe_native_runtime_policy(): pyproject = Path("pyproject.toml").read_text() mutmut_config = pyproject.split("[tool.mutmut]", maxsplit=1)[1].split( "\n[", maxsplit=1, )[0] - workflow = Path(".github/workflows/ci.yml").read_text() + mutation_step = _workflow_step("Run bounded selective mutation gate") + mutation_run = mutation_step["run"] + mutation_env = mutation_step["env"] + quality_job = yaml.safe_load(Path(".github/workflows/ci.yml").read_text())["jobs"][ + "quality-depth" + ] assert "mutate_only_covered_lines = true" in mutmut_config + assert "timeout_constant = 2.0" in mutmut_config assert '"obliteratus/runtime_contracts.py"' in mutmut_config assert '"obliteratus/persistence_contracts.py"' in mutmut_config assert '"obliteratus/evaluation/lm_eval_integration.py"' in mutmut_config + assert '"obliteratus/analysis/numerical_contracts.py"' in mutmut_config + assert '"obliteratus/analysis/whitened_svd.py"' in mutmut_config + assert "required_mutation_targets" in mutmut_config assert '"obliteratus/reporting/report.py"' not in mutmut_config assert '"tests/test_runtime_contracts.py"' in mutmut_config assert '"tests/test_persistence_contracts.py"' in mutmut_config @@ -38,7 +59,98 @@ def test_mutation_campaign_preloads_native_modules_before_covered_line_discovery assert '"tests/test_telemetry_failure_contracts.py"' in Path( "scripts/run_repeat_gate.py", ).read_text() - assert "import torch, yaml; from mutmut.__main__ import cli; cli()" in workflow + assert mutation_env == { + "BLIS_NUM_THREADS": "1", + "MKL_NUM_THREADS": "1", + "NUMEXPR_NUM_THREADS": "1", + "OMP_THREAD_LIMIT": "1", + "OMP_NUM_THREADS": "1", + "OPENBLAS_NUM_THREADS": "1", + "VECLIB_MAXIMUM_THREADS": "1", + } + assert quality_job["timeout-minutes"] == 45 + assert "scripts/check_mutation_targets.py prepare" in mutation_run + assert "scripts/prepare_mutation_coverage.py prepare-coverage --max-children 8" in mutation_run + assert "scripts/prepare_mutation_coverage.py prepare-stats --max-children 8" in mutation_run + assert "scripts/run_prepared_mutmut.py run --max-children 8" in mutation_run + assert "OBLITERATUS_MUTMUT_REUSE_COVERAGE=1" not in mutation_run + assert "scripts/check_mutation_targets.py check" in mutation_run + assert '"$QUALITY_ENV/bin/mutmut" run --max-children 8' not in mutation_run + assert "/usr/bin/time" in mutation_run + timed_block = mutation_run.split("/usr/bin/time", maxsplit=1)[1] + assert timed_block.index("scripts/prepare_mutation_coverage.py prepare-coverage") < ( + timed_block.index("scripts/prepare_mutation_coverage.py prepare-stats") + ) < ( + timed_block.index("scripts/run_prepared_mutmut.py run --max-children 8") + ) + assert "quality-evidence/mutation-time.txt" in timed_block + assert "import torch, yaml; from mutmut.__main__ import cli; cli()" not in mutation_run + + +def test_prepared_mutmut_runner_fails_closed_without_executable(monkeypatch, capsys): + monkeypatch.setattr(sys, "argv", ["run_prepared_mutmut.py", "run"]) + monkeypatch.setattr(run_prepared_mutmut.shutil, "which", lambda _name: None) + + assert run_prepared_mutmut.main() == 1 + assert "mutmut executable is not on PATH" in capsys.readouterr().out + + +def test_prepared_mutmut_runner_execs_with_all_isolation_hooks(monkeypatch): + captured: dict[str, object] = {} + existing_pythonpath = "/existing/pythonpath" + monkeypatch.setenv("PYTHONPATH", existing_pythonpath) + for name in ( + "OBLITERATUS_MUTMUT_REUSE_COVERAGE", + "OBLITERATUS_MUTMUT_REUSE_STATS", + "OBLITERATUS_MUTMUT_SUBPROCESS_PREFLIGHT", + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(sys, "argv", ["run_prepared_mutmut.py", "run", "--max-children", "4"]) + monkeypatch.setattr(run_prepared_mutmut.shutil, "which", lambda _name: "/venv/bin/mutmut") + + def fake_execv(executable, command): + captured["executable"] = executable + captured["command"] = command + captured["pythonpath"] = run_prepared_mutmut.os.environ["PYTHONPATH"] + captured["reuse_coverage"] = run_prepared_mutmut.os.environ[ + "OBLITERATUS_MUTMUT_REUSE_COVERAGE" + ] + captured["reuse_stats"] = run_prepared_mutmut.os.environ[ + "OBLITERATUS_MUTMUT_REUSE_STATS" + ] + captured["subprocess_preflight"] = run_prepared_mutmut.os.environ[ + "OBLITERATUS_MUTMUT_SUBPROCESS_PREFLIGHT" + ] + raise RuntimeError("exec intercepted") + + monkeypatch.setattr(run_prepared_mutmut.os, "execv", fake_execv) + + with pytest.raises(RuntimeError, match="exec intercepted"): + run_prepared_mutmut.main() + + assert captured == { + "executable": "/venv/bin/mutmut", + "command": ["/venv/bin/mutmut", "run", "--max-children", "4"], + "pythonpath": run_prepared_mutmut.os.pathsep.join( + [ + str(run_prepared_mutmut.SITECUSTOMIZE), + str(run_prepared_mutmut.PROJECT_ROOT), + existing_pythonpath, + ], + ), + "reuse_coverage": "1", + "reuse_stats": "1", + "subprocess_preflight": "1", + } + + +def test_mutation_score_floor_is_immutable_across_policy_ci_and_validator(): + policy = json.loads(Path("ci/test-quality-policy.json").read_text(encoding="utf-8")) + workflow = Path(".github/workflows/ci.yml").read_text(encoding="utf-8") + + assert quality.BASELINE_FLOORS["mutation_score"] == 85.0 + assert policy["minimums"]["mutation_score"] == 85.0 + assert "--minimum 85.0" in workflow def _policy(): @@ -111,6 +223,421 @@ def _coverage(): } +def _mutation_project( + tmp_path: Path, + *, + mutmut_table: str | None = None, + source_paths: tuple[str, ...] = ("obliteratus/analysis/numerical_contracts.py",), + meta_by_path: dict[str, str] | None = None, +) -> Path: + """Create a deterministic mutation-target fixture under tmp_path.""" + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text( + mutmut_table + or """ +[tool.mutmut] +required_mutation_targets = [ + "obliteratus/analysis/numerical_contracts.py", +] +""".lstrip(), + encoding="utf-8", + ) + for source_path in source_paths: + source = tmp_path / source_path + source.parent.mkdir(parents=True, exist_ok=True) + source.write_text("def covered_target():\n return 1\n", encoding="utf-8") + for target, metadata in (meta_by_path or {}).items(): + meta = tmp_path / "mutants" / f"{target}.meta" + meta.parent.mkdir(parents=True, exist_ok=True) + meta.write_text(metadata, encoding="utf-8") + return pyproject + + +def test_mutation_target_guard_uses_required_exact_python_targets(tmp_path): + pyproject = _mutation_project( + tmp_path, + mutmut_table=""" +[tool.mutmut] +only_mutate = [ + "obliteratus/config.py", + "obliteratus/analysis/*.py", +] +required_mutation_targets = [ + "obliteratus/analysis/numerical_contracts.py", +] +""".lstrip(), + ) + + assert check_mutation_targets.exact_python_targets( + check_mutation_targets.load_mutmut_config(pyproject), + ) == [Path("obliteratus/analysis/numerical_contracts.py")] + + +@pytest.mark.parametrize( + "bad_target", + [ + "/tmp/escape.py", + "../escape.py", + "obliteratus/../escape.py", + "obliteratus/analysis/*.py", + "README.md", + ], +) +def test_mutation_target_guard_rejects_required_target_escapes_and_non_exact_paths( + tmp_path, bad_target, +): + pyproject = _mutation_project( + tmp_path, + mutmut_table=f""" +[tool.mutmut] +required_mutation_targets = [ + {bad_target!r}, +] +""".lstrip(), + ) + + with pytest.raises(ValueError, match="invalid required mutation target"): + check_mutation_targets.exact_python_targets( + check_mutation_targets.load_mutmut_config(pyproject), + ) + + +def test_mutation_target_guard_rejects_malformed_config_and_metadata(tmp_path): + pyproject = _mutation_project( + tmp_path, + mutmut_table='[tool]\nmutmut = "not-a-table"\n', + ) + with pytest.raises(ValueError, match=r"\[tool\.mutmut\] must be a table"): + check_mutation_targets.load_mutmut_config(pyproject) + + with pytest.raises(ValueError, match="list of strings"): + check_mutation_targets.exact_python_targets({"required_mutation_targets": ["ok.py", 3]}) + + malformed = tmp_path / "mutants/bad.py.meta" + malformed.parent.mkdir(parents=True) + malformed.write_text("{", encoding="utf-8") + with pytest.raises(ValueError, match="is not valid JSON"): + check_mutation_targets.mutant_count(malformed) + + missing_key = tmp_path / "mutants/missing.py.meta" + missing_key.write_text('{"mutants": []}', encoding="utf-8") + with pytest.raises(ValueError, match="exit_code_by_key object"): + check_mutation_targets.mutant_count(missing_key) + + +def test_mutation_target_guard_rejects_missing_configured_source(tmp_path): + pyproject = _mutation_project(tmp_path, source_paths=()) + + with pytest.raises(ValueError, match="configured mutation target does not exist"): + check_mutation_targets.stale_or_empty_targets( + [Path("obliteratus/analysis/numerical_contracts.py")], + project_root=pyproject.parent, + ) + + +def test_mutation_target_guard_removes_all_stale_artifact_kinds(tmp_path): + target = "obliteratus/analysis/numerical_contracts.py" + pyproject = _mutation_project( + tmp_path, + source_paths=(target,), + meta_by_path={target: '{"exit_code_by_key": {}}'}, + ) + artifact_base = tmp_path / "mutants" / target + artifact_base.parent.mkdir(parents=True, exist_ok=True) + for suffix in ("", ".spans"): + (tmp_path / "mutants" / f"{target}{suffix}").write_text("stale", encoding="utf-8") + + assert check_mutation_targets.prepare_required_targets(pyproject) == [Path(target)] + assert not artifact_base.exists() + assert not artifact_base.with_suffix(".py.meta").exists() + assert not artifact_base.with_suffix(".py.spans").exists() + + +def test_mutation_target_guard_never_unlinks_escaped_symlink_artifacts(tmp_path): + target = "obliteratus/analysis/numerical_contracts.py" + pyproject = _mutation_project( + tmp_path, + source_paths=(target,), + meta_by_path={target: '{"exit_code_by_key": {}}'}, + ) + outside = tmp_path / "outside.py" + outside.write_text("do not remove\n", encoding="utf-8") + artifact = tmp_path / "mutants" / target + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.symlink_to(outside) + + with pytest.raises(ValueError, match="escapes project-owned mutants"): + check_mutation_targets.prepare_required_targets(pyproject) + assert outside.exists() + assert artifact.is_symlink() + + +def test_mutation_target_guard_rejects_symlinked_mutants_root_without_deleting_outside( + tmp_path, +): + target = "obliteratus/analysis/numerical_contracts.py" + outside = tmp_path / "outside-mutants" + outside.mkdir() + sentinel = outside / "sentinel.txt" + sentinel.write_text("do not delete\n", encoding="utf-8") + (tmp_path / "mutants").symlink_to(outside, target_is_directory=True) + pyproject = _mutation_project( + tmp_path, + source_paths=(target,), + meta_by_path={}, + ) + + with pytest.raises(ValueError, match="literal project-owned mutants directory"): + check_mutation_targets.prepare_required_targets(pyproject) + with pytest.raises(ValueError, match="literal project-owned mutants directory"): + check_mutation_targets.validate_required_targets(pyproject) + assert sentinel.read_text(encoding="utf-8") == "do not delete\n" + + +@pytest.mark.parametrize("mutants_dir", [Path("/tmp/mutants"), Path("../mutants"), Path("mutants-copy")]) +def test_mutation_target_guard_rejects_unowned_mutants_dir(tmp_path, mutants_dir): + pyproject = _mutation_project(tmp_path) + + with pytest.raises(ValueError, match="mutants directory must be project-owned"): + check_mutation_targets.prepare_required_targets(pyproject, mutants_dir=mutants_dir) + + +def test_mutation_target_guard_cli_prepare_check_and_failure_paths(tmp_path, monkeypatch, capsys): + target = "obliteratus/analysis/numerical_contracts.py" + pyproject = _mutation_project( + tmp_path, + source_paths=(target,), + meta_by_path={target: '{"exit_code_by_key": {"target__mutmut_1": null}}'}, + ) + + monkeypatch.setattr( + sys, + "argv", + ["check_mutation_targets.py", "prepare", "--pyproject", str(pyproject)], + ) + assert check_mutation_targets.main() == 0 + assert "found no stale required targets" in capsys.readouterr().out + + monkeypatch.setattr( + sys, + "argv", + ["check_mutation_targets.py", "check", "--pyproject", str(pyproject)], + ) + assert check_mutation_targets.main() == 0 + assert "mutation target guard passed" in capsys.readouterr().out + + (tmp_path / "mutants" / f"{target}.meta").unlink() + monkeypatch.setattr( + sys, + "argv", + ["check_mutation_targets.py", "check", "--pyproject", str(pyproject)], + ) + assert check_mutation_targets.main() == 1 + assert "produced zero mutants" in capsys.readouterr().out + + +def test_mutation_target_guard_rejects_no_test_required_mutants(tmp_path, monkeypatch): + monkeypatch.setattr( + check_mutation_targets, + "_installed_mutmut_version", + lambda: check_mutation_targets.SUPPORTED_MUTMUT_VERSION, + ) + target = "obliteratus/analysis/numerical_contracts.py" + pyproject = _mutation_project( + tmp_path, + source_paths=(target,), + meta_by_path={ + target: '{"exit_code_by_key": {"target__mutmut_1": 5, "target__mutmut_2": 33}}', + }, + ) + + assert check_mutation_targets.validate_required_targets(pyproject) == [ + "configured mutation target has 2 mutant(s) with no tests: " + "obliteratus/analysis/numerical_contracts.py", + ] + assert check_mutation_targets.MUTMUT_NO_TEST_EXIT_CODES == frozenset({5, 33}) + assert check_mutation_targets.SUPPORTED_MUTMUT_VERSION == "3.7.0" + + +@pytest.mark.parametrize("version", [None, "3.6.0", "3.8.0"]) +def test_mutation_target_guard_fails_closed_before_no_test_code_interpretation( + tmp_path, monkeypatch, version, +): + target = "obliteratus/analysis/numerical_contracts.py" + pyproject = _mutation_project( + tmp_path, + source_paths=(target,), + meta_by_path={target: '{"exit_code_by_key": {"target__mutmut_1": 33}}'}, + ) + monkeypatch.setattr(check_mutation_targets, "_installed_mutmut_version", lambda: version) + + with pytest.raises(ValueError, match="unsupported mutmut version"): + check_mutation_targets.validate_required_targets(pyproject) + + +def test_mutation_target_guard_cli_reports_validation_errors(tmp_path, monkeypatch, capsys): + pyproject = _mutation_project(tmp_path, source_paths=()) + monkeypatch.setattr( + sys, + "argv", + ["check_mutation_targets.py", "prepare", "--pyproject", str(pyproject)], + ) + + assert check_mutation_targets.main() == 1 + assert "configured mutation target does not exist" in capsys.readouterr().out + + +def test_mutation_coverage_manifest_rejects_stale_source_hash(tmp_path, monkeypatch): + source = tmp_path / "pkg/example.py" + mutant = tmp_path / "mutants/pkg/example.py" + source.parent.mkdir(parents=True) + mutant.parent.mkdir(parents=True) + source.write_text("def f():\n return 1\n", encoding="utf-8") + mutant.write_text("mutant", encoding="utf-8") + + manifest = { + "version": 1, + "mutmut_version": prepare_mutation_coverage.SUPPORTED_MUTMUT_VERSION, + "mutate_only_covered_lines": True, + "source_paths": ["pkg/"], + "only_mutate": ["pkg/example.py"], + "pytest_add_cli_args": ["--no-cov"], + "pytest_add_cli_args_test_selection": ["tests/test_example.py"], + "required_mutation_targets": ["pkg/example.py"], + "mutatable_paths": ["pkg/example.py"], + "source_hashes": {"pkg/example.py": "stale"}, + "selected_test_hashes": {"tests/test_example.py": "ok"}, + "hook_hashes": { + "scripts/prepare_mutation_coverage.py": "ok", + "scripts/mutmut_coverage_sitecustomize/sitecustomize.py": "ok", + }, + } + + monkeypatch.chdir(tmp_path) + (tmp_path / "mutants/.covered-lines-prepass.json").write_text( + json.dumps(manifest), + encoding="utf-8", + ) + source.write_text("def f():\n return 2\n", encoding="utf-8") + + monkeypatch.setattr(prepare_mutation_coverage, "assert_supported_mutmut", lambda: None) + monkeypatch.setattr( + prepare_mutation_coverage, + "configured_paths", + lambda: ([Path("pkg/example.py")], [Path("pkg/example.py")]), + ) + monkeypatch.setattr( + prepare_mutation_coverage, + "validate_prepared_artifacts", + lambda *, mutatable, required: { + **manifest, + "source_hashes": { + "pkg/example.py": prepare_mutation_coverage._sha256(source), + }, + }, + ) + + with pytest.raises(RuntimeError, match="manifest is stale for source_hashes"): + prepare_mutation_coverage.validate_manifest() + + +def test_mutation_coverage_prepass_rejects_escaped_cleanup_artifacts(tmp_path, monkeypatch): + outside = tmp_path / "outside.py" + outside.write_text("do not remove\n", encoding="utf-8") + artifact = tmp_path / "mutants/pkg/example.py" + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.symlink_to(outside) + monkeypatch.chdir(tmp_path) + + with pytest.raises(RuntimeError, match="mutation artifact escapes project-owned mutants"): + prepare_mutation_coverage.remove_mutation_artifacts([Path("pkg/example.py")]) + assert outside.exists() + assert artifact.is_symlink() + + +def test_mutation_coverage_manifest_rejects_stale_selected_test_hash(tmp_path, monkeypatch): + source = tmp_path / "pkg/example.py" + test_file = tmp_path / "tests/test_example.py" + hook = tmp_path / "scripts/mutmut_coverage_sitecustomize/sitecustomize.py" + prepass = tmp_path / "scripts/prepare_mutation_coverage.py" + mutant = tmp_path / "mutants/pkg/example.py" + for path in (source, test_file, hook, prepass, mutant): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("original\n", encoding="utf-8") + manifest = { + "version": 1, + "mutmut_version": prepare_mutation_coverage.SUPPORTED_MUTMUT_VERSION, + "mutate_only_covered_lines": True, + "source_paths": ["pkg/"], + "only_mutate": ["pkg/example.py"], + "pytest_add_cli_args": ["--no-cov"], + "pytest_add_cli_args_test_selection": ["tests/test_example.py"], + "required_mutation_targets": ["pkg/example.py"], + "mutatable_paths": ["pkg/example.py"], + "source_hashes": {"pkg/example.py": prepare_mutation_coverage._sha256(source)}, + "selected_test_hashes": {"tests/test_example.py": "stale"}, + "hook_hashes": { + "scripts/prepare_mutation_coverage.py": prepare_mutation_coverage._sha256(prepass), + "scripts/mutmut_coverage_sitecustomize/sitecustomize.py": ( + prepare_mutation_coverage._sha256(hook) + ), + }, + } + monkeypatch.chdir(tmp_path) + (tmp_path / "mutants/.covered-lines-prepass.json").write_text( + json.dumps(manifest), + encoding="utf-8", + ) + monkeypatch.setattr(prepare_mutation_coverage, "assert_supported_mutmut", lambda: None) + monkeypatch.setattr( + prepare_mutation_coverage, + "configured_paths", + lambda: ([Path("pkg/example.py")], [Path("pkg/example.py")]), + ) + monkeypatch.setattr( + prepare_mutation_coverage, + "validate_prepared_artifacts", + lambda *, mutatable, required: { + **manifest, + "selected_test_hashes": { + "tests/test_example.py": prepare_mutation_coverage._sha256(test_file), + }, + }, + ) + + with pytest.raises(RuntimeError, match="manifest is stale for selected_test_hashes"): + prepare_mutation_coverage.validate_manifest() + + +def test_mutation_execution_manifest_rejects_stale_stats_hash(tmp_path, monkeypatch): + stats = tmp_path / "mutants/mutmut-stats.json" + stats.parent.mkdir(parents=True) + stats.write_text( + json.dumps({ + "tests_by_mangled_function_name": {"pkg.x_f": ["tests/test_example.py::test_f"]}, + "duration_by_test": {"tests/test_example.py::test_f": 0.01}, + "stats_time": 0.1, + "function_hashes": {"pkg.x_f": "abc"}, + "function_dependencies": {}, + "config_fingerprint": {}, + "watched_file_hashes": {}, + "git_commit": None, + }), + encoding="utf-8", + ) + manifest = {"stats_hash": "stale"} + + monkeypatch.chdir(tmp_path) + monkeypatch.setattr( + prepare_mutation_coverage, + "validate_coverage_manifest", + lambda: manifest, + ) + + with pytest.raises(RuntimeError, match="execution manifest is stale for stats_hash"): + prepare_mutation_coverage.validate_execution_manifest() + + def test_policy_and_exact_mature_floors_pass(): policy = _policy() assert quality.validate_policy(policy) == [] @@ -122,13 +649,13 @@ def test_policy_and_exact_mature_floors_pass(): def test_floor_regression_requires_structured_reviewed_exception(): policy = _policy() - policy["minimums"]["mutation_score"] = 74 + policy["minimums"]["mutation_score"] = 84 assert quality.validate_policy(policy) == [ - "quality minimum mutation_score cannot move below 75 without an explicit reviewed exception", + "quality minimum mutation_score cannot move below 85 without an explicit reviewed exception", ] policy["threshold_exceptions"] = [{ "threshold": "mutation_score", - "new_value": 74, + "new_value": 84, "reason": "Temporary tool regression", "approved_issue": "https://github.com/elder-plinius/OBLITERATUS/issues/999", "expires": "2026-09-01", diff --git a/tests/test_whitened_svd_oracles.py b/tests/test_whitened_svd_oracles.py new file mode 100644 index 0000000..dff7815 --- /dev/null +++ b/tests/test_whitened_svd_oracles.py @@ -0,0 +1,372 @@ +"""Reference-oracle tests for whitened SVD extraction. + +## Test Context + +- Code to test: `obliteratus/analysis/whitened_svd.py` +- Testing framework: pytest with torch CPU tensors +- Coverage target: repository default, minimum 80% Gate 3 target +- Test types needed: deterministic unit and metamorphic tests +- External dependencies to mock: none; these tests intentionally exercise the + real CPU tensor math and production `WhitenedSVDExtractor` +- Edge cases identified: paired sample permutation, feature-coordinate + permutation, common translation, SVD sign ambiguity, dtype normalization, + singular harmless covariance, zero signal, non-finite input, deterministic + replay, over-requested directions, and `extract_all_layers` intersection order + +The static fixtures below are tiny deterministic activation matrices. Dynamic +fixture factories clone rows into the public list-of-tensors input shape so +tests cannot pass by mutating shared tensors across cases. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pytest +import torch + +from obliteratus.analysis.whitened_svd import WhitenedSVDExtractor + + +REGULARIZATION_EPS = 1e-4 + + +@dataclass(frozen=True) +class ReferenceWhitenedSVD: + directions: torch.Tensor + whitened_directions: torch.Tensor + singular_values: torch.Tensor + variance_explained: float + + +def _activation_pair_fixture( + dtype: torch.dtype = torch.float32, +) -> tuple[torch.Tensor, torch.Tensor]: + harmless = torch.tensor( + [ + [-2.0, -1.0, 0.5, 1.0], + [-1.0, 0.0, -0.5, -1.0], + [0.0, 1.0, 1.5, 0.0], + [1.0, -2.0, -1.5, 2.0], + [2.0, 2.0, 0.0, -2.0], + ], + dtype=dtype, + ) + paired_delta = torch.tensor( + [ + [1.25, -0.50, 0.75, 0.10], + [0.80, 0.25, -0.10, 0.40], + [1.60, -0.75, 0.35, -0.20], + [0.45, 0.90, -0.60, 0.55], + [1.10, -0.20, 0.95, -0.35], + ], + dtype=dtype, + ) + return harmless + paired_delta, harmless + + +def _singular_covariance_fixture() -> tuple[torch.Tensor, torch.Tensor]: + harmless = torch.tensor( + [ + [-2.0, 0.0, 0.0, 1.0], + [-1.0, 0.0, 0.0, 1.0], + [0.0, 0.0, 0.0, 1.0], + [1.0, 0.0, 0.0, 1.0], + [2.0, 0.0, 0.0, 1.0], + ], + ) + harmful = harmless + torch.tensor([1.5, 0.0, 0.0, 0.0]) + return harmful, harmless + + +def _as_public_samples(matrix: torch.Tensor) -> list[torch.Tensor]: + return [row.clone() for row in matrix] + + +def _reference_whitened_svd( + harmful: torch.Tensor, + harmless: torch.Tensor, + *, + n_directions: int, + min_variance_ratio: float = 0.0, +) -> ReferenceWhitenedSVD: + """Independent tiny-tensor reference for the published mathematical contract.""" + harmful = harmful.to(torch.float32).to(torch.float64) + harmless = harmless.to(torch.float32).to(torch.float64) + baseline_mean = harmless.mean(dim=0, keepdim=True) + centered_baseline = harmless - baseline_mean + covariance = centered_baseline.T.mm(centered_baseline) / max(harmless.shape[0] - 1, 1) + eigenvalues, eigenvectors = torch.linalg.eigh(covariance) + eigenvalues = torch.clamp(eigenvalues, min=0.0) + threshold = eigenvalues.max() * min_variance_ratio + kept = eigenvalues >= threshold + kept_values = eigenvalues[kept] + kept_vectors = eigenvectors[:, kept] + + whitening = kept_vectors.mm(torch.diag(torch.rsqrt(kept_values + REGULARIZATION_EPS))) + whitened_delta = (harmful - baseline_mean).mm(whitening) - centered_baseline.mm(whitening) + _, singular_values, right_vectors_t = torch.linalg.svd(whitened_delta, full_matrices=False) + count = min(n_directions, whitened_delta.shape[0], whitened_delta.shape[1]) + whitened_directions = right_vectors_t[:count] + + inverse_whitening = kept_vectors.mm(torch.diag(torch.sqrt(kept_values + REGULARIZATION_EPS))) + original_directions = whitened_directions.mm(inverse_whitening.T) + original_directions = torch.nn.functional.normalize(original_directions, dim=1) + whitened_directions = torch.nn.functional.normalize(whitened_directions, dim=1) + selected_singular_values = singular_values[:count] + variance_explained = ( + selected_singular_values.square().sum() / singular_values.square().sum().clamp(min=1e-12) + ).item() + return ReferenceWhitenedSVD( + directions=original_directions, + whitened_directions=whitened_directions, + singular_values=selected_singular_values, + variance_explained=variance_explained, + ) + + +def _row_space_projector(rows: torch.Tensor, *, tolerance: float = 1e-9) -> torch.Tensor: + rows = rows.to(torch.float64) + _, singular_values, right_vectors_t = torch.linalg.svd(rows, full_matrices=False) + basis = right_vectors_t[singular_values > tolerance] + return basis.T.mm(basis) + + +def _assert_same_subspace(left: torch.Tensor, right: torch.Tensor, *, atol: float = 3e-5) -> None: + assert torch.allclose( + _row_space_projector(left), + _row_space_projector(right), + atol=atol, + rtol=0, + ) + + +def test_matches_independent_reference_not_raw_svd_or_identity_whitening() -> None: + harmful, harmless = _activation_pair_fixture() + result = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ).extract(_as_public_samples(harmful), _as_public_samples(harmless), n_directions=3) + reference = _reference_whitened_svd(harmful, harmless, n_directions=3) + + _assert_same_subspace(result.directions, reference.directions) + _assert_same_subspace(result.whitened_directions, reference.whitened_directions) + assert result.singular_values.double() == pytest.approx( + reference.singular_values, + rel=3e-5, + abs=3e-5, + ) + assert result.variance_explained == pytest.approx(reference.variance_explained, abs=3e-6) + + raw_delta = harmful - harmless + _, _, raw_right_vectors_t = torch.linalg.svd(raw_delta, full_matrices=False) + raw_primary_alignment = torch.dot(result.directions[0], raw_right_vectors_t[0]).abs() + assert raw_primary_alignment < 0.95 + + +def test_joint_sample_permutation_preserves_sign_invariant_refusal_subspace() -> None: + harmful, harmless = _activation_pair_fixture() + permutation = torch.tensor([3, 0, 4, 1, 2]) + extractor = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ) + + original = extractor.extract(_as_public_samples(harmful), _as_public_samples(harmless), 3) + permuted = extractor.extract( + _as_public_samples(harmful[permutation]), + _as_public_samples(harmless[permutation]), + 3, + ) + + _assert_same_subspace(original.directions, permuted.directions) + assert permuted.singular_values == pytest.approx(original.singular_values, rel=3e-5, abs=3e-5) + assert permuted.variance_explained == pytest.approx(original.variance_explained, abs=3e-6) + + +def test_feature_coordinate_permutation_round_trips_through_inverse_mapping() -> None: + harmful, harmless = _activation_pair_fixture() + feature_permutation = torch.tensor([2, 0, 3, 1]) + extractor = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ) + + original = extractor.extract(_as_public_samples(harmful), _as_public_samples(harmless), 3) + permuted = extractor.extract( + _as_public_samples(harmful[:, feature_permutation]), + _as_public_samples(harmless[:, feature_permutation]), + 3, + ) + mapped_back = torch.empty_like(permuted.directions) + mapped_back[:, feature_permutation] = permuted.directions + + _assert_same_subspace(original.directions, mapped_back) + assert permuted.singular_values == pytest.approx(original.singular_values, rel=5e-5, abs=5e-5) + + +def test_common_translation_cannot_change_whitened_svd_oracle_values() -> None: + harmful, harmless = _activation_pair_fixture() + offset = torch.tensor([8.0, -3.0, 0.25, 11.0]) + extractor = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ) + + original = extractor.extract(_as_public_samples(harmful), _as_public_samples(harmless), 3) + translated = extractor.extract( + _as_public_samples(harmful + offset), + _as_public_samples(harmless + offset), + 3, + ) + + _assert_same_subspace(original.directions, translated.directions) + _assert_same_subspace(original.whitened_directions, translated.whitened_directions) + assert translated.singular_values == pytest.approx(original.singular_values, rel=3e-5, abs=3e-5) + assert translated.variance_explained == pytest.approx(original.variance_explained, abs=3e-6) + + +@pytest.mark.parametrize("dtype", [torch.float32, torch.float64]) +def test_float_inputs_follow_float32_output_policy_with_dtype_tolerances( + dtype: torch.dtype, +) -> None: + harmful, harmless = _activation_pair_fixture(dtype) + result = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ).extract(_as_public_samples(harmful), _as_public_samples(harmless), 3) + reference = _reference_whitened_svd(harmful, harmless, n_directions=3) + + assert result.directions.dtype is torch.float32 + assert result.whitened_directions.dtype is torch.float32 + assert result.singular_values.dtype is torch.float32 + _assert_same_subspace(result.directions, reference.directions, atol=4e-5) + assert result.singular_values.double() == pytest.approx( + reference.singular_values, + rel=4e-5, + abs=4e-5, + ) + + +def test_singular_covariance_limits_over_requested_directions_to_effective_rank() -> None: + harmful, harmless = _singular_covariance_fixture() + + result = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.01, + ).extract(_as_public_samples(harmful), _as_public_samples(harmless), n_directions=5) + + assert result.directions.shape == (1, 4) + assert result.whitened_directions.shape == (1, 1) + assert result.singular_values.shape == (1,) + assert torch.count_nonzero(result.directions[0, 1:].abs() > 1e-6) == 0 + assert result.directions.norm() == pytest.approx(1.0) + assert result.variance_explained == pytest.approx(1.0) + assert result.effective_rank == pytest.approx(1.0, abs=1e-6) + + +def test_identical_harmful_and_harmless_inputs_are_rejected_as_no_refusal_signal() -> None: + _, harmless = _activation_pair_fixture() + extractor = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ) + + with pytest.raises(ValueError, match="without activation difference"): + extractor.extract(_as_public_samples(harmless), _as_public_samples(harmless), 2) + + +@pytest.mark.parametrize( + ("poisoned_side", "poisoned_value", "message"), + [ + ("harmful", float("nan"), "finite"), + ("harmless", float("inf"), "finite"), + ], +) +def test_non_finite_activation_values_are_rejected_before_linear_algebra( + poisoned_side: str, + poisoned_value: float, + message: str, +) -> None: + harmful, harmless = _activation_pair_fixture() + target = harmful if poisoned_side == "harmful" else harmless + target[1, 2] = poisoned_value + + with pytest.raises(ValueError, match=message): + WhitenedSVDExtractor().extract(_as_public_samples(harmful), _as_public_samples(harmless), 1) + + +def test_deterministic_replay_returns_byte_stable_cpu_outputs() -> None: + harmful, harmless = _activation_pair_fixture() + extractor = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ) + + first = extractor.extract(_as_public_samples(harmful), _as_public_samples(harmless), 3) + second = extractor.extract(_as_public_samples(harmful), _as_public_samples(harmless), 3) + + assert torch.equal(first.directions, second.directions) + assert torch.equal(first.whitened_directions, second.whitened_directions) + assert torch.equal(first.singular_values, second.singular_values) + assert first.variance_explained == second.variance_explained + assert first.condition_number == second.condition_number + assert first.effective_rank == second.effective_rank + + +def test_extract_all_layers_returns_sorted_harmful_harmless_intersection_only() -> None: + base_harmful, base_harmless = _activation_pair_fixture() + harmful_by_layer = { + 8: _as_public_samples(base_harmful + 0.25), + 2: _as_public_samples(base_harmful), + 5: _as_public_samples(base_harmful * 1.5), + } + harmless_by_layer = { + 9: _as_public_samples(base_harmless), + 5: _as_public_samples(base_harmless * 1.5), + 2: _as_public_samples(base_harmless), + } + + results = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ).extract_all_layers(harmful_by_layer, harmless_by_layer, n_directions=2) + + assert list(results) == [2, 5] + assert [result.layer_idx for result in results.values()] == [2, 5] + assert all(result.directions.shape == (2, 4) for result in results.values()) + + +def test_extract_all_layers_skips_missing_layers_without_stopping_later_matches() -> None: + base_harmful, base_harmless = _activation_pair_fixture() + harmful_by_layer = { + 2: _as_public_samples(base_harmful), + 5: _as_public_samples(base_harmful * 1.5), + 8: _as_public_samples(base_harmful + 0.25), + } + harmless_by_layer = { + 5: _as_public_samples(base_harmless * 1.5), + 8: _as_public_samples(base_harmless), + } + + results = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ).extract_all_layers(harmful_by_layer, harmless_by_layer, n_directions=2) + + assert list(results) == [5, 8] + assert [result.layer_idx for result in results.values()] == [5, 8] + + +def test_extract_all_layers_uses_documented_default_direction_count() -> None: + base_harmful, base_harmless = _activation_pair_fixture() + harmful_by_layer = {5: _as_public_samples(base_harmful)} + harmless_by_layer = {5: _as_public_samples(base_harmless)} + + results = WhitenedSVDExtractor( + regularization_eps=REGULARIZATION_EPS, + min_variance_ratio=0.0, + ).extract_all_layers(harmful_by_layer, harmless_by_layer) + + assert list(results) == [5] + assert results[5].directions.shape == (4, 4)