solar_eval/core/pipeline_compose.py

"""파이프라인 조립 — `extends` 상속 + dot-path override.

왜: 한 축만 바꾼 실험이 파이프라인 파일 **전체 사본**으로 쌓여 왔다.
`pipeline_dev_v24a/b/c` 는 109줄짜리 v24 사본인데 실제 차이는

    repeat_temperatures: [0.0, 0.2]  →  [0.0, 0.0] / [0.0, 0.3] / [0.0, 0.5]

한 줄뿐이다. 사본으로 두면 (1) 실험의 실제 변수가 diff 를 떠야만 보이고,
(2) base 를 고칠 때 사본들이 조용히 갈라진다. 실험의 정의는 사본이 아니라
**diff** 여야 한다.

두 가지 표현을 지원한다. 둘 다 같은 dot-path 문법을 쓴다:

    # 파일로 남기는 실험 (재사용·인용 대상)
    name: pipeline_dev_v24a
    extends: pipeline_dev_v24
    overrides:
      steps.basic_correction.repeat_temperatures: [0.0, 0.0]

    # 파일 없이 한 번 돌려보는 실험
    solar-eval runs start ... -P pipeline_dev_v24 \\
      --set steps.basic_correction.repeat_temperatures='[0.0, 0.5]'

조립은 **로드 시점에 끝난다** — 실행 엔진(`STEP_REGISTRY`)은 `extends` 를
아예 모르고, 완성된 steps 목록만 받는다. 그래서 이 기능은 엔진의 일반성에
영향을 주지 않는다(번역·요약 등 다른 프로젝트도 그대로 쓸 수 있다).

상속은 **1단계로 제한**한다. base 가 또 base 를 갖는 체인은 hydra 계열에서
반복적으로 문제가 된 지점이고("이 값이 어디서 왔는가"를 추적 불가), PoC 규모에서
얻는 게 없다.
"""

import copy
import logging
from collections.abc import Callable, Mapping
from typing import Any

import yaml

logger = logging.getLogger(__name__)

#: dot-path 에서 스텝 목록을 가리키는 최상위 키.
STEPS_KEY = "steps"


class PipelineCompositionError(ValueError):
    """`extends` 해석이나 override 적용이 실패했을 때.

    조용히 넘어가지 않는다 — 오타 난 override 는 "아무 효과 없는 실험"을 만들고,
    그건 잘못된 결론으로 바로 이어진다 (이 엔진이 과거 조용한 no-op 스텝으로
    같은 방식으로 다쳤다: `pipelines/errors.py` 참고).
    """


def compose_pipeline(
    raw: Mapping[str, Any],
    *,
    load_base: Callable[[str], Mapping[str, Any]] | None = None,
    overrides: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """파이프라인 YAML 한 장을 실행 가능한 설정으로 조립한다.

    Args:
        raw: 파이프라인 YAML 을 파싱한 dict.
        load_base: `extends` 대상 이름을 받아 그 YAML dict 를 돌려주는 함수.
            `extends` 를 쓴 파이프라인을 조립하려면 반드시 필요하다.
        overrides: 파일 바깥에서 들어온 override (CLI `--set`). YAML 의
            `overrides` 보다 **나중에** 적용된다.

    Returns:
        `extends`/`overrides` 키가 걷힌, 해석이 끝난 설정 dict.

    Raises:
        PipelineCompositionError: base 를 못 찾거나, 상속이 2단계 이상이거나,
            override 경로가 실재하지 않는 스텝을 가리킬 때.
    """
    base_name = raw.get("extends")
    if base_name is not None and not isinstance(base_name, str):
        raise PipelineCompositionError(f"`extends` must be a pipeline name, got {base_name!r}")

    if base_name:
        if load_base is None:
            raise PipelineCompositionError(
                f"Pipeline extends {base_name!r} but no loader was provided"
            )
        base = dict(load_base(base_name))
        if base.get("extends"):
            raise PipelineCompositionError(
                f"Multi-level inheritance is not supported: {base_name!r} itself extends "
                f"{base['extends']!r}. Flatten it or extend the root pipeline directly."
            )
        composed = _merge_base(base, raw)
    else:
        composed = copy.deepcopy({k: v for k, v in raw.items() if k != "extends"})

    file_overrides = composed.pop("overrides", None) or raw.get("overrides") or {}
    _apply_overrides(composed, file_overrides, origin="pipeline `overrides`")
    _apply_overrides(composed, overrides or {}, origin="--set")
    return composed


def _merge_base(base: dict[str, Any], child: Mapping[str, Any]) -> dict[str, Any]:
    """base 위에 child 의 최상위 키를 얹는다.

    `steps` 는 child 가 통째로 선언했을 때만 교체한다 — 부분 병합은 override 의
    일이고, 여기서까지 하면 "steps 가 어디서 왔는가"가 흐려진다.

    **깊은 복사가 필수다**: 얕게 복사하면 자식의 override 가 base 의 step dict 를
    제자리에서 고쳐, 같은 base 를 쓰는 다른 자식까지 오염된다.
    """
    merged = copy.deepcopy(base)
    for key, value in child.items():
        if key in ("extends", "overrides"):
            continue
        merged[key] = copy.deepcopy(value)
    return merged


def _apply_overrides(config: dict[str, Any], overrides: Mapping[str, Any], *, origin: str) -> None:
    """dot-path override 를 제자리에 적용한다."""
    if not isinstance(overrides, Mapping):
        raise PipelineCompositionError(
            f"{origin} must be a mapping of dot-path → value, got {type(overrides).__name__}"
        )
    for path, value in overrides.items():
        try:
            _set_path(config, str(path), value)
        except PipelineCompositionError as e:
            raise PipelineCompositionError(f"{origin}: {e}") from None


def _set_path(config: dict[str, Any], path: str, value: Any) -> None:
    """`steps.<step>.<field>...` 또는 `<field>...` 경로에 값을 쓴다."""
    parts = [p for p in path.split(".") if p]
    if not parts:
        raise PipelineCompositionError("empty override path")

    if parts[0] == STEPS_KEY:
        if len(parts) < 3:
            raise PipelineCompositionError(
                f"{path!r} — step override needs at least "
                f"'{STEPS_KEY}.<step_name>.<field>' (e.g. "
                f"'{STEPS_KEY}.basic_correction.reasoning_effort')"
            )
        target = _find_step(config, parts[1], path)
        _assign(target, parts[2:], value, path)
        return

    _assign(config, parts, value, path)


def _find_step(config: Mapping[str, Any], step_name: str, path: str) -> dict[str, Any]:
    """이름으로 스텝을 찾는다. 없으면 후보를 실어 raise (오타를 조용히 넘기지 않는다)."""
    steps = config.get(STEPS_KEY)
    if not isinstance(steps, list):
        raise PipelineCompositionError(f"{path!r} — pipeline has no `{STEPS_KEY}` list")

    matches = [s for s in steps if isinstance(s, Mapping) and s.get("name") == step_name]
    if not matches:
        names = [str(s.get("name")) for s in steps if isinstance(s, Mapping)]
        raise PipelineCompositionError(
            f"{path!r} — no step named {step_name!r} (steps: {', '.join(names)})"
        )
    if len(matches) > 1:
        raise PipelineCompositionError(
            f"{path!r} — {step_name!r} matches {len(matches)} steps; step names must be unique "
            f"to be addressable by override"
        )
    return matches[0]  # type: ignore[return-value]


def _assign(target: dict[str, Any], parts: list[str], value: Any, path: str) -> None:
    """중첩 dict 를 따라 내려가며 마지막 키에 값을 쓴다. 중간 dict 는 만들어 준다."""
    cursor: Any = target
    for key in parts[:-1]:
        nxt = cursor.get(key)
        if nxt is None:
            nxt = {}
            cursor[key] = nxt
        elif not isinstance(nxt, dict):
            raise PipelineCompositionError(
                f"{path!r} — {key!r} holds {type(nxt).__name__}, cannot descend into it"
            )
        cursor = nxt
    cursor[parts[-1]] = value


def parse_set_option(raw_values: tuple[str, ...] | list[str]) -> dict[str, Any]:
    """CLI `--set key=value` 목록을 override 맵으로 바꾼다.

    값은 YAML 로 파싱해 타입을 살린다 — `high` 는 문자열, `2` 는 정수,
    `[0.0, 0.5]` 는 리스트가 된다. 파싱에 실패하면 문자열 그대로 둔다.

    Raises:
        PipelineCompositionError: `=` 가 없는 항목이 있을 때.
    """
    overrides: dict[str, Any] = {}
    for item in raw_values:
        if "=" not in item:
            raise PipelineCompositionError(
                f"--set expects 'path=value', got {item!r} "
                f"(e.g. --set steps.basic_correction.reasoning_effort=high)"
            )
        key, _, raw_value = item.partition("=")
        key = key.strip()
        if not key:
            raise PipelineCompositionError(f"--set has an empty path: {item!r}")
        try:
            overrides[key] = yaml.safe_load(raw_value)
        except yaml.YAMLError:
            overrides[key] = raw_value
    return overrides