From 79269da8025e896a3890e74ffa460d9bb9b27b43 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Tue, 30 Jun 2026 21:39:00 -0400
Subject: [PATCH] fix(agent): prompt stop + lazy curator polling + build
 marker; add agent to CI
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Addresses operator reports: Stop never finishes, the agent polls curator
constantly, and stale-cached pages get mistaken for a failed deploy.

- Stop is prompt: flip _running BEFORE any lock so /status + worker loops see
  "stopped" immediately, and add a stop/shrink checkpoint in _process (after
  decode, before the expensive detect+embed) that releases the job and bails —
  so a Stop doesn't wait out heavy GPU work.
- Lazy curator polling: the queue snapshot is fetched only while a browser is
  actually watching (a /status hit within UI_IDLE_GRACE) and on a 5s cadence,
  not a constant background loop. The work loop's own lease/submit is curator's
  only visitor otherwise — nothing polls just to poll.
- Build marker: VERSION is embedded in the page and reported on /status; the UI
  shows a "reload" banner when they differ, so a browser-cached page can't be
  mistaken for "the new image didn't deploy" (complements the no-store header).

CI: the lint lane now also `ruff check`s agent/ and compileall-parses it, so the
GPU agent is linted + syntax-checked before its image builds (build.yml only
`docker build`s it). Fixed the agent's pre-existing UP037/B905 so it passes.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
---
 .forgejo/workflows/ci.yml |  9 ++++++-
 agent/fc_agent/app.py     | 18 ++++++++++++--
 agent/fc_agent/config.py  |  2 +-
 agent/fc_agent/worker.py  | 50 +++++++++++++++++++++++++++++----------
 4 files changed, 62 insertions(+), 17 deletions(-)

diff --git a/.forgejo/workflows/ci.yml b/.forgejo/workflows/ci.yml
index 9a31d7c..ec1dcbf 100644
--- a/.forgejo/workflows/ci.yml
+++ b/.forgejo/workflows/ci.yml
@@ -27,7 +27,14 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Ruff lint
-        run: ruff check backend/ tests/ alembic/
+        # agent/ included so the GPU-agent is linted before its image is built
+        # (build.yml only `docker build`s it — this is where it gets checked).
+        run: ruff check backend/ tests/ alembic/ agent/
+      - name: Agent syntax check
+        # The agent's runtime deps (torch/transformers/ultralytics) aren't in the
+        # CI image, so we can't import it — but compileall parses every module,
+        # catching syntax errors before the image build.
+        run: python -m compileall -q agent/fc_agent
 
   backend-lint-and-test:
     runs-on: python-ci
diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py
index 87be87b..f52435a 100644
--- a/agent/fc_agent/app.py
+++ b/agent/fc_agent/app.py
@@ -12,6 +12,11 @@ from .config import Config
 from .gpu import read_gpu
 from .worker import Worker
 
+# 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-06-30.5 · stop+poll+ci"
+
 logbuf.install()
 cfg = Config.from_env()
 worker = Worker(cfg)
@@ -41,7 +46,7 @@ def _maybe_autostart() -> None:
 
 @app.get("/", response_class=HTMLResponse)
 def index() -> str:
-    return _PAGE
+    return _PAGE.replace("__BUILD__", VERSION)
 
 
 @app.post("/start")
@@ -92,10 +97,12 @@ def status():
     # Pure in-memory read: worker.status() is lock-free and the queue snapshot is
     # kept fresh by a background poller — NO inline curator call, so this can't
     # stall the status view when curator is buried under a big backlog.
+    worker.note_ui()          # a browser is watching → keep the queue snapshot warm
     s = worker.status()
     s["fc_url"] = cfg.fc_url
     s["configured"] = bool(cfg.token)
     s["queue"] = worker.latest_queue()
+    s["build"] = VERSION
     return JSONResponse(s)
 
 
@@ -179,8 +186,11 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
   <div class=brand><span class=logo>◆</span> FabledCurator <span class=sub>GPU agent</span></div>
   <div class=conn><span class="dot" id=dot></span><span id=connlbl>—</span></div>
  </header>
- <p class=meta>Server <code id=fc>—</code> · token <code id=cfg>—</code></p>
+ <p class=meta>Server <code id=fc>—</code> · token <code id=cfg>—</code> · build <code id=build>__BUILD__</code></p>
 
+ <div id=verbanner class=banner style="display:none;background:#3a1212;border-color:#5a1717;color:#ffb3b3">
+  a newer agent version is running — reload this page (Ctrl+Shift+R) to update the controls
+ </div>
  <div id=banner class=banner style=display:none>
   curator unreachable — holding work + retrying, resumes on its own (no restart needed)
  </div>
@@ -227,6 +237,7 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  </section>
 </div>
 <script>
+ const PAGE_BUILD="__BUILD__"
  let CAP=8
  // 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
@@ -262,6 +273,9 @@ _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'
+  // 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
diff --git a/agent/fc_agent/config.py b/agent/fc_agent/config.py
index 5ab7a56..059896d 100644
--- a/agent/fc_agent/config.py
+++ b/agent/fc_agent/config.py
@@ -31,7 +31,7 @@ class Config:
     max_panels: int       # cap panel crops per frame
 
     @classmethod
-    def from_env(cls) -> "Config":
+    def from_env(cls) -> Config:
         return cls(
             fc_url=os.environ.get("FC_URL", "http://localhost:8000").rstrip("/"),
             token=os.environ.get("FC_TOKEN", ""),
diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py
index ba787e1..f09b254 100644
--- a/agent/fc_agent/worker.py
+++ b/agent/fc_agent/worker.py
@@ -28,7 +28,7 @@ from .crops import crop_region
 MAX_BACKOFF_SECONDS = 60.0
 
 
-def _is_transient(exc: "requests.RequestException") -> bool:
+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
@@ -75,6 +75,13 @@ REPROBE_TICKS = 8             # decisions to hold after settling before re-probi
 # data that decides whether a download/compute split is worth building.
 STATS_INTERVAL = 30.0
 
+# The queue snapshot exists only to populate the UI's counts, so it's polled
+# lazily — only while a browser is actually watching (a /status hit in the last
+# UI_IDLE_GRACE seconds), and not on a tight loop. The work loop's own lease/
+# submit calls are the real "is curator up?" signal; nothing polls just to poll.
+QUEUE_POLL_INTERVAL = 5.0
+UI_IDLE_GRACE = 20.0
+
 log = logging.getLogger("fc_agent.worker")
 
 
@@ -109,6 +116,7 @@ class Worker:
         # /status read is instant — never an inline curator HTTP call (which
         # stalls the whole status view when curator is busy).
         self._queue: dict | None = None
+        self._ui_seen = 0.0        # monotonic time of the last UI /status hit
         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
@@ -126,15 +134,21 @@ class Worker:
         self._proposers_lock = threading.Lock()
 
     def _queue_poll_loop(self):
-        """Refresh the curator queue snapshot in the background (every ~3s) so
-        /status is a pure in-memory read. Errors just leave the last snapshot
-        (or None) — the UI shows 'unreachable' without ever blocking."""
+        """Refresh the curator queue snapshot so /status is a pure in-memory read
+        — but ONLY while the UI is being watched (a recent /status hit). No
+        browser open → no polling; the work loop is curator's only visitor.
+        Errors just leave the last snapshot (or None) — never blocks the UI."""
         while True:
-            try:
-                self._queue = self.client.queue_status()
-            except Exception:
-                self._queue = None
-            time.sleep(3.0)
+            if time.monotonic() - self._ui_seen <= UI_IDLE_GRACE:
+                try:
+                    self._queue = self.client.queue_status()
+                except Exception:
+                    self._queue = None
+            time.sleep(QUEUE_POLL_INTERVAL)
+
+    def note_ui(self) -> None:
+        """The UI polled /status — keep the queue snapshot warm for a while."""
+        self._ui_seen = time.monotonic()
 
     def latest_queue(self) -> dict | None:
         return self._queue
@@ -189,9 +203,12 @@ class Worker:
             self._ctrl_thread.start()
 
     def stop(self):
+        # Flip the flag FIRST (atomic bool), before any lock, so /status and the
+        # worker 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()
         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
@@ -272,7 +289,7 @@ class Worker:
             for job in jobs:
                 if slot.stop.is_set() or not self._running:
                     break
-                ok = self._process(job)
+                ok = self._process(job, slot)
                 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)
@@ -407,7 +424,7 @@ class Worker:
                 self._proposers = Proposers(self.cfg)
         return self._proposers
 
-    def _process(self, job: dict) -> bool:
+    def _process(self, job: dict, slot: _Slot) -> 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
@@ -429,6 +446,13 @@ class Worker:
                 frames = [(None, media.load_image(data))]
             self._record("decode", time.monotonic() - _t)
 
+            # Stop/shrink checkpoint: bail BEFORE the expensive detect+embed so a
+            # Stop finishes promptly instead of waiting out heavy GPU work. Hand
+            # the job back to pending for another agent.
+            if not self._running or slot.stop.is_set():
+                self.client.release([job["job_id"]])
+                return True
+
             task = job.get("task") or "ccip"
             embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION
             model_name = (
@@ -527,7 +551,7 @@ class Worker:
                         }))
                 if pending:
                     vecs = embedder.embed_batch([c for c, _ in pending])
-                    for (_c, tmpl), vec in zip(pending, vecs):
+                    for (_c, tmpl), vec in zip(pending, vecs, strict=True):
                         tmpl["siglip_embedding"] = vec
                         tmpl["embedding_version"] = embed_version
                         regions.append(tmpl)