feat: phase 1 & 2 - async architecture, refactoring, and status polling

agent.py

@@ -1,33 +1,21 @@
 import os
 import time
-import shutil
-import requests
-import tempfile
 import logging
 import json
 from typing import TypedDict, Optional
 from langgraph.graph import StateGraph, END
 
-# Unified SDK for both Analyst (Gemini) and Generator (Veo)
+# Import unified SDK
 from google import genai
 from google.genai import types
-from google.cloud import storage
 
+# Import other clients
 from groq import Groq
 from gradio_client import Client, handle_file
-from dotenv import load_dotenv
 
-# --- AUTH SETUP FOR HUGGING FACE ---
-if "GCP_CREDENTIALS_JSON" in os.environ:
-    # logger isn't configured yet, so use print
-    print("🔐 Found GCP Credentials Secret. Setting up auth...")
-    creds_path = "gcp_credentials.json"
-    with open(creds_path, "w") as f:
-        f.write(os.environ["GCP_CREDENTIALS_JSON"])
-    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = creds_path
-
-# Load environment variables
-load_dotenv()
+# Import refactored modules
+from config import Settings
+from utils import download_to_temp, download_blob, save_video_bytes, update_job_status
 
 # Configure Logging
 logging.basicConfig(level=logging.INFO)
@@ -35,6 +23,7 @@ logger = logging.getLogger(__name__)
 
 # State Definition
 class ContinuityState(TypedDict):
+    job_id: Optional[str] # Added job_id
     video_a_url: str
     video_c_url: str
     user_notes: Optional[str]
@@ -44,36 +33,12 @@ class ContinuityState(TypedDict):
     video_a_local_path: Optional[str]
     video_c_local_path: Optional[str]
 
-# --- HELPER FUNCTIONS ---
-def download_to_temp(url):
-    logger.info(f"Downloading: {url}")
-    if os.path.exists(url):
-        return url
-
-    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
-
-def download_blob(gcs_uri, destination_file_name):
-    """Downloads a blob from the bucket."""
-    if not gcs_uri.startswith("gs://"):
-        raise ValueError(f"Invalid GCS URI: {gcs_uri}")
-
-    parts = gcs_uri[5:].split("/", 1)
-    bucket_name = parts[0]
-    source_blob_name = parts[1]
-    storage_client = storage.Client()
-    bucket = storage_client.bucket(bucket_name)
-    blob = bucket.blob(source_blob_name)
-    blob.download_to_filename(destination_file_name)
-    logger.info(f"Downloaded storage object {gcs_uri} to local file {destination_file_name}.")
-
 # --- NODE 1: ANALYST ---
 def analyze_videos(state: ContinuityState) -> dict:
     logger.info("--- 🧐 Analyst Node (Director) ---")
+    job_id = state.get("job_id")
+    
+    update_job_status(job_id, "analyzing", 10, "Director starting analysis...")
 
     video_a_url = state['video_a_url']
     video_c_url = state['video_c_url']
@@ -88,16 +53,23 @@ def analyze_videos(state: ContinuityState) -> dict:
         if not path_c:
             path_c = download_to_temp(video_c_url)
     except Exception as e:
-        logger.error(f"Download failed: {e}")
+        error_msg = f"Download failed: {e}"
+        logger.error(error_msg)
+        update_job_status(job_id, "error", 0, error_msg)
         return {"scene_analysis": "Error downloading", "veo_prompt": "Smooth cinematic transition"}
 
+    update_job_status(job_id, "analyzing", 20, "Director analyzing motion and lighting...")
+
     # 2. Try Gemini 2.0 (With Retry and Wait Loop)
-    client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
+    client = genai.Client(api_key=Settings.GOOGLE_API_KEY)
     transition_prompt = None
     retries = 3
     for attempt in range(retries):
         try:
             logger.info(f"Uploading videos to Gemini... (Attempt {attempt+1})")
+            if attempt > 0:
+                 update_job_status(job_id, "analyzing", 20, f"Retrying analysis (Attempt {attempt+1})...")
+
             file_a = client.files.upload(file=path_a)
             file_c = client.files.upload(file=path_c)
 
@@ -123,6 +95,8 @@ def analyze_videos(state: ContinuityState) -> dict:
             """
             
             logger.info("Generating transition prompt...")
+            update_job_status(job_id, "analyzing", 30, "Director writing scene transition...")
+            
             response = client.models.generate_content(
                 model="gemini-2.0-flash-exp", 
                 contents=[prompt_text, file_a, file_c]
@@ -134,6 +108,7 @@ def analyze_videos(state: ContinuityState) -> dict:
             if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e):
                 wait = 30 * (attempt + 1)
                 logger.warning(f"⚠️ Gemini Quota 429. Retrying in {wait}s...")
+                update_job_status(job_id, "analyzing", 25, f"High traffic, retrying in {wait}s...")
                 time.sleep(wait)
             else:
                 logger.error(f"⚠️ Gemini Error: {e}")
@@ -142,8 +117,9 @@ def analyze_videos(state: ContinuityState) -> dict:
     # 3. Fallback: Groq (Updated Model)
     if not transition_prompt:
         logger.info("Switching to Llama 3.2 (Groq) Fallback...")
+        update_job_status(job_id, "analyzing", 35, "Using backup director (Llama 3.2)...")
         try:
-            groq_client = Groq(api_key=os.environ["GROQ_API_KEY"])
+            groq_client = Groq(api_key=Settings.GROQ_API_KEY)
             fallback_prompt = "Create a smooth, cinematic visual transition that bridges two scenes."
             completion = groq_client.chat.completions.create(
                 model="llama-3.2-90b-vision-preview",
@@ -154,6 +130,8 @@ def analyze_videos(state: ContinuityState) -> dict:
             logger.error(f"❌ Groq also failed: {e}")
             transition_prompt = "Smooth cinematic transition with motion blur matching the scenes."
             
+    update_job_status(job_id, "generating", 40, "Director prompt ready. Starting generation...")
+    
     return {
         "scene_analysis": transition_prompt, 
         "veo_prompt": transition_prompt,
@@ -164,19 +142,24 @@ def analyze_videos(state: ContinuityState) -> dict:
 # --- NODE 2: GENERATOR ---
 def generate_video(state: ContinuityState) -> dict:
     logger.info("--- 🎥 Generator Node ---")
-
+    job_id = state.get("job_id")
+    
     prompt = state.get('veo_prompt', "")
     path_a = state.get('video_a_local_path')
     path_c = state.get('video_c_local_path')
     
+    update_job_status(job_id, "generating", 50, "Veo initializing...")
+    
     if not path_a or not path_c:
-        return {"generated_video_url": "Error: Missing local video paths"}
+        error_msg = "Error: Missing local video paths"
+        update_job_status(job_id, "error", 0, error_msg)
+        return {"generated_video_url": error_msg}
 
     # --- ATTEMPT 1: GOOGLE VEO ---
     try:
         logger.info("⚡ Initializing Google Veo (Unified SDK)...")
-        project_id = os.getenv("GCP_PROJECT_ID")
-        location = os.getenv("GCP_LOCATION", "us-central1")
+        project_id = Settings.GCP_PROJECT_ID
+        location = Settings.GCP_LOCATION
         
         if project_id:
             client = genai.Client(
@@ -186,6 +169,7 @@ def generate_video(state: ContinuityState) -> dict:
             )
             
             logger.info(f"Generating with Veo... Prompt: {prompt[:30]}...")
+            update_job_status(job_id, "generating", 60, "Veo generating video (this takes ~60s)...")
             
             operation = client.models.generate_videos(
                 model='veo-2.0-generate-001',
@@ -198,33 +182,38 @@ def generate_video(state: ContinuityState) -> dict:
             logger.info(f"Waiting for Veo operation {operation.name}...")
             while not operation.done:
                 time.sleep(10)
+                # Pass operation object, not name
                 operation = client.operations.get(operation)
                 logger.info("...still generating...")
                 
             if operation.result and operation.result.generated_videos:
                 video_result = operation.result.generated_videos[0]
                 
+                local_path = None
                 # CASE 1: URI (GCS Bucket)
                 if hasattr(video_result.video, 'uri') and video_result.video.uri:
                     gcs_uri = video_result.video.uri
                     logger.info(f"Veo output saved to GCS: {gcs_uri}")
+                    
                     with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as f:
                         local_path = f.name
+                    
                     download_blob(gcs_uri, local_path)
                     logger.info(f"✅ Veo Video Downloaded (from GCS): {local_path}")
-                    return {"generated_video_url": local_path}
                 
                 # CASE 2: RAW BYTES (Direct Return)
                 elif hasattr(video_result.video, 'video_bytes') and video_result.video.video_bytes:
                     logger.info("Veo returned raw bytes. Saving to local file...")
-                    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as f:
-                        f.write(video_result.video.video_bytes)
-                        local_path = f.name
+                    local_path = save_video_bytes(video_result.video.video_bytes)
                     logger.info(f"✅ Veo Video Saved (from Bytes): {local_path}")
-                    return {"generated_video_url": local_path}
-                    
+                
                 else:
                     logger.warning(f"Veo operation completed but no URI/Bytes found. Result: {video_result}")
+                
+                if local_path:
+                    update_job_status(job_id, "completed", 100, "Done!", video_url=local_path)
+                    return {"generated_video_url": local_path}
+
             else:
                 logger.warning("Veo operation completed with no result.")
                 
@@ -239,6 +228,8 @@ def generate_video(state: ContinuityState) -> dict:
 
     # --- ATTEMPT 2: SVD FALLBACK (Free) ---
     logger.info("🔄 Switching to SVD Fallback...")
+    update_job_status(job_id, "generating", 60, "Switching to SVD fallback...")
+    
     try:
         import cv2
         from PIL import Image
@@ -258,10 +249,12 @@ def generate_video(state: ContinuityState) -> dict:
             start_path = f_start.name
             
         client = Client("multimodalart/stable-video-diffusion")
+        
+        update_job_status(job_id, "generating", 70, "SVD generating video...")
         result = client.predict(
             handle_file(start_path),
             0.0, 0.0, 1, 25,
-            fn_index=0 
+            api_name="/video" 
         )
         logger.info(f"✅ SVD Generated: {result}")
         
@@ -277,11 +270,13 @@ def generate_video(state: ContinuityState) -> dict:
             final_path = result['video']
         else:
             final_path = result
-            
+        
+        update_job_status(job_id, "completed", 100, "Done (SVD)!", video_url=final_path)
         return {"generated_video_url": final_path}
         
     except Exception as e:
         logger.error(f"❌ All Generators Failed. Error: {e}")
+        update_job_status(job_id, "error", 0, f"All generation failed: {e}")
         return {"generated_video_url": f"Error: {str(e)}"}
 
 # Graph Construction
@@ -294,9 +289,10 @@ workflow.add_edge("generator", END)
 app = workflow.compile()
 
 # --- SERVER COMPATIBILITY WRAPPERS ---
-def analyze_only(state_or_path_a, path_c=None):
+def analyze_only(state_or_path_a, path_c=None, job_id=None):
     if isinstance(state_or_path_a, str) and path_c:
         state = {
+            "job_id": job_id,
             "video_a_url": "local",
             "video_c_url": "local",
             "video_a_local_path": state_or_path_a,
@@ -304,12 +300,15 @@ def analyze_only(state_or_path_a, path_c=None):
         }
     else:
         state = state_or_path_a if isinstance(state_or_path_a, dict) else state_or_path_a.dict()
+        if job_id and "job_id" not in state:
+            state["job_id"] = job_id
 
     result = analyze_videos(state)
     return {"prompt": result.get("scene_analysis"), "status": "success"}
 
-def generate_only(prompt, path_a, path_c):
+def generate_only(prompt, path_a, path_c, job_id=None):
     state = {
+        "job_id": job_id,
         "video_a_url": "local",
         "video_c_url": "local",
         "video_a_local_path": path_a,

config.py

@@ -0,0 +1,32 @@
+import os
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+class Settings:
+    GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
+    GCP_PROJECT_ID = os.getenv("GCP_PROJECT_ID")
+    GCP_LOCATION = os.getenv("GCP_LOCATION", "us-central1")
+    GCP_CREDENTIALS_JSON = os.getenv("GCP_CREDENTIALS_JSON")
+    GROQ_API_KEY = os.getenv("GROQ_API_KEY")
+
+    @classmethod
+    def setup_auth(cls):
+        """Sets up Google Application Credentials if JSON is provided in env."""
+        if cls.GCP_CREDENTIALS_JSON:
+            print("🔐 Found GCP Credentials Secret. Setting up auth...")
+            creds_path = "gcp_credentials.json"
+            with open(creds_path, "w") as f:
+                f.write(cls.GCP_CREDENTIALS_JSON)
+            os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = creds_path
+
+    @classmethod
+    def validate(cls):
+        """Validates critical environment variables."""
+        if not cls.GOOGLE_API_KEY:
+            raise ValueError("GOOGLE_API_KEY is missing from environment variables.")
+
+# Run setup and validation immediately on import
+Settings.setup_auth()
+Settings.validate()

server.py

@@ -1,11 +1,12 @@
-from fastapi import FastAPI, HTTPException, UploadFile, Form, File, Body
+from fastapi import FastAPI, HTTPException, UploadFile, Form, File, Body, BackgroundTasks
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, JSONResponse
 import uvicorn
 import os
 import shutil
 import uuid
+import json
 # FIXED IMPORT: Importing from root agent.py instead of continuity_agent
 from agent import analyze_only, generate_only
 
@@ -30,6 +31,7 @@ async def read_root():
 
 @app.post("/analyze")
 async def analyze_endpoint(
+    background_tasks: BackgroundTasks,
     video_a: UploadFile = File(...),
     video_c: UploadFile = File(...)
 ):
@@ -46,8 +48,24 @@ async def analyze_endpoint(
         with open(path_c, "wb") as buffer:
             shutil.copyfileobj(video_c.file, buffer)
         
+        # Call Agent synchronously for analysis (it's relatively fast usually, but could be async too if desired)
+        # For now keeping it sync as per user previous flows, but could be made async easily.
+        # However, user request specifically focused on "generate" being async. 
+        # But wait, analyze also calls Gemini which can be slow. 
+        # Refactoring to also include job_id for analyze might be good practice but not explicitly requested for "analyze" in prompt detail, 
+        # BUT the user request says: "Update analyze_videos node: Call utils.update_job_status...".
+        # So we should probably treat analyze as async too OR just pass the job_id.
+        # But the frontend flow for analyze currently awaits the response to get the prompt.
+        # If we make it async, we break the frontend flow unless we refactor that too.
+        # The prompt says: "Update code.html ... Update the generate button JavaScript." 
+        # It doesn't explicitly say to update the analyze button logic to be async.
+        # However, update_job_status IS called in analyze_videos.
+        # So, we can pass a job_id if we want status updates, but if we await it, the status updates are only useful if polled in parallel.
+        # For now, I will keep analyze synchronous but pass a dummy job_id if we want logging, or just let it block.
+        # Actually, let's keep it blocking as per original server code, but pass a job_id so at least logs are written.
+        
         # Call Agent with local paths
-        result = analyze_only(os.path.abspath(path_a), os.path.abspath(path_c))
+        result = analyze_only(os.path.abspath(path_a), os.path.abspath(path_c), job_id=request_id)
     
         if result.get("status") == "error":
              raise HTTPException(status_code=500, detail=result.get("detail"))
@@ -63,6 +81,7 @@ async def analyze_endpoint(
 
 @app.post("/generate")
 async def generate_endpoint(
+    background_tasks: BackgroundTasks,
     prompt: str = Body(...),
     video_a_path: str = Body(...),
     video_c_path: str = Body(...)
@@ -71,26 +90,34 @@ async def generate_endpoint(
         if not os.path.exists(video_a_path) or not os.path.exists(video_c_path):
              raise HTTPException(status_code=400, detail="Video files not found on server.")
 
-        # Call Agent
-        result = generate_only(prompt, video_a_path, video_c_path)
-        gen_path = result.get("generated_video_url")
-        
-        if not gen_path or "Error" in gen_path:
-            raise HTTPException(status_code=500, detail=f"Generation failed: {gen_path}")
-            
-        final_filename = f"{uuid.uuid4()}_bridge.mp4"
-        final_output_path = os.path.join(OUTPUT_DIR, final_filename)
+        job_id = str(uuid.uuid4())
         
-        if os.path.exists(gen_path):
-             shutil.move(gen_path, final_output_path)
-        else:
-             raise HTTPException(status_code=500, detail="Generated file missing.")
+        # Initialize job status
+        status_file = os.path.join(OUTPUT_DIR, f"{job_id}.json")
+        with open(status_file, "w") as f:
+            json.dump({"status": "queued", "progress": 0, "log": "Job queued..."}, f)
+
+        # Add to background tasks
+        background_tasks.add_task(generate_only, prompt, video_a_path, video_c_path, job_id)
         
-        return {"video_url": f"/outputs/{final_filename}"}
+        return {"job_id": job_id}
         
     except Exception as e:
         print(f"Server Error (Generate): {e}")
         raise HTTPException(status_code=500, detail=str(e))
 
+@app.get("/status/{job_id}")
+async def get_status(job_id: str):
+    file_path = os.path.join(OUTPUT_DIR, f"{job_id}.json")
+    if not os.path.exists(file_path):
+        raise HTTPException(status_code=404, detail="Job not found")
+        
+    try:
+        with open(file_path, "r") as f:
+            data = json.load(f)
+        return data
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error reading status: {e}")
+
 if __name__ == "__main__":
     uvicorn.run("server:app", host="0.0.0.0", port=7860, reload=False)

stitch_continuity_dashboard/code.html

@@ -364,11 +364,11 @@
 
         // Loading State
         const originalContent = btn.innerHTML;
-        btn.innerHTML = `<span class="material-symbols-outlined animate-spin text-[20px] mr-2">progress_activity</span> Generating (This may take ~60s)...`;
         btn.disabled = true;
         btn.classList.add("opacity-70", "cursor-not-allowed");
 
         try {
+          // 1. Start Job
           const response = await fetch("/generate", {
             method: "POST",
             headers: {
@@ -381,29 +381,108 @@
             })
           });
 
-          if (!response.ok) throw new Error("Generation failed.");
+          if (!response.ok) throw new Error("Generation failed to start.");
 
           const data = await response.json();
-
-          // Success: Inject Video
-          const bridgeCard = document.getElementById("bridge-card");
-          if (bridgeCard) {
-            bridgeCard.innerHTML = `
+          const jobId = data.job_id;
+
+          // 2. Poll for Status
+          const pollInterval = setInterval(async () => {
+            try {
+              const statusRes = await fetch(`/status/${jobId}`);
+              if (!statusRes.ok) return;
+
+              const statusData = await statusRes.json();
+
+              // Update Button Text
+              btn.innerHTML = `<span class="material-symbols-outlined animate-spin text-[20px] mr-2">progress_activity</span> ${statusData.log} (${statusData.progress}%)`;
+
+              if (statusData.status === "completed") {
+                clearInterval(pollInterval);
+
+                // Show Video
+                const bridgeCard = document.getElementById("bridge-card");
+                if (bridgeCard) {
+                  // If video_url is a relative path like "outputs/...", make sure it starts with /
+                  let videoUrl = statusData.video_url;
+                  // The backend returns full local path generally, but server logic for update_job_status
+                  // kept the local path. We need to serve it via the /outputs/ mount.
+                  // Wait, utils.save_video_bytes returns absolute path. 
+                  // server.py: app.mount("/outputs", StaticFiles(directory=OUTPUT_DIR), name="outputs")
+                  // config.py/utils.py save to temp folder? 
+                  // Agent returns local path. 
+                  // In server.py for generate_only, we don't have the final file move logic anymore locally in server.py?
+                  // Ah, right. Agent calls update_job_status with the video_url (local path).
+                  // The frontend needs a URL it can access.
+                  // We need to move the file to OUTPUT_DIR or make sure Agent saves to OUTPUT_DIR.
+                  // Agent uses download_to_temp / save_video_bytes which use tempfile.
+                  // So the file in 'video_url' is in /tmp/... 
+                  // This won't be accessible via /outputs/.
+                  // We need to fix this.
+
+                  // This requires a minor fix in server or agent. 
+                  // Since I can't edit server.py inside this JS block logic easily to change python logic...
+                  // I should probably have updated the Agent to save to OUTPUT_DIR or Server to handle the move?
+                  // BUT, the server.py I just wrote launches generate_only in background.
+                  // generate_only returns the result, but since it's in background, we ignore return.
+                  // The only communication channel is the status file.
+                  // The status file contains 'video_url' which is the temp path.
+                  // Frontend cannot access temp path.
+
+                  // Quick FIX:
+                  // The frontend can't move files.
+                  // The AGENT needs to know where to save, OR Utils.
+                  // OR, we update the logic in agent.py to try to move it? No that's messy.
+                  // Let's look at where static files are served. `/outputs`.
+                  // If the file is in /tmp, we can't serve it.
+
+                  // Maybe I should modify `utils.save_video_bytes` to accept a directory?
+                  // Or I can modify `server.py` to launch a wrapper that moves the file?
+
+                  // I will implement a wrapper in server.py in the NEXT step or modify server.py now if I catch it?
+                  // I already wrote server.py.
+                  // I will apply a fix to `agent.py` or `utils.py` to ensure it saves to `outputs/` OR 
+                  // modify `server.py` to wrap the task.
+                  // Wrapping in server.py is cleanest for separation.
+                  // "generate_only" is imported.
+
+                  // Wait, `utils.update_job_status` writes to `outputs/{job_id}.json`.
+                  // If I change `utils.py` to also move the video if provided?
+                  // That seems like a good side effect for a "helper" dealing with job status in this specific app context.
+                  // Or `agent.py` can move it.
+
+                  // Let's handle this in the JS for now assuming the URL *might* work if I fix it in backend in a sec.
+                  // I'll assume the backend will provide a valid relative URL like `/outputs/filename.mp4`.
+
+                  bridgeCard.innerHTML = `
                         <video controls autoplay loop class="w-full h-full object-cover rounded-2xl border-2 border-primary shadow-neon">
-                        <source src="${data.video_url}" type="video/mp4">
+                        <source src="${videoUrl}" type="video/mp4">
                         Your browser does not support the video tag.
                         </video>
                     `;
-            // Reset UI to allow new creation but keep video
-            document.getElementById("analysis-panel").classList.remove("hidden");
-            document.getElementById("review-panel").classList.add("hidden");
-            btn.innerHTML = originalContent; // Restore button just in case
-          }
+
+                  document.getElementById("analysis-panel").classList.remove("hidden");
+                  document.getElementById("review-panel").classList.add("hidden");
+                  btn.innerHTML = originalContent;
+                  btn.disabled = false;
+                  btn.classList.remove("opacity-70", "cursor-not-allowed");
+                }
+              } else if (statusData.status === "error") {
+                clearInterval(pollInterval);
+                alert("❌ Generation Error: " + statusData.log);
+                btn.innerHTML = originalContent;
+                btn.disabled = false;
+                btn.classList.remove("opacity-70", "cursor-not-allowed");
+              }
+
+            } catch (e) {
+              console.error("Polling error", e);
+            }
+          }, 1000);
 
         } catch (error) {
           console.error(error);
-          alert("❌ Generation Failed: " + error.message);
-        } finally {
+          alert("❌ Request Failed: " + error.message);
           btn.innerHTML = originalContent;
           btn.disabled = false;
           btn.classList.remove("opacity-70", "cursor-not-allowed");

utils.py

@@ -0,0 +1,84 @@
+import os
+import shutil
+import requests
+import tempfile
+import logging
+import json
+from google.cloud import storage
+
+# Configure logging for utils
+logger = logging.getLogger(__name__)
+
+def download_to_temp(url):
+    """Downloads a file from a URL to a temporary local file."""
+    logger.info(f"Downloading: {url}")
+    if os.path.exists(url):
+        return url
+
+    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
+
+def download_blob(gcs_uri, destination_file_name):
+    """Downloads a blob from the Google Cloud Storage bucket."""
+    if not gcs_uri.startswith("gs://"):
+        raise ValueError(f"Invalid GCS URI: {gcs_uri}")
+
+    parts = gcs_uri[5:].split("/", 1)
+    bucket_name = parts[0]
+    source_blob_name = parts[1]
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+    blob = bucket.blob(source_blob_name)
+    blob.download_to_filename(destination_file_name)
+    logger.info(f"Downloaded storage object {gcs_uri} to local file {destination_file_name}.")
+
+def save_video_bytes(bytes_data, suffix=".mp4") -> str:
+    """Saves raw video bytes to a temporary local file."""
+    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as f:
+        f.write(bytes_data)
+        local_path = f.name
+    logger.info(f"✅ Video bytes saved to: {local_path}")
+    return local_path
+
+def update_job_status(job_id, status, progress, log=None, video_url=None):
+    """Writes a JSON file to outputs/{job_id}.json with current job status."""
+    if not job_id:
+        return
+
+    output_dir = "outputs"
+    os.makedirs(output_dir, exist_ok=True)
+    
+    # Handle video file move if completed
+    final_video_url = video_url
+    if video_url and os.path.exists(video_url) and status == "completed":
+        try:
+            filename = os.path.basename(video_url)
+            # Ensure unique name or use job_id
+            final_filename = f"{job_id}_final{os.path.splitext(filename)[1]}"
+            destination = os.path.join(output_dir, final_filename)
+            shutil.move(video_url, destination)
+            logger.info(f"Moved video to {destination}")
+            # Set public URL relative to server root
+            final_video_url = f"/outputs/{final_filename}"
+        except Exception as e:
+            logger.error(f"Failed to move output video: {e}")
+
+    file_path = os.path.join(output_dir, f"{job_id}.json")
+    
+    data = {
+        "status": status,
+        "progress": progress,
+        "log": log,
+        "video_url": final_video_url
+    }
+    
+    try:
+        with open(file_path, "w") as f:
+            json.dump(data, f)
+        logger.info(f"Job {job_id} updated: {status} ({progress}%) - {log}")
+    except Exception as e:
+        logger.error(f"Failed to update job status for {job_id}: {e}")