diff --git a/.dockerignore b/.dockerignore index 0dfe444b..2c1ae741 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,5 +1,7 @@ .venv .git +**/__pycache__ +**/*.pyc /checkpoints /datasets /output diff --git a/Dockerfile b/Dockerfile index 529ddb2c..4b901681 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,6 +7,23 @@ ARG CUDA_VERSION=13.0.2 ARG BASE_IMAGE=nvidia/cuda:${CUDA_VERSION}-cudnn-devel-ubuntu24.04 FROM ${BASE_IMAGE} +ARG SOURCE_COMMIT +ARG SOURCE_TREE +ARG SOURCE_DIRTY=1 +ARG BUILD_TIMESTAMP +ARG BASE_IMAGE +ARG CUDA_VERSION +LABEL org.opencontainers.image.revision="${SOURCE_COMMIT}" \ + org.opencontainers.image.created="${BUILD_TIMESTAMP}" \ + com.nvidia.tao.source-tree="${SOURCE_TREE}" \ + com.nvidia.tao.backend="cosmos-framework" +ENV SOURCE_COMMIT="${SOURCE_COMMIT}" \ + SOURCE_TREE="${SOURCE_TREE}" \ + SOURCE_DIRTY="${SOURCE_DIRTY}" \ + BUILD_TIMESTAMP="${BUILD_TIMESTAMP}" \ + PROVENANCE_BASE_IMAGE="${BASE_IMAGE}" \ + CUDA_VERSION="${CUDA_VERSION}" + # Set the DEBIAN_FRONTEND environment variable to avoid interactive prompts during apt operations. ENV DEBIAN_FRONTEND=noninteractive @@ -28,7 +45,8 @@ COPY --from=ghcr.io/astral-sh/uv:0.11.28 /uv /uvx /usr/local/bin/ # Copy from the cache instead of linking since it's a mounted volume ENV UV_LINK_MODE=copy # Cache python downloads -ENV UV_PYTHON_CACHE_DIR=/root/.cache/uv/python +ENV UV_PYTHON_CACHE_DIR=/opt/uv-python-cache \ + UV_PYTHON_INSTALL_DIR=/opt/uv-python # Install just: https://just.systems/man/en/pre-built-binaries.html RUN curl --proto '=https' --tlsv1.2 -sSf https://just.systems/install.sh | bash -s -- --to /usr/local/bin --tag 1.46.0 @@ -40,7 +58,8 @@ WORKDIR /workspace # Install python RUN --mount=type=cache,target=/root/.cache/uv \ --mount=type=bind,source=.python-version,target=.python-version \ - uv python install + uv python install && \ + chmod -R a+rX /opt/uv-python /opt/uv-python-cache # Install into virtual environment RUN echo "$CUDA_VERSION" | sed -E 's/^([0-9]+)\.([0-9]+).*/cu\1\2/' > /root/.cuda-name @@ -49,9 +68,21 @@ RUN --mount=type=cache,target=/root/.cache/uv \ --mount=type=bind,source=pyproject.toml,target=pyproject.toml \ --mount=type=bind,source=.python-version,target=.python-version \ --mount=type=bind,source=packages,target=packages \ - uv sync --locked --no-install-project --no-editable --all-extras --group=$(cat /root/.cuda-name) --group=vllm + uv sync --locked --no-install-project --no-editable --all-extras --group=$(cat /root/.cuda-name)-train ENV PATH="/workspace/.venv/bin:$PATH" +# Package the exact source state into the image so it can run on managed +# platforms without a host bind mount. +COPY . /workspace +RUN --mount=type=cache,target=/root/.cache/uv \ + uv pip install --no-deps . + +RUN /workspace/.venv/bin/python /workspace/docker/write_image_provenance.py && \ + chmod a+rx /workspace /workspace/docker /workspace/docker/entrypoint.sh && \ + chmod -R a+rX /opt/tao /workspace/.venv /workspace/cosmos_framework && \ + test -x /workspace/docker/entrypoint.sh && \ + test -x /workspace/.venv/bin/python + # Triton bundled ptxas doesn't support latest GPU architectures ENV TRITON_PTXAS_PATH="/usr/local/cuda/bin/ptxas" diff --git a/cosmos_framework/callbacks/tao_status.py b/cosmos_framework/callbacks/tao_status.py new file mode 100644 index 00000000..2013a363 --- /dev/null +++ b/cosmos_framework/callbacks/tao_status.py @@ -0,0 +1,541 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +"""TAO-compatible lifecycle and metric logging for Cosmos Framework training.""" + +from __future__ import annotations + +import json +import os +import time +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any + +import torch +import torch.distributed as dist + +from cosmos_framework.utils import distributed, log +from cosmos_framework.utils.callback import Callback + + +def _to_json_value(value: Any) -> Any: + """Recursively convert tensors and array-like values to JSON-safe values.""" + if isinstance(value, dict): + return {str(key): _to_json_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_to_json_value(item) for item in value] + if isinstance(value, torch.Tensor): + value = value.detach() + if value.numel() == 1: + return value.item() + return value.cpu().tolist() + if hasattr(value, "item"): + try: + return value.item() + except (TypeError, ValueError): + pass + if hasattr(value, "tolist"): + try: + return value.tolist() + except (TypeError, ValueError): + pass + return value + + +class _TAOStatusWriter: + """Use TAO Core when available, with a compatible JSON-lines fallback.""" + + def __init__(self, filename: str) -> None: + self.filename = filename + Path(filename).parent.mkdir(parents=True, exist_ok=True) + self._tao_logger = None + self._tao_status = None + self._tao_verbosity = None + + try: + from nvidia_tao_core.loggers.logging import Status, StatusLogger, Verbosity + except ImportError: + log.warning(f"nvidia_tao_core is not installed; writing TAO-compatible JSON records directly to {filename}") + else: + self._tao_status = Status + self._tao_verbosity = Verbosity + self._tao_logger = StatusLogger( + filename=filename, + is_master=True, + verbosity=Verbosity.INFO, + append=True, + ) + + def write( + self, + *, + status: str, + message: str, + data: dict[str, Any] | None = None, + kpi: dict[str, Any] | None = None, + verbosity: str = "INFO", + ) -> None: + data = _to_json_value(data or {}) + kpi = _to_json_value(kpi or {}) + + if self._tao_logger is not None: + status_level = getattr(self._tao_status, status) + verbosity_level = getattr(self._tao_verbosity, verbosity) + self._tao_logger.kpi = kpi + self._tao_logger.write( + data=data, + status_level=status_level, + verbosity_level=verbosity_level, + message=message, + ) + return + + now = datetime.now() + payload: dict[str, Any] = { + **data, + "date": f"{now.month}/{now.day}/{now.year}", + "time": f"{now.hour}:{now.minute}:{now.second}", + "status": status, + "verbosity": verbosity, + "message": message, + } + if kpi: + payload["kpi"] = kpi + with open(self.filename, "a", encoding="utf-8") as status_file: + status_file.write(json.dumps(payload, default=str) + "\n") + + +def write_early_failure(error: BaseException) -> bool: + """Write a terminal TAO record before the callback/config exists. + + The orchestration layer supplies ``TAO_STATUS_FILE`` for direct launches, + or the normal TAO job/result variables. No implicit host path is used. + """ + status_path = os.environ.get("TAO_STATUS_FILE") + if not status_path: + job_id = os.environ.get("TAO_JOB_ID") + results_root = os.environ.get("TAO_RESULTS_ROOT") + if job_id and results_root: + status_path = os.path.join(results_root, job_id, "status.json") + if not status_path: + api_job_id = os.environ.get("TAO_API_JOB_ID") + results_root = os.environ.get("TAO_API_RESULTS_DIR") + if api_job_id and results_root: + status_path = os.path.join(results_root, api_job_id, "status.json") + if not status_path: + return False + _TAOStatusWriter(status_path).write( + status="FAILURE", + verbosity="ERROR", + message=f"Cosmos Framework training failed before callback initialization: {error}", + data={"phase": "preflight_or_initialization", "error_type": type(error).__name__}, + ) + return True + + +class TAOStatusCallback(Callback): + """Write TAO lifecycle, training, and validation records from rank zero. + + The output path is resolved in this order: + + 1. ``status_file_path`` when explicitly configured. + 2. ``$TAO_RESULTS_ROOT/$TAO_JOB_ID/status.json`` (TAO SDK). + 3. ``$TAO_API_RESULTS_DIR/$TAO_API_JOB_ID/status.json`` (TAO API). + 4. ``/status.json`` for direct launches. + """ + + def __init__( + self, + enabled: bool = False, + status_file_path: str | None = None, + experiment_name: str = "", + logging_interval: int = 1, + validation_heartbeat_interval: int = 1, + ) -> None: + if logging_interval < 1: + raise ValueError("logging_interval must be >= 1") + if validation_heartbeat_interval < 1: + raise ValueError("validation_heartbeat_interval must be >= 1") + self.enabled = enabled + self.status_file_path = status_file_path + self.experiment_name = experiment_name + self.logging_interval = logging_interval + self.validation_heartbeat_interval = validation_heartbeat_interval + self._writer: _TAOStatusWriter | None = None + self._train_start_time = 0.0 + self._step_start_time = 0.0 + self._validation_batches = 0 + self._validation_loss_numerator = 0.0 + self._validation_loss_denominator = 0 + self._train_loss_numerator = 0.0 + self._train_loss_denominator = 0 + self.last_training_loss: float | None = None + self.last_validation_loss: float | None = None + + def _is_rank_zero(self) -> bool: + if dist.is_available() and dist.is_initialized(): + return distributed.is_rank0() + return int(os.environ.get("RANK", os.environ.get("LOCAL_RANK", "0"))) == 0 + + def _component_name(self) -> str: + if self.experiment_name: + return self.experiment_name + return getattr(self.config.job, "name", "Cosmos Framework SFT") + + def _resolve_status_file(self) -> str: + if self.status_file_path: + return self.status_file_path + + job_id = os.environ.get("TAO_JOB_ID") + if job_id: + results_root = os.environ.get("TAO_RESULTS_ROOT") + if not results_root: + raise RuntimeError("TAO_RESULTS_ROOT is required when TAO_JOB_ID is set") + return os.path.join(results_root, job_id, "status.json") + + api_job_id = os.environ.get("TAO_API_JOB_ID") + if api_job_id: + results_root = os.environ.get("TAO_API_RESULTS_DIR") + if not results_root: + raise RuntimeError("TAO_API_RESULTS_DIR is required when TAO_API_JOB_ID is set") + return os.path.join(results_root, api_job_id, "status.json") + + return os.path.join(self.config.job.path_local, "status.json") + + def _get_writer(self) -> _TAOStatusWriter | None: + if not self.enabled or not self._is_rank_zero(): + return None + if self._writer is None: + self._writer = _TAOStatusWriter(self._resolve_status_file()) + return self._writer + + def _progress_data(self, iteration: int, seconds_per_step: float | None = None) -> dict[str, Any]: + trainer = getattr(self, "trainer", None) + max_step = int(getattr(trainer, "max_iterations", self.config.trainer.max_iter)) + if seconds_per_step is None: + elapsed = max(time.monotonic() - self._train_start_time, 0.0) + seconds_per_step = elapsed / max(iteration, 1) + eta_seconds = max(max_step - iteration, 0) * seconds_per_step + data = { + "component": self._component_name(), + "step": iteration, + "max_step": max_step, + "time_per_step": str(timedelta(seconds=seconds_per_step)), + "eta": str(timedelta(seconds=eta_seconds)), + } + steps_per_epoch = getattr(trainer, "steps_per_epoch", None) + num_epochs = getattr(trainer, "num_epochs", None) + if steps_per_epoch and num_epochs: + completed_epochs = iteration // steps_per_epoch + if iteration == 0: + epoch = 1 + step_in_epoch = 0 + elif iteration % steps_per_epoch == 0: + epoch = min(completed_epochs, num_epochs) + step_in_epoch = steps_per_epoch + else: + epoch = min(completed_epochs + 1, num_epochs) + step_in_epoch = iteration % steps_per_epoch + data.update( + { + "epoch": epoch, + "max_epoch": num_epochs, + "completed_epochs": min(completed_epochs, num_epochs), + "step_in_epoch": step_in_epoch, + "steps_per_epoch": steps_per_epoch, + "time_per_epoch": str(timedelta(seconds=seconds_per_step * steps_per_epoch)), + } + ) + return data + + def _epoch_label(self, iteration: int) -> str: + progress = self._progress_data(iteration, seconds_per_step=0.0) + if "epoch" not in progress: + return f"training step {iteration}/{progress['max_step']}" + return f"epoch {progress['epoch']}/{progress['max_epoch']}" + + @staticmethod + def _local_token_stats( + loss: torch.Tensor, + data_batch: dict[str, Any], + output_batch: dict[str, Any], + ) -> tuple[torch.Tensor, torch.Tensor]: + numerator = output_batch.get("loss_numerator") + denominator = output_batch.get("loss_denominator") + if numerator is not None and denominator is not None: + return numerator.detach(), denominator.detach().to(dtype=torch.long) + + # Backward-compatible fallback for non-VLM models. It is deliberately + # sample-weighted and is never used by the Cosmos3 VLM path, which + # always emits exact token statistics. + sample_count = next( + ( + int(value.shape[0]) + for value in data_batch.values() + if isinstance(value, torch.Tensor) and value.ndim > 0 + ), + 1, + ) + return ( + loss.detach() * sample_count, + torch.tensor(sample_count, device=loss.device, dtype=torch.long), + ) + + @classmethod + def _global_token_average( + cls, + loss: torch.Tensor, + data_batch: dict[str, Any], + output_batch: dict[str, Any], + ) -> tuple[float, float, int]: + numerator, denominator = cls._local_token_stats(loss, data_batch, output_batch) + numerator = numerator.clone() + denominator = denominator.clone() + if dist.is_available() and dist.is_initialized(): + dist.all_reduce(numerator, op=dist.ReduceOp.SUM) + dist.all_reduce(denominator, op=dist.ReduceOp.SUM) + global_denominator = int(denominator.item()) + global_numerator = float(numerator.item()) + return global_numerator / max(global_denominator, 1), global_numerator, global_denominator + + @staticmethod + def _reduce_accumulator(numerator: float, denominator: int) -> tuple[float, int]: + device = "cuda" if dist.is_available() and dist.is_initialized() else "cpu" + values = torch.tensor([numerator, float(denominator)], dtype=torch.float64, device=device) + if dist.is_available() and dist.is_initialized(): + dist.all_reduce(values, op=dist.ReduceOp.SUM) + return float(values[0].item()), int(values[1].item()) + + def on_train_start(self, model: Any, iteration: int = 0) -> None: + self._train_start_time = time.monotonic() + self._train_loss_numerator = 0.0 + self._train_loss_denominator = 0 + writer = self._get_writer() + if writer is not None: + writer.write( + status="STARTED", + message=f"Starting {self._component_name()} training", + data={ + **self._progress_data(iteration, seconds_per_step=0.0), + "parameter_summary": getattr(model, "parameter_summary", None), + }, + ) + log.info(f"TAO status will be logged to {writer.filename}") + + def on_training_step_start(self, model: Any, data: dict[str, Any], iteration: int = 0) -> None: + self._step_start_time = time.monotonic() + + def on_training_step_batch_end( + self, + model: Any, + data_batch: dict[str, Any], + output_batch: dict[str, Any], + loss: torch.Tensor, + iteration: int = 0, + ) -> None: + numerator, denominator = self._local_token_stats(loss, data_batch, output_batch) + self._train_loss_numerator += float(numerator.item()) + self._train_loss_denominator += int(denominator.item()) + + def on_training_step_end( + self, + model: Any, + data_batch: dict[str, Any], + output_batch: dict[str, Any], + loss: torch.Tensor, + iteration: int = 0, + ) -> None: + interval = int(self.config.trainer.logging_iter) * self.logging_interval + if iteration % interval != 0: + return + + average_loss, numerator, denominator = self._global_token_average(loss, data_batch, output_batch) + writer = self._get_writer() + if writer is None: + return + seconds_per_step = max(time.monotonic() - self._step_start_time, 0.0) + kpi = { + "train/step_loss": average_loss, + "train/step_loss_numerator": numerator, + "train/step_loss_denominator": denominator, + } + progress = self._progress_data(iteration, seconds_per_step=seconds_per_step) + if "epoch" in progress: + message = ( + f"Training epoch {progress['epoch']}/{progress['max_epoch']}, " + f"step {progress['step_in_epoch']}/{progress['steps_per_epoch']} " + f"(global step {iteration}/{progress['max_step']}) - Loss: {average_loss:.6f}" + ) + else: + message = f"Training step {iteration}/{progress['max_step']} - Loss: {average_loss:.6f}" + writer.write( + status="RUNNING", + message=message, + data=progress, + kpi=kpi, + ) + + def on_validation_start(self, model: Any, dataloader_val: Any, iteration: int = 0) -> None: + self._validation_batches = 0 + self._validation_loss_numerator = 0.0 + self._validation_loss_denominator = 0 + writer = self._get_writer() + if writer is not None: + writer.write( + status="RUNNING", + message=f"Starting validation for {self._epoch_label(iteration)}", + data={**self._progress_data(iteration), "phase": "validation_starting"}, + ) + log.info(f"Starting validation for {self._epoch_label(iteration)}") + + def on_validation_step_end( + self, + model: Any, + data_batch: dict[str, Any], + output_batch: dict[str, Any], + loss: torch.Tensor, + iteration: int = 0, + ) -> None: + local_numerator, local_denominator = self._local_token_stats(loss, data_batch, output_batch) + average_loss, _, _ = self._global_token_average(loss, data_batch, output_batch) + self._validation_batches += 1 + self._validation_loss_numerator += float(local_numerator.item()) + self._validation_loss_denominator += int(local_denominator.item()) + + if self._validation_batches % self.validation_heartbeat_interval != 0: + return + writer = self._get_writer() + max_validation_batches = getattr(self.config.trainer, "max_val_iter", None) + batch_progress = ( + f"{self._validation_batches}/{max_validation_batches}" + if max_validation_batches is not None + else str(self._validation_batches) + ) + if writer is not None: + writer.write( + status="RUNNING", + message=( + f"Validation {self._epoch_label(iteration)}, batch {batch_progress} - Loss: {average_loss:.6f}" + ), + data={ + **self._progress_data(iteration), + "phase": "validation_batch_complete", + "validation_batch": self._validation_batches, + "max_validation_batches": max_validation_batches, + }, + kpi={"val/batch_loss": average_loss}, + ) + log.info(f"Validation {self._epoch_label(iteration)}, batch {batch_progress} - Loss: {average_loss:.6f}") + + def on_validation_end(self, model: Any, iteration: int = 0) -> None: + numerator, denominator = self._reduce_accumulator( + self._validation_loss_numerator, self._validation_loss_denominator + ) + if denominator == 0: + log.warning("TAO validation logging saw zero samples; no val/loss record was written") + return + + self.last_validation_loss = numerator / denominator + writer = self._get_writer() + if writer is not None: + writer.write( + status="RUNNING", + message=( + f"Validation complete for {self._epoch_label(iteration)} - Loss: {self.last_validation_loss:.6f}" + ), + data={ + **self._progress_data(iteration), + "phase": "validation_complete", + "validation_batches": self._validation_batches, + "validation_loss_numerator": numerator, + "validation_valid_label_count": denominator, + }, + kpi={ + "val/loss": self.last_validation_loss, + "val/avg_loss": self.last_validation_loss, + "val/loss_numerator": numerator, + "val/valid_label_count": denominator, + }, + ) + log.info(f"Validation loss ({self._epoch_label(iteration)}): {self.last_validation_loss:.6f}") + + def on_save_checkpoint_success(self, iteration: int = 0, elapsed_time: float = 0) -> None: + checkpoint_path = None + checkpoint_root = Path(self.config.job.path_local) / "checkpoints" + latest = checkpoint_root / "latest_checkpoint.txt" + if latest.is_file(): + checkpoint_name = latest.read_text(encoding="utf-8").strip() + if checkpoint_name: + checkpoint_path = str((checkpoint_root / checkpoint_name).resolve()) + writer = self._get_writer() + if writer is not None: + writer.write( + status="RUNNING", + message=f"Checkpoint saved successfully at step {iteration}", + data={ + **self._progress_data(iteration), + "phase": "checkpoint_complete", + "checkpoint_iteration": iteration, + "checkpoint_elapsed_seconds": elapsed_time, + "checkpoint_path": checkpoint_path, + }, + ) + + def on_train_end(self, model: Any, iteration: int = 0) -> None: + numerator, denominator = self._reduce_accumulator( + self._train_loss_numerator, self._train_loss_denominator + ) + if denominator == 0: + raise RuntimeError("TAO metric collection observed zero valid training labels") + self.last_training_loss = numerator / denominator + writer = self._get_writer() + if writer is not None: + writer.write( + status="RUNNING", + message=f"Training complete - token-weighted loss: {self.last_training_loss:.6f}", + data={ + **self._progress_data(iteration), + "phase": "training_complete", + "train_loss_numerator": numerator, + "train_valid_label_count": denominator, + }, + kpi={ + "train/avg_loss": self.last_training_loss, + "train/loss_numerator": numerator, + "train/valid_label_count": denominator, + }, + ) + + def on_app_end(self) -> None: + writer = self._get_writer() + if writer is not None: + writer.write( + status="SUCCESS", + message=f"{self._component_name()} training completed successfully", + data=self._progress_data( + int(getattr(getattr(self, "trainer", None), "max_iterations", self.config.trainer.max_iter)) + ), + kpi={ + **( + {"train/avg_loss": self.last_training_loss} + if self.last_training_loss is not None + else {} + ), + **( + {"val/loss": self.last_validation_loss, "val/avg_loss": self.last_validation_loss} + if self.last_validation_loss is not None + else {} + ), + }, + ) + + def on_exception(self, error: BaseException) -> None: + writer = self._get_writer() + if writer is not None: + writer.write( + status="FAILURE", + verbosity="ERROR", + message=f"{self._component_name()} training failed: {error}", + data={"component": self._component_name(), "error_type": type(error).__name__}, + ) diff --git a/cosmos_framework/callbacks/tao_status_test.py b/cosmos_framework/callbacks/tao_status_test.py new file mode 100644 index 00000000..9c6e87a7 --- /dev/null +++ b/cosmos_framework/callbacks/tao_status_test.py @@ -0,0 +1,115 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import torch + +from cosmos_framework.callbacks.tao_status import TAOStatusCallback + + +def _callback(tmp_path) -> TAOStatusCallback: + callback = TAOStatusCallback( + enabled=True, + status_file_path=str(tmp_path / "status.json"), + experiment_name="test", + ) + callback.config = SimpleNamespace( + job=SimpleNamespace(name="job", path_local=str(tmp_path)), + trainer=SimpleNamespace(max_iter=6, max_val_iter=2, logging_iter=1), + ) + callback.trainer = SimpleNamespace(max_iterations=6, num_epochs=2, steps_per_epoch=3) + return callback + + +def _records(tmp_path) -> list[dict]: + return [json.loads(line) for line in (tmp_path / "status.json").read_text(encoding="utf-8").splitlines()] + + +def test_tao_status_callback_writes_training_validation_and_success(tmp_path) -> None: + callback = _callback(tmp_path) + callback.on_train_start(model=None, iteration=0) + callback.on_training_step_start(model=None, data={}, iteration=0) + callback.on_training_step_batch_end( + model=None, + data_batch={"input_ids": torch.zeros(2, 3)}, + output_batch={"loss_numerator": torch.tensor(6.0), "loss_denominator": torch.tensor(12)}, + loss=torch.tensor(0.5), + iteration=3, + ) + callback.on_training_step_end( + model=None, + data_batch={"input_ids": torch.zeros(2, 3)}, + output_batch={"loss_numerator": torch.tensor(6.0), "loss_denominator": torch.tensor(12)}, + loss=torch.tensor(0.5), + iteration=3, + ) + callback.on_validation_start(model=None, dataloader_val=None, iteration=3) + callback.on_validation_step_end( + model=None, + data_batch={"input_ids": torch.zeros(2, 3)}, + output_batch={"loss_numerator": torch.tensor(2.0), "loss_denominator": torch.tensor(8)}, + loss=torch.tensor(0.25), + iteration=3, + ) + callback.on_validation_end(model=None, iteration=3) + callback.on_train_end(model=None, iteration=6) + callback.on_app_end() + + records = _records(tmp_path) + assert [record["status"] for record in records] == [ + "STARTED", + "RUNNING", + "RUNNING", + "RUNNING", + "RUNNING", + "RUNNING", + "SUCCESS", + ] + assert records[1]["kpi"]["train/step_loss"] == 0.5 + assert records[1]["epoch"] == 1 + assert records[1]["max_epoch"] == 2 + assert records[1]["step_in_epoch"] == 3 + assert records[1]["steps_per_epoch"] == 3 + assert records[1]["max_step"] == 6 + assert records[2]["phase"] == "validation_starting" + assert records[3]["max_validation_batches"] == 2 + assert "epoch 1/2" in records[3]["message"] + assert records[4]["kpi"]["val/avg_loss"] == 0.25 + assert records[4]["kpi"]["val/loss_numerator"] == 2.0 + assert records[4]["kpi"]["val/valid_label_count"] == 8 + assert "epoch 1/2" in records[4]["message"] + assert records[5]["phase"] == "training_complete" + assert records[5]["kpi"]["train/avg_loss"] == 0.5 + assert records[5]["train_loss_numerator"] == 6.0 + assert records[5]["train_valid_label_count"] == 12 + assert records[5]["kpi"]["train/valid_label_count"] == 12 + assert records[-1]["kpi"]["val/loss"] == 0.25 + assert records[-1]["epoch"] == 2 + assert records[-1]["completed_epochs"] == 2 + + +def test_tao_status_callback_reports_checkpoint_event(tmp_path) -> None: + callback = _callback(tmp_path) + checkpoint = tmp_path / "checkpoints" / "epoch_1" + checkpoint.mkdir(parents=True) + (tmp_path / "checkpoints" / "latest_checkpoint.txt").write_text("epoch_1\n") + callback.on_save_checkpoint_success(iteration=3, elapsed_time=1.25) + record = _records(tmp_path)[-1] + assert record["phase"] == "checkpoint_complete" + assert record["checkpoint_iteration"] == 3 + assert record["checkpoint_elapsed_seconds"] == 1.25 + assert record["checkpoint_path"] == str(checkpoint.resolve()) + + +def test_tao_status_callback_writes_failure(tmp_path) -> None: + callback = _callback(tmp_path) + callback.on_train_start(model=None, iteration=0) + callback.on_exception(RuntimeError("boom")) + + records = _records(tmp_path) + assert records[-1]["status"] == "FAILURE" + assert records[-1]["error_type"] == "RuntimeError" diff --git a/cosmos_framework/checkpoint/base.py b/cosmos_framework/checkpoint/base.py index f17ebd2d..e22f2360 100644 --- a/cosmos_framework/checkpoint/base.py +++ b/cosmos_framework/checkpoint/base.py @@ -7,11 +7,11 @@ import torch -from cosmos_framework.utils.config import CheckpointConfig, JobConfig -from cosmos_framework.utils.flags import INTERNAL from cosmos_framework.model._base import ImaginaireModel from cosmos_framework.utils import callback +from cosmos_framework.utils.config import CheckpointConfig, JobConfig from cosmos_framework.utils.easy_io import easy_io +from cosmos_framework.utils.flags import INTERNAL class AbstractCheckpointer(ABC): @@ -92,6 +92,7 @@ def save( scheduler: torch.optim.lr_scheduler.LRScheduler, grad_scaler: torch.amp.GradScaler, iteration: int, + epoch: int | None = None, ) -> None: pass diff --git a/cosmos_framework/checkpoint/dcp.py b/cosmos_framework/checkpoint/dcp.py index ba5f996a..a4ab9398 100644 --- a/cosmos_framework/checkpoint/dcp.py +++ b/cosmos_framework/checkpoint/dcp.py @@ -71,9 +71,9 @@ from cosmos_framework.checkpoint.base import AbstractCheckpointer from cosmos_framework.checkpoint.s3_filesystem import S3StorageReader, S3StorageWriter -from cosmos_framework.utils.config import CheckpointConfig, JobConfig from cosmos_framework.model._base import ImaginaireModel from cosmos_framework.utils import callback, distributed, log, misc +from cosmos_framework.utils.config import CheckpointConfig, JobConfig from cosmos_framework.utils.easy_io import easy_io from cosmos_framework.utils.generator.rand_state import get_rand_state_dict, set_rand_state_dict @@ -748,7 +748,7 @@ def keys_to_resume_during_load(self) -> tuple[set[str], str | None, bool | None] # If the path doesn't end with specific checkpoint, read the latest # checkpoint file to determine the most recent checkpoint iteration. - if not re.search(r"/checkpoints/iter_\d{9}/?$", checkpoint_path): + if not re.search(r"/checkpoints/(?:iter_\d{9}|epoch_\d+)/?$", checkpoint_path): old_ckpt_path = checkpoint_path latest_ckpt_path = os.path.join(checkpoint_path, "checkpoints/latest_checkpoint.txt") @@ -1090,6 +1090,7 @@ def save( scheduler: torch.optim.lr_scheduler.LRScheduler, grad_scaler: torch.amp.GradScaler, iteration: int, + epoch: int | None = None, ) -> None: """Save network weights, optimizer parameters, scheduler parameters to a checkpoint. @@ -1106,7 +1107,7 @@ def save( if self.callbacks is not None: self.callbacks.on_save_checkpoint_start(model, iteration) - checkpoint_file = f"iter_{iteration:09}" + checkpoint_file = f"epoch_{epoch}" if epoch is not None else f"iter_{iteration:09}" # Use rank-specific key for RNG state to ensure each rank saves its own state rng_key = f"rng_state_{dist.get_rank()}" @@ -1129,7 +1130,7 @@ def save( self.callbacks.on_save_checkpoint(model, state_dict=to_save_dict) for k in to_save_dict.keys(): - output_dirname = os.path.join(self.save_dirname, f"iter_{iteration:09}/{k}") + output_dirname = os.path.join(self.save_dirname, checkpoint_file, k) to_save_dict[k] = (to_save_dict[k], output_dirname) if self.async_mode == AsyncMode.ASYNC_WITH_PINNED_MEM: diff --git a/cosmos_framework/checkpoint/dcp_distill.py b/cosmos_framework/checkpoint/dcp_distill.py index a1b5279c..a6921e3a 100644 --- a/cosmos_framework/checkpoint/dcp_distill.py +++ b/cosmos_framework/checkpoint/dcp_distill.py @@ -26,9 +26,6 @@ from torch.distributed.checkpoint.stateful import Stateful from torch.nn.modules.module import _IncompatibleKeys -from cosmos_framework.model._base import ImaginaireModel -from cosmos_framework.utils import log, misc -from cosmos_framework.utils.easy_io import easy_io from cosmos_framework.checkpoint.dcp import ( AsyncMode, CustomLoadPlanner, @@ -38,8 +35,11 @@ DistributedCheckpointer as _DistributedCheckpointer, ) from cosmos_framework.checkpoint.dcp import ModelWrapper as VFMModelWrapper -from cosmos_framework.utils.generator.rand_state import get_rand_state_dict, set_rand_state_dict +from cosmos_framework.model._base import ImaginaireModel from cosmos_framework.model.generator.distillation.optimizer import OptimizerContainerLike, is_optimizer_container +from cosmos_framework.utils import log, misc +from cosmos_framework.utils.easy_io import easy_io +from cosmos_framework.utils.generator.rand_state import get_rand_state_dict, set_rand_state_dict __all__: tuple[str, ...] = ( "DistributedCheckpointer", @@ -324,6 +324,7 @@ def save( scheduler: Any = None, grad_scaler: torch.amp.GradScaler | None = None, iteration: int = 0, + epoch: int | None = None, ) -> None: if self.async_mode == AsyncMode.ASYNC_WITH_PINNED_MEM: self._wait_for_previous_async_checkpoint() @@ -332,7 +333,7 @@ def save( self.callbacks.on_save_checkpoint_start(model, iteration) model_dict = model.model_dict() - checkpoint_file = f"iter_{iteration:09}" + checkpoint_file = f"epoch_{epoch}" if epoch is not None else f"iter_{iteration:09}" rng_key = f"rng_state_{dist.get_rank()}" to_save_dict: dict[str, Any] = { @@ -360,7 +361,7 @@ def save( to_save_dict["dataloader"] = dataloader_wrapper.state_dict() for key in list(to_save_dict.keys()): - output_dirname = os.path.join(self.save_dirname, f"iter_{iteration:09}/{key}") + output_dirname = os.path.join(self.save_dirname, checkpoint_file, key) to_save_dict[key] = (to_save_dict[key], output_dirname) if self.callbacks is not None: diff --git a/cosmos_framework/checkpoint/dummy.py b/cosmos_framework/checkpoint/dummy.py index 3cfa1bab..443241d9 100644 --- a/cosmos_framework/checkpoint/dummy.py +++ b/cosmos_framework/checkpoint/dummy.py @@ -22,6 +22,7 @@ def save( scheduler: torch.optim.lr_scheduler.LRScheduler, grad_scaler: torch.amp.GradScaler, iteration: int, + epoch: int | None = None, ) -> None: pass diff --git a/cosmos_framework/configs/base/reasoner/defaults/callbacks.py b/cosmos_framework/configs/base/reasoner/defaults/callbacks.py index 7d1911f5..37c7a26a 100644 --- a/cosmos_framework/configs/base/reasoner/defaults/callbacks.py +++ b/cosmos_framework/configs/base/reasoner/defaults/callbacks.py @@ -7,22 +7,22 @@ from hydra.core.config_store import ConfigStore -from cosmos_framework.callbacks.manual_gc import ManualGarbageCollection -from cosmos_framework.utils.lazy_config import PLACEHOLDER -from cosmos_framework.utils.lazy_config import LazyCall as L -from cosmos_framework.utils.callback import LowPrecisionCallback, WandBCallback from cosmos_framework.callbacks.dataloader_state import DataLoaderStateCallback - from cosmos_framework.callbacks.grad_clip import GradClip from cosmos_framework.callbacks.hf_export import HFExportCallback from cosmos_framework.callbacks.iter_speed import IterSpeed from cosmos_framework.callbacks.learning_rate_logger import LearningRateLogger from cosmos_framework.callbacks.log_tensor_shape import LogTensorShapeCallback +from cosmos_framework.callbacks.manual_gc import ManualGarbageCollection from cosmos_framework.callbacks.param_count import ParamCount +from cosmos_framework.callbacks.tao_status import TAOStatusCallback from cosmos_framework.callbacks.tokens_per_sec import VLMTokensPerSec from cosmos_framework.callbacks.wandb_log import WandbCallback as WandBCallbackMultiplier from cosmos_framework.callbacks.wandb_vis import VisualizationLoggingCallback from cosmos_framework.configs.base.defaults.callbacks import JOB_MONITOR_CALLBACKS +from cosmos_framework.utils.callback import LowPrecisionCallback, WandBCallback +from cosmos_framework.utils.lazy_config import PLACEHOLDER +from cosmos_framework.utils.lazy_config import LazyCall as L # from cosmos_framework.utils.callback import NVTXCallback @@ -52,6 +52,13 @@ def register_callbacks(): config=PLACEHOLDER, trainer=PLACEHOLDER, ), # reads model.precision; no extra kwarg needed + tao=L(TAOStatusCallback)( + enabled=False, + status_file_path=None, + experiment_name="", + logging_interval=1, + validation_heartbeat_interval=1, + ), # nvtx=L(NVTXCallback)(synchronize=True), ) diff --git a/cosmos_framework/configs/base/reasoner/defaults/policy_config.py b/cosmos_framework/configs/base/reasoner/defaults/policy_config.py index 8c5adccf..526d8241 100644 --- a/cosmos_framework/configs/base/reasoner/defaults/policy_config.py +++ b/cosmos_framework/configs/base/reasoner/defaults/policy_config.py @@ -30,7 +30,21 @@ class PolicyConfig: # 0 < exponent < 1 -> interpolation; e.g. exponent=0.5 gives square-root per-token loss (Qwen3-VL) weighted_ce_exponent: float = 1.0 - # Extra model config + # Parameter-efficient fine-tuning. These fields intentionally live on + # the VLM policy instead of reusing the VFM/MoT model config: the two + # backends have different module names and checkpoint layouts. + lora_enabled: bool = False + lora_rank: int = 16 + lora_alpha: float = 32.0 + lora_dropout: float = 0.0 + lora_target_modules: str = "q_proj,k_proj,v_proj,o_proj" + lora_bias: str = "none" + lora_use_rslora: bool = False + lora_modules_to_save: str = "" + lora_precision: str | None = None + + # Legacy free-form field retained for config compatibility. New recipes + # must use the explicit fields above so PEFT equivalence can be validated. lora: Union[str, None] = None enable_liger_kernel: bool = False trainable_map: Union[str, None] = None @@ -41,6 +55,27 @@ class PolicyConfig: # "sdpa", or "eager" for fallback. attn_implementation: str = "cosmos" + # Qwen3-VL's patch projection is a non-overlapping Conv3d and is therefore + # algebraically equivalent to a linear projection. ``auto`` selects the + # linear implementation on A100 (SM80), where large FSDP runs have exposed + # a cuDNN Conv3d backward failure. No environment-time monkey patch is + # required. + qwen3_vl_patch_embed: str = "auto" + + def __attrs_post_init__(self) -> None: + if self.lora_rank <= 0: + raise ValueError("lora_rank must be positive") + if self.lora_alpha <= 0: + raise ValueError("lora_alpha must be positive") + if not 0.0 <= self.lora_dropout < 1.0: + raise ValueError("lora_dropout must be in [0, 1)") + if self.lora_bias not in {"none", "all", "lora_only"}: + raise ValueError("lora_bias must be one of: none, all, lora_only") + if self.lora_precision not in {None, "float32", "float16", "bfloat16"}: + raise ValueError("lora_precision must be float32, float16, bfloat16, or unset") + if self.qwen3_vl_patch_embed not in {"auto", "linear", "conv3d"}: + raise ValueError("qwen3_vl_patch_embed must be auto, linear, or conv3d") + @attrs.define(slots=False) class VLMModelConfig: diff --git a/cosmos_framework/configs/base/reasoner/experiment/tao_video_sft.py b/cosmos_framework/configs/base/reasoner/experiment/tao_video_sft.py new file mode 100644 index 00000000..d0dd956e --- /dev/null +++ b/cosmos_framework/configs/base/reasoner/experiment/tao_video_sft.py @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +"""Dataset-neutral public imports for TAO video-supervision SFT recipes.""" + +from cosmos_framework.configs.base.reasoner.experiment.wts_vlm import ( + VideoConversationDataset, + VideoSFTProcessor, + tao_task_aware_video_reasoning, + tao_task_aware_video_reasoning_edge, + tao_video_conversation, + tao_video_conversation_edge, +) + +__all__ = [ + "VideoConversationDataset", + "VideoSFTProcessor", + "tao_task_aware_video_reasoning", + "tao_task_aware_video_reasoning_edge", + "tao_video_conversation", + "tao_video_conversation_edge", +] diff --git a/cosmos_framework/configs/base/reasoner/experiment/tao_video_sft_test.py b/cosmos_framework/configs/base/reasoner/experiment/tao_video_sft_test.py new file mode 100644 index 00000000..432fc7fa --- /dev/null +++ b/cosmos_framework/configs/base/reasoner/experiment/tao_video_sft_test.py @@ -0,0 +1,177 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +from __future__ import annotations + +import json +import sys +import types + +import pytest + +# These unit tests exercise dataset/config construction, not video decoding. +# Keep collection deterministic on CPU hosts that do not have FFmpeg/CUDA +# libraries; decoder availability is covered by the image/compute preflight. +torchcodec_video = types.ModuleType("cosmos_framework.utils.generator.torchcodec_video") +torchcodec_video.TorchCodecVideoReader = object +sys.modules.setdefault("cosmos_framework.utils.generator.torchcodec_video", torchcodec_video) + +from cosmos_framework.configs.base.reasoner.experiment.tao_video_sft import ( + VideoConversationDataset, + VideoSFTProcessor, + tao_task_aware_video_reasoning, + tao_task_aware_video_reasoning_edge, + tao_video_conversation_edge, +) +from cosmos_framework.data.generator.local_datasets.tao_vl_reason import ( + TaoVlReasonDaftDataset, + apply_daft_chat_template, + parse_path_list, +) + + +def _install_fake_daft(monkeypatch) -> list[object]: + calls: list[object] = [] + + class FakeDataset: + def __init__(self, **kwargs) -> None: + calls.append(kwargs) + self._raw_length = 2 + + def __getitem__(self, index: int) -> list[dict]: + return [{"role": "assistant", "content": f"daft-{index}"}] + + def fake_template(processor) -> None: + calls.append(processor) + + package = types.ModuleType("nvidia_tao_daft") + datasets = types.ModuleType("nvidia_tao_daft.datasets") + module = types.ModuleType("nvidia_tao_daft.datasets.tao_vl_reason_v1_0") + module.TaoVlReasonV1_0CosmosRLConversationDataset = FakeDataset + module.apply_chat_template_override = fake_template + monkeypatch.setitem(sys.modules, "nvidia_tao_daft", package) + monkeypatch.setitem(sys.modules, "nvidia_tao_daft.datasets", datasets) + monkeypatch.setitem(sys.modules, "nvidia_tao_daft.datasets.tao_vl_reason_v1_0", module) + return calls + + +def test_video_conversation_dataset_resolves_media_paths_and_limit(tmp_path) -> None: + media = tmp_path / "videos" + media.mkdir() + (media / "clip.mp4").write_bytes(b"video") + annotations = tmp_path / "annotations.json" + annotations.write_text( + json.dumps( + [ + { + "video": "clip.mp4", + "conversations": [ + {"from": "human", "value": "