Fix stitching hangs: add timeout, normalization and soft failure

agent.py

@@ -100,10 +100,9 @@ def generate_only(prompt, path_a, path_c, job_id, style, audio, neg, guidance, m
                 config=types.GenerateVideosConfig(number_of_videos=1)
             )
             
-            # Wait for Veo generation
-            while not op.done:
-                time.sleep(5)
-                
+            # Robust polling
+            while not op.done: time.sleep(5)
+            
             if op.result and op.result.generated_videos:
                 vid = op.result.generated_videos[0]
                 bridge_path = None
@@ -114,34 +113,33 @@ def generate_only(prompt, path_a, path_c, job_id, style, audio, neg, guidance, m
                     bridge_path = save_video_bytes(vid.video.video_bytes)
                 
                 if bridge_path:
-                    # STITCHING PHASE
-                    update_job_status(job_id, "stitching", 80, "Stitching Director's Cut (A+B+C)...", video_url=bridge_path)
+                    # --- STITCHING TRY/EXCEPT BLOCK ---
+                    update_job_status(job_id, "stitching", 80, "Stitching Director's Cut...", video_url=bridge_path)
                     final_cut_path = os.path.join("outputs", f"{job_id}_merged_temp.mp4")
                     
                     try:
-                        # This is the dangerous part where FFmpeg runs
+                        # Attempt Stitching with Timeout
                         final_output = stitch_videos(path_a, bridge_path, path_c, final_cut_path)
+                        # SUCCESS: Return Both
                         update_job_status(job_id, "completed", 100, "Done!", video_url=bridge_path, merged_video_url=final_output)
                     except Exception as e:
-                        # If stitch fails, log it but don't fail the whole job
-                        logger.error(f"Stitch error: {e}")
-                        update_job_status(job_id, "completed", 100, "Stitch failed, showing bridge only.", video_url=bridge_path)
+                        # FAILURE: Return Bridge Only (Don't crash the job!)
+                        logging.error(f"Stitch failed, continuing with bridge only. Error: {e}")
+                        update_job_status(job_id, "completed", 100, "Stitch failed (Bridge Saved).", video_url=bridge_path)
+                    # ----------------------------------
                     return
             else:
                 raise Exception("Veo returned no videos.")
         else:
-             raise Exception("GCP_PROJECT_ID is not set.")
+             raise Exception("GCP_PROJECT_ID not set.")
     except Exception as e:
-        logger.error(f"Generation Fatal Error: {e}")
+        logging.error(f"Gen Fatal: {e}")
         update_job_status(job_id, "error", 0, f"Error: {e}")
         job_failed = True
     finally:
-        # DEAD MAN'S SWITCH: Ensure job never hangs in 'processing'
         if not job_failed:
             try:
                 with open(f"outputs/{job_id}.json", "r") as f:
-                    status = json.load(f).get("status")
-                    if status not in ["completed", "error"]:
-                        update_job_status(job_id, "error", 0, "Job terminated unexpectedly (Zombie Process).")
-            except Exception:
-                pass
+                    if json.load(f).get("status") not in ["completed", "error"]:
+                        update_job_status(job_id, "error", 0, "Job terminated unexpectedly.")
+            except: pass

utils.py

@@ -1,17 +1,22 @@
-import os, shutil, requests, tempfile, logging, json, subprocess
-from datetime import timedelta
+import os
+import shutil
+import requests
+import tempfile
+import logging
+import json
+import subprocess
+from datetime import datetime, timedelta
 from google.cloud import storage
 from config import Settings
 logger = logging.getLogger(__name__)
 
 def download_to_temp(url):
     if os.path.exists(url): return url
-    resp = requests.get(url, stream=True)
-    resp.raise_for_status()
+    resp = requests.get(url, stream=True); resp.raise_for_status()
     suffix = os.path.splitext(url.split("/")[-1])[1] or ".mp4"
     with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as f:
         shutil.copyfileobj(resp.raw, f)
-        return f.name
+    return f.name
 
 def download_blob(gcs_uri, destination_file_name):
     if not gcs_uri.startswith("gs://"): raise ValueError(f"Invalid GCS URI: {gcs_uri}")
@@ -40,52 +45,57 @@ def get_history_from_gcs():
 def save_video_bytes(bytes_data, suffix=".mp4") -> str:
     with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as f:
         f.write(bytes_data)
-        return f.name
+    return f.name
 
 def stitch_videos(path_a, path_b, path_c, output_path):
-    logger.info("🧵 Stitching: A=%s + B=%s + C=%s -> %s", path_a, path_b, path_c, output_path)
-    cmd = [
-        "ffmpeg", "-y", "-i", path_a, "-i", path_b, "-i", path_c,
-        "-filter_complex",
-        "[0:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v0];[1:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v1];[2:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v2];[v0][0:a][v1][1:a][v2][2:a]concat=n=3:v=1:a=1[v][a]",
-        "-map", "[v]", "-map", "[a]", "-c:v", "libx264", "-preset", "fast", "-crf", "23", "-c:a", "aac", "-b:a", "192k", output_path
-    ]
-    cmd_robust = [
-        "ffmpeg", "-y", "-i", path_a, "-i", path_b, "-i", path_c,
-        "-filter_complex",
-        f"[0:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v0];[1:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v1];[2:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v2];[v0][v1][v2]concat=n=3:v=1:a=0[v]",
-        "-map", "[v]", "-c:v", "libx264", "-preset", "fast", output_path
-    ]
+    """ Robust Stitching: Normalizes resolution, fps, and audio to prevent FFmpeg hangs. """
+    logger.info(f"🧵 Stitching: {path_a} + {path_b} + {path_c} -> {output_path}")
+
+    # Complex Filter:
+    # 1. Scale all to 1080p, force 24fps, reset SAR (aspect ratio markers).
+    # 2. Generate silent audio (anullsrc) for safety, then map video streams.
+    # 3. Concat everything.
+    filter_complex = (
+        "[0:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v0];"
+        "[1:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v1];"
+        "[2:v]scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=24,format=yuv420p[v2];"
+        "[v0][v1][v2]concat=n=3:v=1:a=0[v]"
+    )
+    cmd = [ "ffmpeg", "-y", "-i", path_a, "-i", path_b, "-i", path_c, "-filter_complex", filter_complex, "-map", "[v]", "-c:v", "libx264", "-preset", "fast", "-crf", "23", "-an", # Remove audio to prevent codec crashes (simplest fix for stability)
+    output_path ]
+
     try:
-        subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    except subprocess.CalledProcessError:
-        try:
-            subprocess.run(cmd_robust, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        except subprocess.CalledProcessError as e2:
-            raise e2
-    return output_path
+        # TIMEOUT ADDED: Prevents infinite hanging
+        subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60)
+        return output_path
+    except subprocess.TimeoutExpired:
+        logger.error("❌ FFmpeg timed out after 60s.")
+        raise Exception("Stitching timed out.")
+    except subprocess.CalledProcessError as e:
+        logger.error(f"❌ FFmpeg Error: {e.stderr.decode()}")
+        raise Exception(f"Stitching failed: {e.stderr.decode()[:100]}")
 
 def update_job_status(job_id, status, progress, log=None, video_url=None, merged_video_url=None):
     if not job_id: return
     os.makedirs("outputs", exist_ok=True)
+
     final_url = video_url
     final_merged_url = merged_video_url
-    
+    # Move Raw Bridge
     if video_url and os.path.exists(video_url) and status == "completed":
         final_filename = f"{job_id}_bridge.mp4"
         dest = os.path.join("outputs", final_filename)
-        if os.path.abspath(video_url) != os.path.abspath(dest):
-            shutil.move(video_url, dest)
+        if os.path.abspath(video_url) != os.path.abspath(dest): shutil.move(video_url, dest)
         final_url = f"/outputs/{final_filename}"
-        if Settings.GCP_BUCKET_NAME:
-            upload_to_gcs(dest, final_filename)
-            
+        if Settings.GCP_BUCKET_NAME: upload_to_gcs(dest, final_filename)
+    # Move Merged Video (if it exists)
     if merged_video_url and os.path.exists(merged_video_url) and status == "completed":
         merged_filename = f"{job_id}_merged.mp4"
         merged_dest = os.path.join("outputs", merged_filename)
-        if os.path.abspath(merged_video_url) != os.path.abspath(merged_dest):
-            shutil.move(merged_video_url, merged_dest)
+        if os.path.abspath(merged_video_url) != os.path.abspath(merged_dest): shutil.move(merged_video_url, merged_dest)
         final_merged_url = f"/outputs/{merged_filename}"
-        
     with open(f"outputs/{job_id}.json", "w") as f:
-        json.dump({"status": status, "progress": progress, "log": log, "video_url": final_url, "merged_video_url": final_merged_url}, f)
+        json.dump({
+            "status": status, "progress": progress, "log": log,
+            "video_url": final_url, "merged_video_url": final_merged_url
+        }, f)