Merge pull request 'fix(agent): stream videos via ffmpeg-from-URL (no full download) — env-agnostic' (#181) from dev into main

agent/fc_agent/app.py

@@ -17,7 +17,7 @@ from .worker import Worker
 # Bump on every agent change. The page embeds this and /status reports it; the UI
 # warns to reload when they differ — so a stale browser-cached page can't be
 # mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.)
-VERSION = "2026-07-01.5 · name the real fetch-failure reason"
+VERSION = "2026-07-01.6 · stream videos via ffmpeg-from-URL (no full download)"
 
 logbuf.install()
 cfg = Config.from_env()

agent/fc_agent/client.py

@@ -121,6 +121,16 @@ class FcClient:
         r.raise_for_status()
         return r.content
 
+    def is_reachable(self) -> bool:
+        """Cheap 'is curator responding at all right now?' check. Used to decide,
+        when a video can't be sampled, between a transient outage (keep retrying —
+        survives a redeploy) and an unprocessable file (fail it, don't loop)."""
+        try:
+            r = self.s.get(f"{self.base}/api/gpu/status", timeout=5)
+            return r.status_code < 500
+        except requests.RequestException:
+            return False
+
     def queue_status(self) -> dict:
         # Short timeout: this backs the UI /status poll, so a busy curator must
         # not hang the page for long (the GPU meters poll /gpu separately).

agent/fc_agent/config.py

@@ -41,6 +41,8 @@ class Config:
     frame_dedupe_distance: int  # video frames whose dHash differs by < this many
     #                             bits are near-duplicates, dropped before detect;
     #                             higher keeps more frames, 0 disables
+    ffmpeg_timeout: float  # hard ceiling (s) for ffmpeg-from-URL video sampling;
+    #                        generous so a SLOW media link still completes
 
     @classmethod
     def from_env(cls) -> Config:
@@ -69,4 +71,5 @@ class Config:
             max_regions=int(os.environ.get("MAX_REGIONS", "128")),
             dedupe_iou=float(os.environ.get("DEDUPE_IOU", "0.85")),
             frame_dedupe_distance=int(os.environ.get("FRAME_DEDUPE_DISTANCE", "8")),
+            ffmpeg_timeout=float(os.environ.get("FFMPEG_TIMEOUT", "1200")),
         )

agent/fc_agent/media.py

@@ -85,32 +85,48 @@ def load_image(data: bytes) -> Image.Image:
     return to_rgb(Image.open(io.BytesIO(data)))
 
 
-def sample_frames(
+# ffmpeg reconnect flags — resume a dropped HTTP transfer (a slow/contended media
-    data: bytes, interval_seconds: float, max_frames: int
+# store can cut a long stream) instead of failing the whole job. Relies only on
+# HTTP + Range, which every FC deployment serves → environment-agnostic.
+_RECONNECT = [
+    "-reconnect", "1", "-reconnect_streamed", "1",
+    "-reconnect_on_network_error", "1", "-reconnect_delay_max", "5",
+]
+
+
+def _collect_frames(
+    tmp: str, interval: float, cap: int
 ) -> list[tuple[float, Image.Image]]:
-    """Extract up to max_frames frames at one-every-interval_seconds via ffmpeg.
+    out: list[tuple[float, Image.Image]] = []
-    Returns [(timestamp_seconds, frame)]. Empty on failure (caller falls back)."""
+    names = sorted(n for n in os.listdir(tmp) if n.startswith("f_"))
+    for i, name in enumerate(names[:cap]):
+        with Image.open(os.path.join(tmp, name)) as im:
+            out.append((round(i * interval, 2), to_rgb(im)))
+    return out
+
+
+def sample_frames_from_url(
+    url: str, interval_seconds: float, max_frames: int,
+    *, headers: str = "", timeout: float = 1200.0,
+) -> list[tuple[float, Image.Image]]:
+    """Sample frames by pointing ffmpeg STRAIGHT at the media URL — it Range-reads
+    only the video index + up to max_frames worth of content, so the agent never
+    downloads the whole file (VR/4K originals run 800MB+ and would buffer ~1GB in
+    RAM and get cut off mid-download). Reconnect flags resume a dropped transfer;
+    the generous timeout lets it finish however slow the link is — completion over
+    speed, and environment-agnostic (any HTTP+Range source). Empty on failure."""
     interval = max(0.5, float(interval_seconds or 4.0))
     cap = max(1, int(max_frames or 64))
+    hdr = ["-headers", headers] if headers else []
     with tempfile.TemporaryDirectory() as tmp:
-        src = os.path.join(tmp, "in")
-        with open(src, "wb") as fh:
-            fh.write(data)
         pattern = os.path.join(tmp, "f_%05d.jpg")
         try:
             subprocess.run(
-                [
+                ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr,
-                    "ffmpeg", "-nostdin", "-loglevel", "error", "-i", src,
+                 "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
-                    "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
+                 "-q:v", "3", pattern],
-                    "-q:v", "3", pattern,
+                check=True, timeout=timeout,
-                ],
-                check=True, timeout=600,
             )
         except (subprocess.SubprocessError, FileNotFoundError):
             return []
-        out: list[tuple[float, Image.Image]] = []
+        return _collect_frames(tmp, interval, cap)
-        names = sorted(n for n in os.listdir(tmp) if n.startswith("f_"))
-        for i, name in enumerate(names[:cap]):
-            with Image.open(os.path.join(tmp, name)) as im:
-                out.append((round(i * interval, 2), to_rgb(im)))
-        return out

agent/fc_agent/worker.py

@@ -132,6 +132,11 @@ class Worker:
     def __init__(self, cfg: Config):
         self.cfg = cfg
         self.client = FcClient(cfg.fc_url, cfg.token, cfg.agent_id)
+        # Sent to ffmpeg-from-URL so video sampling works whether or not a
+        # deployment gates /images behind the bearer token (FC's is public).
+        self._auth_header = (
+            f"Authorization: Bearer {cfg.token}\r\n" if cfg.token else ""
+        )
         self._lock = threading.Lock()
         self._running = False
         self._auto = bool(cfg.auto_scale)      # autoscale the downloader count
@@ -480,26 +485,42 @@ class Worker:
     def _download_decode(self, job: dict):
         """Fetch the image bytes and decode → [(frame_time, PIL.Image)]. Videos
         are sampled into frames (ffmpeg). Records the download + decode timings."""
-        _t = time.monotonic()
-        data = self.client.fetch_image(job["image_url"])
-        self._record("download", time.monotonic() - _t)
-        _t = time.monotonic()
         if media.is_video(job.get("mime", "")):
-            frames = media.sample_frames(
+            # Stream the video: ffmpeg reads the media URL directly and Range-reads
-                data, job.get("frame_interval_seconds", 4.0),
+            # only the frames it needs, so we NEVER pull the whole file (VR/4K
+            # originals are 800MB+ — buffering that in RAM and getting cut off
+            # mid-download was the failure loop). Environment-agnostic + resilient.
+            _t = time.monotonic()
+            url = f"{self.cfg.fc_url}{job['image_url']}"
+            frames = media.sample_frames_from_url(
+                url, job.get("frame_interval_seconds", 4.0),
                 job.get("max_frames", 64),
-            ) or [(None, media.load_image(data))]
+                headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout,
-            # Temporal dedup: a near-static video sampled into many frames re-runs
+            )
-            # the whole detect+embed chain on ~identical frames. Drop near-dup
+            self._record("decode", time.monotonic() - _t)
-            # frames HERE (decode stage, CPU) so the GPU never sees them.
+            if not frames:
+                # Couldn't sample. If curator is up, the file is unprocessable →
+                # a job fault (fail it, don't re-lease forever). If curator is
+                # unreachable, it's transient → let the loop back off + retry
+                # (survives a redeploy). ConnectionError is caught as transient.
+                if self.client.is_reachable():
+                    raise RuntimeError("no frames sampled from video (unprocessable)")
+                raise requests.ConnectionError("curator unreachable during video sampling")
+            # Temporal dedup: a near-static video re-runs the whole detect+embed
+            # chain on ~identical frames — drop near-dups HERE (CPU) pre-GPU.
             if len(frames) > 1 and self.cfg.frame_dedupe_distance > 0:
                 kept = media.dedupe_frames(frames, self.cfg.frame_dedupe_distance)
                 if len(kept) < len(frames):
                     log.info("job %s: video frames %d→%d (near-dup dedup)",
                              job.get("job_id"), len(frames), len(kept))
                 frames = kept
-        else:
+            return frames
-            frames = [(None, media.load_image(data))]
+        # Stills: download the bytes and decode.
+        _t = time.monotonic()
+        data = self.client.fetch_image(job["image_url"])
+        self._record("download", time.monotonic() - _t)
+        _t = time.monotonic()
+        frames = [(None, media.load_image(data))]
         self._record("decode", time.monotonic() - _t)
         return frames