feat(agent): autoscale the worker count (throughput hill-climb), Auto default-on

The new per-job workload (3 detectors + several SigLIP embeds) is far more
GPU-bound than the old I/O-bound CCIP pass, so the right worker count shifted and
is hard to guess. Add an Auto mode (default ON) that finds it:

- _control_loop samples jobs/sec + GPU util/VRAM every ~6s and hill-climbs the
  target: grow while throughput keeps improving and VRAM stays under budget,
  revert a step that doesn't help, back off under memory pressure (VRAM >= 90%),
  then settle and periodically re-probe (the GPU/IO balance shifts over a run).
- A manual concurrency set is an override → leaves Auto; an "Auto" toggle in the
  control UI re-enables it. status() reports `auto`; the dial reflects the
  auto-chosen count (read-only) while Auto is on.
- AUTO_SCALE env (default on) + compose doc. Agent py-compiled (outside CI).

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

agent/fc_agent/worker.py

@@ -48,6 +48,16 @@ MAX_CONCURRENCY = 32
 DEFAULT_EMBED_MODEL = "google/siglip-so400m-patch14-384"
 DEFAULT_EMBED_VERSION = "siglip-so400m-patch14-384"
 
+# Autoscaler (when Auto is on): a throughput hill-climb that finds the worker
+# count on its own — grows while jobs/sec keeps rising and VRAM stays under
+# budget, backs off when a step stops helping or memory gets tight, then settles
+# and periodically re-probes (the workload's GPU/IO balance shifts).
+CONTROL_INTERVAL = 6.0     # seconds between control decisions
+VRAM_HI = 0.90             # back off above this fraction of VRAM used
+UTIL_HI = 96               # GPU util% considered saturated
+TPUT_MARGIN = 0.10         # a step up must beat the baseline by this to "help"
+REPROBE_TICKS = 5          # ticks to hold after settling before re-probing up
+
 
 class _Slot:
     """One worker loop. `inflight` = jobs leased but not yet processed, so a
@@ -66,6 +76,9 @@ class Worker:
         self._lock = threading.Lock()
         self._running = False
         self._target = max(1, min(MAX_CONCURRENCY, cfg.concurrency))
+        self._auto = bool(cfg.auto_scale)      # autoscale worker count
+        self._ctrl_stop = threading.Event()
+        self._ctrl_thread: threading.Thread | None = None
         self._slots: list[_Slot] = []
         self.processed = 0
         self.errors = 0
@@ -85,20 +98,43 @@ class Worker:
         with self._lock:
             self._running = True
             self._reconcile_locked()
+        # (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):
+        self._ctrl_stop.set()
         with self._lock:
             self._running = False
             slots, self._slots = self._slots, []
         for s in slots:
             s.stop.set()          # each slot releases its inflight on exit
 
-    def set_concurrency(self, n: int):
+    def set_auto(self, on: bool):
         with self._lock:
+            self._auto = bool(on)
+
+    def set_concurrency(self, n: int):
+        # A manual set is an override → leave Auto.
+        with self._lock:
+            self._auto = False
             self._target = max(1, min(MAX_CONCURRENCY, int(n)))
             if self._running:
                 self._reconcile_locked()
 
+    def _apply_step(self, delta: int) -> bool:
+        """Nudge the target by delta (bounded). Returns True if it changed."""
+        with self._lock:
+            new = max(1, min(MAX_CONCURRENCY, self._target + delta))
+            if new == self._target:
+                return False
+            self._target = new
+            if self._running:
+                self._reconcile_locked()
+            return True
+
     def _reconcile_locked(self):
         while len(self._slots) < self._target:
             slot = _Slot()
@@ -113,6 +149,7 @@ class Worker:
                 "state": "running" if self._running else "stopped",
                 "concurrency": self._target,
                 "max_concurrency": MAX_CONCURRENCY,
+                "auto": self._auto,
                 "workers": len(self._slots),
                 "active": self._active,
                 "processed": self.processed,
@@ -171,6 +208,58 @@ class Worker:
         a pool-shrink doesn't hang for a full backoff window."""
         slot.stop.wait(timeout=seconds)
 
+    # --- autoscaler --------------------------------------------------------
+    def _control_loop(self):
+        """Throughput hill-climb (Auto mode): grow the pool while jobs/sec keeps
+        improving and VRAM stays under budget; revert a step that doesn't help;
+        back off under memory pressure; settle, then periodically re-probe."""
+        import time as _t
+
+        from . import gpu as gpumod
+
+        prev_p, prev_t = self.processed, _t.monotonic()
+        base_tput = None     # throughput baseline the current climb is judged against
+        last_dir = 0         # direction of the last applied step (+1 / -1 / 0)
+        cooldown = 0         # ticks to wait (post-settle / post-backoff) before acting
+        while not self._ctrl_stop.wait(CONTROL_INTERVAL):
+            if not (self._running and self._auto):
+                prev_p, prev_t = self.processed, _t.monotonic()
+                base_tput, last_dir, cooldown = None, 0, 0
+                continue
+            now = _t.monotonic()
+            dt = max(1e-3, now - prev_t)
+            tput = (self.processed - prev_p) / dt
+            prev_p, prev_t = self.processed, now
+
+            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
+
+            if vram >= VRAM_HI:                      # memory pressure → always shrink
+                self._apply_step(-1)
+                base_tput, last_dir, cooldown = None, 0, 2
+                continue
+            if cooldown > 0:
+                cooldown -= 1
+                continue
+            if base_tput is None:                    # establish a baseline + probe up
+                base_tput = tput
+                last_dir = 1 if self._apply_step(1) else 0
+                if last_dir == 0:                    # already at the cap
+                    base_tput, cooldown = None, REPROBE_TICKS
+                continue
+            if last_dir > 0:
+                if tput > base_tput * (1 + TPUT_MARGIN) and util < UTIL_HI:
+                    base_tput = tput                 # the step helped → keep climbing
+                    if not self._apply_step(1):
+                        base_tput, last_dir, cooldown = None, 0, REPROBE_TICKS
+                else:                                # didn't help → revert + settle
+                    self._apply_step(-1)
+                    base_tput, last_dir, cooldown = None, 0, REPROBE_TICKS
+            else:
+                base_tput = None                     # settled → re-probe next cycle
+
     def _ensure_embedder(self, model_name: str):
         if self._embedder is not None:
             return self._embedder