"""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"]