test: add Gate 3 numerical oracle contracts

This commit is contained in:
Joseph Magly
2026-08-15 20:09:21 -04:00
parent 369417fbb4
commit 51529ad1d7
25 changed files with 3443 additions and 170 deletions
+61
View File
@@ -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 37, 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.
+4 -2
View File
@@ -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.
+27 -5
View File
@@ -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()
+1 -1
View File
@@ -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": [
+2
View File
@@ -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"
]
},
+24 -151
View File
@@ -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
+195
View File
@@ -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,
)
+41 -2
View File
@@ -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(
+92
View File
@@ -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()]
+8
View File
@@ -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]
+4 -1
View File
@@ -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
+224
View File
@@ -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())
+1 -1
View File
@@ -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,
}
@@ -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
+451
View File
@@ -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)
+43
View File
@@ -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())
+200
View File
@@ -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."""
+110
View File
@@ -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
+8
View File
@@ -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
+84 -1
View File
@@ -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)],
+711
View File
@@ -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))
+110
View File
@@ -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)
+77
View File
@@ -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) == [
+533 -6
View File
@@ -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",
+372
View File
@@ -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)