From ba749c68b9762789ca8682722e6e85630f0215dc Mon Sep 17 00:00:00 2001
From: Joseph Magly <1159087+jmagly@users.noreply.github.com>
Date: Sat, 15 Aug 2026 01:52:44 -0400
Subject: [PATCH] test: enforce gate 2 coverage and vertical contracts
---
.github/workflows/conditional-tests.yml | 40 ++
ci/test-quality-policy.json | 8 +-
ci/test-risk-map.json | 35 +-
pyproject.toml | 2 +-
scripts/check_conditional_policy.py | 128 ++++
scripts/check_quality_policy.py | 8 +-
scripts/run_repeat_gate.py | 8 +
tests/test_bayesian_optimizer_contracts.py | 462 ++++++++++++
tests/test_conditional_evidence_freshness.py | 99 +++
tests/test_informed_pipeline_contracts.py | 715 +++++++++++++++++++
tests/test_model_profile_contracts.py | 103 +++
tests/test_offline_integration.py | 162 ++++-
tests/test_package_export_contracts.py | 35 +
tests/test_quality_policy.py | 12 +-
tests/test_remaining_cpu_contracts.py | 164 +++++
tests/test_strategy_navigation_contracts.py | 126 ++++
tests/test_sweep_contracts.py | 67 ++
17 files changed, 2134 insertions(+), 40 deletions(-)
create mode 100644 tests/test_bayesian_optimizer_contracts.py
create mode 100644 tests/test_conditional_evidence_freshness.py
create mode 100644 tests/test_informed_pipeline_contracts.py
create mode 100644 tests/test_model_profile_contracts.py
create mode 100644 tests/test_package_export_contracts.py
create mode 100644 tests/test_remaining_cpu_contracts.py
create mode 100644 tests/test_strategy_navigation_contracts.py
create mode 100644 tests/test_sweep_contracts.py
diff --git a/.github/workflows/conditional-tests.yml b/.github/workflows/conditional-tests.yml
index 8c1de43..ad78aef 100644
--- a/.github/workflows/conditional-tests.yml
+++ b/.github/workflows/conditional-tests.yml
@@ -3,6 +3,10 @@ name: Conditional tests
on:
workflow_dispatch:
inputs:
+ candidate_sha:
+ description: Candidate commit SHA expected in software conditional evidence
+ type: string
+ default: ""
run_model:
description: Run pinned tiny-model download and evaluation gates
type: boolean
@@ -31,6 +35,14 @@ on:
description: Run the least-privileged SSH provider gate
type: boolean
default: false
+ stale_evidence_reason:
+ description: Maintainer reason for accepting older software conditional evidence
+ type: string
+ default: ""
+ stale_evidence_issue:
+ description: OBLITERATUS issue URL approving older software conditional evidence
+ type: string
+ default: ""
schedule:
- cron: "17 6 * * 0"
release:
@@ -44,6 +56,9 @@ concurrency:
cancel-in-progress: false
env:
+ CONDITIONAL_CANDIDATE_SHA: ${{ github.event.inputs.candidate_sha || github.sha }}
+ CONDITIONAL_STALE_EVIDENCE_ISSUE: ${{ github.event.inputs.stale_evidence_issue || '' }}
+ CONDITIONAL_STALE_EVIDENCE_REASON: ${{ github.event.inputs.stale_evidence_reason || '' }}
PIP_DISABLE_PIP_VERSION_CHECK: "1"
PIP_NO_INPUT: "1"
UV_VERSION: "0.12.4"
@@ -95,6 +110,15 @@ jobs:
run: >-
"$CONDITIONAL_ENV/bin/python" scripts/run_conditional_gate.py
external-evaluation
+ - name: Validate software evidence freshness
+ run: >-
+ "$CONDITIONAL_ENV/bin/python" scripts/check_conditional_policy.py
+ --candidate-sha "$CONDITIONAL_CANDIDATE_SHA"
+ --evidence-dir conditional-evidence
+ --require-gate model-download-runtime
+ --require-gate external-evaluation
+ --stale-evidence-reason "$CONDITIONAL_STALE_EVIDENCE_REASON"
+ --stale-evidence-issue "$CONDITIONAL_STALE_EVIDENCE_ISSUE"
- name: Upload model-runtime evidence
if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
@@ -128,6 +152,14 @@ jobs:
run: >-
"$CONDITIONAL_ENV/bin/python" scripts/run_conditional_gate.py
network-services
+ - name: Validate software evidence freshness
+ run: >-
+ "$CONDITIONAL_ENV/bin/python" scripts/check_conditional_policy.py
+ --candidate-sha "$CONDITIONAL_CANDIDATE_SHA"
+ --evidence-dir conditional-evidence
+ --require-gate network-services
+ --stale-evidence-reason "$CONDITIONAL_STALE_EVIDENCE_REASON"
+ --stale-evidence-issue "$CONDITIONAL_STALE_EVIDENCE_ISSUE"
- name: Upload network evidence
if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
@@ -161,6 +193,14 @@ jobs:
run: >-
"$CONDITIONAL_ENV/bin/python" scripts/run_conditional_gate.py
operator-ui
+ - name: Validate software evidence freshness
+ run: >-
+ "$CONDITIONAL_ENV/bin/python" scripts/check_conditional_policy.py
+ --candidate-sha "$CONDITIONAL_CANDIDATE_SHA"
+ --evidence-dir conditional-evidence
+ --require-gate operator-ui
+ --stale-evidence-reason "$CONDITIONAL_STALE_EVIDENCE_REASON"
+ --stale-evidence-issue "$CONDITIONAL_STALE_EVIDENCE_ISSUE"
- name: Upload UI evidence
if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
diff --git a/ci/test-quality-policy.json b/ci/test-quality-policy.json
index 842063c..12184c1 100644
--- a/ci/test-quality-policy.json
+++ b/ci/test-quality-policy.json
@@ -1,11 +1,11 @@
{
"schema_version": 1,
"minimums": {
- "repository_statement": 70.0,
- "repository_branch": 55.0,
+ "repository_statement": 75.0,
+ "repository_branch": 60.0,
"changed_line": 90.0,
- "mature_cpu_statement": 90.0,
- "mature_cpu_branch": 78.0,
+ "mature_cpu_statement": 92.0,
+ "mature_cpu_branch": 80.0,
"mutation_score": 75.0,
"warning_budget": 0
},
diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json
index cb7591c..ea2f35c 100644
--- a/ci/test-risk-map.json
+++ b/ci/test-risk-map.json
@@ -25,6 +25,7 @@
"tests/test_cli.py",
"tests/test_cli_boundaries.py",
"tests/test_module_imports.py",
+ "tests/test_package_export_contracts.py",
"tests/conditional/test_operator_ui.py"
]
},
@@ -48,9 +49,12 @@
"tests/test_abliterate.py",
"tests/test_abliterate_extended.py",
"tests/test_auto_obliterate.py",
+ "tests/test_bayesian_optimizer_contracts.py",
"tests/test_informed_pipeline.py",
+ "tests/test_informed_pipeline_contracts.py",
"tests/test_offline_integration.py",
"tests/test_runner_boundaries.py",
+ "tests/test_sweep_contracts.py",
"tests/test_checkpoint_atomicity.py",
"tests/test_persistence_contracts.py",
"tests/test_persistence_pipeline.py"
@@ -82,6 +86,7 @@
"tests/test_loader_boundaries.py",
"tests/test_mlx_backend_boundaries.py",
"tests/test_model_profile.py",
+ "tests/test_model_profile_contracts.py",
"tests/test_runtime_contracts.py",
"tests/test_study_presets.py"
]
@@ -206,7 +211,8 @@
"tests/test_heretic_eval.py",
"tests/test_lm_eval_reporting_contracts.py",
"tests/test_metrics.py",
- "tests/test_property_contracts.py"
+ "tests/test_property_contracts.py",
+ "tests/test_remaining_cpu_contracts.py"
]
},
{
@@ -237,6 +243,7 @@
],
"required_tests": [
"tests/test_strategies.py",
+ "tests/test_strategy_navigation_contracts.py",
"tests/test_gemma4_support.py",
"tests/test_study_presets.py"
]
@@ -303,7 +310,11 @@
"path": "obliteratus/evaluation/advanced_metrics.py",
"risk_class": "cpu-contract",
"risk": "refusal detection, confidence intervals, and robustness metrics",
- "required_tests": ["tests/test_advanced_metrics.py", "tests/test_property_contracts.py"],
+ "required_tests": [
+ "tests/test_advanced_metrics.py",
+ "tests/test_property_contracts.py",
+ "tests/test_remaining_cpu_contracts.py"
+ ],
"conditional_gates": []
},
{
@@ -363,14 +374,18 @@
"path": "obliteratus/auto_obliterate.py",
"risk_class": "mixed-runtime",
"risk": "automated search state, retry, scoring, and checkpoint behavior",
- "required_tests": ["tests/test_auto_obliterate.py"],
+ "required_tests": ["tests/test_auto_obliterate.py", "tests/test_remaining_cpu_contracts.py"],
"conditional_gates": ["model-download-runtime"]
},
{
"path": "obliteratus/bayesian_optimizer.py",
"risk_class": "conditional-runtime",
"risk": "optional optimizer trials over repeated live model mutation and evaluation",
- "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_model_download_runtime.py"],
+ "required_tests": [
+ "tests/test_bayesian_optimizer_contracts.py",
+ "tests/test_module_imports.py",
+ "tests/conditional/test_model_download_runtime.py"
+ ],
"conditional_gates": ["model-download-runtime"]
},
{
@@ -399,7 +414,11 @@
"path": "obliteratus/informed_pipeline.py",
"risk_class": "mixed-runtime",
"risk": "multi-stage pipeline orchestration and stage-result contracts",
- "required_tests": ["tests/test_informed_pipeline.py", "tests/test_offline_integration.py"],
+ "required_tests": [
+ "tests/test_informed_pipeline.py",
+ "tests/test_informed_pipeline_contracts.py",
+ "tests/test_offline_integration.py"
+ ],
"conditional_gates": ["model-download-runtime"]
},
{
@@ -413,7 +432,11 @@
"path": "obliteratus/sweep.py",
"risk_class": "conditional-runtime",
"risk": "parameter sweeps over repeated mutation and evaluation",
- "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_model_download_runtime.py"],
+ "required_tests": [
+ "tests/test_module_imports.py",
+ "tests/test_sweep_contracts.py",
+ "tests/conditional/test_model_download_runtime.py"
+ ],
"conditional_gates": ["model-download-runtime"]
},
{
diff --git a/pyproject.toml b/pyproject.toml
index e432961..5d44cc0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -95,7 +95,7 @@ addopts = [
"--strict-markers",
"--cov=obliteratus",
"--cov-report=term-missing",
- "--cov-fail-under=70",
+ "--cov-fail-under=75",
]
filterwarnings = [
"error",
diff --git a/scripts/check_conditional_policy.py b/scripts/check_conditional_policy.py
index e38e79d..891addb 100644
--- a/scripts/check_conditional_policy.py
+++ b/scripts/check_conditional_policy.py
@@ -5,12 +5,21 @@ from __future__ import annotations
import argparse
import json
+import re
from pathlib import Path
REQUIRED_GATE_FIELDS = {
"id", "job", "marker", "runner", "prerequisites", "expected_cost", "coverage_paths"
}
+SOFTWARE_ONLY_GATES = (
+ "model-download-runtime",
+ "external-evaluation",
+ "network-services",
+ "operator-ui",
+)
+SHA = re.compile(r"^[0-9a-f]{40}$")
+ISSUE_URL = "https://github.com/elder-plinius/OBLITERATUS/issues/"
def validate(policy_path: Path, quality_path: Path, workflow_path: Path) -> list[str]:
@@ -67,6 +76,86 @@ def validate(policy_path: Path, quality_path: Path, workflow_path: Path) -> list
return errors
+def _load_json_object(path: Path, label: str, errors: list[str]) -> dict:
+ try:
+ value = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError) as exc:
+ errors.append(f"cannot read {label}: {exc}")
+ return {}
+ if not isinstance(value, dict):
+ errors.append(f"{label} root must be an object")
+ return {}
+ return value
+
+
+def _valid_stale_exception(reason: str | None, issue: str | None) -> bool:
+ return (
+ isinstance(reason, str)
+ and bool(reason.strip())
+ and isinstance(issue, str)
+ and issue.startswith(ISSUE_URL)
+ )
+
+
+def validate_evidence(
+ policy_path: Path,
+ evidence_dir: Path,
+ *,
+ candidate_sha: str,
+ required_gates: list[str] | None = None,
+ stale_exception_reason: str | None = None,
+ stale_exception_issue: str | None = None,
+) -> list[str]:
+ """Validate selected software-only conditional evidence against a candidate SHA."""
+
+ errors: list[str] = []
+ policy = _load_json_object(policy_path, "conditional policy", errors)
+ if errors:
+ return errors
+
+ if SHA.fullmatch(candidate_sha) is None:
+ errors.append("candidate SHA must be a 40-character lowercase hex commit")
+
+ gates = policy.get("gates")
+ policy_gate_ids = {
+ gate.get("id")
+ for gate in gates
+ if isinstance(gates, list) and isinstance(gate, dict)
+ } if isinstance(gates, list) else set()
+ requested = required_gates or list(SOFTWARE_ONLY_GATES)
+ for gate_id in requested:
+ if gate_id not in SOFTWARE_ONLY_GATES:
+ errors.append(f"hardware or credential gate is not software-only: {gate_id}")
+ if gate_id not in policy_gate_ids:
+ errors.append(f"conditional policy does not define gate {gate_id}")
+
+ exception = _valid_stale_exception(stale_exception_reason, stale_exception_issue)
+ if (stale_exception_reason or stale_exception_issue) and not exception:
+ errors.append(
+ "stale evidence exception requires a non-empty reason and an OBLITERATUS issue URL",
+ )
+
+ for gate_id in requested:
+ evidence = _load_json_object(
+ evidence_dir / f"{gate_id}.json",
+ f"conditional evidence {gate_id}",
+ errors,
+ )
+ if not evidence:
+ continue
+ if evidence.get("gate") != gate_id:
+ errors.append(f"conditional evidence {gate_id} records gate {evidence.get('gate')!r}")
+ if evidence.get("status") != "passed":
+ errors.append(f"conditional evidence {gate_id} did not pass: {evidence.get('status')!r}")
+ evidence_sha = evidence.get("git_sha")
+ if evidence_sha != candidate_sha and not exception:
+ errors.append(
+ f"conditional evidence {gate_id} git_sha {evidence_sha!r} "
+ f"does not match candidate {candidate_sha}",
+ )
+ return errors
+
+
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--policy", type=Path, default=Path("ci/conditional-test-policy.json"))
@@ -74,8 +163,47 @@ def main() -> int:
parser.add_argument(
"--workflow", type=Path, default=Path(".github/workflows/conditional-tests.yml")
)
+ parser.add_argument(
+ "--evidence-dir",
+ type=Path,
+ help="validate software-only conditional evidence files in this directory",
+ )
+ parser.add_argument(
+ "--candidate-sha",
+ help="40-character candidate commit SHA required for evidence freshness validation",
+ )
+ parser.add_argument(
+ "--require-gate",
+ action="append",
+ default=[],
+ help="software-only gate that must have current passed evidence (repeatable)",
+ )
+ parser.add_argument(
+ "--stale-evidence-reason",
+ default="",
+ help="maintainer reason for accepting older software conditional evidence",
+ )
+ parser.add_argument(
+ "--stale-evidence-issue",
+ default="",
+ help="OBLITERATUS issue URL approving older software conditional evidence",
+ )
args = parser.parse_args()
errors = validate(args.policy, args.quality, args.workflow)
+ if args.evidence_dir is not None:
+ if not args.candidate_sha:
+ errors.append("evidence freshness validation requires --candidate-sha")
+ else:
+ errors.extend(
+ validate_evidence(
+ args.policy,
+ args.evidence_dir,
+ candidate_sha=args.candidate_sha,
+ required_gates=args.require_gate or None,
+ stale_exception_reason=args.stale_evidence_reason or None,
+ stale_exception_issue=args.stale_evidence_issue or None,
+ ),
+ )
if errors:
for error in errors:
print(f"ERROR: {error}")
diff --git a/scripts/check_quality_policy.py b/scripts/check_quality_policy.py
index 516b28f..3f7408d 100644
--- a/scripts/check_quality_policy.py
+++ b/scripts/check_quality_policy.py
@@ -13,11 +13,11 @@ from typing import Any
BASELINE_FLOORS = {
- "repository_statement": 70.0,
- "repository_branch": 55.0,
+ "repository_statement": 75.0,
+ "repository_branch": 60.0,
"changed_line": 90.0,
- "mature_cpu_statement": 90.0,
- "mature_cpu_branch": 78.0,
+ "mature_cpu_statement": 92.0,
+ "mature_cpu_branch": 80.0,
"mutation_score": 75.0,
"warning_budget": 0.0,
}
diff --git a/scripts/run_repeat_gate.py b/scripts/run_repeat_gate.py
index d9d9eb5..555b56a 100644
--- a/scripts/run_repeat_gate.py
+++ b/scripts/run_repeat_gate.py
@@ -15,18 +15,26 @@ from xml.etree import ElementTree
DEFAULT_TESTS = (
+ "tests/test_bayesian_optimizer_contracts.py",
"tests/test_config.py",
"tests/test_config_properties.py",
+ "tests/test_conditional_evidence_freshness.py",
"tests/test_coverage_thresholds.py",
"tests/test_evaluation_reporting_contracts.py",
"tests/test_lm_eval_reporting_contracts.py",
+ "tests/test_informed_pipeline_contracts.py",
+ "tests/test_model_profile_contracts.py",
"tests/test_numerical_contracts.py",
+ "tests/test_package_export_contracts.py",
"tests/test_persistence_contracts.py",
"tests/test_property_contracts.py",
"tests/test_advanced_metrics.py",
"tests/test_metrics.py",
"tests/test_remote_contracts.py",
+ "tests/test_remaining_cpu_contracts.py",
"tests/test_runtime_contracts.py",
+ "tests/test_strategy_navigation_contracts.py",
+ "tests/test_sweep_contracts.py",
"tests/test_telemetry_failure_contracts.py",
)
HASH_SEEDS = ("0", "1", "8675309")
diff --git a/tests/test_bayesian_optimizer_contracts.py b/tests/test_bayesian_optimizer_contracts.py
new file mode 100644
index 0000000..dfddb16
--- /dev/null
+++ b/tests/test_bayesian_optimizer_contracts.py
@@ -0,0 +1,462 @@
+"""CPU-safe contract tests for Bayesian optimization helpers."""
+
+from __future__ import annotations
+
+import builtins
+import sys
+import types
+
+import pytest
+import torch
+import torch.nn as nn
+
+from obliteratus import bayesian_optimizer as bo
+
+
+class _Pipeline:
+ def __init__(self):
+ self.refusal_directions = {}
+ self.handle = None
+ self._strong_layers = []
+ self.harmful_prompts = ["harmful one", "harmful two"]
+ self.use_chat_template = False
+ self.freed = 0
+ self.logs = []
+
+ def _get_model_device(self, _model):
+ return torch.device("cpu")
+
+ def _maybe_apply_chat_template(self, prompts):
+ return [f"{prompt}" for prompt in prompts]
+
+ def _free_gpu_memory(self):
+ self.freed += 1
+
+ def log(self, message):
+ self.logs.append(message)
+
+
+def _install_fake_optuna(monkeypatch, study=None):
+ optuna = types.ModuleType("optuna")
+ optuna.logging = types.SimpleNamespace(WARNING=30, set_verbosity=lambda _level: None)
+ optuna.create_study = lambda **_kwargs: study
+
+ samplers = types.ModuleType("optuna.samplers")
+
+ class TPESampler:
+ def __init__(self, **_kwargs):
+ pass
+
+ samplers.TPESampler = TPESampler
+ monkeypatch.setitem(sys.modules, "optuna", optuna)
+ monkeypatch.setitem(sys.modules, "optuna.samplers", samplers)
+
+
+class _FakeTrial:
+ number = 0
+
+ params = {
+ "attn_max_weight": 0.8,
+ "attn_peak_position": 0.0,
+ "attn_min_weight": 0.1,
+ "attn_spread": 0.6,
+ "mlp_max_weight": 0.6,
+ "mlp_peak_position": 1.0,
+ "mlp_min_weight": 0.2,
+ "mlp_spread": 0.6,
+ "dir_idx": 1.0,
+ }
+
+ values = (0.2, 0.1)
+
+ def __init__(self, number=0, params=None, values=None):
+ self.number = number
+ if params is not None:
+ self.params = params
+ if values is not None:
+ self.values = values
+
+ def suggest_float(self, name, _low, _high):
+ return self.params[name]
+
+
+class _FakeStudy:
+ def __init__(self, best_trials):
+ self.best_trials = best_trials
+ self.enqueued = []
+ self.objective_values = []
+
+ def enqueue_trial(self, params):
+ self.enqueued.append(params)
+
+ def optimize(self, objective, n_trials, show_progress_bar):
+ assert show_progress_bar is False
+ for number in range(n_trials):
+ self.objective_values.append(objective(_FakeTrial(number=number)))
+
+
+class _TokenBatch(dict):
+ def __init__(self):
+ super().__init__(input_ids=torch.tensor([[1, 2]]))
+
+
+class _ReferenceTokenizer:
+ def __call__(self, *_args, **_kwargs):
+ return _TokenBatch()
+
+
+class _Layer(nn.Module):
+ def __init__(self):
+ super().__init__()
+ self.self_attn = nn.Module()
+ self.self_attn.o_proj = nn.Linear(2, 2)
+ self.mlp = nn.Module()
+ self.mlp.down_proj = nn.Linear(2, 2)
+
+
+class _ReferenceModel(nn.Module):
+ def __init__(self, layers):
+ super().__init__()
+ self.model = nn.Module()
+ self.model.layers = nn.ModuleList(layers)
+
+ def forward(self, **_kwargs):
+ return types.SimpleNamespace(logits=torch.tensor([[[0.0, 1.0]]]))
+
+
+def _optimization_pipeline(layers):
+ pipeline = _Pipeline()
+ pipeline.handle = types.SimpleNamespace(
+ model=_ReferenceModel(layers),
+ tokenizer=_ReferenceTokenizer(),
+ architecture="llama",
+ )
+ pipeline._strong_layers = list(range(len(layers)))
+ pipeline.refusal_directions = {
+ idx: torch.tensor([float(idx + 1), 1.0])
+ for idx in pipeline._strong_layers
+ }
+ pipeline.norm_preserve = True
+ pipeline.projections = []
+ pipeline.moe_calls = []
+
+ def project_out(module, direction, names, norm_preserve, regularization):
+ pipeline.projections.append(
+ {
+ "module": module,
+ "direction": direction.detach().clone(),
+ "names": tuple(names),
+ "norm_preserve": norm_preserve,
+ "regularization": regularization,
+ }
+ )
+ for name in names:
+ proj = getattr(module, name, None)
+ if proj is not None and hasattr(proj, "weight"):
+ proj.weight.data.add_(10.0)
+ return 1
+ return 0
+
+ def project_moe(module, direction, **kwargs):
+ pipeline.moe_calls.append((module, direction.detach().clone(), kwargs))
+
+ pipeline._project_out_advanced = project_out
+ pipeline._project_moe_experts = project_moe
+ return pipeline
+
+
+def test_parametric_layer_weight_boundaries():
+ assert bo._parametric_layer_weight(0, 1, 0.8, 0.5, 0.1, 0.2) == pytest.approx(0.8)
+
+ # At the peak, the kernel returns the maximum weight.
+ assert bo._parametric_layer_weight(2, 5, 0.9, 0.5, 0.1, 0.25) == pytest.approx(0.9)
+
+ # At the tent edge, it reaches the minimum weight.
+ assert bo._parametric_layer_weight(1, 5, 0.9, 0.5, 0.1, 0.25) == pytest.approx(0.1)
+
+ # Outside the spread cutoff, the layer is skipped.
+ assert bo._parametric_layer_weight(0, 5, 0.9, 0.5, 0.1, 0.24) == pytest.approx(0.0)
+
+ # Tiny or negative spread is clamped to 0.01.
+ assert bo._parametric_layer_weight(0, 101, 0.7, 0.0, 0.2, -1.0) == pytest.approx(0.7)
+
+
+def test_interpolate_direction_handles_empty_clamps_exact_and_normalized_interpolation():
+ pipeline = _Pipeline()
+ assert torch.equal(bo._interpolate_direction(pipeline, layer_idx=3, float_dir_idx=1.0), torch.zeros(1))
+
+ pipeline.refusal_directions = {
+ 2: torch.tensor([3.0, 0.0]),
+ 5: torch.tensor([0.0, 4.0]),
+ 9: torch.tensor([1.0, 1.0]),
+ }
+
+ low = bo._interpolate_direction(pipeline, layer_idx=5, float_dir_idx=-10.0)
+ assert torch.allclose(low, torch.tensor([1.0, 0.0]))
+
+ high = bo._interpolate_direction(pipeline, layer_idx=5, float_dir_idx=99.0)
+ assert torch.allclose(high, torch.tensor([2**-0.5, 2**-0.5]))
+
+ exact = bo._interpolate_direction(pipeline, layer_idx=5, float_dir_idx=1.0)
+ assert torch.allclose(exact, torch.tensor([0.0, 1.0]))
+
+ interpolated = bo._interpolate_direction(pipeline, layer_idx=5, float_dir_idx=0.5)
+ expected = torch.tensor([1.5, 2.0])
+ expected = expected / expected.norm()
+ assert torch.allclose(interpolated, expected)
+ assert interpolated.norm().item() == pytest.approx(1.0)
+
+
+def test_run_bayesian_optimization_returns_empty_when_optuna_missing(monkeypatch):
+ real_import = builtins.__import__
+
+ def fake_import(name, *args, **kwargs):
+ if name == "optuna" or name.startswith("optuna."):
+ raise ImportError("no optuna in this test")
+ return real_import(name, *args, **kwargs)
+
+ monkeypatch.setattr(builtins, "__import__", fake_import)
+
+ pipeline = _Pipeline()
+ pipeline.handle = object()
+ pipeline._strong_layers = [0]
+
+ assert bo.run_bayesian_optimization(pipeline) == {}
+
+
+def test_run_bayesian_optimization_returns_empty_without_handle_or_layers(monkeypatch):
+ _install_fake_optuna(monkeypatch)
+
+ pipeline = _Pipeline()
+ pipeline.handle = None
+ pipeline._strong_layers = [0]
+ assert bo.run_bayesian_optimization(pipeline) == {}
+
+ pipeline.handle = types.SimpleNamespace(model=object(), tokenizer=object())
+ pipeline._strong_layers = []
+ assert bo.run_bayesian_optimization(pipeline) == {}
+
+
+def test_run_bayesian_optimization_returns_empty_when_reference_logits_fail(monkeypatch):
+ _install_fake_optuna(monkeypatch)
+
+ class RaisingTokenizer:
+ def __call__(self, *_args, **_kwargs):
+ raise RuntimeError("tokenization failed")
+
+ pipeline = _Pipeline()
+ pipeline.handle = types.SimpleNamespace(
+ model=object(),
+ tokenizer=RaisingTokenizer(),
+ )
+ pipeline._strong_layers = [0]
+
+ assert bo.run_bayesian_optimization(pipeline, n_kl_prompts=2) == {}
+ assert pipeline.freed == 1
+ assert "Failed to collect reference logits" in pipeline.logs[-1]
+
+
+def test_measure_refusal_rate_restores_chat_template_and_counts_generation_failures():
+ class TokenBatch(dict):
+ def __init__(self, token_count):
+ super().__init__(input_ids=torch.arange(token_count).unsqueeze(0))
+
+ class Tokenizer:
+ def __init__(self):
+ self.calls = 0
+
+ def __call__(self, prompt, **_kwargs):
+ self.calls += 1
+ return TokenBatch(3 if "one" in prompt else 4)
+
+ def decode(self, token_ids, **_kwargs):
+ if token_ids.numel() == 0:
+ return ""
+ return "I cannot assist with that request"
+
+ class Model:
+ def __init__(self):
+ self.calls = 0
+
+ def generate(self, **kwargs):
+ self.calls += 1
+ if self.calls == 2:
+ raise RuntimeError("generation failed")
+ input_ids = kwargs["input_ids"]
+ continuation = torch.tensor([[99, 100]])
+ return torch.cat([input_ids, continuation], dim=1)
+
+ pipeline = _Pipeline()
+ pipeline.use_chat_template = False
+ pipeline.handle = types.SimpleNamespace(model=Model(), tokenizer=Tokenizer())
+
+ rate = bo._measure_refusal_rate(pipeline, n_prompts=2, max_new_tokens=4)
+
+ assert rate == pytest.approx(0.5)
+ assert pipeline.use_chat_template is False
+ assert pipeline.freed == 1
+
+
+def test_measure_kl_divergence_skips_failures_and_missing_reference_logits():
+ class TokenBatch(dict):
+ def __init__(self):
+ super().__init__(input_ids=torch.tensor([[1, 2]]))
+
+ class Tokenizer:
+ def __call__(self, prompt, **_kwargs):
+ if "bad" in prompt:
+ raise RuntimeError("tokenization failed")
+ return TokenBatch()
+
+ class Output:
+ logits = torch.tensor([[[0.0, 1.0, 2.0]]])
+
+ class Model:
+ def __call__(self, **_kwargs):
+ return Output()
+
+ pipeline = _Pipeline()
+ pipeline.handle = types.SimpleNamespace(model=Model(), tokenizer=Tokenizer())
+ reference = [torch.tensor([0.0, 1.0, 2.0])]
+
+ kl = bo._measure_kl_divergence(
+ pipeline,
+ reference_logits=reference,
+ prompts=["good", "bad", "ignored because no reference"],
+ )
+
+ assert kl == pytest.approx(0.0)
+ assert pipeline.freed == 1
+
+
+def test_run_bayesian_optimization_pareto_path_enqueues_warm_start_and_restores(monkeypatch):
+ layers = [_Layer(), _Layer()]
+ originals = [
+ layer.self_attn.o_proj.weight.detach().clone()
+ for layer in layers
+ ] + [
+ layer.mlp.down_proj.weight.detach().clone()
+ for layer in layers
+ ]
+ best_params = {
+ "attn_max_weight": 0.8,
+ "attn_peak_position": 0.0,
+ "attn_min_weight": 0.1,
+ "attn_spread": 0.6,
+ "mlp_max_weight": 0.6,
+ "mlp_peak_position": 1.0,
+ "mlp_min_weight": 0.2,
+ "mlp_spread": 0.6,
+ "dir_idx": 1.0,
+ }
+ worse_params = {**best_params, "dir_idx": 0.0}
+ study = _FakeStudy(
+ best_trials=[
+ _FakeTrial(params=worse_params, values=(0.5, 0.1)),
+ _FakeTrial(params=best_params, values=(0.2, 0.3)),
+ ]
+ )
+ _install_fake_optuna(monkeypatch, study)
+ monkeypatch.setattr(bo, "_measure_refusal_rate", lambda *_args, **_kwargs: 0.25)
+ monkeypatch.setattr(bo, "_measure_kl_divergence", lambda *_args, **_kwargs: 0.1)
+
+ pipeline = _optimization_pipeline(layers)
+ pipeline._informed_warm_start = {
+ "max_weight": 0.8,
+ "peak_position": 0.25,
+ "min_weight": 0.02,
+ "spread": 0.2,
+ "mlp_scale": 0.5,
+ "dir_idx": 0.75,
+ }
+
+ result = bo.run_bayesian_optimization(
+ pipeline,
+ n_trials=2,
+ n_refusal_prompts=3,
+ n_kl_prompts=1,
+ )
+
+ assert study.enqueued == [{
+ "attn_max_weight": 0.8,
+ "attn_peak_position": 0.25,
+ "attn_min_weight": 0.02,
+ "attn_spread": 0.2,
+ "mlp_max_weight": 0.4,
+ "mlp_peak_position": 0.25,
+ "mlp_min_weight": 0.02,
+ "mlp_spread": 0.2,
+ "dir_idx": 0.75,
+ }]
+ assert study.objective_values == [(0.25, 0.1), (0.25, 0.1)]
+ assert result == {0: pytest.approx(0.6), 1: pytest.approx(0.7)}
+ assert pipeline._bayesian_attn_scale == pytest.approx(0.8)
+ assert pipeline._bayesian_mlp_scale == pytest.approx(0.6)
+ assert any("Applying interpolated direction" in msg for msg in pipeline.logs)
+
+ restored = [
+ layer.self_attn.o_proj.weight
+ for layer in layers
+ ] + [
+ layer.mlp.down_proj.weight
+ for layer in layers
+ ]
+ for live, original in zip(restored, originals, strict=True):
+ assert torch.allclose(live, original)
+
+ assert len(pipeline.projections) == 8
+ regularizations = sorted({round(call["regularization"], 6) for call in pipeline.projections})
+ assert regularizations == [0.2, 0.4, 1.0]
+ assert all(call["norm_preserve"] is True for call in pipeline.projections)
+ assert all(call["direction"].shape == (2, 1) for call in pipeline.projections)
+
+ expected_direction = torch.tensor([2.0, 1.0])
+ expected_direction = expected_direction / expected_direction.norm()
+ assert torch.allclose(pipeline.refusal_directions[0], expected_direction)
+ assert torch.allclose(pipeline.refusal_directions[1], expected_direction)
+
+
+def test_run_bayesian_optimization_no_pareto_uses_objective_best_and_restores_after_projection_errors(
+ monkeypatch,
+):
+ layers = [_Layer()]
+ original_attn = layers[0].self_attn.o_proj.weight.detach().clone()
+ original_mlp = layers[0].mlp.down_proj.weight.detach().clone()
+ study = _FakeStudy(best_trials=[])
+ _install_fake_optuna(monkeypatch, study)
+ monkeypatch.setattr(bo, "_measure_refusal_rate", lambda *_args, **_kwargs: 0.4)
+ monkeypatch.setattr(bo, "_measure_kl_divergence", lambda *_args, **_kwargs: 0.2)
+
+ pipeline = _optimization_pipeline(layers)
+
+ def raising_project(*_args, **_kwargs):
+ layers[0].self_attn.o_proj.weight.data.add_(5.0)
+ layers[0].mlp.down_proj.weight.data.add_(7.0)
+ raise RuntimeError("projection failed")
+
+ pipeline._project_out_advanced = raising_project
+
+ result = bo.run_bayesian_optimization(
+ pipeline,
+ n_trials=1,
+ n_refusal_prompts=1,
+ n_kl_prompts=1,
+ )
+
+ assert study.enqueued == [{
+ "attn_max_weight": 0.9,
+ "attn_peak_position": 0.0,
+ "attn_min_weight": 0.05,
+ "attn_spread": 0.3,
+ "mlp_max_weight": 0.6,
+ "mlp_peak_position": 0.0,
+ "mlp_min_weight": 0.05,
+ "mlp_spread": 0.3,
+ "dir_idx": 0.0,
+ }]
+ assert study.objective_values == [(0.4, 0.2)]
+ assert result == {0: pytest.approx(0.3)}
+ assert any("Using best combined score: 0.5000" in msg for msg in pipeline.logs)
+ assert pipeline.moe_calls == []
+ assert torch.allclose(layers[0].self_attn.o_proj.weight, original_attn)
+ assert torch.allclose(layers[0].mlp.down_proj.weight, original_mlp)
diff --git a/tests/test_conditional_evidence_freshness.py b/tests/test_conditional_evidence_freshness.py
new file mode 100644
index 0000000..81973a3
--- /dev/null
+++ b/tests/test_conditional_evidence_freshness.py
@@ -0,0 +1,99 @@
+"""Tests for software-only conditional evidence freshness policy."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from scripts import check_conditional_policy
+
+
+ROOT = Path(__file__).parents[1]
+SHA = "0123456789abcdef0123456789abcdef01234567"
+OLD_SHA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+
+
+def _write_evidence(path: Path, gate: str, *, git_sha: str = SHA, status: str = "passed") -> None:
+ path.write_text(
+ json.dumps({
+ "schema_version": 1,
+ "gate": gate,
+ "status": status,
+ "git_sha": git_sha,
+ }),
+ encoding="utf-8",
+ )
+
+
+def test_software_evidence_accepts_exact_candidate_sha(tmp_path):
+ _write_evidence(tmp_path / "network-services.json", "network-services")
+
+ assert check_conditional_policy.validate_evidence(
+ ROOT / "ci" / "conditional-test-policy.json",
+ tmp_path,
+ candidate_sha=SHA,
+ required_gates=["network-services"],
+ ) == []
+
+
+def test_software_evidence_rejects_stale_failed_or_mismatched_records(tmp_path):
+ _write_evidence(tmp_path / "network-services.json", "wrong-gate", git_sha=OLD_SHA)
+ _write_evidence(tmp_path / "operator-ui.json", "operator-ui", status="failed")
+
+ errors = check_conditional_policy.validate_evidence(
+ ROOT / "ci" / "conditional-test-policy.json",
+ tmp_path,
+ candidate_sha=SHA,
+ required_gates=["network-services", "operator-ui"],
+ )
+
+ assert "conditional evidence network-services records gate 'wrong-gate'" in errors
+ assert (
+ "conditional evidence network-services git_sha "
+ f"'{OLD_SHA}' does not match candidate {SHA}"
+ ) in errors
+ assert "conditional evidence operator-ui did not pass: 'failed'" in errors
+
+
+def test_maintainer_exception_only_allows_sha_mismatch(tmp_path):
+ _write_evidence(tmp_path / "external-evaluation.json", "external-evaluation", git_sha=OLD_SHA)
+
+ assert check_conditional_policy.validate_evidence(
+ ROOT / "ci" / "conditional-test-policy.json",
+ tmp_path,
+ candidate_sha=SHA,
+ required_gates=["external-evaluation"],
+ stale_exception_reason="Gate reviewed against equivalent conditional surface.",
+ stale_exception_issue="https://github.com/elder-plinius/OBLITERATUS/issues/123",
+ ) == []
+
+
+def test_exception_requires_reason_and_issue_url(tmp_path):
+ _write_evidence(tmp_path / "external-evaluation.json", "external-evaluation", git_sha=OLD_SHA)
+
+ errors = check_conditional_policy.validate_evidence(
+ ROOT / "ci" / "conditional-test-policy.json",
+ tmp_path,
+ candidate_sha=SHA,
+ required_gates=["external-evaluation"],
+ stale_exception_reason="reviewed",
+ )
+
+ assert (
+ "stale evidence exception requires a non-empty reason and an OBLITERATUS issue URL"
+ ) in errors
+ assert (
+ "conditional evidence external-evaluation git_sha "
+ f"'{OLD_SHA}' does not match candidate {SHA}"
+ ) in errors
+
+
+def test_evidence_freshness_does_not_make_hardware_lanes_mandatory(tmp_path):
+ _write_evidence(tmp_path / "cuda-runtime.json", "cuda-runtime")
+
+ assert check_conditional_policy.validate_evidence(
+ ROOT / "ci" / "conditional-test-policy.json",
+ tmp_path,
+ candidate_sha=SHA,
+ required_gates=["cuda-runtime"],
+ ) == ["hardware or credential gate is not software-only: cuda-runtime"]
diff --git a/tests/test_informed_pipeline_contracts.py b/tests/test_informed_pipeline_contracts.py
new file mode 100644
index 0000000..defbbc3
--- /dev/null
+++ b/tests/test_informed_pipeline_contracts.py
@@ -0,0 +1,715 @@
+"""Deterministic contracts for analysis-informed orchestration boundaries."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+from obliteratus.informed_pipeline import InformedAbliterationPipeline
+
+
+@pytest.fixture
+def pipeline(tmp_path):
+ return InformedAbliterationPipeline(
+ model_name="fixture/model",
+ output_dir=str(tmp_path / "output"),
+ on_log=lambda _message: None,
+ )
+
+
+def test_run_informed_executes_the_documented_stage_order(pipeline, monkeypatch):
+ calls = []
+ output = pipeline.output_dir
+ for name in (
+ "_summon",
+ "_probe",
+ "_analyze",
+ "_distill_informed",
+ "_excise_informed",
+ "_verify_and_compensate",
+ ):
+ monkeypatch.setattr(pipeline, name, lambda name=name: calls.append(name))
+ monkeypatch.setattr(
+ pipeline,
+ "_rebirth_informed",
+ lambda: calls.append("_rebirth_informed") or output,
+ )
+ ticks = iter((10.0, 12.5))
+ monkeypatch.setattr("obliteratus.informed_pipeline.time.time", lambda: next(ticks))
+
+ result, report = pipeline.run_informed()
+
+ assert result == output
+ assert calls == [
+ "_summon",
+ "_probe",
+ "_analyze",
+ "_distill_informed",
+ "_excise_informed",
+ "_verify_and_compensate",
+ "_rebirth_informed",
+ ]
+ assert report.total_duration == 2.5
+
+
+def test_analyze_runs_only_enabled_modules_and_records_duration(pipeline, monkeypatch):
+ pipeline._run_cone = False
+ pipeline._run_defense = False
+ calls = []
+ events = []
+ monkeypatch.setattr(pipeline, "_analyze_alignment_imprint", lambda: calls.append("alignment"))
+ monkeypatch.setattr(pipeline, "_analyze_cone_geometry", lambda: calls.append("cone"))
+ monkeypatch.setattr(pipeline, "_analyze_cross_layer", lambda: calls.append("cross-layer"))
+ monkeypatch.setattr(pipeline, "_analyze_defense_robustness", lambda: calls.append("defense"))
+ monkeypatch.setattr(pipeline, "_analyze_sparsity", lambda: calls.append("sparsity"))
+ monkeypatch.setattr(pipeline, "_derive_configuration", lambda: calls.append("derive"))
+ monkeypatch.setattr(pipeline, "_emit", lambda *args, **kwargs: events.append((args, kwargs)))
+ ticks = iter((20.0, 21.25))
+ monkeypatch.setattr("obliteratus.informed_pipeline.time.time", lambda: next(ticks))
+
+ pipeline._analyze()
+
+ assert calls == ["alignment", "cross-layer", "sparsity", "derive"]
+ assert pipeline._report.analysis_duration == 1.25
+ assert events[0][0][:2] == ("analyze", "running")
+ assert events[-1][0][:2] == ("analyze", "done")
+ assert events[-1][1]["duration"] == 1.25
+
+
+def test_alignment_imprint_skips_zero_directions(pipeline, monkeypatch):
+ detector_calls = []
+
+ class Detector:
+ def detect_imprint(self, directions):
+ detector_calls.append(directions)
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.alignment_imprint.AlignmentImprintDetector",
+ Detector,
+ )
+ pipeline._harmful_means = {0: torch.ones(1, 3)}
+ pipeline._harmless_means = {0: torch.ones(1, 3)}
+
+ pipeline._analyze_alignment_imprint()
+
+ assert detector_calls == []
+ assert pipeline._insights.detected_alignment_method == "unknown"
+
+
+def test_alignment_imprint_normalizes_directions_and_copies_result(pipeline, monkeypatch):
+ captured = {}
+ result = SimpleNamespace(
+ predicted_method="dpo",
+ confidence=0.8,
+ dpo_probability=0.8,
+ rlhf_probability=0.1,
+ cai_probability=0.05,
+ sft_probability=0.05,
+ gini_coefficient=0.3,
+ effective_rank=2.0,
+ cross_layer_smoothness=0.7,
+ tail_layer_bias=0.2,
+ )
+
+ class Detector:
+ def detect_imprint(self, directions):
+ captured.update(directions)
+ return result
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.alignment_imprint.AlignmentImprintDetector",
+ Detector,
+ )
+ pipeline._harmful_means = {1: torch.tensor([[3.0, 4.0]])}
+ pipeline._harmless_means = {1: torch.zeros(1, 2)}
+
+ pipeline._analyze_alignment_imprint()
+
+ assert torch.allclose(captured[1], torch.tensor([0.6, 0.8]))
+ assert pipeline._insights.detected_alignment_method == "dpo"
+ assert pipeline._insights.alignment_confidence == 0.8
+ assert pipeline._insights.alignment_probabilities == {
+ "dpo": 0.8,
+ "rlhf": 0.1,
+ "cai": 0.05,
+ "sft": 0.05,
+ }
+
+
+def test_cone_geometry_aggregates_layers_and_keeps_strongest_categories(
+ pipeline,
+ monkeypatch,
+):
+ category = SimpleNamespace(
+ category="safety",
+ direction=torch.tensor([1.0, 0.0]),
+ specificity=0.7,
+ strength=2.0,
+ )
+ results = iter(
+ (
+ SimpleNamespace(
+ is_polyhedral=True,
+ cone_dimensionality=3.0,
+ mean_pairwise_cosine=0.2,
+ general_direction=torch.tensor([2.0, 0.0]),
+ category_directions=[category],
+ category_count=1,
+ ),
+ SimpleNamespace(
+ is_polyhedral=True,
+ cone_dimensionality=5.0,
+ mean_pairwise_cosine=0.4,
+ general_direction=torch.tensor([1.0, 0.0]),
+ category_directions=[],
+ category_count=0,
+ ),
+ )
+ )
+
+ class Analyzer:
+ def analyze_layer(self, *_args, **_kwargs):
+ return next(results)
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.concept_geometry.ConceptConeAnalyzer",
+ Analyzer,
+ )
+ pipeline._harmful_acts = {idx: [torch.ones(1, 2)] for idx in range(4)}
+ pipeline._harmless_acts = {idx: [torch.zeros(1, 2)] for idx in range(4)}
+
+ pipeline._analyze_cone_geometry()
+
+ assert pipeline._insights.cone_is_polyhedral is True
+ assert pipeline._insights.cone_dimensionality == 4.0
+ assert pipeline._insights.mean_pairwise_cosine == pytest.approx(0.3)
+ assert torch.equal(
+ pipeline._insights.per_category_directions["safety"],
+ torch.tensor([1.0, 0.0]),
+ )
+ assert pipeline._insights.direction_specificity == {"safety": 0.7}
+
+
+def test_cone_geometry_preserves_defaults_when_no_layers_are_eligible(pipeline, monkeypatch):
+ class Analyzer:
+ def analyze_layer(self, *_args, **_kwargs):
+ raise AssertionError("no layer should be analyzed")
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.concept_geometry.ConceptConeAnalyzer",
+ Analyzer,
+ )
+ pipeline._harmful_acts = {0: [torch.ones(1, 2)]}
+ pipeline._harmless_acts = {}
+
+ pipeline._analyze_cone_geometry()
+
+ assert pipeline._insights.cone_is_polyhedral is False
+ assert pipeline._insights.cone_dimensionality == 1.0
+
+
+def test_cross_layer_analysis_selects_the_strongest_cluster_representatives(
+ pipeline,
+ monkeypatch,
+):
+ result = SimpleNamespace(
+ clusters=[[0, 1], [2]],
+ cluster_count=2,
+ direction_persistence_score=0.75,
+ mean_adjacent_cosine=0.6,
+ )
+
+ class Analyzer:
+ def __init__(self, cluster_threshold):
+ assert cluster_threshold == 0.85
+
+ def analyze(self, directions):
+ assert set(directions) == {0, 1, 2}
+ return result
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.cross_layer.CrossLayerAlignmentAnalyzer",
+ Analyzer,
+ )
+ pipeline._harmful_means = {
+ 0: torch.tensor([[1.0, 0.0]]),
+ 1: torch.tensor([[3.0, 0.0]]),
+ 2: torch.tensor([[0.0, 2.0]]),
+ }
+ pipeline._harmless_means = {idx: torch.zeros(1, 2) for idx in range(3)}
+
+ pipeline._analyze_cross_layer()
+
+ assert pipeline._insights.direction_clusters == [[0, 1], [2]]
+ assert pipeline._insights.cluster_count == 2
+ assert pipeline._insights.direction_persistence == 0.75
+ assert pipeline._insights.cluster_representative_layers == [1, 2]
+
+
+def test_cross_layer_analysis_requires_two_nonzero_directions(pipeline, monkeypatch):
+ class Analyzer:
+ def __init__(self, **_kwargs):
+ raise AssertionError("analyzer should not be constructed")
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.cross_layer.CrossLayerAlignmentAnalyzer",
+ Analyzer,
+ )
+ pipeline._harmful_means = {0: torch.ones(1, 2)}
+ pipeline._harmless_means = {0: torch.zeros(1, 2)}
+
+ pipeline._analyze_cross_layer()
+
+ assert pipeline._insights.cluster_count == 0
+
+
+def test_defense_analysis_restores_directions_and_copies_profile(pipeline, monkeypatch):
+ original = {9: torch.tensor([9.0])}
+ observed = {}
+
+ class Evaluator:
+ def __init__(self, candidate):
+ observed["temporary"] = dict(candidate.refusal_directions)
+
+ def profile_defense(self):
+ return SimpleNamespace(
+ estimated_robustness="high",
+ self_repair_estimate=0.6,
+ entanglement_score=0.4,
+ )
+
+ def map_entanglement(self):
+ return SimpleNamespace(
+ most_entangled_layers=[2],
+ least_entangled_layers=[0],
+ )
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.defense_robustness.DefenseRobustnessEvaluator",
+ Evaluator,
+ )
+ pipeline.refusal_directions = original
+ pipeline._harmful_means = {0: torch.tensor([[0.0, 2.0]])}
+ pipeline._harmless_means = {0: torch.zeros(1, 2)}
+
+ pipeline._analyze_defense_robustness()
+
+ assert torch.equal(observed["temporary"][0], torch.tensor([0.0, 1.0]))
+ assert pipeline.refusal_directions is original
+ assert pipeline._insights.estimated_robustness == "high"
+ assert pipeline._insights.self_repair_estimate == 0.6
+ assert pipeline._insights.entangled_layers == [2]
+ assert pipeline._insights.clean_layers == [0]
+
+
+def test_sparsity_analysis_builds_a_plan_from_compatible_ffn_weights(
+ pipeline,
+ monkeypatch,
+):
+ projection = SimpleNamespace(weight=SimpleNamespace(data=torch.eye(2)))
+ layer = SimpleNamespace()
+ plan = SimpleNamespace(
+ mean_refusal_sparsity_index=0.65,
+ recommended_sparsity=0.2,
+ most_sparse_layer=0,
+ most_dense_layer=0,
+ )
+ captured = {}
+
+ class Surgeon:
+ def __init__(self, auto_sparsity):
+ assert auto_sparsity is True
+
+ def plan_surgery(self, weights, directions):
+ captured["weights"] = weights
+ captured["directions"] = directions
+ return plan
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.sparse_surgery.SparseDirectionSurgeon",
+ Surgeon,
+ )
+ monkeypatch.setattr(
+ "obliteratus.strategies.utils.get_layer_modules",
+ lambda _handle: [layer],
+ )
+ monkeypatch.setattr(
+ "obliteratus.strategies.utils.get_ffn_module",
+ lambda _layer, _arch: SimpleNamespace(down_proj=projection),
+ )
+ pipeline.handle = SimpleNamespace(architecture="gpt2")
+ pipeline._harmful_means = {0: torch.tensor([[0.0, 2.0]])}
+ pipeline._harmless_means = {0: torch.zeros(1, 2)}
+
+ pipeline._analyze_sparsity()
+
+ assert set(captured["weights"]) == {0}
+ assert torch.equal(captured["directions"][0], torch.tensor([0.0, 1.0]))
+ assert pipeline._insights.mean_refusal_sparsity_index == 0.65
+ assert pipeline._insights.recommended_sparsity == 0.2
+
+
+@pytest.mark.parametrize(
+ ("method", "budget"),
+ [("dpo", 0.5), ("rlhf", 0.3), ("cai", 0.2), ("sft", 0.4), ("unknown", 0.35)],
+)
+def test_bayesian_warm_start_sets_alignment_specific_kl_budget(
+ pipeline,
+ method,
+ budget,
+):
+ pipeline._insights.detected_alignment_method = method
+
+ pipeline._configure_bayesian_warm_start()
+
+ assert pipeline.kl_budget == budget
+ assert pipeline._bayesian_trials == 50
+ assert pipeline.layer_adaptive_strength is True
+ assert pipeline.float_layer_interpolation is True
+ assert pipeline.use_kl_optimization is True
+
+
+def test_bayesian_warm_start_uses_strongest_cluster_and_entanglement(pipeline):
+ pipeline._harmful_means = {
+ 0: torch.tensor([[1.0, 0.0]]),
+ 1: torch.tensor([[4.0, 0.0]]),
+ 2: torch.tensor([[2.0, 0.0]]),
+ 3: torch.tensor([[1.0, 0.0]]),
+ }
+ pipeline._harmless_means = {idx: torch.zeros(1, 2) for idx in range(4)}
+ pipeline._insights.cluster_representative_layers = [0, 1]
+ pipeline._insights.direction_clusters = [[0, 1], [2, 3]]
+ pipeline._insights.direction_persistence = 0.5
+ pipeline._insights.entanglement_score = 0.8
+
+ pipeline._configure_bayesian_warm_start()
+
+ warm = pipeline._informed_warm_start
+ assert warm["peak_position"] == pytest.approx(1 / 3)
+ assert warm["spread"] == pytest.approx(1 / 3)
+ assert warm["min_weight"] == 0.1
+ assert warm["attn_scale"] == 0.7
+ assert warm["mlp_scale"] == 0.4
+
+
+def test_excise_informed_routes_sparse_and_dense_paths(pipeline, monkeypatch):
+ calls = []
+ monkeypatch.setattr(pipeline, "_excise_sparse", lambda: calls.append("sparse"))
+ monkeypatch.setattr(
+ pipeline,
+ "_configure_bayesian_warm_start",
+ lambda: calls.append("warm-start"),
+ )
+ monkeypatch.setattr(pipeline, "_excise", lambda: calls.append("dense"))
+
+ pipeline._insights.use_sparse_surgery = True
+ pipeline._excise_informed()
+ pipeline._insights.use_sparse_surgery = False
+ pipeline._excise_informed()
+
+ assert calls == ["sparse", "warm-start", "dense"]
+
+
+def test_verify_compensation_stops_when_no_residual_layers(pipeline, monkeypatch):
+ calls = []
+
+ def verify():
+ calls.append("verify")
+ pipeline._quality_metrics = {"refusal_rate": 0.9, "kl_divergence": 0.1}
+
+ monkeypatch.setattr(pipeline, "_verify", verify)
+ monkeypatch.setattr(pipeline, "_probe", lambda: calls.append("probe"))
+
+ def distill():
+ calls.append("distill")
+ pipeline._strong_layers = []
+
+ monkeypatch.setattr(pipeline, "_distill_inner", distill)
+ monkeypatch.setattr(pipeline, "_excise_informed", lambda: calls.append("excise"))
+
+ pipeline._verify_and_compensate()
+
+ assert calls == ["verify", "probe", "distill"]
+ assert pipeline._report.ouroboros_passes == 1
+ assert pipeline._report.final_refusal_rate == 0.9
+
+
+def test_verify_compensation_stops_at_kl_ceiling(pipeline, monkeypatch):
+ outcomes = iter(
+ (
+ {"refusal_rate": 0.9, "kl_divergence": 0.1},
+ {"refusal_rate": 0.8, "kl_divergence": 0.8},
+ )
+ )
+ calls = []
+
+ def verify():
+ calls.append("verify")
+ pipeline._quality_metrics = next(outcomes)
+
+ monkeypatch.setattr(pipeline, "_verify", verify)
+ monkeypatch.setattr(pipeline, "_probe", lambda: calls.append("probe"))
+ monkeypatch.setattr(pipeline, "_distill_inner", lambda: setattr(pipeline, "_strong_layers", [1]))
+ monkeypatch.setattr(pipeline, "_excise_informed", lambda: calls.append("excise"))
+ pipeline.kl_budget = 0.3
+
+ pipeline._verify_and_compensate()
+
+ assert calls == ["verify", "probe", "excise", "verify"]
+ assert pipeline._report.ouroboros_passes == 1
+ assert pipeline._report.final_refusal_rate == 0.8
+
+
+def test_rebirth_writes_model_tokenizer_and_research_metadata(pipeline, monkeypatch):
+ saved = []
+
+ class Artifact:
+ def __init__(self, name):
+ self.name = name
+
+ def save_pretrained(self, path):
+ saved.append((self.name, Path(path)))
+
+ pipeline.handle = SimpleNamespace(model=Artifact("model"), tokenizer=Artifact("tokenizer"))
+ pipeline._strong_layers = [1, 3]
+ pipeline._quality_metrics = {"refusal_rate": 0.1}
+ pipeline._insights.detected_alignment_method = "dpo"
+ pipeline._insights.recommended_layers = [1, 3]
+ pipeline._report.analysis_duration = 1.2
+ pipeline._report.total_duration = 3.4
+ pipeline._report.ouroboros_passes = 1
+ pipeline._report.final_refusal_rate = 0.1
+ events = []
+ monkeypatch.setattr(pipeline, "_emit", lambda *args, **kwargs: events.append((args, kwargs)))
+ ticks = iter((5.0, 5.5))
+ monkeypatch.setattr("obliteratus.informed_pipeline.time.time", lambda: next(ticks))
+
+ result = pipeline._rebirth_informed()
+
+ assert result == pipeline.output_dir
+ assert saved == [("model", result), ("tokenizer", result)]
+ metadata = json.loads((result / "abliteration_metadata.json").read_text())
+ assert metadata["analysis_insights"]["detected_alignment_method"] == "dpo"
+ assert metadata["derived_config"]["layers_used"] == [1, 3]
+ assert metadata["pipeline_stats"]["ouroboros_passes"] == 1
+ assert metadata["quality_metrics"] == {"refusal_rate": 0.1}
+ assert events[0][0][:2] == ("rebirth", "running")
+ assert events[-1][0][:2] == ("rebirth", "done")
+
+
+def test_distill_single_direction_honors_recommended_and_skipped_layers(
+ pipeline,
+ monkeypatch,
+):
+ pipeline.handle = SimpleNamespace(hidden_size=4096, total_params=3_000_000_000)
+ pipeline.n_directions = 1
+ pipeline._harmful_means = {
+ 0: torch.tensor([[3.0, 4.0]]),
+ 1: torch.tensor([[0.0, 2.0]]),
+ }
+ pipeline._harmless_means = {idx: torch.zeros(1, 2) for idx in range(2)}
+ pipeline._insights.recommended_layers = [0, 1, 99]
+ pipeline._insights.skip_layers = [1]
+ events = []
+ monkeypatch.setattr(pipeline, "_emit", lambda *args, **kwargs: events.append((args, kwargs)))
+ ticks = iter((1.0, 1.5))
+ monkeypatch.setattr("obliteratus.informed_pipeline.time.time", lambda: next(ticks))
+
+ pipeline._distill_informed()
+
+ assert torch.allclose(pipeline.refusal_directions[0], torch.tensor([0.6, 0.8]))
+ assert pipeline.refusal_subspaces[0].shape == (1, 2)
+ assert pipeline._strong_layers == [0]
+ assert events[0][0][:2] == ("distill", "running")
+ assert events[-1][1]["strong_layers"] == [0]
+
+
+def test_distill_svd_sanitizes_nonfinite_input_and_enriches_category_directions(
+ pipeline,
+ monkeypatch,
+):
+ pipeline.handle = SimpleNamespace(hidden_size=4096, total_params=3_000_000_000)
+ pipeline.n_directions = 3
+ pipeline.use_whitened_svd = False
+ pipeline._harmful_means = {0: torch.tensor([[2.0, 0.0, 0.0]])}
+ pipeline._harmless_means = {0: torch.zeros(1, 3)}
+ pipeline._harmful_acts = {
+ 0: [
+ torch.tensor([[2.0, 0.0, 0.0]]),
+ torch.tensor([[0.0, float("nan"), 0.0]]),
+ torch.tensor([[0.0, 0.0, 1.0]]),
+ ]
+ }
+ pipeline._harmless_acts = {0: [torch.zeros(1, 3) for _ in range(3)]}
+ pipeline._insights.cone_is_polyhedral = True
+ pipeline._insights.per_category_directions = {
+ "one": torch.tensor([0.0, 1.0, 0.0]),
+ "two": torch.tensor([0.0, 0.0, 1.0]),
+ }
+ monkeypatch.setattr(pipeline, "_select_layers_knee", lambda ranked: [ranked[0][0]])
+ monkeypatch.setattr(pipeline, "_emit", lambda *_args, **_kwargs: None)
+
+ pipeline._distill_informed()
+
+ assert pipeline._strong_layers == [0]
+ assert pipeline.refusal_subspaces[0].shape[0] >= 2
+ assert torch.isfinite(pipeline.refusal_subspaces[0]).all()
+
+
+def test_distill_uses_whitened_extractor_for_multi_direction_models(
+ pipeline,
+ monkeypatch,
+):
+ result = SimpleNamespace(
+ directions=torch.tensor([[1.0, 0.0], [0.0, 1.0]]),
+ singular_values=torch.tensor([3.0, 1.0]),
+ )
+
+ class Extractor:
+ def extract(self, harmful, harmless, *, n_directions, layer_idx):
+ assert harmful is pipeline._harmful_acts[0]
+ assert harmless is pipeline._harmless_acts[0]
+ assert (n_directions, layer_idx) == (2, 0)
+ return result
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.whitened_svd.WhitenedSVDExtractor",
+ Extractor,
+ )
+ pipeline.handle = SimpleNamespace(hidden_size=4096, total_params=3_000_000_000)
+ pipeline.n_directions = 2
+ pipeline.use_whitened_svd = True
+ pipeline._harmful_means = {0: torch.tensor([[1.0, 0.0]])}
+ pipeline._harmless_means = {0: torch.zeros(1, 2)}
+ pipeline._harmful_acts = {0: [torch.ones(1, 2)]}
+ pipeline._harmless_acts = {0: [torch.zeros(1, 2)]}
+ monkeypatch.setattr(pipeline, "_select_layers_knee", lambda _ranked: [0])
+ monkeypatch.setattr(pipeline, "_emit", lambda *_args, **_kwargs: None)
+
+ pipeline._distill_informed()
+
+ assert torch.equal(pipeline.refusal_subspaces[0], result.directions)
+ assert torch.equal(pipeline.refusal_directions[0], result.directions[0])
+
+
+def test_distill_leace_falls_back_per_layer_after_extractor_failure(
+ pipeline,
+ monkeypatch,
+):
+ leace_result = SimpleNamespace(
+ direction=torch.tensor([0.0, 1.0]),
+ generalized_eigenvalue=4.0,
+ erasure_loss=0.2,
+ )
+
+ class Extractor:
+ def extract(self, _harmful, _harmless, *, layer_idx):
+ if layer_idx == 0:
+ return leace_result
+ raise RuntimeError("singular fixture")
+
+ monkeypatch.setattr("obliteratus.analysis.leace.LEACEExtractor", Extractor)
+ pipeline.handle = SimpleNamespace(hidden_size=4096, total_params=3_000_000_000)
+ pipeline.direction_method = "leace"
+ pipeline.n_directions = 1
+ pipeline._harmful_means = {
+ 0: torch.tensor([[0.0, 1.0]]),
+ 1: torch.tensor([[1.0, 0.0]]),
+ }
+ pipeline._harmless_means = {idx: torch.zeros(1, 2) for idx in range(2)}
+ pipeline._harmful_acts = {idx: [torch.ones(1, 2)] for idx in range(2)}
+ pipeline._harmless_acts = {idx: [torch.zeros(1, 2)] for idx in range(2)}
+ monkeypatch.setattr(pipeline, "_select_layers_knee", lambda _ranked: [0, 1])
+ monkeypatch.setattr(pipeline, "_emit", lambda *_args, **_kwargs: None)
+
+ pipeline._distill_informed()
+
+ assert torch.equal(pipeline.refusal_directions[0], leace_result.direction)
+ assert torch.equal(pipeline.refusal_directions[1], torch.tensor([1.0, 0.0]))
+
+
+def test_sparse_excision_projects_attention_and_ffn_with_iterative_reprobe(
+ pipeline,
+ monkeypatch,
+):
+ class Layer(torch.nn.Module):
+ def __init__(self):
+ super().__init__()
+ self.anchor = torch.nn.Parameter(torch.ones(1))
+
+ layer = Layer()
+ attention = SimpleNamespace(o_proj=torch.nn.Linear(2, 2, bias=False))
+ ffn = SimpleNamespace(down_proj=torch.nn.Linear(2, 2, bias=False))
+ calls = []
+
+ class Surgeon:
+ def __init__(self, *, sparsity, auto_sparsity):
+ assert sparsity == 0.25
+ assert auto_sparsity is True
+
+ def apply_sparse_projection(self, weight, direction):
+ calls.append((weight.clone(), direction.clone()))
+ return weight * 0.5
+
+ monkeypatch.setattr(
+ "obliteratus.analysis.sparse_surgery.SparseDirectionSurgeon",
+ Surgeon,
+ )
+ monkeypatch.setattr(
+ "obliteratus.strategies.utils.get_layer_modules",
+ lambda _handle: [layer],
+ )
+ monkeypatch.setattr(
+ "obliteratus.strategies.utils.get_attention_module",
+ lambda _layer, _arch: attention,
+ )
+ monkeypatch.setattr(
+ "obliteratus.strategies.utils.get_ffn_module",
+ lambda _layer, _arch: ffn,
+ )
+ pipeline.handle = SimpleNamespace(architecture="gpt2")
+ pipeline._insights.recommended_sparsity = 0.25
+ pipeline._strong_layers = [0]
+ pipeline.refusal_subspaces = {0: torch.eye(2)}
+ pipeline.refinement_passes = 2
+ pipeline.true_iterative_refinement = True
+ monkeypatch.setattr(pipeline, "_probe", lambda: calls.append("probe"))
+ monkeypatch.setattr(pipeline, "_distill_inner", lambda: calls.append("distill"))
+ events = []
+ monkeypatch.setattr(pipeline, "_emit", lambda *args, **kwargs: events.append((args, kwargs)))
+
+ pipeline._excise_sparse()
+
+ projection_calls = [call for call in calls if isinstance(call, tuple)]
+ assert len(projection_calls) == 8
+ assert calls.count("probe") == 1
+ assert calls.count("distill") == 1
+ assert events[-1][1]["modified_count"] == 8
+
+
+def test_verify_compensation_stops_when_kl_rises_sharply(pipeline, monkeypatch):
+ outcomes = iter(
+ (
+ {"refusal_rate": 0.9, "kl_divergence": 0.1},
+ {"refusal_rate": 0.8, "kl_divergence": 0.12},
+ {"refusal_rate": 0.7, "kl_divergence": 0.2},
+ )
+ )
+
+ def verify():
+ pipeline._quality_metrics = next(outcomes)
+
+ monkeypatch.setattr(pipeline, "_verify", verify)
+ monkeypatch.setattr(pipeline, "_probe", lambda: None)
+ monkeypatch.setattr(pipeline, "_distill_inner", lambda: setattr(pipeline, "_strong_layers", [0]))
+ monkeypatch.setattr(pipeline, "_excise_informed", lambda: None)
+ pipeline.kl_budget = 1.0
+
+ pipeline._verify_and_compensate()
+
+ assert pipeline._report.ouroboros_passes == 2
+ assert pipeline._report.final_refusal_rate == 0.7
diff --git a/tests/test_model_profile_contracts.py b/tests/test_model_profile_contracts.py
new file mode 100644
index 0000000..0e47464
--- /dev/null
+++ b/tests/test_model_profile_contracts.py
@@ -0,0 +1,103 @@
+"""Contracts for model-profile estimation and defaults."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from obliteratus.model_profile import (
+ ModelProfile,
+ default_self_improve_params,
+ estimate_active_params_b,
+ estimate_total_params,
+ profile_model,
+)
+
+
+def test_estimate_total_params_prefers_explicit_counts():
+ for key in ("num_parameters", "n_params", "total_params"):
+ cfg = {key: 12345}
+ assert estimate_total_params(cfg) == 12345
+
+
+@pytest.mark.parametrize(
+ "cfg",
+ [
+ {"hidden_size": 0, "num_hidden_layers": 2},
+ {"hidden_size": 128, "num_hidden_layers": 0},
+ {"hidden_size": -1, "num_hidden_layers": 2},
+ ],
+)
+def test_estimate_total_params_rejects_invalid_or_zero_dimensions(cfg):
+ assert estimate_total_params(cfg) is None
+
+
+def test_estimate_total_params_and_active_params_cover_moe_shapes():
+ cfg = {
+ "hidden_size": 4096,
+ "num_hidden_layers": 32,
+ "num_attention_heads": 32,
+ "num_key_value_heads": 8,
+ "head_dim": 128,
+ "intermediate_size": 14336,
+ "num_local_experts": 8,
+ "num_experts_per_tok": 2,
+ "moe_intermediate_size": 28672,
+ "vocab_size": 32000,
+ }
+
+ total = estimate_total_params(cfg)
+ assert total is not None
+ assert total > 0
+
+ active = estimate_active_params_b(cfg, total / 1e9)
+ assert active > 0
+ assert active < total / 1e9
+
+
+def test_profile_model_uses_local_config_when_safetensors_absent(tmp_path):
+ model_dir = tmp_path / "toy"
+ model_dir.mkdir()
+ (model_dir / "config.json").write_text(
+ json.dumps(
+ {
+ "model_type": "toy",
+ "hidden_size": 64,
+ "num_hidden_layers": 2,
+ "num_attention_heads": 4,
+ "intermediate_size": 128,
+ "vocab_size": 320,
+ }
+ )
+ )
+
+ profile = profile_model(str(model_dir), dtype="float16")
+ assert profile.source == "local_config"
+ assert profile.total_params is not None
+ assert profile.total_params > 0
+ assert profile.dtype == "float16"
+
+
+def test_mid_size_defaults_and_modelprofile_round_trip():
+ profile = ModelProfile(
+ model="mid",
+ source="test",
+ total_params=int(10e9),
+ total_params_b=10.0,
+ active_params_b=6.0,
+ num_layers=24,
+ hidden_size=4096,
+ intermediate_size=14336,
+ vocab_size=32000,
+ model_type="qwen",
+ dtype="bfloat16",
+ )
+
+ defaults = default_self_improve_params(profile)
+ assert defaults["n_directions"] == 3
+ assert defaults["refinement_passes"] == 1
+ assert defaults["verify_sample_size"] == 40
+ assert defaults["residue_weight"] == 5
+
+ assert profile.to_json()["total_params"] == int(10e9)
diff --git a/tests/test_offline_integration.py b/tests/test_offline_integration.py
index 90e3723..83238ff 100644
--- a/tests/test_offline_integration.py
+++ b/tests/test_offline_integration.py
@@ -10,6 +10,7 @@ from pathlib import Path
import pytest
import torch
+import yaml
from datasets import Dataset
from transformers import AutoModelForCausalLM, AutoTokenizer
@@ -21,6 +22,53 @@ 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]
+
+
+def _offline_cli_env(home: Path) -> dict[str, str]:
+ return {
+ **os.environ,
+ "HOME": str(home),
+ "HF_HOME": str(home / "hf"),
+ "HF_DATASETS_OFFLINE": "1",
+ "HF_HUB_DISABLE_TELEMETRY": "1",
+ "HF_HUB_OFFLINE": "1",
+ "TRANSFORMERS_OFFLINE": "1",
+ }
+
+
+def _obliteratus_import_origin(workdir: Path, env: dict[str, str]) -> Path:
+ origin = subprocess.run(
+ [sys.executable, "-I", "-c", "import obliteratus; print(obliteratus.__file__)"],
+ cwd=workdir,
+ env=env,
+ check=True,
+ capture_output=True,
+ text=True,
+ timeout=30,
+ )
+ return Path(origin.stdout.strip()).resolve()
+
+
+def _package_origin_mode(origin: Path) -> str:
+ if REPOSITORY_ROOT in origin.parents:
+ return "source-backed"
+ if {"site-packages", "dist-packages"} & set(origin.parts):
+ return "installed-artifact"
+ return "unknown"
+
+
+def _require_installed_artifact_import(origin: Path) -> None:
+ mode = _package_origin_mode(origin)
+ if mode == "source-backed":
+ pytest.xfail(
+ "current interpreter imports obliteratus from the source checkout; "
+ "the installed-artifact CLI contract requires a non-editable or wheel install"
+ )
+ assert mode == "installed-artifact", (
+ "obliteratus must import from an installed artifact for this contract; "
+ f"origin={origin}"
+ )
def _state_dict(path: Path) -> dict[str, torch.Tensor]:
@@ -105,26 +153,10 @@ def test_installed_wheel_cli_loads_local_model_without_repository_imports(tmp_pa
isolated_workdir.mkdir()
isolated_home = tmp_path / "home"
isolated_home.mkdir()
- env = {
- **os.environ,
- "HOME": str(isolated_home),
- "HF_HOME": str(isolated_home / "hf"),
- "HF_DATASETS_OFFLINE": "1",
- "HF_HUB_DISABLE_TELEMETRY": "1",
- "HF_HUB_OFFLINE": "1",
- "TRANSFORMERS_OFFLINE": "1",
- }
+ env = _offline_cli_env(isolated_home)
- origin = subprocess.run(
- [sys.executable, "-I", "-c", "import obliteratus; print(obliteratus.__file__)"],
- cwd=isolated_workdir,
- env=env,
- check=True,
- capture_output=True,
- text=True,
- timeout=30,
- )
- assert "site-packages" in origin.stdout
+ origin = _obliteratus_import_origin(isolated_workdir, env)
+ _require_installed_artifact_import(origin)
result = subprocess.run(
[
@@ -150,6 +182,98 @@ def test_installed_wheel_cli_loads_local_model_without_repository_imports(tmp_pa
assert "gpt2" in result.stdout.lower()
+def test_installed_package_cli_executes_offline_checkpoint_to_report_slice(tmp_path):
+ source = build_tiny_offline_model(tmp_path / "source")
+ checkpoint = tmp_path / "checkpoint"
+ dataset_dir = tmp_path / "dataset"
+ dataset_dir.mkdir()
+ (dataset_dir / "samples.jsonl").write_text(
+ json.dumps({"text": "hello world safe test"}) + "\n",
+ )
+ study_output = tmp_path / "study-results"
+ config_path = tmp_path / "study.yaml"
+ config_path.write_text(
+ yaml.safe_dump(
+ {
+ "model": {
+ "name": str(checkpoint),
+ "device": "cpu",
+ "dtype": "float32",
+ },
+ "dataset": {
+ "name": str(dataset_dir),
+ "split": "train",
+ "max_samples": 1,
+ },
+ "strategies": [{"name": "layer_removal"}],
+ "metrics": ["perplexity"],
+ "batch_size": 1,
+ "max_length": 8,
+ "output_dir": str(study_output),
+ },
+ ),
+ )
+ isolated_workdir = tmp_path / "outside-repository"
+ isolated_workdir.mkdir()
+ isolated_home = tmp_path / "home"
+ isolated_home.mkdir()
+ env = _offline_cli_env(isolated_home)
+ origin = _obliteratus_import_origin(isolated_workdir, env)
+ _require_installed_artifact_import(origin)
+
+ subprocess.run(
+ [
+ sys.executable,
+ "-I",
+ "-m",
+ "obliteratus",
+ "obliterate",
+ str(source),
+ "--output-dir",
+ str(checkpoint),
+ "--device",
+ "cpu",
+ "--dtype",
+ "float32",
+ "--method",
+ "basic",
+ "--n-directions",
+ "1",
+ "--refinement-passes",
+ "1",
+ "--verify-sample-size",
+ "1",
+ "--refusal-max-tokens",
+ "1",
+ ],
+ cwd=isolated_workdir,
+ env=env,
+ check=True,
+ capture_output=True,
+ text=True,
+ timeout=120,
+ )
+ assert (checkpoint / "abliteration_metadata.json").is_file()
+ AutoModelForCausalLM.from_pretrained(checkpoint, local_files_only=True)
+
+ subprocess.run(
+ [sys.executable, "-I", "-m", "obliteratus", "run", str(config_path)],
+ cwd=isolated_workdir,
+ env=env,
+ check=True,
+ capture_output=True,
+ text=True,
+ timeout=120,
+ )
+
+ report = json.loads((study_output / "results.json").read_text())
+ assert report["model_name"] == checkpoint.name
+ assert report["baseline_metrics"]["perplexity"] > 0
+ assert len(report["results"]) == 1
+ assert report["results"][0]["strategy"] == "layer_removal"
+ assert (study_output / "results.csv").is_file()
+
+
def test_study_runner_evaluates_ablates_restores_and_reports(
tmp_path,
monkeypatch,
diff --git a/tests/test_package_export_contracts.py b/tests/test_package_export_contracts.py
new file mode 100644
index 0000000..94b0150
--- /dev/null
+++ b/tests/test_package_export_contracts.py
@@ -0,0 +1,35 @@
+"""Contracts for every documented lazy package export."""
+
+from __future__ import annotations
+
+import pytest
+
+import obliteratus
+
+
+@pytest.mark.parametrize(
+ "name",
+ [
+ "AbliterationPipeline",
+ "InformedAbliterationPipeline",
+ "save_contribution",
+ "load_contributions",
+ "aggregate_results",
+ "TourneyRunner",
+ "TourneyResult",
+ "get_adaptive_recommendation",
+ "AdaptiveRecommendation",
+ "RemoteRunner",
+ "RemoteConfig",
+ "Watchtower",
+ "get_watchtower",
+ "AutoObliterator",
+ ],
+)
+def test_documented_lazy_export_resolves(name):
+ assert getattr(obliteratus, name) is not None
+
+
+def test_unknown_lazy_export_raises_attribute_error():
+ with pytest.raises(AttributeError, match="has no attribute 'not_an_export'"):
+ getattr(obliteratus, "not_an_export")
diff --git a/tests/test_quality_policy.py b/tests/test_quality_policy.py
index 277f468..c91299a 100644
--- a/tests/test_quality_policy.py
+++ b/tests/test_quality_policy.py
@@ -67,9 +67,9 @@ def _coverage():
"obliteratus/pure.py": {
"summary": {
"num_statements": 100,
- "covered_lines": 90,
+ "covered_lines": 92,
"num_branches": 100,
- "covered_branches": 78,
+ "covered_branches": 80,
},
},
"obliteratus/external.py": {
@@ -89,8 +89,8 @@ def test_policy_and_exact_mature_floors_pass():
assert quality.validate_policy(policy) == []
measurement, failures = quality.validate_mature_cpu_scope(_coverage(), policy)
assert failures == []
- assert measurement["line_percent"] == 90
- assert measurement["branch_percent"] == 78
+ assert measurement["line_percent"] == 92
+ assert measurement["branch_percent"] == 80
def test_floor_regression_requires_structured_reviewed_exception():
@@ -122,10 +122,10 @@ def test_exclusions_require_unique_traceable_environment_boundaries():
def test_mature_scope_rejects_regression_and_stale_exclusion():
policy = _policy()
report = _coverage()
- report["files"]["obliteratus/pure.py"]["summary"]["covered_lines"] = 89
+ report["files"]["obliteratus/pure.py"]["summary"]["covered_lines"] = 91
_, failures = quality.validate_mature_cpu_scope(report, policy)
assert failures == [
- "mature CPU line coverage 89.00% is below the 90.00% floor",
+ "mature CPU line coverage 91.00% is below the 92.00% floor",
]
del report["files"]["obliteratus/external.py"]
_, failures = quality.measure_mature_cpu_scope(report, policy)
diff --git a/tests/test_remaining_cpu_contracts.py b/tests/test_remaining_cpu_contracts.py
new file mode 100644
index 0000000..db3fc33
--- /dev/null
+++ b/tests/test_remaining_cpu_contracts.py
@@ -0,0 +1,164 @@
+"""Cheap deterministic contracts for remaining CPU-only surfaces."""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+import torch
+
+from obliteratus.auto_obliterate import AutoObliterator, IterationResult
+from obliteratus.evaluation.advanced_metrics import (
+ AbliterationEvalResult,
+ _is_degenerate,
+ _is_refusal_detailed,
+ activation_cosine_similarity,
+ effective_rank,
+ format_eval_report,
+ linear_cka,
+ refusal_projection_magnitude,
+ refusal_rate_with_ci,
+ token_kl_divergence,
+)
+from obliteratus.evaluation.benchmarks import BenchmarkRunner
+
+
+class _TinyTokenizer:
+ def __call__(self, prompt, return_tensors="pt", truncation=True, max_length=256):
+ return {"input_ids": torch.tensor([[1, 2, 3]])}
+
+ def encode(self, letter, add_special_tokens=False):
+ return [ord(letter)]
+
+ def decode(self, tokens, skip_special_tokens=True):
+ return ""
+
+
+class _TinyModel:
+ def __init__(self):
+ self._p = torch.nn.Parameter(torch.zeros(1))
+
+ def parameters(self):
+ return iter([self._p])
+
+ def __call__(self, **_inputs):
+ return SimpleNamespace(logits=torch.tensor([[[0.1, 0.9, 0.0, -0.1]]]))
+
+ def generate(self, **_inputs):
+ return torch.tensor([[1, 2, 3, 4]])
+
+
+def test_auto_obliterate_resume_loads_valid_state_and_saves_round_trip(tmp_path, monkeypatch):
+ monkeypatch.setattr("obliteratus.auto_obliterate.Path.home", staticmethod(lambda: tmp_path))
+ output_base = tmp_path / "auto"
+ state_file = output_base / "auto_state.json"
+ state_file.parent.mkdir(parents=True)
+ state_file.write_text(
+ json.dumps(
+ {
+ "model_id": "demo",
+ "iterations": [
+ {
+ "iteration": 1,
+ "method": "aggressive",
+ "prompt_volume": 4,
+ "time_seconds": 1.0,
+ "output_dir": str(output_base / "iter_1"),
+ }
+ ],
+ }
+ )
+ )
+
+ ob = AutoObliterator("demo", max_iterations=2, output_base=str(output_base))
+ assert ob._resume_from == 1
+ assert len(ob._result.iterations) == 1
+
+ ob._result.iterations.append(
+ IterationResult(iteration=2, method="nuclear", prompt_volume=8, output_dir=str(output_base / "iter_2"))
+ )
+ ob._save_state()
+ saved = json.loads(state_file.read_text())
+ assert saved["iterations"][0]["iteration"] == 1
+ assert saved["iterations"][1]["method"] == "nuclear"
+
+
+def test_auto_obliterate_quarantines_invalid_state_and_prompt_fallback(monkeypatch, tmp_path):
+ monkeypatch.setattr("obliteratus.auto_obliterate.Path.home", staticmethod(lambda: tmp_path))
+ output_base = tmp_path / "auto"
+ state_file = output_base / "auto_state.json"
+ state_file.parent.mkdir(parents=True)
+ state_file.write_text("{not-json")
+
+ ob = AutoObliterator("demo", output_base=str(output_base))
+ quarantined = list(output_base.glob("auto_state.json.corrupt-*"))
+ assert quarantined
+ assert ob._resume_from == 0
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "obliteratus.prompts",
+ SimpleNamespace(
+ load_dataset_source=Mock(side_effect=RuntimeError("boom")),
+ DEFAULT_HARMFUL=("h1",),
+ DEFAULT_HARMLESS=("n1",),
+ ),
+ )
+ harmful, harmless = AutoObliterator._get_expanded_prompts(0)
+ assert harmful == ["h1"]
+ assert harmless == ["n1"]
+
+
+def test_benchmark_runner_generation_truthfulness_and_number_branches():
+ runner = BenchmarkRunner(_TinyModel(), _TinyTokenizer(), device="cpu")
+ assert runner._generate_short("prompt") == ""
+ assert runner._check_truthfulness("No, that is a myth.", "no", "yes") is True
+ assert runner._check_truthfulness("yes", "no", "yes") is False
+ assert runner._extract_number("The answer is 42.") == 42.0
+ assert runner._extract_number("no number here") is None
+
+
+def test_advanced_metrics_tensor_validation_refusal_and_report_edges():
+ with pytest.raises(ValueError, match="identical shapes"):
+ token_kl_divergence(torch.zeros(1, 2, 3), torch.zeros(1, 3, 3))
+
+ with pytest.raises(ValueError, match="must not be empty"):
+ token_kl_divergence(torch.zeros(1, 0, 3), torch.zeros(1, 0, 3))
+
+ with pytest.raises(ValueError, match="same sample count"):
+ linear_cka(torch.zeros(1, 2), torch.zeros(2, 2))
+
+ with pytest.raises(ValueError, match="non-zero norm"):
+ refusal_projection_magnitude(torch.ones(2, 3), torch.zeros(3))
+
+ assert _is_degenerate("!!!!!")
+ assert _is_refusal_detailed("I cannot help with that.", "combined")[0] is True
+ assert _is_refusal_detailed(" ", "combined") == (False, "")
+
+ report = format_eval_report(
+ AbliterationEvalResult(
+ refusal_rate_harmful=0.0,
+ refusal_rate_harmless=None,
+ kl_divergence=0.1,
+ perplexity=None,
+ coherence_score=0.2,
+ mean_activation_cosine=None,
+ mean_cka=0.3,
+ )
+ )
+ assert "unavailable" in report
+ assert "excellent" in report
+
+ assert refusal_rate_with_ci([], mode="combined") == {
+ "available": False,
+ "rate": None,
+ "ci_lower": None,
+ "ci_upper": None,
+ "n_samples": 0,
+ "refusal_count": 0,
+ }
+
+ assert effective_rank(torch.eye(2)) == pytest.approx(2.0)
+ assert activation_cosine_similarity(torch.ones(2, 3), torch.ones(2, 3)) == pytest.approx(1.0)
diff --git a/tests/test_strategy_navigation_contracts.py b/tests/test_strategy_navigation_contracts.py
new file mode 100644
index 0000000..40eb2ca
--- /dev/null
+++ b/tests/test_strategy_navigation_contracts.py
@@ -0,0 +1,126 @@
+"""Contracts for strategy navigation and head/embedding fallback behavior."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+import torch
+from torch import nn
+
+from obliteratus.models.loader import ModelHandle
+from obliteratus.strategies.base import AblationSpec
+from obliteratus.strategies.head_pruning import HeadPruningStrategy
+from obliteratus.strategies.utils import (
+ get_attention_module,
+ get_embedding_module,
+ get_ffn_module,
+ get_layer_modules,
+)
+
+
+class _DummyTokenizer:
+ pad_token = ""
+ eos_token = ""
+
+
+class _LlamaLayer(nn.Module):
+ def __init__(self):
+ super().__init__()
+ self.self_attn = nn.Module()
+ self.self_attn.q_proj = nn.Linear(8, 8, bias=True)
+ self.self_attn.k_proj = nn.Linear(8, 8, bias=True)
+ self.self_attn.v_proj = nn.Linear(8, 8, bias=True)
+ self.self_attn.o_proj = nn.Linear(8, 8, bias=True)
+ self.mlp = nn.Module()
+ self.mlp.down_proj = nn.Linear(8, 8, bias=True)
+
+
+class _Qwen35MoeLayer(nn.Module):
+ def __init__(self, *, with_primary_attn: bool):
+ super().__init__()
+ if with_primary_attn:
+ self.self_attn = nn.Module()
+ else:
+ self.linear_attn = nn.Module()
+ self.mlp = nn.Module()
+
+
+class _Qwen35MoeModel(nn.Module):
+ def __init__(self):
+ super().__init__()
+ self.model = nn.Module()
+ self.model.layers = nn.ModuleList(
+ [_Qwen35MoeLayer(with_primary_attn=True), _Qwen35MoeLayer(with_primary_attn=False)]
+ )
+ self.model.embed_tokens = nn.Embedding(32, 8)
+
+
+class _NoEmbeddingModel(nn.Module):
+ def __init__(self):
+ super().__init__()
+ self.model = nn.Module()
+ self.model.layers = nn.ModuleList([nn.Module()])
+
+
+def _handle(model: nn.Module, *, architecture: str, hidden_size: int = 8, num_layers: int = 1, num_heads: int = 2):
+ return ModelHandle(
+ model=model,
+ tokenizer=_DummyTokenizer(),
+ config=SimpleNamespace(
+ model_type=architecture,
+ hidden_size=hidden_size,
+ num_hidden_layers=num_layers,
+ num_attention_heads=num_heads,
+ intermediate_size=hidden_size * 4,
+ ),
+ model_name="test-model",
+ task="causal_lm",
+ )
+
+
+def test_strategy_navigation_resolves_fallback_layers_and_missing_attention():
+ handle = _handle(_Qwen35MoeModel(), architecture="qwen3_5_moe", num_layers=2)
+
+ layers = get_layer_modules(handle)
+ assert len(layers) == 2
+ assert get_attention_module(layers[0], handle.architecture) is layers[0].self_attn
+ assert get_attention_module(layers[1], handle.architecture) is layers[1].linear_attn
+ assert get_ffn_module(layers[0], handle.architecture) is layers[0].mlp
+
+ broken = nn.Module()
+ with pytest.raises(AttributeError):
+ get_attention_module(broken, "qwen3_5_moe")
+
+
+def test_head_pruning_zeros_qkv_and_output_slices_for_standard_attention():
+ model = nn.Module()
+ model.model = nn.Module()
+ model.model.layers = nn.ModuleList([_LlamaLayer()])
+ handle = _handle(model, architecture="llama")
+
+ spec = AblationSpec(
+ strategy_name="head_pruning",
+ component="layer_0_head_1",
+ description="test",
+ metadata={"layer_idx": 0, "head_idx": 1},
+ )
+ HeadPruningStrategy().apply(handle, spec)
+
+ attn = get_attention_module(get_layer_modules(handle)[0], handle.architecture)
+ head_dim = handle.hidden_size // handle.num_heads
+ start = head_dim
+ end = start + head_dim
+ for proj_name in ("q_proj", "k_proj", "v_proj"):
+ proj = getattr(attn, proj_name)
+ assert torch.all(proj.weight[start:end, :] == 0)
+ assert torch.all(proj.bias[start:end] == 0)
+ assert torch.all(attn.o_proj.weight[:, start:end] == 0)
+
+
+def test_embedding_navigation_uses_first_embedding_and_fails_without_one():
+ handle = _handle(_Qwen35MoeModel(), architecture="qwen3_5_moe")
+ assert get_embedding_module(handle) is handle.model.model.embed_tokens
+
+ with pytest.raises(RuntimeError, match="Cannot locate embedding module"):
+ get_embedding_module(_handle(_NoEmbeddingModel(), architecture="qwen3_5_moe"))
diff --git a/tests/test_sweep_contracts.py b/tests/test_sweep_contracts.py
new file mode 100644
index 0000000..511da31
--- /dev/null
+++ b/tests/test_sweep_contracts.py
@@ -0,0 +1,67 @@
+"""Deterministic orchestration contracts for hyperparameter sweeps."""
+
+from __future__ import annotations
+
+import json
+
+from obliteratus.sweep import SweepConfig, _param_grid, run_sweep
+
+
+def test_param_grid_is_stable_and_crosses_sorted_keys():
+ assert _param_grid({"zeta": [1, 2], "alpha": ["a", "b"]}) == [
+ {"alpha": "a", "zeta": 1},
+ {"alpha": "a", "zeta": 2},
+ {"alpha": "b", "zeta": 1},
+ {"alpha": "b", "zeta": 2},
+ ]
+
+
+def test_run_sweep_records_success_failure_seeds_and_incremental_json(
+ tmp_path,
+ monkeypatch,
+):
+ created = []
+
+ class Pipeline:
+ def __init__(self, **kwargs):
+ self.kwargs = kwargs
+ self._quality_metrics = {"score": kwargs["seed"]}
+ self._stage_durations = {"probe": 0.25}
+ self._strong_layers = [1, 3]
+ created.append(self)
+
+ def run(self):
+ if self.kwargs["strength"] == 2 and self.kwargs["seed"] == 11:
+ raise RuntimeError("intentional sweep failure")
+
+ monkeypatch.setattr("obliteratus.abliterate.AbliterationPipeline", Pipeline)
+ output = tmp_path / "sweep"
+ config = SweepConfig(
+ model_name="fixture/model",
+ sweep_params={"strength": [1, 2]},
+ fixed_params={"method": "basic"},
+ output_dir=str(output),
+ seed=10,
+ n_seeds=2,
+ )
+
+ results = run_sweep(config)
+
+ assert len(results) == 4
+ assert [result.seed for result in results] == [10, 11, 10, 11]
+ assert results[0].params == {"strength": 1}
+ assert results[0].quality_metrics == {"score": 10}
+ assert results[0].stage_durations == {"probe": 0.25}
+ assert results[0].strong_layers == [1, 3]
+ assert results[-1].error == "intentional sweep failure"
+ assert results[-1].quality_metrics == {}
+ assert [item.kwargs["output_dir"] for item in created] == [
+ str(output / f"run_{index:03d}") for index in range(4)
+ ]
+ assert all(item.kwargs["model_name"] == "fixture/model" for item in created)
+ assert all(item.kwargs["method"] == "basic" for item in created)
+
+ saved = json.loads((output / "sweep_results.json").read_text())
+ assert len(saved) == 4
+ assert saved[0]["quality_metrics"] == {"score": 10}
+ assert saved[-1]["error"] == "intentional sweep failure"