From 713a11e39418fc3f690201091db195159eecc3ab Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Wed, 1 Jul 2026 16:51:26 -0400
Subject: [PATCH] fix(agent): server-side rate metrics + killable-on-stop
 ffmpeg
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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/app.py    | 30 ++++++-----------------
 agent/fc_agent/media.py  | 48 +++++++++++++++++++++++++++++-------
 agent/fc_agent/worker.py | 53 ++++++++++++++++++++++++++++++++++------
 3 files changed, 91 insertions(+), 40 deletions(-)
diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py
index c6cb747..6ae7a1a 100644
--- a/agent/fc_agent/app.py
+++ b/agent/fc_agent/app.py
@@ -21,7 +21,7 @@ log = logging.getLogger("fc_agent.app")
 # 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.8 · real start/stop state machine (no more stuck 'stopping')"
+VERSION = "2026-07-01.9 · server-computed jobs/min + downloads/min (poll-rate independent)"
 
 logbuf.install()
 cfg = Config.from_env()
@@ -252,11 +252,6 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 <script>
  const PAGE_BUILD="__BUILD__"
  let CAP=8
- // Smoothed derived rates (jobs/min, downloads/min) + the deltas they're built
- // from. The raw buffer/on-GPU/downloader counts change many times a second, so
- // a poll only ever samples noise; these EWMA the two monotonic counters into a
- // stable, readable throughput instead. Reset to null on load → re-converge.
- let jpmv=null, dpmv=null, lastP=null, lastD=null, lastRateT=null
  // Optimistic transitional state on click, then apply the POST's own status
  // response (it returns worker.status()) for instant feedback — don't wait on the
  // separate /status poll, which can lag behind the curator queue call.
@@ -308,23 +303,12 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
   // backend truthfully lands on "stopped".
   startbtn.disabled=(st!=='stopped')
   stopbtn.disabled=!(running||st==='starting')
-  // Derived rates over the poll deltas — j/min ≈ GPU throughput, dl/min ≈ fetch
-  // throughput. EWMA-smoothed so a lull between jobs doesn't blank them; clamped
-  // at 0 so an agent restart (counters reset to 0 → negative delta) can't spike
-  // them; the dt<30s guard drops a backgrounded-tab gap that would read as huge.
-  const now=performance.now()
-  if(lastRateT!=null && s.processed!=null){
-   const dt=(now-lastRateT)/1000
-   if(dt>0.5 && dt<30){
-    const jp=Math.max(0,60*(s.processed-lastP)/dt)
-    const dp=Math.max(0,60*((s.downloaded||0)-lastD)/dt)
-    jpmv=(jpmv==null)?jp:0.4*jp+0.6*jpmv
-    dpmv=(dpmv==null)?dp:0.4*dp+0.6*dpmv
-   }
-  }
-  lastP=s.processed; lastD=(s.downloaded||0); lastRateT=now
-  jpm.textContent=(jpmv==null)?'—':Math.round(jpmv)
-  dpm.textContent=(dpmv==null)?'—':Math.round(dpmv)
+  // Throughput rates arrive READY from the backend (jobs/min ≈ GPU throughput,
+  // dl/min ≈ fetch throughput), computed there on a fixed cadence — so they show
+  // a real number no matter how often this tab polls (a backgrounded tab throttles
+  // its timers, which used to leave a client-side delta-rate blank forever).
+  jpm.textContent=(s.jobs_per_min!=null)?Math.round(s.jobs_per_min):'—'
+  dpm.textContent=(s.downloads_per_min!=null)?Math.round(s.downloads_per_min):'—'
   done.textContent=s.processed
   err.textContent=s.errors; err.className='n'+(s.errors>0?' warn':'')
   waited.textContent=s.transient||0
diff --git a/agent/fc_agent/media.py b/agent/fc_agent/media.py
index 4af1f1e..ba70754 100644
--- a/agent/fc_agent/media.py
+++ b/agent/fc_agent/media.py
@@ -5,6 +5,7 @@ import io
 import os
 import subprocess
 import tempfile
+import time
 
 from PIL import Image, ImageFile
 
@@ -105,28 +106,57 @@ def _collect_frames(
     return out
 
 
+def _terminate(proc: subprocess.Popen) -> None:
+    """Stop an ffmpeg cleanly, then hard-kill if it ignores SIGTERM."""
+    proc.terminate()
+    try:
+        proc.wait(timeout=2)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        try:
+            proc.wait(timeout=2)
+        except subprocess.TimeoutExpired:
+            pass
+
+
 def sample_frames_from_url(
     url: str, interval_seconds: float, max_frames: int,
-    *, headers: str = "", timeout: float = 1200.0,
+    *, headers: str = "", timeout: float = 1200.0, should_stop=None,
 ) -> 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."""
+    the timeout is the per-video ceiling (a slow/reconnecting stream can otherwise
+    run for minutes). `should_stop` is polled while ffmpeg runs so a Stop KILLS the
+    subprocess at once — otherwise a downloader stuck in a long decode keeps the
+    agent "working" long after Stop. Empty on failure / stop / timeout."""
     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:
         pattern = os.path.join(tmp, "f_%05d.jpg")
+        cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr,
+               "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
+               "-q:v", "3", pattern]
         try:
-            subprocess.run(
-                ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr,
-                 "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
-                 "-q:v", "3", pattern],
-                check=True, timeout=timeout,
+            proc = subprocess.Popen(
+                cmd, stdin=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
             )
-        except (subprocess.SubprocessError, FileNotFoundError):
+        except (OSError, ValueError):
+            return []
+        # Poll rather than block, so a Stop (or the per-video timeout) can kill a
+        # slow/wedged ffmpeg promptly instead of waiting it out.
+        start = time.monotonic()
+        while True:
+            try:
+                proc.wait(timeout=0.5)
+                break
+            except subprocess.TimeoutExpired:
+                if (should_stop and should_stop()) or (time.monotonic() - start > timeout):
+                    _terminate(proc)
+                    return []
+        if proc.returncode != 0:
             return []
         return _collect_frames(tmp, interval, cap)
diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py
index 27fd09c..2841845 100644
--- a/agent/fc_agent/worker.py
+++ b/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():
-- 
2.52.0

