diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py
index 2c28e04..c6cb747 100644
--- a/agent/fc_agent/app.py
+++ b/agent/fc_agent/app.py
@@ -6,6 +6,8 @@ bandwidth for throughput), and watch GPU load + buffer occupancy + progress +
the server-side queue. Config is env-seeded; the downloader count is adjustable
here on the fly (GPU consumers autoscale between 1 and 2 on their own).
"""
+import logging
+
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse
@@ -14,10 +16,12 @@ from .config import Config
from .gpu import read_gpu
from .worker import Worker
+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-01.7 · status: smoothed jobs/min + downloads/min rates"
+VERSION = "2026-07-01.8 · real start/stop state machine (no more stuck 'stopping')"
logbuf.install()
cfg = Config.from_env()
@@ -53,12 +57,14 @@ def index() -> str:
@app.post("/start")
def start():
+ log.info("UI: Start button pressed") # the press; worker logs the transition
worker.start()
return JSONResponse(worker.status())
@app.post("/stop")
def stop():
+ log.info("UI: Stop button pressed")
worker.stop()
return JSONResponse(worker.status())
@@ -260,10 +266,12 @@ _PAGE = """
// /status refresh (now always fast) recovers the true state either way.
const ac=new AbortController(); const to=setTimeout(()=>ac.abort(),8000)
try{ applyStatus(await (await fetch('/'+p,{method:'POST',signal:ac.signal})).json()) }
- catch{ /* leave the periodic refresh to recover the real state */ }
- finally{ clearTimeout(to); startbtn.disabled=false; stopbtn.disabled=false }
+ catch{ refresh() /* on abort/error, repaint the real state from /status */ }
+ finally{ clearTimeout(to) }
}
function pending(label){
+ // Instant optimistic feedback on click; applyStatus (POST response, then the
+ // periodic poll) then owns the real state + which buttons are enabled.
state.textContent=label; state.className='n busy'
dot.className='dot amber'
startbtn.disabled=true; stopbtn.disabled=true
@@ -284,15 +292,22 @@ _PAGE = """
}
function applyStatus(s){
CAP=s.max_concurrency||8; capn.textContent=CAP
- const running=s.state==='running'
+ // The backend owns the state now (stopped|starting|running|stopping) and drives
+ // every transition, so the pill is always truthful — no client-side guessing
+ // from active>0, which used to wedge on "stopping" forever.
+ const st=s.state||'stopped'
+ const running=st==='running'
+ const busy=(st==='starting'||st==='stopping')
// 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
- state.textContent=draining?'stopping':s.state
- state.className='n'+(draining?' busy':'')
+ state.textContent=st
+ state.className='n'+(busy?' busy':'')
+ // Buttons follow the real state so you can't fight a transition: Start only
+ // from stopped; Stop only while up; both disabled through "stopping" until the
+ // backend truthfully lands on "stopped".
+ startbtn.disabled=(st!=='stopped')
+ stopbtn.disabled=!(running||st==='starting')
// Derived rates over the poll deltas — j/min ≈ GPU throughput, dl/min ≈ fetch
// throughput. EWMA-smoothed so a lull between jobs doesn't blank them; clamped
// at 0 so an agent restart (counters reset to 0 → negative delta) can't spike
@@ -331,9 +346,16 @@ _PAGE = """
if('configured' in s){
const ok=s.configured
fc.textContent=s.fc_url; cfg.textContent=ok?'set':'MISSING'
- dot.className='dot '+(!ok?'red':(running?(s.queue?'green':'amber'):'amber'))
- connlbl.textContent=!ok?'no token':(running?(s.queue?'running':'running · curator unreachable'):'stopped')
- banner.style.display=(running && !s.queue)?'block':'none'
+ // Pill colour + label track the real state: green only when running AND
+ // curator is answering; amber for the transient states + a running-but-
+ // unreachable curator; grey when stopped; red with no token.
+ let dc='dot', lbl='stopped'
+ if(!ok){ dc='dot red'; lbl='no token' }
+ else if(st==='running'){ dc='dot '+(s.queue?'green':'amber'); lbl=s.queue?'running':'running · curator unreachable' }
+ else if(st==='starting'){ dc='dot amber'; lbl='starting…' }
+ else if(st==='stopping'){ dc='dot amber'; lbl='stopping…' }
+ dot.className=dc; connlbl.textContent=lbl
+ banner.style.display=(st==='running' && !s.queue)?'block':'none'
queue.textContent=s.queue?('queue · pending '+s.queue.pending+' · in flight '+s.queue.leased+' · done '+s.queue.done+' · errored '+s.queue.error):'queue · unreachable'
}
}
diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py
index 8f57faa..27fd09c 100644
--- a/agent/fc_agent/worker.py
+++ b/agent/fc_agent/worker.py
@@ -133,6 +133,23 @@ STATS_INTERVAL = 30.0
QUEUE_POLL_INTERVAL = 5.0
UI_IDLE_GRACE = 20.0
+# Stop backstop: pressing Stop signals every worker thread to wind down and hand
+# its lease back; the UI shows "stopping" until those threads have actually
+# exited, then "stopped" — so the state the operator sees is always truthful.
+# If a thread wedges (a stuck submit/release to an overloaded curator), NEVER
+# hang in "stopping": declare "stopped" after this many seconds and let the lone
+# straggler finish detached. This is the guarantee the old active>0 UI hack lacked.
+STOPPING_TIMEOUT = 20.0
+
+# The lifecycle the operator sees. `_state` is the source of truth the UI reads;
+# `_running` is the low-level flag the worker THREADS gate on (True while
+# starting/running; False from the Stop press onward, so they wind down at once).
+# stopped → starting (spun up, not yet reached curator)
+# → running (leased at least once — curator is answering)
+# → stopping (Stop pressed; threads winding down + handing work back)
+# → stopped (threads exited, or the timeout backstop fired)
+STOPPED, STARTING, RUNNING, STOPPING = "stopped", "starting", "running", "stopping"
+
log = logging.getLogger("fc_agent.worker")
@@ -147,6 +164,12 @@ class Worker:
)
self._lock = threading.Lock()
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
+ # the worker threads have actually exited (or the timeout backstop fires),
+ # then STOPPED. `_running` flips False the instant Stop is pressed.
+ self._state = STOPPED
+ self._shutdown_thread: threading.Thread | None = None
self._auto = bool(cfg.auto_scale) # autoscale the downloader count
self._dl_target = max(1, min(DL_MAX, cfg.concurrency))
self._consumer_target = 1 # GPU is fast — start with one
@@ -311,42 +334,88 @@ class Worker:
# --- control -----------------------------------------------------------
def start(self):
+ # Only a fully-stopped worker can start — ignore a click while already
+ # up or still winding down (the UI also disables the button then). This
+ # keeps a Start during "stopping" from racing the shutdown monitor's
+ # lease-release against a fresh pool's new leases.
with self._lock:
+ if self._state != STOPPED:
+ return
self._running = True
+ self._state = STARTING # → RUNNING on the first successful lease
self._dl_target = max(1, self._dl_target)
self._consumer_target = max(1, self._consumer_target)
self._reconcile_locked()
+ log.info("▶ start → starting (spinning up, awaiting curator)")
# (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):
- # Flip the flag FIRST (atomic bool), before any lock, so /status and the
- # 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()
+ def _mark_running(self) -> None:
+ """STARTING → RUNNING once a downloader has actually leased from curator —
+ so 'running' means 'curator is answering', not just 'Start was clicked'.
+ First caller wins; the rest are a cheap no-op."""
+ if self._state != STARTING:
+ return
with self._lock:
+ if self._state != STARTING:
+ return
+ self._state = RUNNING
+ log.info("state → running (curator reachable — leasing)")
+
+ def stop(self):
+ # Enter STOPPING and signal every worker thread to wind down. The handler
+ # returns at once — the actual wait-for-exit + lease-release runs in a
+ # background monitor, so a slow curator can NEVER block the Stop button or
+ # wedge the state (the bug: the old UI faked "stopping" from active>0 and
+ # sat there forever when a consumer hung mid-release).
+ with self._lock:
+ if self._state not in (STARTING, RUNNING):
+ return # already stopping/stopped
+ self._state = STOPPING
+ self._running = False # every thread's _stopped() now trips → wind down
dls, self._dls = self._dls, []
cons, self._consumers = self._consumers, []
- self._active = 0 # no consumers left → the meter reads 0 at once;
- # any lagging decrement is clamped (see _bump)
+ self._active = 0
+ self._ctrl_stop.set()
for _, ev in dls:
ev.set()
for _, ev in cons:
ev.set()
- # Wake any consumer blocked on an empty buffer.
+ # Wake any consumer blocked on an empty buffer so it sees the stop at once.
for _ in range(CONSUMER_MAX):
try:
self._buffer.put_nowait(None)
except queue.Full:
break
- # Drain the buffer + release every still-held lease in one shot so orphaned
- # work is re-leased at once. A downloader/consumer mid-flight may also
- # release its own job — a duplicate release is a harmless no-op.
+ log.info("■ stop → stopping (winding down, handing work back)")
+ threads = [t for t, _ in dls] + [t for t, _ in cons]
+ self._shutdown_thread = threading.Thread(
+ target=self._finish_stop, args=(threads,), daemon=True)
+ self._shutdown_thread.start()
+
+ def _finish_stop(self, threads: list[threading.Thread]) -> None:
+ """Wait for the signalled worker threads to actually exit, then land on
+ STOPPED — so 'stopped' is truthful. Bounded by STOPPING_TIMEOUT so a
+ wedged submit/release can never hold the UI in 'stopping' forever."""
+ deadline = time.monotonic() + STOPPING_TIMEOUT
+ for t in threads:
+ t.join(timeout=max(0.0, deadline - time.monotonic()))
+ # Hand back every lease still held in one shot (background — no handler
+ # block); a straggler releasing its own job later is a harmless dup.
self._drain_and_release()
+ with self._lock:
+ if self._state == STOPPING: # a Start may have re-entered; if so, leave it
+ self._state = STOPPED
+ self._active = 0 # any straggler's -1 is clamped (see _bump)
+ stragglers = sum(1 for t in threads if t.is_alive())
+ if stragglers:
+ log.info("state → stopped (%d thread(s) still winding down past %ds — detached)",
+ stragglers, int(STOPPING_TIMEOUT))
+ else:
+ log.info("state → stopped (all work handed back)")
def _drain_and_release(self) -> None:
while True:
@@ -419,7 +488,7 @@ class Worker:
# under the GIL) and this backs the UI poll — it must NEVER be able to
# block behind a thread holding _lock, or the whole status view freezes.
return {
- "state": "running" if self._running else "stopped",
+ "state": self._state, # stopped|starting|running|stopping (truthful)
"concurrency": self._dl_target, # the UI dial = downloader count
"max_concurrency": DL_MAX,
"auto": self._auto,
@@ -456,6 +525,7 @@ class Worker:
with self._timed("lease"):
jobs = self.client.lease(self.cfg.batch_size)
backoff = self.cfg.poll_idle_seconds # server answered → reset
+ self._mark_running() # curator answered → leave "starting"
except Exception:
# curator unreachable (redeploy, network drop): wait it out with
# exponential backoff, capped — resume on our own when it returns.