Diff Coverage

    # Optional accelerator fields that affect memory estimation.
    owss = model.get("optimizer_weight_shard_size")
    if owss and owss > 0:
        accel["optimizer_weight_shard_size"] = owss

    use_sp = model.get("use_seq_parallel", True)
    accel.setdefault("use_seq_parallel", bool(use_sp))

    gc_dict: Dict[str, Any] = {"activation_checkpoint": recompute}
    recompute_slice = model.get("recompute_slice_activation")
    if recompute_slice is not None:
        gc_dict["recompute_slice_activation"] = bool(recompute_slice)

    model_dict = _build_model_dict(model)

    hp_yaml: dict = {

        )
        if hasattr(target_ccfg.parser, "config_shard_emb"):
            target_ccfg.parser.config_shard_emb()
        if hasattr(target_ccfg.parser, "config_shard_recompute"):
            target_ccfg.parser.config_shard_recompute()
        target_ccfg.parser.config_dp_tp_exp(target_ccfg)
        target_ccfg.parser.config_optimizer_shard(target_ccfg)
        target_ccfg.parser.config_comm_flag(target_ccfg)
        if fr is not None:

                self._get_cfg_attr(overrides, "capacity_factor", None)
                or self._get_cfg_attr(overrides, "cap_fact", None)
            )
            if cap_val is not None:
                self.ccfg.cap_fact = max(1, float(cap_val))
            self.ccfg.gmm = (
                self._get_cfg_attr(overrides, "use_gmm", False)
                or self._get_cfg_attr(overrides, "gmm", False)
            )

        # optimizers.  Matches the MF parser's
        # ``self.ccfg.optimizer = self.config.optimizer.type``.
        opt_type = self._get_cfg_attr(optimizer, "type", None)
        if opt_type:
            self.ccfg.optimizer = str(opt_type)

    def _parse_recompute(self):
        """Parse recompute mode.

        gc = self._get_cfg_attr(train_raw, "gradient_checkpointing", Config({}))
        ac_mode = str(self._get_cfg_attr(gc, "activation_checkpoint", "none"))

        if full_rec_override is not None:
            self.ccfg.full_rec = full_rec_override
        else:
            self.ccfg.full_rec = ac_mode == "full"

        if sel_rec_override is not None:
            self.ccfg.sel_rec = sel_rec_override
        else:
            self.ccfg.sel_rec = ac_mode == "selective"

        self.ccfg.rec_op = Config({

        if self.ccfg.n_exp > 1 and self.ccfg.etp == 0:
            self.ccfg.etp = 1
        try:
            self.config_dp_tp_exp(self.ccfg)
        except TypeError:
            logger.warning(
                "MoE config_dp_tp_exp failed for d=%d t=%d ep=%d etp=%d "
                "n_exp=%d — clamping to minimum values.",
                self.ccfg.d, self.ccfg.t, self.ccfg.ep,
                self.ccfg.etp, self.ccfg.n_exp,

                "n_exp=%d — clamping to minimum values.",
                self.ccfg.d, self.ccfg.t, self.ccfg.ep,
                self.ccfg.etp, self.ccfg.n_exp,
            )
            self.ccfg.d_exp = max(1, self.ccfg.d_exp)
            self.ccfg.t_exp = max(1, self.ccfg.t_exp)
            self.ccfg.hff_exp = max(1, self.ccfg.hff_exp)
            self.ccfg.n_exp = max(1, self.ccfg.n_exp)

    def _init_offset(self):
        """Initialize the pipeline offset.

        model_raw = self._get_cfg_attr(self.config, "model", Config({}))
        overrides = self._get_cfg_attr(model_raw, "config_overrides", Config({}))
        explicit = self._get_cfg_attr(overrides, "offset", None)
        if isinstance(explicit, list):
            self.ccfg.offset = list(explicit)
        elif isinstance(explicit, int):
            if explicit == 0:
                self.ccfg.offset = 0
            else:
                self.ccfg.offset = [explicit] * self.ccfg.p
        else:
            self.ccfg.offset = [0] * self.ccfg.p

    def config_shard_emb(self) -> None:

        write_resolved_yaml,
    )

    if not os.path.isfile(cli_args.search_config):
        cli_parser.error(f"search-config not found: {cli_args.search_config}")
    if not os.path.isfile(cli_args.yaml_config):
        cli_parser.error(f"yaml-config not found: {cli_args.yaml_config}")

    set_verbose_level(cli_args.verbosity)

    search_cfg = read_search_config(cli_args.search_config)

    errors = validate(search_cfg)
    hard_errors = [e for e in errors if e.severity == "error"]
    warnings = [e for e in errors if e.severity == "warning"]
    for w in warnings:
        logger.warning("%s: %s", w.field_path, w.message)
    if hard_errors:
        for e in hard_errors:
            logger.error("%s: %s", e.field_path, e.message)
        cli_parser.error(

    search_cfg.resolved_strategy = result

    output_dir = cli_args.output_dir or "."
    if not os.path.isdir(output_dir):
        os.makedirs(output_dir, exist_ok=True)
    resolve_path = os.path.join(output_dir, "resolved.yaml")
    write_resolved_yaml(search_cfg, cli_args.yaml_config, resolve_path)
    logger.output("Resolved strategy written to %s", resolve_path)
    logger.output(