From db4a60c5db338e85c0ca375f2a96af924eb0624d Mon Sep 17 00:00:00 2001 From: Davids048 Date: Fri, 2 Oct 2026 09:55:15 +0000 Subject: [PATCH] [refactor]: add ordered config resolution with provenance Add fastvideo/api/resolution.py. resolve_generator_config parses a raw GeneratorConfig mapping and runs an ordered list of steps over it. Each step receives a read-only ResolutionView and returns {dotted_path: value}; it never assigns fields. The driver records every decision as (source, values), where source is the step's qualified name, and the last decision for a path wins. Explicit paths are taken from the keys of the raw mapping, so "the user set this" is never inferred from defaults. The result is a frozen ResolvedGeneratorConfig with the same nesting as GeneratorConfig, per-path provenance, and with_override(source, values) as the only way to change it after resolution; every override is logged. The module is not wired into FastVideoArgs, compat.py, or any entry point yet, so runtime behavior is unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012M91wnVFmPEvJ9h39r5BH7 --- fastvideo/api/resolution.py | 431 ++++++++++++++++++ .../api/test_resolution_generator_config.py | 250 ++++++++++ 2 files changed, 681 insertions(+) create mode 100644 fastvideo/api/resolution.py create mode 100644 fastvideo/tests/api/test_resolution_generator_config.py diff --git a/fastvideo/api/resolution.py b/fastvideo/api/resolution.py new file mode 100644 index 000000000..31aaacb16 --- /dev/null +++ b/fastvideo/api/resolution.py @@ -0,0 +1,431 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Ordered, traceable resolution of a typed config into a frozen result. + +Resolution turns the user's raw typed config (``GeneratorConfig``) into the +final values that runtime code reads. It follows three rules: + +* The raw input is kept as it was given. Explicit paths, the dotted leaf + paths that the user wrote, are recorded from the raw mapping when + resolution starts, so "the user set this" is never inferred from defaults. +* Resolution runs an ordered list of steps. A step is a plain function that + receives a read-only :class:`ResolutionView` and returns + ``{dotted_path: value}``. A step never assigns fields. The driver records + each returned mapping as a decision ``(source, values)``, where ``source`` + is the qualified name of the step; the last decision for a path wins. +* The result is frozen. :class:`ResolvedGeneratorConfig` mirrors the nested + structure of ``GeneratorConfig`` (``resolved.engine.offload.dit``), rejects + attribute assignment at every level, and reports the provenance of every + path. The only way to change a resolved config is + :meth:`ResolvedGeneratorConfig.with_override`, which validates the paths, + returns a separate frozen result, and appends ``(source, values)`` to an + override log. + +Paths address dataclass fields with dots (``engine.parallelism.sp_size``). +Dict-valued fields such as ``pipeline.experimental`` accept keys that the +input does not have (``pipeline.experimental.attention_backend``). A path +through a plain value, or through an optional nested config that is +``None``, is rejected. +""" +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +from copy import deepcopy +import dataclasses +from types import MappingProxyType +from typing import Any + +from fastvideo.api.parser import parse_config +from fastvideo.api.request_metadata import _record_value_paths +from fastvideo.api.schema import GeneratorConfig + +# Source name reported for a path that no step or override decided. +INPUT_SOURCE = "input" + +ResolutionStep = Callable[["ResolutionView"], Mapping[str, Any]] + +# One recorded change: (step or override source, {dotted_path: value}). +Decision = tuple[str, dict[str, Any]] + + +class ResolutionError(ValueError): + """A step or override named a path that the config does not have.""" + + +@dataclasses.dataclass(frozen=True) +class PathProvenance: + """Where the final value of one path came from. + + ``source`` is the qualified name of the step, or the source name of the + override, that last set the path or an enclosing path; it is + :data:`INPUT_SOURCE` when nothing changed the input value. ``raw_value`` + is the parsed input value including schema defaults, and ``raw_present`` + is false for dict keys that only resolution added. + """ + + path: str + value: Any + source: str + raw_value: Any + raw_present: bool + explicit: bool + + +class _Struct: + """A nested config dataclass, held as its class and a dict of field values.""" + + __slots__ = ("config_class", "fields") + + def __init__(self, config_class: type[Any], fields: dict[str, Any]): + self.config_class = config_class + self.fields = fields + + def __getstate__(self) -> tuple[type[Any], dict[str, Any]]: + return self.config_class, self.fields + + def __setstate__(self, state: tuple[type[Any], dict[str, Any]]) -> None: + self.config_class, self.fields = state + + +def _is_dataclass_instance(value: Any) -> bool: + return dataclasses.is_dataclass(value) and not isinstance(value, type) + + +def _to_tree(value: Any) -> Any: + """Copy a value into the resolution tree. + + Dataclasses become ``_Struct`` nodes, and read-only mappings that a step + took from :meth:`ResolutionView.get` become dicts again, so the tree only + holds plain, picklable containers. + """ + if _is_dataclass_instance(value): + return _Struct(type(value), + {field.name: _to_tree(getattr(value, field.name)) + for field in dataclasses.fields(value)}) + if isinstance(value, dict | MappingProxyType): + return {key: _to_tree(child) for key, child in value.items()} + if isinstance(value, list): + return [_to_tree(child) for child in value] + if isinstance(value, tuple): + return tuple(_to_tree(child) for child in value) + return deepcopy(value) + + +def _to_plain(node: Any) -> Any: + """Copy a resolution tree into plain nested dicts, lists, and values.""" + if isinstance(node, _Struct): + return {name: _to_plain(child) for name, child in node.fields.items()} + if isinstance(node, dict): + return {key: _to_plain(child) for key, child in node.items()} + if isinstance(node, list): + return [_to_plain(child) for child in node] + return deepcopy(node) + + +def _to_config(node: Any) -> Any: + """Rebuild typed config dataclasses from a resolution tree.""" + if isinstance(node, _Struct): + return node.config_class(**{name: _to_config(child) for name, child in node.fields.items()}) + return deepcopy(node) + + +def _freeze(value: Any) -> Any: + """Read-only copy of a leaf value: dicts become mapping proxies, lists and sets become tuples and frozensets.""" + if isinstance(value, dict): + return MappingProxyType({key: _freeze(child) for key, child in value.items()}) + if isinstance(value, list | tuple): + return tuple(_freeze(child) for child in value) + if isinstance(value, set): + return frozenset(_freeze(child) for child in value) + return deepcopy(value) + + +def _split(path: str) -> list[str] | None: + """Dotted path parts, or ``None`` when ``path`` is not a non-empty dotted string.""" + if not isinstance(path, str) or not path or any(not part for part in path.split(".")): + return None + return path.split(".") + + +def _lookup(tree: _Struct, path: str) -> tuple[bool, Any]: + """Return ``(present, value)`` for a dotted path, walking structure fields and dict keys.""" + parts = _split(path) + if parts is None: + return False, None + node: Any = tree + for part in parts: + if isinstance(node, _Struct) and part in node.fields: + node = node.fields[part] + elif isinstance(node, dict) and part in node: + node = node[part] + else: + return False, None + return True, node + + +def _assign(tree: _Struct, path: str, value: Any, source: str) -> None: + """Set one dotted path in the tree, rejecting paths that the config does not have. + + Intermediate parts must name existing structure fields or existing dict + keys. The last part must name an existing structure field, or any key of + a dict-valued field. A field that holds a nested structure accepts only a + dataclass instance; an optional nested config that is ``None`` accepts a + dataclass instance to install it. + """ + parts = _split(path) + if parts is None: + raise ResolutionError(f"{source}: {path!r} is not a dotted config path") + node: Any = tree + for depth, part in enumerate(parts[:-1]): + if isinstance(node, _Struct) and part in node.fields: + node = node.fields[part] + elif isinstance(node, dict) and part in node: + node = node[part] + else: + walked = ".".join(parts[:depth + 1]) + raise ResolutionError(f"{source}: {path!r} goes through {walked!r}, which is not a config path") + last = parts[-1] + if isinstance(node, dict): + node[last] = _to_tree(value) + return + if not isinstance(node, _Struct): + parent = ".".join(parts[:-1]) + raise ResolutionError(f"{source}: {path!r} goes through {parent!r}, which holds a plain value") + if last not in node.fields: + raise ResolutionError(f"{source}: {path!r} is not a field of {node.config_class.__name__}") + if isinstance(node.fields[last], _Struct) and not _is_dataclass_instance(value): + raise ResolutionError(f"{source}: {path!r} is a nested config; set its fields or pass a dataclass instance") + node.fields[last] = _to_tree(value) + + +def _apply(tree: _Struct, source: str, values: Mapping[str, Any]) -> Decision: + """Validate one decision on a copy of the tree, then apply it, so a rejected path leaves ``tree`` unchanged.""" + if not isinstance(values, Mapping): + raise ResolutionError(f"{source}: expected a mapping of config paths, got {type(values).__name__}") + trial = deepcopy(tree) + for path, value in values.items(): + _assign(trial, path, value, source) + for path, value in values.items(): + _assign(tree, path, value, source) + return source, {path: _to_plain(_to_tree(value)) for path, value in values.items()} + + +def _is_explicit(explicit_paths: frozenset[str], path: str) -> bool: + """A path is explicit when the user wrote it or wrote a path inside it.""" + return path in explicit_paths or any(written.startswith(path + ".") for written in explicit_paths) + + +def _last_source(log: Sequence[Decision], path: str) -> str: + """Source of the latest log entry that set ``path`` or an enclosing path.""" + for source, values in reversed(log): + for decided in values: + if path == decided or path.startswith(decided + "."): + return source + return INPUT_SOURCE + + +class ResolutionView: + """Read-only view that resolution steps receive. + + Reads return the latest decision for a path, else the input value. + Nested structures come back as read-only nodes and container values as + read-only copies, so a step can change the config only by returning a + decision. + """ + + __slots__ = ("_tree", "_explicit_paths", "_decisions") + + def __init__(self, tree: _Struct, explicit_paths: frozenset[str], decisions: list[Decision]): + self._tree = tree + self._explicit_paths = explicit_paths + self._decisions = decisions + + def get(self, path: str) -> Any: + present, value = _lookup(self._tree, path) + if not present: + raise ResolutionError(f"{path!r} is not a path of this config") + return _FrozenNode(deepcopy(value)) if isinstance(value, _Struct) else _freeze(value) + + def is_explicit(self, path: str) -> bool: + return _is_explicit(self._explicit_paths, path) + + def decided_by(self, path: str) -> str: + return _last_source(self._decisions, path) + + +class _FrozenNode: + """Read-only attribute access to one nested structure of a resolved config.""" + + __slots__ = ("_struct", ) + + def __init__(self, struct: _Struct): + object.__setattr__(self, "_struct", struct) + + def __getattr__(self, name: str) -> Any: + struct = object.__getattribute__(self, "_struct") + if name not in struct.fields: + raise AttributeError(f"{struct.config_class.__name__} has no field {name!r}") + child = struct.fields[name] + return _FrozenNode(child) if isinstance(child, _Struct) else _freeze(child) + + def __setattr__(self, name: str, value: Any) -> None: + raise AttributeError(f"resolved config is read-only; cannot set {name!r}") + + def __delattr__(self, name: str) -> None: + raise AttributeError(f"resolved config is read-only; cannot delete {name!r}") + + def __repr__(self) -> str: + struct = object.__getattribute__(self, "_struct") + return f"{struct.config_class.__name__}({_to_plain(struct)!r})" + + +class _ResolvedTree(_FrozenNode): + """Frozen resolution result of any typed config, with per-path provenance and an override log. + + Fields are read by attribute with the same nesting as the typed config. + The methods below are reserved names and take precedence over top-level + fields of the same name. + """ + + __slots__ = ("_raw_tree", "_explicit_paths", "_decisions", "_override_log") + + def __init__( + self, + tree: _Struct, + raw_tree: _Struct, + explicit_paths: frozenset[str], + decisions: Sequence[Decision], + override_log: Sequence[Decision], + ): + super().__init__(tree) + object.__setattr__(self, "_raw_tree", raw_tree) + object.__setattr__(self, "_explicit_paths", explicit_paths) + object.__setattr__(self, "_decisions", tuple(decisions)) + object.__setattr__(self, "_override_log", tuple(override_log)) + + def __reduce__(self) -> tuple[Any, ...]: + state = object.__getattribute__ + return type(self), (state(self, "_struct"), state(self, "_raw_tree"), state(self, "_explicit_paths"), + state(self, "_decisions"), state(self, "_override_log")) + + @property + def decisions(self) -> tuple[Decision, ...]: + """Resolution decisions in the order that the steps made them.""" + return deepcopy(object.__getattribute__(self, "_decisions")) + + @property + def override_log(self) -> tuple[Decision, ...]: + """Changes applied after resolution by :meth:`with_override`, in order.""" + return deepcopy(object.__getattribute__(self, "_override_log")) + + def is_explicit(self, path: str) -> bool: + return _is_explicit(object.__getattribute__(self, "_explicit_paths"), path) + + def provenance(self, path: str) -> PathProvenance: + """Final value, deciding source, input value, and explicit flag of one path.""" + present, value = _lookup(object.__getattribute__(self, "_struct"), path) + if not present: + raise ResolutionError(f"{path!r} is not a path of this config") + raw_present, raw_value = _lookup(object.__getattribute__(self, "_raw_tree"), path) + log = object.__getattribute__(self, "_decisions") + object.__getattribute__(self, "_override_log") + return PathProvenance( + path=path, + value=_to_plain(value), + source=_last_source(log, path), + raw_value=_to_plain(raw_value), + raw_present=raw_present, + explicit=self.is_explicit(path), + ) + + def provenance_table(self) -> list[PathProvenance]: + """Provenance of every leaf field and of every path that a step or override set, sorted by path.""" + paths: set[str] = set() + + def walk(node: _Struct, prefix: str) -> None: + for name, child in node.fields.items(): + path = f"{prefix}.{name}" if prefix else name + if isinstance(child, _Struct): + walk(child, path) + else: + paths.add(path) + + walk(object.__getattribute__(self, "_struct"), "") + for _, values in object.__getattribute__(self, "_decisions") + object.__getattribute__(self, "_override_log"): + paths.update(values) + return [self.provenance(path) for path in sorted(paths)] + + def with_override(self, source: str, values: Mapping[str, Any]) -> Any: + """Return a separate frozen result with ``values`` applied and ``(source, values)`` added to the override log. + + Use this for decisions that can only be made after resolution, such + as ones that depend on the device that a worker binds. Paths are + validated exactly as for resolution steps; this object is unchanged. + """ + if not isinstance(source, str) or not source: + raise ResolutionError("an override needs a non-empty source name") + tree = deepcopy(object.__getattribute__(self, "_struct")) + override = _apply(tree, source, values) + return type(self)( + tree, + object.__getattribute__(self, "_raw_tree"), + object.__getattribute__(self, "_explicit_paths"), + object.__getattribute__(self, "_decisions"), + object.__getattribute__(self, "_override_log") + (override, ), + ) + + def to_dict(self) -> dict[str, Any]: + """Plain nested dict copy of the resolved values.""" + return _to_plain(object.__getattribute__(self, "_struct")) + + +class ResolvedGeneratorConfig(_ResolvedTree): + """Frozen result of resolving a ``GeneratorConfig``, with the provenance of every path.""" + + __slots__ = () + + def with_override(self, source: str, values: Mapping[str, Any]) -> ResolvedGeneratorConfig: + return super().with_override(source, values) + + def to_config(self) -> GeneratorConfig: + """Typed ``GeneratorConfig`` copy of the resolved values, for code that consumes the typed config.""" + return _to_config(object.__getattribute__(self, "_struct")) + + +def _run_steps(typed_config: Any, explicit_paths: frozenset[str], + steps: Sequence[ResolutionStep]) -> tuple[_Struct, _Struct, list[Decision]]: + """Run ``steps`` in order over a parsed typed config; return the resolved tree, the input tree, and decisions.""" + raw_tree = _to_tree(typed_config) + tree = deepcopy(raw_tree) + decisions: list[Decision] = [] + view = ResolutionView(tree, explicit_paths, decisions) + for step in steps: + source = getattr(step, "__qualname__", repr(step)) + values = step(view) + if values: + decisions.append(_apply(tree, source, values)) + return tree, raw_tree, decisions + + +def resolve_generator_config(raw: Mapping[str, Any], steps: Sequence[ResolutionStep]) -> ResolvedGeneratorConfig: + """Parse a raw generator config mapping, run ``steps`` in order, and freeze the result. + + ``raw`` is the merged YAML, CLI-override, and keyword input in the nested + shape of ``GeneratorConfig``. Every leaf written in ``raw`` is an + explicit path. Each step sees the decisions of the steps before it. + """ + if not isinstance(raw, Mapping): + raise TypeError(f"expected a raw config mapping, got {type(raw).__name__}") + explicit_paths: set[str] = set() + _record_value_paths(raw, "", explicit_paths) + tree, raw_tree, decisions = _run_steps(parse_config(GeneratorConfig, raw), frozenset(explicit_paths), steps) + return ResolvedGeneratorConfig(tree, raw_tree, frozenset(explicit_paths), decisions, ()) + + +__all__ = [ + "INPUT_SOURCE", + "PathProvenance", + "ResolutionError", + "ResolutionStep", + "ResolutionView", + "ResolvedGeneratorConfig", + "resolve_generator_config", +] diff --git a/fastvideo/tests/api/test_resolution_generator_config.py b/fastvideo/tests/api/test_resolution_generator_config.py new file mode 100644 index 000000000..256311bf7 --- /dev/null +++ b/fastvideo/tests/api/test_resolution_generator_config.py @@ -0,0 +1,250 @@ +# SPDX-License-Identifier: Apache-2.0 +"""CPU tests for resolving a GeneratorConfig into a frozen ResolvedGeneratorConfig.""" +import pickle + +import pytest +import yaml + +from fastvideo.api.resolution import ( + INPUT_SOURCE, + ResolutionError, + ResolutionView, + ResolvedGeneratorConfig, + resolve_generator_config, +) +from fastvideo.api.schema import GeneratorConfig, QuantizationConfig + +MODEL_PATH = "/models/wan" + + +def fill_dit_offload_if_unset(view: ResolutionView) -> dict: + if view.is_explicit("engine.offload.dit"): + return {} + return {"engine.offload.dit": False} + + +def set_sp_size_two(view: ResolutionView) -> dict: + return {"engine.parallelism.sp_size": 2} + + +def set_sp_size_four(view: ResolutionView) -> dict: + return {"engine.parallelism.sp_size": 4} + + +def test_explicit_input_beats_fill_if_unset_step() -> None: + explicit = resolve_generator_config( + {"model_path": MODEL_PATH, "engine": {"offload": {"dit": True}}}, + [fill_dit_offload_if_unset], + ) + assert explicit.engine.offload.dit is True + assert explicit.provenance("engine.offload.dit").source == INPUT_SOURCE + assert explicit.provenance("engine.offload.dit").explicit is True + assert explicit.decisions == () + + unset = resolve_generator_config({"model_path": MODEL_PATH}, [fill_dit_offload_if_unset]) + assert unset.engine.offload.dit is False + provenance = unset.provenance("engine.offload.dit") + assert provenance.source == "fill_dit_offload_if_unset" + assert provenance.raw_value is True + assert provenance.explicit is False + + +def test_later_step_overrides_earlier_step_and_provenance_names_it() -> None: + seen: dict = {} + + def check_previous_decision(view: ResolutionView) -> dict: + seen["sp_size"] = view.get("engine.parallelism.sp_size") + seen["decided_by"] = view.decided_by("engine.parallelism.sp_size") + return {} + + resolved = resolve_generator_config( + {"model_path": MODEL_PATH}, + [set_sp_size_two, check_previous_decision, set_sp_size_four], + ) + + assert seen == {"sp_size": 2, "decided_by": "set_sp_size_two"} + assert resolved.engine.parallelism.sp_size == 4 + assert resolved.provenance("engine.parallelism.sp_size").source == "set_sp_size_four" + assert resolved.provenance("engine.parallelism.sp_size").raw_value == -1 + assert resolved.provenance("engine.parallelism.tp_size").source == INPUT_SOURCE + assert resolved.decisions == ( + ("set_sp_size_two", {"engine.parallelism.sp_size": 2}), + ("set_sp_size_four", {"engine.parallelism.sp_size": 4}), + ) + + +@pytest.mark.parametrize( + "path", + [ + "engine.not_a_field", + "engine.num_gpus.extra", + "engine.quantization.transformer_quant", + "pipeline.experimental.missing_section.key", + "engine..num_gpus", + ], +) +def test_unknown_paths_are_rejected_with_step_name(path: str) -> None: + + def write_bad_path(view: ResolutionView) -> dict: + return {path: 1} + + with pytest.raises(ResolutionError, match="write_bad_path"): + resolve_generator_config({"model_path": MODEL_PATH}, [write_bad_path]) + + +def test_wildcard_dicts_and_optional_configs_accept_new_values() -> None: + + def add_experimental_key(view: ResolutionView) -> dict: + return {"pipeline.experimental.attention_backend": "SAGE_ATTN"} + + def install_quantization(view: ResolutionView) -> dict: + return {"engine.quantization": QuantizationConfig()} + + def set_quantization_field(view: ResolutionView) -> dict: + return {"engine.quantization.transformer_quant": "fp8"} + + resolved = resolve_generator_config( + {"model_path": MODEL_PATH}, + [add_experimental_key, install_quantization, set_quantization_field], + ) + + assert resolved.pipeline.experimental["attention_backend"] == "SAGE_ATTN" + provenance = resolved.provenance("pipeline.experimental.attention_backend") + assert provenance.raw_present is False + assert provenance.source.endswith("add_experimental_key") + assert resolved.engine.quantization.transformer_quant == "fp8" + + +def test_step_must_return_a_mapping() -> None: + + def return_list(view: ResolutionView) -> list: + return ["engine.num_gpus"] + + with pytest.raises(ResolutionError, match="return_list"): + resolve_generator_config({"model_path": MODEL_PATH}, [return_list]) + + +def test_assignment_raises_at_every_nesting_level() -> None: + resolved = resolve_generator_config( + {"model_path": MODEL_PATH, "pipeline": {"experimental": {"flag": [1, 2]}}}, + [], + ) + + with pytest.raises(AttributeError): + resolved.model_path = "/other" + with pytest.raises(AttributeError): + resolved.engine = None + with pytest.raises(AttributeError): + resolved.engine.num_gpus = 8 + with pytest.raises(AttributeError): + resolved.engine.offload.dit = False + with pytest.raises(AttributeError): + del resolved.engine.parallelism + with pytest.raises(TypeError): + resolved.pipeline.experimental["flag"] = 3 + with pytest.raises(AttributeError): + resolved.pipeline.experimental["flag"].append(3) + assert resolved.pipeline.experimental["flag"] == (1, 2) + + def mutate_view_value(view: ResolutionView) -> dict: + view.get("pipeline.experimental")["flag"] = 3 + return {} + + with pytest.raises(TypeError): + resolve_generator_config({"model_path": MODEL_PATH}, [mutate_view_value]) + + +def test_post_freeze_change_is_validated_and_logged() -> None: + resolved = resolve_generator_config({"model_path": MODEL_PATH}, [set_sp_size_two]) + + changed = resolved.with_override("worker.bind_device", {"engine.offload.dit": False}) + + assert isinstance(changed, ResolvedGeneratorConfig) + assert changed.engine.offload.dit is False + assert resolved.engine.offload.dit is True + assert resolved.override_log == () + assert changed.override_log == (("worker.bind_device", {"engine.offload.dit": False}), ) + assert changed.decisions == resolved.decisions + assert changed.provenance("engine.offload.dit").source == "worker.bind_device" + assert changed.provenance("engine.parallelism.sp_size").source == "set_sp_size_two" + assert changed.to_config().engine.offload.dit is False + + with pytest.raises(ResolutionError, match="worker.bind_device"): + changed.with_override("worker.bind_device", {"engine.offload.dit": False, "engine.not_a_field": 1}) + assert changed.override_log == (("worker.bind_device", {"engine.offload.dit": False}), ) + + +def test_pickle_round_trip_keeps_values_and_provenance() -> None: + resolved = resolve_generator_config( + {"model_path": MODEL_PATH, "pipeline": {"experimental": {"flag": True}}}, + [fill_dit_offload_if_unset, set_sp_size_two], + ).with_override("worker.bind_device", {"engine.parallelism.sp_size": 8}) + + restored = pickle.loads(pickle.dumps(resolved)) + + assert isinstance(restored, ResolvedGeneratorConfig) + assert restored.to_dict() == resolved.to_dict() + assert restored.to_config() == resolved.to_config() + assert restored.provenance_table() == resolved.provenance_table() + assert restored.decisions == resolved.decisions + assert restored.override_log == resolved.override_log + assert restored.is_explicit("pipeline.experimental.flag") is True + with pytest.raises(AttributeError): + restored.engine.num_gpus = 2 + + +def test_view_reports_explicitness_for_yaml_mapping() -> None: + raw = yaml.safe_load(""" +model_path: /models/wan +engine: + num_gpus: 1 + parallelism: + sp_size: 2 + quantization: + transformer_quant: fp8 +pipeline: + preset_overrides: {} + experimental: + attention_backend: SAGE_ATTN +""") + queries = [ + "model_path", + "revision", + "engine.num_gpus", + "engine.parallelism", + "engine.parallelism.sp_size", + "engine.parallelism.tp_size", + "engine.offload", + "engine.offload.dit", + "engine.quantization.transformer_quant", + "engine.quantization.text_encoder_quant", + "pipeline.preset_overrides", + "pipeline.experimental.attention_backend", + "pipeline.experimental.other_key", + ] + seen: dict = {} + + def record_explicitness(view: ResolutionView) -> dict: + seen.update({path: view.is_explicit(path) for path in queries}) + return {} + + resolved = resolve_generator_config(raw, [record_explicitness]) + + assert seen == { + "model_path": True, + "revision": False, + # Written with its default value; still explicit because the user wrote it. + "engine.num_gpus": True, + "engine.parallelism": True, + "engine.parallelism.sp_size": True, + "engine.parallelism.tp_size": False, + "engine.offload": False, + "engine.offload.dit": False, + "engine.quantization.transformer_quant": True, + "engine.quantization.text_encoder_quant": False, + "pipeline.preset_overrides": True, + "pipeline.experimental.attention_backend": True, + "pipeline.experimental.other_key": False, + } + assert all(resolved.is_explicit(path) is explicit for path, explicit in seen.items()) + assert isinstance(resolved.to_config(), GeneratorConfig)