expert_groups: add exp_legal expert group (MultiLegalPile)

connito/shared/config.py

@@ -203,6 +203,14 @@ class DatasetSourceCfg(BaseConfig):
     name: str | None = None
     weight: PositiveFloat = 1.0
     text_column: str = "text"
+    # Authorize HF's `load_dataset` to execute the dataset repo's custom
+    # builder script. Required for sources that ship a `<name>.py` loader
+    # (e.g. joelniklaus/Multi_Legal_Pile). Opt-in per source so a single
+    # malicious dataset can't piggyback through a globally-enabled flag —
+    # operators consent to executing remote code one dataset at a time.
+    # Pin the source's revision via `eval_source_revision_pin` when
+    # turning this on to bound the surface to a reviewed SHA.
+    trust_remote_code: bool = False
 
     @model_validator(mode="after")
     def _validate_non_empty(self):
@@ -481,10 +489,16 @@ class ExpertCfg(BaseConfig):
 
 
 class TaskCfg(BaseConfig):
+    # `expert_group_name` is locked so operators can't drift back to
+    # `exp_math` after the subnet-wide switch to the legal expert group —
+    # `auto_update_config` resets any non-default value on load and logs a
+    # one-time reset warning (see docs/exp-legal-migration-plan.md for the
+    # activation checklist). `helper_group_id` and `routing_mode` stay
+    # locked for the natural-routing (2Fnat) consensus contract.
     _LOCKED_FIELDS: ClassVar[frozenset[str]] = frozenset({
         "expert_group_name", "helper_group_id", "routing_mode",
     })
-    expert_group_name: str = "exp_math"
+    expert_group_name: str = "exp_legal"
     load_all_expert_groups: bool = False
     base_path: Path = Path("expert_groups")
     path: Path | None = None
@@ -569,8 +583,18 @@ def _refresh_paths(self) -> None:
         # ckpt paths — always start from the class default (relative)
         base_ckpt = root / Path(ckpt_cls.model_fields["base_checkpoint_path"].default)
         self.ckpt.base_checkpoint_path = base_ckpt
+        # Include expert_group_name in the leaf so switching the active expert
+        # group automatically writes/resumes from a fresh, group-isolated
+        # directory — no run_name bump needed, and no risk of resuming another
+        # group's (incompatible) checkpoints. Re-derived after locked-field
+        # reset via _update_by_task -> _refresh_paths, so the path always
+        # tracks the *effective* group.
         self.ckpt.checkpoint_path = (
-            base_ckpt / self.chain.coldkey_name / self.chain.hotkey_name / self.run.run_name
+            base_ckpt
+            / self.chain.coldkey_name
+            / self.chain.hotkey_name
+            / self.run.run_name
+            / self.task.expert_group_name
         )
         self.ckpt.validator_checkpoint_path = (
             base_ckpt / Path(ckpt_cls.model_fields["validator_checkpoint_path"].default)
@@ -661,7 +685,25 @@ def from_path(cls, path: str | Path, auto_update_config: bool = False) -> "Worke
             data = yaml.safe_load(f) or {}
         instance = cls(**data)
         instance._prompt_new_fields(yaml_data=data, config_path=path, auto_update=auto_update_config)
+        pre_lock_group = instance.task.expert_group_name
         instance.check_and_prompt_locked(config_path=path, auto_update=auto_update_config)
+        # Locked-field enforcement may have just reset task.expert_group_name
+        # (the exp_legal activation path: a YAML still saying exp_math gets
+        # reset to the locked default). task.path / task.exp were derived at
+        # construction from the PRE-reset name, so re-derive them — otherwise
+        # the process persists "exp_legal" to disk but keeps RUNNING exp_math
+        # (wrong group_id on chain commits) until a second restart. Observed
+        # live on the pioneer validator, 2026-07-11 11:49 UTC.
+        if instance.task.expert_group_name != pre_lock_group:
+            logger.info(
+                "Locked-field reset changed the active task — re-deriving task config",
+                old_task=pre_lock_group,
+                new_task=instance.task.expert_group_name,
+            )
+            # Pass the name explicitly: the no-arg form of _update_by_task
+            # reloads task.exp from the STALE task.path before refreshing
+            # paths, so the exp config would still be the old group's.
+            instance._update_by_task(expert_group_name=instance.task.expert_group_name)
         return instance
 
     def _prompt_new_fields(

connito/shared/dataloader.py

@@ -101,12 +101,23 @@ def get_tokenised_dataset(
     ):
         split_name = "train" if train else "validation"
 
-        def _load_streaming_split(ds_name: str, ds_config: str | None = None):
+        def _load_streaming_split(
+            ds_name: str,
+            ds_config: str | None = None,
+            trust_remote_code: bool = False,
+        ):
             """Helper to load a dataset split safely, falling back to 'train' if 'validation' is missing."""
             try:
-                load_kwargs = {"streaming": True, "revision": "main"}
+                load_kwargs: dict[str, Any] = {"streaming": True, "revision": "main"}
                 if ds_config is not None:
                     load_kwargs["name"] = ds_config
+                if trust_remote_code:
+                    # Authorize HF to execute the dataset repo's custom
+                    # builder script. Opt-in per source via
+                    # DatasetSourceCfg.trust_remote_code; never on by
+                    # default. See config.DatasetSourceCfg for the
+                    # rationale.
+                    load_kwargs["trust_remote_code"] = True
 
                 ds = load_dataset(ds_name, **load_kwargs)
                 if split_name in ds:
@@ -224,6 +235,7 @@ def ensure_string(example: dict[str, Any], source_text_column: str):
             ds_config = _source_value(source, "name")
             text_column = _source_value(source, "text_column", "text")
             weight = float(_source_value(source, "weight", 1.0))
+            trust_remote_code = bool(_source_value(source, "trust_remote_code", False))
 
             if not ds_name:
                 raise ValueError("Each dataset source must define a non-empty 'path'.")
@@ -248,7 +260,11 @@ def ensure_string(example: dict[str, Any], source_text_column: str):
                 )
                 source_split = load_streaming_shard(pick, split_name=split_name)
             else:
-                source_split = _load_streaming_split(ds_name, ds_config=ds_config)
+                source_split = _load_streaming_split(
+                    ds_name,
+                    ds_config=ds_config,
+                    trust_remote_code=trust_remote_code,
+                )
 
             source_split = source_split.select_columns([text_column])
             source_split = source_split.map(

connito/test/test_locked_task_rederive.py

@@ -0,0 +1,70 @@
+"""Activation-order regression: when the locked `task.expert_group_name`
+is auto-reset on load (the exp_legal activation path), the derived
+`task.path` / `task.exp` must be re-derived in the same load.
+
+Observed live on the pioneer validator (2026-07-11 11:49 UTC): a config
+YAML still saying `exp_math` was reset to the locked default `exp_legal`
+and persisted to disk — but the process kept RUNNING exp_math
+(`task_path=/app/expert_groups/exp_math`, chain commits `group_id: 0`)
+because `task.path`/`task.exp` had been derived at construction, before
+`check_and_prompt_locked` ran. Every fleet validator would have needed a
+second restart to actually switch groups.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import yaml
+
+from connito.shared.config import MinerConfig
+
+
+def _write_cfg(tmp_path: Path, expert_group_name: str) -> Path:
+    cfg = {
+        "task": {"expert_group_name": expert_group_name},
+        # Pre-filled wallet identifiers so from_path skips the chain lookup.
+        "chain": {"hotkey_ss58": "test-hk", "coldkey_ss58": "test-ck", "uid": 0},
+    }
+    p = tmp_path / "config.yaml"
+    p.write_text(yaml.safe_dump(cfg))
+    return p
+
+
+def test_locked_reset_rederives_task_path(tmp_path: Path) -> None:
+    # Run from the repo root so relative expert_groups/<name> resolves.
+    assert Path("expert_groups/exp_legal/config.yaml").exists(), (
+        f"run from repo root (cwd={os.getcwd()})"
+    )
+    cfg_path = _write_cfg(tmp_path, "exp_math")
+
+    config = MinerConfig.from_path(cfg_path, auto_update_config=True)
+
+    # The locked default flipped the name...
+    assert config.task.expert_group_name == "exp_legal"
+    # ...and the DERIVED task state must have followed in the same load:
+    assert config.task.path is not None and config.task.path.name == "exp_legal"
+    assert config.task.exp.group_id == 3, (
+        f"stale task.exp — still group_id={config.task.exp.group_id} "
+        "(exp_math=0): check_and_prompt_locked reset the name without "
+        "re-deriving task.path/task.exp"
+    )
+    # ...including the checkpoint path, which is group-scoped so the switch
+    # writes/resumes from a fresh dir instead of the prior group's checkpoints.
+    assert config.ckpt.checkpoint_path is not None
+    assert config.ckpt.checkpoint_path.name == "exp_legal", (
+        f"checkpoint_path leaf must track the effective group, got "
+        f"{config.ckpt.checkpoint_path}"
+    )
+    # And the persisted YAML matches what the process actually runs.
+    persisted = yaml.safe_load(cfg_path.read_text())
+    assert persisted["task"]["expert_group_name"] == "exp_legal"
+
+
+def test_no_reset_no_rederive_noise(tmp_path: Path) -> None:
+    # A config already on the locked default loads once, stays exp_legal.
+    cfg_path = _write_cfg(tmp_path, "exp_legal")
+    config = MinerConfig.from_path(cfg_path, auto_update_config=True)
+    assert config.task.expert_group_name == "exp_legal"
+    assert config.task.exp.group_id == 3