feat(agent): per-stage timing breakdown (lease/download/decode/gpu/submit)

Instrument the job pipeline so we can see where wall-clock actually goes and
decide — on data, not theory — whether a download/compute split is worth
building. Each stage is timed per job and a rolling breakdown is logged every
30s to the agent console, e.g.:

  timing/30s — lease 8ms · download 310ms · decode 40ms · gpu 165ms · submit 70ms | wall/job 585ms (214 jobs)

- lease timed around client.lease() in the slot loop (per batch).
- download = fetch_image; decode = image/frame decode; gpu = detect + CCIP +
  batched embed; submit = the results POST. One-time model load is excluded
  from the gpu figure.
- Thread-safe accumulator (stage -> [sum, count]) summarised + reset by a small
  daemon reporter thread; logs only when there was work.

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

agent/fc_agent/worker.py

@@ -70,6 +70,11 @@ TPUT_ALPHA = 0.5              # throughput EWMA weight
 TPUT_MARGIN = 0.08            # a grow must lift smoothed jobs/s by this to "help"
 REPROBE_TICKS = 8             # decisions to hold after settling before re-probing
 
+# How often to log the per-stage timing breakdown (lease/download/decode/gpu/
+# submit) so the operator can see where a job's wall-clock actually goes — the
+# data that decides whether a download/compute split is worth building.
+STATS_INTERVAL = 30.0
+
 log = logging.getLogger("fc_agent.worker")
 
 
@@ -106,6 +111,12 @@ class Worker:
         self._queue: dict | None = None
         self._queue_thread = threading.Thread(target=self._queue_poll_loop, daemon=True)
         self._queue_thread.start()
+        # Per-stage timing: stage -> [sum_seconds, count], summarised to the log
+        # every STATS_INTERVAL so we can see where wall-clock goes per job.
+        self._timing: dict[str, list[float]] = {}
+        self._timing_lock = threading.Lock()
+        self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True)
+        self._stats_thread.start()
         # The crop embedder (SigLIP-family) is built lazily on the first job that
         # needs it, from the model the server announces — one shared instance.
         self._embedder = None
@@ -131,6 +142,41 @@ class Worker:
     def util_smooth(self) -> float | None:
         return self._util_smooth
 
+    def _record(self, stage: str, seconds: float) -> None:
+        with self._timing_lock:
+            s = self._timing.get(stage)
+            if s is None:
+                self._timing[stage] = [seconds, 1]
+            else:
+                s[0] += seconds
+                s[1] += 1
+
+    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."""
+        while True:
+            time.sleep(STATS_INTERVAL)
+            with self._timing_lock:
+                snap = {k: (v[0], v[1]) for k, v in self._timing.items() if v[1]}
+                self._timing = {}
+            if not snap:
+                continue
+            order = ["lease", "download", "decode", "gpu", "submit"]
+            parts = [
+                f"{st} {1000 * snap[st][0] / snap[st][1]:.0f}ms"
+                for st in order if st in snap
+            ]
+            jobs = (snap.get("gpu") or snap.get("download") or (0, 0))[1]
+            # Per-job wall time across the compute path (lease is per-batch, so
+            # it's shown separately above, not folded into this figure).
+            per_job = sum(
+                snap[st][0] for st in ("download", "decode", "gpu", "submit")
+                if st in snap
+            )
+            pj_ms = 1000 * per_job / jobs if jobs else 0
+            log.info("timing/%ds — %s | wall/job %.0fms (%d jobs)",
+                     int(STATS_INTERVAL), " · ".join(parts), pj_ms, jobs)
+
     # --- control -----------------------------------------------------------
     def start(self):
         with self._lock:
@@ -209,7 +255,9 @@ class Worker:
         backoff = self.cfg.poll_idle_seconds
         while not slot.stop.is_set() and self._running:
             try:
+                _t = time.monotonic()
                 jobs = self.client.lease(self.cfg.batch_size)
+                self._record("lease", time.monotonic() - _t)
                 backoff = self.cfg.poll_idle_seconds      # server answered → reset
             except Exception:
                 # curator unreachable (redeploy, network drop): wait it out with
@@ -367,7 +415,11 @@ class Worker:
         re-leased rather than fail()ed into its attempt budget."""
         self._bump(active=1)
         try:
+            _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),
@@ -375,6 +427,7 @@ class Worker:
                 ) or [(None, media.load_image(data))]
             else:
                 frames = [(None, media.load_image(data))]
+            self._record("decode", time.monotonic() - _t)
 
             task = job.get("task") or "ccip"
             embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION
@@ -388,7 +441,8 @@ class Worker:
             # new model, #1190) → image_record.siglip_embedding. Mean-pool video
             # frames, matching the server's tag_and_embed. No regions.
             if task == "embed":
-                embedder = self._ensure_embedder(model_name)
+                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(
@@ -396,7 +450,10 @@ class Worker:
                     ).tolist()
                 else:
                     vec = vecs[0]
+                self._record("gpu", time.monotonic() - _t)
+                _t = time.monotonic()
                 self.client.submit_embedding(job["job_id"], vec, embed_version)
+                self._record("submit", time.monotonic() - _t)
                 self._bump(processed=1)
                 return True
 
@@ -419,6 +476,7 @@ 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:
                 # FIGURE boxes: imgutils detect_person ∪ general COCO person,
                 # NMS-merged → CCIP identity (+ a concept crop). Covers anime +
@@ -473,7 +531,10 @@ class Worker:
                         tmpl["siglip_embedding"] = vec
                         tmpl["embedding_version"] = embed_version
                         regions.append(tmpl)
+            self._record("gpu", time.monotonic() - _t_gpu)
+            _t = time.monotonic()
             self.client.submit(job["job_id"], regions, replace_kinds)
+            self._record("submit", time.monotonic() - _t)
             self._bump(processed=1)
             return True
         except requests.RequestException as exc: