feat(surgery): add validated Qwen3.8 output-writer contract

This commit is contained in:
Joseph Magly
2026-08-26 22:32:14 -04:00
parent f09ae6b276
commit 5aa11828a1
10 changed files with 465 additions and 25 deletions
+2
View File
@@ -55,6 +55,7 @@
],
"paths": [
"obliteratus/abliterate.py",
"obliteratus/models/qwen35_contracts.py",
"obliteratus/models/offload_surgery.py",
"obliteratus/persistence_contracts.py",
"obliteratus/auto_obliterate.py",
@@ -71,6 +72,7 @@
"required_tests": [
"tests/test_abliterate.py",
"tests/test_abliterate_extended.py",
"tests/test_qwen35_contracts.py",
"tests/test_auto_obliterate.py",
"tests/test_bayesian_optimizer_contracts.py",
"tests/test_informed_pipeline.py",
+53 -19
View File
@@ -37,6 +37,11 @@ import torch.nn as nn
from obliteratus import device as dev # noqa: E402 — must import before CUDA setup
from obliteratus.models.quant_dequant import FP8_DTYPES as _FP8_DTYPES
from obliteratus.models.qwen35_contracts import (
Qwen35ContractError,
Qwen35LayerTargets,
validate_qwen38_27b_projection_contract,
)
# Module attributes that hold quantization scale tensors; their presence on a
# module means a uint8/float8 ``weight`` is packed quantized data, not a plain
@@ -1622,12 +1627,20 @@ class AbliterationPipeline:
"deterministic coherence prompt",
stage="baseline",
)
self._fail_validation(
"architecture_support",
0.0,
"Qwen3.5/Qwen3.8 hybrid Gated DeltaNet surgery has no validated "
"projection allowlist; no weights were modified",
stage="baseline",
try:
manifest = validate_qwen38_27b_projection_contract(self.handle)
except Qwen35ContractError as error:
self._fail_validation(
"architecture_support",
0.0,
f"Qwen3.5/Qwen3.8 hybrid surgery remains blocked: {error}; "
"no weights were modified",
stage="baseline",
)
self._qwen35_projection_manifest = manifest
self.log(
"Validated Qwen3.8-27B text-only surgery manifest: "
"48 linear_attn.out_proj + 16 self_attn.o_proj + 64 mlp.down_proj"
)
def _enforce_perplexity_guardrail(self, perplexity: float) -> None:
@@ -3965,6 +3978,9 @@ class AbliterationPipeline:
total_modified = 0
total_neurons_masked = 0
total_sae_projections = 0
qwen_manifest: tuple[Qwen35LayerTargets, ...] | None = getattr(
self, "_qwen35_projection_manifest", None
)
# Resolve every selected layer's meta-resident parameter before any
# surgery. Unknown or unsupported Accelerate layouts therefore fail
@@ -4235,7 +4251,12 @@ class AbliterationPipeline:
multi_dir = subspace.shape[0] > 1 and self.norm_preserve
saved_layer_norms: dict[str, float] = {}
if multi_dir:
saved_layer_norms = self._capture_layer_weight_norms(layers[idx])
allowed_norms = (
qwen_manifest[idx].parameter_names if qwen_manifest else None
)
saved_layer_norms = self._capture_layer_weight_norms(
layers[idx], parameter_names=allowed_norms
)
# Disable per-direction norm preservation when doing multi-
# direction subspace projection (will restore once afterward)
@@ -4273,14 +4294,17 @@ class AbliterationPipeline:
_text_cfg or config, "num_kv_shared_layers",
getattr(config, "num_kv_shared_layers", 0)
) or 0
_attn_names = attention_projection_names(
projection_target=self.projection_target,
layer_index=idx,
num_layers=len(layers),
num_kv_shared_layers=_kv_shared_layers,
output_names=_ATTN_OUT_NAMES,
input_names=_ATTN_IN_NAMES,
)
if qwen_manifest:
_attn_names = [qwen_manifest[idx].mixer_output]
else:
_attn_names = attention_projection_names(
projection_target=self.projection_target,
layer_index=idx,
num_layers=len(layers),
num_kv_shared_layers=_kv_shared_layers,
output_names=_ATTN_OUT_NAMES,
input_names=_ATTN_IN_NAMES,
)
count += self._project_out_advanced(
attn, d, _attn_names,
norm_preserve=dir_norm_preserve,
@@ -4372,7 +4396,7 @@ class AbliterationPipeline:
projection_row_fraction=self.projection_row_fraction,
offload_roots=(layers[idx],),
)
elif self.projection_target == "output":
elif self.projection_target == "output" or qwen_manifest:
if self.project_biases:
ffn_count += self._project_bias(
ffn,
@@ -4591,7 +4615,7 @@ class AbliterationPipeline:
# Project using the direction from the last strong layer (closest
# to the output).
lm_head_count = 0
if self._strong_layers and self.handle:
if self._strong_layers and self.handle and not qwen_manifest:
last_strong = max(self._strong_layers)
model = self.handle.model
if last_strong in self.refusal_subspaces:
@@ -4653,7 +4677,12 @@ class AbliterationPipeline:
# of the full reflection strength. Only the PRIMARY direction is
# projected to limit representation damage.
embed_count = 0
if self.project_embeddings and self._strong_layers and self.handle:
if (
self.project_embeddings
and self._strong_layers
and self.handle
and not qwen_manifest
):
first_strong = min(self._strong_layers)
model = self.handle.model
if first_strong in self.refusal_directions:
@@ -5385,7 +5414,10 @@ class AbliterationPipeline:
)
@staticmethod
def _capture_layer_weight_norms(layer: nn.Module) -> dict[str, float]:
def _capture_layer_weight_norms(
layer: nn.Module,
parameter_names: frozenset[str] | None = None,
) -> dict[str, float]:
"""Capture Frobenius norms of ALL weight matrices in a transformer layer.
Used for correct multi-direction norm preservation: capture once before
@@ -5416,6 +5448,8 @@ class AbliterationPipeline:
if identity in seen:
continue
param_name = f"{module_name}.weight" if module_name else "weight"
if parameter_names is not None and param_name not in parameter_names:
continue
if weight is not None and weight.device.type == "meta":
data = resolve_logical_parameter(
module,
+19
View File
@@ -173,9 +173,11 @@ def detect_architecture(
num_active_experts = 0
total_params_b = 0.0
is_reasoning = False
is_qwen35_hybrid = False
# ── Step 1: Extract info from config if available ────────────────
if config is not None:
is_qwen35_hybrid = str(getattr(config, "model_type", "")).startswith("qwen3_5")
# Mistral 3 is a multimodal wrapper whose text_config may be either a
# dense Mistral backbone or a Mistral 4 MoE backbone. Profile the text
# config when present without replacing the outer config used by the
@@ -183,6 +185,7 @@ def detect_architecture(
text_config = getattr(config, "text_config", None)
profile_config = text_config or config
model_type = getattr(profile_config, "model_type", "")
is_qwen35_hybrid = is_qwen35_hybrid or str(model_type).startswith("qwen3_5")
# Check for MoE via config attributes
for attr in _TOTAL_EXPERT_ATTRS:
@@ -252,6 +255,8 @@ def detect_architecture(
if pattern_re.search(name_lower):
is_reasoning = True
break
if is_qwen35_hybrid:
is_reasoning = True
# ── Step 3: Classify architecture ────────────────────────────────
size = classify_architecture_size(
@@ -282,6 +287,20 @@ def detect_architecture(
)
_apply_recommended_defaults(profile)
if is_qwen35_hybrid:
profile.profile_label = "Validated Qwen3.5 Hybrid"
profile.profile_description = (
"Hybrid Gated DeltaNet/full-attention reasoning model. Use the "
"architecture-validated output-writer manifest; generic aggressive "
"input/gate projection is unsupported."
)
profile.recommended_method = "advanced"
profile.method_overrides = {
"n_directions": 4,
"refinement_passes": 1,
"use_chat_template": True,
"projection_target": "all",
}
return profile
+16
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import logging
import os
import re
import tempfile
from dataclasses import dataclass, field
from typing import Optional
@@ -808,6 +809,21 @@ def load_model(
if task == "classification":
config.num_labels = num_labels
load_kwargs["config"] = config
if task == "causal_lm" and getattr(config, "model_type", "") == "qwen3_5":
import transformers
version_match = re.match(r"^(\d+)\.(\d+)", transformers.__version__)
version = tuple(map(int, version_match.groups())) if version_match else (0, 0)
if version < (5, 15):
raise RuntimeError(
"Qwen3.8 requires transformers>=5.15 for the validated hybrid "
"runtime and multidimensional-position fixes"
)
load_kwargs["attn_implementation"] = "sdpa"
logger.info(
"Loading Qwen3.8 through AutoModelForCausalLM as an explicit text-only "
"derivative; vision and MTP checkpoint tensors are not part of the output."
)
# Quantization support (requires bitsandbytes)
if load_policy.quantization_backend == "native":
+107
View File
@@ -0,0 +1,107 @@
"""Fail-closed projection contract for the Qwen3.8-27B text backbone."""
from __future__ import annotations
from dataclasses import dataclass
import torch.nn as nn
from obliteratus.models.loader import ModelHandle
from obliteratus.strategies.utils import get_layer_modules
class Qwen35ContractError(RuntimeError):
"""The loaded Qwen hybrid does not match the validated surgery layout."""
@dataclass(frozen=True)
class Qwen35LayerTargets:
"""Semantic residual-writer targets for one decoder layer."""
mixer_attribute: str
mixer_output: str
ffn_attribute: str = "mlp"
ffn_output: str = "down_proj"
@property
def parameter_names(self) -> frozenset[str]:
return frozenset(
{
f"{self.mixer_attribute}.{self.mixer_output}.weight",
f"{self.ffn_attribute}.{self.ffn_output}.weight",
}
)
_QWEN38_27B_LAYER_TYPES = tuple(
layer_type
for _ in range(16)
for layer_type in ("linear_attention", "linear_attention", "linear_attention", "full_attention")
)
def validate_qwen38_27b_projection_contract(
handle: ModelHandle,
) -> tuple[Qwen35LayerTargets, ...]:
"""Validate and return the complete Qwen3.8-27B output-writer manifest.
This intentionally supports only the researched dense 27B topology. Other
Qwen3.5/Qwen3.8 sizes, MoE variants, or changed module shapes remain blocked.
"""
if str(handle.architecture).lower() != "qwen3_5":
raise Qwen35ContractError(
f"unsupported Qwen hybrid architecture {handle.architecture!r}"
)
model_id = str(getattr(handle, "model_name", "")).rstrip("/").lower()
if model_id != "qwen/qwen3.8-27b":
raise Qwen35ContractError(
"validated Qwen hybrid surgery currently supports only Qwen/Qwen3.8-27B"
)
text_config = getattr(handle.config, "text_config", None)
if text_config is None:
raise Qwen35ContractError("Qwen3.8-27B is missing its nested text_config")
layer_types = tuple(getattr(text_config, "layer_types", ()) or ())
if layer_types != _QWEN38_27B_LAYER_TYPES:
raise Qwen35ContractError(
"Qwen3.8-27B layer_types do not match the validated 48 DeltaNet / "
"16 full-attention topology"
)
if (
getattr(text_config, "hidden_size", None) != 5120
or getattr(text_config, "intermediate_size", None) != 17408
or getattr(text_config, "num_hidden_layers", None) != 64
):
raise Qwen35ContractError("Qwen3.8-27B text dimensions do not match the validated contract")
layers = get_layer_modules(handle)
if len(layers) != len(layer_types):
raise Qwen35ContractError(
f"Qwen3.8-27B exposes {len(layers)} layers; expected {len(layer_types)}"
)
targets: list[Qwen35LayerTargets] = []
for index, (layer, layer_type) in enumerate(zip(layers, layer_types, strict=True)):
if layer_type == "linear_attention":
target = Qwen35LayerTargets("linear_attn", "out_proj")
expected_mixer_shape = (5120, 6144)
else:
target = Qwen35LayerTargets("self_attn", "o_proj")
expected_mixer_shape = (5120, 6144)
try:
mixer_output = getattr(getattr(layer, target.mixer_attribute), target.mixer_output)
ffn_output = getattr(getattr(layer, target.ffn_attribute), target.ffn_output)
except AttributeError as error:
raise Qwen35ContractError(
f"Qwen3.8-27B layer {index} is missing a required residual writer"
) from error
if not isinstance(mixer_output, nn.Linear) or tuple(mixer_output.weight.shape) != expected_mixer_shape:
raise Qwen35ContractError(
f"Qwen3.8-27B layer {index} mixer output has unexpected type or shape"
)
if not isinstance(ffn_output, nn.Linear) or tuple(ffn_output.weight.shape) != (5120, 17408):
raise Qwen35ContractError(
f"Qwen3.8-27B layer {index} MLP output has unexpected type or shape"
)
targets.append(target)
return tuple(targets)
@@ -76,12 +76,12 @@ def test_pinned_mistral4_config_resolves_composite_contract_without_remote_code(
@pytest.mark.gpu
def test_qwen38_bf16_pristine_baseline_blocks_unvalidated_surgery(tmp_path):
"""Operator-gated 27B regression: healthy stock model, zero modified weights."""
def test_qwen38_bf16_pristine_baseline_validates_projection_contract(tmp_path):
"""Operator-gated 27B regression: healthy stock model and exact manifest."""
if os.environ.get("OBLITERATUS_QWEN38_E2E") != "1":
pytest.skip("set OBLITERATUS_QWEN38_E2E=1 on a >=80 GiB GPU runner")
from obliteratus.abliterate import AbliterationPipeline, PipelineValidationError
from obliteratus.abliterate import AbliterationPipeline
pipeline = AbliterationPipeline(
QWEN38_MODEL,
@@ -93,6 +93,10 @@ def test_qwen38_bf16_pristine_baseline_blocks_unvalidated_surgery(tmp_path):
pipeline._capture_stock_baseline()
assert pipeline._stock_baseline["perplexity"] > 0
assert pipeline._stock_baseline["coherence"] > 0
with pytest.raises(PipelineValidationError, match="no validated projection allowlist"):
pipeline._validate_architecture_surgery_support()
pipeline._validate_architecture_surgery_support()
assert len(pipeline._qwen35_projection_manifest) == 64
assert sum(
target.mixer_attribute == "linear_attn"
for target in pipeline._qwen35_projection_manifest
) == 48
assert pipeline._excise_modified_count is None
+1 -1
View File
@@ -241,7 +241,7 @@ class TestPipelineInit:
pipeline = AbliterationPipeline(model_name="Qwen/Qwen3.8-27B")
pipeline.handle = SimpleNamespace(architecture=architecture)
with pytest.raises(PipelineValidationError, match="no validated projection allowlist"):
with pytest.raises(PipelineValidationError, match="remains blocked"):
pipeline._validate_architecture_surgery_support()
assert pipeline._excise_modified_count is None
+18
View File
@@ -557,6 +557,24 @@ class TestEdgeCases:
assert profile.arch_class == ArchitectureClass.DENSE
assert profile.reasoning_class == ReasoningClass.STANDARD
def test_qwen38_hybrid_does_not_recommend_generic_aggressive_surgery(self):
text_config = SimpleNamespace(
model_type="qwen3_5_text",
num_hidden_layers=64,
hidden_size=5120,
intermediate_size=17408,
vocab_size=248320,
)
config = SimpleNamespace(model_type="qwen3_5", text_config=text_config)
profile = detect_architecture("Qwen/Qwen3.8-27B", config=config)
assert profile.reasoning_class == ReasoningClass.REASONING
assert profile.profile_label == "Validated Qwen3.5 Hybrid"
assert profile.recommended_method == "advanced"
assert profile.method_overrides["n_directions"] == 4
assert profile.method_overrides["refinement_passes"] == 1
def test_unknown_model_type_in_config(self):
"""Unknown model_type should not cause MoE classification."""
class MockConfig:
+13
View File
@@ -512,6 +512,19 @@ def test_mistral3_load_uses_image_text_class_and_keeps_remote_code_disabled(
loader_boundary.model_class.from_pretrained.assert_not_called()
def test_qwen35_load_is_explicit_text_only_sdpa(loader_boundary, caplog):
caplog.set_level("INFO")
loader_boundary.config.model_type = "qwen3_5"
loader_boundary.config.architectures = ["Qwen3_5ForConditionalGeneration"]
handle = loader.load_model("Qwen/Qwen3.8-27B", skip_snapshot=True)
assert handle.architecture == "qwen3_5"
kwargs = loader_boundary.model_class.from_pretrained.call_args.kwargs
assert kwargs["attn_implementation"] == "sdpa"
assert "text-only derivative" in caplog.text
def test_model_handle_metadata_snapshot_restore_summary_and_cleanup(tmp_path):
model = _model()
nested = SimpleNamespace(
+227
View File
@@ -0,0 +1,227 @@
"""Contracts for the explicitly supported Qwen3.8-27B residual writers."""
from types import SimpleNamespace
import pytest
import torch
from torch import nn
from obliteratus.abliterate import AbliterationPipeline
from obliteratus.models.qwen35_contracts import (
Qwen35LayerTargets,
Qwen35ContractError,
validate_qwen38_27b_projection_contract,
)
class _LinearAttentionLayer(nn.Module):
def __init__(self):
super().__init__()
with torch.device("meta"):
self.linear_attn = nn.Module()
self.linear_attn.out_proj = nn.Linear(6144, 5120, bias=False)
self.linear_attn.in_proj_qkv = nn.Linear(5120, 10240, bias=False)
self.linear_attn.in_proj_z = nn.Linear(5120, 6144, bias=False)
self.mlp = nn.Module()
self.mlp.down_proj = nn.Linear(17408, 5120, bias=False)
self.mlp.gate_proj = nn.Linear(5120, 17408, bias=False)
self.mlp.up_proj = nn.Linear(5120, 17408, bias=False)
class _FullAttentionLayer(nn.Module):
def __init__(self):
super().__init__()
with torch.device("meta"):
self.self_attn = nn.Module()
self.self_attn.o_proj = nn.Linear(6144, 5120, bias=False)
self.self_attn.q_proj = nn.Linear(5120, 12288, bias=False)
self.self_attn.k_proj = nn.Linear(5120, 1024, bias=False)
self.self_attn.v_proj = nn.Linear(5120, 1024, bias=False)
self.mlp = nn.Module()
self.mlp.down_proj = nn.Linear(17408, 5120, bias=False)
self.mlp.gate_proj = nn.Linear(5120, 17408, bias=False)
self.mlp.up_proj = nn.Linear(5120, 17408, bias=False)
def _handle():
layer_types = [
layer_type
for _ in range(16)
for layer_type in (
"linear_attention",
"linear_attention",
"linear_attention",
"full_attention",
)
]
layers = nn.ModuleList(
_LinearAttentionLayer() if kind == "linear_attention" else _FullAttentionLayer()
for kind in layer_types
)
text_config = SimpleNamespace(
layer_types=layer_types,
hidden_size=5120,
intermediate_size=17408,
num_hidden_layers=64,
)
return SimpleNamespace(
architecture="qwen3_5",
model_name="Qwen/Qwen3.8-27B",
config=SimpleNamespace(text_config=text_config),
model=SimpleNamespace(model=SimpleNamespace(layers=layers)),
num_layers=64,
)
def test_qwen38_manifest_contains_only_residual_writers():
targets = validate_qwen38_27b_projection_contract(_handle())
assert len(targets) == 64
assert sum(target.mixer_attribute == "linear_attn" for target in targets) == 48
assert sum(target.mixer_attribute == "self_attn" for target in targets) == 16
assert targets[0].parameter_names == frozenset(
{"linear_attn.out_proj.weight", "mlp.down_proj.weight"}
)
assert targets[3].parameter_names == frozenset(
{"self_attn.o_proj.weight", "mlp.down_proj.weight"}
)
forbidden = {
"in_proj_qkv",
"in_proj_z",
"q_proj",
"k_proj",
"v_proj",
"gate_proj",
"up_proj",
"lm_head",
}
assert all(
not any(name in parameter for name in forbidden)
for target in targets
for parameter in target.parameter_names
)
def test_qwen38_manifest_fails_closed_on_topology_change():
handle = _handle()
handle.config.text_config.layer_types[0] = "full_attention"
with pytest.raises(Qwen35ContractError, match="48 DeltaNet"):
validate_qwen38_27b_projection_contract(handle)
def test_qwen38_manifest_fails_closed_on_writer_shape_change():
handle = _handle()
with torch.device("meta"):
handle.model.model.layers[0].linear_attn.out_proj = nn.Linear(5120, 5120, bias=False)
with pytest.raises(Qwen35ContractError, match="unexpected type or shape"):
validate_qwen38_27b_projection_contract(handle)
def test_qwen_output_only_projection_preserves_forbidden_tensors_bit_exactly():
torch.manual_seed(7)
layer = nn.Module()
layer.linear_attn = nn.Module()
layer.linear_attn.out_proj = nn.Linear(4, 4, bias=False)
layer.linear_attn.in_proj_qkv = nn.Linear(4, 8, bias=False)
layer.linear_attn.in_proj_z = nn.Linear(4, 4, bias=False)
layer.mlp = nn.Module()
layer.mlp.down_proj = nn.Linear(8, 4, bias=False)
layer.mlp.gate_proj = nn.Linear(4, 8, bias=False)
layer.mlp.up_proj = nn.Linear(4, 8, bias=False)
target = Qwen35LayerTargets("linear_attn", "out_proj")
forbidden_before = {
name: parameter.detach().clone()
for name, parameter in layer.named_parameters()
if name not in target.parameter_names
}
allowed_before = {
name: parameter.detach().clone()
for name, parameter in layer.named_parameters()
if name in target.parameter_names
}
direction = torch.tensor([[1.0], [0.0], [0.0], [0.0]])
saved_norms = AbliterationPipeline._capture_layer_weight_norms(
layer, parameter_names=target.parameter_names
)
AbliterationPipeline._project_out_advanced(
layer.linear_attn,
direction,
[target.mixer_output],
norm_preserve=False,
)
AbliterationPipeline._project_out_advanced(
layer.mlp,
direction,
[target.ffn_output],
norm_preserve=False,
)
AbliterationPipeline._restore_layer_weight_norms(layer, saved_norms)
current = dict(layer.named_parameters())
assert all(torch.equal(current[name], value) for name, value in forbidden_before.items())
assert all(not torch.equal(current[name], value) for name, value in allowed_before.items())
def test_qwen_excise_route_does_not_touch_inputs_gates_or_lm_head():
torch.manual_seed(11)
layer = nn.Module()
layer.linear_attn = nn.Module()
layer.linear_attn.out_proj = nn.Linear(4, 4, bias=False)
layer.linear_attn.in_proj_qkv = nn.Linear(4, 8, bias=False)
layer.linear_attn.in_proj_z = nn.Linear(4, 4, bias=False)
layer.mlp = nn.Module()
layer.mlp.down_proj = nn.Linear(8, 4, bias=False)
layer.mlp.gate_proj = nn.Linear(4, 8, bias=False)
layer.mlp.up_proj = nn.Linear(4, 8, bias=False)
model = nn.Module()
model.model = nn.Module()
model.model.layers = nn.ModuleList([layer])
model.lm_head = nn.Linear(4, 16, bias=False)
handle = SimpleNamespace(
architecture="qwen3_5",
config=SimpleNamespace(num_attention_heads=1),
model=model,
num_layers=1,
)
pipeline = AbliterationPipeline(
model_name="Qwen/Qwen3.8-27B",
method="advanced",
projection_target="all",
)
pipeline.handle = handle
pipeline._qwen35_projection_manifest = (
Qwen35LayerTargets("linear_attn", "out_proj"),
)
pipeline._strong_layers = [0]
pipeline.refusal_subspaces = {
0: torch.tensor([[1.0, 0.0, 0.0, 0.0], [0.0, 1.0, 0.0, 0.0]])
}
pipeline.refusal_directions = {0: pipeline.refusal_subspaces[0][0]}
pipeline._layer_excise_weights = {}
pipeline._on_log = lambda _message: None
before = {name: value.detach().clone() for name, value in model.named_parameters()}
pipeline._excise()
after = dict(model.named_parameters())
allowed = {
"model.layers.0.linear_attn.out_proj.weight",
"model.layers.0.mlp.down_proj.weight",
}
assert all(not torch.equal(after[name], before[name]) for name in allowed)
assert all(
torch.equal(after[name], before[name])
for name in before.keys() - allowed
)
@pytest.mark.parametrize("architecture", ["qwen3_5_text", "qwen3_5_moe"])
def test_other_qwen35_variants_remain_blocked(architecture):
handle = _handle()
handle.architecture = architecture
with pytest.raises(Qwen35ContractError, match="unsupported Qwen hybrid architecture"):
validate_qwen38_27b_projection_contract(handle)