diff --git a/.github/workflows/ci-macos-mlx.yml b/.github/workflows/ci-macos-mlx.yml index daf968bd9..62054c839 100644 --- a/.github/workflows/ci-macos-mlx.yml +++ b/.github/workflows/ci-macos-mlx.yml @@ -8,6 +8,8 @@ on: - "fastvideo/mlx_runtime/**" - "fastvideo/tests/mlx/**" - "fastvideo/tests/platforms/test_mps_vsa_error.py" + - "fastvideo/tests/platforms/test_cpu_sdpa.py" + - "fastvideo/platforms/cpu.py" - "fastvideo/platforms/mps.py" - "fastvideo/platforms/__init__.py" - "fastvideo/__init__.py" @@ -79,6 +81,12 @@ jobs: fastvideo/tests/mlx/test_mlx_compile_parity.py \ fastvideo/tests/mlx/test_mlx_checkpoint.py \ fastvideo/tests/mlx/test_mlx_checkpoint_compat.py \ + fastvideo/tests/mlx/test_mlx_affine_dq_gemm.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_parity.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_vsa.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_vsa_regressions.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_fast_mode.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_fast_spatial.py \ fastvideo/tests/mlx/test_mlx_fastwan_benchmark.py \ fastvideo/tests/mlx/test_taehv_decode.py \ fastvideo/tests/mlx/test_frame_upsample.py \ @@ -91,7 +99,8 @@ jobs: fastvideo/tests/mlx/test_mlx_rife_interpolation.py::test_rife_download_unavailable_has_specific_error \ fastvideo/tests/mlx/test_mlx_rife_interpolation.py::test_rife_backend_regression_is_not_skip_eligible \ fastvideo/tests/platforms/test_mps_vsa_error.py \ - -q + fastvideo/tests/platforms/test_cpu_sdpa.py \ + -v -s -o faulthandler_timeout=120 # Same tests on MLX's CPU backend. Hosted macOS runners are scarce and # slower to schedule; this Linux job gives fast PR signal on the identical @@ -136,6 +145,12 @@ jobs: fastvideo/tests/mlx/test_mlx_compile_parity.py \ fastvideo/tests/mlx/test_mlx_checkpoint.py \ fastvideo/tests/mlx/test_mlx_checkpoint_compat.py \ + fastvideo/tests/mlx/test_mlx_affine_dq_gemm.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_parity.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_vsa.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_vsa_regressions.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_fast_mode.py \ + fastvideo/tests/mlx/test_mlx_minimax_h3_fast_spatial.py \ fastvideo/tests/mlx/test_mlx_fastwan_benchmark.py \ fastvideo/tests/mlx/test_taehv_decode.py \ fastvideo/tests/mlx/test_frame_upsample.py \ @@ -148,4 +163,5 @@ jobs: fastvideo/tests/mlx/test_mlx_rife_interpolation.py::test_rife_download_unavailable_has_specific_error \ fastvideo/tests/mlx/test_mlx_rife_interpolation.py::test_rife_backend_regression_is_not_skip_eligible \ fastvideo/tests/platforms/test_mps_vsa_error.py \ - -q + fastvideo/tests/platforms/test_cpu_sdpa.py \ + -v -s -o faulthandler_timeout=120 diff --git a/.gitignore b/.gitignore index 0bb1ca629..5f3453ddb 100644 --- a/.gitignore +++ b/.gitignore @@ -133,6 +133,9 @@ fastvideo/tests/ssim/reference_videos/** !fastvideo/tests/ssim/reference_videos/**/*.mp4 !fastvideo/tests/ssim/reference_videos/**/*.png +# Local H3 MLX kernel / exactness benches (JSON, logs, frames, videos) +.kernel_bench/ + # Editor logs and local Python version pins (accidentally committed) *.nvimlog .nvimlog diff --git a/apps/fastvideo_studio/database.py b/apps/fastvideo_studio/database.py index 9969e2b35..264699bd3 100644 --- a/apps/fastvideo_studio/database.py +++ b/apps/fastvideo_studio/database.py @@ -9,6 +9,7 @@ from __future__ import annotations import contextlib import logging +import json import sqlite3 import threading from pathlib import Path @@ -46,8 +47,13 @@ DEFAULT_SETTINGS: dict[str, Any] = { def _sqlite_row_get(row: sqlite3.Row, key: str, default: Any) -> Any: - """Like dict.get for sqlite3.Row (Row has no .get on Python 3.10).""" - return row[key] if key in row else default # noqa: SIM401 + """Like dict.get for sqlite3.Row (Row has no .get on Python 3.10). + + NOTE: `key in row` tests Row *values*, not column names, so the membership + check has to go through .keys() -- otherwise every lookup falls back to the + default and jobs restored from the database lose their stored fields. + """ + return row[key] if key in row.keys() else default # noqa: SIM401, SIM118 def _get_db_path(data_dir: Path) -> Path: @@ -83,6 +89,9 @@ def _migrate_db(conn: sqlite3.Connection) -> None: _add_column_if_missing(conn, "jobs", "fps", "INTEGER", "24") _add_column_if_missing(conn, "jobs", "workload_type", "TEXT", "'t2v'") _add_column_if_missing(conn, "jobs", "image_path", "TEXT", "''") + _add_column_if_missing(conn, "jobs", "name", "TEXT", "''") + _add_column_if_missing(conn, "jobs", "last_image_path", "TEXT", "''") + _add_column_if_missing(conn, "jobs", "references_json", "TEXT", "''") _add_column_if_missing(conn, "jobs", "job_type", "TEXT", "'inference'") _add_column_if_missing(conn, "jobs", "data_path", "TEXT", "''") _add_column_if_missing(conn, "jobs", "max_train_steps", "INTEGER", "1000") @@ -242,7 +251,8 @@ class Database: self._execute( """ INSERT INTO jobs ( - id, model_id, prompt, workload_type, image_path, job_type, status, + id, model_id, name, prompt, workload_type, image_path, + last_image_path, references_json, job_type, status, created_at, started_at, finished_at, error, output_path, log_file_path, num_inference_steps, num_frames, height, width, guidance_scale, guidance_rescale, fps, seed, num_gpus, dit_cpu_offload, @@ -254,14 +264,17 @@ class Database: dmd_use_vsa, dmd_vsa_sparsity, dmd_denoising_steps, real_score_guidance_scale, generator_update_interval, real_score_model_path, fake_score_model_path - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( job["id"], job["model_id"], + job.get("name", ""), job["prompt"], job.get("workload_type", "t2v"), job.get("image_path", ""), + job.get("last_image_path", ""), + json.dumps(job.get("references") or []), job.get("job_type", "inference"), job["status"], job["created_at"], @@ -540,9 +553,12 @@ def _row_to_job(row: sqlite3.Row) -> dict[str, Any]: result = { "id": row["id"], "model_id": row["model_id"], + "name": _sqlite_row_get(row, "name", "") or "", "prompt": row["prompt"], "workload_type": _sqlite_row_get(row, "workload_type", "t2v"), "image_path": _sqlite_row_get(row, "image_path", "") or "", + "last_image_path": _sqlite_row_get(row, "last_image_path", "") or "", + "references": _sqlite_row_get(row, "references_json", "") or "", "job_type": _sqlite_row_get(row, "job_type", "inference"), "status": row["status"], "created_at": row["created_at"], diff --git a/apps/fastvideo_studio/job_runner.py b/apps/fastvideo_studio/job_runner.py index dbe389516..e3316d833 100644 --- a/apps/fastvideo_studio/job_runner.py +++ b/apps/fastvideo_studio/job_runner.py @@ -10,7 +10,9 @@ from __future__ import annotations import atexit import collections import contextlib +import copy import enum +import json import logging import logging.handlers import multiprocessing as mp @@ -123,10 +125,13 @@ class LogBufferHandler(logging.Handler): class Job: id: str model_id: str - prompt: str + name: str = "" + prompt: str = "" workload_type: str = "t2v" job_type: str = "inference" image_path: str = "" + last_image_path: str = "" + references: list[dict[str, Any]] = field(default_factory=list) status: JobStatus = JobStatus.PENDING created_at: float = field(default_factory=time.time) started_at: float | None = None @@ -145,6 +150,7 @@ class Job: negative_prompt: str = "" num_gpus: int = 1 dit_cpu_offload: bool = False + dit_layerwise_offload: bool = False text_encoder_cpu_offload: bool = False vae_cpu_offload: bool = False image_encoder_cpu_offload: bool = False @@ -180,10 +186,13 @@ class Job: return { "id": self.id, "model_id": self.model_id, + "name": self.name, "prompt": self.prompt, "workload_type": self.workload_type, "job_type": self.job_type, "image_path": self.image_path, + "last_image_path": self.last_image_path, + "references": self.references, "status": self.status.value, "created_at": self.created_at, "started_at": self.started_at, @@ -202,6 +211,7 @@ class Job: "negative_prompt": self.negative_prompt, "num_gpus": self.num_gpus, "dit_cpu_offload": self.dit_cpu_offload, + "dit_layerwise_offload": self.dit_layerwise_offload, "text_encoder_cpu_offload": self.text_encoder_cpu_offload, "vae_cpu_offload": self.vae_cpu_offload, "image_encoder_cpu_offload": self.image_encoder_cpu_offload, @@ -232,6 +242,75 @@ class Job: } +MINIMAX_H3_REF2VA_PIPELINE = "MiniMaxH3Ref2VAModularPipeline" + + +def _build_h3_references(raw: list[dict[str, Any]]) -> list[Any]: + """Turn the API's reference dicts into MiniMaxH3Reference objects. + + Imported lazily so the API server starts without pulling in fastvideo. + """ + from fastvideo.pipelines.basic.minimax_h3 import MiniMaxH3Reference + + built = [] + for i, ref in enumerate(raw): + source = (ref or {}).get("source") + if not source: + raise ValueError(f"reference {i} has no source") + if not os.path.isfile(source): + raise ValueError(f"reference {i} source not found: {source}") + kwargs: dict[str, Any] = { + "source": source, + "media_type": (ref.get("media_type") or "image"), + } + for opt in ("soundtrack", "fps", "sample_rate"): + if ref.get(opt) not in (None, ""): + kwargs[opt] = ref[opt] + built.append(MiniMaxH3Reference(**kwargs)) + return built + + +JOB_LOG_FILENAME = "out.log" + + +def _job_log_path(output_dir: str, job_id: str) -> str: + """Each job's log lives beside its outputs: //out.log.""" + return os.path.join(output_dir, job_id, JOB_LOG_FILENAME) + + +def _decode_references(value: Any) -> list[dict[str, Any]]: + """Reference lists round-trip through the DB as JSON text.""" + if not value: + return [] + if isinstance(value, list): + return list(value) + try: + decoded = json.loads(value) + except (TypeError, ValueError): + logger.warning("Could not decode stored references: %r", value) + return [] + return list(decoded) if isinstance(decoded, list) else [] + + +def _generator_is_alive(generator: Any) -> bool: + """True if the generator's worker processes are all still running. + + A cached VideoGenerator holds a MultiprocExecutor whose workers are separate + processes; nothing notices when they exit. Probing `proc.is_alive()` is what + the executor itself uses during shutdown. Anything unexpected in the object + graph is treated as alive so a probe failure can never wedge the cache. + """ + executor = getattr(generator, "executor", None) + workers = getattr(executor, "workers", None) + if not workers: + return True + try: + return all(w.proc.is_alive() for w in workers) + except Exception: + logger.debug("Worker liveness probe failed", exc_info=True) + return True + + class JobRunner: """Manages video generation jobs, their execution, and generator caching.""" @@ -276,7 +355,7 @@ class JobRunner: """Populate job's log buffer from its log file if it exists.""" path = job.log_file_path if not path: - path = os.path.join(self.log_dir, f"{job.id}.log") + path = _job_log_path(self.output_dir, job.id) if not os.path.isfile(path): return try: @@ -314,10 +393,13 @@ class JobRunner: job = Job( id=row["id"], model_id=row["model_id"], + name=row.get("name", "") or "", prompt=row["prompt"], workload_type=row.get("workload_type", "t2v"), job_type=row.get("job_type", "inference"), image_path=row.get("image_path", "") or "", + last_image_path=row.get("last_image_path", "") or "", + references=_decode_references(row.get("references")), data_path=row.get("data_path", "") or "", max_train_steps=row.get("max_train_steps", 1000), train_batch_size=row.get("train_batch_size", 1), @@ -350,6 +432,7 @@ class JobRunner: negative_prompt=row.get("negative_prompt", "") or "", num_gpus=row.get("num_gpus", 1), dit_cpu_offload=row.get("dit_cpu_offload", False), + dit_layerwise_offload=row.get("dit_layerwise_offload", False), text_encoder_cpu_offload=row.get("text_encoder_cpu_offload", False), vae_cpu_offload=row.get("vae_cpu_offload", False), image_encoder_cpu_offload=row.get("image_encoder_cpu_offload", False), @@ -394,9 +477,12 @@ class JobRunner: job_id: str, model_id: str, prompt: str, + name: str = "", workload_type: str = "t2v", job_type: str = "inference", image_path: str = "", + last_image_path: str = "", + references: list[dict[str, Any]] | None = None, data_path: str = "", max_train_steps: int = 1000, train_batch_size: int = 1, @@ -422,6 +508,7 @@ class JobRunner: num_gpus: int = 1, negative_prompt: str = "", dit_cpu_offload: bool = False, + dit_layerwise_offload: bool = False, text_encoder_cpu_offload: bool = False, vae_cpu_offload: bool = False, image_encoder_cpu_offload: bool = False, @@ -435,10 +522,13 @@ class JobRunner: job = Job( id=job_id, model_id=model_id, + name=(name or "").strip(), prompt=prompt.strip(), workload_type=workload_type or "t2v", job_type=job_type or "inference", image_path=image_path or "", + last_image_path=last_image_path or "", + references=list(references or []), data_path=data_path or "", max_train_steps=max_train_steps, train_batch_size=train_batch_size, @@ -464,6 +554,7 @@ class JobRunner: negative_prompt=negative_prompt or "", num_gpus=num_gpus, dit_cpu_offload=dit_cpu_offload, + dit_layerwise_offload=dit_layerwise_offload, text_encoder_cpu_offload=text_encoder_cpu_offload, vae_cpu_offload=vae_cpu_offload, image_encoder_cpu_offload=image_encoder_cpu_offload, @@ -521,6 +612,84 @@ class JobRunner: logger.info("Deleted job %s", job.id) return True + CONFIG_FIELDS: tuple[str, ...] = ( + "model_id", + "name", + "prompt", + "workload_type", + "job_type", + "image_path", + "last_image_path", + "references", + "negative_prompt", + "num_inference_steps", + "num_frames", + "height", + "width", + "guidance_scale", + "guidance_rescale", + "fps", + "seed", + "num_gpus", + "dit_cpu_offload", + "dit_layerwise_offload", + "text_encoder_cpu_offload", + "vae_cpu_offload", + "image_encoder_cpu_offload", + "use_fsdp_inference", + "enable_torch_compile", + "vsa_sparsity", + "tp_size", + "sp_size", + "data_path", + "max_train_steps", + "train_batch_size", + "learning_rate", + "num_latent_t", + "validation_dataset_file", + "lora_rank", + "dmd_use_vsa", + "dmd_vsa_sparsity", + "dmd_denoising_steps", + "real_score_guidance_scale", + "generator_update_interval", + "real_score_model_path", + "fake_score_model_path", + ) + + def duplicate_job(self, job_id: str, new_job_id: str) -> Job: + """Create a new pending job with an existing job's configuration. + + Runtime state (status, timings, logs, outputs) is not carried over. + """ + with self._jobs_lock: + source = self._jobs.get(job_id) + if source is None: + raise ValueError(f"Job {job_id} not found") + config = {f: copy.deepcopy(getattr(source, f)) for f in self.CONFIG_FIELDS} + return self.create_job(job_id=new_job_id, **config) + + #: Editable exactly when startable: the same set start_job() accepts. + EDITABLE_STATUSES = (JobStatus.PENDING, JobStatus.FAILED, JobStatus.STOPPED) + + def update_job_config(self, job_id: str, updates: dict[str, Any]) -> Job: + """Edit the configuration of a job that has not produced a result.""" + with self._jobs_lock: + job = self._jobs.get(job_id) + if job is None: + raise ValueError(f"Job {job_id} not found") + if job.status not in self.EDITABLE_STATUSES: + allowed = ", ".join(s.value for s in self.EDITABLE_STATUSES) + raise ValueError(f"Job is {job.status.value}; only {allowed} jobs can be edited. " + "Duplicate it instead.") + unknown = set(updates) - set(self.CONFIG_FIELDS) + if unknown: + raise ValueError(f"Not editable: {', '.join(sorted(unknown))}") + for field_name, value in updates.items(): + setattr(job, field_name, value) + self._save_job(job) + return job + def start_job(self, job_id: str) -> Job: """Start (or restart) a pending / stopped / failed job. @@ -623,6 +792,8 @@ class JobRunner: workload_type: str, num_gpus: int, dit_cpu_offload: bool = False, + dit_layerwise_offload: bool = False, + override_pipeline_cls_name: str | None = None, text_encoder_cpu_offload: bool = False, vae_cpu_offload: bool = False, image_encoder_cpu_offload: bool = False, @@ -638,6 +809,10 @@ class JobRunner: workload_type, num_gpus, dit_cpu_offload, + dit_layerwise_offload, + # Ref2VA loads different DiT weights (transformer_ref), so the + # override must key the cache or a t2v/i2v generator gets reused. + override_pipeline_cls_name, text_encoder_cpu_offload, vae_cpu_offload, image_encoder_cpu_offload, @@ -650,8 +825,21 @@ class JobRunner: # Generators are cached by model_id and configuration parameters with self._generators_lock: - if cache_key in self._generators: - return self._generators[cache_key] + cached = self._generators.get(cache_key) + if cached is not None: + if _generator_is_alive(cached): + return cached + # Workers can exit while a generator sits idle in the cache; + # reusing it fails every later job with the same config. + logger.warning( + "Cached generator for %s has dead workers; reloading.", + model_id, + ) + self._generators.pop(cache_key, None) + try: + cached.shutdown() + except Exception: + logger.debug("Shutdown of the dead generator failed", exc_info=True) # Import lazily so starting the server is fast even without a GPU. from fastvideo import VideoGenerator @@ -677,6 +865,11 @@ class JobRunner: gen = VideoGenerator.from_pretrained( model_id, workload_type=workload_type, + num_gpus=num_gpus, + dit_layerwise_offload=dit_layerwise_offload, + **({ + "override_pipeline_cls_name": override_pipeline_cls_name + } if override_pipeline_cls_name else {}), dit_cpu_offload=dit_cpu_offload, text_encoder_cpu_offload=text_encoder_cpu_offload, vae_cpu_offload=vae_cpu_offload, @@ -706,10 +899,9 @@ class JobRunner: def _run_training_job(self, job: Job): """Run a finetuning, distillation, or LoRA job via subprocess.""" buf = job._log_buf - os.makedirs(self.log_dir, exist_ok=True) - job.log_file_path = os.path.join(self.log_dir, f"{job.id}.log") job_output_dir = os.path.join(self.output_dir, job.id) os.makedirs(job_output_dir, exist_ok=True) + job.log_file_path = _job_log_path(self.output_dir, job.id) if not job.data_path or not os.path.isdir(job.data_path): job.status = JobStatus.FAILED @@ -827,8 +1019,8 @@ class JobRunner: def _run_inference_job(self, job: Job): buf = job._log_buf - os.makedirs(self.log_dir, exist_ok=True) - job.log_file_path = os.path.join(self.log_dir, f"{job.id}.log") + os.makedirs(os.path.join(self.output_dir, job.id), exist_ok=True) + job.log_file_path = _job_log_path(self.output_dir, job.id) # Add file handler to persist logs file_handler = logging.FileHandler(job.log_file_path, mode='w', encoding='utf-8') @@ -875,62 +1067,45 @@ class JobRunner: buf.phase = "loading model" logger.info("Loading model...") - # Run generator creation in a background thread so we - # can poll _stop_event while the (potentially slow) - # model download / load is in progress. - _gen_result: list[Any] = [] - _gen_error: list[BaseException] = [] + # The generator MUST be created on this thread: building it spawns + # the executor's worker processes, and they are torn down if the + # creating thread exits. Running it in a helper thread (to poll + # _stop_event during load) made every collective_rpc fail with + # ConnectionResetError. + if job._stop_event.is_set(): + job.status = JobStatus.STOPPED + job.finished_at = time.time() + self._save_job(job) + logger.warning("Job %s stopped before model loading", job.id) + buf.phase = "stopped" + return - def _load_generator() -> None: - try: - gen = self._get_or_create_generator( - job.model_id, - job.workload_type, - job.num_gpus, - dit_cpu_offload=job.dit_cpu_offload, - text_encoder_cpu_offload=(job.text_encoder_cpu_offload), - vae_cpu_offload=job.vae_cpu_offload, - image_encoder_cpu_offload=(job.image_encoder_cpu_offload), - use_fsdp_inference=job.use_fsdp_inference, - enable_torch_compile=(job.enable_torch_compile), - vsa_sparsity=job.vsa_sparsity, - tp_size=job.tp_size, - sp_size=job.sp_size, - log_queue=log_queue, - ) - _gen_result.append(gen) - except BaseException as exc: - _gen_error.append(exc) - - loader = threading.Thread( - target=_load_generator, - daemon=True, + generator = self._get_or_create_generator( + job.model_id, + job.workload_type, + job.num_gpus, + dit_cpu_offload=job.dit_cpu_offload, + dit_layerwise_offload=job.dit_layerwise_offload, + override_pipeline_cls_name=(MINIMAX_H3_REF2VA_PIPELINE if job.references else None), + text_encoder_cpu_offload=(job.text_encoder_cpu_offload), + vae_cpu_offload=job.vae_cpu_offload, + image_encoder_cpu_offload=(job.image_encoder_cpu_offload), + use_fsdp_inference=job.use_fsdp_inference, + enable_torch_compile=(job.enable_torch_compile), + vsa_sparsity=job.vsa_sparsity, + tp_size=job.tp_size, + sp_size=job.sp_size, + log_queue=log_queue, ) - loader.start() - - while loader.is_alive(): - if job._stop_event.is_set(): - job.status = JobStatus.STOPPED - job.finished_at = time.time() - self._save_job(job) - logger.warning( - "Job %s stopped during model loading", - job.id, - ) - buf.phase = "stopped" - return - loader.join(timeout=0.5) - - if _gen_error: - raise _gen_error[0] - - generator = _gen_result[0] buf.phase = "generating" logger.info("Starting generation for job %s (model=%s)", job.id, job.model_id) + # Without a name FastVideo derives the filename from the prompt. + safe_name = re.sub(r'[\\/:*?"<>|]+', "", job.name).strip().strip(".") + output_target = (os.path.join(job_output_dir, f"{safe_name[:80]}.mp4") if safe_name else job_output_dir) gen_kwargs: dict[str, Any] = { "prompt": job.prompt, - "output_path": job_output_dir, + "output_path": output_target, "save_video": True, "num_inference_steps": job.num_inference_steps, "num_frames": job.num_frames, @@ -945,6 +1120,12 @@ class JobRunner: } if job.image_path: gen_kwargs["image_path"] = job.image_path + if job.references: + gen_kwargs["references"] = _build_h3_references(job.references) + if job.last_image_path: + # _prepare_fl2va requires a PIL image, not a path. + from PIL import Image as _PILImage + gen_kwargs["last_image"] = _PILImage.open(job.last_image_path) generator.generate_video(**gen_kwargs) buf.phase = "saving" @@ -977,7 +1158,7 @@ class JobRunner: except Exception as exception: error_msg = str(exception) - logger.error("Critical error in job thread: %s", error_msg) + logger.exception("Critical error in job thread: %s", error_msg) job.status = JobStatus.FAILED job.error = f"Critical error ({type(exception).__name__}): {error_msg}" job.finished_at = time.time() diff --git a/apps/fastvideo_studio/models/create_job_request.py b/apps/fastvideo_studio/models/create_job_request.py index 525362d18..47b862d6d 100644 --- a/apps/fastvideo_studio/models/create_job_request.py +++ b/apps/fastvideo_studio/models/create_job_request.py @@ -1,15 +1,20 @@ # SPDX-License-Identifier: Apache-2.0 """Request model for creating a job.""" +from typing import Any + from pydantic import BaseModel class CreateJobRequest(BaseModel): model_id: str + name: str = "" prompt: str workload_type: str = "t2v" job_type: str = "inference" image_path: str = "" + last_image_path: str = "" + references: list[dict[str, Any]] | None = None data_path: str = "" max_train_steps: int = 1000 train_batch_size: int = 1 @@ -28,6 +33,7 @@ class CreateJobRequest(BaseModel): seed: int = 1024 num_gpus: int = 1 dit_cpu_offload: bool = False + dit_layerwise_offload: bool = False text_encoder_cpu_offload: bool = False vae_cpu_offload: bool = False image_encoder_cpu_offload: bool = False diff --git a/apps/fastvideo_studio/server.py b/apps/fastvideo_studio/server.py index d2301c72f..d4fe3e481 100644 --- a/apps/fastvideo_studio/server.py +++ b/apps/fastvideo_studio/server.py @@ -18,6 +18,7 @@ import argparse import contextlib import logging import os +import re import shutil import signal import time @@ -116,7 +117,30 @@ def list_models(workload_type: str | None = None) -> list[dict[str, Any]]: return _available_models +def _safe_upload_name(filename: str | None, ext: str) -> str: + """A filesystem-safe version of the client's filename, keeping it readable. + + Uploads live under a per-file uuid directory, so the basename does not have + to be unique -- only safe. Keeping the original name means the path stays + self-describing wherever it travels: the database, job logs, and payloads + copied back out to the API. + """ + stem = os.path.basename(filename or "").rsplit(".", 1)[0] + stem = re.sub(r"[^A-Za-z0-9._-]+", "_", stem).strip("._-") + return f"{stem[:80] or 'upload'}{ext}" + + +def _upload_destination(ext: str, filename: str | None) -> str: + """//""" + directory = os.path.join(upload_dir, uuid.uuid4().hex) + os.makedirs(directory, exist_ok=True) + return os.path.join(directory, _safe_upload_name(filename, ext)) + + ALLOWED_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp"} +ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".mkv", ".webm", ".avi"} +ALLOWED_AUDIO_EXTENSIONS = {".wav", ".mp3", ".flac", ".m4a", ".ogg"} +ALLOWED_MEDIA_EXTENSIONS = (ALLOWED_IMAGE_EXTENSIONS | ALLOWED_VIDEO_EXTENSIONS | ALLOWED_AUDIO_EXTENSIONS) @app.post("/api/upload-image") @@ -136,8 +160,7 @@ async def upload_image(file: Annotated[UploadFile, File()], ) -> dict[str, str]: f"{', '.join(ALLOWED_IMAGE_EXTENSIONS)}"), ) os.makedirs(upload_dir, exist_ok=True) - unique_name = f"{uuid.uuid4().hex}{ext}" - dest_path = os.path.join(upload_dir, unique_name) + dest_path = _upload_destination(ext, file.filename) try: contents = await file.read() with open(dest_path, "wb") as f: @@ -150,6 +173,47 @@ async def upload_image(file: Annotated[UploadFile, File()], ) -> dict[str, str]: return {"path": os.path.abspath(dest_path)} +@app.post("/api/upload-media") +async def upload_media(file: Annotated[UploadFile, File()], ) -> dict[str, str]: + """Upload an image, video or audio file for Ref2VA references. + + Returns the absolute path plus the media_type MiniMax-H3 expects, so the + caller does not have to re-derive it from the extension. + """ + global upload_dir # noqa: PLW0603 + if not upload_dir: + raise HTTPException( + status_code=503, + detail="Upload directory not configured", + ) + ext = Path(file.filename or "").suffix.lower() + if ext not in ALLOWED_MEDIA_EXTENSIONS: + raise HTTPException( + status_code=400, + detail=(f"Invalid file type. Allowed: " + f"{', '.join(sorted(ALLOWED_MEDIA_EXTENSIONS))}"), + ) + if ext in ALLOWED_VIDEO_EXTENSIONS: + media_type = "video" + elif ext in ALLOWED_AUDIO_EXTENSIONS: + media_type = "audio" + else: + media_type = "image" + + os.makedirs(upload_dir, exist_ok=True) + dest_path = _upload_destination(ext, file.filename) + try: + contents = await file.read() + with open(dest_path, "wb") as f: + f.write(contents) + except OSError as e: + raise HTTPException( + status_code=500, + detail=f"Failed to save upload: {e}", + ) from e + return {"path": os.path.abspath(dest_path), "media_type": media_type} + + ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".webm", ".avi", ".mov", ".mkv"} @@ -282,10 +346,13 @@ def create_job(req: CreateJobRequest) -> dict[str, Any]: job = job_runner.create_job( job_id=str(uuid.uuid4()), model_id=req.model_id, + name=req.name or "", prompt=req.prompt, workload_type=req.workload_type or "t2v", job_type=job_type, image_path=req.image_path or "", + last_image_path=req.last_image_path or "", + references=req.references or [], data_path=data_path, max_train_steps=req.max_train_steps, train_batch_size=req.train_batch_size, @@ -304,6 +371,7 @@ def create_job(req: CreateJobRequest) -> dict[str, Any]: seed=req.seed, num_gpus=req.num_gpus, dit_cpu_offload=req.dit_cpu_offload, + dit_layerwise_offload=req.dit_layerwise_offload, text_encoder_cpu_offload=req.text_encoder_cpu_offload, vae_cpu_offload=req.vae_cpu_offload, image_encoder_cpu_offload=req.image_encoder_cpu_offload, @@ -336,6 +404,28 @@ def create_job(req: CreateJobRequest) -> dict[str, Any]: return job.to_dict() +@app.post("/api/jobs/{job_id}/duplicate", status_code=201) +def duplicate_job(job_id: str) -> dict[str, Any]: + """Create a new pending job with the same configuration as an existing one.""" + try: + job = job_runner.duplicate_job(job_id, str(uuid.uuid4())) + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) from e + return job.to_dict() + + +@app.patch("/api/jobs/{job_id}") +def update_job(job_id: str, updates: dict[str, Any]) -> dict[str, Any]: + """Edit a pending job's configuration. Started jobs cannot be edited.""" + try: + job = job_runner.update_job_config(job_id, updates) + except ValueError as e: + detail = str(e) + status = 404 if "not found" in detail else 400 + raise HTTPException(status_code=status, detail=detail) from e + return job.to_dict() + + @app.post("/api/jobs/{job_id}/start") def start_job(job_id: str) -> dict[str, Any]: """Start (or restart) a pending / stopped / failed job.""" diff --git a/apps/fastvideo_studio/src/components/jobs/CreateJobModal.tsx b/apps/fastvideo_studio/src/components/jobs/CreateJobModal.tsx index 1abccf4a6..77b7233ad 100644 --- a/apps/fastvideo_studio/src/components/jobs/CreateJobModal.tsx +++ b/apps/fastvideo_studio/src/components/jobs/CreateJobModal.tsx @@ -22,15 +22,35 @@ import { useStore } from '@/hooks/useStore'; import { defaultOptionsStore } from '@/stores/defaultOptions'; import { createJob, + updateJob, getDatasets, getModels, uploadImage, + uploadMedia, type CreateJobRequest, type Model, } from '@/lib/api'; import { getDefaultModelForWorkload } from '@/lib/defaultOptions'; import { WORKLOAD_OPTIONS } from '@/lib/jobConfig'; import type { JobType } from '@/lib/types'; +import { + H3_MAX_REFERENCES, + labelReferences, + referencePromptSeed, + validateReferences, + type H3Reference, +} from '@/lib/h3References'; +import { + EMPTY_H3_PROMPT_FIELDS, + H3_PROMPT_SECTIONS, + H3_SECTION_HINTS, + H3_SECTION_LABELS, + isEmptyPromptFields, + parseH3Prompt, + serializeH3Prompt, + type H3PromptFields, +} from '@/lib/h3Prompt'; +import { jobToFormFields, type JobLike } from '@/lib/jobToFields'; export interface CreateJobModalProps { isOpen: boolean; @@ -38,6 +58,10 @@ export interface CreateJobModalProps { onSuccess: () => void; jobType: JobType; workloadType: string; + /** When set, the modal edits this pending job instead of creating a new one. */ + editingJob?: JobLike | null; + /** Show the configuration without allowing changes (started/finished jobs). */ + readOnly?: boolean; } export default function CreateJobModal({ @@ -46,6 +70,8 @@ export default function CreateJobModal({ onSuccess, jobType, workloadType, + editingJob, + readOnly = false, }: CreateJobModalProps) { const { options } = useStore(defaultOptionsStore); @@ -55,8 +81,22 @@ export default function CreateJobModal({ const [models, setModels] = React.useState([]); const [modelId, setModelId] = React.useState(''); + const [name, setName] = React.useState(''); const [prompt, setPrompt] = React.useState(''); const [imagePath, setImagePath] = React.useState(''); + const [lastImagePath, setLastImagePath] = React.useState(''); + const [references, setReferences] = React.useState([]); + const [isUploadingReference, setIsUploadingReference] = React.useState(false); + const [referenceError, setReferenceError] = React.useState(null); + const [promptFields, setPromptFields] = React.useState( + EMPTY_H3_PROMPT_FIELDS, + ); + const [useGuidedPrompt, setUseGuidedPrompt] = React.useState(true); + const [lastImageFileName, setLastImageFileName] = React.useState(''); + const [isUploadingLastImage, setIsUploadingLastImage] = React.useState(false); + const [lastImageUploadError, setLastImageUploadError] = React.useState< + string | null + >(null); const [imageFileName, setImageFileName] = React.useState(''); const [isUploadingImage, setIsUploadingImage] = React.useState(false); const [negativePrompt, setNegativePrompt] = React.useState(''); @@ -70,12 +110,50 @@ export default function CreateJobModal({ const [seed, setSeed] = React.useState(1024); const [numGpus, setNumGpus] = React.useState(1); const [ditCpuOffload, setDitCpuOffload] = React.useState(false); + const [ditLayerwiseOffload, setDitLayerwiseOffload] = React.useState(false); const [textEncoderCpuOffload, setTextEncoderCpuOffload] = React.useState(false); const [vaeCpuOffload, setVaeCpuOffload] = React.useState(false); const [imageEncoderCpuOffload, setImageEncoderCpuOffload] = React.useState(false); const [useFsdpInference, setUseFsdpInference] = React.useState(false); + + // H3 is the only registered model with an end frame or references. + const supportsLastImage = modelId.toLowerCase().includes('minimax-h3'); + const usingReferences = supportsLastImage && references.length > 0; + + // JobCard re-renders on every job-list poll, so `editingJob` is a fresh + // object each time. Effects must depend on these, never on the object. + const editingJobId = editingJob?.id ?? null; + const editingJobModelId = editingJob?.model_id ?? null; + + // Layerwise offload and FSDP compete for the DiT weights and FastVideoArgs + // silently picks a winner (fastvideo_args.py:859); resolve it visibly here. + // dit_cpu_offload is deliberately not interlocked -- it is a modifier, not a + // competing strategy. + const handleDitLayerwiseOffloadChange = React.useCallback((next: boolean) => { + setDitLayerwiseOffload(next); + if (next) { + setUseFsdpInference(false); + } + }, []); + + const handleUseFsdpInferenceChange = React.useCallback((next: boolean) => { + setUseFsdpInference(next); + if (next) { + setDitLayerwiseOffload(false); + } + }, []); + + const handleNumGpusChange = React.useCallback((next: number) => { + setNumGpus(next); + if (next > 1) { + // Dropping back to one GPU leaves FSDP alone: single-GPU FSDP is a + // valid way to reach its CPU offload (docs/inference/offloading.md). + setUseFsdpInference(true); + setDitLayerwiseOffload(false); + } + }, []); const [enableTorchCompile, setEnableTorchCompile] = React.useState(false); const [vsaSparsity, setVsaSparsity] = React.useState(0); const [tpSize, setTpSize] = React.useState(-1); @@ -126,6 +204,47 @@ export default function CreateJobModal({ const justOpened = isOpen && !justOpenedRef.current; justOpenedRef.current = isOpen; if (!justOpened) return; + if (editingJob) { + // Must not fall through to the defaults below: a partially-seeded + // form silently edits values the user never saw. + const f = jobToFormFields(editingJob); + setModelId(f.modelId); + setName(f.name); + setPrompt(f.prompt); + setNegativePrompt(f.negativePrompt); + setImagePath(f.imagePath); + setImageFileName(f.imagePath.split('/').pop() ?? ''); + setLastImagePath(f.lastImagePath); + setLastImageFileName(f.lastImagePath.split('/').pop() ?? ''); + setReferences(f.references); + setPromptFields(f.promptFields ?? EMPTY_H3_PROMPT_FIELDS); + setUseGuidedPrompt(f.promptFields !== null); + setNumInferenceSteps(f.numInferenceSteps); + setNumFrames(f.numFrames); + setHeight(f.height); + setWidth(f.width); + setGuidanceScale(f.guidanceScale); + setGuidanceRescale(f.guidanceRescale); + setFps(f.fps); + setSeed(f.seed); + setNumGpus(f.numGpus); + setDitCpuOffload(f.ditCpuOffload); + setDitLayerwiseOffload(f.ditLayerwiseOffload); + setTextEncoderCpuOffload(f.textEncoderCpuOffload); + setVaeCpuOffload(f.vaeCpuOffload); + setImageEncoderCpuOffload(f.imageEncoderCpuOffload); + setUseFsdpInference(f.useFsdpInference); + setEnableTorchCompile(f.enableTorchCompile); + setVsaSparsity(f.vsaSparsity); + setTpSize(f.tpSize); + setSpSize(f.spSize); + setReferenceError(null); + setModelLoadError(null); + setImageUploadError(null); + setLastImageUploadError(null); + setSubmitError(null); + return; + } const opts = options; setNumInferenceSteps(opts.numInferenceSteps); setNumFrames(workloadType === 't2i' ? 1 : opts.numFrames); @@ -137,6 +256,7 @@ export default function CreateJobModal({ setSeed(opts.seed); setNumGpus(opts.numGpus); setDitCpuOffload(opts.ditCpuOffload); + setDitLayerwiseOffload(opts.ditLayerwiseOffload ?? false); setTextEncoderCpuOffload(opts.textEncoderCpuOffload); setVaeCpuOffload(opts.vaeCpuOffload); setImageEncoderCpuOffload(opts.imageEncoderCpuOffload); @@ -151,8 +271,16 @@ export default function CreateJobModal({ inferenceWorkload as 't2v' | 'i2v' | 't2i', ), ); + setName(''); setImagePath(''); setImageFileName(''); + setLastImagePath(''); + setLastImageFileName(''); + setLastImageUploadError(null); + setReferences([]); + setReferenceError(null); + setPromptFields(EMPTY_H3_PROMPT_FIELDS); + setUseGuidedPrompt(true); setSelectedDatasetId(''); setSelectedValidationDatasetId(''); setModelLoadError(null); @@ -168,7 +296,7 @@ export default function CreateJobModal({ setRealScoreModelPath(''); setFakeScoreModelPath(''); } - }, [isOpen, workloadType, inferenceWorkload, options]); + }, [isOpen, workloadType, inferenceWorkload, options, editingJobId]); // Load the models available for this workload. React.useEffect(() => { @@ -188,7 +316,16 @@ export default function CreateJobModal({ opts, inferenceWorkload as 't2v' | 'i2v' | 't2i', ); - const chosen = ids.includes(defaultId) ? defaultId : (list[0]?.id ?? ''); + // When editing, the job's own model wins over the workload default -- + // this resolves after the seeding effect, so choosing a default here + // would silently swap the model out from under the user. + const editedId = editingJobModelId; + const chosen = + editedId && ids.includes(editedId) + ? editedId + : ids.includes(defaultId) + ? defaultId + : (list[0]?.id ?? ''); setModelId(chosen); if (workloadType === 'dmd_t2v') { setRealScoreModelPath(chosen); @@ -210,7 +347,7 @@ export default function CreateJobModal({ return () => { stale = true; }; - }, [isOpen, inferenceWorkload, workloadType]); + }, [isOpen, inferenceWorkload, workloadType, editingJobModelId]); // Training jobs need a dataset; load the ready datasets when relevant. React.useEffect(() => { @@ -262,6 +399,106 @@ export default function CreateJobModal({ } } + async function handleLastImageChange( + e: React.ChangeEvent, + ) { + const file = e.target.files?.[0]; + if (!file) { + setLastImagePath(''); + setLastImageFileName(''); + setLastImageUploadError(null); + return; + } + setIsUploadingLastImage(true); + setLastImageFileName(file.name); + setLastImageUploadError(null); + try { + const { path } = await uploadImage(file); + setLastImagePath(path); + } catch (error) { + console.error('Failed to upload end image:', error); + setLastImagePath(''); + setLastImageFileName(''); + setLastImageUploadError( + error instanceof Error + ? `${error.message}. Choose the image again to retry.` + : 'The image could not be uploaded. Choose it again to retry.', + ); + } finally { + setIsUploadingLastImage(false); + } + } + + async function handleAddReference( + e: React.ChangeEvent, + ) { + const file = e.target.files?.[0]; + e.target.value = ''; // allow re-picking the same file + if (!file) return; + setIsUploadingReference(true); + setReferenceError(null); + try { + const { path, media_type } = await uploadMedia(file); + const next: H3Reference[] = [ + ...references, + { + id: `${Date.now()}-${file.name}`, + source: path, + media_type, + fileName: file.name, + }, + ]; + setReferences(next); + setReferenceError(validateReferences(next)); + } catch (error) { + console.error('Failed to upload reference:', error); + setReferenceError( + error instanceof Error ? error.message : 'The file could not be uploaded.', + ); + } finally { + setIsUploadingReference(false); + } + } + + function removeReference(id: string) { + const next = references.filter((r) => r.id !== id); + setReferences(next); + setReferenceError(validateReferences(next)); + } + + function seedPromptFields() { + setPromptFields({ + ...EMPTY_H3_PROMPT_FIELDS, + ...referencePromptSeed(references), + }); + setUseGuidedPrompt(true); + } + + function setPromptField(section: string, value: string) { + setPromptFields((prev) => ({ ...prev, [section]: value })); + } + + // Switching between the guided fields and the raw editor keeps whatever was + // typed: serialize on the way out, parse back on the way in. + function toggleGuidedPrompt() { + if (useGuidedPrompt) { + if (!isEmptyPromptFields(promptFields)) { + setPrompt(serializeH3Prompt(promptFields)); + } + setUseGuidedPrompt(false); + } else { + const parsed = parseH3Prompt(prompt); + if (parsed) setPromptFields(parsed); + setUseGuidedPrompt(true); + } + } + + function clearLastImage() { + setLastImagePath(''); + setLastImageFileName(''); + setLastImageUploadError(null); + } + function clearImage() { setImagePath(''); setImageFileName(''); @@ -271,7 +508,16 @@ export default function CreateJobModal({ async function handleSubmit(e: React.FormEvent) { e.preventDefault(); - if (isInference && workloadType === 'i2v' && !imagePath) return; + if (isInference && workloadType === 'i2v' && !imagePath && !usingReferences) + return; + if (usingReferences && validateReferences(references)) return; + if ( + usingReferences && + useGuidedPrompt && + isEmptyPromptFields(promptFields) && + !prompt.trim() + ) + return; // Send the dataset id; the backend resolves it to the on-disk media dir. const effectiveDataPath = selectedDatasetId ?? ''; if (!isInference && !selectedDatasetId) return; @@ -285,14 +531,35 @@ export default function CreateJobModal({ try { const payload: CreateJobRequest = { model_id: modelId, - prompt, + name: name.trim(), + prompt: + usingReferences && useGuidedPrompt && !isEmptyPromptFields(promptFields) + ? serializeH3Prompt(promptFields) + : prompt, workload_type: workloadType, job_type: effectiveJobType, ...(isInference ? { - ...(workloadType === 'i2v' && imagePath + // Ref2VA and the FL2VA keyframes are mutually exclusive: + // _prepare_ref2va rejects image_path/last_image_path outright + // when references are present. + ...(workloadType === 'i2v' && !usingReferences && imagePath ? { image_path: imagePath } : {}), + ...(workloadType === 'i2v' && + supportsLastImage && + !usingReferences && + lastImagePath + ? { last_image_path: lastImagePath } + : {}), + ...(workloadType === 'i2v' && supportsLastImage && references.length + ? { + references: references.map((r) => ({ + source: r.source, + media_type: r.media_type, + })), + } + : {}), negative_prompt: negativePrompt, num_inference_steps: numInferenceSteps, num_frames: numFrames, @@ -304,6 +571,7 @@ export default function CreateJobModal({ seed, num_gpus: numGpus, dit_cpu_offload: ditCpuOffload, + dit_layerwise_offload: ditLayerwiseOffload, text_encoder_cpu_offload: textEncoderCpuOffload, vae_cpu_offload: vaeCpuOffload, image_encoder_cpu_offload: imageEncoderCpuOffload, @@ -334,7 +602,14 @@ export default function CreateJobModal({ : {}), }), }; - await createJob(payload); + if (editingJob) { + await updateJob( + editingJob.id, + payload as unknown as Record, + ); + } else { + await createJob(payload); + } onSuccess(); onClose(); } catch (err) { @@ -356,9 +631,9 @@ export default function CreateJobModal({ const workloadLabel = WORKLOAD_OPTIONS[jobType]?.find((o) => o.type === workloadType)?.label ?? ''; - const title = `New ${jobType.charAt(0).toUpperCase() + jobType.slice(1)} Job${ - workloadLabel ? ` (${workloadLabel})` : '' - }`; + const title = `${readOnly ? 'View' : editingJob ? 'Edit' : 'New'} ${ + jobType.charAt(0).toUpperCase() + jobType.slice(1) + } Job${workloadLabel ? ` (${workloadLabel})` : ''}`; return ( + {/* disabled cascades to every control inside; display:contents + keeps the parent's flex layout. */} +
+ + setName(e.target.value)} + placeholder="Shown on the job card and used for the output filename" + disabled={isSubmitting} + /> + + {imageFileName && ( @@ -462,24 +754,169 @@ export default function CreateJobModal({ )} - -