Merge pull request 'Agent: DRY pass + Status rate-metrics + real start/stop state machine' (#182) from dev into main

2026-07-01 16:30:32 -04:00
parent 5ba9871ef0 6282e753a9
commit ed20df905b
5 changed files with 363 additions and 244 deletions
@@ -6,6 +6,8 @@ bandwidth for throughput), and watch GPU load + buffer occupancy + progress +
 the server-side queue. Config is env-seeded; the downloader count is adjustable
 here on the fly (GPU consumers autoscale between 1 and 2 on their own).
 """
+import logging
+
 from fastapi import FastAPI, Request
 from fastapi.responses import HTMLResponse, JSONResponse

@@ -14,10 +16,12 @@ from .config import Config
 from .gpu import read_gpu
 from .worker import Worker

+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.6 · stream videos via ffmpeg-from-URL (no full download)"
+VERSION = "2026-07-01.8 · real start/stop state machine (no more stuck 'stopping')"

 logbuf.install()
 cfg = Config.from_env()
@@ -53,12 +57,14 @@ def index() -> str:

@app.post("/start")
 def start():
+    log.info("UI: Start button pressed")   # the press; worker logs the transition
    worker.start()
    return JSONResponse(worker.status())


@app.post("/stop")
 def stop():
+    log.info("UI: Stop button pressed")
    worker.stop()
    return JSONResponse(worker.status())

@@ -218,11 +224,11 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  <div class=card-h>Status</div>
  <div class=tiles>
   <div class=tile><div class=n id=state>—</div><div class=l>state</div></div>
-   <div class=tile><div class=n id=dln>0</div><div class=l>downloaders</div></div>
-   <div class=tile><div class=n id=bufn>—</div><div class=l>buffer</div></div>
-   <div class=tile><div class=n id=active>0</div><div class=l>on GPU</div></div>
+   <div class=tile><div class=n id=jpm>—</div><div class=l>jobs / min</div></div>
+   <div class=tile><div class=n id=dpm>—</div><div class=l>downloads / min</div></div>
   <div class=tile><div class="n ok" id=done>0</div><div class=l>processed</div></div>
   <div class=tile><div class=n id=err>0</div><div class=l>errors</div></div>
+   <div class=tile><div class=n id=waited>0</div><div class=l>waited out</div></div>
  </div>
  <div class=meters>
   <div class=meter><div class=meter-h><span>GPU util</span><b id=utillbl>—</b></div>
@@ -232,7 +238,7 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
   <div class=meter><div class=meter-h><span>buffer occupancy</span><b id=buflbl>—</b></div>
    <div class=bar><i id=bufbar></i></div></div>
  </div>
-  <div class=queue id=pipe>consumers — · waited out 0</div>
+  <div class=queue id=pipe>downloaders — · consumers — · on GPU 0</div>
  <div class=queue id=queue>queue —</div>
 </section>

@@ -246,6 +252,11 @@ _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.
@@ -255,10 +266,12 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  // /status refresh (now always fast) recovers the true state either way.
  const ac=new AbortController(); const to=setTimeout(()=>ac.abort(),8000)
  try{ applyStatus(await (await fetch('/'+p,{method:'POST',signal:ac.signal})).json()) }
-  catch{ /* leave the periodic refresh to recover the real state */ }
-  finally{ clearTimeout(to); startbtn.disabled=false; stopbtn.disabled=false }
+  catch{ refresh() /* on abort/error, repaint the real state from /status */ }
+  finally{ clearTimeout(to) }
 }
 function pending(label){
+  // Instant optimistic feedback on click; applyStatus (POST response, then the
+  // periodic poll) then owns the real state + which buttons are enabled.
  state.textContent=label; state.className='n busy'
  dot.className='dot amber'
  startbtn.disabled=true; stopbtn.disabled=true
@@ -279,21 +292,45 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 }
 function applyStatus(s){
  CAP=s.max_concurrency||8; capn.textContent=CAP
-  const running=s.state==='running'
+  // The backend owns the state now (stopped|starting|running|stopping) and drives
+  // every transition, so the pill is always truthful — no client-side guessing
+  // from active>0, which used to wedge on "stopping" forever.
+  const st=s.state||'stopped'
+  const running=st==='running'
+  const busy=(st==='starting'||st==='stopping')
  // Stale-page guard: if the server is a newer build than this page, the cached
  // controls may misbehave — tell the operator to reload.
  if(s.build && s.build!==PAGE_BUILD) verbanner.style.display='block'
-  // "stopping" = stopped but in-flight jobs are still draining; resolves to
-  // "stopped" on its own once active hits 0.
-  const draining=!running && s.active>0
-  state.textContent=draining?'stopping':s.state
-  state.className='n'+(draining?' busy':'')
-  dln.textContent=(s.downloaders!=null?s.downloaders:'—')
-  bufn.textContent=(s.buffer!=null?(s.buffer+'/'+s.buffer_max):'—')
-  active.textContent=s.active; active.className='n'+(s.active>0?' busy':'')
+  state.textContent=st
+  state.className='n'+(busy?' busy':'')
+  // Buttons follow the real state so you can't fight a transition: Start only
+  // from stopped; Stop only while up; both disabled through "stopping" until the
+  // 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)
  done.textContent=s.processed
  err.textContent=s.errors; err.className='n'+(s.errors>0?' warn':'')
-  pipe.textContent='consumers '+(s.consumers!=null?s.consumers:'—')+' · waited out '+(s.transient||0)
+  waited.textContent=s.transient||0
+  // Instantaneous pool state → demoted to the sub-line, where its jumpiness reads
+  // as live churn rather than a "broken" headline metric.
+  pipe.textContent='downloaders '+(s.downloaders!=null?s.downloaders:'—')+' · consumers '+(s.consumers!=null?s.consumers:'—')+' · on GPU '+(s.active||0)
  // Buffer occupancy bar (also driven here so it tracks the /status cadence).
  if(s.buffer!=null && s.buffer_max){ const p=Math.round(100*s.buffer/s.buffer_max)
   buflbl.textContent=s.buffer+' / '+s.buffer_max; bufbar.style.width=p+'%' }
@@ -309,9 +346,16 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  if('configured' in s){
   const ok=s.configured
   fc.textContent=s.fc_url; cfg.textContent=ok?'set':'MISSING'
-   dot.className='dot '+(!ok?'red':(running?(s.queue?'green':'amber'):'amber'))
-   connlbl.textContent=!ok?'no token':(running?(s.queue?'running':'running · curator unreachable'):'stopped')
-   banner.style.display=(running && !s.queue)?'block':'none'
+   // Pill colour + label track the real state: green only when running AND
+   // curator is answering; amber for the transient states + a running-but-
+   // unreachable curator; grey when stopped; red with no token.
+   let dc='dot', lbl='stopped'
+   if(!ok){ dc='dot red'; lbl='no token' }
+   else if(st==='running'){ dc='dot '+(s.queue?'green':'amber'); lbl=s.queue?'running':'running · curator unreachable' }
+   else if(st==='starting'){ dc='dot amber'; lbl='starting…' }
+   else if(st==='stopping'){ dc='dot amber'; lbl='stopping…' }
+   dot.className=dc; connlbl.textContent=lbl
+   banner.style.display=(st==='running' && !s.queue)?'block':'none'
   queue.textContent=s.queue?('queue · pending '+s.queue.pending+' · in flight '+s.queue.leased+' · done '+s.queue.done+' · errored '+s.queue.error):'queue · unreachable'
  }
 }
@@ -44,6 +44,31 @@ class FcClient:
        s.mount("https://", adapter)
        return s

+    def _submit(self, path: str, payload: dict) -> dict:
+        """POST to a submit endpoint on the RETRYING session (by submit time the
+        GPU work is done — a blip must not throw it away), raise on a hard error,
+        and return the parsed JSON. `agent_id` is added to every body."""
+        r = self._submit_s.post(
+            f"{self.base}{path}",
+            json={"agent_id": self.agent_id, **payload},
+            timeout=120,
+        )
+        r.raise_for_status()
+        return r.json()
+
+    def _post_quiet(self, path: str, payload: dict) -> None:
+        """Fire-and-forget POST on the main session — heartbeat/fail/release are
+        best-effort, so a transport error is swallowed (the worker's own retry and
+        the server's orphan-recovery cover a lost call). `agent_id` is added."""
+        try:
+            self.s.post(
+                f"{self.base}{path}",
+                json={"agent_id": self.agent_id, **payload},
+                timeout=30,
+            )
+        except requests.RequestException:
+            pass
+
    def lease(self, batch_size: int) -> list[dict]:
        r = self.s.post(
            f"{self.base}/api/gpu/jobs/lease",
@@ -54,62 +79,27 @@ class FcClient:
        return r.json().get("jobs", [])

    def submit(self, job_id: int, regions: list[dict], replace_kinds: list[str]) -> dict:
-        r = self._submit_s.post(
-            f"{self.base}/api/gpu/jobs/submit",
-            json={
-                "agent_id": self.agent_id, "job_id": job_id,
-                "regions": regions, "replace_kinds": replace_kinds,
-            },
-            timeout=120,
-        )
-        r.raise_for_status()
-        return r.json()
+        return self._submit("/api/gpu/jobs/submit", {
+            "job_id": job_id, "regions": regions, "replace_kinds": replace_kinds,
+        })

    def submit_embedding(self, job_id: int, embedding: list, version: str) -> dict:
        """Post a whole-image SigLIP embedding (the 'embed' task) → image_record."""
-        r = self._submit_s.post(
-            f"{self.base}/api/gpu/jobs/submit_embedding",
-            json={
-                "agent_id": self.agent_id, "job_id": job_id,
-                "embedding": embedding, "embedding_version": version,
-            },
-            timeout=120,
-        )
-        r.raise_for_status()
-        return r.json()
+        return self._submit("/api/gpu/jobs/submit_embedding", {
+            "job_id": job_id, "embedding": embedding, "embedding_version": version,
+        })

    def heartbeat(self, job_ids: list[int]) -> None:
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/heartbeat",
-                json={"agent_id": self.agent_id, "job_ids": job_ids},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
+        self._post_quiet("/api/gpu/jobs/heartbeat", {"job_ids": job_ids})

    def fail(self, job_id: int, error: str) -> None:
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/fail",
-                json={"agent_id": self.agent_id, "job_id": job_id, "error": error},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
+        self._post_quiet("/api/gpu/jobs/fail", {"job_id": job_id, "error": error})

    def release(self, job_ids: list[int]) -> None:
        # Graceful hand-back on stop so orphaned work is re-leased at once.
        if not job_ids:
            return
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/release",
-                json={"agent_id": self.agent_id, "job_ids": job_ids},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
+        self._post_quiet("/api/gpu/jobs/release", {"job_ids": job_ids})

    def fetch_image(self, image_url: str) -> bytes:
        # image_url is a server-relative path ("/images/...").
@@ -8,6 +8,11 @@ import os
 from dataclasses import dataclass


+def _bool_env(name: str, default: str = "") -> bool:
+    """A boolean env var — present + truthy ('1'/'true'/'yes') → True."""
+    return os.environ.get(name, default).lower() in ("1", "true", "yes")
+
+
@dataclass
 class Config:
    fc_url: str          # base URL of the FabledCurator web service
@@ -57,8 +62,8 @@ class Config:
            poll_idle_seconds=float(os.environ.get("POLL_IDLE_SECONDS", "10")),
            embed_dtype=os.environ.get("SIGLIP_DTYPE", "float16"),
            embed_model_override=os.environ.get("EMBED_MODEL_NAME", ""),
-            auto_start=os.environ.get("AUTO_START", "").lower() in ("1", "true", "yes"),
-            auto_scale=os.environ.get("AUTO_SCALE", "true").lower() in ("1", "true", "yes"),
+            auto_start=_bool_env("AUTO_START"),
+            auto_scale=_bool_env("AUTO_SCALE", "true"),
            person_weights=os.environ.get("PERSON_WEIGHTS", "yolo11n.pt"),
            person_conf=float(os.environ.get("PERSON_CONF", "0.35")),
            anatomy_weights=os.environ.get("ANATOMY_WEIGHTS", ""),
@@ -202,14 +202,17 @@ class Proposers:
            boxes += self._person.detect(image)
        return nms_merge(boxes)[: self.cfg.max_figures]   # nms_merge is score-desc

-    def components(self, image):
-        if self._anatomy is None:
+    @staticmethod
+    def _top(detector, image, cap: int):
+        """Top-`cap` detections by score from an optional proposer (None → the
+        proposer is off → []). Shared by the anatomy + panel proposers, which
+        differ only in which detector and which cap."""
+        if detector is None:
            return []
-        items = sorted(self._anatomy.detect(image), key=lambda b: b[1], reverse=True)
-        return items[: self.cfg.max_components]
+        return sorted(detector.detect(image), key=lambda b: b[1], reverse=True)[:cap]
+
+    def components(self, image):
+        return self._top(self._anatomy, image, self.cfg.max_components)

    def panels(self, image):
-        if self._panel is None:
-            return []
-        items = sorted(self._panel.detect(image), key=lambda b: b[1], reverse=True)
-        return items[: self.cfg.max_panels]
+        return self._top(self._panel, image, self.cfg.max_panels)
@@ -28,6 +28,7 @@ work on a blip), release-on-stop (hand leases back at once), region caps + video
 early-exit (bound pathological jobs). Stop drains BOTH pools and releases every
 held lease immediately so orphaned work is re-picked without waiting out the TTL.
 """
+import contextlib
 import logging
 import queue
 import threading
@@ -74,6 +75,13 @@ def _transient_reason(exc: requests.RequestException) -> str:
    return type(exc).__name__


+def _ewma(prev: float | None, x: float, alpha: float) -> float:
+    """Exponentially-weighted moving average: seed with the first sample, then
+    fold each new sample in at weight `alpha`. Smooths the noisy control signals
+    (buffer occupancy, GPU util, throughput) the autoscaler decides on."""
+    return x if prev is None else alpha * x + (1 - alpha) * prev
+
+
 # Pipeline sizing. Downloaders are I/O-bound, but every download streams a full
 # original (large videos included) THROUGH curator's single Python file-serving
 # path — so the ceiling is deliberately modest: too many concurrent large-file
@@ -125,6 +133,23 @@ STATS_INTERVAL = 30.0
 QUEUE_POLL_INTERVAL = 5.0
 UI_IDLE_GRACE = 20.0

+# 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.
+# If a thread wedges (a stuck submit/release to an overloaded curator), NEVER
+# hang in "stopping": declare "stopped" after this many seconds and let the lone
+# straggler finish detached. This is the guarantee the old active>0 UI hack lacked.
+STOPPING_TIMEOUT = 20.0
+
+# The lifecycle the operator sees. `_state` is the source of truth the UI reads;
+# `_running` is the low-level flag the worker THREADS gate on (True while
+# starting/running; False from the Stop press onward, so they wind down at once).
+#   stopped → starting (spun up, not yet reached curator)
+#           → running  (leased at least once — curator is answering)
+#           → stopping (Stop pressed; threads winding down + handing work back)
+#           → stopped  (threads exited, or the timeout backstop fired)
+STOPPED, STARTING, RUNNING, STOPPING = "stopped", "starting", "running", "stopping"
+
 log = logging.getLogger("fc_agent.worker")


@@ -139,6 +164,12 @@ class Worker:
        )
        self._lock = threading.Lock()
        self._running = False
+        # The lifecycle state the UI shows (see the STOPPED/STARTING/... consts).
+        # It stays STOPPING — a truthful "winding down" — from the Stop press until
+        # the worker threads have actually exited (or the timeout backstop fires),
+        # then STOPPED. `_running` flips False the instant Stop is pressed.
+        self._state = STOPPED
+        self._shutdown_thread: threading.Thread | None = None
        self._auto = bool(cfg.auto_scale)      # autoscale the downloader count
        self._dl_target = max(1, min(DL_MAX, cfg.concurrency))
        self._consumer_target = 1              # GPU is fast — start with one
@@ -154,6 +185,10 @@ 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.errors = 0
        self.transient = 0        # jobs handed back due to a server outage (NOT
        #                           failed) — the "waiting out curator" counter
@@ -188,16 +223,38 @@ class Worker:
        with self._held_lock:
            self._held.discard(job_id)

-    def _release_owned(self, job_ids: list[int]) -> None:
-        """Hand a set of still-held leases back to curator and drop them from the
-        held set — used when a downloader exits (stop/shrink) still owning leases
-        it hadn't yet buffered."""
+    def _release(self, job_ids: list[int]) -> None:
+        """Hand still-held leases back to curator and drop them from the held set —
+        the single hand-back path for both a downloader exiting (stop/shrink) with
+        unbuffered leases and a consumer releasing one job it can't finish. Empty
+        list is a no-op; a duplicate release is a harmless server no-op."""
        if not job_ids:
            return
        self.client.release(job_ids)
        for jid in job_ids:
            self._unhold(jid)

+    def _fail(self, jid: int, image_id, exc: Exception, verb: str = "failed") -> None:
+        """Terminal job-fault path: count the error, log it, tell curator the job
+        failed, and drop the lease. `verb` lets a caller say 'failed to decode'."""
+        self._bump(errors=1)
+        log.warning("job %s (image %s) %s: %s", jid, image_id, verb, str(exc)[:200])
+        self.client.fail(jid, str(exc)[:500])
+        self._unhold(jid)
+
+    def _stopped(self, stop_evt: threading.Event) -> bool:
+        """The shared 'should I bail now?' check — the worker is stopping (global
+        flag) or this thread's own stop event is set. Lock-free atomic reads."""
+        return not self._running or stop_evt.is_set()
+
+    def _abort_if_stopped(self, jid: int, stop_evt: threading.Event) -> bool:
+        """If we're stopping, hand this job's lease back (so a Stop drains promptly
+        instead of finishing heavy work) and return True so the caller bails."""
+        if self._stopped(stop_evt):
+            self._release([jid])
+            return True
+        return False
+
    # --- background loops ---------------------------------------------------
    def _heartbeat_loop(self) -> None:
        """Keep every held lease alive so buffered jobs waiting on the GPU aren't
@@ -243,6 +300,17 @@ class Worker:
                s[0] += seconds
                s[1] += 1

+    @contextlib.contextmanager
+    def _timed(self, stage: str):
+        """Time a pipeline stage and fold it into the per-stage breakdown — but
+        ONLY on a clean exit: an exception (a failed download, a stopped GPU pass)
+        propagates through the `yield` and skips the record, so it never skews the
+        stage average with work that didn't complete. Matches the old inline
+        monotonic()/_record() pairs, where _record ran only if no exception fired."""
+        t = time.monotonic()
+        yield
+        self._record(stage, time.monotonic() - t)
+
    def _stats_loop(self) -> None:
        """Log a per-stage timing breakdown every STATS_INTERVAL (only when there
        was work), so the operator can see the download/decode/gpu/submit split.
@@ -266,42 +334,88 @@ class Worker:

    # --- control -----------------------------------------------------------
    def start(self):
+        # Only a fully-stopped worker can start — ignore a click while already
+        # up or still winding down (the UI also disables the button then). This
+        # keeps a Start during "stopping" from racing the shutdown monitor's
+        # lease-release against a fresh pool's new leases.
        with self._lock:
+            if self._state != STOPPED:
+                return
            self._running = True
+            self._state = STARTING     # → RUNNING on the first successful lease
            self._dl_target = max(1, self._dl_target)
            self._consumer_target = max(1, self._consumer_target)
            self._reconcile_locked()
+        log.info("▶ start → starting (spinning up, awaiting curator)")
        # (Re)start the autoscaler control loop.
        if self._ctrl_thread is None or not self._ctrl_thread.is_alive():
            self._ctrl_stop.clear()
            self._ctrl_thread = threading.Thread(target=self._control_loop, daemon=True)
            self._ctrl_thread.start()

-    def stop(self):
-        # Flip the flag FIRST (atomic bool), before any lock, so /status and the
-        # loops observe "stopped" immediately even if _lock is momentarily held —
-        # the state can never lag behind the click.
-        self._running = False
-        self._ctrl_stop.set()
+    def _mark_running(self) -> None:
+        """STARTING → RUNNING once a downloader has actually leased from curator —
+        so 'running' means 'curator is answering', not just 'Start was clicked'.
+        First caller wins; the rest are a cheap no-op."""
+        if self._state != STARTING:
+            return
        with self._lock:
+            if self._state != STARTING:
+                return
+            self._state = RUNNING
+        log.info("state → running (curator reachable — leasing)")
+
+    def stop(self):
+        # Enter STOPPING and signal every worker thread to wind down. The handler
+        # returns at once — the actual wait-for-exit + lease-release runs in a
+        # background monitor, so a slow curator can NEVER block the Stop button or
+        # wedge the state (the bug: the old UI faked "stopping" from active>0 and
+        # sat there forever when a consumer hung mid-release).
+        with self._lock:
+            if self._state not in (STARTING, RUNNING):
+                return            # already stopping/stopped
+            self._state = STOPPING
+            self._running = False     # every thread's _stopped() now trips → wind down
            dls, self._dls = self._dls, []
            cons, self._consumers = self._consumers, []
-            self._active = 0      # no consumers left → the meter reads 0 at once;
-            #                       any lagging decrement is clamped (see _bump)
+            self._active = 0
+        self._ctrl_stop.set()
        for _, ev in dls:
            ev.set()
        for _, ev in cons:
            ev.set()
-        # Wake any consumer blocked on an empty buffer.
+        # Wake any consumer blocked on an empty buffer so it sees the stop at once.
        for _ in range(CONSUMER_MAX):
            try:
                self._buffer.put_nowait(None)
            except queue.Full:
                break
-        # Drain the buffer + release every still-held lease in one shot so orphaned
-        # work is re-leased at once. A downloader/consumer mid-flight may also
-        # release its own job — a duplicate release is a harmless no-op.
+        log.info("■ stop → stopping (winding down, handing work back)")
+        threads = [t for t, _ in dls] + [t for t, _ in cons]
+        self._shutdown_thread = threading.Thread(
+            target=self._finish_stop, args=(threads,), daemon=True)
+        self._shutdown_thread.start()
+
+    def _finish_stop(self, threads: list[threading.Thread]) -> None:
+        """Wait for the signalled worker threads to actually exit, then land on
+        STOPPED — so 'stopped' is truthful. Bounded by STOPPING_TIMEOUT so a
+        wedged submit/release can never hold the UI in 'stopping' forever."""
+        deadline = time.monotonic() + STOPPING_TIMEOUT
+        for t in threads:
+            t.join(timeout=max(0.0, deadline - time.monotonic()))
+        # Hand back every lease still held in one shot (background — no handler
+        # block); a straggler releasing its own job later is a harmless dup.
        self._drain_and_release()
+        with self._lock:
+            if self._state == STOPPING:      # a Start may have re-entered; if so, leave it
+                self._state = STOPPED
+                self._active = 0             # any straggler's -1 is clamped (see _bump)
+        stragglers = sum(1 for t in threads if t.is_alive())
+        if stragglers:
+            log.info("state → stopped (%d thread(s) still winding down past %ds — detached)",
+                     stragglers, int(STOPPING_TIMEOUT))
+        else:
+            log.info("state → stopped (all work handed back)")

    def _drain_and_release(self) -> None:
        while True:
@@ -374,7 +488,7 @@ class Worker:
        # under the GIL) and this backs the UI poll — it must NEVER be able to
        # block behind a thread holding _lock, or the whole status view freezes.
        return {
-            "state": "running" if self._running else "stopped",
+            "state": self._state,   # stopped|starting|running|stopping (truthful)
            "concurrency": self._dl_target,        # the UI dial = downloader count
            "max_concurrency": DL_MAX,
            "auto": self._auto,
@@ -384,13 +498,15 @@ class Worker:
            "buffer_max": BUFFER_MAX,
            "active": self._active,
            "processed": self.processed,
+            "downloaded": self.downloaded,
            "errors": self.errors,
            "transient": self.transient,
        }

-    def _bump(self, *, processed=0, errors=0, active=0, transient=0):
+    def _bump(self, *, processed=0, downloaded=0, errors=0, active=0, transient=0):
        with self._lock:
            self.processed += processed
+            self.downloaded += downloaded
            self.errors += errors
            self.transient += transient
            # Clamp at 0: a Stop resets _active to 0, so a consumer that was
@@ -404,12 +520,12 @@ class Worker:
        on any exit path it releases whatever it still owns so nothing is stranded
        holding a lease."""
        backoff = self.cfg.poll_idle_seconds
-        while self._running and not stop_evt.is_set():
+        while not self._stopped(stop_evt):
            try:
-                _t = time.monotonic()
-                jobs = self.client.lease(self.cfg.batch_size)
-                self._record("lease", time.monotonic() - _t)
+                with self._timed("lease"):
+                    jobs = self.client.lease(self.cfg.batch_size)
                backoff = self.cfg.poll_idle_seconds      # server answered → reset
+                self._mark_running()      # curator answered → leave "starting"
            except Exception:
                # curator unreachable (redeploy, network drop): wait it out with
                # exponential backoff, capped — resume on our own when it returns.
@@ -425,7 +541,7 @@ class Worker:
            owned = [j["job_id"] for j in jobs]   # released on any early exit
            for job in jobs:
                jid = job["job_id"]
-                if not self._running or stop_evt.is_set():
+                if self._stopped(stop_evt):
                    break
                try:
                    frames = self._download_decode(job)
@@ -436,45 +552,37 @@ class Worker:
                        # NOT the job's fault. Hand back this job + the rest of the
                        # batch and back the whole loop off.
                        self._bump(transient=1)
-                        self.client.release([jid])
-                        self._unhold(jid)
+                        self._release([jid])
                        # Name the ACTUAL failure — "curator unreachable" was
                        # printed for every transient, hiding whether a single
                        # file's transfer stalled (ReadTimeout, curator fine) or
                        # curator itself is down (ConnectTimeout/ConnectionError).
                        log.info("fetch failed job %s (image %s, %s) — released, backing off",
                                 jid, job.get("image_id"), _transient_reason(exc))
-                        self._release_owned(owned)
+                        self._release(owned)
                        owned = []
                        if not stop_evt.wait(backoff):
                            backoff = min(backoff * 2, MAX_BACKOFF_SECONDS)
                        break
                    # a job-specific HTTP fault (404 image gone, 400) → fail it
-                    self._bump(errors=1)
-                    log.warning("job %s (image %s) failed: %s",
-                                jid, job.get("image_id"), str(exc)[:200])
-                    self.client.fail(jid, str(exc)[:500])
-                    self._unhold(jid)
+                    self._fail(jid, job.get("image_id"), exc)
                    continue
                except Exception as exc:   # noqa: BLE001 — bad media → the job's fault
                    owned.remove(jid)
-                    self._bump(errors=1)
-                    log.warning("job %s (image %s) failed to decode: %s",
-                                jid, job.get("image_id"), str(exc)[:200])
-                    self.client.fail(jid, str(exc)[:500])
-                    self._unhold(jid)
+                    self._fail(jid, job.get("image_id"), exc, verb="failed to decode")
                    continue
                # Blocks on a full buffer (backpressure) but wakes promptly on stop.
                if self._put((job, frames), stop_evt):
+                    self._bump(downloaded=1)   # fetched+decoded into the buffer
                    owned.remove(jid)      # ownership handed to the buffer/consumer
                else:
                    break                  # stopped while waiting for buffer space
-            self._release_owned(owned)
+            self._release(owned)

    def _put(self, item, stop_evt: threading.Event) -> bool:
        """Push onto the bounded buffer, blocking while it's full but rechecking
        stop so a shrink/Stop can't hang a full-buffer window. False = stopped."""
-        while self._running and not stop_evt.is_set():
+        while not self._stopped(stop_evt):
            try:
                self._buffer.put(item, timeout=0.5)
                return True
@@ -490,14 +598,13 @@ class Worker:
            # 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),
-                headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout,
-            )
-            self._record("decode", time.monotonic() - _t)
+            with self._timed("decode"):
+                frames = media.sample_frames_from_url(
+                    url, job.get("frame_interval_seconds", 4.0),
+                    job.get("max_frames", 64),
+                    headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout,
+                )
            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
@@ -516,18 +623,16 @@ class Worker:
                frames = kept
            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)
+        with self._timed("download"):
+            data = self.client.fetch_image(job["image_url"])
+        with self._timed("decode"):
+            frames = [(None, media.load_image(data))]
        return frames

    # --- GPU consumer pool -------------------------------------------------
    def _consumer(self, stop_evt: threading.Event):
        """Pull decoded jobs off the buffer and run detect + embed + submit."""
-        while self._running and not stop_evt.is_set():
+        while not self._stopped(stop_evt):
            try:
                item = self._buffer.get(timeout=1.0)
            except queue.Empty:
@@ -535,9 +640,7 @@ class Worker:
            if item is None:                 # stop sentinel
                continue
            job, frames = item
-            if not self._running or stop_evt.is_set():
-                self.client.release([job["job_id"]])
-                self._unhold(job["job_id"])
+            if self._abort_if_stopped(job["job_id"], stop_evt):
                continue
            self._bump(active=1)
            try:
@@ -572,9 +675,7 @@ class Worker:
        it so a Stop drains promptly instead of finishing heavy GPU work."""
        jid = job["job_id"]
        try:
-            if not self._running or stop_evt.is_set():
-                self.client.release([jid])
-                self._unhold(jid)
+            if self._abort_if_stopped(jid, stop_evt):
                return False

            task = job.get("task") or "ccip"
@@ -590,22 +691,16 @@ class Worker:
            # frames, matching the server's tag_and_embed. No regions.
            if task == "embed":
                embedder = self._ensure_embedder(model_name)   # one-time model load
-                _t = time.monotonic()
-                vecs = [embedder.embed(frame) for _, frame in frames]
-                if len(vecs) > 1:
-                    vec = np.mean(
-                        np.asarray(vecs, dtype=np.float32), axis=0
-                    ).tolist()
-                else:
-                    vec = vecs[0]
-                self._record("gpu", time.monotonic() - _t)
-                if not self._running or stop_evt.is_set():
-                    self.client.release([jid])
-                    self._unhold(jid)
+                with self._timed("gpu"):
+                    vecs = [embedder.embed(frame) for _, frame in frames]
+                    vec = (
+                        np.mean(np.asarray(vecs, dtype=np.float32), axis=0).tolist()
+                        if len(vecs) > 1 else vecs[0]
+                    )
+                if self._abort_if_stopped(jid, stop_evt):
                    return False
-                _t = time.monotonic()
-                self.client.submit_embedding(jid, vec, embed_version)
-                self._record("submit", time.monotonic() - _t)
+                with self._timed("submit"):
+                    self.client.submit_embedding(jid, vec, embed_version)
                self._unhold(jid)
                return True

@@ -628,84 +723,82 @@ class Worker:
            ccip_ev = self.cfg.ccip_model or "ccip-default"
            dv = f"person-{self.cfg.detector_level}"

-            _t_gpu = time.monotonic()   # detect + CCIP + batched embed = "gpu"
-            for t, frame in frames:
-                # Bail promptly on Stop instead of grinding through every frame of
-                # a long video before the caller can hand the job back.
-                if not self._running or stop_evt.is_set():
-                    break
-                # FIGURE boxes: imgutils detect_person ∪ general COCO person,
-                # NMS-merged → CCIP identity (+ a concept crop). Covers anime +
-                # Western/realistic figures.
-                base = models.detect_figures(frame, self.cfg.detector_level)
-                figs = proposers.figures(frame, base)
-                if not figs:
-                    figs = [((0.0, 0.0, 1.0, 1.0), 1.0, "whole")]  # whole-frame fallback
+            # detect + CCIP + batched embed across every (deduped) frame = "gpu".
+            with self._timed("gpu"):
+                for t, frame in frames:
+                    # Bail promptly on Stop instead of grinding through every frame
+                    # of a long video before the caller can hand the job back.
+                    if self._stopped(stop_evt):
+                        break
+                    # FIGURE boxes: imgutils detect_person ∪ general COCO person,
+                    # NMS-merged → CCIP identity (+ a concept crop). Covers anime +
+                    # Western/realistic figures.
+                    base = models.detect_figures(frame, self.cfg.detector_level)
+                    figs = proposers.figures(frame, base)
+                    if not figs:
+                        figs = [((0.0, 0.0, 1.0, 1.0), 1.0, "whole")]  # whole-frame fallback

-                # Collect every crop that needs a SigLIP embedding, then embed
-                # them in ONE batched forward pass (huge GPU-util + throughput
-                # win vs one forward per crop). CCIP runs per figure inline.
-                pending = []   # (crop, region-template-without-embedding)
-                for bbox, score, _label in figs:
-                    crop = crop_region(frame, bbox)
-                    if crop is None:
+                    # Collect every crop that needs a SigLIP embedding, then embed
+                    # them in ONE batched forward pass (huge GPU-util + throughput
+                    # win vs one forward per crop). CCIP runs per figure inline.
+                    pending = []   # (crop, region-template-without-embedding)
+                    for bbox, score, _label in figs:
+                        crop = crop_region(frame, bbox)
+                        if crop is None:
+                            continue
+                        if want_ccip:
+                            regions.append({
+                                "kind": "figure", "bbox": list(bbox), "frame_time": t,
+                                "score": score,
+                                "ccip_embedding": models.ccip_vector(
+                                    crop, self.cfg.ccip_model or None
+                                ),
+                                "embedding_version": ccip_ev, "detector_version": dv,
+                            })
+                        if want_siglip:
+                            pending.append((crop, {
+                                "kind": "concept", "bbox": list(bbox), "frame_time": t,
+                                "score": score, "detector_version": dv,
+                            }))
+                    if not want_siglip:
                        continue
-                    if want_ccip:
-                        regions.append({
-                            "kind": "figure", "bbox": list(bbox), "frame_time": t,
-                            "score": score,
-                            "ccip_embedding": models.ccip_vector(
-                                crop, self.cfg.ccip_model or None
-                            ),
-                            "embedding_version": ccip_ev, "detector_version": dv,
-                        })
-                    if want_siglip:
-                        pending.append((crop, {
-                            "kind": "concept", "bbox": list(bbox), "frame_time": t,
-                            "score": score, "detector_version": dv,
-                        }))
-                if not want_siglip:
-                    continue
-                # ANATOMY components (booru_yolo) + PANELS → concept/panel crops.
-                for bbox, score, label in proposers.components(frame):
-                    crop = crop_region(frame, bbox)
-                    if crop is not None:
-                        pending.append((crop, {
-                            "kind": "concept", "bbox": list(bbox), "frame_time": t,
-                            "score": score, "detector_version": f"booru:{label}",
-                        }))
-                for bbox, score, _label in proposers.panels(frame):
-                    crop = crop_region(frame, bbox)
-                    if crop is not None:
-                        pending.append((crop, {
-                            "kind": "panel", "bbox": list(bbox), "frame_time": t,
-                            "score": score, "detector_version": "panel",
-                        }))
-                if pending:
-                    # Drop near-duplicate crops (a figure box ≈ an anatomy
-                    # component, overlapping booru classes) before the embed so we
-                    # never SigLIP the same region twice — saves GPU and a slot
-                    # against max_regions. High-IoU + kind-aware, so intentional
-                    # nested crops (figure ⊃ head) survive.
-                    pending = dedupe_crops(pending, self.cfg.dedupe_iou)
-                    vecs = embedder.embed_batch([c for c, _ in pending])
-                    for (_c, tmpl), vec in zip(pending, vecs, strict=True):
-                        tmpl["siglip_embedding"] = vec
-                        tmpl["embedding_version"] = embed_version
-                        regions.append(tmpl)
-                # Stop once we have enough: a long video (image 81602 = a 156 MB
-                # mp4, 64 sampled frames × ~32 regions) would otherwise burn ~38s
-                # of GPU across every frame before the submit is even truncated.
-                # Bounds the WORK, not just the POST body.
-                if len(regions) >= self.cfg.max_regions:
-                    break
-            self._record("gpu", time.monotonic() - _t_gpu)
+                    # ANATOMY components (booru_yolo) + PANELS → concept/panel crops.
+                    for bbox, score, label in proposers.components(frame):
+                        crop = crop_region(frame, bbox)
+                        if crop is not None:
+                            pending.append((crop, {
+                                "kind": "concept", "bbox": list(bbox), "frame_time": t,
+                                "score": score, "detector_version": f"booru:{label}",
+                            }))
+                    for bbox, score, _label in proposers.panels(frame):
+                        crop = crop_region(frame, bbox)
+                        if crop is not None:
+                            pending.append((crop, {
+                                "kind": "panel", "bbox": list(bbox), "frame_time": t,
+                                "score": score, "detector_version": "panel",
+                            }))
+                    if pending:
+                        # Drop near-duplicate crops (a figure box ≈ an anatomy
+                        # component, overlapping booru classes) before the embed so
+                        # we never SigLIP the same region twice — saves GPU and a
+                        # slot against max_regions. High-IoU + kind-aware, so
+                        # intentional nested crops (figure ⊃ head) survive.
+                        pending = dedupe_crops(pending, self.cfg.dedupe_iou)
+                        vecs = embedder.embed_batch([c for c, _ in pending])
+                        for (_c, tmpl), vec in zip(pending, vecs, strict=True):
+                            tmpl["siglip_embedding"] = vec
+                            tmpl["embedding_version"] = embed_version
+                            regions.append(tmpl)
+                    # Stop once we have enough: a long video (image 81602 = a 156 MB
+                    # mp4, 64 sampled frames × ~32 regions) would otherwise burn
+                    # ~38s of GPU across every frame before the submit is even
+                    # truncated. Bounds the WORK, not just the POST body.
+                    if len(regions) >= self.cfg.max_regions:
+                        break

            # A Stop mid-frame-loop leaves partial regions — don't submit those;
            # hand the whole job back so another agent redoes it cleanly.
-            if not self._running or stop_evt.is_set():
-                self.client.release([jid])
-                self._unhold(jid)
+            if self._abort_if_stopped(jid, stop_evt):
                return False

            # Backstop: never submit an unbounded pile of regions (a pathological
@@ -718,9 +811,8 @@ class Worker:
                regions = regions[: self.cfg.max_regions]
                log.info("job %s: capped regions %d→%d (dropped %d)",
                         jid, len(regions) + dropped, len(regions), dropped)
-            _t = time.monotonic()
-            self.client.submit(jid, regions, replace_kinds)
-            self._record("submit", time.monotonic() - _t)
+            with self._timed("submit"):
+                self.client.submit(jid, regions, replace_kinds)
            self._unhold(jid)
            return True
        except requests.RequestException as exc:
@@ -731,21 +823,12 @@ class Worker:
                self._bump(transient=1)
                log.info("submit failed job %s (%s) — released, re-lease later",
                         jid, _transient_reason(exc))
-                self.client.release([jid])
-                self._unhold(jid)
+                self._release([jid])
                return False
-            self._bump(errors=1)
-            log.warning("job %s (image %s) failed: %s",
-                        jid, job.get("image_id"), str(exc)[:200])
-            self.client.fail(jid, str(exc)[:500])
-            self._unhold(jid)
+            self._fail(jid, job.get("image_id"), exc)
            return False
        except Exception as exc:   # noqa: BLE001 — a genuine job fault: report it
-            self._bump(errors=1)
-            log.warning("job %s (image %s) failed: %s",
-                        jid, job.get("image_id"), str(exc)[:200])
-            self.client.fail(jid, str(exc)[:500])
-            self._unhold(jid)
+            self._fail(jid, job.get("image_id"), exc)
            return False

    # --- autoscaler --------------------------------------------------------
@@ -786,16 +869,12 @@ class Worker:
                continue

            occ = self._buffer.qsize() / BUFFER_MAX
-            occ_ewma = occ if occ_ewma is None else (
-                OCC_ALPHA * occ + (1 - OCC_ALPHA) * occ_ewma
-            )
+            occ_ewma = _ewma(occ_ewma, occ, OCC_ALPHA)
            g = gpumod.read_gpu() or {}
            mt = g.get("mem_total_mb") or 0
            vram = (g.get("mem_used_mb", 0) / mt) if mt else 0.0
            util = g.get("util_pct", 0) or 0
-            util_ewma = util if util_ewma is None else (
-                UTIL_ALPHA * util + (1 - UTIL_ALPHA) * util_ewma
-            )
+            util_ewma = _ewma(util_ewma, util, UTIL_ALPHA)
            self._util_smooth = util_ewma

            # Memory pressure overrides the cadence — react immediately.
@@ -815,9 +894,7 @@ class Worker:
            now = time.monotonic()
            inst = (self.processed - prev_p) / max(1e-3, now - prev_t)
            prev_p, prev_t = self.processed, now
-            tput_ewma = inst if tput_ewma is None else (
-                TPUT_ALPHA * inst + (1 - TPUT_ALPHA) * tput_ewma
-            )
+            tput_ewma = _ewma(tput_ewma, inst, TPUT_ALPHA)
            fail_delta = self.transient - prev_fail
            prev_fail = self.transient