fix(agent): server-side rate metrics + killable-on-stop ffmpeg

Two follow-ups from live debugging of "work/min never populates" and
"stopped never reached".

1) jobs/min + downloads/min are now computed in the BACKEND on a fixed
   cadence (_rate_loop, EWMA) and reported ready-to-show. The rates were
   derived client-side from poll deltas with a dt<30s guard — but a
   backgrounded/unfocused browser tab throttles its timers to ~1/min, so
   every delta exceeded 30s and the guard blanked the rates forever. A
   server-side rate is independent of how often the tab polls. Frontend just
   displays s.jobs_per_min / s.downloads_per_min. VERSION → .9.

2) ffmpeg video sampling is now killable on Stop. A downloader stuck in a
   slow/reconnecting decode (observed: 47s, 230s for one video) couldn't see
   the stop signal until ffmpeg returned, so Stop detached still-running
   threads and work kept flowing long after — "stopped" that wasn't really
   stopped. sample_frames_from_url now runs ffmpeg via Popen and polls a
   `should_stop` callback every 0.5s, terminating (then killing) the process
   at once on Stop or the per-video timeout. A stop-killed job is handed back
   (transient), not failed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

agent/fc_agent/worker.py

@@ -133,6 +133,14 @@ STATS_INTERVAL = 30.0
 QUEUE_POLL_INTERVAL = 5.0
 UI_IDLE_GRACE = 20.0
 
+# Throughput rates (jobs/min, downloads/min) are computed HERE, on a fixed
+# wall-clock cadence, and reported ready-to-show — NOT derived in the browser
+# from poll deltas. A background/unfocused tab throttles its timers to ~1/min,
+# which made a client-side delta-rate reject every sample and read blank forever;
+# a server-side EWMA is immune to how often (or seldom) the UI actually polls.
+RATE_INTERVAL = 3.0
+RATE_ALPHA = 0.3
+
 # Stop backstop: pressing Stop signals every worker thread to wind down and hand
 # its lease back; the UI shows "stopping" until those threads have actually
 # exited, then "stopped" — so the state the operator sees is always truthful.
@@ -185,14 +193,17 @@ class Worker:
         self._held: set[int] = set()
         self._held_lock = threading.Lock()
         self.processed = 0
-        self.downloaded = 0       # jobs fetched+decoded into the buffer (monotonic)
-        #                           — the UI derives a smoothed downloads/min from it,
-        #                           since the instantaneous buffer/active gauges move
-        #                           faster than any sane poll can show.
+        self.downloaded = 0       # jobs fetched+decoded into the buffer (monotonic);
+        #                           feeds the server-side downloads/min rate below.
         self.errors = 0
         self.transient = 0        # jobs handed back due to a server outage (NOT
         #                           failed) — the "waiting out curator" counter
         self._active = 0          # jobs currently mid-GPU (consumers busy)
+        # Smoothed throughput rates the UI just displays (jobs/min, downloads/min),
+        # computed on a fixed cadence by _rate_loop so they're independent of how
+        # often the browser polls (see RATE_INTERVAL). Decay to 0 when work stops.
+        self._jpm = 0.0
+        self._dpm = 0.0
         self._util_smooth: float | None = None   # EWMA GPU util (set by control loop)
         # Curator queue snapshot, refreshed by a background poller so the UI
         # /status read is instant — never an inline curator HTTP call (which
@@ -206,6 +217,7 @@ class Worker:
         self._timing: dict[str, list[float]] = {}
         self._timing_lock = threading.Lock()
         threading.Thread(target=self._stats_loop, daemon=True).start()
+        threading.Thread(target=self._rate_loop, daemon=True).start()
         # The crop embedder (SigLIP-family) and region proposers are built lazily
         # on the first job that needs them and SHARED across all consumers — one
         # instance, so a 2nd consumer adds concurrent inference, not N× VRAM.
@@ -332,6 +344,24 @@ class Worker:
             log.info("timing/%ds — %s (%d jobs)",
                      int(STATS_INTERVAL), " · ".join(parts), jobs)
 
+    def _rate_loop(self) -> None:
+        """Compute the jobs/min + downloads/min the UI shows, on a fixed cadence
+        from the monotonic counters — EWMA-smoothed, clamped so a counter reset
+        (agent restart) can't spike them, decaying to 0 when work stops. Doing
+        this here (not in the browser) makes the numbers independent of the poll
+        rate, so a throttled/unfocused tab still shows a real rate."""
+        prev_p, prev_d, prev_t = self.processed, self.downloaded, time.monotonic()
+        while True:
+            time.sleep(RATE_INTERVAL)
+            now = time.monotonic()
+            dt = now - prev_t
+            if dt > 0:
+                jp = max(0.0, 60.0 * (self.processed - prev_p) / dt)
+                dp = max(0.0, 60.0 * (self.downloaded - prev_d) / dt)
+                self._jpm = RATE_ALPHA * jp + (1 - RATE_ALPHA) * self._jpm
+                self._dpm = RATE_ALPHA * dp + (1 - RATE_ALPHA) * self._dpm
+            prev_p, prev_d, prev_t = self.processed, self.downloaded, now
+
     # --- control -----------------------------------------------------------
     def start(self):
         # Only a fully-stopped worker can start — ignore a click while already
@@ -499,6 +529,8 @@ class Worker:
             "active": self._active,
             "processed": self.processed,
             "downloaded": self.downloaded,
+            "jobs_per_min": round(self._jpm, 1),        # ready-to-show throughput
+            "downloads_per_min": round(self._dpm, 1),
             "errors": self.errors,
             "transient": self.transient,
         }
@@ -544,7 +576,7 @@ class Worker:
                 if self._stopped(stop_evt):
                     break
                 try:
-                    frames = self._download_decode(job)
+                    frames = self._download_decode(job, stop_evt)
                 except requests.RequestException as exc:
                     owned.remove(jid)
                     if _is_transient(exc):
@@ -590,7 +622,7 @@ class Worker:
                 continue
         return False
 
-    def _download_decode(self, job: dict):
+    def _download_decode(self, job: dict, stop_evt: threading.Event):
         """Fetch the image bytes and decode → [(frame_time, PIL.Image)]. Videos
         are sampled into frames (ffmpeg). Records the download + decode timings."""
         if media.is_video(job.get("mime", "")):
@@ -604,10 +636,15 @@ class Worker:
                     url, job.get("frame_interval_seconds", 4.0),
                     job.get("max_frames", 64),
                     headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout,
+                    should_stop=lambda: self._stopped(stop_evt),
                 )
             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
+                # Stop killed ffmpeg → NOT the job's fault; raise transient so the
+                # downloader releases (not fails) it as it winds down.
+                if self._stopped(stop_evt):
+                    raise requests.ConnectionError("stopped during video sampling")
+                # Else 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():