fix(agent): stream videos via ffmpeg-from-URL instead of downloading the whole file

The failing "poison" jobs were 800MB+ 4K VR videos: the agent pulled the ENTIRE
file into memory (r.content) just to sample a few frames, which buffered ~1GB in
RAM and — on any slow/contended media store — got cut off mid-download
(ChunkedEncodingError), failed, and re-leased forever. Measured the media read at
~4–6 MB/s (raw off the share, curator out of the path), so no serving-layer tweak
helps; the file simply shouldn't be fully downloaded.

Environment-agnostic fix (works for any deployment, completes even when slow):
- media.sample_frames_from_url(): point ffmpeg straight at curator's /images URL.
  It Range-reads only the video index + up to max_frames of content — never the
  whole file — and reconnect flags resume a dropped transfer instead of failing.
  Generous, env-tunable timeout (FFMPEG_TIMEOUT, default 1200s) = completion over
  speed. Removes the bytes-based sample_frames (dead once videos stream).
- worker._download_decode: videos now stream (no fetch_image, no RAM blowup);
  stills still download+decode. On an ffmpeg miss, probe curator liveness
  (client.is_reachable) → fail the job if curator is up (unprocessable file, stops
  the infinite re-lease) vs release if curator is down (transient, survives a
  redeploy). Auth header passed so it works whether or not /images is gated.

Build marker 2026-07-01.6. Refs issue #1225.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa

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(
-                data, job.get("frame_interval_seconds", 4.0),
+            # Stream the video: ffmpeg reads the media URL directly and Range-reads
+            # 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))]
-            # Temporal dedup: a near-static video sampled into many frames re-runs
-            # the whole detect+embed chain on ~identical frames. Drop near-dup
-            # frames HERE (decode stage, CPU) so the GPU never sees them.
+                headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout,
+            )
+            self._record("decode", time.monotonic() - _t)
+            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:
-            frames = [(None, media.load_image(data))]
+            return frames
+        # 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