Update api/ltx_server_refactored.py

api/ltx_server_refactored.py

@@ -1,769 +1,423 @@
-# ltx_server.py — VideoService (beta 1.1)
-# Sempre output_type="latent"; no final: VAE (bloco inteiro) → pixels → MP4.
-# Ignora UserWarning/FutureWarning e injeta VAE no manager com dtype/device corretos.
-# --- 0. WARNINGS E AMBIENTE ---
+# FILE: api/ltx_server_refactored_complete.py
+# DESCRIPTION: Final orchestrator for LTX-Video generation.
+# This version includes the fix for the narrative generation overlap bug and
+# consolidates all previous refactoring and debugging improvements.
 
-import warnings
-warnings.filterwarnings("ignore", category=UserWarning)
-warnings.filterwarnings("ignore", category=FutureWarning)
-warnings.filterwarnings("ignore", message=".*")
-from huggingface_hub import logging
-logging.set_verbosity_error()
-logging.set_verbosity_warning()
-logging.set_verbosity_info()
-logging.set_verbosity_debug()
-LTXV_DEBUG=1
-LTXV_FRAME_LOG_EVERY=8
-import os, subprocess, shlex, tempfile
-import torch
+import gc
 import json
-import numpy as np
-import random
+import logging
 import os
-import shlex
-import yaml
-from typing import List, Dict
-from pathlib import Path
-import imageio
-from PIL import Image # Import adicionado para handle_media_upload_for_dims
-import tempfile
-from huggingface_hub import hf_hub_download
-import sys
-import subprocess
-import gc
 import shutil
-import contextlib
+import sys
+import tempfile
 import time
-import traceback
-from einops import rearrange
-import torch.nn.functional as F
-from managers.vae_manager import vae_manager_singleton
-from tools.video_encode_tool import video_encode_tool_singleton
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+import random
+import torch
+import yaml
+import numpy as np
+from huggingface_hub import hf_hub_download
+
+# ==============================================================================
+# --- SETUP E IMPORTAÇÕES DO PROJETO ---
+# ==============================================================================
+
+# Configuração de logging e supressão de warnings
+import warnings
+warnings.filterwarnings("ignore")
+logging.getLogger("huggingface_hub").setLevel(logging.ERROR)
+log_level = os.environ.get("ADUC_LOG_LEVEL", "INFO").upper()
+logging.basicConfig(level=log_level, format='[%(levelname)s] [%(name)s] %(message)s')
+
+# --- Constantes de Configuração ---
 DEPS_DIR = Path("/data")
 LTX_VIDEO_REPO_DIR = DEPS_DIR / "LTX-Video"
+RESULTS_DIR = Path("/app/output")
+DEFAULT_FPS = 24.0
+FRAMES_ALIGNMENT = 8
+LTX_REPO_ID = "Lightricks/LTX-Video"
 
-# CORREÇÃO: Movido run_setup para o início para garantir que seja definido antes de ser chamado.
-def run_setup():
-    setup_script_path = "setup.py"
-    if not os.path.exists(setup_script_path):
-        print("[DEBUG] 'setup.py' não encontrado. Pulando clonagem de dependências.")
-        return
-    try:
-        print("[DEBUG] Executando setup.py para dependências...")
-        subprocess.run([sys.executable, setup_script_path], check=True)
-        print("[DEBUG] Setup concluído com sucesso.")
-    except subprocess.CalledProcessError as e:
-        print(f"[DEBUG] ERRO no setup.py (code {e.returncode}). Abortando.")
-        sys.exit(1)
-        
-if not LTX_VIDEO_REPO_DIR.exists():
-    print(f"[DEBUG] Repositório não encontrado em {LTX_VIDEO_REPO_DIR}. Rodando setup...")
-    run_setup()
-
+# Garante que a biblioteca LTX-Video seja importável
 def add_deps_to_path():
     repo_path = str(LTX_VIDEO_REPO_DIR.resolve())
-    if str(LTX_VIDEO_REPO_DIR.resolve()) not in sys.path:
+    if repo_path not in sys.path:
         sys.path.insert(0, repo_path)
-        print(f"[DEBUG] Repo adicionado ao sys.path: {repo_path}")
-def _query_gpu_processes_via_nvml(device_index: int) -> List[Dict]:
-    try:
-        import psutil
-        import pynvml as nvml
-        nvml.nvmlInit()
-        handle = nvml.nvmlDeviceGetHandleByIndex(device_index)
-        try:
-            procs = nvml.nvmlDeviceGetComputeRunningProcesses_v3(handle)
-        except Exception:
-            procs = nvml.nvmlDeviceGetComputeRunningProcesses(handle)
-        results = []
-        for p in procs:
-            pid = int(p.pid)
-            used_mb = None
-            try:
-                if getattr(p, "usedGpuMemory", None) is not None and p.usedGpuMemory not in (0,):
-                    used_mb = max(0, int(p.usedGpuMemory) // (1024 * 1024))
-            except Exception:
-                used_mb = None
-            name = "unknown"
-            user = "unknown"
-            try:
-                import psutil
-                pr = psutil.Process(pid)
-                name = pr.name()
-                user = pr.username()
-            except Exception:
-                pass
-            results.append({"pid": pid, "name": name, "user": user, "used_mb": used_mb})
-        nvml.nvmlShutdown()
-        return results
-    except Exception:
-        return []
-def _query_gpu_processes_via_nvidiasmi(device_index: int) -> List[Dict]:
-    cmd = f"nvidia-smi -i {device_index} --query-compute-apps=pid,process_name,used_memory --format=csv,noheader,nounits"
-    try:
-        out = subprocess.check_output(shlex.split(cmd), stderr=subprocess.STDOUT, text=True, timeout=2.0)
-    except Exception:
-        return []
-    results = []
-    for line in out.strip().splitlines():
-        parts = [p.strip() for p in line.split(",")]
-        if len(parts) >= 3:
-            try:
-                pid = int(parts[0]); name = parts[1]; used_mb = int(parts[2])
-                user = "unknown"
-                try:
-                    import psutil
-                    pr = psutil.Process(pid)
-                    user = pr.username()
-                except Exception:
-                    pass
-                results.append({"pid": pid, "name": name, "user": user, "used_mb": used_mb})
-            except Exception:
-                continue
-    return results
-def calculate_new_dimensions(orig_w, orig_h, divisor=8):
-    if orig_w == 0 or orig_h == 0:
-        return 512, 512
-    if orig_w >= orig_h:
-        aspect_ratio = orig_w / orig_h
-        new_h = 512
-        new_w = new_h * aspect_ratio
-    else:
-        aspect_ratio = orig_h / orig_w
-        new_w = 512
-        new_h = new_w * aspect_ratio
-    final_w = int(round(new_w / divisor)) * divisor
-    final_h = int(round(new_h / divisor)) * divisor
-    final_w = max(divisor, final_w)
-    final_h = max(divisor, final_h)
-    print(f"[Dimension Calc] Original: {orig_w}x{orig_h} -> Calculado: {new_w:.0f}x{new_h:.0f} -> Final (divisível por {divisor}): {final_w}x{final_h}")
-    return final_h, final_w
-def handle_media_upload_for_dims(filepath, current_h, current_w):
-    # CORREÇÃO: Gradio (`gr`) não deve ser usado no backend. Retornando tupla diretamente.
-    if not filepath or not os.path.exists(str(filepath)):
-        return current_h, current_w
-    try:
-        if str(filepath).lower().endswith(('.png', '.jpg', '.jpeg', '.webp')):
-            with Image.open(filepath) as img:
-                orig_w, orig_h = img.size
-        else: 
-            with imageio.get_reader(filepath) as reader:
-                meta = reader.get_meta_data()
-                orig_w, orig_h = meta.get('size', (current_w, current_h))
-        new_h, new_w = calculate_new_dimensions(orig_w, orig_h)
-        return new_h, new_w
-    except Exception as e:
-        print(f"Erro ao processar mídia para dimensões: {e}")
-        return current_h, current_w
-def _gpu_process_table(processes: List[Dict], current_pid: int) -> str:
-    if not processes:
-        return "  - Processos ativos: (nenhum)\n"
-    processes = sorted(processes, key=lambda x: (x.get("used_mb") or 0), reverse=True)
-    lines = ["  - Processos ativos (PID | USER | NAME | VRAM MB):"]
-    for p in processes:
-        star = "*" if p["pid"] == current_pid else " "
-        used_str = str(p["used_mb"]) if p.get("used_mb") is not None else "N/A"
-        lines.append(f"    {star} {p['pid']} | {p['user']} | {p['name']} | {used_str}")
-    return "\n".join(lines) + "\n"
-def log_tensor_info(tensor, name="Tensor"):
-    if not isinstance(tensor, torch.Tensor):
-        print(f"\n[INFO] '{name}' não é tensor.")
-        return
-    print(f"\n--- Tensor: {name} ---")
-    print(f"  - Shape: {tuple(tensor.shape)}")
-    print(f"  - Dtype: {tensor.dtype}")
-    print(f"  - Device: {tensor.device}")
-    if tensor.numel() > 0:
-        try:
-            print(f"  - Min: {tensor.min().item():.4f}  Max: {tensor.max().item():.4f}  Mean: {tensor.mean().item():.4f}")
-        except Exception:
-            pass
-    print("------------------------------------------\n")
+        logging.info(f"[ltx_server] LTX-Video repository added to sys.path: {repo_path}")
+
 add_deps_to_path()
-from ltx_video.pipelines.pipeline_ltx_video import ConditioningItem, LTXMultiScalePipeline
-from ltx_video.utils.skip_layer_strategy import SkipLayerStrategy
-from ltx_video.models.autoencoders.vae_encode import un_normalize_latents, normalize_latents
-from ltx_video.pipelines.pipeline_ltx_video import adain_filter_latent
-from api.ltx.inference import (
-    create_ltx_video_pipeline,
-    create_latent_upsampler,
-    load_image_to_tensor_with_resize_and_crop,
-    seed_everething,
-    calculate_padding,
-    load_media_file,
-)
+
+# --- Módulos da nossa Arquitetura ---
+try:
+    from api.gpu_manager import gpu_manager
+    from managers.vae_manager import vae_manager_singleton
+    from tools.video_encode_tool import video_encode_tool_singleton
+    from api.ltx.ltx_utils import (
+        build_ltx_pipeline_on_cpu,
+        seed_everything,
+        load_image_to_tensor_with_resize_and_crop,
+        ConditioningItem,
+    )
+    from api.utils.debug_utils import log_function_io
+except ImportError as e:
+    logging.critical(f"A crucial import from the local API/architecture failed. Error: {e}", exc_info=True)
+    sys.exit(1)
+
+# ==============================================================================
+# --- FUNÇÕES AUXILIARES DO ORQUESTRADOR ---
+# ==============================================================================
+
+@log_function_io
+def calculate_padding(orig_h: int, orig_w: int, target_h: int, target_w: int) -> Tuple[int, int, int, int]:
+    """Calculates symmetric padding required to meet target dimensions."""
+    pad_h = target_h - orig_h
+    pad_w = target_w - orig_w
+    pad_top = pad_h // 2
+    pad_bottom = pad_h - pad_top
+    pad_left = pad_w // 2
+    pad_right = pad_w - pad_left
+    return (pad_left, pad_right, pad_top, pad_bottom)
+
+# ==============================================================================
+# --- CLASSE DE SERVIÇO (O ORQUESTRADOR) ---
+# ==============================================================================
+
 class VideoService:
+    """
+    Orchestrates the high-level logic of video generation, delegating low-level
+    tasks to specialized managers and utility modules.
+    """
+
+    @log_function_io
     def __init__(self):
         t0 = time.perf_counter()
-        print("[DEBUG] Inicializando VideoService...")
-        self.debug = os.getenv("LTXV_DEBUG", "1") == "1"
-        self.frame_log_every = int(os.getenv("LTXV_FRAME_LOG_EVERY", "8"))
+        logging.info("Initializing VideoService Orchestrator...")
+        RESULTS_DIR.mkdir(parents=True, exist_ok=True)
+
+        target_main_device_str = str(gpu_manager.get_ltx_device())
+        target_vae_device_str = str(gpu_manager.get_ltx_vae_device())
+        logging.info(f"LTX allocated to devices: Main='{target_main_device_str}', VAE='{target_vae_device_str}'")
+
         self.config = self._load_config()
-        print(f"[DEBUG] Config carregada (precision={self.config.get('precision')}, sampler={self.config.get('sampler')})")
-        self.device = "cuda" if torch.cuda.is_available() else "cpu"
-        print(f"[DEBUG] Device selecionado: {self.device}")
-        self.last_memory_reserved_mb = 0.0
-        self._tmp_dirs = set(); self._tmp_files = set(); self._last_outputs = []
+        self._resolve_model_paths_from_cache()
 
-        self.pipeline, self.latent_upsampler = self._load_models()
-        print(f"[DEBUG] Pipeline e Upsampler carregados. Upsampler ativo? {bool(self.latent_upsampler)}")
+        self.pipeline, self.latent_upsampler = build_ltx_pipeline_on_cpu(self.config)
 
-        print(f"[DEBUG] Movendo modelos para {self.device}...")
-        self.pipeline.to(self.device)
-        if self.latent_upsampler:
-            self.latent_upsampler.to(self.device)
+        self.main_device = torch.device("cpu")
+        self.vae_device = torch.device("cpu")
+        self.move_to_device(main_device_str=target_main_device_str, vae_device_str=target_vae_device_str)
 
         self._apply_precision_policy()
-        print(f"[DEBUG] runtime_autocast_dtype = {getattr(self, 'runtime_autocast_dtype', None)}")
-
-        vae_manager_singleton.attach_pipeline(
-            self.pipeline,
-            device=self.device,
-            autocast_dtype=self.runtime_autocast_dtype
-        )
-        print(f"[DEBUG] VAE manager conectado: has_vae={hasattr(self.pipeline, 'vae')} device={self.device}")
+        vae_manager_singleton.attach_pipeline(self.pipeline, device=self.vae_device, autocast_dtype=self.runtime_autocast_dtype)
+        logging.info(f"VideoService ready. Startup time: {time.perf_counter()-t0:.2f}s")
+
+    def _load_config(self) -> Dict:
+        """Loads the YAML configuration file."""
+        config_path = LTX_VIDEO_REPO_DIR / "configs" / "ltxv-13b-0.9.8-distilled-fp8.yaml"
+        logging.info(f"Loading config from: {config_path}")
+        with open(config_path, "r") as file:
+            return yaml.safe_load(file)
 
-        if self.device == "cuda":
-            torch.cuda.empty_cache()
-            self._log_gpu_memory("Após carregar modelos")
-
-        print(f"[DEBUG] VideoService pronto. boot_time={time.perf_counter()-t0:.3f}s")
-
-    def _log_gpu_memory(self, stage_name: str):
-        if self.device != "cuda":
-            return
-        device_index = torch.cuda.current_device() if torch.cuda.is_available() else 0
-        current_reserved_b = torch.cuda.memory_reserved(device_index)
-        current_reserved_mb = current_reserved_b / (1024 ** 2)
-        total_memory_b = torch.cuda.get_device_properties(device_index).total_memory
-        total_memory_mb = total_memory_b / (1024 ** 2)
-        peak_reserved_mb = torch.cuda.max_memory_reserved(device_index) / (1024 ** 2)
-        delta_mb = current_reserved_mb - getattr(self, "last_memory_reserved_mb", 0.0)
-        processes = _query_gpu_processes_via_nvml(device_index) or _query_gpu_processes_via_nvidiasmi(device_index)
-        print(f"\n--- [LOG GPU] {stage_name} (cuda:{device_index}) ---")
-        print(f"  - Reservado: {current_reserved_mb:.2f} MB / {total_memory_mb:.2f} MB  (Δ={delta_mb:+.2f} MB)")
-        if peak_reserved_mb > getattr(self, "last_memory_reserved_mb", 0.0):
-            print(f"  - Pico reservado (nesta fase): {peak_reserved_mb:.2f} MB")
-        print(_gpu_process_table(processes, os.getpid()), end="")
-        print("--------------------------------------------------\n")
-        self.last_memory_reserved_mb = current_reserved_mb
-
-    def _register_tmp_dir(self, d: str):
-        if d and os.path.isdir(d):
-            self._tmp_dirs.add(d); print(f"[DEBUG] Registrado tmp dir: {d}")
-
-    def _register_tmp_file(self, f: str):
-        if f and os.path.exists(f):
-            self._tmp_files.add(f); print(f"[DEBUG] Registrado tmp file: {f}")
-
-    def finalize(self, keep_paths=None, extra_paths=None, clear_gpu=True):
-        print("[DEBUG] Finalize: iniciando limpeza...")
-        keep = set(keep_paths or []); extras = set(extra_paths or [])
-        removed_files = 0
-        for f in list(self._tmp_files | extras):
-            try:
-                if f not in keep and os.path.isfile(f):
-                    os.remove(f); removed_files += 1; print(f"[DEBUG] Removido arquivo tmp: {f}")
-            except Exception as e:
-                print(f"[DEBUG] Falha removendo arquivo {f}: {e}")
-            finally:
-                self._tmp_files.discard(f)
-        removed_dirs = 0
-        for d in list(self._tmp_dirs):
-            try:
-                if d not in keep and os.path.isdir(d):
-                    shutil.rmtree(d, ignore_errors=True); removed_dirs += 1; print(f"[DEBUG] Removido diretório tmp: {d}")
-            except Exception as e:
-                print(f"[DEBUG] Falha removendo diretório {d}: {e}")
-            finally:
-                self._tmp_dirs.discard(d)
-        print(f"[DEBUG] Finalize: arquivos removidos={removed_files}, dirs removidos={removed_dirs}")
-        gc.collect()
+    def _resolve_model_paths_from_cache(self):
+        """Finds the absolute paths to model files in the cache and updates the in-memory config."""
+        logging.info("Resolving model paths from Hugging Face cache...")
+        cache_dir = os.environ.get("HF_HOME")
         try:
-            if clear_gpu and torch.cuda.is_available():
-                torch.cuda.empty_cache()
-                try:
-                    torch.cuda.ipc_collect()
-                except Exception:
-                    pass
+            main_ckpt_path = hf_hub_download(repo_id=LTX_REPO_ID, filename=self.config["checkpoint_path"], cache_dir=cache_dir)
+            self.config["checkpoint_path"] = main_ckpt_path
+            logging.info(f"  -> Main checkpoint resolved to: {main_ckpt_path}")
+
+            if self.config.get("spatial_upscaler_model_path"):
+                upscaler_path = hf_hub_download(repo_id=LTX_REPO_ID, filename=self.config["spatial_upscaler_model_path"], cache_dir=cache_dir)
+                self.config["spatial_upscaler_model_path"] = upscaler_path
+                logging.info(f"  -> Spatial upscaler resolved to: {upscaler_path}")
         except Exception as e:
-            print(f"[DEBUG] Finalize: limpeza GPU falhou: {e}")
-        try:
-            self._log_gpu_memory("Após finalize")
-        except Exception as e:
-            print(f"[DEBUG] Log GPU pós-finalize falhou: {e}")
-
-    def _load_config(self):
-        base = LTX_VIDEO_REPO_DIR / "configs"
-        candidates = [
-            base / "ltxv-13b-0.9.8-dev-fp8.yaml",
-            base / "ltxv-13b-0.9.8-distilled-fp8.yaml",
-            base / "ltxv-13b-0.9.8-distilled.yaml",
-        ]
-        for cfg in candidates:
-            if cfg.exists():
-                print(f"[DEBUG] Config selecionada: {cfg}")
-                with open(cfg, "r") as file:
-                    return yaml.safe_load(file)
-        cfg = base / "ltxv-13b-0.9.8-distilled-fp8.yaml"
-        print(f"[DEBUG] Config fallback: {cfg}")
-        with open(cfg, "r") as file:
-            return yaml.safe_load(file)
-
-    def _load_models(self):
-        t0 = time.perf_counter()
-        LTX_REPO = "Lightricks/LTX-Video"
-        print("[DEBUG] Baixando checkpoint principal...")
-        distilled_model_path = hf_hub_download(
-            repo_id=LTX_REPO,
-            filename=self.config["checkpoint_path"],
-            local_dir=os.getenv("HF_HOME"),
-            cache_dir=os.getenv("HF_HOME_CACHE"),
-            token=os.getenv("HF_TOKEN"),
-        )
-        self.config["checkpoint_path"] = distilled_model_path
-        print(f"[DEBUG] Checkpoint em: {distilled_model_path}")
-
-        print("[DEBUG] Baixando upscaler espacial...")
-        spatial_upscaler_path = hf_hub_download(
-            repo_id=LTX_REPO,
-            filename=self.config["spatial_upscaler_model_path"],
-            local_dir=os.getenv("HF_HOME"),
-            cache_dir=os.getenv("HF_HOME_CACHE"),
-            token=os.getenv("HF_TOKEN")
-        )
-        self.config["spatial_upscaler_model_path"] = spatial_upscaler_path
-        print(f"[DEBUG] Upscaler em: {spatial_upscaler_path}")
-
-        print("[DEBUG] Construindo pipeline...")
-        pipeline = create_ltx_video_pipeline(
-            ckpt_path=self.config["checkpoint_path"],
-            precision=self.config["precision"],
-            text_encoder_model_name_or_path=self.config["text_encoder_model_name_or_path"],
-            sampler=self.config["sampler"],
-            device="cpu",
-            enhance_prompt=False,
-            prompt_enhancer_image_caption_model_name_or_path=self.config["prompt_enhancer_image_caption_model_name_or_path"],
-            prompt_enhancer_llm_model_name_or_path=self.config["prompt_enhancer_llm_model_name_or_path"],
-        )
-        print("[DEBUG] Pipeline pronto.")
-
-        latent_upsampler = None
-        if self.config.get("spatial_upscaler_model_path"):
-            print("[DEBUG] Construindo latent_upsampler...")
-            latent_upsampler = create_latent_upsampler(self.config["spatial_upscaler_model_path"], device="cpu")
-            print("[DEBUG] Upsampler pronto.")
-        print(f"[DEBUG] _load_models() tempo total={time.perf_counter()-t0:.3f}s")
-        return pipeline, latent_upsampler
-
-    def _promote_fp8_weights_to_bf16(self, module):
-        if not isinstance(module, torch.nn.Module):
-            print("[DEBUG] Promoção FP8→BF16 ignorada: alvo não é nn.Module.")
-            return
-        f8 = getattr(torch, "float8_e4m3fn", None)
-        if f8 is None:
-            print("[DEBUG] torch.float8_e4m3fn indisponível.")
-            return
-        p_cnt = b_cnt = 0
-        for _, p in module.named_parameters(recurse=True):
-            try:
-                if p.dtype == f8:
-                    with torch.no_grad():
-                        p.data = p.data.to(torch.bfloat16); p_cnt += 1
-            except Exception:
-                pass
-        for _, b in module.named_buffers(recurse=True):
-            try:
-                if hasattr(b, "dtype") and b.dtype == f8:
-                    b.data = b.data.to(torch.bfloat16); b_cnt += 1
-            except Exception:
-                pass
-        print(f"[DEBUG] FP8→BF16: params_promoted={p_cnt}, buffers_promoted={b_cnt}")
-
-    @torch.no_grad()
-    def _upsample_latents_internal(self, latents: torch.Tensor) -> torch.Tensor:
-        if not self.latent_upsampler:
-            raise ValueError("Latent Upsampler não está carregado.")
-        self.latent_upsampler.to(self.device)
-        self.pipeline.vae.to(self.device)
-        print(f"[DEBUG-UPSAMPLE] Shape de entrada: {tuple(latents.shape)}")
-        latents = un_normalize_latents(latents, self.pipeline.vae, vae_per_channel_normalize=True)
-        upsampled_latents = self.latent_upsampler(latents)
-        upsampled_latents = normalize_latents(upsampled_latents, self.pipeline.vae, vae_per_channel_normalize=True)
-        print(f"[DEBUG-UPSAMPLE] Shape de saída: {tuple(upsampled_latents.shape)}")
-        return upsampled_latents
+            logging.critical(f"Failed to resolve model paths. Ensure setup.py ran correctly. Error: {e}", exc_info=True)
+            sys.exit(1)
+
+    @log_function_io
+    def move_to_device(self, main_device_str: str, vae_device_str: str):
+        """Moves pipeline components to their designated target devices."""
+        target_main_device = torch.device(main_device_str)
+        target_vae_device = torch.device(vae_device_str)
+        logging.info(f"Moving LTX models -> Main Pipeline: {target_main_device}, VAE: {target_vae_device}")
+
+        self.main_device = target_main_device
+        self.pipeline.to(self.main_device)
+        self.vae_device = target_vae_device
+        self.pipeline.vae.to(self.vae_device)
+        if self.latent_upsampler: self.latent_upsampler.to(self.main_device)
+        logging.info("LTX models successfully moved to target devices.")
+
+    def move_to_cpu(self):
+        """Moves all LTX components to CPU to free VRAM for other services."""
+        self.move_to_device(main_device_str="cpu", vae_device_str="cpu")
+        if torch.cuda.is_available(): torch.cuda.empty_cache()
+
+    def finalize(self):
+        """Cleans up GPU memory after a generation task."""
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+            try: torch.cuda.ipc_collect();
+            except Exception: pass
+
+    # ==========================================================================
+    # --- LÓGICA DE NEGÓCIO: ORQUESTRADOR PÚBLICO UNIFICADO ---
+    # ==========================================================================
+
+    @log_function_io
+    def generate_low_resolution(self, prompt: str, **kwargs) -> Tuple[Optional[str], Optional[str], Optional[int]]:
+        """
+        [UNIFIED ORCHESTRATOR] Generates a low-resolution video from a prompt.
+        Handles both single-line and multi-line prompts transparently.
+        """
+        logging.info("Starting unified low-resolution generation (random seed)...")
+        used_seed = self._get_random_seed()
+        seed_everything(used_seed)
+        logging.info(f"Using randomly generated seed: {used_seed}")
+
+        prompt_list = [p.strip() for p in prompt.splitlines() if p.strip()]
+        if not prompt_list: raise ValueError("Prompt is empty or contains no valid lines.")
         
-    def _apply_precision_policy(self):
-        prec = str(self.config.get("precision", "")).lower()
-        self.runtime_autocast_dtype = torch.float32
-        print(f"[DEBUG] Aplicando política de precisão: {prec}")
-        if prec == "float8_e4m3fn":
-            self.runtime_autocast_dtype = torch.bfloat16
-            force_promote = os.getenv("LTXV_FORCE_BF16_ON_FP8", "0") == "1"
-            print(f"[DEBUG] FP8 detectado. force_promote={force_promote}")
-            if force_promote and hasattr(torch, "float8_e4m3fn"):
-                try:
-                    self._promote_fp8_weights_to_bf16(self.pipeline)
-                except Exception as e:
-                    print(f"[DEBUG] Promoção FP8→BF16 na pipeline falhou: {e}")
-                try:
-                    if self.latent_upsampler:
-                        self._promote_fp8_weights_to_bf16(self.latent_upsampler)
-                except Exception as e:
-                    print(f"[DEBUG] Promoção FP8→BF16 no upsampler falhou: {e}")
-        elif prec == "bfloat16":
-            self.runtime_autocast_dtype = torch.bfloat16
-        elif prec == "mixed_precision":
-            self.runtime_autocast_dtype = torch.float16
-        else:
-            self.runtime_autocast_dtype = torch.float32
-
-    def _prepare_conditioning_tensor(self, filepath, height, width, padding_values):
-        print(f"[DEBUG] Carregando condicionamento: {filepath}")
-        tensor = load_image_to_tensor_with_resize_and_crop(filepath, height, width)
-        tensor = torch.nn.functional.pad(tensor, padding_values)
-        out = tensor.to(self.device, dtype=self.runtime_autocast_dtype) if self.device == "cuda" else tensor.to(self.device)
-        print(f"[DEBUG] Cond shape={tuple(out.shape)} dtype={out.dtype} device={out.device}")
-        return out
-
-    def _dividir_latentes_por_tamanho(self, latents_brutos, num_latente_por_chunk: int, overlap: int = 1):
-        sum_latent = latents_brutos.shape[2]
-        chunks = []
-        if num_latente_por_chunk >= sum_latent:
-            return [latents_brutos.clone().detach()] # CORREÇÃO: Retornar uma lista e clonar
-        # CORREÇÃO: Lógica de chunking simplificada e corrigida para evitar estouro de índice
-        start = 0
-        while start < sum_latent:
-            end = min(start + num_latente_por_chunk, sum_latent)
-            # Para o overlap, pegamos um pouco do chunk anterior, exceto para o primeiro
-            overlap_start = max(0, start - overlap)
-            
-            # O chunk a ser processado vai de `overlap_start` até `end`
-            # mas o chunk "real" para junção posterior seria de `start` a `end`
-            # A lógica atual já faz um overlap simples, vamos refinar
-            effective_end = min(start + num_latente_por_chunk, sum_latent)
-            chunk = latents_brutos[:, :, start:effective_end, :, :].clone().detach()
-            
-            # Adiciona overlap no final se não for o último chunk
-            if effective_end < sum_latent:
-                overlap_end = min(effective_end + overlap, sum_latent)
-                chunk = latents_brutos[:, :, start:overlap_end, :, :].clone().detach()
+        is_narrative = len(prompt_list) > 1
+        logging.info(f"Generation mode detected: {'Narrative' if is_narrative else 'Simple'} ({len(prompt_list)} scene(s)).")
 
-            print(f"[DEBUG] Chunk: start={start}, end={chunk.shape[2]}, total_latents={sum_latent}")
-            chunks.append(chunk)
-            
-            # Avança para o próximo chunk
-            if start + num_latente_por_chunk >= sum_latent:
-                break
-            start += num_latente_por_chunk
-            
-        return chunks
-   
-    def _get_total_frames(self, video_path: str) -> int:
-        cmd = [
-            "ffprobe", "-v", "error", "-select_streams", "v:0", "-count_frames",
-            "-show_entries", "stream=nb_read_frames", "-of", "default=nokey=1:noprint_wrappers=1", video_path
-        ]
-        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
-        return int(result.stdout.strip())    
-
-    def _gerar_lista_com_transicoes(self, pasta: str, video_paths: list[str], crossfade_frames: int = 8) -> list[str]:
-        # Esta função parece complexa e propensa a erros com nomes de arquivo.
-        # Por segurança, mantendo a lógica original, mas corrigindo possíveis bugs de `shell=True`
-        # e garantindo que os arquivos existam.
-        if len(video_paths) <= 1:
-            return video_paths # Não há o que fazer
-            
-        nova_lista_intermediaria = []
-        # Primeiro, cria todos os vídeos podados
-        videos_podados = []
-        for i, base in enumerate(video_paths):
-            video_podado = os.path.join(pasta, f"podado_{i}.mp4")
-            total_frames = self._get_total_frames(base)
-            
-            start_frame = crossfade_frames if i > 0 else 0
-            end_frame = total_frames - crossfade_frames if i < len(video_paths) - 1 else total_frames
-            
-            # Pular poda se não houver frames suficientes
-            if start_frame >= end_frame:
-                continue
-
-            cmd = [
-                'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', base,
-                '-vf', f'trim=start_frame={start_frame}:end_frame={end_frame},setpts=PTS-STARTPTS',
-                '-an', video_podado
-            ]
-            subprocess.run(cmd, check=True)
-            videos_podados.append(video_podado)
-
-        # Agora, cria as transições e monta a lista final
-        lista_final = [videos_podados[0]]
-        for i in range(len(video_paths) - 1):
-            video_anterior = video_paths[i]
-            video_seguinte = video_paths[i+1]
-            
-            # Extrai fade_fim do anterior
-            fade_fim_path = os.path.join(pasta, f"fade_fim_{i}.mp4")
-            total_frames_anterior = self._get_total_frames(video_anterior)
-            cmd_fim = [
-                'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', video_anterior,
-                '-vf', f'trim=start_frame={total_frames_anterior - crossfade_frames},setpts=PTS-STARTPTS',
-                '-an', fade_fim_path
-            ]
-            subprocess.run(cmd_fim, check=True)
-
-            # Extrai fade_ini do seguinte
-            fade_ini_path = os.path.join(pasta, f"fade_ini_{i+1}.mp4")
-            cmd_ini = [
-                'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', video_seguinte,
-                '-vf', f'trim=end_frame={crossfade_frames},setpts=PTS-STARTPTS', '-an', fade_ini_path
-            ]
-            subprocess.run(cmd_ini, check=True)
-
-            # Cria a transição
-            transicao_path = os.path.join(pasta, f"transicao_{i}_{i+1}.mp4")
-            cmd_blend = [
-                'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error',
-                '-i', fade_fim_path, '-i', fade_ini_path,
-                '-filter_complex', f'[0:v][1:v]blend=all_expr=\'A*(1-T/{crossfade_frames})+B*(T/{crossfade_frames})\',format=yuv420p',
-                '-frames:v', str(crossfade_frames), transicao_path
-            ]
-            subprocess.run(cmd_blend, check=True)
-
-            lista_final.append(transicao_path)
-            lista_final.append(videos_podados[i+1])
-            
-        return lista_final
+        num_chunks = len(prompt_list)
+        total_frames = self._calculate_aligned_frames(kwargs.get("duration", 4.0))
+        frames_per_chunk = max(FRAMES_ALIGNMENT, (total_frames // num_chunks // FRAMES_ALIGNMENT) * FRAMES_ALIGNMENT)
         
-    def _concat_mp4s_no_reencode(self, mp4_list: List[str], out_path: str):
-        if not mp4_list:
-            raise ValueError("A lista de MP4s para concatenar está vazia.")
-        # Se houver apenas um vídeo, apenas o copie/mova
-        if len(mp4_list) == 1:
-            shutil.move(mp4_list[0], out_path)
-            print(f"[DEBUG] Apenas um vídeo, movido para: {out_path}")
-            return
-            
-        with tempfile.NamedTemporaryFile("w", delete=False, suffix=".txt") as f:
-            for mp4 in mp4_list:
-                f.write(f"file '{os.path.abspath(mp4)}'\n")
-            list_path = f.name
-    
-        cmd = f"ffmpeg -y -f concat -safe 0 -i {list_path} -c copy {out_path}"
-        print(f"[DEBUG] Concat: {cmd}")
-    
-        try:
-            subprocess.check_call(shlex.split(cmd))
-        finally:
-            try:
-                os.remove(list_path)
-            except Exception:
-                pass   
-   
-    def generate(
-        self,
-        prompt,
-        negative_prompt,
-        mode="text-to-video",
-        start_image_filepath=None,
-        middle_image_filepath=None,
-        middle_frame_number=None,
-        middle_image_weight=1.0,
-        end_image_filepath=None,
-        end_image_weight=1.0,
-        input_video_filepath=None,
-        height=512,
-        width=704,
-        duration=2.0,
-        frames_to_use=9, # Parâmetro não utilizado, mas mantido por consistência
-        seed=42,
-        randomize_seed=True,
-        guidance_scale=3.0,
-        improve_texture=True,
-        progress_callback=None,
-        external_decode=True, # Parâmetro não utilizado, mas mantido
-    ):
-        t_all = time.perf_counter()
-        print(f"[DEBUG] generate() begin mode={mode} improve_texture={improve_texture}")
-        if self.device == "cuda":
-            torch.cuda.empty_cache(); torch.cuda.reset_peak_memory_stats()
-        self._log_gpu_memory("Início da Geração")
-
-        if mode == "image-to-video" and not start_image_filepath:
-            raise ValueError("A imagem de início é obrigatória para o modo image-to-video")
-        used_seed = random.randint(0, 2**32 - 1) if randomize_seed else int(seed)
-        seed_everething(used_seed); print(f"[DEBUG] Seed usado: {used_seed}")
-        FPS = 24.0; MAX_NUM_FRAMES = 2570
-        target_frames_rounded = round(duration * FPS)
-        n_val = round((float(target_frames_rounded) - 1.0) / 8.0)
-        actual_num_frames = max(9, min(MAX_NUM_FRAMES, int(n_val * 8 + 1)))
-        height_padded = ((height - 1) // 8 + 1) * 8
-        width_padded = ((width - 1) // 8 + 1) * 8
-        padding_values = calculate_padding(height, width, height_padded, width_padded)
-        generator = torch.Generator(device=self.device).manual_seed(used_seed)
+        # Overlap must be N*8+1 frames. 9 is the smallest practical value.
+        overlap_frames = 9 if is_narrative else 0
+        if is_narrative:
+            logging.info(f"Narrative mode: Using overlap of {overlap_frames} frames between chunks.")
         
-        conditioning_items = []
-        if mode == "image-to-video":
-            start_tensor = self._prepare_conditioning_tensor(start_image_filepath, height, width, padding_values)
-            conditioning_items.append(ConditioningItem(start_tensor, 0, 1.0))
-            if middle_image_filepath and middle_frame_number is not None:
-                middle_tensor = self._prepare_conditioning_tensor(middle_image_filepath, height, width, padding_values)
-                safe_middle_frame = max(0, min(int(middle_frame_number), actual_num_frames - 1))
-                conditioning_items.append(ConditioningItem(middle_tensor, safe_middle_frame, float(middle_image_weight)))
-            if end_image_filepath:
-                end_tensor = self._prepare_conditioning_tensor(end_image_filepath, height, width, padding_values)
-                last_frame_index = actual_num_frames - 1
-                conditioning_items.append(ConditioningItem(end_tensor, last_frame_index, float(end_image_weight)))
-            print(f"[DEBUG] Conditioning items: {len(conditioning_items)}")
-
-        call_kwargs = {
-            "prompt": prompt, "negative_prompt": negative_prompt, "height": height_padded, "width": width_padded,
-            "num_frames": actual_num_frames, "frame_rate": int(FPS), "generator": generator, "output_type": "latent",
-            "conditioning_items": conditioning_items if conditioning_items else None, "media_items": None,
-            "decode_timestep": self.config["decode_timestep"], "decode_noise_scale": self.config["decode_noise_scale"],
-            "stochastic_sampling": self.config["stochastic_sampling"], "image_cond_noise_scale": 0.01, "is_video": True,
-            "vae_per_channel_normalize": True, "mixed_precision": (self.config["precision"] == "mixed_precision"),
-            "offload_to_cpu": False, "enhance_prompt": False, "skip_layer_strategy": SkipLayerStrategy.AttentionValues,
-        }
+        temp_latent_paths = []
+        overlap_condition_item = None
         
-        # CORREÇÃO: Inicialização de listas
-        latents_list = []
-        temp_dir = tempfile.mkdtemp(prefix="ltxv_"); self._register_tmp_dir(temp_dir)
-        results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
-            
         try:
-            if improve_texture:
-                ctx = torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype) if self.device == "cuda" else contextlib.nullcontext()
-                with ctx:
-                    if not self.latent_upsampler:
-                        raise ValueError("Upscaler espacial não carregado, mas 'improve_texture' está ativo.")
-                    
-                    print("\n--- INICIANDO ETAPA 1: GERAÇÃO BASE (FIRST PASS) ---")
-                    t_pass1 = time.perf_counter()
-                    first_pass_config = self.config.get("first_pass", {}).copy()
-                    first_pass_config.pop("num_inference_steps", None)
-                    downscale_factor = self.config.get("downscale_factor", 0.6666666)
-                    vae_scale_factor = self.pipeline.vae_scale_factor
-                    x_width = int(width_padded * downscale_factor)
-                    downscaled_width = x_width - (x_width % vae_scale_factor)
-                    x_height = int(height_padded * downscale_factor)
-                    downscaled_height = x_height - (x_height % vae_scale_factor)
-                    print(f"[DEBUG] First Pass Dims: Original Pad ({width_padded}x{height_padded}) -> Downscaled ({downscaled_width}x{downscaled_height})")
-                    
-                    first_pass_kwargs = call_kwargs.copy()
-                    first_pass_kwargs.update({
-                        "output_type": "latent", "width": downscaled_width, "height": downscaled_height,
-                        "guidance_scale": float(guidance_scale), **first_pass_config
-                    })
-                    
-                    print(f"[DEBUG] First Pass: Gerando em {downscaled_width}x{downscaled_height}...")
-                    # CORREÇÃO: Usar self.pipeline, não a variável deletada 'pipeline'
-                    latents = self.pipeline(**first_pass_kwargs).images
-                    log_tensor_info(latents, "Latentes Base (First Pass)")
-                    print(f"[DEBUG] First Pass concluída em {time.perf_counter() - t_pass1:.2f}s")
-                    
-                with ctx:
-                    print("\n--- INICIANDO ETAPA 2: UPSCALE DOS LATENTES ---")
-                    t_upscale = time.perf_counter()
-                    upsampled_latents = self._upsample_latents_internal(latents)
-                    upsampled_latents = adain_filter_latent(latents=upsampled_latents, reference_latents=latents)
-                    print(f"[DEBUG] Upscale de Latentes concluído em {time.perf_counter() - t_upscale:.2f}s")
-                    
-                    # CORREÇÃO: Manter latentes originais para AdaIN e passar latentes com upscale para o second pass
-                    reference_latents_cpu = latents.detach().to("cpu", non_blocking=True)
-                    latents_to_refine = upsampled_latents
-                    del upsampled_latents; del latents; gc.collect(); torch.cuda.empty_cache()
+            for i, chunk_prompt in enumerate(prompt_list):
+                logging.info(f"Processing scene {i+1}/{num_chunks}: '{chunk_prompt[:50]}...'")
                 
-                # CORREÇÃO: Lógica de chunking para o second pass
-                latents_parts = self._dividir_latentes_por_tamanho(latents_to_refine, 32, 8) # Exemplo: chunks de 32 frames com 8 de overlap
-                del latents_to_refine
-
-                with ctx:
-                    for i, latents_chunk in enumerate(latents_parts):
-                        print(f"\n--- INICIANDO ETAPA 3.{i+1}: REFINAMENTO DE TEXTURA (SECOND PASS) ---")
-                        # CORREÇÃO: AdaIN precisa de latents de referência com mesmo H/W, o que não é o caso aqui.
-                        # Vamos aplicar AdaIN com o próprio chunk para normalização, ou pular. Pulando por simplicidade.
-                        
-                        second_pass_config = self.config.get("second_pass", {}).copy()
-                        second_pass_config.pop("num_inference_steps", None)
-                        
-                        # O tamanho do second pass deve ser o tamanho do latente de entrada (após upscale)
-                        second_pass_height, second_pass_width = latents_chunk.shape[3] * 8, latents_chunk.shape[4] * 8
-
-                        print(f"[DEBUG] Second Pass Dims: Target ({second_pass_width}x{second_pass_height})")
-                        t_pass2 = time.perf_counter()
-                        second_pass_kwargs = call_kwargs.copy()
-                        second_pass_kwargs.update({
-                           "output_type": "latent", "width": second_pass_width, "height": second_pass_height,
-                           "latents": latents_chunk.to(self.device), # Mover chunk para GPU
-                           "guidance_scale": float(guidance_scale),
-                           "num_frames": latents_chunk.shape[2], # Usar o número de frames do chunk
-                           **second_pass_config
-                        })
-                        print(f"[DEBUG] Second Pass: Refinando chunk {i+1}/{len(latents_parts)}...")
-                        final_latents = self.pipeline(**second_pass_kwargs).images
-                        log_tensor_info(final_latents, "Latentes Finais (Pós-Second Pass)")
-                        print(f"[DEBUG] Second part Pass concluída em {time.perf_counter() - t_pass2:.2f}s")
-                        latents_cpu = final_latents.detach().to("cpu", non_blocking=True)
-                        latents_list.append(latents_cpu)
-                        del final_latents; del latents_chunk; gc.collect(); torch.cuda.empty_cache()
-            else: 
-                ctx = torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype) if self.device == "cuda" else contextlib.nullcontext()
-                with ctx:    
-                    print("\n--- INICIANDO GERAÇÃO DE ETAPA ÚNICA ---")
-                    t_single = time.perf_counter()
-                    single_pass_call_kwargs = call_kwargs.copy()
-                    # CORREÇÃO: `pipeline_instance` não existe, usar `self.pipeline`.
-                    latents_single_pass = self.pipeline(**single_pass_call_kwargs).images
-                    log_tensor_info(latents_single_pass, "Latentes Finais (Etapa Única)")
-                    print(f"[DEBUG] Etapa única concluída em {time.perf_counter() - t_single:.2f}s")
-                    latents_cpu = latents_single_pass.detach().to("cpu", non_blocking=True)
-                    latents_list.append(latents_cpu) # CORREÇÃO: aqui deve ser latents_cpu, não latents_single_pass
-                    del latents_single_pass; gc.collect(); torch.cuda.empty_cache()
-
-            # --- ETAPA FINAL: DECODIFICAÇÃO E CODIFICAÇÃO MP4 ---
-            print("\n--- INICIANDO ETAPA FINAL: DECODIFICAÇÃO E MONTAGEM ---")
-            partes_mp4 = []
-            for i, latents in enumerate(latents_list):
-                print(f"[DEBUG] Decodificando partição {i+1}/{len(latents_list)}: {tuple(latents.shape)}")
-                output_video_path = os.path.join(temp_dir, f"output_{used_seed}_{i}.mp4")
+                if i < num_chunks - 1:
+                    current_frames_base = frames_per_chunk
+                else: # Last chunk takes all remaining frames
+                    processed_frames_base = (num_chunks - 1) * frames_per_chunk
+                    current_frames_base = total_frames - processed_frames_base
                 
-                pixel_tensor = vae_manager_singleton.decode(
-                    latents.to(self.device, non_blocking=True),
-                    decode_timestep=float(self.config.get("decode_timestep", 0.05))
-                )
-                log_tensor_info(pixel_tensor, "Pixel tensor (VAE saída)")
+                current_frames = current_frames_base + (overlap_frames if i > 0 else 0)
+                # Ensure final frame count for generation is N*8+1
+                current_frames = self._align(current_frames, alignment_rule='n*8+1')
+
+                current_conditions = kwargs.get("initial_conditions", []) if i == 0 else []
+                if overlap_condition_item: current_conditions.append(overlap_condition_item)
 
-                video_encode_tool_singleton.save_video_from_tensor(
-                   pixel_tensor, output_video_path, fps=call_kwargs["frame_rate"], progress_callback=progress_callback
+                chunk_latents = self._generate_single_chunk_low(
+                    prompt=chunk_prompt, num_frames=current_frames, seed=used_seed + i,
+                    conditioning_items=current_conditions, **kwargs
                 )
-                partes_mp4.append(output_video_path)
-                del pixel_tensor; del latents; gc.collect(); torch.cuda.empty_cache()
-
-            final_vid = os.path.join(results_dir, f"final_video_{used_seed}.mp4")
-            if len(partes_mp4) > 1:
-                # A função _gerar_lista_com_transicoes é complexa, usando uma concatenação direta como fallback robusto.
-                # Para usar a transição, a lógica de overlap na divisão de latentes precisa ser perfeita.
-                print("[DEBUG] Múltiplas partes geradas, concatenando...")
-                partes_mp4_fade = self._gerar_lista_com_transicoes(pasta=temp_dir, video_paths=partes_mp4, crossfade_frames=8)
-                self._concat_mp4s_no_reencode(partes_mp4_fade, final_vid)
-            else:
-                shutil.move(partes_mp4[0], final_vid)
+                if chunk_latents is None: raise RuntimeError(f"Failed to generate latents for scene {i+1}.")
 
-            self._log_gpu_memory("Fim da Geração")
-            return final_vid, used_seed
+                if is_narrative and i < num_chunks - 1:
+                    overlap_latents = chunk_latents[:, :, -overlap_frames:, :, :].clone()
+                    overlap_condition_item = ConditioningItem(media_item=overlap_latents, media_frame_number=0, conditioning_strength=1.0)
+                
+                if i > 0:
+                    chunk_latents = chunk_latents[:, :, overlap_frames:, :, :]
+                
+                chunk_path = RESULTS_DIR / f"temp_chunk_{i}_{used_seed}.pt"
+                torch.save(chunk_latents.cpu(), chunk_path)
+                temp_latent_paths.append(chunk_path)
             
+            base_filename = "narrative_video" if is_narrative else "single_video"
+            return self._finalize_generation(temp_latent_paths, base_filename, used_seed)
         except Exception as e:
-            print("[DEBUG] EXCEÇÃO NA GERAÇÃO:")
-            print("".join(traceback.format_exception(type(e), e, e.__traceback__)))
-            raise
-        
+            logging.error(f"Error during unified generation: {e}", exc_info=True)
+            return None, None, None
         finally:
-            gc.collect()
-            if torch.cuda.is_available():
-                torch.cuda.empty_cache()
-                torch.cuda.ipc_collect()
-            self.finalize(keep_paths=[]) # O resultado final já foi movido
-
-print("Criando instância do VideoService. O carregamento do modelo começará agora...")
-video_generation_service = VideoService()
+            for path in temp_latent_paths:
+                if path.exists(): path.unlink()
+            self.finalize()
+
+    # ==========================================================================
+    # --- UNIDADES DE TRABALHO E HELPERS INTERNOS ---
+    # ==========================================================================
+
+    # --- NOVA FUNÇÃO DE LOG DEDICADA ---
+    def _log_conditioning_items(self, items: List[ConditioningItem]):
+        """
+        Logs detailed information about a list of ConditioningItem objects.
+        This is a dedicated debug helper function.
+        """
+        # Só imprime o log se o nível de logging for DEBUG
+        if logging.getLogger().isEnabledFor(logging.INFO):
+            log_str = ["\n" + "="*25 + " INFO: Conditioning Items " + "="*25]
+            if not items:
+                log_str.append("  -> Lista de conditioning_items está vazia.")
+            else:
+                for i, item in enumerate(items):
+                    if hasattr(item, 'media_item') and isinstance(item.media_item, torch.Tensor):
+                        t = item.media_item
+                        log_str.append(
+                            f"  -> Item [{i}]: "
+                            f"Tensor(shape={list(t.shape)}, "
+                            f"device='{t.device}', "
+                            f"dtype={t.dtype}), "
+                            f"Target Frame = {item.media_frame_number}, "
+                            f"Strength = {item.conditioning_strength:.2f}"
+                        )
+                    else:
+                        log_str.append(f"  -> Item [{i}]: Não contém um tensor válido.")
+            log_str.append("="*75 + "\n")
+            
+            # Usa o logger de debug para imprimir a mensagem completa
+            logging.info("\n".join(log_str))
+
+
+    @log_function_io
+    def _generate_single_chunk_low(self, **kwargs) -> Optional[torch.Tensor]:
+        """[WORKER] Calls the pipeline to generate a single chunk of latents."""
+        height_padded, width_padded = (self._align(d) for d in (kwargs['height'], kwargs['width']))
+        downscale_factor = self.config.get("downscale_factor", 0.6666666)
+        vae_scale_factor = self.pipeline.vae_scale_factor
+        downscaled_height = self._align(int(height_padded * downscale_factor), vae_scale_factor)
+        downscaled_width = self._align(int(width_padded * downscale_factor), vae_scale_factor)
+
+        # 1. Começa com a configuração padrão
+        first_pass_config = self.config.get("first_pass", {}).copy()
+        
+        # 2. Aplica os overrides da UI, se existirem
+        if kwargs.get("ltx_configs_override"):
+            self._apply_ui_overrides(first_pass_config, kwargs.get("ltx_configs_override"))
+
+        # 3. Monta o dicionário de argumentos SEM conditioning_items primeiro
+        pipeline_kwargs = {
+            "prompt": kwargs['prompt'],
+            "negative_prompt": kwargs['negative_prompt'],
+            "height": downscaled_height,
+            "width": downscaled_width,
+            "num_frames": kwargs['num_frames'],
+            "frame_rate": int(DEFAULT_FPS),
+            "generator": torch.Generator(device=self.main_device).manual_seed(kwargs['seed']),
+            "output_type": "latent",
+            #"conditioning_items": conditioning_items if conditioning_items else None,
+            "media_items": None,
+            "decode_timestep": self.config["decode_timestep"],
+            "decode_noise_scale": self.config["decode_noise_scale"],
+            "stochastic_sampling": self.config["stochastic_sampling"],
+            "image_cond_noise_scale": 0.01,
+            "is_video": True,
+            "vae_per_channel_normalize": True,
+            "mixed_precision": (self.config["precision"] == "mixed_precision"),
+            "offload_to_cpu": False,
+            "enhance_prompt": False,
+            #"skip_layer_strategy": SkipLayerStrategy.AttentionValues,
+            **first_pass_config
+        }
+        
+        # --- Bloco de Logging para Depuração ---
+        # 4. Loga os argumentos do pipeline (sem os tensores de condição)
+        logging.info(f"\n[Info] Pipeline Arguments (BASE):\n {json.dumps(pipeline_kwargs, indent=2, default=str)}\n")
+        
+        # Loga os conditioning_items separadamente com a nossa função helper
+        conditioning_items_list = kwargs.get('conditioning_items')
+        self._log_conditioning_items(conditioning_items_list)
+        # --- Fim do Bloco de Logging ---
+
+        # 5. Adiciona os conditioning_items ao dicionário
+        pipeline_kwargs['conditioning_items'] = conditioning_items_list
+        
+        # 6. Executa o pipeline com o dicionário completo
+        with torch.autocast(device_type=self.main_device.type, dtype=self.runtime_autocast_dtype, enabled="cuda" in self.main_device.type):
+            latents_raw = self.pipeline(**pipeline_kwargs).images
+        
+        return latents_raw.to(self.main_device)
+
+    
+    @log_function_io
+    def _finalize_generation(self, temp_latent_paths: List[Path], base_filename: str, seed: int) -> Tuple[str, str, int]:
+        """Consolidates latents, decodes them to video, and saves final artifacts."""
+        logging.info("Finalizing generation: decoding latents to video.")
+        all_tensors_cpu = [torch.load(p) for p in temp_latent_paths]
+        final_latents = torch.cat(all_tensors_cpu, dim=2)
+        
+        final_latents_path = RESULTS_DIR / f"latents_{base_filename}_{seed}.pt"
+        torch.save(final_latents, final_latents_path)
+        logging.info(f"Final latents saved to: {final_latents_path}")
+        
+        pixel_tensor = vae_manager_singleton.decode(
+            final_latents, decode_timestep=float(self.config.get("decode_timestep", 0.05))
+        )
+        video_path = self._save_and_log_video(pixel_tensor, f"{base_filename}_{seed}")
+        return str(video_path), str(final_latents_path), seed
+
+    @log_function_io
+    def prepare_condition_items(self, items_list: List, height: int, width: int, num_frames: int) -> List[ConditioningItem]:
+        """[UNIFIED] Prepares ConditioningItems from a mixed list of file paths and tensors."""
+        if not items_list: return []
+        height_padded, width_padded = self._align(height), self._align(width)
+        padding_values = calculate_padding(height, width, height_padded, width_padded)
+        
+        conditioning_items = []
+        for media_item, frame, weight in items_list:
+            if isinstance(media_item, str):
+                tensor = load_image_to_tensor_with_resize_and_crop(media_item, height, width)
+                tensor = torch.nn.functional.pad(tensor, padding_values)
+                tensor = tensor.to(self.main_device, dtype=self.runtime_autocast_dtype)
+            elif isinstance(media_item, torch.Tensor):
+                tensor = media_item.to(self.main_device, dtype=self.runtime_autocast_dtype)
+            else:
+                logging.warning(f"Unknown conditioning media type: {type(media_item)}. Skipping.")
+                continue
+            
+            safe_frame = max(0, min(int(frame), num_frames - 1))
+            conditioning_items.append(ConditioningItem(tensor, safe_frame, float(weight)))
+        return conditioning_items
+
+    def _apply_ui_overrides(self, config_dict: Dict, overrides: Dict):
+        """Applies advanced settings from the UI to a config dictionary."""
+        # Override step counts
+        for key in ["num_inference_steps", "skip_initial_inference_steps", "skip_final_inference_steps"]:
+            ui_value = overrides.get(key)
+            if ui_value and ui_value > 0:
+                config_dict[key] = ui_value
+                logging.info(f"Override: '{key}' set to {ui_value} by UI.")
+        
+    def _save_and_log_video(self, pixel_tensor: torch.Tensor, base_filename: str) -> Path:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = os.path.join(temp_dir, f"{base_filename}.mp4")
+            video_encode_tool_singleton.save_video_from_tensor(pixel_tensor, temp_path, fps=DEFAULT_FPS)
+            final_path = RESULTS_DIR / f"{base_filename}.mp4"
+            shutil.move(temp_path, final_path)
+            logging.info(f"Video saved successfully to: {final_path}")
+            return final_path
+    
+    def _apply_precision_policy(self):
+        precision = str(self.config.get("precision", "bfloat16")).lower()
+        if precision in ["float8_e4m3fn", "bfloat16"]: self.runtime_autocast_dtype = torch.bfloat16
+        elif precision == "mixed_precision": self.runtime_autocast_dtype = torch.float16
+        else: self.runtime_autocast_dtype = torch.float32
+        logging.info(f"Runtime precision policy set for autocast: {self.runtime_autocast_dtype}")
+
+    def _align(self, dim: int, alignment: int = FRAMES_ALIGNMENT, alignment_rule: str = 'default') -> int:
+        """Aligns a dimension to the nearest multiple of `alignment`."""
+        if alignment_rule == 'n*8+1':
+             return ((dim - 1) // alignment) * alignment + 1
+        return ((dim - 1) // alignment + 1) * alignment
+    
+    def _calculate_aligned_frames(self, duration_s: float, min_frames: int = 1) -> int:
+        num_frames = int(round(duration_s * DEFAULT_FPS))
+        # Para a duração total, sempre arredondamos para cima para o múltiplo de 8 mais próximo
+        aligned_frames = self._align(num_frames, alignment=FRAMES_ALIGNMENT)
+        return max(aligned_frames, min_frames)
+
+    def _get_random_seed(self) -> int:
+        """Always generates and returns a new random seed."""
+        return random.randint(0, 2**32 - 1)
+
+# ==============================================================================
+# --- INSTANCIAÇÃO SINGLETON ---
+# ==============================================================================
+
+video_generation_service = VideoService()
+logging.info("Global VideoService orchestrator instance created successfully.")