configuration_source_qwen3.py

"""Configuration for CLOSE latent reasoning on Qwen3-VL.

The current ``raw_evidence_question`` path keeps the complete layer-``ell_star``
visual memory as immutable evidence and recurrently updates the text-only
question state by cross-attending to it. Its selected interface averages all
four residual recurrent states and may use a training-only restoration target
from the normal multimodal question state. Historical slot-workspace fields are
retained so old checkpoints remain loadable, but they are mutually exclusive
with the raw-evidence path.

Every architectural hyperparameter of Method §3 lives here -- nothing is
hardcoded in ``modeling_close_qwen3_vl.py``.

Layer-index convention (used consistently across the repo)
----------------------------------------------------------
``ell_star`` is **0-indexed and inclusive**: it is the index of the last decoder
layer belonging to the lower branch. With ``num_hidden_layers == 28``::

    F_{<=l*} = language_model.layers[: ell_star + 1]     # lower_slice
    F_{>l*}  = language_model.layers[ell_star + 1 :]     # upper_slice

so ``ell_star=18`` means layers 0..18 run before the split and 19..27 after it.
``ell_star`` is measured once by ``scripts/localize_read_point.py`` (§3.1) and
frozen before any workspace training.

Careful with ``Qwen3VLConfig``: its ``__setattr__`` forwards any attribute
whose name already exists in ``text_config.__dict__`` down to the sub-config, and
its ``__init__`` builds ``text_config`` from ``**kwargs``. Workspace fields are
therefore declared as explicit named parameters so they never enter ``kwargs``
and never shadow a text-config field. ``tests/test_config_roundtrip.py`` guards
this.
"""

from __future__ import annotations

import math

from transformers.models.qwen3_vl.configuration_qwen3_vl import Qwen3VLConfig

#: How the ``S`` workspace slots are assigned M-RoPE positions once they are
#: spliced into the replacement cache above ``ell_star``. Qwen3-VL uses 3D
#: M-RoPE (t, h, w) with ``mrope_section=[16, 24, 24]``, so slots -- which have
#: no spatial extent -- need an explicit convention.
WORKSPACE_ROPE_MODES = (
    # Slots continue the 1-D text position sequence directly after Q*, with
    # t == h == w. Treats the workspace as "more text".
    "continue",
    # Slots reuse the (t, h, w) span the image tokens occupied in the original
    # multimodal prefill, subsampled to S positions. Preserves the read point's
    # positional geometry but reintroduces image-derived indices.
    "image_span",
    # All slots share one position (the first position after Q*). Makes the
    # workspace order-free, matching the set semantics of Perceiver-style slots.
    "shared",
)


class CloseQwen3VLConfig(Qwen3VLConfig):
    r"""Configuration for :class:`CloseQwen3VLForConditionalGeneration`.

    Args:
        ell_star: 0-indexed inclusive last layer of the lower branch. ``None``
            until §3.1 localization has run; the model refuses to build with
            ``None`` rather than falling back to a guessed depth.
        num_workspace_steps: the fixed recurrent horizon (paper notation ``T``;
            per-item segment counts ``n`` exist only in the losses, never here).
            The trainable parameter count is independent of it because
            ``f_theta`` is shared across steps, so it is a free experimental
            axis. ``0`` is legal and is the no-recurrence floor: ``Z^(0)`` goes
            straight to the decoder.
        num_workspace_slots: ``S``, slots in ``Z``. Free capacity knob.
        workspace_width: ``d_w``, the slot width. Free capacity knob -- the
            method's claim is path exclusivity (all visual information routed
            through ``Z``), NOT an information bottleneck, so ``S`` and ``d_w``
            carry no methodological commitment. Kept below ``d`` (4096 for the
            7B) since ``P_Z`` projects up at the splice.
        workspace_num_heads: attention heads inside ``r_theta`` / ``f_theta``.
        read_num_blocks: cross-attention blocks in ``r_theta``. The *read* is
            once regardless of depth: all blocks see the same ``V*`` in a single
            forward, and ``V*`` is discarded afterwards.
        transition_num_blocks: blocks in the shared recurrent ``f_theta``.
        workspace_ffn_mult: FFN expansion inside workspace blocks.
        workspace_dropout: dropout inside workspace blocks.
        workspace_rope_mode: one of :data:`WORKSPACE_ROPE_MODES`.
        adapter_rank: rank of the low-rank adapter on the post-``ell_star``
            backbone layers. ``0`` disables it (frozen backbone only).
        adapter_alpha: LoRA-style scaling; effective scale is
            ``adapter_alpha / adapter_rank``.
        adapter_dropout: dropout probability applied only to the LoRA branch.
        lambda_traj: legacy trajectory-loss weight. The current method sets it
            to zero and trains with final-answer CE only; it remains serialized
            solely so historical checkpoints can still be inspected.
    """

    model_type = "close_qwen3_vl"

    def __init__(
        self,
        ell_star: int | None = None,
        num_workspace_steps: int = 4,
        num_workspace_slots: int = 32,
        workspace_width: int = 512,
        workspace_num_heads: int = 8,
        read_num_blocks: int = 1,
        transition_num_blocks: int = 2,
        workspace_ffn_mult: int = 4,
        workspace_dropout: float = 0.0,
        workspace_rope_mode: str = "continue",
        adapter_rank: int = 16,
        adapter_alpha: int = 32,
        adapter_dropout: float = 0.0,
        adapter_exclude_layers: tuple = (),
        lambda_traj: float = 0.0,
        read_shortcut: bool = False,
        splice_scale_match: bool = False,
        read_gating: bool = False,
        gate_init_eps: float = 0.02,
        read_residual_pure: bool = False,
        residual_norm_cap: float = 0.0,
        pure_warm_start: bool = False,
        final_verify: bool = False,
        mid_read_step: int = 0,
        persistent_slot_id: bool = False,
        mid_read_alpha: float = 1.0,
        evidence_reasoning: bool = False,
        raw_evidence_question: bool = False,
        raw_attention_source_layer: int = -1,
        raw_transition_ffn: bool | None = None,
        raw_state_replacement: bool = False,
        raw_question_anchor: bool = False,
        raw_mean_aggregation: bool = False,
        counterfactual_residual_recurrence: bool = False,
        visual_counterfactual_recurrence: bool = False,
        visual_cumulative_recurrence: bool = False,
        visual_full_state_recurrence: bool = False,
        recurrence_only_adapter: bool = False,
        counterfactual_beta: float = 1.0,
        crr_concat_aggregation: bool = False,
        crr_policy_rank: int = 0,
        crr_policy_alpha: float = 32.0,
        counterfactual_reliance_loss: bool = False,
        transition_aware_reliance_loss: bool = False,
        counterfactual_step_retention_loss: bool = False,
        functional_state_loss: bool = False,
        per_example_answer_loss: bool = False,
        interface_loss: bool = False,
        read_anchor: bool = False,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.ell_star = ell_star
        self.num_workspace_steps = num_workspace_steps
        self.num_workspace_slots = num_workspace_slots
        self.workspace_width = workspace_width
        self.workspace_num_heads = workspace_num_heads
        self.read_num_blocks = read_num_blocks
        self.transition_num_blocks = transition_num_blocks
        self.workspace_ffn_mult = workspace_ffn_mult
        self.workspace_dropout = workspace_dropout
        self.workspace_rope_mode = workspace_rope_mode
        self.adapter_rank = adapter_rank
        self.adapter_alpha = adapter_alpha
        self.adapter_dropout = adapter_dropout
        # E run: dense full-FT 층은 LoRA 대상에서 제외 (이중 파라미터화 방지)
        self.adapter_exclude_layers = list(adapter_exclude_layers or ())
        self.lambda_traj = lambda_traj
        # v2 anti-collapse switches (defaults False so v1 checkpoints reload).
        # read_shortcut: r_theta adds an orthogonally-projected pooled-V* term to
        # every slot, so Z^(0) is image-dependent from step 0 instead of relying
        # on a random cross-attention finding signal.
        # splice_scale_match: P_Z(Z) rows are RMS-matched to the question rows at
        # the splice. Unmatched, slot states are ~O(1) against layer-20 hiddens
        # in the hundreds -- the decoder could ignore them for free, and did
        # (probe 20143: visual sensitivity 0.94 -> 1.0000 over training).
        self.read_shortcut = read_shortcut
        self.splice_scale_match = splice_scale_match
        # v6 candidate: gated re-reads DURING the recurrence. Every re-read
        # goes through r_theta into Z, so the mediation invariant (decoder sees
        # only [Q*; P_Z(Z)]) is untouched; what changes is that the workspace
        # may consult V* again mid-reasoning, scaled by a learned scalar gate.
        # The gate is free at EVERY step -- no hand-coded schedule. Degeneration
        # into answer-time retrieval is disciplined by the objective itself:
        # L_traj pins each intermediate state to its teacher segment, so a
        # wait-then-fetch trajectory misses its intermediate targets. The gate
        # bias initialises negative, i.e. training starts from the proven
        # no-reread regime and departs only where gradients demand it.
        # Default False: v4/v5 checkpoints reload bit-identically.
        self.read_gating = read_gating
        # Initial gate open-rate epsilon_gamma: b_gamma = logit(eps). Sigmoid
        # never reaches exactly 0 at finite logits, so this is near-identity
        # (not bitwise) initialisation w.r.t. the read-once path.
        self.gate_init_eps = gate_init_eps
        # v7: always-on PURE visual residual rereads. Every step adds
        # Z = U + W_o XAttn(LN(U) -> queries, V* -> keys/values): no self-attn,
        # no FFN, no output norm, no biases in the branch, W_o zero-init.
        # Motivation (battery, ep1 ckpt 20752): read_once hurt badly while
        # null_step (blank memory at one step) was harmless -- the gated
        # reader's value was extra recurrent DEPTH, not vision. This branch
        # removes every path that can change Z without V* content, so any
        # benefit it shows IS visual by construction.
        self.read_residual_pure = read_residual_pure
        # v7.1 (advisor): relative-norm trust region on the visual residual.
        #   s_k = min(1, rho * RMS(U) / (RMS(Delta) + eps));  Z = U + s_k * Delta
        # Unlike a fixed alpha, the optimizer cannot undo this by rescaling
        # W_o (s_k adapts); ||Delta_used|| <~ rho * ||U|| structurally, and
        # V* = 0 => Delta = 0 keeps the zero-memory identity. 0.0 disables.
        self.residual_norm_cap = residual_norm_cap
        # v7.1: initialise pure_attn K/V projections from the initial visual
        # reader's v_in (same [d_w, d] shape) and Q from its first read
        # block's in-proj slice, instead of random xavier -- the ep1 finding
        # was that Q/K/V stayed ~random while only W_o trained.
        self.pure_warm_start = pure_warm_start
        # MAIN METHOD (advisor-confirmed): Final-Verified Recurrent Workspace.
        # Initial read -> PURE latent recurrence (no mid-step visual anything)
        # -> exactly ONE pure visual verification at the end:
        #   Z_out = U^(T) + W_o XAttn(LN(U^(T)), V*),  alpha = 1, W_o = 0 init.
        # Uses the same bias-free verifier module as read_residual_pure; the
        # two flags differ only in WHERE the branch fires (every step vs once).
        self.final_verify = final_verify
        # MAIN (advisor 최종 확정): Mid-Read Recurrent Workspace. 고정 상수
        # k* = floor(T/2) = 4 (스텝 sweep 산물이 아니라 판독 전후 동수 전이의
        # midpoint)에서 pure visual read 1회:
        #   Z^(4) = U^(4) + W_o XAttn(LN(U^(4)), V*),  alpha=1, W_o=0 init.
        # 논리적 step 4 == python 루프 index 3 (0-based) -- off-by-one은
        # tests/test_mid_read.py가 상태 분기 위치로 고정한다. 0이면 비활성.
        self.mid_read_step = mid_read_step
        # advisor(2026-08-06): E_slot(=workspace_read.slots)을 매 recurrent
        # step 입력과 mid-read query에 일시 제공. state에는 누적하지 않는다.
        self.persistent_slot_id = persistent_slot_id
        # 고정 read 스케일 Z = U + alpha*W_o A (advisor scale-mismatch 검사).
        # 1.0 = 기존과 비트동일. eval의 gate_overrides alpha는 절대값으로 이를 대체.
        self.mid_read_alpha = mid_read_alpha
        # advisor 최종안: E(persistent evidence) / R(recurrent reasoning) 분리.
        # states = [E^(0), R^(1..T)], 결합 Z^(T)=E+R^(T)는 prefill에서.
        self.evidence_reasoning = evidence_reasoning
        # Current replacement candidate: no learned evidence/reasoning slots.
        # E is the complete raw V* sequence and R0 is the text-only Q* sequence.
        # A single shared CrossAttn transition produces R1..RT. Residual mode
        # decodes RT; the replacement candidate decodes Q*+HT. Cross-attention
        # uses the backbone's native GQA shape and
        # is initialized from the first upper layer by default. Checkpoints
        # created before the CrossAttn-only decision did not serialize
        # raw_transition_ffn; None therefore preserves their historical FFN
        # for faithful evaluation, while every new training launch passes False.
        self.raw_evidence_question = bool(raw_evidence_question)
        self.raw_attention_source_layer = int(raw_attention_source_layer)
        self.raw_transition_ffn = (
            True if raw_transition_ffn is None else bool(raw_transition_ffn)
        )
        # Experimental simplification of the same raw-E/Q path.  The residual
        # carrier is replaced, not augmented:
        #   H0=Q*, Hk=CrossAttn(H{k-1}, E), decoder_latent=Q*+HT.
        # False preserves all residual raw-E/Q checkpoints and active runs.
        self.raw_state_replacement = bool(raw_state_replacement)
        # Question-anchored replacement keeps the fixed text scaffold in every
        # recurrent query without restoring the old recurrent identity path:
        #   Z0=Q*, Zk=Q*+CrossAttn(Z{k-1}, E), decoder_latent=ZT.
        # This replaces the one-time Q*+HT decoder interface above.
        self.raw_question_anchor = bool(raw_question_anchor)
        # Q-shaped recurrent candidate selected by the V* interface oracle:
        #   R0=Q*, Rk=R{k-1}+CrossAttn(R{k-1}, V*)
        #   R_agg=mean(R1..RT)
        # This is parameter-free and replaces both question-anchored state
        # replacement and final-state-only decoding.
        self.raw_mean_aggregation = bool(raw_mean_aggregation)
        # Counterfactual Residual Recurrence (CRR): the image-conditioned and
        # text-only question states are propagated by the same native layer
        # F=L(ell_star+1).  Only their counterfactual residual is recurrently
        # refined; there are no slots, V* memory, custom attention modules, or
        # inference-time auxiliary modules.  beta is the fixed relaxed-update
        # coefficient; an optional batch-swap reliance loss is training-only.
        self.counterfactual_residual_recurrence = bool(
            counterfactual_residual_recurrence
        )
        # Qwen3 counterpart of the promoted Qwen2.5 full-state CRR path.  The
        # multimodal scaffold is used only inside the shared recurrent cell;
        # answer decoding still receives a question-shaped state and a
        # text-only prefix cache.
        self.visual_counterfactual_recurrence = bool(
            visual_counterfactual_recurrence
        )
        self.visual_cumulative_recurrence = bool(
            visual_cumulative_recurrence
        )
        self.visual_full_state_recurrence = bool(
            visual_full_state_recurrence
        )
        self.recurrence_only_adapter = bool(recurrence_only_adapter)
        self.counterfactual_beta = float(counterfactual_beta)
        # Bias-free Concat(C1..CT) -> C_agg, initialized to the exact
        # historical mean interface for checkpoint-compatible warm starts.
        self.crr_concat_aggregation = bool(crr_concat_aggregation)
        # Optional latent-only policy adapter.  It corrects the deterministic
        # C2..CT means but never touches B, C1, or answer-token decoding.  Rank 0
        # is the exact historical CRR path; RL checkpoints serialize rank > 0.
        self.crr_policy_rank = int(crr_policy_rank)
        self.crr_policy_alpha = float(crr_policy_alpha)
        # Training-only CRR intervention.  A batch-matched residual from a
        # different-answer example replaces the factual residual before the
        # upper decoder.  The model returns the factual-minus-swapped answer
        # score gap; the Trainer applies the margin loss.  Evaluation and
        # generation never execute the extra branch.
        self.counterfactual_reliance_loss = bool(
            counterfactual_reliance_loss
        )
        self.transition_aware_reliance_loss = bool(
            transition_aware_reliance_loss
        )
        # Training-only functional no-regression objective.  Gold-answer NLL
        # is measured through the actual upper decoder at every recurrent CRR
        # prefix; inference remains the unchanged T-step mean interface.
        self.counterfactual_step_retention_loss = bool(
            counterfactual_step_retention_loss
        )
        # Training-only self-distillation target.  The multimodal lower pass
        # already computed for V* also provides its image-conditioned question
        # rows.  They supervise R_agg at the split boundary, but are never
        # exposed to the inference decoder.
        self.functional_state_loss = bool(functional_state_loss)
        # Token CE lets long caption answers dominate one-token visual-choice
        # examples.  The per-example variant first averages valid answer tokens
        # within each sample, then averages samples.  It changes training/eval
        # loss reduction only and has no inference-time effect.
        self.per_example_answer_loss = bool(per_example_answer_loss)
        # ER v2 (advisor 2026-08-11): explicit evidence read —
        # R^(k) = f_theta(R^(k-1), Q*; E), E는 별도 memory로 cross-attn 읽기.
        # (E+R 합산-감산 제거. evidence_reasoning=True 전제.)
        self.explicit_evidence_read = bool(kwargs.pop("explicit_evidence_read", False))
        # Historical field name retained for checkpoint compatibility. In ER
        # v2.2 this means R^(0) is a batch-broadcast set of learnable slots; it
        # is deliberately NOT initialized from Q*. Q* conditions every shared
        # transition instead.
        self.r_init_question = bool(kwargs.pop("r_init_question", False))
        # ER v2.1 (advisor 2026-08-11): evidence read의 competitive normalization —
        # slot 축 softmax 후 evidence 축 재정규화 (Slot Attention식 경쟁).
        # 파라미터/모듈/loss 무추가; evidence_read의 W_qkv/W_o 재사용.
        self.competitive_evidence_read = bool(
            kwargs.pop("competitive_evidence_read", False))
        # ER v2.2 C: k>=1 traj 감독 부재 시 twin 체인 제거 (states = answer 체인).
        self.er_single_chain = bool(kwargs.pop("er_single_chain", False))
        # ER v2.2 C (§5-7): decoder latent = softmax(temporal_logits)로 가중한
        # R1..RT 결합 (전역 스칼라 T개). er_single_chain 전제. (legacy)
        self.temporal_aggregation = bool(kwargs.pop("temporal_aggregation", False))
        # ER v2.2 C 확정판: R_agg = Linear(Concat(R1..RT), bias=False),
        # 초기 W=[I/T,...,I/T] (R_agg=mean(R1..RT)). er_single_chain 전제.
        self.concat_aggregation = bool(kwargs.pop("concat_aggregation", False))
        # ER v2: 워크스페이스 블록의 rank-r 부분공간 attention/FFN (0 = dense).
        # d_w=d에서 state width 유지와 파라미터 폭증을 분리 (advisor).
        self.workspace_low_rank = int(kwargs.pop("workspace_low_rank", 0))
        # advisor: single-layer text-side interface alignment (bridge-only).
        # teacher = frozen-base mm 경로 q_end@l*+1 (adapter off, sg);
        # student = detach(R^T) splice의 l*+1 한 층 재계산. 추론 시 무존재.
        self.interface_loss = interface_loss
        # Training-only visual anchor switch. False in the current answer-only
        # method; when false the model does not even materialize pooled V* for
        # the Trainer.
        self.read_anchor = bool(read_anchor)
        if evidence_reasoning and persistent_slot_id:
            raise ValueError("evidence_reasoning은 persistent_slot_id와 동시 사용 불가")
        if evidence_reasoning and not mid_read_step:
            raise ValueError("evidence_reasoning은 mid_read_step > 0 필요")
        self.validate()

    # -- derived views -----------------------------------------------------

    @property
    def num_decoder_layers(self) -> int:
        return self.text_config.num_hidden_layers

    @property
    def backbone_width(self) -> int:
        """``d`` -- backbone hidden width (4096 for the 7B)."""
        return self.text_config.hidden_size

    @property
    def lower_slice(self) -> slice:
        """Layers forming ``F_{<=l*}``."""
        self._require_ell_star()
        return slice(0, self.ell_star + 1)

    @property
    def upper_slice(self) -> slice:
        """Layers forming ``F_{>l*}``."""
        self._require_ell_star()
        return slice(self.ell_star + 1, self.num_decoder_layers)

    # -- validation --------------------------------------------------------

    def _require_ell_star(self) -> None:
        if self.ell_star is None:
            raise ValueError(
                "ell_star is unset. Run scripts/localize_read_point.py (Method 3.1) "
                "and pass the measured layer explicitly; it must not be guessed."
            )

    def validate(self) -> None:
        """Reject configurations that silently break the §3.2 contract."""
        if self.ell_star is not None:
            n = self.num_decoder_layers
            # Both branches must be non-empty: an empty lower branch means there
            # is no V* to read, an empty upper branch means the workspace never
            # reaches the decoder.
            if not 0 <= self.ell_star <= n - 2:
                raise ValueError(
                    f"ell_star={self.ell_star} out of range for {n} decoder layers; "
                    f"expected 0 <= ell_star <= {n - 2} so both branches are non-empty."
                )
        # K=0 is the no-recurrence control (read only), not a misconfiguration.
        if self.num_workspace_steps < 0:
            raise ValueError("num_workspace_steps (K) must be >= 0.")
        if self.num_workspace_slots < 1:
            raise ValueError("num_workspace_slots (S) must be >= 1.")
        if self.workspace_num_heads < 1:
            raise ValueError("workspace_num_heads must be >= 1.")
        if self.workspace_width % self.workspace_num_heads != 0:
            raise ValueError(
                f"workspace_width={self.workspace_width} must be divisible by "
                f"workspace_num_heads={self.workspace_num_heads}."
            )
        if self.workspace_width > self.backbone_width:
            raise ValueError(
                f"workspace_width (d_w={self.workspace_width}) cannot exceed "
                f"backbone width (d={self.backbone_width}). Native-width d_w=d is "
                "the current method and uses an Identity decoder interface."
            )
        if self.workspace_rope_mode not in WORKSPACE_ROPE_MODES:
            raise ValueError(
                f"workspace_rope_mode={self.workspace_rope_mode!r} not in "
                f"{WORKSPACE_ROPE_MODES}."
            )
        if self.adapter_rank < 0:
            raise ValueError("adapter_rank must be >= 0 (0 disables the adapter).")
        if not math.isfinite(self.adapter_dropout) or not 0.0 <= self.adapter_dropout < 1.0:
            raise ValueError("adapter_dropout must be finite and in [0, 1).")
        if getattr(self, "crr_policy_rank", 0) < 0:
            raise ValueError("crr_policy_rank must be >= 0.")
        if not math.isfinite(getattr(self, "crr_policy_alpha", 0.0)) or getattr(
            self, "crr_policy_alpha", 0.0
        ) <= 0.0:
            raise ValueError("crr_policy_alpha must be finite and > 0.")
        if self.read_num_blocks < 1 or self.transition_num_blocks < 1:
            raise ValueError("read_num_blocks and transition_num_blocks must be >= 1.")
        if self.workspace_ffn_mult < 1:
            raise ValueError("workspace_ffn_mult must be >= 1.")
        mid = int(getattr(self, "mid_read_step", 0) or 0)
        if mid < 0 or mid > self.num_workspace_steps:
            raise ValueError(
                f"mid_read_step={mid} must be in [0, num_workspace_steps="
                f"{self.num_workspace_steps}]."
            )
        low_rank = int(getattr(self, "workspace_low_rank", 0) or 0)
        if low_rank < 0:
            raise ValueError("workspace_low_rank must be >= 0.")
        if low_rank and low_rank % self.workspace_num_heads != 0:
            raise ValueError(
                f"workspace_low_rank={low_rank} must be divisible by "
                f"workspace_num_heads={self.workspace_num_heads}."
            )
        if getattr(self, "competitive_evidence_read", False) and not getattr(
            self, "explicit_evidence_read", False
        ):
            raise ValueError(
                "competitive_evidence_read requires explicit_evidence_read=True; "
                "otherwise the flag has no execution path."
            )
        if getattr(self, "explicit_evidence_read", False) and not getattr(
            self, "evidence_reasoning", False
        ):
            raise ValueError(
                "explicit_evidence_read requires evidence_reasoning=True; otherwise "
                "the evidence-read module is allocated but never called."
            )
        if getattr(self, "r_init_question", False) and not getattr(
            self, "evidence_reasoning", False
        ):
            raise ValueError(
                "r_init_question (legacy name for learnable R0 slots) requires "
                "evidence_reasoning=True; otherwise reasoning_slots are unused."
            )
        if (getattr(self, "concat_aggregation", False)
                or getattr(self, "temporal_aggregation", False)):
            if self.num_workspace_steps < 1:
                raise ValueError("reasoning aggregation requires num_workspace_steps >= 1.")
            if not getattr(self, "evidence_reasoning", False):
                raise ValueError(
                    "reasoning aggregation requires evidence_reasoning=True; "
                    "otherwise the aggregation module is unused."
                )
        if getattr(self, "concat_aggregation", False) and not getattr(
            self, "er_single_chain", False
        ):
            raise ValueError("concat_aggregation requires er_single_chain=True.")
        if getattr(self, "raw_evidence_question", False):
            if self.workspace_width != self.backbone_width:
                raise ValueError(
                    "raw_evidence_question requires native workspace_width == "
                    "backbone_width: E=V* and R0=Q* are used without projections."
                )
            if self.num_workspace_steps < 1:
                raise ValueError(
                    "raw_evidence_question requires num_workspace_steps >= 1."
                )
            source_layer = (
                self.ell_star + 1
                if self.raw_attention_source_layer < 0
                else self.raw_attention_source_layer
            )
            if not self.ell_star < source_layer < self.text_config.num_hidden_layers:
                raise ValueError(
                    "raw_attention_source_layer must be an upper-backbone layer, "
                    f"got ell_star={self.ell_star}, source={source_layer}."
                )
            incompatible = {
                "evidence_reasoning": bool(getattr(self, "evidence_reasoning", False)),
                "explicit_evidence_read": bool(getattr(self, "explicit_evidence_read", False)),
                "competitive_evidence_read": bool(getattr(self, "competitive_evidence_read", False)),
                "r_init_question": bool(getattr(self, "r_init_question", False)),
                "er_single_chain": bool(getattr(self, "er_single_chain", False)),
                "temporal_aggregation": bool(getattr(self, "temporal_aggregation", False)),
                "concat_aggregation": bool(getattr(self, "concat_aggregation", False)),
                "read_shortcut": bool(getattr(self, "read_shortcut", False)),
                "splice_scale_match": bool(getattr(self, "splice_scale_match", False)),
                "read_gating": bool(getattr(self, "read_gating", False)),
                "read_residual_pure": bool(getattr(self, "read_residual_pure", False)),
                "final_verify": bool(getattr(self, "final_verify", False)),
                "mid_read_step": bool(getattr(self, "mid_read_step", 0)),
                "persistent_slot_id": bool(getattr(self, "persistent_slot_id", False)),
                "interface_loss": bool(getattr(self, "interface_loss", False)),
                "read_anchor": bool(getattr(self, "read_anchor", False)),
            }
            active = [name for name, enabled in incompatible.items() if enabled]
            if active:
                raise ValueError(
                    "raw_evidence_question replaces the slot workspace and is "
                    f"incompatible with: {', '.join(active)}"
                )
            if self.raw_state_replacement and self.raw_transition_ffn:
                raise ValueError(
                    "raw_state_replacement is CrossAttn-only and is incompatible "
                    "with raw_transition_ffn=True."
                )
            if self.raw_question_anchor and not self.raw_state_replacement:
                raise ValueError(
                    "raw_question_anchor requires raw_state_replacement=True."
                )
            if self.raw_mean_aggregation and (
                self.raw_state_replacement or self.raw_question_anchor
            ):
                raise ValueError(
                    "raw_mean_aggregation replaces raw_state_replacement and "
                    "raw_question_anchor; use the residual recurrence."
                )
            if self.functional_state_loss and not self.raw_mean_aggregation:
                raise ValueError(
                    "functional_state_loss requires raw_mean_aggregation=True "
                    "so its target is the exact decoder-facing recurrence."
                )
        elif getattr(self, "raw_state_replacement", False):
            raise ValueError(
                "raw_state_replacement requires raw_evidence_question=True."
            )
        elif getattr(self, "raw_question_anchor", False):
            raise ValueError(
                "raw_question_anchor requires raw_evidence_question=True."
            )
        elif getattr(self, "raw_mean_aggregation", False):
            raise ValueError(
                "raw_mean_aggregation requires raw_evidence_question=True."
            )
        elif getattr(self, "functional_state_loss", False):
            raise ValueError(
                "functional_state_loss requires raw_evidence_question=True."
            )
        if (
            getattr(self, "counterfactual_reliance_loss", False)
            and not getattr(self, "counterfactual_residual_recurrence", False)
        ):
            raise ValueError(
                "counterfactual_reliance_loss requires "
                "counterfactual_residual_recurrence=True."
            )
        if (
            getattr(self, "transition_aware_reliance_loss", False)
            and not getattr(self, "counterfactual_reliance_loss", False)
        ):
            raise ValueError(
                "transition_aware_reliance_loss requires "
                "counterfactual_reliance_loss=True."
            )
        if (
            getattr(self, "transition_aware_reliance_loss", False)
            and self.num_workspace_steps < 2
        ):
            raise ValueError(
                "transition_aware_reliance_loss requires at least two steps."
            )
        if (
            getattr(self, "counterfactual_step_retention_loss", False)
            and not getattr(self, "counterfactual_residual_recurrence", False)
        ):
            raise ValueError(
                "counterfactual_step_retention_loss requires "
                "counterfactual_residual_recurrence=True."
            )
        if (
            getattr(self, "counterfactual_step_retention_loss", False)
            and self.num_workspace_steps < 2
        ):
            raise ValueError(
                "counterfactual_step_retention_loss requires at least two steps."
            )
        if getattr(self, "crr_concat_aggregation", False) and not getattr(
            self, "counterfactual_residual_recurrence", False
        ):
            raise ValueError(
                "crr_concat_aggregation requires "
                "counterfactual_residual_recurrence=True."
            )
        if getattr(self, "counterfactual_residual_recurrence", False):
            if getattr(self, "raw_evidence_question", False):
                raise ValueError(
                    "counterfactual_residual_recurrence replaces "
                    "raw_evidence_question; enable exactly one q-state method."
                )
            if self.num_workspace_steps < 1:
                raise ValueError(
                    "counterfactual_residual_recurrence requires "
                    "num_workspace_steps >= 1."
                )
            if self.crr_policy_rank > 0 and self.num_workspace_steps < 2:
                raise ValueError(
                    "crr_policy_rank > 0 requires num_workspace_steps >= 2."
                )
            if not 0.0 <= self.counterfactual_beta <= 1.0:
                raise ValueError(
                    "counterfactual_beta must lie in [0, 1], got "
                    f"{self.counterfactual_beta}."
                )
            incompatible = {
                "evidence_reasoning": bool(getattr(self, "evidence_reasoning", False)),
                "explicit_evidence_read": bool(getattr(self, "explicit_evidence_read", False)),
                "competitive_evidence_read": bool(getattr(self, "competitive_evidence_read", False)),
                "r_init_question": bool(getattr(self, "r_init_question", False)),
                "er_single_chain": bool(getattr(self, "er_single_chain", False)),
                "temporal_aggregation": bool(getattr(self, "temporal_aggregation", False)),
                "concat_aggregation": bool(getattr(self, "concat_aggregation", False)),
                "read_shortcut": bool(getattr(self, "read_shortcut", False)),
                "splice_scale_match": bool(getattr(self, "splice_scale_match", False)),
                "read_gating": bool(getattr(self, "read_gating", False)),
                "read_residual_pure": bool(getattr(self, "read_residual_pure", False)),
                "final_verify": bool(getattr(self, "final_verify", False)),
                "mid_read_step": bool(getattr(self, "mid_read_step", 0)),
                "persistent_slot_id": bool(getattr(self, "persistent_slot_id", False)),
                "interface_loss": bool(getattr(self, "interface_loss", False)),
                "read_anchor": bool(getattr(self, "read_anchor", False)),
                "functional_state_loss": bool(getattr(self, "functional_state_loss", False)),
            }
            active = [name for name, enabled in incompatible.items() if enabled]
            if active:
                raise ValueError(
                    "counterfactual_residual_recurrence replaces the slot/raw "
                    f"workspace and is incompatible with: {', '.join(active)}"
                )
        elif getattr(self, "crr_policy_rank", 0) > 0:
            raise ValueError(
                "crr_policy_rank > 0 requires "
                "counterfactual_residual_recurrence=True."
            )
        if getattr(self, "visual_counterfactual_recurrence", False):
            if not getattr(self, "counterfactual_residual_recurrence", False):
                raise ValueError(
                    "visual_counterfactual_recurrence requires "
                    "counterfactual_residual_recurrence=True."
                )
            if not getattr(self, "visual_cumulative_recurrence", False):
                raise ValueError(
                    "Qwen3 visual recurrence currently requires the promoted "
                    "cumulative full-state path."
                )
            if not getattr(self, "visual_full_state_recurrence", False):
                raise ValueError(
                    "Qwen3 visual recurrence currently requires "
                    "visual_full_state_recurrence=True."
                )
            incompatible = {
                "counterfactual_reliance_loss": bool(
                    getattr(self, "counterfactual_reliance_loss", False)
                ),
                "transition_aware_reliance_loss": bool(
                    getattr(self, "transition_aware_reliance_loss", False)
                ),
                "counterfactual_step_retention_loss": bool(
                    getattr(self, "counterfactual_step_retention_loss", False)
                ),
                "crr_concat_aggregation": bool(
                    getattr(self, "crr_concat_aggregation", False)
                ),
                "crr_policy_rank": bool(getattr(self, "crr_policy_rank", 0)),
            }
            active = [name for name, enabled in incompatible.items() if enabled]
            if active:
                raise ValueError(
                    "Qwen3 full-state visual recurrence uses final-answer CE "
                    "and final-state decoding only; incompatible with: "
                    + ", ".join(active)
                )
        elif getattr(self, "visual_cumulative_recurrence", False) or getattr(
            self, "visual_full_state_recurrence", False
        ):
            raise ValueError(
                "visual cumulative/full-state flags require "
                "visual_counterfactual_recurrence=True."
            )
        if getattr(self, "recurrence_only_adapter", False):
            if not getattr(self, "visual_full_state_recurrence", False):
                raise ValueError(
                    "recurrence_only_adapter requires "
                    "visual_full_state_recurrence=True."
                )
            if self.adapter_rank <= 0:
                raise ValueError(
                    "recurrence_only_adapter requires adapter_rank > 0."
                )
            recurrent_layer = int(self.ell_star) + 1
            if recurrent_layer in set(self.adapter_exclude_layers):
                raise ValueError(
                    "recurrence_only_adapter cannot exclude its recurrent "
                    f"layer {recurrent_layer} from LoRA."
                )
        modes = [bool(getattr(self, m, False)) for m in
                 ("read_gating", "read_residual_pure", "final_verify")]
        modes.append(bool(getattr(self, "mid_read_step", 0)))
        if sum(modes) > 1:
            raise ValueError(
                "read_gating / read_residual_pure / final_verify are mutually "
                "exclusive reread modes."
            )


__all__ = ["CloseQwen3VLConfig", "WORKSPACE_ROPE_MODES"]