Add live video streaming for in-progress renders

app/routers/runs.py

@@ -955,3 +955,40 @@ async def purge_failed_runs(
 
     logger.info(f"Purged {len(deleted)} failed runs")
     return {"purged": len(deleted), "run_ids": deleted}
+
+
+@router.get("/{run_id}/stream")
+async def stream_run_output(
+    run_id: str,
+    request: Request,
+):
+    """Stream the video output of a running render.
+    
+    Returns the partial video file as it's being written,
+    allowing live preview of the render progress.
+    """
+    from fastapi.responses import StreamingResponse, FileResponse
+    from pathlib import Path
+    import os
+
+    # Check for the streaming output file in the shared cache
+    cache_dir = os.environ.get("CACHE_DIR", "/data/cache")
+    stream_path = Path(cache_dir) / "streaming" / run_id / "output.mp4"
+
+    if not stream_path.exists():
+        raise HTTPException(404, "Stream not available yet")
+
+    file_size = stream_path.stat().st_size
+    if file_size == 0:
+        raise HTTPException(404, "Stream not ready")
+
+    # Return the file with headers that allow streaming of growing file
+    return FileResponse(
+        path=str(stream_path),
+        media_type="video/mp4",
+        headers={
+            "Accept-Ranges": "bytes",
+            "Cache-Control": "no-cache, no-store, must-revalidate",
+            "X-Content-Size": str(file_size),
+        }
+    )

tasks/streaming.py

@@ -255,7 +255,12 @@ def run_stream(
     # Create temp directory for work
     work_dir = Path(tempfile.mkdtemp(prefix="stream_"))
     recipe_path = work_dir / "recipe.sexp"
-    output_path = work_dir / output_name
+
+    # Write output to shared cache for live streaming access
+    cache_dir = Path(os.environ.get("CACHE_DIR", "/data/cache"))
+    stream_dir = cache_dir / "streaming" / run_id
+    stream_dir.mkdir(parents=True, exist_ok=True)
+    output_path = stream_dir / "output.mp4"  # Always mp4 for streaming
 
     # Create symlinks to effect directories so relative paths work
     (work_dir / "sexp_effects").symlink_to(sexp_effects_dir)
@@ -438,7 +443,9 @@ def run_stream(
         }
 
     finally:
-        # Cleanup temp directory
+        # Cleanup temp directory and streaming directory
         import shutil
         if work_dir.exists():
             shutil.rmtree(work_dir, ignore_errors=True)
+        if stream_dir.exists():
+            shutil.rmtree(stream_dir, ignore_errors=True)