| """Supported Safin-1 architecture; inactive historical metadata remains loadable.""" |
|
|
| RETIRED_ARCHITECTURE_FLAGS = ( |
| "hsd_text_only_mlp", "hsd_text_only_in_proj", "hsd_anchor_isolated_in_proj", |
| "hsd_text_single_pass_delta", "hsd_anchor_mlp", "hsd_full_attention_anchor_tokens", |
| ) |
|
|
|
|
| def validate_safin1_runtime(config): |
| get = config.get if isinstance(config, dict) else lambda name, default: getattr(config, name, default) |
| enabled = [name for name in RETIRED_ARCHITECTURE_FLAGS if get(name, False)] |
| if enabled: |
| raise ValueError(f"Retired Safin-1 architecture options cannot be enabled: {enabled}") |
| for name, expected in (("hsd_local_state_mode", "full"), ("hsd_history_read_q_mode", "shared_local")): |
| value = get(name, expected) |
| if value != expected: |
| raise ValueError(f"Safin-1 requires {name}={expected!r}; retired variant {value!r} is unsupported") |
|
|
|
|
| def validate_safin1_training_runtime(config): |
| """Reject training settings that have no implementation before model setup.""" |
| validate_safin1_runtime(config) |
| get = config.get if isinstance(config, dict) else lambda name, default: getattr(config, name, default) |
| for name, expected in (("hsd_history_mix_mode", "additive"), |
| ("hsd_route_weight_momentum_gamma", 0.0)): |
| value = get(name, expected) |
| if value != expected: |
| raise ValueError( |
| f"Safin-1 training requires {name}={expected!r}; " |
| f"{value!r} is not implemented and would not change the computation." |
| ) |
|
|
|
|
| FORBIDDEN_TRAINING_ENV = ( |
| "HSD_DEBUG_DETACH_TEXT_O", "HSD_DEBUG_ZERO_TEXT_O", |
| *(name.upper() for name in RETIRED_ARCHITECTURE_FLAGS), |
| ) |
|
|
|
|
| def validate_training_environment(environ=None, *, config=None, require_march=False): |
| """Reject inherited experimental switches before loading or allocating a model.""" |
| if environ is None: |
| import os |
| environ = os.environ |
| if require_march or is_march_training_config(config) or str(environ.get("USE_HIST_LINEAR_ATTN", "")).strip().lower() in _TRUE: |
| for name in ("USE_HIST_LINEAR_ATTN", "USE_MCORE_GDN", "SWIFT_USE_MCORE_GDN"): |
| value = str(environ.get(name, "")).strip().lower() |
| if value and value not in _TRUE: |
| raise ValueError(f"Safin-1/MARCH training requires {name}=1; conflicting value {value!r}") |
| if config is not None: |
| validate_runtime_environment(config, environ) |
| enabled = [name for name in FORBIDDEN_TRAINING_ENV |
| if str(environ.get(name, "")).strip().lower() in {"1", "true", "yes", "on"}] |
| if enabled: |
| raise ValueError(f"Unsupported training environment options: {enabled}; unset them before training") |
| for name, expected in (("HSD_LOCAL_STATE_MODE", "full"), |
| ("HSD_HISTORY_READ_Q_MODE", "shared_local")): |
| value = str(environ.get(name, "")).strip().lower() |
| if value and value != expected: |
| raise ValueError(f"Unsupported training environment {name}={value!r}; only {expected!r} is supported") |
|
|
|
|
| _TRUE = {"1", "true", "yes", "on"} |
| _FALSE = {"0", "false", "no", "off"} |
|
|
|
|
| def is_march_training_config(config): |
| """Recognize the architecture independently of process-local switches.""" |
| if config is None: |
| return False |
| get = config.get if isinstance(config, dict) else lambda name, default=None: getattr(config, name, default) |
| for name in ("model_type", "hf_model_type", "llm_model_type"): |
| model_type = str(get(name, "") or "") |
| if model_type.startswith(("safin1_", "historical_qwen3_5")): |
| return True |
| if int(get("anchor_chunk_size", 0) or 0) > 0 and int(get("anchor_token_num", 0) or 0) > 0: |
| return True |
| nested = get("hf_config") |
| return nested is not None and nested is not config and is_march_training_config(nested) |
|
|
|
|
| def validate_runtime_environment(config, environ): |
| """Legacy runtime environment values may confirm, but never override, config. |
| |
| Performance-only kernel tuning variables have no serialized config field |
| and remain available. ModelConfig supplies all serialized runtime defaults. |
| """ |
| import json |
| import math |
|
|
| values = config if isinstance(config, dict) else vars(config) |
| for name, raw in environ.items(): |
| field = "hsd_read_block_size" if name == "HIST_READ_BLOCK_SIZE" else name.lower() |
| if (not field.startswith("hsd_") or field == "hsd_runtime_config" |
| or field not in values or not str(raw).strip()): |
| continue |
| expected = values[field] |
| raw = str(raw).strip() |
| try: |
| if isinstance(expected, bool): |
| if raw.lower() not in _TRUE | _FALSE: |
| raise ValueError("expected a boolean") |
| actual = raw.lower() in _TRUE |
| elif isinstance(expected, int): |
| actual = int(raw) |
| elif isinstance(expected, float): |
| actual = float(raw) |
| if not math.isfinite(actual): |
| raise ValueError("expected a finite number") |
| else: |
| try: |
| actual = json.loads(raw) |
| except ValueError: |
| actual = None if raw.lower() in {"none", "null"} else raw |
| if isinstance(expected, str): |
| actual = str(actual).strip().lower() |
| expected = expected.strip().lower() |
| if isinstance(expected, tuple): |
| expected = list(expected) |
| except (TypeError, ValueError) as exc: |
| raise ValueError(f"Invalid training environment {name}={raw!r}: {exc}") from exc |
| if actual != expected: |
| raise ValueError( |
| f"Training environment {name}={raw!r} conflicts with serialized {field}={expected!r}. " |
| "Unset the environment variable and use --hsd-runtime-config or an explicit CLI field; " |
| "training and saved configuration must agree." |
| ) |
|
|
|
|
| def march_training_enabled(config, environ=None): |
| if environ is None: |
| import os |
| environ = os.environ |
| enabled = is_march_training_config(config) or str(environ.get("USE_HIST_LINEAR_ATTN", "")).strip().lower() in _TRUE |
| validate_training_environment(environ, config=config, require_march=enabled) |
| return enabled |
|
|
