diff --git a/agent/docker-compose.yml b/agent/docker-compose.yml index 9b1adb1..0f07cfa 100644 --- a/agent/docker-compose.yml +++ b/agent/docker-compose.yml @@ -10,6 +10,13 @@ # 4. Open http://localhost:8770 → Start. Pause/Stop hands the GPU back. # docker compose down to stop the container entirely. # +# Surviving a curator redeploy (you're away, can't touch the agent): +# - A running agent rides out curator being unreachable on its own — it retries +# leasing with capped backoff and resumes when the server is back. In-flight +# work is handed back (not failed), so a redeploy never poisons good jobs. +# - AUTO_START=1 (below) also resumes the worker if the AGENT container itself +# restarts (host reboot / crash via `restart: unless-stopped`) — no click. +# # Needs the NVIDIA Container Toolkit installed on the host for --gpus. services: @@ -24,6 +31,12 @@ services: CCIP_MODEL: ${CCIP_MODEL:-} DETECTOR_LEVEL: ${DETECTOR_LEVEL:-m} BATCH_SIZE: ${BATCH_SIZE:-4} + # Resume the worker automatically on container start (survive a reboot / + # crash-restart while you're away). Set to 0 to require a manual Start. + AUTO_START: ${AUTO_START:-1} + # 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} volumes: # Persist the downloaded ONNX models so restarts are fast. - fc-agent-models:/models diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index c7fb992..9252c2f 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -16,6 +16,16 @@ worker = Worker(cfg) app = FastAPI(title="FabledCurator GPU agent") +@app.on_event("startup") +def _maybe_autostart() -> None: + # With AUTO_START set, a container restart (host reboot, or `restart: + # unless-stopped` after a crash) resumes the worker on its own — the slots + # then ride out a still-down curator via lease backoff. Lets the agent + # survive a redeploy with nobody at the desktop to click Start. + if cfg.auto_start and cfg.token: + worker.start() + + @app.get("/", response_class=HTMLResponse) def index() -> str: return _PAGE @@ -86,6 +96,10 @@ _PAGE = """ 0
active now 0
processed 0
errors + 0
waited out + +
GPU — …
@@ -103,7 +117,10 @@ _PAGE = """ const s=await (await fetch('/status')).json() CAP=s.max_concurrency||8; capn.textContent=CAP state.textContent=s.state; active.textContent=s.active; done.textContent=s.processed - err.textContent=s.errors; fc.textContent=s.fc_url + err.textContent=s.errors; fc.textContent=s.fc_url; wait.textContent=s.transient||0 + // Running but the queue read failed → curator is unreachable; show we're + // riding it out rather than erroring. + banner.style.display=(s.state==='running' && !s.queue)?'block':'none' if(document.activeElement!==conc) conc.value=s.concurrency conc.max=CAP cfg.textContent=s.configured?'set':'MISSING' diff --git a/agent/fc_agent/config.py b/agent/fc_agent/config.py index 0a289fb..42dee5c 100644 --- a/agent/fc_agent/config.py +++ b/agent/fc_agent/config.py @@ -16,6 +16,8 @@ class Config: embed_dtype: str # torch dtype for the crop embedder: float16|float32 embed_model_override: str # force a SigLIP-family model ("" → use the one # the server announces in the lease) + auto_start: bool # start the worker pool on boot (so a container restart + # resumes processing without anyone clicking Start) @classmethod def from_env(cls) -> "Config": @@ -30,4 +32,5 @@ 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"), ) diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 799a3dd..8b42277 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -10,13 +10,32 @@ Stop (or shrinking the pool) RELEASES a slot's still-leased jobs immediately so orphaned work is re-picked at once rather than waiting out the lease. """ import threading -import time + +import requests from . import media, models from .client import FcClient from .config import Config from .crops import crop_region +# Cap on the lease-retry backoff: when curator is unreachable (e.g. you redeploy +# it while away), each slot retries leasing with exponential backoff up to this +# many seconds, then resumes within this window once the server is back — no +# restart needed. +MAX_BACKOFF_SECONDS = 60.0 + + +def _is_transient(exc: "requests.RequestException") -> bool: + """A server/transport problem (wait it out) vs a job-specific fault (fail it). + No response → connection refused/timeout → curator is down → transient. With + a response: 5xx, auth (401/403, e.g. a token blip on redeploy), 408/409/429 + (timeout / our lease reclaimed / rate-limited) are all 'not this job's fault'. + A specific 4xx like 404 (image gone) / 400 IS the job's fault → fail it.""" + resp = getattr(exc, "response", None) + if resp is None: + return True + return resp.status_code >= 500 or resp.status_code in (401, 403, 408, 409, 429) + # Generous cap: the pipeline is usually I/O-bound (downloading + decoding images # over HTTP), so the GPU stays underused until many workers overlap that I/O. # Push it up while watching the GPU util + VRAM in the UI. @@ -49,6 +68,8 @@ class Worker: self._slots: list[_Slot] = [] self.processed = 0 self.errors = 0 + self.transient = 0 # jobs handed back due to a server outage (NOT + # failed) — the "waiting out curator" counter self._active = 0 # slots currently mid-image # The crop embedder (SigLIP-family) is built lazily on the first job that # needs it, from the model the server announces — one shared instance. @@ -92,31 +113,48 @@ class Worker: "active": self._active, "processed": self.processed, "errors": self.errors, + "transient": self.transient, } - def _bump(self, *, processed=0, errors=0, active=0): + def _bump(self, *, processed=0, errors=0, active=0, transient=0): with self._lock: self.processed += processed self.errors += errors + self.transient += transient self._active += active # --- per-slot loop ----------------------------------------------------- def _loop(self, slot: _Slot): + backoff = self.cfg.poll_idle_seconds while not slot.stop.is_set() and self._running: try: jobs = self.client.lease(self.cfg.batch_size) + backoff = self.cfg.poll_idle_seconds # server answered → reset except Exception: - time.sleep(self.cfg.poll_idle_seconds) + # curator unreachable (redeploy, network drop): wait it out with + # exponential backoff, capped — resume on our own when it returns. + self._interruptible_sleep(slot, backoff) + backoff = min(backoff * 2, MAX_BACKOFF_SECONDS) continue if not jobs: - time.sleep(self.cfg.poll_idle_seconds) + self._interruptible_sleep(slot, self.cfg.poll_idle_seconds) continue slot.inflight = [j["job_id"] for j in jobs] for job in jobs: if slot.stop.is_set() or not self._running: break - self._process(job) + ok = self._process(job) slot.inflight = [i for i in slot.inflight if i != job["job_id"]] + if not ok: + # Server went away mid-batch: hand the rest back (best effort) + # and back off instead of hammering a recovering server or + # burning the jobs' attempt budgets on fail(). + if slot.inflight: + self.client.release(slot.inflight) + slot.inflight = [] + self._interruptible_sleep(slot, backoff) + backoff = min(backoff * 2, MAX_BACKOFF_SECONDS) + break if slot.inflight: self.client.heartbeat(slot.inflight) # Graceful hand-back of anything leased but not processed. @@ -124,6 +162,11 @@ class Worker: self.client.release(slot.inflight) slot.inflight = [] + def _interruptible_sleep(self, slot: _Slot, seconds: float): + """Sleep, but wake immediately if the slot is told to stop — so a Stop or + a pool-shrink doesn't hang for a full backoff window.""" + slot.stop.wait(timeout=seconds) + def _ensure_embedder(self, model_name: str): if self._embedder is not None: return self._embedder @@ -133,7 +176,12 @@ class Worker: self._embedder = CropEmbedder(model_name, self.cfg.embed_dtype) return self._embedder - def _process(self, job: dict): + def _process(self, job: dict) -> bool: + """Process one job. Returns True when handled (completed, or hard-failed + because the job itself is bad) and False on a TRANSPORT error (curator + unreachable / 5xx / our lease was reclaimed mid-flight) — which is not + the job's fault, so the caller backs off and the job is left to be + re-leased rather than fail()ed into its attempt budget.""" self._bump(active=1) try: data = self.client.fetch_image(job["image_url"]) @@ -203,8 +251,24 @@ class Worker: }) self.client.submit(job["job_id"], regions, replace_kinds) self._bump(processed=1) - except Exception as exc: # noqa: BLE001 — report + move on + return True + except requests.RequestException as exc: + if _is_transient(exc): + # curator down/redeploying, a 5xx, or our lease was reclaimed + # while we worked. NOT the job's fault — hand it back (best + # effort; no-ops if the server is still down, then the server's + # orphan-recovery reclaims it) and signal the loop to wait. + self._bump(transient=1) + self.client.release([job["job_id"]]) + return False + # A job-specific HTTP fault (404 image gone, 400) → fail it so it + # doesn't re-lease forever. self._bump(errors=1) self.client.fail(job["job_id"], str(exc)[:500]) + return True + except Exception as exc: # noqa: BLE001 — a genuine job fault: report it + self._bump(errors=1) + self.client.fail(job["job_id"], str(exc)[:500]) + return True finally: self._bump(active=-1)