从一段吉他音频，
到一串 DSP 控制指令。

$scriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
$projectDir = (Resolve-Path (Join-Path $scriptDir "..")).Path
$mainDir = (Resolve-Path (Join-Path $projectDir "Main")).Path
$workspaceDir = Split-Path -Parent $projectDir

if (-not $DataProjectRoot) {
    $localManifest = Join-Path $projectDir "Audio\Samples\Wet_Main\manifest_full.jsonl"
    $v2ProjectRoot = Join-Path $workspaceDir "AI_Guitar_Pedal_V2"
    $v2Manifest = Join-Path $v2ProjectRoot "Audio\Samples\Wet_Main\manifest_full.jsonl"
    if (Test-Path $localManifest) {
        $DataProjectRoot = $projectDir
    } elseif (Test-Path $v2Manifest) {
        $DataProjectRoot = $v2ProjectRoot
    } else {
        throw "No complete Phase 2 data tree found. Pass -DataProjectRoot explicitly."
    }
}
$DataProjectRoot = Resolve-ExistingPath -PathValue $DataProjectRoot

def bootstrap_storage_env_from_argv(argv: Sequence[str]) -> None:
    raw_e_root = early_arg_value(argv, "--e_root") or E_ROOT
    try:
        e_root = resolve_e_root(raw_e_root)
        storage = ensure_storage_tree(e_root)
        apply_env_defaults(storage, set_temp_tmp=True)
    except Exception:
        return

    def __init__(self, token_to_id: dict[str, int]):
        if not token_to_id:
            raise ValueError("DSP control vocabulary is empty")
        ids = list(token_to_id.values())
        if len(ids) != len(set(ids)):
            raise ValueError("DSP control vocabulary contains duplicate token IDs")
        if sorted(ids) != list(range(len(ids))):
            raise ValueError("DSP control token IDs must be contiguous from zero")

        self.token_to_id = dict(token_to_id)
        self.id_to_token = {token_id: token for token, token_id in token_to_id.items()}
        self.vocab_size = len(token_to_id)
        self.pad_token_id = self._required_id("<PAD>")
        self.bos_token_id = self._required_id("<BOS>")
        self.eos_token_id = self._required_id("<EOS>")

    def decode_ids(self, token_ids: Iterable[int]) -> list[str]:
        tokens = []
        for token_id in token_ids:
            token_id = int(token_id)
            if token_id not in self.id_to_token:
                raise ValueError(f"Unknown DSP control token ID: {token_id}")
            tokens.append(self.id_to_token[token_id])
        return tokens

def read_program_token_ids(
    program_path: Path,
    tokenizer: DSPControlTokenizer,
    strip_bos: bool,
) -> List[int]:
    payload = json.loads(program_path.read_text(encoding="utf-8"))
    token_ids = payload.get("model_token_ids")
    if not isinstance(token_ids, list) or not token_ids:
        raise ValueError(f"program missing model_token_ids: {program_path}")
    out = [int(x) for x in token_ids]
    if strip_bos and out and out[0] == tokenizer.bos_token_id:
        out = out[1:]
    if not out or out[-1] != tokenizer.eos_token_id:
        raise ValueError(f"program token sequence must end with EOS: {program_path}")
    bad = [x for x in out if x < 0 or x >= tokenizer.vocab_size]
    if bad:
        raise ValueError(f"program token id out of vocabulary: {program_path}, examples={bad[:10]}")
    return out

            cached = program_cache[program_path]
            rows.append(
                {
                    "audio": str(audio_path),
                    "labels": cached["labels"],
                    "program_hash": cached["program_hash"],
                    "target_length": cached["target_length"],
                }
            )

def token_hash(token_ids: Sequence[int]) -> str:
    text = ",".join(str(int(x)) for x in token_ids)
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def split_by_program_hash(
    rows: Sequence[Dict[str, Any]],
    eval_ratio: float,
    seed: int,
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    if not (0.0 <= eval_ratio < 1.0):
        raise ValueError(f"eval_ratio must be in [0,1), got {eval_ratio}")
    grouped: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
    for row in rows:
        grouped[str(row["program_hash"])].append(row)

    keys = sorted(grouped)
    rng = random.Random(seed)
    rng.shuffle(keys)
    if eval_ratio <= 0.0 or len(keys) <= 1:
        eval_keys = set()
    else:
        eval_count = int(round(len(keys) * eval_ratio))
        eval_count = max(1, eval_count)
        eval_count = min(eval_count, len(keys) - 1)
        eval_keys = set(keys[:eval_count])

    train_rows: List[Dict[str, Any]] = []
    eval_rows: List[Dict[str, Any]] = []
    for key in keys:
        if key in eval_keys:
            eval_rows.extend(grouped[key])
        else:
            train_rows.extend(grouped[key])
    rng.shuffle(train_rows)
    rng.shuffle(eval_rows)
    if not train_rows:
        raise ValueError("No train rows after program-hash split.")
    return train_rows, eval_rows

def load_audio_stereo_16k(path: Path, target_sr: int = TARGET_SR) -> np.ndarray:
    audio, sr = sf.read(str(path), always_2d=True)
    audio = np.asarray(audio, dtype=np.float32)
    if audio.size == 0:
        raise ValueError(f"Empty audio file: {path}")

    if audio.shape[1] == 1:
        audio = np.repeat(audio, STEREO_CHANNELS, axis=1)
    elif audio.shape[1] < STEREO_CHANNELS:
        raise ValueError(f"Expected at least one audio channel, got shape {audio.shape}: {path}")
    elif audio.shape[1] > STEREO_CHANNELS:
        audio = audio[:, :STEREO_CHANNELS]

    stereo = audio.T
    if int(sr) != int(target_sr):
        stereo = np.stack(
            [
                librosa.resample(channel, orig_sr=int(sr), target_sr=int(target_sr))
                for channel in stereo
            ],
            axis=0,
        )
    return np.asarray(stereo, dtype=np.float32)

def fix_audio_length(waveform: np.ndarray, max_len: int) -> np.ndarray:
    if waveform.ndim != 2 or waveform.shape[0] != STEREO_CHANNELS:
        raise ValueError(f"Expected stereo waveform with shape [2, T], got {waveform.shape}")
    if waveform.shape[1] > max_len:
        waveform = waveform[:, :max_len]
    elif waveform.shape[1] < max_len:
        waveform = np.pad(
            waveform,
            ((0, 0), (0, max_len - waveform.shape[1])),
            mode="constant",
        )
    return waveform.astype(np.float32, copy=False)

def stereo_to_features(
    feature_extractor: AutoFeatureExtractor,
    waveform: np.ndarray,
) -> torch.Tensor:
    if waveform.ndim != 2 or waveform.shape[0] != STEREO_CHANNELS:
        raise ValueError(f"Expected stereo waveform with shape [2, T], got {waveform.shape}")

    channel_features: list[torch.Tensor] = []
    for channel in waveform:
        features = feature_extractor(
            channel,
            sampling_rate=TARGET_SR,
            return_tensors="pt",
        ).input_features
        channel_features.append(features[0].cpu().to(torch.float32))

    if len(channel_features) != STEREO_CHANNELS:
        raise ValueError(f"Expected {STEREO_CHANNELS} channel feature tensors")
    if channel_features[0].shape != channel_features[1].shape:
        raise ValueError(
            "Stereo channel feature shapes differ: "
            f"{channel_features[0].shape} vs {channel_features[1].shape}"
        )
    return torch.cat(channel_features, dim=0).to(torch.float32)

def feature_cache_base_dir(
    feature_cache_root: Path,
    model_name: str,
    max_seconds: float,
) -> Path:
    model_id = normalize_model_name_for_cache(model_name)
    model_tag = re.sub(r"[^A-Za-z0-9._-]+", "_", model_id).strip("_")
    max_tag = f"{max_seconds:.3f}".replace(".", "p")
    return feature_cache_root / model_tag / f"sr{TARGET_SR}_max{max_tag}_{AUDIO_CHANNEL_MODE}"

def load_or_compute_input_features(
    wav_path: Path,
    feature_extractor: AutoFeatureExtractor,
    max_len: int,
    use_feature_cache: bool,
    cache_path: Path,
    save_on_miss: bool,
) -> torch.Tensor:
    if use_feature_cache and cache_path.exists():
        tensor = torch.load(cache_path, map_location="cpu")
        if not isinstance(tensor, torch.Tensor):
            raise ValueError(f"Invalid cached tensor format: {cache_path}")
        if tensor.ndim < 2 or int(tensor.shape[0]) != STEREO_FEATURE_BINS:
            raise ValueError(
                f"Cached features are not stereo {STEREO_FEATURE_BINS}-bin tensors: "
                f"{cache_path} shape={tuple(tensor.shape)}"
            )
        return tensor.to(torch.float32)

    tensor = compute_input_features(
        wav_path=wav_path,
        feature_extractor=feature_extractor,
        max_len=max_len,
    )
    if use_feature_cache and save_on_miss:
        atomic_torch_save(tensor, cache_path)
    return tensor

def atomic_torch_save(tensor: torch.Tensor, path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + f".tmp_{os.getpid()}")
    torch.save(tensor.cpu(), tmp)
    os.replace(tmp, path)

    def __getitem__(self, idx: int) -> Dict[str, torch.Tensor]:
        row = self.rows[idx]
        wav_path = Path(str(row["audio"])).expanduser().resolve()
        cache_path = feature_cache_path_for_audio(
            wav_path=wav_path,
            audio_root=self.audio_root,
            cache_base=self.cache_base,
        )
        input_features = load_or_compute_input_features(
            wav_path=wav_path,
            feature_extractor=self.feature_extractor,
            max_len=self.max_len,
            use_feature_cache=self.use_feature_cache,
            cache_path=cache_path,
            save_on_miss=self.use_feature_cache,
        )
        return {
            "input_features": input_features,
            "labels": torch.tensor(row["labels"], dtype=torch.long),
        }

class Seq2SeqCollator:
    def __init__(self, pad_token_id: int, label_pad_token_id: int = -100) -> None:
        self.pad_token_id = int(pad_token_id)
        self.label_pad_token_id = int(label_pad_token_id)

    def __call__(self, items: List[Dict[str, torch.Tensor]]) -> Dict[str, torch.Tensor]:
        input_features = torch.stack([item["input_features"] for item in items], dim=0)
        labels = [item["labels"] for item in items]
        max_len = max(int(label.shape[0]) for label in labels)
        padded = torch.full(
            (len(labels), max_len),
            self.label_pad_token_id,
            dtype=torch.long,
        )
        for idx, label in enumerate(labels):
            padded[idx, : label.shape[0]] = label
        return {"input_features": input_features, "labels": padded}

def build_model(args: argparse.Namespace, tokenizer: DSPControlTokenizer) -> WhisperForConditionalGeneration:
    config = WhisperConfig.from_pretrained(args.model_name)
    requested_target_positions = int(args.max_target_length)
    if requested_target_positions <= 0:
        raise ValueError(f"max_target_length must be positive, got {requested_target_positions}")
    config.max_target_positions = max(
        requested_target_positions,
        int(getattr(config, "max_target_positions", 0) or 0),
    )
    config.vocab_size = int(tokenizer.vocab_size)
    config.pad_token_id = int(tokenizer.pad_token_id)
    config.bos_token_id = int(tokenizer.bos_token_id)
    config.eos_token_id = int(tokenizer.eos_token_id)
    config.decoder_start_token_id = int(tokenizer.bos_token_id)
    config.forced_decoder_ids = None
    config.suppress_tokens = []
    config.begin_suppress_tokens = []

    model = WhisperForConditionalGeneration.from_pretrained(
        args.model_name,
        config=config,
        ignore_mismatched_sizes=True,
    )
    adapt_whisper_model_for_stereo(model)
    model.config.use_cache = False
    if hasattr(model, "generation_config"):
        model.generation_config.pad_token_id = int(tokenizer.pad_token_id)
        model.generation_config.bos_token_id = int(tokenizer.bos_token_id)
        model.generation_config.eos_token_id = int(tokenizer.eos_token_id)
        model.generation_config.decoder_start_token_id = int(tokenizer.bos_token_id)
        model.generation_config.forced_decoder_ids = None
        model.generation_config.suppress_tokens = []
        model.generation_config.begin_suppress_tokens = []
        model.generation_config.max_length = int(args.max_target_length)
    if args.gradient_checkpointing:
        model.gradient_checkpointing_enable()
    return model

def adapt_whisper_model_for_stereo(model: torch.nn.Module) -> None:
    conv1 = whisper_encoder_first_conv(model)
    if conv1.in_channels == STEREO_FEATURE_BINS:
        setattr(model.config, "num_mel_bins", STEREO_FEATURE_BINS)
        return
    if conv1.in_channels != MONO_FEATURE_BINS:
        raise RuntimeError(
            f"Unsupported Whisper conv1 input channels: {conv1.in_channels}; "
            f"expected {MONO_FEATURE_BINS} or {STEREO_FEATURE_BINS}"
        )

    new_conv = torch.nn.Conv1d(
        in_channels=STEREO_FEATURE_BINS,
        out_channels=conv1.out_channels,
        kernel_size=conv1.kernel_size,
        stride=conv1.stride,
        padding=conv1.padding,
        dilation=conv1.dilation,
        groups=conv1.groups,
        bias=conv1.bias is not None,
        padding_mode=conv1.padding_mode,
        device=conv1.weight.device,
        dtype=conv1.weight.dtype,
    )
    with torch.no_grad():
        new_conv.weight[:, :MONO_FEATURE_BINS, :] = conv1.weight * 0.5
        new_conv.weight[:, MONO_FEATURE_BINS:, :] = conv1.weight * 0.5
        if conv1.bias is not None and new_conv.bias is not None:
            new_conv.bias.copy_(conv1.bias)

    candidates = [
        getattr(model, "encoder", None),
        getattr(getattr(model, "model", None), "encoder", None),
        getattr(getattr(model, "whisper", None), "encoder", None),
    ]
    for encoder in candidates:
        if encoder is not None and getattr(encoder, "conv1", None) is conv1:
            encoder.conv1 = new_conv
            break
    setattr(model.config, "num_mel_bins", STEREO_FEATURE_BINS)

def shift_tokens_right(input_ids: torch.Tensor, pad_token_id: int, decoder_start_token_id: int):
    """
    Shift input ids one token to the right.
    """
    shifted_input_ids = input_ids.new_zeros(input_ids.shape)
    shifted_input_ids[:, 1:] = input_ids[:, :-1].clone()
    shifted_input_ids[:, 0] = decoder_start_token_id

    if pad_token_id is None:
        raise ValueError("self.model.config.pad_token_id has to be defined.")
    # replace possible -100 values in labels by `pad_token_id`
    shifted_input_ids.masked_fill_(shifted_input_ids == -100, pad_token_id)

    return shifted_input_ids

        if labels is not None:
            if labels.shape[1] > self.max_target_positions:
                raise ValueError(
                    f"Labels' sequence length {labels.shape[1]} cannot exceed the maximum allowed length of {self.max_target_positions} tokens."
                )
            if decoder_input_ids is None and decoder_inputs_embeds is None:
                decoder_input_ids = shift_tokens_right(
                    labels, self.config.pad_token_id, self.config.decoder_start_token_id
                )

        outputs: Seq2SeqModelOutput = self.model(
            input_features,
            attention_mask=attention_mask,
            decoder_input_ids=decoder_input_ids,
            encoder_outputs=encoder_outputs,
            decoder_attention_mask=decoder_attention_mask,
            past_key_values=past_key_values,
            decoder_inputs_embeds=decoder_inputs_embeds,
            decoder_position_ids=decoder_position_ids,
            use_cache=use_cache,
            **kwargs,
        )
        lm_logits = self.proj_out(outputs.last_hidden_state)

        loss = None
        if labels is not None:
            loss_fct = CrossEntropyLoss()
            # move labels to correct device to enable PP
            labels = labels.to(lm_logits.device)
            loss = loss_fct(lm_logits.view(-1, self.config.vocab_size), labels.reshape(-1))

    def on_step_end(self, args, state, control, **kwargs):
        step = int(getattr(state, "global_step", 0))
        if not self._should_check(step):
            return control

        ram = get_ram_usage()
        if ram is not None and float(ram["ratio"]) >= self.ram_threshold:
            if (step - self._last_ram_warn_step) >= self.repeat_warning_steps:
                self._last_ram_warn_step = step
                print(
                    "[MemWarn][RAM] "
                    f"step={step}, usage={float(ram['ratio']) * 100:.1f}%, "
                    f"used={_format_bytes_gb(int(ram['used']))}, "
                    f"total={_format_bytes_gb(int(ram['total']))}"
                )

        for item in get_vram_usage():
            device = int(item["device"])
            ratio = float(item["ratio"])
            if ratio < self.vram_threshold:
                continue
            last_step = self._last_vram_warn_step.get(device, -10**9)
            if (step - last_step) < self.repeat_warning_steps:
                continue
            self._last_vram_warn_step[device] = step
            print(
                "[MemWarn][VRAM] "
                f"step={step}, cuda:{device}, usage={ratio * 100:.1f}%, "
                f"used={_format_bytes_gb(int(item['used']))}, "
                f"total={_format_bytes_gb(int(item['total']))}"
            )
        return control

    def on_step_end(self, args, state, control, **kwargs):
        if self._pause_handled or not self.pause_flag_file.exists():
            return control

        print("[PAUSE] detected -> saving checkpoint then stopping.")
        try:
            if self._trainer is None:
                raise RuntimeError("Pause callback has no trainer reference.")
            if self._trainer.args.should_save:
                self._save_pause_checkpoint()
                print(f"[PAUSE] checkpoint saved: {self.pause_ckpt_dir}")
                if self.remove_pause_flag and self.pause_flag_file.exists():
                    self.pause_flag_file.unlink()
                    print(f"[PAUSE] flag removed: {self.pause_flag_file}")
            else:
                print("[PAUSE] non-saving process received pause signal, stopping.")
        except Exception as exc:
            print(f"[PAUSE] failed to save pause checkpoint: {exc}")

        self._pause_handled = True
        control.should_training_stop = True
        return control

def resolve_resume_checkpoint(
    args: argparse.Namespace,
    pause_ckpt_dir: Path,
) -> Tuple[Path | None, str]:
    if str(args.resume_from_checkpoint).strip():
        resume_path = Path(str(args.resume_from_checkpoint)).expanduser().resolve()
        if not resume_path.exists():
            raise FileNotFoundError(f"resume_from_checkpoint not found: {resume_path}")
        if not is_stereo_checkpoint(resume_path):
            raise ValueError(
                f"Explicit resume checkpoint is not stereo-compatible: {resume_path} "
                f"(num_mel_bins={checkpoint_input_feature_bins(resume_path)}, "
                f"expected={STEREO_FEATURE_BINS})"
            )
        return resume_path, "explicit"

    if not bool(args.auto_resume):
        return None, "disabled"

    if is_trainer_checkpoint(pause_ckpt_dir) and is_stereo_checkpoint(pause_ckpt_dir):
        return pause_ckpt_dir, "pause_ckpt"
    if is_trainer_checkpoint(pause_ckpt_dir):
        print(
            f"[Resume] skipping mono/incompatible pause checkpoint: {pause_ckpt_dir} "
            f"(num_mel_bins={checkpoint_input_feature_bins(pause_ckpt_dir)})"
        )

    latest_ckpt = find_latest_trainer_checkpoint(Path(str(args.output_dir)))
    if latest_ckpt is not None and is_stereo_checkpoint(latest_ckpt):
        return latest_ckpt, "latest_checkpoint"
    if latest_ckpt is not None:
        print(
            f"[Resume] skipping mono/incompatible latest checkpoint: {latest_ckpt} "
            f"(num_mel_bins={checkpoint_input_feature_bins(latest_ckpt)})"
        )
    return None, "none"

    kwargs: Dict[str, Any] = {
        "output_dir": str(Path(args.output_dir).expanduser().resolve()),
        "logging_dir": str(Path(args.log_dir).expanduser().resolve()),
        "per_device_train_batch_size": int(args.per_device_train_batch_size),
        "per_device_eval_batch_size": int(args.per_device_eval_batch_size),
        "gradient_accumulation_steps": int(args.gradient_accumulation_steps),
        "learning_rate": float(args.learning_rate),
        "num_train_epochs": float(args.num_train_epochs),
        "max_steps": int(args.max_steps),
        "fp16": bool(args.fp16 and torch.cuda.is_available()),
        "save_strategy": strategy,
        "logging_strategy": "steps",
        "logging_steps": 10,
        "save_total_limit": max(1, int(args.save_total_limit)),
        "report_to": [],
        "run_name": sanitize_experiment_name(args.experiment_name),
        "remove_unused_columns": False,
        "seed": int(args.seed),
        "gradient_checkpointing": bool(args.gradient_checkpointing),
        "dataloader_num_workers": int(args.num_workers),
        "dataloader_pin_memory": True,
        "eval_accumulation_steps": max(1, int(args.eval_accumulation_steps)),
    }
    if int(args.max_steps) <= 0:
        kwargs.pop("max_steps")

    params = inspect.signature(TrainingArguments.__init__).parameters

    model = build_model(args, tokenizer)
    trainer = Trainer(
        model=model,
        args=build_training_args(args, has_eval=eval_dataset is not None),
        train_dataset=train_dataset,
        eval_dataset=eval_dataset,
        data_collator=Seq2SeqCollator(pad_token_id=tokenizer.pad_token_id),
        compute_metrics=compute_token_metrics if eval_dataset is not None else None,
        preprocess_logits_for_metrics=preprocess_logits_for_metrics
        if eval_dataset is not None
        else None,
    )

    resume_ckpt, resume_source = resolve_resume_checkpoint(args, pause_ckpt_dir)
    if resume_ckpt is not None:
        print(f"[Resume] source={resume_source}, resume_from_checkpoint={resume_ckpt}")
        trainer.train(resume_from_checkpoint=str(resume_ckpt))
    else:
        print(f"[Resume] source={resume_source}, starting from scratch")
        trainer.train()

def preprocess_logits_for_metrics(logits, labels):
    if isinstance(logits, tuple):
        logits = logits[0]
    return torch.argmax(logits, dim=-1)

def compute_token_metrics(eval_pred) -> Dict[str, float]:
    preds = np.asarray(eval_pred.predictions)
    labels = np.asarray(eval_pred.label_ids)
    if preds.shape != labels.shape:
        preds = preds.reshape(labels.shape)
    mask = labels != -100
    if not np.any(mask):
        return {"token_acc": 0.0, "seq_exact": 0.0}
    token_acc = float(np.mean(preds[mask] == labels[mask]))
    seq_ok = []
    for pred_row, label_row, mask_row in zip(preds, labels, mask):
        seq_ok.append(bool(np.all(pred_row[mask_row] == label_row[mask_row])))
    return {
        "token_acc": token_acc,
        "seq_exact": float(np.mean(seq_ok)) if seq_ok else 0.0,
    }

        kwargs["load_best_model_at_end"] = True
        kwargs["metric_for_best_model"] = "eval_token_acc"
        kwargs["greater_is_better"] = True

    if eval_dataset is not None:
        metrics = trainer.evaluate()
        print(
            f"[Eval] eval_token_acc={metrics.get('eval_token_acc', 0.0):.6f}, "
            f"eval_seq_exact={metrics.get('eval_seq_exact', 0.0):.6f}"
        )
    else:
        metrics = {}

    export_dir.mkdir(parents=True, exist_ok=True)
    trainer.save_model(str(export_dir))
    feature_extractor.save_pretrained(str(export_dir))
    shutil.copy2(vocab_path, export_dir / "dsp_control_vocab.json")

    summary = {
        "objective": "audio_to_dsp_control_program_tokens",
        "desensitized_from_preset_number": True,
        "uses_preset_number_as_label": False,
        "target_source_field": "target_program_json.model_token_ids",
        "manifest": str(manifest_path),
        "vocab": str(vocab_path),
        "program_root": str(program_root),
        "vocab_size": int(tokenizer.vocab_size),
        "audio_root": str(audio_root),
        "total_rows": int(len(rows)),
        "train_rows": int(len(train_rows)),
        "eval_rows": int(len(eval_rows)),
        "unique_program_hashes": int(program_hash_count),
        "train_program_hashes": int(len({r["program_hash"] for r in train_rows})),
        "eval_program_hashes": int(len({r["program_hash"] for r in eval_rows})),
        "target_length": {
            "min": int(min(lengths)),
            "max": int(max(lengths)),
            "mean": float(np.mean(lengths)),
        },
        "metrics": metrics,
        "training_config": {
            "model_name": str(args.model_name),
            "audio_channel_mode": AUDIO_CHANNEL_MODE,
            "input_feature_bins": int(STEREO_FEATURE_BINS),
            "max_seconds": float(args.max_seconds),
            "max_target_length": int(args.max_target_length),
            "strip_bos_from_labels": bool(args.strip_bos_from_labels),
            "eval_ratio": float(args.eval_ratio),
            "num_train_epochs": float(args.num_train_epochs),
            "max_steps": int(args.max_steps),
            "checkpoint_strategy": str(args.checkpoint_strategy),
            "save_steps": int(args.save_steps),
            "eval_steps": int(args.eval_steps),
            "save_total_limit": int(args.save_total_limit),
            "per_device_train_batch_size": int(args.per_device_train_batch_size),
            "per_device_eval_batch_size": int(args.per_device_eval_batch_size),
            "gradient_accumulation_steps": int(args.gradient_accumulation_steps),
            "learning_rate": float(args.learning_rate),
            "gradient_checkpointing": bool(args.gradient_checkpointing),
            "fp16": bool(args.fp16 and torch.cuda.is_available()),
            "gpu_mem_frac": float(args.gpu_mem_frac),
            "auto_resume": bool(args.auto_resume),
            "resume_from_checkpoint": str(resume_ckpt) if resume_ckpt is not None else "",
            "resume_source": str(resume_source),
        },
        "feature_cache_base": str(cache_base),
        "e_root": str(e_root),
        "output_dir": str(output_dir),
        "log_dir": str(log_dir),
        "export_dir": str(export_dir),
        "pause_flag_file": str(pause_flag_file),
        "pause_ckpt_dir": str(pause_ckpt_dir),
        "env": env_effective,
    }

    (export_dir / "run_summary.json").write_text(
        json.dumps(summary, indent=2, ensure_ascii=False),
        encoding="utf-8",
    )
    print(f"Parameter sequence model exported to: {export_dir}")