diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json index c0fdd1f..540a5ef 100644 --- a/ci/test-risk-map.json +++ b/ci/test-risk-map.json @@ -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", diff --git a/obliteratus/abliterate.py b/obliteratus/abliterate.py index 365047b..be5eae4 100644 --- a/obliteratus/abliterate.py +++ b/obliteratus/abliterate.py @@ -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, diff --git a/obliteratus/architecture_profiles.py b/obliteratus/architecture_profiles.py index 1cb260c..7ab4ca9 100644 --- a/obliteratus/architecture_profiles.py +++ b/obliteratus/architecture_profiles.py @@ -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 diff --git a/obliteratus/models/loader.py b/obliteratus/models/loader.py index 35afc9f..475026a 100644 --- a/obliteratus/models/loader.py +++ b/obliteratus/models/loader.py @@ -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": diff --git a/obliteratus/models/qwen35_contracts.py b/obliteratus/models/qwen35_contracts.py new file mode 100644 index 0000000..d5dbadf --- /dev/null +++ b/obliteratus/models/qwen35_contracts.py @@ -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) diff --git a/tests/conditional/test_model_download_runtime.py b/tests/conditional/test_model_download_runtime.py index 3015466..25077c0 100644 --- a/tests/conditional/test_model_download_runtime.py +++ b/tests/conditional/test_model_download_runtime.py @@ -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 diff --git a/tests/test_abliterate.py b/tests/test_abliterate.py index e0be4a8..13e5e87 100644 --- a/tests/test_abliterate.py +++ b/tests/test_abliterate.py @@ -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 diff --git a/tests/test_architecture_profiles.py b/tests/test_architecture_profiles.py index c8c5cf5..1aab9db 100644 --- a/tests/test_architecture_profiles.py +++ b/tests/test_architecture_profiles.py @@ -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: diff --git a/tests/test_loader_boundaries.py b/tests/test_loader_boundaries.py index 06b0e3f..9e8efe2 100644 --- a/tests/test_loader_boundaries.py +++ b/tests/test_loader_boundaries.py @@ -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( diff --git a/tests/test_qwen35_contracts.py b/tests/test_qwen35_contracts.py new file mode 100644 index 0000000..27e57e7 --- /dev/null +++ b/tests/test_qwen35_contracts.py @@ -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)