solar_eval/core/task_discovery.py

"""데이터셋 파일에서 태스크를 발견한다 — 매니페스트에 손으로 적지 않는다.

왜: 태스크는 사실상 "데이터셋 파일 + 채점 설정"인데, 파일이 있다는 사실을
`project.yaml` 에 한 줄씩 옮겨 적어 왔다(chosun-proofreading 은 59줄). 손으로
옮겨 적는 목록은 반드시 원본과 갈라진다 — 실제로 데이터셋이 사라진 뒤에도
매니페스트에만 남은 유령 태스크 두 개(`critical-issue`·`edge-case`)가 생겼고,
`solar-eval tasks list` 에는 보이지만 돌리면 죽는 상태로 방치됐다.

발견 규칙은 파일 이름에서 태스크 이름을 만든다:

    discover:
      - glob: critical_issue/*_v2.jsonl   # datasets/ 기준 상대 glob
        name: ci-v2-{stem}                # {stem} = 파일 stem, '_' → '-'
        strip_suffix: _v2                 # stem 에서 먼저 떼는 꼬리표

**명시 태스크가 이긴다.** 이미 `tasks:` 에 있는 dataset_path 는 발견 대상에서
빠지므로, 규칙에 안 맞는 이름(`paragraph`, `yul-ryul`)은 예전 그대로 적어 두면
된다. 과거 run 이 그 이름으로 기록돼 있어 이름을 바꾸면 evalhub 에서 계보가
끊어지기 때문에, 이 우선순위는 타협 대상이 아니다.
"""

import logging
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)

#: `name` 템플릿에서 파일 stem 으로 치환되는 자리표시자.
STEM_PLACEHOLDER = "{stem}"


class TaskDiscoveryError(ValueError):
    """발견 규칙이 잘못됐을 때 — 조용히 빈 목록을 만들지 않는다."""


def discover_tasks(config: dict[str, Any], data_root: Path | str | None) -> list[dict[str, Any]]:
    """`discover:` 규칙으로 태스크 목록을 만든다.

    Args:
        config: project.yaml dict. `discover`, `dataset.repo`, `tasks` 를 읽는다.
        data_root: 데이터 루트(projects_dir). `dataset.repo` 가 이 아래에 있다.

    Returns:
        새로 발견된 태스크 dict 목록 (명시 태스크와 중복되지 않는 것만).
        `discover` 가 없으면 빈 목록.

    Raises:
        TaskDiscoveryError: 규칙에 `glob`/`name` 이 없거나 이름이 충돌할 때.
    """
    rules = config.get("discover")
    if not rules:
        return []
    if not isinstance(rules, list):
        raise TaskDiscoveryError("`discover` must be a list of rules")

    dataset_root = _dataset_root(config, data_root)
    if dataset_root is None or not dataset_root.is_dir():
        logger.warning(
            "Task discovery skipped for %s: dataset root not found (%s)",
            config.get("name"),
            dataset_root,
        )
        return []

    claimed_paths = {
        t.get("dataset_path") for t in config.get("tasks", []) or [] if isinstance(t, dict)
    }
    claimed_names = {t.get("name") for t in config.get("tasks", []) or [] if isinstance(t, dict)}

    discovered: list[dict[str, Any]] = []
    seen_names: dict[str, str] = {}
    for rule in rules:
        discovered.extend(_apply_rule(rule, dataset_root, claimed_paths, claimed_names, seen_names))
    return sorted(discovered, key=lambda t: t["name"])


def _apply_rule(
    rule: Any,
    dataset_root: Path,
    claimed_paths: set[Any],
    claimed_names: set[Any],
    seen_names: dict[str, str],
) -> list[dict[str, Any]]:
    """규칙 하나를 적용해 태스크를 만든다."""
    if not isinstance(rule, dict):
        raise TaskDiscoveryError(f"discover rule must be a mapping, got {type(rule).__name__}")
    pattern = rule.get("glob")
    name_template = rule.get("name")
    if not pattern or not name_template:
        raise TaskDiscoveryError(f"discover rule needs `glob` and `name`: {rule!r}")
    if STEM_PLACEHOLDER not in name_template:
        raise TaskDiscoveryError(
            f"discover `name` must contain {STEM_PLACEHOLDER}: {name_template!r}"
        )

    strip_suffix = rule.get("strip_suffix") or ""
    extra = {k: v for k, v in rule.items() if k not in ("glob", "name", "strip_suffix")}

    tasks: list[dict[str, Any]] = []
    matches = sorted(dataset_root.glob(pattern))
    if not matches:
        logger.warning("discover rule matched no files: %s (under %s)", pattern, dataset_root)

    for file in matches:
        if not file.is_file():
            continue
        dataset_path = file.relative_to(dataset_root).as_posix()
        if dataset_path in claimed_paths:
            continue  # 명시 태스크가 이미 가리키는 파일 — 이름 계보를 지킨다
        task_name = _task_name(file.stem, strip_suffix, name_template)
        if task_name in claimed_names:
            continue
        previous = seen_names.get(task_name)
        if previous is not None and previous != dataset_path:
            raise TaskDiscoveryError(
                f"discover produced duplicate task name {task_name!r} "
                f"for {previous} and {dataset_path}"
            )
        seen_names[task_name] = dataset_path
        tasks.append({"name": task_name, "dataset_path": dataset_path, **extra})
    return tasks


def _task_name(stem: str, strip_suffix: str, template: str) -> str:
    """파일 stem → 태스크 이름. 꼬리표를 떼고 `_` 를 `-` 로 바꾼다."""
    if strip_suffix and stem.endswith(strip_suffix):
        stem = stem[: -len(strip_suffix)]
    return template.replace(STEM_PLACEHOLDER, stem.replace("_", "-"))


def _dataset_root(config: dict[str, Any], data_root: Path | str | None) -> Path | None:
    """`dataset.repo` 를 데이터 루트 기준으로 푼다. 원격(HF) 데이터셋은 스캔하지 않는다."""
    dataset = config.get("dataset") or {}
    if dataset.get("source") not in (None, "local"):
        return None
    repo = dataset.get("repo")
    if not repo:
        return None
    repo_path = Path(repo)
    if repo_path.is_absolute():
        return repo_path
    if data_root is None:
        return None
    return Path(data_root) / repo_path