fix: core evaluator/cache/utils correctness fixes (#1378)

* fix: core evaluator/cache/utils correctness fixes

- cache.save_to_cache: accumulate serializable_obj per item (previously
  mutated subitems but never appended, so it always pickled an empty list)
- response_cache: recursively fingerprint nested task groups so subtasks
  get real fingerprints instead of falling back to empty strings
- evaluator:
  - redact api_key/token/hf_token-style values from results.config
  - clone padding Instances so [pad_source]*n no longer aliases one object
  - auto-init a gloo process group and inject lm._rank/_world_size for
    multi-rank launches whose backend never called init_process_group
  - fix torchrun all_gather_into_tensor (1-D input tensor, world_size buffer)
  - defensively skip docs with no requests in the postprocess loop
- utils:
  - repo-root-anchored git commit/branch helpers with env fallbacks
    (LMMS_EVAL_GIT_COMMIT/GITHUB_SHA/GITHUB_REF_NAME) and detached-HEAD handling
  - build_eval_output_dir helper
  - resolve relative dataset_kwargs.data_files to absolute paths for local
    csv/json/parquet builders
  - py3.13 re.split(maxsplit=) keyword fix

* fix(evaluator): gate gloo auto-init to the non-accelerate path; isolate padding clones per slot

Author: Luodian

lmms_eval/caching/cache.py

@@ -44,10 +44,13 @@ def save_to_cache(file_name, obj):
     serializable_obj = []
 
     for item in obj:
+        serializable_item = []
         for subitem in item:
             if hasattr(subitem, "arguments"):  # we need to handle the arguments specially since doc_to_visual is callable method and not serializable
                 serializable_arguments = tuple(arg if not callable(arg) else None for arg in subitem.arguments)
                 subitem.arguments = serializable_arguments
+            serializable_item.append(subitem)
+        serializable_obj.append(serializable_item)
 
     eval_logger.debug(f"Saving {file_path} to cache...")
     try:

lmms_eval/caching/response_cache.py

@@ -407,14 +407,21 @@ def create(
             cache_root = os.path.dirname(os.path.abspath(cache_root))
         cache_root = os.path.abspath(cache_root)
 
-        # Task fingerprints
+        # Task fingerprints — recursively process nested groups so subtasks
+        # get proper fingerprints instead of falling back to empty strings.
         task_fingerprints = {}
         if task_dict:
-            for tname, tobj in task_dict.items():
-                if hasattr(tobj, "dump_config"):
-                    cfg_str = json.dumps(tobj.dump_config(), sort_keys=True, default=str)
-                    cfg_str = _FUNC_ADDR_RE.sub(">", cfg_str)
-                    task_fingerprints[tname] = hash_string(cfg_str)[:16]
+
+            def _collect_fingerprints(d):
+                for tname, tobj in d.items():
+                    if hasattr(tobj, "dump_config"):
+                        cfg_str = json.dumps(tobj.dump_config(), sort_keys=True, default=str)
+                        cfg_str = _FUNC_ADDR_RE.sub(">", cfg_str)
+                        task_fingerprints[tname] = hash_string(cfg_str)[:16]
+                    elif isinstance(tobj, dict):
+                        _collect_fingerprints(tobj)
+
+            _collect_fingerprints(task_dict)
 
         # Model fingerprint
         if isinstance(model_args, dict):

lmms_eval/evaluator.py

@@ -73,6 +73,43 @@
 )
 
 IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tif", ".tiff")
+_SENSITIVE_CONFIG_KEYS = {
+    "api_key",
+    "client_secret",
+    "hf_token",
+    "huggingface_hub_token",
+    "huggingfacehub_api_token",
+    "token",
+}
+# Suffixes also catch prefixed variants: openai_api_key, github_token, my_secret, db_password.
+_SENSITIVE_KEY_SUFFIXES = ("api_key", "_token", "_secret", "password")
+_SECRET_ASSIGNMENT_RE = re.compile(r"(?i)(^|[,\s])(\w*(?:api_key|token|secret|password))=([^,\s]+)")
+_HF_TOKEN_VALUE_RE = re.compile(r"\bhf_[A-Za-z0-9]{20,}\b")
+
+
+def _is_sensitive_config_key(key) -> bool:
+    key_lower = str(key).lower()
+    return key_lower in _SENSITIVE_CONFIG_KEYS or key_lower.endswith(_SENSITIVE_KEY_SUFFIXES)
+
+
+def _redact_eval_config_secrets(value):
+    if isinstance(value, dict):
+        return {key: ("[REDACTED]" if _is_sensitive_config_key(key) else _redact_eval_config_secrets(item)) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return type(value)(_redact_eval_config_secrets(item) for item in value)
+    if isinstance(value, str):
+        value = _SECRET_ASSIGNMENT_RE.sub(r"\1\2=[REDACTED]", value)
+        return _HF_TOKEN_VALUE_RE.sub("[REDACTED]", value)
+    return value
+
+
+def _clone_padding_request(pad_source: Instance) -> Instance:
+    pad_instance = copy.copy(pad_source)
+    pad_instance.metadata = dict(pad_source.metadata or {})
+    pad_instance.metadata["__padding_only__"] = True
+    pad_instance.resps = []
+    pad_instance.token_counts = []
+    return pad_instance
 
 
 def _enable_reentrant_filelocks() -> None:
@@ -531,6 +568,7 @@ def _adjust_config(task_dict):
         # add info about execution
         results["config"].update(
             {
+                "model_backend": f"{type(lm).__module__}.{type(lm).__name__} ({task_type})",
                 "batch_size": batch_size,
                 "batch_sizes": (list(lm.batch_sizes.values()) if hasattr(lm, "batch_sizes") else []),
                 "device": device,
@@ -556,6 +594,8 @@ def _adjust_config(task_dict):
                     resolved[key] = str(value)
             results["config"]["resolved_cli_args"] = resolved
 
+        results["config"] = _redact_eval_config_secrets(results["config"])
+
         results["git_hash"] = get_git_commit_hash()
         results["git_branch"] = get_git_branch_name()
         results["lmms_eval_version"] = get_lmms_eval_version_string()
@@ -862,6 +902,7 @@ def evaluate(
     local_rank = int(os.environ.get("LOCAL_RANK", 0))
     global_rank = int(os.environ.get("RANK", 0))
     world_size = int(os.environ.get("WORLD_SIZE", 1))
+
     eval_logger.info(f"Running on rank {global_rank} (local rank {local_rank})")
 
     def _infer_task_request_type(task_obj: Task) -> Optional[str]:
@@ -882,8 +923,32 @@ def _infer_task_request_type(task_obj: Task) -> Optional[str]:
         if not all("bypass" not in getattr(task_output.task, "_metric_fn_list", {}).keys() for task_output in eval_tasks):
             raise ValueError("log_samples must be True for 'bypass' metric-only tasks")
 
-    if distributed_executor_backend == "accelerate" and not hasattr(lm, "accelerator"):
-        lm.accelerator = Accelerator()
+    if distributed_executor_backend == "accelerate":
+        if not hasattr(lm, "accelerator"):
+            lm.accelerator = Accelerator()
+    else:
+        # Torchrun/native path: auto-init a process group for multi-rank runs
+        # whose model backend never called init_process_group (e.g. diffusers
+        # video-gen models), so evaluator collectives don't crash. gloo needs no
+        # GPU. Gated out of the accelerate path on purpose: pre-initializing here
+        # would make Accelerator() adopt this gloo group instead of building NCCL,
+        # silently running GPU collectives on CPU.
+        if world_size > 1 and dist.is_available() and not dist.is_initialized():
+            if os.environ.get("MASTER_ADDR") and os.environ.get("MASTER_PORT"):
+                dist.init_process_group(backend="gloo")
+                eval_logger.info(f"evaluator: auto-initialized gloo process group (world={world_size})")
+            else:
+                eval_logger.warning(
+                    f"evaluator: WORLD_SIZE={world_size} but MASTER_ADDR/MASTER_PORT are unset; skipping torch.distributed auto-init. Distributed collectives will fail unless the model backend initializes the process group itself."
+                )
+
+    # Inject rank/world_size into the model so that logging (tqdm disable)
+    # and rank-conditional logic work on non-zero ranks.  Many simple models
+    # only read LOCAL_RANK for device binding and leave _rank/_world_size
+    # at their defaults (0, 1).
+    if world_size > 1:
+        lm._rank = global_rank
+        lm._world_size = world_size
 
     for task_output in eval_tasks:
         task = task_output.task
@@ -950,8 +1015,8 @@ def _infer_task_request_type(task_obj: Task) -> Optional[str]:
                 instances_rnk = torch.tensor(len(task._instances), device=lm.device)
                 gathered_item = lm.accelerator.gather(instances_rnk).cpu().detach().numpy().tolist()
             elif distributed_executor_backend == "torchrun":
-                instances_rnk = torch.tensor(len(task._instances), device=lm.device)
-                gathered_item = torch.zeros(world_size * 1, dtype=instances_rnk.dtype, device=lm.device)
+                instances_rnk = torch.tensor([len(task._instances)], device=lm.device)
+                gathered_item = torch.zeros(world_size, dtype=instances_rnk.dtype, device=lm.device)
                 dist.all_gather_into_tensor(gathered_item, instances_rnk)
                 gathered_item = gathered_item.cpu().detach().numpy().tolist()
             else:
@@ -1003,7 +1068,8 @@ def _infer_task_request_type(task_obj: Task) -> Optional[str]:
                 eval_logger.warning(f"Running {reqtype} requests but could not find a pad source request on rank {global_rank}; skipping rank padding.")
             else:
                 for _ in range(padding_requests[reqtype]):
-                    cloned_reqs.extend([pad_source] * pad_source.repeats)
+                    pad_instance = _clone_padding_request(pad_source)
+                    cloned_reqs.extend([pad_instance] * pad_instance.repeats)
 
         # run requests through model (with optional response cache)
         if reqtype == "generate_until_agentic":
@@ -1124,6 +1190,12 @@ def _infer_task_request_type(task_obj: Task) -> Optional[str]:
             pbar = tqdm(total=total_docs, desc="Postprocessing", disable=(RANK != 0))
             for doc_id, doc in doc_iterator:
                 requests = instances_by_doc_id[doc_id]
+                # Defensive skip: if a doc landed in this rank's shard but has
+                # no requests (e.g. a split mismatch during a partial rerun),
+                # skip rather than crashing at `requests[0].doc` below.
+                if not requests:
+                    pbar.update(1)
+                    continue
 
                 # Strip reasoning tags before scoring
                 if reasoning_tags is not None:

lmms_eval/utils.py

@@ -76,7 +76,7 @@ def escaped_split(text, sep_char, maxsplit=-1):
         return text
     maxsplit = max(0, maxsplit)
 
-    return re.split(r"(?<!\\)" + sep_char, text, maxsplit)
+    return re.split(r"(?<!\\)" + sep_char, text, maxsplit=maxsplit)
 
 
 def handle_arg_string(arg):
@@ -775,27 +775,65 @@ def run_task_tests(task_list: List[str]):
         raise ValueError(f"Not all tests for the specified tasks ({task_list}) ran successfully! Error code: {pytest_return_val}")
 
 
+def _lmms_eval_repo_root() -> pathlib.Path:
+    return pathlib.Path(__file__).resolve().parents[1]
+
+
+def _git_output(*args: str) -> str:
+    return (
+        subprocess.check_output(
+            ["git", "-C", str(_lmms_eval_repo_root()), *args],
+            stderr=subprocess.DEVNULL,
+        )
+        .strip()
+        .decode()
+    )
+
+
+def _first_env(*names: str) -> str | None:
+    for name in names:
+        value = os.environ.get(name, "").strip()
+        if value:
+            return value
+    return None
+
+
+def _normalize_branch_name(branch: str) -> str:
+    return branch.removeprefix("remotes/origin/")
+
+
+def _is_exact_git_ref_name(branch: str) -> bool:
+    return bool(branch and branch != "undefined" and "^" not in branch and "~" not in branch)
+
+
 def get_git_commit_hash():
-    """
-    Gets the git commit hash of your current repo (if it exists).
-    Source: https://github.com/EleutherAI/gpt-neox/blob/b608043be541602170bfcfb8ec9bf85e8a0799e0/megatron/neox_arguments/neox_args.py#L42
-    """
     try:
-        git_hash = subprocess.check_output(["git", "describe", "--always"]).strip()
-        git_hash = git_hash.decode()
+        git_hash = _git_output("describe", "--always")
     except (subprocess.CalledProcessError, FileNotFoundError):
-        # FileNotFoundError occurs when git not installed on system
-        git_hash = None
+        git_hash = _first_env("LMMS_EVAL_GIT_COMMIT", "GIT_COMMIT", "GITHUB_SHA")
     return git_hash
 
 
 def get_git_branch_name():
-    """Gets the current git branch name (if in a repo)."""
     try:
-        branch = subprocess.check_output(["git", "rev-parse", "--abbrev-ref", "HEAD"]).strip()
-        return branch.decode()
+        branch = _git_output("branch", "--show-current")
+        if branch:
+            return _normalize_branch_name(branch)
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        pass
+
+    branch = _first_env("LMMS_EVAL_GIT_BRANCH", "GIT_BRANCH", "BRANCH_NAME", "GITHUB_REF_NAME")
+    if branch:
+        return _normalize_branch_name(branch)
+
+    try:
+        branch = _git_output("name-rev", "--name-only", "--exclude=tags/*", "HEAD")
+        if _is_exact_git_ref_name(branch):
+            return _normalize_branch_name(branch)
     except (subprocess.CalledProcessError, FileNotFoundError):
-        return None
+        pass
+
+    return "detached" if get_git_commit_hash() else None
 
 
 def get_lmms_eval_version_string():
@@ -919,6 +957,58 @@ def import_function(loader, node):
         raise ImportError(f"Failed to import function '{function_name}' from module '{module_name}'. " f"Tried relative path '{module_path}' and absolute import.") from ex
 
 
+_LOCAL_BUILDERS = frozenset({"csv", "json", "parquet", "text", "pandas", "arrow"})
+
+
+def _looks_like_url(value: str) -> bool:
+    return value.startswith(("http://", "https://", "s3://", "gs://", "ftp://", "ftps://"))
+
+
+def _resolve_local_data_files(value, yaml_dir: str):
+    """Rewrite relative ``data_files`` paths to absolute paths anchored at ``yaml_dir``.
+
+    Conservative: only rewrites a string when it's a relative path AND the
+    resolved absolute path exists on disk. Leaves absolute paths, URLs, and
+    non-existent relatives untouched. Recurses into dict/list to handle the
+    ``data_files: {train: <path>, validation: <path>}`` and ``data_files:
+    [<path>, <path>]`` shapes that ``datasets.load_dataset`` accepts.
+    """
+    if isinstance(value, str):
+        if not value or _looks_like_url(value) or os.path.isabs(value):
+            return value
+        candidate = os.path.join(yaml_dir, value)
+        if os.path.isfile(candidate):
+            return candidate
+        return value
+    if isinstance(value, dict):
+        return {k: _resolve_local_data_files(v, yaml_dir) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_resolve_local_data_files(v, yaml_dir) for v in value]
+    return value
+
+
+def _maybe_resolve_yaml_local_data_files(yaml_config: dict, yaml_dir: str) -> dict:
+    """Resolve ``dataset_kwargs.data_files`` relative paths against ``yaml_dir``.
+
+    Only applied for the generic local builders (``csv``, ``json``,
+    ``parquet``, ``text``, ``pandas``, ``arrow``). For HF hub dataset
+    paths (e.g. ``lmms-lab-eval/ssv2``) ``data_files`` is left as-is to
+    preserve the upstream behavior of letting HF resolve relative paths
+    against the dataset's hub root.
+    """
+    dataset_path = yaml_config.get("dataset_path")
+    if not isinstance(dataset_path, str) or dataset_path not in _LOCAL_BUILDERS:
+        return yaml_config
+    dataset_kwargs = yaml_config.get("dataset_kwargs")
+    if not isinstance(dataset_kwargs, dict) or "data_files" not in dataset_kwargs:
+        return yaml_config
+    resolved = _resolve_local_data_files(dataset_kwargs["data_files"], yaml_dir)
+    if resolved is dataset_kwargs["data_files"]:
+        return yaml_config
+    new_kwargs = {**dataset_kwargs, "data_files": resolved}
+    return {**yaml_config, "dataset_kwargs": new_kwargs}
+
+
 def load_yaml_config(yaml_path=None, yaml_config=None, yaml_dir=None, mode="full"):
     if mode == "simple":
         constructor_fn = ignore_constructor
@@ -962,8 +1052,8 @@ def load_yaml_config(yaml_path=None, yaml_config=None, yaml_dir=None, mode="full
                 raise ex
 
         final_yaml_config.update(yaml_config)
-        return final_yaml_config
-    return yaml_config
+        return _maybe_resolve_yaml_local_data_files(final_yaml_config, yaml_dir)
+    return _maybe_resolve_yaml_local_data_files(yaml_config, yaml_dir)
 
 
 def regex_replace(string, pattern, repl, count: int = 0):