bvandeusen
afef95a87d
The agent workload is download-bound (download 400–5462ms vs GPU ~300–600ms), so the old N-slot serial chain (each slot: lease→download→decode→GPU→submit) left the fast GPU idle during every download. Rearchitect worker.py into a producer/consumer pipeline: downloader pool (autoscaled by BUFFER OCCUPANCY) → bounded queue → 1–2 GPU consumers (detect+embed→submit) - Downloaders are I/O-bound → many overlap; the autoscaler now tunes DOWNLOADER count by buffer fill (empty = GPU starving → add; full = outpacing GPU → add a 2nd consumer if it has util/VRAM headroom and lifts throughput, else trim). - Bounded buffer (12) = backpressure: a full buffer blocks downloaders, capping RAM + lease look-ahead. VRAM pressure sheds a consumer immediately. - Heartbeat thread keeps every held lease alive (buffered jobs wait on the GPU; curator's 180s TTL would otherwise reclaim them mid-buffer). - Preserves all resilience: lease exp-backoff, submit-path retry (#169), release-on-stop, region caps + video early-exit (#171). Stop drains BOTH pools and releases every held lease at once (single held-set as source of truth). - Consumers SHARE one embedder + proposers instance (a 2nd consumer adds concurrent inference, not N× VRAM — bounds the VRAM creep seen with N slots). - UI reworked for the pipeline: tiles show downloaders · buffer · on-GPU · processed · errors, a buffer-occupancy meter, and a consumers/waited-out line; the dial now tunes downloaders. Build marker 2026-07-01.1. Also fix the operator-flagged detector warning: yolo11n + the comic-panel model threw "'Conv' object has no attribute 'bn'" on every image (ultralytics' load- time Conv+BN fusion on a version-mismatched graph), silently disabling 2 of 3 crop proposers and spamming the log per image. Disable that fusion (unfused inference is correct, marginally slower) and permanently self-disable a proposer on the first inference failure instead of re-throwing forever. Refs milestone 122. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
352 lines
17 KiB
Python
352 lines
17 KiB
Python
"""FastAPI control surface for the agent (served on localhost).
|
||
|
||
Start / stop the download→GPU pipeline, tune the downloader count live (the
|
||
workload is download-bound, so downloaders are the dial that trades desktop
|
||
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).
|
||
"""
|
||
from fastapi import FastAPI, Request
|
||
from fastapi.responses import HTMLResponse, JSONResponse
|
||
|
||
from . import logbuf
|
||
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-07-01.1 · download/GPU pipeline"
|
||
|
||
logbuf.install()
|
||
cfg = Config.from_env()
|
||
worker = Worker(cfg)
|
||
app = FastAPI(title="FabledCurator GPU agent")
|
||
|
||
|
||
@app.middleware("http")
|
||
async def _no_store(request, call_next):
|
||
# The control page is a static string and the status/gpu/logs polls are
|
||
# live data — never let the browser cache either, or a freshly-pulled agent
|
||
# image still shows the OLD UI until a hard refresh (operator-flagged
|
||
# 2026-06-30).
|
||
resp = await call_next(request)
|
||
resp.headers["Cache-Control"] = "no-store"
|
||
return resp
|
||
|
||
|
||
@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.replace("__BUILD__", VERSION)
|
||
|
||
|
||
@app.post("/start")
|
||
def start():
|
||
worker.start()
|
||
return JSONResponse(worker.status())
|
||
|
||
|
||
@app.post("/stop")
|
||
def stop():
|
||
worker.stop()
|
||
return JSONResponse(worker.status())
|
||
|
||
|
||
@app.post("/concurrency")
|
||
async def concurrency(request: Request):
|
||
body = await request.json()
|
||
worker.set_concurrency(int(body.get("value", 1)))
|
||
return JSONResponse(worker.status())
|
||
|
||
|
||
@app.post("/auto")
|
||
async def auto(request: Request):
|
||
body = await request.json()
|
||
worker.set_auto(bool(body.get("value", True)))
|
||
return JSONResponse(worker.status())
|
||
|
||
|
||
@app.get("/gpu")
|
||
def gpu():
|
||
# GPU meters poll this on their own fast cadence. It only reads local
|
||
# nvidia-smi — no curator round-trip — so the util/VRAM bars stay live even
|
||
# when /status is slow waiting on the (sometimes busy) curator queue call.
|
||
g = read_gpu() or {}
|
||
us = worker.util_smooth()
|
||
if us is not None:
|
||
g["util_smooth"] = round(us, 1) # autoscaler's EWMA — the UI bar tracks this
|
||
return JSONResponse(g)
|
||
|
||
|
||
@app.get("/logs")
|
||
def logs():
|
||
return JSONResponse({"lines": list(logbuf.LINES)})
|
||
|
||
|
||
@app.get("/status")
|
||
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)
|
||
|
||
|
||
_PAGE = """<!doctype html><html><head><meta charset=utf-8>
|
||
<meta name=viewport content="width=device-width,initial-scale=1">
|
||
<title>FabledCurator · GPU agent</title>
|
||
<style>
|
||
:root{--bg:#0f1216;--panel:#181c22;--panel2:#1e232b;--bd:#2a313b;--fg:#e9edf2;
|
||
--mut:#8b97a6;--acc:#e8923a;--grn:#46c46a;--red:#e8584d;--amb:#e8b23a}
|
||
*{box-sizing:border-box}
|
||
body{font:14px/1.5 system-ui,-apple-system,Segoe UI,Roboto,sans-serif;margin:0;
|
||
background:radial-gradient(1200px 600px at 50% -10%,#1a2029,#0f1216);color:var(--fg)}
|
||
.wrap{max-width:820px;margin:0 auto;padding:28px 20px 28px;height:100vh;
|
||
box-sizing:border-box;overflow:hidden;display:flex;flex-direction:column}
|
||
header{display:flex;align-items:center;justify-content:space-between;margin-bottom:4px}
|
||
.brand{display:flex;align-items:center;gap:10px;font-size:19px;font-weight:700;letter-spacing:.2px}
|
||
.logo{color:var(--acc);font-size:20px}
|
||
.brand .sub{color:var(--mut);font-weight:600;font-size:13px;text-transform:uppercase;letter-spacing:.12em}
|
||
.conn{display:flex;align-items:center;gap:8px;color:var(--mut);font-size:13px;font-weight:600}
|
||
.dot{width:9px;height:9px;border-radius:50%;background:var(--mut);box-shadow:0 0 0 0 rgba(0,0,0,0)}
|
||
.dot.green{background:var(--grn);box-shadow:0 0 10px 1px rgba(70,196,106,.5)}
|
||
.dot.amber{background:var(--amb)} .dot.red{background:var(--red)}
|
||
.meta{color:var(--mut);margin:0 0 18px;font-size:13px}
|
||
code{background:#11151a;border:1px solid var(--bd);padding:2px 7px;border-radius:6px;
|
||
font:12px ui-monospace,SFMono-Regular,Menlo,monospace;color:#cdd6e0}
|
||
.card{background:linear-gradient(180deg,var(--panel),var(--panel2));border:1px solid var(--bd);
|
||
border-radius:14px;padding:16px 18px;margin-bottom:14px;box-shadow:0 1px 0 rgba(255,255,255,.02) inset}
|
||
.card-h{font-size:11px;font-weight:800;letter-spacing:.12em;text-transform:uppercase;
|
||
color:var(--mut);margin-bottom:14px}
|
||
.controls{display:flex;align-items:center;gap:10px;flex-wrap:wrap}
|
||
.spacer{flex:1}
|
||
.btn{font:600 14px system-ui;padding:.5rem 1rem;border:1px solid transparent;border-radius:9px;
|
||
cursor:pointer;color:#fff;transition:.12s}
|
||
.btn:hover{transform:translateY(-1px)}
|
||
.btn[disabled]{opacity:.45;pointer-events:none;transform:none}
|
||
@keyframes pulse{0%,100%{opacity:1}50%{opacity:.4}}
|
||
.tile .n.busy{color:var(--acc);animation:pulse 1s ease-in-out infinite}
|
||
.btn.start{background:linear-gradient(180deg,#2f9c4c,#247a3c)}
|
||
.btn.stop{background:linear-gradient(180deg,#3a3f48,#2a2f37);color:#e9edf2;border-color:var(--bd)}
|
||
.switch{display:inline-flex;align-items:center;gap:8px;cursor:pointer;font-weight:600;user-select:none}
|
||
.switch input{display:none}
|
||
.switch .track{width:38px;height:22px;border-radius:11px;background:#2a313b;position:relative;transition:.15s}
|
||
.switch .track:after{content:"";position:absolute;top:2px;left:2px;width:18px;height:18px;border-radius:50%;
|
||
background:#cdd6e0;transition:.15s}
|
||
.switch input:checked+.track{background:var(--acc)}
|
||
.switch input:checked+.track:after{transform:translateX(16px);background:#fff}
|
||
.stepper{display:inline-flex;align-items:center;gap:6px}
|
||
.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;
|
||
color:var(--fg);border:1px solid var(--bd);border-radius:8px}
|
||
.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}
|
||
.tile .n{font:800 22px ui-monospace,monospace;line-height:1.1}
|
||
.tile .n.warn{color:var(--red)} .tile .n.ok{color:var(--grn)}
|
||
.tile .l{font-size:10px;text-transform:uppercase;letter-spacing:.06em;color:var(--mut);margin-top:4px}
|
||
.meters{display:flex;flex-direction:column;gap:10px;margin-bottom:14px}
|
||
.meter-h{display:flex;justify-content:space-between;font-size:12px;color:var(--mut);margin-bottom:4px}
|
||
.meter-h b{color:var(--fg);font-variant-numeric:tabular-nums}
|
||
.bar{height:9px;border-radius:5px;background:#11151a;border:1px solid var(--bd);overflow:hidden}
|
||
.bar>i{display:block;height:100%;width:0;background:linear-gradient(90deg,#3a7d57,var(--grn));transition:width .4s}
|
||
#utilbar{background:linear-gradient(90deg,#9a5a1f,var(--acc))}
|
||
#bufbar{background:linear-gradient(90deg,#2f5a9a,#4a86d8)}
|
||
.queue{font:13px ui-monospace,monospace;color:var(--mut)}
|
||
.banner{margin:0 0 14px;padding:.7rem .9rem;border-radius:10px;background:#3a2f12;
|
||
border:1px solid #5a4a17;color:#ffd98a;font-size:13px}
|
||
.logs-h{display:flex;align-items:center;justify-content:space-between}
|
||
.grow{flex:1;display:flex;flex-direction:column;min-height:0}
|
||
.grow .logs{flex:1;min-height:0}
|
||
.copybtn{font:600 11px system-ui;letter-spacing:.04em;text-transform:uppercase;
|
||
background:#262c34;color:var(--fg);border:1px solid var(--bd);border-radius:7px;
|
||
padding:5px 11px;cursor:pointer}
|
||
.copybtn:hover{border-color:var(--acc)}
|
||
.logs{margin:0;background:#0b0e12;border:1px solid var(--bd);border-radius:10px;padding:12px;
|
||
overflow:auto;font:12px/1.55 ui-monospace,SFMono-Regular,Menlo,monospace;
|
||
color:#b9c4d0;white-space:pre-wrap;word-break:break-word}
|
||
</style></head><body>
|
||
<div class=wrap>
|
||
<header>
|
||
<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> · 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>
|
||
|
||
<section class=card>
|
||
<div class=card-h>Control</div>
|
||
<div class=controls>
|
||
<button class="btn start" id=startbtn onclick=act('start')>▶ Start</button>
|
||
<button class="btn stop" id=stopbtn onclick=act('stop')>■ Stop</button>
|
||
<div class=spacer></div>
|
||
<label class=switch><input type=checkbox id=autochk onchange="setauto(this.checked)"><span class=track></span>Auto</label>
|
||
<div class=stepper>
|
||
<button class=step onclick=setc(-1)>−</button>
|
||
<input id=conc type=number min=1 value=1 onchange="setv(this.value)">
|
||
<button class=step onclick=setc(1)>+</button>
|
||
</div>
|
||
</div>
|
||
<div class=hint id=conchint>auto-tuning downloaders to keep the GPU fed · max <b id=capn>8</b></div>
|
||
</section>
|
||
|
||
<section class=card>
|
||
<div class=card-h>Status</div>
|
||
<div class=tiles>
|
||
<div class=tile><div class=n id=state>—</div><div class=l>state</div></div>
|
||
<div class=tile><div class=n id=dln>0</div><div class=l>downloaders</div></div>
|
||
<div class=tile><div class=n id=bufn>—</div><div class=l>buffer</div></div>
|
||
<div class=tile><div class=n id=active>0</div><div class=l>on GPU</div></div>
|
||
<div class=tile><div class="n ok" id=done>0</div><div class=l>processed</div></div>
|
||
<div class=tile><div class=n id=err>0</div><div class=l>errors</div></div>
|
||
</div>
|
||
<div class=meters>
|
||
<div class=meter><div class=meter-h><span>GPU util</span><b id=utillbl>—</b></div>
|
||
<div class=bar><i id=utilbar></i></div></div>
|
||
<div class=meter><div class=meter-h><span>VRAM</span><b id=vramlbl>—</b></div>
|
||
<div class=bar><i id=gpubar></i></div></div>
|
||
<div class=meter><div class=meter-h><span>buffer occupancy</span><b id=buflbl>—</b></div>
|
||
<div class=bar><i id=bufbar></i></div></div>
|
||
</div>
|
||
<div class=queue id=pipe>consumers — · waited out 0</div>
|
||
<div class=queue id=queue>queue —</div>
|
||
</section>
|
||
|
||
<section class="card grow">
|
||
<div class="card-h logs-h">Logs
|
||
<button class=copybtn id=copybtn onclick=copyLogs()>Copy</button>
|
||
</div>
|
||
<pre class=logs id=logs>waiting for activity…</pre>
|
||
</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
|
||
// separate /status poll, which can lag behind the curator queue call.
|
||
async function act(p){
|
||
pending(p==='start'?'starting':'stopping')
|
||
// Abort a slow POST after 8s so the buttons never stay stuck — the periodic
|
||
// /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 }
|
||
}
|
||
function pending(label){
|
||
state.textContent=label; state.className='n busy'
|
||
dot.className='dot amber'
|
||
startbtn.disabled=true; stopbtn.disabled=true
|
||
}
|
||
function setc(d){ if(conc.disabled)return; setv((parseInt(conc.value||'1'))+d) }
|
||
async function setv(v){
|
||
v=Math.max(1,Math.min(CAP,parseInt(v)||1)); conc.value=v
|
||
await fetch('/concurrency',{method:'POST',headers:{'Content-Type':'application/json'},
|
||
body:JSON.stringify({value:v})});refresh()
|
||
}
|
||
async function setauto(on){
|
||
await fetch('/auto',{method:'POST',headers:{'Content-Type':'application/json'},
|
||
body:JSON.stringify({value:on})});refresh()
|
||
}
|
||
async function refresh(){
|
||
let s; try{ s=await (await fetch('/status')).json() }catch{ return }
|
||
applyStatus(s)
|
||
}
|
||
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
|
||
state.textContent=draining?'stopping':s.state
|
||
state.className='n'+(draining?' busy':'')
|
||
dln.textContent=(s.downloaders!=null?s.downloaders:'—')
|
||
bufn.textContent=(s.buffer!=null?(s.buffer+'/'+s.buffer_max):'—')
|
||
active.textContent=s.active; active.className='n'+(s.active>0?' busy':'')
|
||
done.textContent=s.processed
|
||
err.textContent=s.errors; err.className='n'+(s.errors>0?' warn':'')
|
||
pipe.textContent='consumers '+(s.consumers!=null?s.consumers:'—')+' · waited out '+(s.transient||0)
|
||
// 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+'%' }
|
||
// Auto on → dial reflects the auto-chosen count (read-only); off → manual.
|
||
if(document.activeElement!==autochk) autochk.checked=!!s.auto
|
||
conc.disabled=!!s.auto; conc.style.opacity=s.auto?0.55:1
|
||
conchint.textContent=s.auto?('auto-tuning downloaders to keep the GPU fed · max '+CAP):('manual downloaders · max '+CAP)
|
||
if(document.activeElement!==conc) conc.value=s.concurrency
|
||
conc.max=CAP
|
||
// Connection pill + queue come only from the /status poll (the Start/Stop POST
|
||
// responses skip the slow curator call to stay snappy) — guard so an action
|
||
// response doesn't blank them.
|
||
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'
|
||
queue.textContent=s.queue?('queue · pending '+s.queue.pending+' · in flight '+s.queue.leased+' · done '+s.queue.done+' · errored '+s.queue.error):'queue · unreachable'
|
||
}
|
||
}
|
||
// GPU meters poll their OWN endpoint on a fast cadence — kept off /status so a
|
||
// slow curator queue call can't freeze the bars (they only stale on refresh).
|
||
let UAVG=null // smoothed util for the bar (raw util swings 0↔99; show the trend)
|
||
async function refreshGpu(){
|
||
let g; try{ g=await (await fetch('/gpu')).json() }catch{ return }
|
||
if(g && g.util_pct!=null){
|
||
// Prefer the agent's own EWMA (util_smooth) when running; otherwise smooth
|
||
// the raw reading here so a stopped agent's bar still glides, not jumps.
|
||
const raw=g.util_pct
|
||
UAVG = (g.util_smooth!=null) ? g.util_smooth
|
||
: (UAVG==null ? raw : 0.25*raw + 0.75*UAVG)
|
||
const used=g.mem_used_mb, tot=g.mem_total_mb||1
|
||
utillbl.textContent=Math.round(UAVG)+'% · '+g.temp_c+'°C'; utilbar.style.width=Math.round(UAVG)+'%'
|
||
vramlbl.textContent=used+' / '+tot+' MB'; gpubar.style.width=Math.round(100*used/tot)+'%'
|
||
} else { UAVG=null; utillbl.textContent='n/a'; vramlbl.textContent='n/a (CPU?)'; utilbar.style.width='0%'; gpubar.style.width='0%' }
|
||
}
|
||
async function refreshLogs(){
|
||
try{
|
||
const r=await (await fetch('/logs')).json()
|
||
const el=logs, atBottom=el.scrollHeight-el.scrollTop-el.clientHeight<40
|
||
el.textContent=(r.lines&&r.lines.length)?r.lines.join('\\n'):'waiting for activity…'
|
||
if(atBottom) el.scrollTop=el.scrollHeight
|
||
}catch{}
|
||
}
|
||
async function copyLogs(){
|
||
const txt=logs.textContent||''
|
||
try{ await navigator.clipboard.writeText(txt) }
|
||
catch{ const t=document.createElement('textarea'); t.value=txt; document.body.appendChild(t);
|
||
t.select(); try{document.execCommand('copy')}catch{}; t.remove() }
|
||
copybtn.textContent='Copied'; setTimeout(()=>{copybtn.textContent='Copy'},1200)
|
||
}
|
||
refresh(); refreshGpu(); refreshLogs()
|
||
setInterval(refresh,3000); setInterval(refreshGpu,1500); setInterval(refreshLogs,2500)
|
||
</script></body></html>"""
|