feat(agent): global bandwidth cap — the agent can't saturate the desktop's network

One shared TokenBucket (default 8 MB/s; BANDWIDTH_LIMIT_MB_S, 0 = unlimited; live MB/s dial + net readout in the control UI) is charged by every still download (streamed chunk reads) and every ffmpeg video stream (metered from outside via /proc/<pid>/io and SIGSTOP/SIGCONTed into budget). Why: D1 re-measurement 2026-07-02 — the idle link moves ~38 MB/s, but 8 unthrottled downloaders bufferbloated it to ~1-1.5 MB/s PER STREAM (operator's browser included). Capping the aggregate keeps the desktop usable and still beats the collapsed sweep throughput it replaces. Agent build 2026-07-02.4. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
2026-07-02 11:20:45 -04:00
parent 31c416bc7b
commit 95d2ae1d58
8 changed files with 304 additions and 8 deletions
@@ -38,6 +38,10 @@ services:
      # spot + backs off under VRAM pressure). On by default; toggle live in the
      # control UI. Set to 0 to start in manual mode.
      AUTO_SCALE: ${AUTO_SCALE:-1}
      # Aggregate download cap in MB/s (stills + video streams combined), so the
      # agent can't saturate the desktop's network and wreck browsing — WiFi
      # especially. 0 = unlimited; tunable live in the control UI.
      BANDWIDTH_LIMIT_MB_S: ${BANDWIDTH_LIMIT_MB_S:-8}
      # Crop embedder (SigLIP concept bag): float16 keeps VRAM low on a shared
      # desktop GPU; the model itself is announced by the server.
      SIGLIP_DTYPE: ${SIGLIP_DTYPE:-float16}
@@ -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-02.3 · poison guard: repeated transient failures fail the job (with reason) instead of releasing it forever"
+VERSION = "2026-07-02.4 · bandwidth governor: aggregate download cap (MB/s dial) so the agent can't saturate the desktop's network"
 logbuf.install()
 cfg = Config.from_env()
@@ -83,6 +83,13 @@ async def auto(request: Request):
    return JSONResponse(worker.status())
@app.post("/bandwidth")
 async def bandwidth(request: Request):
    body = await request.json()
    worker.set_bandwidth(float(body.get("value", 0)))
    return JSONResponse(worker.status())
@app.get("/gpu")
 def gpu():
    # GPU meters poll this on their own fast cadence. It only reads local
@@ -161,8 +168,9 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 .step{background:#262c34;color:var(--fg);border:1px solid var(--bd);border-radius:8px;
  width:30px;height:32px;font:700 16px system-ui;cursor:pointer}
 .step:hover{border-color:var(--acc)}
- #conc{width:3.4rem;height:32px;text-align:center;font:700 16px system-ui;background:#11151a;
+ #conc,#bw{width:3.4rem;height:32px;text-align:center;font:700 16px system-ui;background:#11151a;
  color:var(--fg);border:1px solid var(--bd);border-radius:8px}
 .unit{color:var(--mut);font-size:12px;font-weight:600}
 .hint{color:var(--mut);font-size:12px;margin-top:12px}
 .tiles{display:grid;grid-template-columns:repeat(6,1fr);gap:8px;margin-bottom:16px}
 .tile{background:#13171d;border:1px solid var(--bd);border-radius:10px;padding:12px 8px;text-align:center}
@@ -216,6 +224,10 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
    <input id=conc type=number min=1 value=1 onchange="setv(this.value)">
    <button class=step onclick=setc(1)>+</button>
   </div>
   <div class=stepper title="aggregate download cap, downloads + video streams combined — 0 = unlimited">
    <input id=bw type=number min=0 step=1 value=8 onchange="setbw(this.value)">
    <span class=unit>MB/s</span>
   </div>
  </div>
  <div class=hint id=conchint>auto-tuning downloaders to keep the GPU fed · max 8</div>
 </section>
@@ -281,6 +293,11 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  await fetch('/auto',{method:'POST',headers:{'Content-Type':'application/json'},
   body:JSON.stringify({value:on})});refresh()
 }
 async function setbw(v){
  v=Math.max(0,parseFloat(v)||0); bw.value=v
  await fetch('/bandwidth',{method:'POST',headers:{'Content-Type':'application/json'},
   body:JSON.stringify({value:v})});refresh()
 }
 async function refresh(){
  let s; try{ s=await (await fetch('/status')).json() }catch{ return }
  applyStatus(s)
@@ -318,6 +335,9 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  // 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)
   +' · net '+(s.net_mb_s!=null?s.net_mb_s.toFixed(1):'—')+' MB/s'
   +(s.bandwidth_limit_mb_s>0?(' / cap '+s.bandwidth_limit_mb_s):'')
  if(document.activeElement!==bw && s.bandwidth_limit_mb_s!=null) bw.value=s.bandwidth_limit_mb_s
  // 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+'%' }
@@ -101,15 +101,26 @@ class FcClient:
            return
        self._post_quiet("/api/gpu/jobs/release", {"job_ids": job_ids})
-    def fetch_image(self, image_url: str) -> bytes:
+    def fetch_image(self, image_url: str, throttle=None) -> bytes:
        # image_url is a server-relative path ("/images/...").
        # timeout=(connect, read): the read timeout is BETWEEN-BYTES, not total,
        # so a large-but-flowing download still completes — but a stuck/dead
        # connection (curator overloaded) fails in 60s instead of hanging a
        # downloader for 180s and piling up concurrent stuck requests on curator.
-        r = self.s.get(f"{self.base}{image_url}", timeout=(10, 60))
+        # With a throttle (the worker's shared TokenBucket), the body is streamed
-        r.raise_for_status()
+        # in chunks and each chunk is charged to the global bandwidth budget —
-        return r.content
+        # pausing between reads lets TCP flow control pace curator's send side.
        with self.s.get(
            f"{self.base}{image_url}", timeout=(10, 60), stream=throttle is not None
        ) as r:
            r.raise_for_status()
            if throttle is None:
                return r.content
            buf = bytearray()
            for chunk in r.iter_content(chunk_size=262_144):
                throttle.take(len(chunk))
                buf.extend(chunk)
            return bytes(buf)
    def is_reachable(self) -> bool:
        """Cheap 'is curator responding at all right now?' check. Used to decide,
@@ -48,6 +48,9 @@ class Config:
    #                             higher keeps more frames, 0 disables
    ffmpeg_timeout: float  # hard ceiling (s) for ffmpeg-from-URL video sampling;
    #                        generous so a SLOW media link still completes
    bandwidth_limit_mb_s: float  # aggregate download cap in MEGABYTES/s across
    #                              all downloaders + video streams (0 = unlimited);
    #                              tunable live from the agent UI
    @classmethod
    def from_env(cls) -> Config:
@@ -77,4 +80,11 @@ class Config:
            dedupe_iou=float(os.environ.get("DEDUPE_IOU", "0.85")),
            frame_dedupe_distance=int(os.environ.get("FRAME_DEDUPE_DISTANCE", "8")),
            ffmpeg_timeout=float(os.environ.get("FFMPEG_TIMEOUT", "1200")),
            # Default 8 MB/s (~64 Mbit/s): ~20% of the measured ~300 Mbit/s home
            # WiFi, so browsing stays snappy while the agent works — yet MORE
            # sweep throughput than the self-inflicted congestion collapse this
            # replaces (2026-07-02: 8 unthrottled downloaders bufferbloated the
            # link to ~1-1.5 MB/s per stream, browser included). Raise it (or 0)
            # from the agent UI on wired/faster networks.
            bandwidth_limit_mb_s=float(os.environ.get("BANDWIDTH_LIMIT_MB_S", "8")),
        )
@@ -4,12 +4,15 @@ instances, each with a timestamp."""
 import io
 import logging
 import os
 import signal
 import subprocess
 import tempfile
 import time
 from PIL import Image, ImageFile
 from .throttle import PidReadMeter
 log = logging.getLogger("fc_agent.media")
 # Load slightly-truncated images (a few missing trailing bytes) instead of
@@ -111,6 +114,12 @@ def _collect_frames(
 def _terminate(proc: subprocess.Popen) -> None:
    """Stop an ffmpeg cleanly, then hard-kill if it ignores SIGTERM."""
    try:
        # A bandwidth-paused (SIGSTOPped) process can't receive SIGTERM until it
        # resumes — always CONT first so termination is prompt, not queued.
        proc.send_signal(signal.SIGCONT)
    except OSError:
        pass
    proc.terminate()
    try:
        proc.wait(timeout=2)
@@ -122,9 +131,33 @@ def _terminate(proc: subprocess.Popen) -> None:
            pass
 def _pause(proc: subprocess.Popen, seconds: float, should_stop) -> bool:
    """SIGSTOP ffmpeg for ~`seconds` of bandwidth debt, staying responsive to
    Stop. While paused, the kernel socket buffer fills and TCP flow control
    stalls curator's send side — that's the throttle. SIGCONT is ALWAYS sent
    before returning. False = a Stop arrived mid-pause."""
    try:
        proc.send_signal(signal.SIGSTOP)
    except OSError:
        return True                     # already exited — nothing to pause
    try:
        end = time.monotonic() + seconds
        while (left := end - time.monotonic()) > 0:
            if should_stop and should_stop():
                return False
            time.sleep(min(0.5, left))
        return True
    finally:
        try:
            proc.send_signal(signal.SIGCONT)
        except OSError:
            pass
 def sample_frames_from_url(
    url: str, interval_seconds: float, max_frames: int,
    *, headers: str = "", timeout: float = 1200.0, should_stop=None,
    governor=None,
 ) -> tuple[list[tuple[float, Image.Image]], str | None]:
    """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
@@ -133,7 +166,10 @@ def sample_frames_from_url(
    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.
+    agent "working" long after Stop. `governor` (the worker's shared TokenBucket)
    meters ffmpeg's network reads from outside via /proc/<pid>/io and SIGSTOPs
    the process into budget, so video streaming honors the same aggregate
    bandwidth cap as still downloads.
    Returns (frames, reason): frames is empty on failure/stop/timeout, and
    `reason` then carries the SPECIFIC cause (ffmpeg's stderr tail / timeout) so
@@ -168,6 +204,7 @@ def sample_frames_from_url(
                    cmd, stdin=subprocess.DEVNULL,
                    stdout=subprocess.DEVNULL, stderr=errf,
                )
                meter = PidReadMeter(proc.pid) if governor is not None else None
                # 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()
@@ -184,6 +221,17 @@ def sample_frames_from_url(
                            log.warning("ffmpeg timed out after %.0fs: %s",
                                        timeout, url)
                            return [], f"ffmpeg timed out after {timeout:.0f}s"
                        if meter is not None:
                            read = meter.delta()
                            if read is None:      # /proc gone → stop governing
                                meter = None
                            elif (debt := governor.charge(read)) > 0:
                                # Over budget: pause ffmpeg until the bucket
                                # recovers. Pause time counts toward `timeout`
                                # (it stays the wedge backstop either way).
                                if not _pause(proc, debt, should_stop):
                                    _terminate(proc)
                                    return [], "stopped"
        except (OSError, ValueError) as exc:
            return [], f"ffmpeg not runnable: {exc}"
        frames = _collect_frames(tmp, interval, cap)
@@ -0,0 +1,111 @@
 """Global download-bandwidth governor (one token bucket for the whole agent).
 The agent lives on someone's desktop and shares that desktop's network —
 typically WiFi, where saturating the link doesn't just slow other apps: it
 bufferbloats the airtime (RTT 21→45ms) and collapses EVERY connection,
 the operator's browser included. Measured 2026-07-02: the idle link moved
 ~38 MB/s single-stream, but under the 8-downloader sweep every stream on the
 machine crawled at ~1-1.5 MB/s. So the cap is on the AGGREGATE, not per
 stream: still downloads pump their chunks through take(), and ffmpeg video
 streams — whose sockets live in a subprocess we can't wrap — are metered from
 outside via /proc/<pid>/io and paused (SIGSTOP) into budget using charge()'s
 debt signal; TCP flow control then stalls the sender while ffmpeg sleeps.
 Accounting is post-paid (charge the bytes first, then wait out any debt): the
 bytes have already crossed the network by the time we count them, and it means
 a chunk larger than one second of budget can never deadlock the bucket.
 Stdlib-only on purpose — unit-tested in CI, where the agent's ML deps
 don't exist.
 """
 import threading
 import time
 class TokenBucket:
    """Thread-safe token bucket in bytes/second. rate 0 = unlimited.
    `consumed` is the monotonic total of bytes charged (throttled or not) —
    the worker's rate loop derives the UI's "net MB/s" readout from it.
    """
    def __init__(self, rate_bytes_per_s: float = 0.0):
        self._cond = threading.Condition()
        self._rate = max(0.0, float(rate_bytes_per_s))
        # Burst = one second of budget: enough that chunked reads stay smooth,
        # small enough that a burst can't meaningfully lift the average.
        self._level = self._rate
        self._stamp = time.monotonic()
        self.consumed = 0
    @property
    def rate(self) -> float:
        return self._rate
    def set_rate(self, rate_bytes_per_s: float) -> None:
        """Retune live (the UI dial). Waiters re-check immediately, so raising
        the cap (or lifting it with 0) unblocks a mid-download wait at once."""
        with self._cond:
            self._refill_locked()          # settle elapsed time at the OLD rate
            self._rate = max(0.0, float(rate_bytes_per_s))
            self._level = min(self._level, self._rate)
            self._cond.notify_all()
    def _refill_locked(self) -> None:
        now = time.monotonic()
        self._level = min(self._rate, self._level + (now - self._stamp) * self._rate)
        self._stamp = now
    def take(self, n: int) -> None:
        """Charge n bytes and block until the budget recovers (stills path)."""
        with self._cond:
            self.consumed += n
            if self._rate <= 0:
                return
            self._refill_locked()
            self._level -= n
            while self._level < 0:
                # Wake early on set_rate; cap the wait so a big debt is paid in
                # re-checked slices rather than one long uninterruptible sleep.
                self._cond.wait(min(-self._level / self._rate, 0.5))
                if self._rate <= 0:
                    return
                self._refill_locked()
    def charge(self, n: int) -> float:
        """Charge n bytes WITHOUT blocking; return seconds of debt (0 = within
        budget). The ffmpeg governor can't block the subprocess's own reads, so
        it SIGSTOPs the process for (about) the returned debt instead."""
        with self._cond:
            self.consumed += n
            if self._rate <= 0:
                return 0.0
            self._refill_locked()
            self._level -= n
            return max(0.0, -self._level / self._rate)
 class PidReadMeter:
    """Cumulative read-bytes meter for a subprocess, via /proc/<pid>/io.
    `rchar` counts every read() syscall's bytes — for a streaming ffmpeg the
    network reads dominate, so the delta is a good-enough aggregate-bandwidth
    signal (it's a governor, not a billing meter). Returns None when /proc is
    unavailable (process exited, or a non-Linux host): the caller then simply
    doesn't govern — degrade to unthrottled rather than break video sampling.
    """
    def __init__(self, pid: int):
        self._path = f"/proc/{pid}/io"
        self._last = 0
    def delta(self) -> int | None:
        try:
            with open(self._path, "rb") as f:
                for line in f:
                    if line.startswith(b"rchar:"):
                        total = int(line.split()[1])
                        d, self._last = total - self._last, total
                        return max(0, d)
        except (OSError, ValueError):
            return None
        return None
@@ -42,6 +42,7 @@ from .client import FcClient
 from .config import Config
 from .crops import crop_region
 from .detectors import dedupe_crops
 from .throttle import TokenBucket
 # Cap on the lease-retry backoff: when curator is unreachable (e.g. you redeploy
 # it while away), a downloader retries leasing with exponential backoff up to this
@@ -181,6 +182,10 @@ class Worker:
            f"Authorization: Bearer {cfg.token}\r\n" if cfg.token else ""
        )
        self._lock = threading.Lock()
        # ONE bandwidth budget for everything the agent pulls (still downloads
        # and ffmpeg video streams): the agent shares a desktop's network, so
        # the polite bound is on the aggregate — see throttle.py for why.
        self.throttle = TokenBucket(cfg.bandwidth_limit_mb_s * 1_048_576)
        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
@@ -219,6 +224,7 @@ class Worker:
        # often the browser polls (see RATE_INTERVAL). Decay to 0 when work stops.
        self._jpm = 0.0
        self._dpm = 0.0
        self._net_mb_s = 0.0      # smoothed aggregate download rate (UI readout)
        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
@@ -382,6 +388,7 @@ class Worker:
        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()
        prev_b = self.throttle.consumed
        while True:
            time.sleep(RATE_INTERVAL)
            now = time.monotonic()
@@ -389,9 +396,12 @@ class Worker:
            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)
                nb = max(0.0, (self.throttle.consumed - prev_b) / dt / 1_048_576)
                self._jpm = RATE_ALPHA * jp + (1 - RATE_ALPHA) * self._jpm
                self._dpm = RATE_ALPHA * dp + (1 - RATE_ALPHA) * self._dpm
                self._net_mb_s = RATE_ALPHA * nb + (1 - RATE_ALPHA) * self._net_mb_s
            prev_p, prev_d, prev_t = self.processed, self.downloaded, now
            prev_b = self.throttle.consumed
    # --- control -----------------------------------------------------------
    def start(self):
@@ -494,6 +504,14 @@ class Worker:
        with self._lock:
            self._auto = bool(on)
    def set_bandwidth(self, mb_s: float):
        # Live-retunes the shared bucket; a downloader blocked mid-wait re-checks
        # immediately (set_rate notifies), so raising the cap takes effect now.
        self.throttle.set_rate(max(0.0, float(mb_s)) * 1_048_576)
        log.info("bandwidth cap set to %s",
                 "unlimited" if self.throttle.rate <= 0
                 else f"{self.throttle.rate / 1_048_576:g} MB/s")
    def set_concurrency(self, n: int):
        # The UI dial tunes the DOWNLOADER count. A manual set is an override →
        # leave Auto so the autoscaler stops fighting the operator.
@@ -564,6 +582,8 @@ class Worker:
            "downloads_per_min": round(self._dpm, 1),
            "errors": self.errors,
            "transient": self.transient,
            "bandwidth_limit_mb_s": round(self.throttle.rate / 1_048_576, 1),
            "net_mb_s": round(self._net_mb_s, 1),       # observed aggregate rate
        }
    def _bump(self, *, processed=0, downloaded=0, errors=0, active=0, transient=0):
@@ -682,6 +702,7 @@ class Worker:
                    job.get("max_frames", 64),
                    headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout,
                    should_stop=lambda: self._stopped(stop_evt),
                    governor=self.throttle,
                )
            if not frames:
                # Stop killed ffmpeg → NOT the job's fault; raise transient so the
@@ -709,7 +730,7 @@ class Worker:
            return frames
        # Stills: download the bytes and decode.
        with self._timed("download"):
-            data = self.client.fetch_image(job["image_url"])
+            data = self.client.fetch_image(job["image_url"], throttle=self.throttle)
        with self._timed("decode"):
            frames = [(None, media.load_image(data))]
        return frames
@@ -0,0 +1,71 @@
 """Agent bandwidth governor (TokenBucket + PidReadMeter) — pure stdlib, so it
 runs in the unit lane even though the rest of the agent (torch/ultralytics)
 can't be imported in CI. Timing asserts use generous margins: they check the
 throttle's ORDER of magnitude, not scheduler precision."""
 import os
 import threading
 import time
 from agent.fc_agent.throttle import PidReadMeter, TokenBucket
 def test_unlimited_never_blocks_and_counts():
    b = TokenBucket(0)
    t0 = time.monotonic()
    for _ in range(100):
        b.take(10_000_000)
    assert time.monotonic() - t0 < 0.2
    assert b.consumed == 1_000_000_000
 def test_take_enforces_average_rate():
    # 400 KB/s with a 1s burst: the first 400KB is free (burst), the next
    # 200KB is 0.5s of debt the caller must wait out.
    b = TokenBucket(400_000)
    b.take(400_000)
    t0 = time.monotonic()
    b.take(200_000)
    elapsed = time.monotonic() - t0
    assert 0.35 <= elapsed < 3.0
 def test_set_rate_zero_unblocks_a_waiter():
    b = TokenBucket(1_000)            # 1 KB/s → a 1MB take would wait ~17 min
    started = threading.Event()
    def blocked_take():
        started.set()
        b.take(1_000_000)
    th = threading.Thread(target=blocked_take, daemon=True)
    th.start()
    started.wait(1.0)
    time.sleep(0.1)                   # let it enter the wait loop
    b.set_rate(0)                     # lift the cap live (the UI dial)
    th.join(timeout=2.0)
    assert not th.is_alive()
 def test_charge_reports_debt_without_blocking():
    b = TokenBucket(1_000)
    t0 = time.monotonic()
    assert b.charge(1_000) == 0.0     # covered by the 1s burst
    debt = b.charge(2_000)            # 2s over budget → ~2s of debt
    assert time.monotonic() - t0 < 0.2
    assert 1.5 <= debt <= 2.1
    assert b.consumed == 3_000
 def test_pid_read_meter_tracks_own_reads():
    m = PidReadMeter(os.getpid())
    first = m.delta()                 # cumulative rchar since process start
    assert first is not None and first > 0
    with open("/dev/zero", "rb") as f:
        f.read(1_048_576)
    grown = m.delta()
    assert grown is not None and grown >= 1_048_576
 def test_pid_read_meter_missing_pid_degrades_to_none():
    # PID 2^22+ is above the default pid_max — guaranteed absent.
    assert PidReadMeter(2**22 + 12345).delta() is None