feat: per-GPU gpu-jobs queue + worker-mode gate (no hidden runs)

Make EVERY studio render flow through the org's visible gpu-jobs queue and eliminate the direct-to-127.0.0.1:8188 bypass. Studio-side only (cloud CLI / worker two-lane claim owned by the cloud agent). - gpu_dispatch: per-GPU targeting. X-Target-GPU -> taskQueue "gpu:<identity>" (namespace stays gpu-jobs); an explicit target always enqueues. Untargeted = shared "gpu-jobs" lane. - server.post_prompt: in --worker-mode refuse a direct /prompt POST without X-Worker-Token (403) via worker_client.reject_untrusted_worker_submit (one policy gate). The only accepted execution is a claimed job over /v1/worker/execute. Legacy prompt_router push is OFF unless STUDIO_LEGACY_PUSH_ROUTER=1 -> gpu-jobs enqueue is the ONE submit path. - worker_client: /v1/worker/execute validates X-Worker-Token, runs the graph on the local prompt_queue, scopes outputs to the job's org, returns {prompt_id}. - studio_home: /v1/render-queue + /v1/nodes are AUTHORITATIVE — read the cloud tasks engine's gpu-jobs activities (real claiming identity, status -> queued/running/done/failed, lastHeartbeatTime, per-node depth). render_jobs.json is only a ~2s pre-claim placeholder. X-Target-GPU forwarded through _queue_prompt. - UI: "Run on <gpu>" chips (studio_home.html/shell.js) set X-Target-GPU; in-flight rows show the REAL worker + queued/running state. Tests: gpu_dispatch_test (targeting), worker_seam_test (gate 403 + claim seam). 378 passed. Follow-up (cloud agent): content/studio_render.go:86 convergence.
2026-07-21 12:45:59 -07:00
parent 537149fd36
commit 91cbfd81d4
13 changed files with 675 additions and 124 deletions
@@ -97,6 +97,24 @@ shared Hanzo IAM (standard OIDC code flow). No embedding.
  visor_client}.py`): `/v1/workers/register` (heartbeat), `/v1/workers` (list),
  `/v1/worker/execute` (in-cluster push fast path). Worker↔coordinator trust via
  `X-Worker-Token` (`STUDIO_WORKER_TOKEN`, KMS-sourced).
+- **Per-GPU gpu-jobs queue** (`middleware/{gpu_dispatch,studio_home}.py`, `server.py`) —
+  the ONE submit path onto a GPU. A model-less coordinator's `POST /prompt` enqueues to
+  the cloud tasks engine (`POST {STUDIO_CLOUD_API_URL}/v1/tasks/namespaces/gpu-jobs/activities`,
+  user bearer → org-scoped); a BYO worker (`hanzo gpu connect`) claims + runs it.
+  **Targeting**: the Home "Run on <gpu>" chip sets `X-Target-GPU: <identity>`;
+  `dispatch_if_worker` enqueues on that machine's own lane `taskQueue="gpu:<identity>"`
+  (namespace stays `gpu-jobs`), else the shared `gpu-jobs` lane any worker drains —
+  `X-Target-GPU` rides through `_queue_prompt` so fix/compose/video/rerun all honor it.
+  **Authoritative visible queue**: `/v1/render-queue` + `/v1/nodes` read the tasks
+  engine's activity list — REAL claiming `identity`, status
+  SCHEDULED/STARTED/COMPLETED/FAILED → queued/running/done/failed, `lastHeartbeatTime`
+  liveness, per-node depth — not a local shadow; `render_jobs.json` survives only as a
+  ~2s pre-claim placeholder. **No hidden runs**: in `--worker-mode` a direct `/prompt`
+  POST is refused 403 unless it carries `X-Worker-Token` — the only accepted execution is
+  a claimed job handed in over `POST /v1/worker/execute`
+  (`worker_client.reject_untrusted_worker_submit`, one policy gate). Legacy push
+  (`prompt_router`) is OFF unless `STUDIO_LEGACY_PUSH_ROUTER=1`. Tests:
+  `middleware_test/{gpu_dispatch_test,worker_seam_test}.py`.
 - **Durable render queue** (`middleware/tasks_queue.py`, `docs/federation.md`):
  `SqlitePromptQueue` — crash-durable render queue in a single SQLite file
  (stdlib `sqlite3`, WAL, **zero external processes**). `STUDIO_QUEUE_DB=<path>`
@@ -456,7 +456,13 @@ def start_studio(asyncio_loop=None):

    # In worker mode, add worker execution routes and register with coordinator
    if args.worker_mode:
-        from middleware.worker_client import add_worker_routes
+        from middleware.worker_client import add_worker_routes, WORKER_TOKEN
+        # FAIL CLOSED: a worker box executes jobs the coordinator hands it over a
+        # secret-authenticated seam. Without STUDIO_WORKER_TOKEN that seam (and the
+        # worker-mode /prompt gate) would accept un-tokened callers — silently
+        # reopening the hidden-run hole. Refuse to start rather than run wide open.
+        if not WORKER_TOKEN:
+            sys.exit("worker-mode requires STUDIO_WORKER_TOKEN (coordinator shared secret) — refusing to start")
        add_worker_routes(prompt_server.routes, prompt_server)
        logging.info("Worker mode enabled — worker_id=%s coordinator=%s",
                      args.worker_id, args.coordinator_url)
@@ -224,21 +224,39 @@ def dispatch_if_worker(request, org_id: str, prompt_id: str, prompt: dict) -> bo
        tok = _bearer(request)
        if not tok:
            return False
+        # Per-GPU targeting: the UI's "Run on <gpu>" affordance sets X-Target-GPU to a
+        # machine identity. A targeted render is enqueued on THAT machine's own lane
+        # ("gpu:<identity>") which only that worker claims; an untargeted render rides
+        # the shared "gpu-jobs" lane any worker drains. The NAMESPACE is "gpu-jobs"
+        # either way — targeting is a taskQueue, not a second queue. One enqueue path.
+        target = (request.headers.get("X-Target-GPU") or "").strip()
        online = _online_gpu_nodes(tok)
+        # Honor a target ONLY if it is one of the caller-org's OWN online GPUs. This
+        # blocks a forged lane AND sanitizes the value before it becomes a taskQueue
+        # (and later a queue-row label): an unknown/offline target falls back to the
+        # shared lane rather than enqueuing onto a "gpu:<bogus>" lane no worker will
+        # ever claim (a 30-min schedule-to-start hang) — and a "<img onerror>" target
+        # never reaches the queue. Cross-org is already impossible (per-org shard); this
+        # is defense in depth.
+        if target and target not in {_node_label(m) for m in online}:
+            target = ""
+        task_queue = f"gpu:{target}" if target else JOBS_NS
        # No visible worker AND this box can load THIS graph → render locally. No
        # visible worker on a box that lacks the graph's models → still enqueue and
        # wait (never a doomed local validation), even with a stray unrelated model.
-        if not online and has_models_for(prompt):
+        # A validated target ALWAYS enqueues (the user asked for that online GPU).
+        if not target and not online and has_models_for(prompt):
            return False
-        # Which node runs it: a single visible GPU is definitive; with several — or an
-        # org-blind read that saw none — the claimer is decided at pick-up: "gpu".
-        node = _node_label(online[0]) if len(online) == 1 else "gpu"
+        # Which node runs it: the explicit target; else a single visible GPU is
+        # definitive; with several — or an org-blind read that saw none — the claimer
+        # is decided at pick-up: "gpu".
+        node = target or (_node_label(online[0]) if len(online) == 1 else "gpu")
        staged = _collect_input_images(prompt, org_id)
        body = json.dumps({
            "activityId": prompt_id,
            "runId": prompt_id,
            "activityType": {"name": "studio.render"},
-            "taskQueue": JOBS_NS,
+            "taskQueue": task_queue,
            # 1200s: a heavy Qwen edit on a cold GB10 (model reload + full 30-step
            # sample) can exceed 600s; the worker heartbeats each step, so this is
            # the ceiling for a genuinely long single render, not idle slack.
@@ -144,9 +144,12 @@
    gpus.classList.toggle("on", online > 0);
    $("hznodes").innerHTML = nodes.length ? nodes.map(function (n) {
      var s = byNode[n.name] || { running: null, depth: 0 };
+      // Authoritative depth from /v1/nodes (gpu-jobs activities) when present; the
+      // running-render label still comes from the org's live queue/status.
+      var depth = (typeof n.depth === "number") ? n.depth : s.depth;
      var sub = (n.online ? "online" : "offline") +
        (s.running ? " · running: " + esc(String(s.running).slice(0, 40)) : (n.online ? " · idle" : "")) +
-        " · queue " + s.depth;
+        " · queue " + depth;
      return '<div class="hz-node ' + (n.online ? "on" : "") + '"><span class="hz-dot"></span>' +
        '<div class="hz-nn"><b>' + esc(n.name) + '</b><div class="hz-nsub">' + sub + "</div></div></div>";
    }).join("") : '<div class="hz-nsub">No GPU nodes connected. Run <b>hanzo gpu connect</b> on a box to add one.</div>';
@@ -136,7 +136,9 @@ dialog .drop{border:1.5px dashed var(--line);border-radius:9px;padding:14px;text
 .wmeta{font-size:.72rem;color:var(--dim);margin-top:3px;display:flex;gap:8px;flex-wrap:wrap;align-items:center}
 .wnode{border:1px solid var(--line);border-radius:6px;padding:0 6px;color:var(--txt)}
 .qhead{display:flex;gap:8px;flex-wrap:wrap;align-items:center;margin:0 0 14px;font-size:.8rem;color:var(--dim)}
-.nodebadge{display:inline-flex;align-items:center;gap:6px;border:1px solid var(--line);border-radius:99px;padding:3px 11px;background:var(--panel)}
+.nodebadge{display:inline-flex;align-items:center;gap:6px;border:1px solid var(--line);border-radius:99px;padding:3px 11px;background:var(--panel);cursor:pointer;user-select:none}
+.nodebadge:hover{border-color:var(--dim)}
+.nodebadge.sel{border-color:var(--ok);box-shadow:0 0 0 1px var(--ok) inset}
 .nodebadge .dot{width:7px;height:7px;border-radius:50%;background:#555}
 .nodebadge.on .dot{background:var(--ok)} .nodebadge.off{opacity:.6}
 .qhead .mstat{margin-left:auto;color:var(--dim)}
@@ -573,7 +575,7 @@ let TEMPLATE_NAMES={};
 async function renderTemplate(slug){
  const name=TEMPLATE_NAMES[slug]||slug;
  toast('Queueing '+name+'…');
-  const r=await fetch('/v1/templates/render',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({name:slug})});
+  const r=await fetch('/v1/templates/render',{method:'POST',headers:genHeaders(),body:JSON.stringify({name:slug})});
  const j=await r.json().catch(()=>({}));
  toast(r.ok?'Render queued → see Queue & History':'Failed: '+(errMsg(j)||('error '+r.status)));
 }
@@ -606,7 +608,7 @@ async function create(){
    await enqueue(m3btn,'…',async()=>{
      if(!(await gpuOnline()))toast('⚠ No GPU online — queues until a BYO-GPU connects');
      const body=libs.length?{path:libs[0]}:{upload:ups[0]};
-      const r=await fetch('/v1/library/model3d',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(body)});
+      const r=await fetch('/v1/library/model3d',{method:'POST',headers:genHeaders(),body:JSON.stringify(body)});
      const j=await r.json().catch(()=>({}));
      if(r.ok){ toast('3D queued → see Queue & History'); track('model3d',{from:'composer'}); $('pin').value=''; clearTray(); clearLane(); }
      else if(!(await healStale(j))) toast('Failed: '+(errMsg(j)||('error '+r.status)));
@@ -622,7 +624,7 @@ async function create(){
    await enqueue(vbtn,'…',async()=>{
      if(!(await gpuOnline()))toast('⚠ No GPU online — queues until a BYO-GPU connects');
      const body={prompt:t}; if(libs.length)body.path=libs[0]; else if(ups.length)body.upload=ups[0];
-      const r=await fetch('/v1/library/video',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(body)});
+      const r=await fetch('/v1/library/video',{method:'POST',headers:genHeaders(),body:JSON.stringify(body)});
      const j=await r.json().catch(()=>({}));
      if(r.ok){ toast('Video queued → see Queue & History'); track('video',{from:'composer'}); $('pin').value=''; clearTray(); clearLane(); }
      else if(!(await healStale(j))) toast('Failed: '+(errMsg(j)||('error '+r.status)));
@@ -637,9 +639,9 @@ async function create(){
  await enqueue(btn,'…',async()=>{
    if(!(await gpuOnline()))toast('⚠ No GPU online — queues until a BYO-GPU connects');
    let r;
-    if(n===1){ r=await fetch('/v1/library/fix',{method:'POST',headers:{'Content-Type':'application/json'},
+    if(n===1){ r=await fetch('/v1/library/fix',{method:'POST',headers:genHeaders(),
        body:JSON.stringify(libs.length?{path:libs[0],instruction:t}:{upload:ups[0],instruction:t})}); }
-    else{ r=await fetch('/v1/library/compose',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({paths:libs,uploads:ups,instruction:t})}); }
+    else{ r=await fetch('/v1/library/compose',{method:'POST',headers:genHeaders(),body:JSON.stringify({paths:libs,uploads:ups,instruction:t})}); }
    const j=await r.json().catch(()=>({}));
    if(r.ok){ toast('Queued → see Queue & History'); track(n===1?'fix':'compose',{from:'composer'}); $('pin').value=''; clearTray(); }
    else if(!(await healStale(j))) toast('Failed: '+(errMsg(j)||('error '+r.status)));
@@ -772,7 +774,7 @@ async function bulk(status){
  if(!SEL.size)return;
  const paths=[...SEL];
  pushUndo(paths.map(p=>({path:p,prev:(DATA.assets.find(x=>x.path===p)||{}).status})));
-  for(const p of paths){await fetch('/v1/library/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path:p,status})});
+  for(const p of paths){await fetch('/v1/library/status',{method:'POST',headers:genHeaders(),body:JSON.stringify({path:p,status})});
    const a=DATA.assets.find(x=>x.path===p);if(a)a.status=status;}
  toast(paths.length+(status==='deleted'?' archived — ⌘Z to undo':status==='draft'?' recovered':' → '+status)); clearSel(); paintChips();
 }
@@ -782,7 +784,7 @@ const UNDO=[];
 function pushUndo(items){items=(items||[]).filter(it=>it&&it.path);if(items.length){UNDO.push(items);if(UNDO.length>100)UNDO.shift();}}
 async function undoLast(){
  const e=UNDO.pop(); if(!e){toast('Nothing to undo');return;}
-  await Promise.all(e.map(it=>fetch('/v1/library/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path:it.path,status:it.prev})})));
+  await Promise.all(e.map(it=>fetch('/v1/library/status',{method:'POST',headers:genHeaders(),body:JSON.stringify({path:it.path,status:it.prev})})));
  e.forEach(it=>{const a=DATA.assets.find(x=>x.path===it.path);if(a)a.status=it.prev;});
  render(); paintChips();
  toast('Restored '+e.length+(e.length>1?' items':' item')+' — ⌘Z');
@@ -886,7 +888,7 @@ async function sendCompose(){
  const t=$('comptext').value.trim(); if(t.length<3){toast('Describe how to combine them');return;}
  const paths=[...SEL]; toast('Composing from '+paths.length+' assets…');
  const stack=await resolveDest($('compdest'));
-  const r=await fetch('/v1/library/compose',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({paths,instruction:t,stack})});
+  const r=await fetch('/v1/library/compose',{method:'POST',headers:genHeaders(),body:JSON.stringify({paths,instruction:t,stack})});
  const j=await r.json().catch(()=>({}));
  if(r.ok){ $('compdlg').close(); toast(`Compose queued (${j.refs} refs) → see Queue & History`); track('compose',{refs:j.refs}); clearSel(); }
  else if(!(await healStale(j))){ $('gpuwarn2').style.display='block'; $('gpuwarn2').textContent='Couldn’t queue: '+(errMsg(j)||('error '+r.status)); }
@@ -906,6 +908,11 @@ let WL=[], CTX=null;
 // live queue status (positions/ETAs/node) computed SERVER-SIDE; QLOCAL seeds the
 // client-side elapsed tick between polls. NODES = the org's GPU nodes for badges.
 let QST={items:{},medians:{},lanes:{},now:0}, QLOCAL=Date.now(), NODES=[];
+let TARGET_GPU=null;        // null = any GPU (shared "gpu-jobs" lane); else a machine id
+// ONE header factory for every generation POST: forwards the picked GPU so the render
+// lands on THAT machine (dispatch → taskQueue "gpu:<id>"). Harmless on non-render routes.
+function genHeaders(){const h={'Content-Type':'application/json'};if(TARGET_GPU)h['X-Target-GPU']=TARGET_GPU;return h;}
+function pickGPU(name){TARGET_GPU=name||null;toast(TARGET_GPU?('Next render → '+TARGET_GPU):'Next render → any available GPU');renderQhead();}
 const WSTATUS={queued:['⏳','queued'],running:['●','running'],done:['✓','done'],failed:['⚠','failed'],cancelled:['✕','cancelled']};
 const fmtDur=s=>{ s=Math.max(0,Math.round(s)); if(s<60)return s+'s'; const m=Math.floor(s/60),r=s%60; if(s<3600)return m+'m'+(r?(' '+r+'s'):''); return Math.floor(s/3600)+'h '+Math.floor((s%3600)/60)+'m'; };
 const fmtEta=s=>{ s=Math.max(0,Math.round(s)); return s<60?('~'+s+'s est'):('~'+Math.round(s/60)+' min est'); };
@@ -914,10 +921,10 @@ function wlThumb(it){ if(it.output)return thumbURL({path:it.output});
  if(it.parents&&it.parents[0])return thumbURL({path:it.parents[0]}); return ''; }
 function qmetaHTML(it,e){
  if(!e)return '';
-  const node=`<span class="wnode">⚙ ${(e.node||it.node||'gpu')}</span>`;
+  const node=`<span class="wnode">⚙ ${esc(e.node||it.node||'gpu')}</span>`;
  if(e.status==='running'){
    const el=(e.elapsed_seconds||0)+(Date.now()-QLOCAL)/1000;
-    return `<div class="wmeta">${node} · #${e.position}/${e.of} · running <span id="el-${it.id}">${fmtDur(el)}</span>${e.eta_seconds!=null?(' · ~'+fmtDur(e.eta_seconds)+' left'):''}</div>`;
+    return `<div class="wmeta">${node} · #${e.position}/${e.of} · running <span id="el-${esc(it.id)}">${fmtDur(el)}</span>${e.eta_seconds!=null?(' · ~'+fmtDur(e.eta_seconds)+' left'):''}</div>`;
  }
  return `<div class="wmeta">${node} · #${e.position}/${e.of} in line · ${fmtEta(e.eta_seconds||0)}</div>`;
 }
@@ -927,22 +934,22 @@ function wlRow(it){
  const [ic,lbl]=WSTATUS[status]||['·',status||''];
  const th=wlThumb(it), when=it.ts?new Date(it.ts*1000).toLocaleString():'';
  const active=status==='queued'||status==='running';
-  return `<div class="qrow" style="cursor:pointer" onclick="openContext('${it.id}')">
+  return `<div class="qrow" style="cursor:pointer" data-open="${esc(it.id)}">
    <div class="qthumb" style="${th?`background-image:url('${th}');background-size:cover;background-position:center`:'display:flex;align-items:center;justify-content:center;color:var(--dim)'}">${th?'':'🗂'}</div>
-    <div class="qtitle">${it.kind} · <span class="wst ${status}">${ic} ${lbl}</span>
-      <div class="qdesc">${(it.prompt||'(no prompt)').replace(/</g,'&lt;')}</div>
-      ${active&&e?qmetaHTML(it,e):`<div class="qsub">${when}${it.node?(' · ⚙ '+it.node):''}${it.error?` · <span style="color:#ff8f8f">${it.error.replace(/</g,'&lt;')}</span>`:''}</div>`}</div>
-    ${active?`<button class="mini" onclick="event.stopPropagation();openAmend('${it.id}')" title="Amend prompt/refs (re-queues)">Edit</button><button class="mini" onclick="event.stopPropagation();cancelJob('${it.id}')">Cancel</button>`:''}
-    <button class="mini" onclick="event.stopPropagation();openContext('${it.id}')">Open</button>
-    ${(!active&&it.output)?`<button class="mini" onclick="event.stopPropagation();archiveFromQueue('${qq(it.output)}')" title="Move this render to Archive (recoverable)">🗄</button>`:''}
-    ${active?'':`<button class="mini" onclick="event.stopPropagation();delWork('${it.id}')" title="Delete this item from the queue" style="color:#f85149">🗑</button>`}</div>`;
+    <div class="qtitle">${esc(it.kind)} · <span class="wst ${status}">${ic} ${lbl}</span>
+      <div class="qdesc">${esc(it.prompt||'(no prompt)')}</div>
+      ${active&&e?qmetaHTML(it,e):`<div class="qsub">${when}${it.node?(' · ⚙ '+esc(it.node)):''}${it.error?` · <span style="color:#ff8f8f">${esc(it.error)}</span>`:''}</div>`}</div>
+    ${active?`<button class="mini" data-amend="${esc(it.id)}" title="Amend prompt/refs (re-queues)">Edit</button><button class="mini" data-cancel="${esc(it.id)}">Cancel</button>`:''}
+    <button class="mini" data-open="${esc(it.id)}">Open</button>
+    ${(!active&&it.output)?`<button class="mini" data-arch="${esc(it.output)}" title="Move this render to Archive (recoverable)">🗄</button>`:''}
+    ${active?'':`<button class="mini" data-del="${esc(it.id)}" title="Delete this item from the queue" style="color:#f85149">🗑</button>`}</div>`;
 }
 // Delete a queue/history ENTRY (the render request row) — distinct from archiving the
 // OUTPUT image. Optimistic: drop it locally + re-render, then persist.
 async function delWork(id){
  if(!id)return;
  WL=WL.filter(x=>x.id!==id);          // optimistic: drop locally, re-render
-  try{await fetch('/v1/worklog/delete',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({id})});}catch(_){}
+  try{await fetch('/v1/worklog/delete',{method:'POST',headers:genHeaders(),body:JSON.stringify({id})});}catch(_){}
  toast('Removed from queue'); worklog();
 }
 function delCtx(){ if(!CTX||!CTX.id)return; ctxdlg.close(); delWork(CTX.id); }
@@ -950,7 +957,7 @@ async function archiveFromQueue(path){
  if(!path)return;
  const a=DATA.assets.find(x=>x.path===path);
  pushUndo([{path,prev:(a||{}).status||'draft'}]);
-  try{const r=await fetch('/v1/library/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path,status:'deleted'})});
+  try{const r=await fetch('/v1/library/status',{method:'POST',headers:genHeaders(),body:JSON.stringify({path,status:'deleted'})});
    if(!r.ok)throw 0; if(a)a.status='deleted';
    toast('Archived — in 🗄 Archive · ⌘Z to undo'); worklog(); render(); paintChips();
  }catch(_){toast('Couldn’t archive');}
@@ -960,8 +967,12 @@ async function loadQueueStatus(){ let d; try{d=await (await fetch('/v1/queue/sta
 async function loadNodes(){ try{const d=await (await fetch('/v1/nodes')).json(); NODES=d.nodes||[];}catch(_){NODES=[];} renderQhead(); }
 function renderQhead(){
  if(!$('qhead'))return;
-  const badges=(NODES||[]).map(n=>`<span class="nodebadge ${n.online?'on':'off'}"><span class="dot"></span>${n.name}${n.online?'':' · offline'}</span>`).join('')
-    +`<span class="nodebadge on"><span class="dot"></span>studio pod</span>`;
+  // Clickable chips: "Any GPU" (shared lane) + one per node — click to aim your NEXT
+  // render at that machine. Selected chip carries a ✓; each shows its live queue depth.
+  const chip=(arg,label,online,sel,depth)=>`<span class="nodebadge ${online?'on':'off'}${sel?' sel':''}" role="button" tabindex="0" onclick="pickGPU(${arg})" title="${arg==='null'?'Run on any available GPU':'Run my next render on '+label}"><span class="dot"></span>${label}${online===false?' · offline':''}${depth?` · ${depth} in queue`:''}${sel?' ✓':''}</span>`;
+  const badges=chip("null","Any GPU",true,!TARGET_GPU,0)
+    +(NODES||[]).map(n=>chip("'"+String(n.name).replace(/'/g,"\\'")+"'",n.name,n.online,TARGET_GPU===n.name,(n.depth||0))).join('')
+    +`<span class="nodebadge on" title="This studio pod coordinates; renders run on the GPU nodes"><span class="dot"></span>studio pod</span>`;
  const m=QST.medians||{}, ks=Object.keys(m);
  const ms=ks.length?('median '+ks.map(k=>k+' ~'+fmtDur(m[k])).join(' · ')):'';
  $('qhead').innerHTML=badges+(ms?`<span class="mstat">${ms}</span>`:'');
@@ -1062,15 +1073,20 @@ async function runs(){
    const eta=running?'rendering now':(pos!=null?`~${(pos+1)*ETA_MIN} min · #${pos+1} in line`:'queued');
    return `<div class="qrow">
      <div class="qthumb" style="${thumb?`background-image:url('${thumb}');background-size:cover;background-position:center`:'display:flex;align-items:center;justify-content:center;color:var(--dim)'}">${thumb?'':(running?'▶':'⏳')}</div>
-      <div class="qtitle">${title}<div class="qdesc">${desc||'(no description)'}</div><div class="qsub">${running?'<span style=color:var(--ok)>● </span>':''}${eta}${refs.length>1?` · ${refs.length} refs`:''}</div></div>
-      ${running?'':`<button class="mini" onclick="cancelJob('${id}')">Remove</button>`}</div>`;};
+      <div class="qtitle">${esc(title)}<div class="qdesc">${esc(desc)||'(no description)'}</div><div class="qsub">${running?'<span style=color:var(--ok)>● </span>':''}${eta}${refs.length>1?` · ${refs.length} refs`:''}</div></div>
+      ${running?'':`<button class="mini" data-cancel="${esc(id)}">Remove</button>`}</div>`;};
  // BYO-GPU renders (fix/compose on a connected GPU) — rendered on the worker,
  // shown here so a dispatched fix is visible (was invisible before).
  let gjobs=[]; try{gjobs=(await (await fetch('/v1/render-queue')).json()).jobs||[];}catch(_){}
-  const gjRow=j=>`<div class="qrow">
-    <div class="qthumb" style="${j.refs&&j.refs[0]?`background-image:url('/v1/library/file?thumb=1&input=1&path=${encodeURIComponent(j.refs[0])}');background-size:cover;background-position:center`:'display:flex;align-items:center;justify-content:center;color:var(--ok)'}">${j.refs&&j.refs[0]?'':'▶'}</div>
-    <div class="qtitle">${j.prefix||'render'}<div class="qsub"><span style=color:var(--ok)>● </span>rendering on ${j.node||'your GPU'} · ${Math.max(0,Math.round((j.age||0)/60))}m elapsed${QST.items[j.id]&&QST.items[j.id].eta_seconds!=null?(' · ~'+fmtDur(QST.items[j.id].eta_seconds)+' left'):''}</div></div>
-    ${j.id?`<button class="mini" onclick="cancelJob('${j.id}')">Cancel</button>`:''}</div>`;
+  const gjRow=j=>{const running=(j.status||'running')==='running';
+    const where=esc((j.node&&j.node!=='queued')?j.node:'your GPU');
+    const sub=running
+      ?`<span style=color:var(--ok)>● </span>rendering on ${where} · ${Math.max(0,Math.round((j.age||0)/60))}m${QST.items[j.id]&&QST.items[j.id].eta_seconds!=null?(' · ~'+fmtDur(QST.items[j.id].eta_seconds)+' left'):''}`
+      :`⏳ queued${(j.node&&j.node!=='queued')?(' → '+esc(j.node)):''}`;
+    return `<div class="qrow">
+    <div class="qthumb" style="${j.refs&&j.refs[0]?`background-image:url('/v1/library/file?thumb=1&input=1&path=${encodeURIComponent(j.refs[0])}');background-size:cover;background-position:center`:'display:flex;align-items:center;justify-content:center;color:var(--ok)'}">${j.refs&&j.refs[0]?'':(running?'▶':'⏳')}</div>
+    <div class="qtitle">${esc(j.prefix||'render')}<div class="qsub">${sub}</div></div>
+    ${j.id?`<button class="mini" data-cancel="${esc(j.id)}">Cancel</button>`:''}</div>`;};
  const gjHtml=gjobs.map(gjRow).join('');
  $('qrun').innerHTML=gjHtml + qr.map(it=>qitem(it,true,null)).join('') || '<div class="qsub" style="padding:6px 2px">Idle — nothing rendering.</div>';
  $('qpend').innerHTML=qp.map((it,i)=>qitem(it,false,i)).join('')||'<div class="qsub" style="padding:6px 2px">Nothing queued.</div>';
@@ -1108,16 +1124,16 @@ async function runs(){
      </div></div>`;}).join('')||'<div class="empty">No runs yet.</div>';
  lazyObserve();
 }
-async function clearQueue(){if(!await ask('Clear all queued jobs?'))return;await fetch('/queue',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({clear:true})});toast('Queue cleared');runs();}
-async function rate(key,stars){await fetch('/v1/library/rate',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({key,stars})});RATINGS[key]={...(RATINGS[key]||{}),stars};runs();}
-async function rateNote(key,notes){await fetch('/v1/library/rate',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({key,notes})});toast('Note saved — the AI will use it next time');}
+async function clearQueue(){if(!await ask('Clear all queued jobs?'))return;await fetch('/queue',{method:'POST',headers:genHeaders(),body:JSON.stringify({clear:true})});toast('Queue cleared');runs();}
+async function rate(key,stars){await fetch('/v1/library/rate',{method:'POST',headers:genHeaders(),body:JSON.stringify({key,stars})});RATINGS[key]={...(RATINGS[key]||{}),stars};runs();}
+async function rateNote(key,notes){await fetch('/v1/library/rate',{method:'POST',headers:genHeaders(),body:JSON.stringify({key,notes})});toast('Note saved — the AI will use it next time');}
 // ── Train-the-model feedback: a 👍/👎 vote + a written note on any render, logged to
 // the org's append-only training dataset (feedback.jsonl) the model trainer consumes.
 // Keyed by the image path, so a vote persists with the asset (not just the run). ──
 async function castVote(path,v){
  const cur=RATINGS[path]||{}, nv=cur.vote===v?0:v;               // click the same thumb again = clear
  RATINGS[path]={...cur,vote:nv}; render(); zPaintRate();
-  try{await fetch('/v1/library/rate',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({key:path,path,vote:nv})});}catch(_){}
+  try{await fetch('/v1/library/rate',{method:'POST',headers:genHeaders(),body:JSON.stringify({key:path,path,vote:nv})});}catch(_){}
  toast(nv===1?'👍 Thanks — added to the training set':nv===-1?'👎 Noted — add a 💬 note so the model learns why':'Vote cleared'); track('rate',{score:nv});
  refreshFbCount();
 }
@@ -1133,7 +1149,7 @@ async function sendCritique(e){e.preventDefault();const t=($('fbtext').value||''
  const el=$('fbthread');const empty=el.querySelector('.empty');if(empty)el.innerHTML='';
  el.insertAdjacentHTML('beforeend',fbBubble('user',t));
  const pend=document.createElement('div');pend.style.cssText='align-self:flex-start;color:var(--dim);font-size:.84rem;padding:4px 8px';pend.textContent='…';el.appendChild(pend);el.scrollTop=el.scrollHeight;
-  try{const r=await fetch('/v1/library/critique',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path:FBPATH,message:t})});
+  try{const r=await fetch('/v1/library/critique',{method:'POST',headers:genHeaders(),body:JSON.stringify({path:FBPATH,message:t})});
    const j=await r.json();pend.remove();
    el.insertAdjacentHTML('beforeend',fbBubble('ai',j.reply||'Noted — the next render will avoid it.'));el.scrollTop=el.scrollHeight;
    RATINGS[FBPATH]={...(RATINGS[FBPATH]||{}),notes:t};track('rate',{score:FBVOTE,noted:true});refreshFbCount();zPaintRate();
@@ -1141,7 +1157,7 @@ async function sendCritique(e){e.preventDefault();const t=($('fbtext').value||''
  return false;}
 function fbSetVote(v){FBVOTE=FBVOTE===v?0:v;fbPaintVote();
  if(FBPATH){RATINGS[FBPATH]={...(RATINGS[FBPATH]||{}),vote:FBVOTE};
-    fetch('/v1/library/rate',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({key:FBPATH,path:FBPATH,vote:FBVOTE})}).catch(()=>{});
+    fetch('/v1/library/rate',{method:'POST',headers:genHeaders(),body:JSON.stringify({key:FBPATH,path:FBPATH,vote:FBVOTE})}).catch(()=>{});
    track('rate',{score:FBVOTE});refreshFbCount();zPaintRate();}}
 function fbPaintVote(){$('fbup').classList.toggle('brand',FBVOTE===1);$('fbdown').classList.toggle('brand',FBVOTE===-1);}
 async function refreshFbCount(){try{const s=await (await fetch('/v1/feedback/stats')).json();const b=$('fbexport');if(b)b.textContent='⬇ Feedback'+(s.total?' ('+s.total+')':'');}catch(_){}}
@@ -1154,20 +1170,20 @@ function qVote(path,v){
  const sel=(window.CSS&&CSS.escape)?CSS.escape(path):path;
  const fb=document.querySelector('.fb[data-op="'+sel+'"]');
  if(fb){fb.querySelector('.vote.up').classList.toggle('on',nv===1);fb.querySelector('.vote.down').classList.toggle('on',nv===-1);}
-  try{fetch('/v1/library/rate',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({key:path,path,vote:nv})});}catch(_){}
+  try{fetch('/v1/library/rate',{method:'POST',headers:genHeaders(),body:JSON.stringify({key:path,path,vote:nv})});}catch(_){}
  toast(nv===1?'👍 Thanks — rated for training':nv===-1?'👎 Noted — add a 💬 note so the model learns why':'Vote cleared'); refreshFbCount();
 }
 async function qReject(path){
  REJECTED.add(path);
  const a=DATA.assets.find(x=>x.path===path); pushUndo([{path,prev:(a||{}).status||'draft'}]);
  if(a)a.status='deleted';
-  try{await fetch('/v1/library/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path,status:'deleted'})});}catch(_){}
+  try{await fetch('/v1/library/status',{method:'POST',headers:genHeaders(),body:JSON.stringify({path,status:'deleted'})});}catch(_){}
  toast('Archived — the model will learn it was a bad result (⌘Z to undo)'); paintChips();
  if(typeof runs==='function')runs();
 }
 let FAVKEY=null,FAVTITLE=null;
 async function toggleFav(key,title){const nv=!FAVS[key];
-  await fetch('/v1/favorites',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({key,favorite:nv})});
+  await fetch('/v1/favorites',{method:'POST',headers:genHeaders(),body:JSON.stringify({key,favorite:nv})});
  if(nv)FAVS[key]={ts:Math.floor(Date.now()/1000)};else delete FAVS[key];
  if(nv){FAVKEY=key;FAVTITLE=title;$('tmplname').value=title;$('tmpldesc').value='';
    $('tmplsteps').innerHTML='This favorited run has an embedded workflow. Saving turns it into a reusable KIND:<br>1 · name it &amp; describe what it makes<br>2 · it appears under Templates for your org<br>3 · run it anytime with new inputs';
@@ -1180,21 +1196,21 @@ async function saveTemplate(){const nm=$('tmplname').value.trim();if(!nm){toast(
  // Editor's ★ Template also writes to.
  let g=null; try{const wr=await fetch('/v1/workflow?id='+encodeURIComponent(FAVKEY)); if(wr.ok)g=(await wr.json()).graph;}catch(_){}
  if(!g){$('tmpldlg').close();toast('No graph on this run — open it in the Editor and use ★ Template');return;}
-  const r=await fetch('/v1/templates',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({name:nm,graph:g})});
+  const r=await fetch('/v1/templates',{method:'POST',headers:genHeaders(),body:JSON.stringify({name:nm,graph:g})});
  const j=await r.json().catch(()=>({}));
  $('tmpldlg').close();toast(r.ok?('★ Saved "'+nm+'" — see Templates'):('Failed: '+(errMsg(j)||('error '+r.status))));
  if(r.ok&&!$('templates').hidden)loadOrgTemplates();
 }
-async function setSt(i,status){const a=DATA.assets[i];const prev=a.status;const r=await fetch('/v1/library/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path:a.path,status})});if(r.ok){pushUndo([{path:a.path,prev}]);a.status=status;render();paintChips();toast(`${a.path.split('/').pop()} → ${status} — ⌘Z`);}else toast('Failed');}
+async function setSt(i,status){const a=DATA.assets[i];const prev=a.status;const r=await fetch('/v1/library/status',{method:'POST',headers:genHeaders(),body:JSON.stringify({path:a.path,status})});if(r.ok){pushUndo([{path:a.path,prev}]);a.status=status;render();paintChips();toast(`${a.path.split('/').pop()} → ${status} — ⌘Z`);}else toast('Failed');}
 // Optimistic trash: mark deleted locally + re-render immediately (feels instant),
 // then persist. On failure, revert. This is the always-visible 🗑 on each card.
 async function quickDel(path){
  const a=DATA.assets.find(x=>x.path===path); if(!a)return;
  const prev=a.status; pushUndo([{path,prev}]); a.status='deleted'; render(); paintChips(); toast('Moved to Archive — ⌘Z to undo');
-  try{const r=await fetch('/v1/library/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path,status:'deleted'})});
+  try{const r=await fetch('/v1/library/status',{method:'POST',headers:genHeaders(),body:JSON.stringify({path,status:'deleted'})});
    if(!r.ok)throw 0;}catch(_){a.status=prev;UNDO.pop();render();paintChips();toast('Delete failed — restored');}
 }
-async function rerun(i,btn){const a=DATA.assets[i];await enqueue(btn,'Queueing…',async()=>{toast('Queueing re-run…');const r=await fetch('/v1/library/rerun',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path:a.path})});await r.json().catch(()=>({}));toast(r.ok?`Re-run queued — see Runs`:'Failed');});}
+async function rerun(i,btn){const a=DATA.assets[i];await enqueue(btn,'Queueing…',async()=>{toast('Queueing re-run…');const r=await fetch('/v1/library/rerun',{method:'POST',headers:genHeaders(),body:JSON.stringify({path:a.path})});await r.json().catch(()=>({}));toast(r.ok?`Re-run queued — see Runs`:'Failed');});}
 // Dedupe: server hashes visible assets, soft-deletes exact byte-identical copies
 // (keeps one), returns the moved paths so ⌘Z undoes the whole sweep at once.
 async function dedup(){
@@ -1212,7 +1228,7 @@ async function dedup(){
 // server enforces archive-only, so this can never destroy a live gallery image. ──
 async function purgePaths(paths){
  if(!paths.length)return;
-  let r; try{ r=await fetch('/v1/library/purge',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({paths})}); }
+  let r; try{ r=await fetch('/v1/library/purge',{method:'POST',headers:genHeaders(),body:JSON.stringify({paths})}); }
  catch(_){ toast('Delete failed'); return; }
  if(!r.ok){ toast('Delete failed'); return; }
  const j=await r.json().catch(()=>({removed:paths}));
@@ -1234,7 +1250,7 @@ async function emptyArchive(){
  const n=DATA.assets.filter(a=>a.status==='deleted').length;
  if(!n){ toast('Archive is empty'); return; }
  if(!await ask('Permanently delete all '+n+' archived item'+(n>1?'s':'')+'? This cannot be undone.'))return;
-  let r; try{ r=await fetch('/v1/library/purge',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({all:true})}); }
+  let r; try{ r=await fetch('/v1/library/purge',{method:'POST',headers:genHeaders(),body:JSON.stringify({all:true})}); }
  catch(_){ toast('Empty Archive failed'); return; }
  if(!r.ok){ toast('Empty Archive failed'); return; }
  const j=await r.json().catch(()=>({removed:[]}));
@@ -1326,11 +1342,11 @@ async function sendFix(){const t=$('fixtext').value.trim();if(t.length<3){showFi
  if(MASKMODE&&maskPainted()){ mask=await uploadMask(); if(!mask){showFixErr('Couldn’t upload the painted area — try again.');return;} }
  let r;
  if(FIXMODE==='amend'){ // atomic swap: cancel the queued job + re-dispatch amended
-    r=await fetch('/v1/queue/amend',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({prompt_id:AMENDID,instruction:t,paths,uploads,stack})});
+    r=await fetch('/v1/queue/amend',{method:'POST',headers:genHeaders(),body:JSON.stringify({prompt_id:AMENDID,instruction:t,paths,uploads,stack})});
  }else if(!FIXREFS.length){ // single base image → the low-denoise edit (optionally masked)
-    r=await fetch('/v1/library/fix',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path:FIXPATH,instruction:t,stack,strength,mask})});
+    r=await fetch('/v1/library/fix',{method:'POST',headers:genHeaders(),body:JSON.stringify({path:FIXPATH,instruction:t,stack,strength,mask})});
  }else{               // base + references → multi-reference compose
-    r=await fetch('/v1/library/compose',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({paths:[FIXPATH,...paths],instruction:t,uploads,stack})});
+    r=await fetch('/v1/library/compose',{method:'POST',headers:genHeaders(),body:JSON.stringify({paths:[FIXPATH,...paths],instruction:t,uploads,stack})});
  }
  const j=await r.json().catch(()=>({}));
  if(r.ok){ $('fixdlg').close(); toast(j.note || (FIXMODE==='amend'?'Amended — re-queued (position reset)':(mask?'Area edit queued → see Queue & History':'Edit queued → see Queue & History'))); track('fix',{mode:FIXMODE,masked:!!mask}); if(typeof runs==='function')runs(); }
@@ -1366,7 +1382,7 @@ async function uploadMask(){                            // export the canvas (bl
 }
 async function cancelJob(id){
  if(!await ask('Cancel this job?'))return;
-  const r=await fetch('/v1/queue/cancel',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({prompt_id:id})});
+  const r=await fetch('/v1/queue/cancel',{method:'POST',headers:genHeaders(),body:JSON.stringify({prompt_id:id})});
  const j=await r.json().catch(()=>({}));
  if(r.ok)toast('Cancelled'+(j.canceled==='gpu'?' — if it was already rendering on your GPU it may still finish':''));
  else toast('Cancel failed: '+(errMsg(j)||('error '+r.status)));
@@ -1433,7 +1449,7 @@ async function loadSession(){
    m.innerHTML=orgs.map(o=>`<div class="omi" data-o="${o}" style="padding:8px 11px;border-radius:8px;cursor:pointer;font-size:.85rem;color:${o===s.org?'var(--txt)':'var(--dim)'}">${o===s.org?'● ':'○ '}${o}</div>`).join('')
      +`<div style="height:1px;background:var(--line);margin:5px 4px"></div>`+(s.console_url?`<a href="${s.console_url}" style="display:block;padding:8px 11px;color:var(--dim);text-decoration:none;font-size:.85rem">Console ↗</a>`:'')+`<a href="/logout" style="display:block;padding:8px 11px;color:var(--dim);text-decoration:none;font-size:.85rem">Sign out</a>`;
    document.body.appendChild(m);
-    m.querySelectorAll('.omi').forEach(el=>el.onclick=async()=>{try{await fetch('/v1/session/org',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({org:el.dataset.o})});}catch(e){} location.reload();});
+    m.querySelectorAll('.omi').forEach(el=>el.onclick=async()=>{try{await fetch('/v1/session/org',{method:'POST',headers:genHeaders(),body:JSON.stringify({org:el.dataset.o})});}catch(e){} location.reload();});
    setTimeout(()=>document.addEventListener('click',function h(e){if(!m.contains(e.target)&&!$('user').contains(e.target)){m.remove();document.removeEventListener('click',h);}}),0);
  };
 }
@@ -1478,6 +1494,18 @@ async function updateLivebar(){
 let STACKS=[], STACKSET=new Set(), CURSTACK=null, DRAG=null, SVDRAG=null, SVI=-1, DELSTACK=null;
 const esc=s=>String(s==null?'':s).replace(/[&<>"]/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;'}[c]));
 const qq=s=>String(s==null?'':s).replace(/\\/g,'\\\\').replace(/'/g,"\\'");
+// One delegated handler for every queue/worklog row action. Job ids and output paths
+// are client/worker-controlled data (prompt_id, X-Target-GPU, filenames), so they ride
+// esc'd data-* attributes and are NEVER raw-concatenated into an onclick — closing the
+// stored-XSS class at the sink. Most-specific action wins; a bare row click opens it.
+document.addEventListener('click',ev=>{
+  const t=ev.target; if(!t||!t.closest)return; let el;
+  if(el=t.closest('[data-cancel]')){ev.stopPropagation();cancelJob(el.getAttribute('data-cancel'));}
+  else if(el=t.closest('[data-amend]')){ev.stopPropagation();openAmend(el.getAttribute('data-amend'));}
+  else if(el=t.closest('[data-del]')){ev.stopPropagation();delWork(el.getAttribute('data-del'));}
+  else if(el=t.closest('[data-arch]')){ev.stopPropagation();archiveFromQueue(el.getAttribute('data-arch'));}
+  else if(el=t.closest('[data-open]')){ev.stopPropagation();openContext(el.getAttribute('data-open'));}
+});
 const assetByPath=p=>DATA.assets.find(a=>a.path===p);
 const stackIdOf=p=>{for(const s of STACKS)if((s.assetIds||[]).includes(p))return s.id;return null;};
 // Archived Stacks stay OUT of the gallery/pickers, but their members stay hidden
@@ -1535,14 +1563,14 @@ function dragOverStack(e){if(!DRAG)return;e.preventDefault();e.currentTarget.cla
 async function dropOnStack(e,id){e.preventDefault();e.currentTarget.classList.remove('drop-target');const src=DRAG;DRAG=null;if(!src)return;await addAssetsToStack(id,[src]);}
 // ── Create / add ────────────────────────────────────────────────────────────────
 async function createStack(name,ids){
-  const r=await fetch('/v1/stacks',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({name,assetIds:ids})});
+  const r=await fetch('/v1/stacks',{method:'POST',headers:genHeaders(),body:JSON.stringify({name,assetIds:ids})});
  if(!r.ok){toast('Couldn’t create Stack');return null;}
  const s=await r.json();await load();toast('Stack created: '+(s.name||'Untitled Stack'));announce('Stack created');track('stack_create',{n:(ids||[]).length});return s;
 }
 async function createStackFromDrag(a,b){const name=await ask('Name this Stack','Untitled Stack');if(name===null)return;await createStack(name||'Untitled Stack',[a,b]);}
 async function newStackPrompt(){const name=await ask('Name this Stack','Untitled Stack');if(name===null)return;const s=await createStack(name||'Untitled Stack',[]);if(s)openStack(s.id);}
 async function addAssetsToStack(id,ids){
-  const r=await fetch('/v1/stacks/'+encodeURIComponent(id)+'/assets',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({assetIds:ids})});
+  const r=await fetch('/v1/stacks/'+encodeURIComponent(id)+'/assets',{method:'POST',headers:genHeaders(),body:JSON.stringify({assetIds:ids})});
  if(!r.ok){toast('Couldn’t add to Stack');return null;}
  const s=await r.json();await load();if(CURSTACK&&CURSTACK.id===id){CURSTACK=s;svRender();}toast('Added to Stack');announce('Added to Stack');return s;
 }
@@ -1563,12 +1591,12 @@ function moveSelTo(ev){if(!SEL.size)return;const ids=[...SEL];
 }
 async function returnToGallery(ids){
  const by={};for(const p of ids){const sid=stackIdOf(p);if(sid)(by[sid]=by[sid]||[]).push(p);}
-  for(const sid in by)await fetch('/v1/stacks/'+encodeURIComponent(sid)+'/assets',{method:'DELETE',headers:{'Content-Type':'application/json'},body:JSON.stringify({assetIds:by[sid]})});
+  for(const sid in by)await fetch('/v1/stacks/'+encodeURIComponent(sid)+'/assets',{method:'DELETE',headers:genHeaders(),body:JSON.stringify({assetIds:by[sid]})});
  await load();toast('Returned to gallery');
 }
 function downloadSel(){if(!SEL.size)return;for(const p of [...SEL])dl(imgURL({path:p}),p.split('/').pop());toast('Downloading '+SEL.size);}
 async function tagSel(){if(!SEL.size)return;const t=await ask('Add tags (comma-separated)','');if(t===null)return;const tags=t.split(',').map(x=>x.trim()).filter(Boolean);if(!tags.length)return;
-  const r=await fetch('/v1/library/tags',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({paths:[...SEL],tags})});toast(r.ok?('Tagged '+SEL.size):'Tag failed');clearSel();await load();}
+  const r=await fetch('/v1/library/tags',{method:'POST',headers:genHeaders(),body:JSON.stringify({paths:[...SEL],tags})});toast(r.ok?('Tagged '+SEL.size):'Tag failed');clearSel();await load();}
 // ── Stack detail view ─────────────────────────────────────────────────────────
 async function openStack(id){
  const s=await fetchStack(id);if(!s){toast('Stack not found');await load();return;}
@@ -1628,18 +1656,18 @@ function svImgMenu(path,idx,ev){ev&&ev.stopPropagation();
    {label:'🗑 Delete image',danger:true,on:()=>svDeleteImg(path)},
  ],ev);
 }
-async function setCover(path){const r=await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id),{method:'PATCH',headers:{'Content-Type':'application/json'},body:JSON.stringify({cover:path})});if(r.ok){CURSTACK=await r.json();svRender();await loadStacks();toast('Cover updated');}else toast('Couldn’t set cover');}
-async function svRemove(paths){const r=await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id)+'/assets',{method:'DELETE',headers:{'Content-Type':'application/json'},body:JSON.stringify({assetIds:paths})});if(r.ok){CURSTACK=await r.json();svRender();await loadStacks();toast('Removed — back in gallery');announce('Removed from Stack');}else toast('Couldn’t remove');}
+async function setCover(path){const r=await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id),{method:'PATCH',headers:genHeaders(),body:JSON.stringify({cover:path})});if(r.ok){CURSTACK=await r.json();svRender();await loadStacks();toast('Cover updated');}else toast('Couldn’t set cover');}
+async function svRemove(paths){const r=await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id)+'/assets',{method:'DELETE',headers:genHeaders(),body:JSON.stringify({assetIds:paths})});if(r.ok){CURSTACK=await r.json();svRender();await loadStacks();toast('Removed — back in gallery');announce('Removed from Stack');}else toast('Couldn’t remove');}
 function svRemoveSel(){if(!SEL.size)return;const ids=[...SEL];SEL.clear();SELMODE=false;updateBulk();svRemove(ids);}
 function svFix(path){if(!assetByPath(path)){toast('Image unavailable');return;}openFixPath(path,'');}
-async function rerunInStack(path){await enqueue(null,'',async()=>{toast('Queueing into this Stack…');const r=await fetch('/v1/library/rerun',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path,stack:CURSTACK.id})});toast(r.ok?'Queued → lands in this Stack':'Failed');});}
+async function rerunInStack(path){await enqueue(null,'',async()=>{toast('Queueing into this Stack…');const r=await fetch('/v1/library/rerun',{method:'POST',headers:genHeaders(),body:JSON.stringify({path,stack:CURSTACK.id})});toast(r.ok?'Queued → lands in this Stack':'Failed');});}
 async function svDeleteImg(path){if(!await ask('Delete this image? (recoverable in Archive)'))return;
-  await fetch('/v1/library/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path,status:'deleted'})});
-  await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id)+'/assets',{method:'DELETE',headers:{'Content-Type':'application/json'},body:JSON.stringify({assetIds:[path]})});
+  await fetch('/v1/library/status',{method:'POST',headers:genHeaders(),body:JSON.stringify({path,status:'deleted'})});
+  await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id)+'/assets',{method:'DELETE',headers:genHeaders(),body:JSON.stringify({assetIds:[path]})});
  CURSTACK=await fetchStack(CURSTACK.id);await load();svRender();toast('Archived');}
 async function svDeleteSel(){if(!SEL.size)return;if(!await ask('Delete '+SEL.size+' image(s)? (recoverable)'))return;const ids=[...SEL];SEL.clear();SELMODE=false;updateBulk();
-  (async()=>{for(const p of ids)await fetch('/v1/library/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({path:p,status:'deleted'})});
-    await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id)+'/assets',{method:'DELETE',headers:{'Content-Type':'application/json'},body:JSON.stringify({assetIds:ids})});
+  (async()=>{for(const p of ids)await fetch('/v1/library/status',{method:'POST',headers:genHeaders(),body:JSON.stringify({path:p,status:'deleted'})});
+    await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id)+'/assets',{method:'DELETE',headers:genHeaders(),body:JSON.stringify({assetIds:ids})});
    CURSTACK=await fetchStack(CURSTACK.id);await load();svRender();toast('Archived');})();}
 function svDownload(){if(!SEL.size)return;for(const p of [...SEL])dl(imgURL({path:p}),p.split('/').pop());toast('Downloading '+SEL.size);}
 function svMoveTo(){if(!SEL.size)return;moveSelTo({clientX:innerWidth/2,clientY:130,stopPropagation(){}});}
@@ -1651,7 +1679,7 @@ function svDragOver(e,idx){if(SVDRAG===null)return;e.preventDefault();e.currentT
 async function svDrop(e,idx){e.preventDefault();e.currentTarget.classList.remove('drop-target');if(SVDRAG===null||SVDRAG===idx){SVDRAG=null;return;}
  const before=CURSTACK.assets,arr=before.slice(),[m]=arr.splice(SVDRAG,1);arr.splice(idx,0,m);SVDRAG=null;
  CURSTACK.assets=arr;svRender();                                    // optimistic
-  const r=await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id)+'/order',{method:'PATCH',headers:{'Content-Type':'application/json'},body:JSON.stringify({assetIds:arr.map(x=>x.assetId)})});
+  const r=await fetch('/v1/stacks/'+encodeURIComponent(CURSTACK.id)+'/order',{method:'PATCH',headers:genHeaders(),body:JSON.stringify({assetIds:arr.map(x=>x.assetId)})});
  if(!r.ok){CURSTACK.assets=before;svRender();toast('Couldn’t save order — reverted');}
  else{CURSTACK=await r.json();svRender();announce('Order saved');}
 }
@@ -1662,7 +1690,7 @@ function svOpen(i){if(!CURSTACK||!CURSTACK.assets[i])return;SVI=i;const path=CUR
  announce('Image '+(i+1)+' of '+CURSTACK.assets.length);}
 function svStep(d){if(SVI<0||!CURSTACK)return;const n=CURSTACK.assets.length;SVI=(SVI+d+n)%n;svOpen(SVI);}
 // ── Header actions: rename / describe / add images / share / settings ─────────
-async function patchStack(id,fields){const r=await fetch('/v1/stacks/'+encodeURIComponent(id),{method:'PATCH',headers:{'Content-Type':'application/json'},body:JSON.stringify(fields)});if(!r.ok){toast('Update failed');return null;}const s=await r.json();if(CURSTACK&&CURSTACK.id===id){CURSTACK=s;$('svname').value=s.name;}await loadStacks();if(!CURSTACK)render();return s;}
+async function patchStack(id,fields){const r=await fetch('/v1/stacks/'+encodeURIComponent(id),{method:'PATCH',headers:genHeaders(),body:JSON.stringify(fields)});if(!r.ok){toast('Update failed');return null;}const s=await r.json();if(CURSTACK&&CURSTACK.id===id){CURSTACK=s;$('svname').value=s.name;}await loadStacks();if(!CURSTACK)render();return s;}
 async function renameCurStack(v){if(!CURSTACK)return;await patchStack(CURSTACK.id,{name:(v||'').trim()||'Untitled Stack'});toast('Renamed');}
 async function renameStack(id){const s=STACKS.find(x=>x.id===id);const n=await ask('Rename Stack',(s&&s.name)||'');if(n===null)return;await patchStack(id,{name:n.trim()||'Untitled Stack'});toast('Renamed');}
 async function describeCurStack(v){if(!CURSTACK)return;await patchStack(CURSTACK.id,{description:v||''});}
@@ -1670,12 +1698,12 @@ function shareLink(url){if(navigator.clipboard&&navigator.clipboard.writeText)na
 function shareStack(id){shareLink(location.origin+location.pathname+'#stack='+id);}
 function shareCurStack(){if(CURSTACK)shareStack(CURSTACK.id);}
 function downloadStack(id){(async()=>{const st=(CURSTACK&&CURSTACK.id===id)?CURSTACK:await fetchStack(id);if(!st)return;for(const m of st.assets)dl(imgURL({path:m.assetId}),m.assetId.split('/').pop());toast('Downloading '+st.assets.length);})();}
-async function archiveStack(id){const r=await fetch('/v1/stacks/'+encodeURIComponent(id),{method:'PATCH',headers:{'Content-Type':'application/json'},body:JSON.stringify({archived:true})});if(r.ok){if(CURSTACK&&CURSTACK.id===id)closeStack();await load();toast('Archived');}else toast('Couldn’t archive');}
+async function archiveStack(id){const r=await fetch('/v1/stacks/'+encodeURIComponent(id),{method:'PATCH',headers:genHeaders(),body:JSON.stringify({archived:true})});if(r.ok){if(CURSTACK&&CURSTACK.id===id)closeStack();await load();toast('Archived');}else toast('Couldn’t archive');}
 async function unstackStack(id){if(!await ask('Unstack? The container is removed and every image returns to the gallery.'))return;const r=await fetch('/v1/stacks/'+encodeURIComponent(id)+'/unstack',{method:'POST'});if(r.ok){if(CURSTACK&&CURSTACK.id===id)closeStack();await load();toast('Unstacked — images preserved');announce('Unstacked');}else toast('Couldn’t unstack');}
 function pickAddImages(){if(!CURSTACK)return;const f=$('svaddfile');f.onchange=async e=>{const files=e.target.files;if(!files||!files.length)return;toast('Adding '+files.length+' image(s)…');const added=[];
  for(const file of files){const fd=new FormData();fd.append('image',file,file.name||'upload.png');try{const r=await fetch('/v1/library/upload',{method:'POST',body:fd});const j=await r.json().catch(()=>({}));if(r.ok&&j.path)added.push(j.path);}catch(_){}}
  if(added.length)await addAssetsToStack(CURSTACK.id,added);else toast('No images added');e.target.value='';};f.click();}
-function setAutoStack(on){fetch('/v1/stacks/settings',{method:'PATCH',headers:{'Content-Type':'application/json'},body:JSON.stringify({autoStack:!!on})}).then(()=>toast(on?'Batch generations will auto-stack':'Auto-stack off'),()=>toast('Couldn’t save setting'));}
+function setAutoStack(on){fetch('/v1/stacks/settings',{method:'PATCH',headers:genHeaders(),body:JSON.stringify({autoStack:!!on})}).then(()=>toast(on?'Batch generations will auto-stack':'Auto-stack off'),()=>toast('Couldn’t save setting'));}
 // ── Delete-Stack confirmation ─────────────────────────────────────────────────
 function deleteStackPrompt(id){DELSTACK=id;const r=document.querySelector('input[name=delmode][value=only]');if(r)r.checked=true;$('delstackdlg').showModal();}
 async function confirmDeleteStack(){const id=DELSTACK;if(!id)return;const mode=(document.querySelector('input[name=delmode]:checked')||{}).value||'only';
@@ -830,7 +830,10 @@ async def _queue_prompt(request: web.Request, server, graph: dict) -> str:
    nothing queued' — the error was swallowed)."""
    port = getattr(server, "port", None) or int(os.environ.get("PORT", "8188"))
    headers = {"Content-Type": "application/json"}
-    for h in ("Authorization", "Cookie"):
+    # Forward the caller's identity AND their per-GPU target (X-Target-GPU) so a fix/
+    # compose/video/rerun dispatched server-side lands on the machine the user picked —
+    # one targeting path, honored by dispatch_if_worker for every generation surface.
+    for h in ("Authorization", "Cookie", "X-Target-GPU"):
        if request.headers.get(h):
            headers[h] = request.headers[h]
    try:
@@ -895,6 +898,123 @@ def _render_job_node(org: str, pid: str) -> str:
    return (j.get("node") if j else None) or "gpu"


+# ── Authoritative gpu-jobs queue (the tasks engine is the source of truth) ───────────
+# The visible queue reads the org's activities STRAIGHT from the cloud tasks engine
+# (GET /v1/tasks/namespaces/gpu-jobs/activities, org-scoped by the caller's bearer via
+# the tasks gate) — not a local shadow. So every row shows the REAL worker that claimed
+# it (activity.identity) and its REAL status, and nothing renders on a GPU without
+# appearing here (a claimed job IS an activity). render_jobs.json survives ONLY as an
+# optimistic placeholder for the ~2s between enqueue and first claim.
+_JOBS_NS = os.environ.get("STUDIO_GPU_JOBS_NS", "gpu-jobs")
+_PHASE = {"SCHEDULED": "queued", "PENDING": "queued", "STARTED": "running",
+          "RUNNING": "running", "COMPLETED": "done", "SUCCEEDED": "done",
+          "FAILED": "failed", "CANCELED": "cancelled", "CANCELLED": "cancelled"}
+
+
+def _epoch(ts) -> int | None:
+    """Best-effort seconds-since-epoch from a tasks timestamp (RFC3339 string or a
+    number in s/ms). None for empty/unset so age math can skip it."""
+    if not ts:
+        return None
+    if isinstance(ts, (int, float)):
+        return int(ts if ts < 1e12 else ts / 1000)
+    if isinstance(ts, str):
+        try:
+            from datetime import datetime
+            return int(datetime.fromisoformat(ts.strip().replace("Z", "+00:00")).timestamp())
+        except Exception:
+            return None
+    return None
+
+
+def _act_type_name(a: dict) -> str:
+    t = a.get("type") or a.get("activityType") or {}
+    return t.get("name", "") if isinstance(t, dict) else ""
+
+
+def _activity_phase(a: dict) -> str:
+    return _PHASE.get(str(a.get("status", "")).upper(), "queued")
+
+
+def _prompt_title_refs(prompt: dict) -> tuple:
+    """A human title (SaveImage/SaveVideo prefix) + first LoadImage ref for a thumbnail,
+    derived from the activity's OWN input so a row is self-describing without the shadow."""
+    from middleware.gpu_dispatch import SAVE_NODES
+    title, refs = "render", []
+    for n in (prompt or {}).values():
+        if not isinstance(n, dict):
+            continue
+        ct = n.get("class_type", "")
+        if ct in SAVE_NODES:
+            title = (n.get("inputs", {}) or {}).get("filename_prefix") or title
+        elif "LoadImage" in ct:
+            im = (n.get("inputs", {}) or {}).get("image")
+            if im:
+                refs.append(im)
+    return title.split("/")[-1], refs[:1]
+
+
+def _job_from_activity(a: dict, now: int) -> dict:
+    """Map a StandaloneActivity → the UI's job row: REAL claimer + REAL status."""
+    ex = a.get("execution") or {}
+    pid = ex.get("runId") or ex.get("workflowId") or a.get("activityId") or ""
+    inp = a.get("input") if isinstance(a.get("input"), dict) else {}
+    title, refs = _prompt_title_refs(inp.get("prompt") or {})
+    tq = str(a.get("taskQueue") or "")
+    # identity is the machine that CLAIMED it; before pick-up the "gpu:<name>" lane
+    # still names the targeted machine so a queued targeted job shows where it's headed.
+    node = a.get("identity") or (tq[4:] if tq.startswith("gpu:") else None) or "queued"
+    start = _epoch(a.get("startTime")) or _epoch(a.get("scheduleTime"))
+    beat = _epoch(a.get("lastHeartbeatTime"))
+    return {"id": pid, "node": node, "status": _activity_phase(a),
+            "prefix": title, "refs": refs,
+            "age": max(0, now - start) if start else 0,
+            "beat_age": (now - beat) if beat else None}
+
+
+async def _list_gpu_jobs(request) -> list | None:
+    """The caller-org's gpu-jobs activities from the tasks engine. Returns None (not [])
+    when the tasks API can't be reached, so a transient blip falls back to the local
+    placeholder rather than flashing an empty queue. Tenant-scoped by the bearer."""
+    from middleware.gpu_dispatch import CLOUD, _bearer
+    tok = _bearer(request)
+    if not tok:
+        return []
+    url = f"{CLOUD}/v1/tasks/namespaces/{_JOBS_NS}/activities?pageSize=100"
+    try:
+        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=8)) as s:
+            async with s.get(url, headers={"Authorization": tok}) as r:
+                if r.status != 200:
+                    return None
+                data = await r.json()
+    except Exception:
+        return None
+    acts = data.get("activities") if isinstance(data, dict) else None
+    return acts if isinstance(acts, list) else []
+
+
+def _render_queue_placeholder(root, now: int, exclude: set) -> list:
+    """Optimistic pre-claim rows from render_jobs.json: ONLY jobs the authoritative list
+    hasn't surfaced yet and that are seconds-young — the bridge across the ~2s between
+    enqueue and first claim. Anything older is the engine's to report."""
+    try:
+        jobs = json.loads((root / "render_jobs.json").read_text())
+    except Exception:
+        return []
+    out = []
+    for j in jobs:
+        pid = j.get("id")
+        if not pid or pid in exclude:
+            continue
+        age = now - int(j.get("ts", 0))
+        if age < 0 or age > 30:
+            continue
+        out.append({"id": pid, "node": j.get("node") or "queued", "status": "queued",
+                    "prefix": j.get("prefix") or "render", "refs": (j.get("refs") or [])[:1],
+                    "age": age, "beat_age": None})
+    return out
+
+
 def _output_mtime(root: Path | None, rel: str) -> int | None:
    try:
        return int((root / rel).stat().st_mtime) if root and rel else None
@@ -2080,44 +2200,31 @@ def add_studio_home_routes(routes: web.RouteTableDef, server) -> None:

    @routes.get("/v1/render-queue")
    async def render_queue(request: web.Request):
-        """The org's IN-FLIGHT BYO-GPU renders (fix/compose/rerun dispatched to a
-        connected GPU). These run on the worker's gpu-jobs queue, NOT the pod's
-        local /queue, so the UI could not see them — this is what makes a queued
-        fix visible. Drops entries whose output has landed in the library or that
-        have aged out (~25 min). Cheap: reads one small per-org JSON."""
+        """The org's IN-FLIGHT BYO-GPU renders — read AUTHORITATIVELY from the cloud
+        gpu-jobs tasks engine: each row carries the REAL claiming GPU (activity.identity)
+        and REAL status, so a dispatched fix/compose/rerun is visible AND attributed to
+        the machine actually running it — and nothing can render on a GPU without
+        appearing here. render_jobs.json is used ONLY as a ~2s pre-claim placeholder for
+        a job the engine hasn't listed yet. Terminal (done/failed/cancelled) and stale
+        rows drop off. Tenant-scoped by the caller's bearer via the tasks gate."""
        org = _org_of(request)
        root = _library_root(org)
-        if root is None:
-            return web.json_response({"jobs": []})
-        jf = root / "render_jobs.json"
-        try:
-            jobs = json.loads(jf.read_text())
-        except Exception:
-            jobs = []
        now = int(time.time())
-        live = []
-        for j in jobs:
-            age = now - int(j.get("ts", 0))
-            if age > 1500:            # aged out (worker deadline is 1200s)
-                continue
-            # Completed only when the ACTUAL output for THIS job exists: the render
-            # saves to "<outPrefix>_NNNNN_.png". Check precisely (glob the exact
-            # output prefix) — not a fuzzy base match, which collided with the source
-            # image and made the job disappear from the queue after 30s.
-            out_prefix = j.get("outPrefix") or j.get("prefix") or ""
-            landed = False
-            if out_prefix:
-                try:
-                    parent = root / os.path.dirname(out_prefix)
-                    stem = os.path.basename(out_prefix)
-                    landed = any(any(parent.glob(stem + "*" + e)) for e in _SAVE_EXT)
-                except Exception:
-                    landed = False
-            if landed and age > 20:
-                continue
-            j["age"] = age
-            live.append(j)
-        return web.json_response({"jobs": live})
+        acts = await _list_gpu_jobs(request)
+        live, seen = [], set()
+        if acts is not None:
+            for a in acts:
+                if _act_type_name(a) not in ("studio.render", ""):
+                    continue
+                job = _job_from_activity(a, now)
+                if not job["id"]:
+                    continue
+                seen.add(job["id"])
+                if job["status"] in ("done", "failed", "cancelled") or job["age"] > 1500:
+                    continue
+                live.append(job)
+        placeholder = _render_queue_placeholder(root, now, seen) if root is not None else []
+        return web.json_response({"jobs": placeholder + live})

    @routes.post("/v1/library/status")
    async def set_status(request: web.Request):
@@ -2359,20 +2466,35 @@ def add_studio_home_routes(routes: web.RouteTableDef, server) -> None:

    @routes.get("/v1/nodes")
    async def get_nodes(request: web.Request):
-        """The caller-org's GPU nodes (online/offline) for the queue header badges,
-        plus this studio pod. The fleet is token-scoped, so only the caller's org's
-        nodes are ever returned."""
+        """The caller-org's GPU nodes (online/offline) for the queue header badges, plus
+        this studio pod — each annotated with LIVE queue depth read from the authoritative
+        gpu-jobs activities (jobs whose claiming identity, or targeted 'gpu:<name>' lane,
+        is this node). The fleet is token-scoped, so only the caller-org's nodes return."""
        try:
            from middleware.gpu_dispatch import _bearer, _fleet_machines, _node_label
        except ImportError:
            from .middleware.gpu_dispatch import _bearer, _fleet_machines, _node_label
+        now = int(time.time())
+        depth: dict = {}
+        for a in (await _list_gpu_jobs(request) or []):
+            if _act_type_name(a) not in ("studio.render", ""):
+                continue
+            ph = _activity_phase(a)
+            if ph not in ("queued", "running"):
+                continue
+            d = depth.setdefault(_job_from_activity(a, now)["node"], {"queued": 0, "running": 0})
+            d[ph] += 1
        nodes = []
        try:
            tok = _bearer(request)
            for m in (_fleet_machines(tok) if tok else []):
                if not m.get("gpu"):
                    continue
-                nodes.append({"name": _node_label(m), "online": m.get("status") == "online"})
+                name = _node_label(m)
+                d = depth.get(name, {"queued": 0, "running": 0})
+                nodes.append({"name": name, "online": m.get("status") == "online",
+                              "queued": d["queued"], "running": d["running"],
+                              "depth": d["queued"] + d["running"]})
        except Exception:
            nodes = []
        return web.json_response({"nodes": nodes, "pod": {"name": "studio pod", "online": True}})
@@ -7,6 +7,7 @@ When Studio runs with --worker-mode, this module handles:
 3. Reporting device capabilities (GPU model, VRAM, etc.)
 """
 import asyncio
+import hmac
 import logging
 import os
 import time as _time
@@ -17,6 +18,7 @@ import aiohttp
 from aiohttp import web

 import execution
+from studio.cli_args import args

 logger = logging.getLogger(__name__)

@@ -31,10 +33,31 @@ WORKER_TOKEN = os.environ.get("STUDIO_WORKER_TOKEN", "")


 def verify_worker_token(request) -> bool:
-    """True if the request carries the coordinator shared secret (or none is set)."""
+    """True iff the request carries the coordinator shared secret. FAILS CLOSED: in
+    worker mode the secret is MANDATORY (main.py refuses to boot without it, and the
+    KMS /v1 migration has emptied that secret before), so an empty/unset token is a
+    valid single-trust-domain dev signal ONLY when NOT in worker mode — never a bypass
+    that reopens the hidden-run hole on a worker box. Constant-time comparison."""
    if not WORKER_TOKEN:
-        return True
-    return request.headers.get("X-Worker-Token", "") == WORKER_TOKEN
+        return not args.worker_mode
+    return hmac.compare_digest(request.headers.get("X-Worker-Token", ""), WORKER_TOKEN)
+
+
+def reject_untrusted_worker_submit(request, worker_mode: bool):
+    """The ONE policy gate for worker-mode /prompt (decomplected: one function, one
+    place). In worker mode this process is a render BACKEND, not a front — the only
+    accepted submit is a job claimed off the org's gpu-jobs queue and handed in over the
+    coordinator seam (POST /v1/worker/execute with X-Worker-Token). A direct /prompt POST
+    without a valid token is refused (403), so no render can reach a GPU without appearing
+    in the org's visible queue. Returns an aiohttp Response to send, or None to allow.
+    Not worker mode → always allowed (returns None)."""
+    if worker_mode and not verify_worker_token(request):
+        return web.json_response({"error": {
+            "type": "worker_mode_forbidden",
+            "message": "worker-mode renders run only from the gpu-jobs queue "
+                       "(POST /v1/worker/execute); direct /prompt is disabled.",
+            "details": "", "extra_info": {}}, "node_errors": {}}, status=403)
+    return None


 def _worker_headers() -> dict:
@@ -172,6 +195,8 @@ def add_worker_routes(routes: web.RouteTableDef, prompt_server):
            return web.json_response({"error": "No prompt provided"}, status=400)

        prompt = json_data["prompt"]
+        if not isinstance(prompt, dict):
+            return web.json_response({"error": "prompt must be an object"}, status=400)
        prompt_id = str(json_data.get("prompt_id", uuid.uuid4()))

        partial_execution_targets = json_data.get("partial_execution_targets")
@@ -197,8 +222,19 @@ def add_worker_routes(routes: web.RouteTableDef, prompt_server):
            if sensitive_val in extra_data:
                sensitive[sensitive_val] = extra_data.pop(sensitive_val)

-        number = float(json_data.get("number", 0))
+        try:
+            number = float(json_data.get("number", 0))
+        except (TypeError, ValueError):
+            return web.json_response({"error": "number must be numeric"}, status=400)
        extra_data["create_time"] = int(_time.time() * 1000)
+        # `org` scopes outputs on THIS worker's LOCAL disk only — a worker-LOCAL label,
+        # never a gallery destination. The durable upload path re-derives the gallery org
+        # from the user's IAM token (not this body field), so a forged body `org` can at
+        # most mislabel a local temp dir, never write into another tenant's gallery. The
+        # prompt worker binds folder_paths.set_execution_org from extra_data["org_id"].
+        org = json_data.get("org") or extra_data.get("org_id")
+        if org:
+            extra_data["org_id"] = org

        # Queue locally — the worker's prompt_worker thread will pick it up
        prompt_server.prompt_queue.put(
@@ -1,6 +1,6 @@
 [project]
 name = "hanzo-studio"
-version = "0.17.22"
+version = "0.17.23"
 readme = "README.md"
 license = { file = "LICENSE" }
 requires-python = ">=3.10"
@@ -1043,6 +1043,15 @@ class PromptServer():
        async def post_prompt(request):
            logging.info("got prompt")

+            # Execution-engine (worker) mode: this process IS a BYO-GPU render backend,
+            # not a front. A direct /prompt POST without the coordinator token is refused
+            # so no render reaches a GPU without appearing in the org's visible queue.
+            # One way onto a GPU: enqueue -> claim -> execute. (Policy: worker_client.)
+            from middleware.worker_client import reject_untrusted_worker_submit
+            _gate = reject_untrusted_worker_submit(request, args.worker_mode)
+            if _gate is not None:
+                return _gate
+
            # Billing: check balance before accepting prompt
            if args.enable_billing and args.billing_check_balance:
                iam_user = request.get("iam_user")
@@ -1062,7 +1071,11 @@ class PromptServer():
            json_data = self.trigger_on_prompt(json_data)

            # --- Prompt routing: check if this should go to a GPU worker ---
-            if not args.worker_mode:
+            # Legacy in-cluster PUSH federation (prompt_router -> compute_config
+            # workers) is OFF by default: the gpu-jobs enqueue below (dispatch_if_worker)
+            # is the ONE submit path. Set STUDIO_LEGACY_PUSH_ROUTER=1 only to resurrect
+            # the old push seam. Keeps exactly one way onto a GPU in prod.
+            if not args.worker_mode and os.environ.get("STUDIO_LEGACY_PUSH_ROUTER") == "1":
                org_id = self._get_org_id(request)
                try:
                    route_result = await prompt_router.route_prompt(org_id, json_data)
@@ -162,6 +162,84 @@ def test_advisory_gate_false_when_nothing_online(monkeypatch):
    assert g.has_online_gpu_advisory("Bearer user") is False


+# ── Per-GPU targeting: X-Target-GPU → taskQueue "gpu:<identity>" ──────────────────
+
+class _ReqH(_Req):
+    """_Req plus an arbitrary header map (so a test can set X-Target-GPU)."""
+    def __init__(self, auth="Bearer u", headers=None):
+        super().__init__(auth)
+        if headers:
+            self.headers.update(headers)
+
+
+def _capture_body(monkeypatch):
+    """Capture the enqueue POST body so a test can assert taskQueue + the URL namespace."""
+    import json as _json
+    sent: dict = {}
+
+    def fake_urlopen(req, timeout=0):
+        sent["url"] = req.full_url
+        sent["body"] = _json.loads(req.data.decode()) if req.data else {}
+        return _Resp()
+
+    monkeypatch.setattr(g.urllib.request, "urlopen", fake_urlopen)
+    monkeypatch.setattr(g, "_collect_input_images", lambda prompt, org: [])
+    monkeypatch.setattr(g, "_track_dispatched",
+                        lambda org, pid, prompt, node="gpu": sent.update(node=node))
+    return sent
+
+
+def test_untargeted_uses_shared_gpu_jobs_lane(monkeypatch):
+    """No X-Target-GPU → the shared 'gpu-jobs' lane any worker drains; namespace unchanged."""
+    monkeypatch.setattr(g, "_online_gpu_nodes", lambda tok: [])
+    monkeypatch.setattr(g, "has_models_for", lambda prompt: False)
+    sent = _capture_body(monkeypatch)
+    assert g.dispatch_if_worker(_Req("Bearer acme"), "acme", "pid", {}) is True
+    assert sent["body"]["taskQueue"] == "gpu-jobs"            # shared lane
+    assert "namespaces/gpu-jobs/activities" in sent["url"]    # namespace is always gpu-jobs
+    assert sent["node"] == "gpu"
+
+
+def test_x_target_gpu_routes_to_that_machine_lane(monkeypatch):
+    """A targeted render — where the target IS one of the caller-org's own online GPUs —
+    enqueues on the machine's OWN lane 'gpu:<identity>' (namespace still gpu-jobs) and is
+    tracked to that node, so only spark claims it. An in-fleet target ALWAYS enqueues,
+    even on a box that could render the graph locally."""
+    monkeypatch.setattr(g, "_online_gpu_nodes",
+                        lambda tok: [{"name": "spark", "status": "online", "gpu": True}])
+    monkeypatch.setattr(g, "has_models_for", lambda prompt: True)   # would render local…
+    sent = _capture_body(monkeypatch)
+    ok = g.dispatch_if_worker(_ReqH("Bearer acme", {"X-Target-GPU": "spark"}), "acme", "pid", {})
+    assert ok is True                                        # …but an in-fleet target forces the queue
+    assert sent["body"]["taskQueue"] == "gpu:spark"          # the machine's own lane
+    assert "namespaces/gpu-jobs/activities" in sent["url"]   # namespace unchanged
+    assert sent["node"] == "spark"                           # tracked to the targeted GPU
+
+
+def test_out_of_fleet_target_falls_back_to_shared(monkeypatch):
+    """A target that is NOT one of the caller-org's own online GPUs is ignored — the job
+    rides the shared lane, never a 'gpu:<bogus>' lane no worker claims (a 30-min hang),
+    and an injected label (XSS payload) never becomes a taskQueue / queue-row node."""
+    monkeypatch.setattr(g, "_online_gpu_nodes",
+                        lambda tok: [{"name": "spark", "status": "online", "gpu": True}])
+    monkeypatch.setattr(g, "has_models_for", lambda prompt: False)
+    sent = _capture_body(monkeypatch)
+    ok = g.dispatch_if_worker(
+        _ReqH("Bearer acme", {"X-Target-GPU": "<img src=x onerror=alert(1)>"}), "acme", "pid", {})
+    assert ok is True
+    assert sent["body"]["taskQueue"] == "gpu-jobs"           # bogus target → shared lane, not gpu:<payload>
+    assert sent["node"] == "spark"                           # node decided by the real fleet, not the header
+
+
+def test_blank_target_header_falls_back_to_shared(monkeypatch):
+    """An empty/whitespace X-Target-GPU is not a target — shared lane."""
+    monkeypatch.setattr(g, "_online_gpu_nodes", lambda tok: [])
+    monkeypatch.setattr(g, "has_models_for", lambda prompt: False)
+    sent = _capture_body(monkeypatch)
+    assert g.dispatch_if_worker(_ReqH("Bearer acme", {"X-Target-GPU": "  "}), "acme", "pid", {}) is True
+    assert sent["body"]["taskQueue"] == "gpu-jobs"
+
+
 # ── A BYO-GPU video job records its REAL prefix, not the "render" fallback ─────────
 def test_track_savevideo(tmp_path, monkeypatch):
    import json
@@ -254,21 +254,33 @@ def test_landed_video(orgaware):
 # ── 4. the live BYO-GPU queue drops a video job when its mp4 lands ─────────────────
@pytest.mark.asyncio
 async def test_render_queue_video(orgaware, monkeypatch):
+    """The render-queue is AUTHORITATIVE: it reads the org's gpu-jobs activities from the
+    tasks engine and shows the REAL claiming GPU (activity.identity), dropping terminal
+    jobs. A STARTED studio.render is live; a COMPLETED one falls off."""
    outd, _ = orgaware
    root = Path(outd("acme"))
    _lib(root, [])
-    now = int(time.time())
-    (root / "render_jobs.json").write_text(json.dumps(
-        [{"id": "p", "outPrefix": "video/x", "ts": now - 40, "node": "spark"}]))
    render_queue = _handler(_FakeServer(), "GET", "/v1/render-queue", monkeypatch)
+    base = {
+        "execution": {"runId": "p", "workflowId": "p"},
+        "type": {"name": "studio.render"},
+        "taskQueue": "gpu:spark", "identity": "spark",
+        "input": {"prompt": {"12": {"class_type": "SaveVideo",
+                                    "inputs": {"filename_prefix": "video/x"}}}},
+    }

+    async def _started(_req):
+        return [{**base, "status": "STARTED"}]
+    monkeypatch.setattr(sh, "_list_gpu_jobs", _started)
    resp = await render_queue(_Req(org="acme"))
-    assert len(json.loads(resp.text)["jobs"]) == 1              # not landed yet → live
+    jobs = json.loads(resp.text)["jobs"]
+    assert len(jobs) == 1 and jobs[0]["node"] == "spark"       # real claimer shown

-    (root / "video").mkdir(parents=True, exist_ok=True)
-    (root / "video" / "x_00001_.mp4").write_bytes(b"\x00\x00\x00 ftyp")
+    async def _completed(_req):
+        return [{**base, "status": "COMPLETED"}]
+    monkeypatch.setattr(sh, "_list_gpu_jobs", _completed)
    resp = await render_queue(_Req(org="acme"))
-    assert json.loads(resp.text)["jobs"] == []                 # mp4 landed → dropped
+    assert json.loads(resp.text)["jobs"] == []                 # engine says done → dropped


 # ── 5. worklog params + per-kind ETA bucketing ────────────────────────────────────
@@ -0,0 +1,164 @@
+"""worker-mode /prompt gate + the /v1/worker/execute claim seam.
+
+Covers:
+- verify_worker_token / reject_untrusted_worker_submit — the ONE policy gate that makes
+  a direct /prompt POST on a worker box refuse un-tokened submits (403), so no render
+  reaches a GPU without appearing in the org's gpu-jobs queue.
+- POST /v1/worker/execute — validates X-Worker-Token and RUNS the claimed graph on the
+  local prompt_queue, returning {prompt_id}, scoped to the job's org.
+"""
+import sys
+from pathlib import Path
+
+import pytest
+from aiohttp import web
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+# CPU CI runners have no NVIDIA driver: force studio's CPU mode BEFORE importing
+# worker_client (which pulls execution -> studio.model_management, whose import-time
+# device probe would otherwise call torch.cuda and crash on a GPU-less box). Same
+# switch the studio pod boots with (--cpu); harmless on a GPU box.
+import studio.cli_args
+studio.cli_args.args.cpu = True
+
+from middleware import worker_client as wc
+
+
+class _Req:
+    def __init__(self, headers=None):
+        self.headers = headers or {}
+
+
+# ── The gate policy: worker-mode /prompt refuses un-tokened submits ───────────────
+
+def test_empty_token_dev_bypass_ONLY_outside_worker_mode(monkeypatch):
+    """FAIL CLOSED: an empty/unset STUDIO_WORKER_TOKEN is a valid single-trust-domain dev
+    signal ONLY when NOT in worker mode. On a worker box (worker_mode=True) an empty token
+    must NOT open /prompt or /v1/worker/execute — that was the fail-open hole (KMS /v1
+    migration has emptied the secret before)."""
+    monkeypatch.setattr(wc, "WORKER_TOKEN", "")
+    monkeypatch.setattr(wc.args, "worker_mode", False)
+    assert wc.verify_worker_token(_Req()) is True            # local dev, single trust domain
+    monkeypatch.setattr(wc.args, "worker_mode", True)
+    assert wc.verify_worker_token(_Req()) is False           # worker box + empty token → refuse
+
+
+def test_verify_worker_token_enforced(monkeypatch):
+    monkeypatch.setattr(wc, "WORKER_TOKEN", "s3cret")
+    assert wc.verify_worker_token(_Req({"X-Worker-Token": "s3cret"})) is True
+    assert wc.verify_worker_token(_Req({"X-Worker-Token": "nope"})) is False
+    assert wc.verify_worker_token(_Req()) is False
+    # constant-time path (hmac.compare_digest) accepts only the exact secret
+    assert wc.verify_worker_token(_Req({"X-Worker-Token": "s3cret "})) is False
+
+
+def test_gate_allows_when_not_worker_mode(monkeypatch):
+    monkeypatch.setattr(wc, "WORKER_TOKEN", "s3cret")
+    assert wc.reject_untrusted_worker_submit(_Req(), worker_mode=False) is None
+
+
+def test_gate_403_in_worker_mode_without_token(monkeypatch):
+    monkeypatch.setattr(wc, "WORKER_TOKEN", "s3cret")
+    resp = wc.reject_untrusted_worker_submit(_Req(), worker_mode=True)
+    assert resp is not None and resp.status == 403           # direct /prompt refused
+
+
+def test_gate_403_in_worker_mode_with_EMPTY_token(monkeypatch):
+    """Fail closed even when the secret is empty (KMS-emptied): worker mode + empty token
+    still refuses a direct /prompt (403) — runtime belt to main.py's refuse-to-start guard."""
+    monkeypatch.setattr(wc, "WORKER_TOKEN", "")
+    monkeypatch.setattr(wc.args, "worker_mode", True)
+    resp = wc.reject_untrusted_worker_submit(_Req(), worker_mode=True)
+    assert resp is not None and resp.status == 403
+
+
+def test_gate_allows_worker_mode_with_token(monkeypatch):
+    monkeypatch.setattr(wc, "WORKER_TOKEN", "s3cret")
+    ok = _Req({"X-Worker-Token": "s3cret"})
+    assert wc.reject_untrusted_worker_submit(ok, worker_mode=True) is None
+
+
+# ── The claim seam: POST /v1/worker/execute validates token + runs locally ────────
+
+class _Queue:
+    def __init__(self):
+        self.items = []
+
+    def put(self, item):
+        self.items.append(item)
+
+
+class _NRM:
+    def apply_replacements(self, prompt):
+        pass
+
+
+class _Server:
+    def __init__(self):
+        self.prompt_queue = _Queue()
+        self.node_replace_manager = _NRM()
+
+
+@pytest.fixture
+def worker_app(monkeypatch):
+    import execution
+
+    async def fake_validate(pid, prompt, targets=None):
+        return (True, None, list(prompt.keys()), {})
+
+    monkeypatch.setattr(execution, "validate_prompt", fake_validate)
+    monkeypatch.setattr(wc, "WORKER_TOKEN", "s3cret")
+    server = _Server()
+    app = web.Application()
+    routes = web.RouteTableDef()
+    wc.add_worker_routes(routes, server)
+    app.add_routes(routes)
+    app["srv"] = server
+    return app
+
+
+@pytest.mark.asyncio
+async def test_worker_execute_rejects_without_token(worker_app, aiohttp_client):
+    client = await aiohttp_client(worker_app)
+    r = await client.post("/v1/worker/execute",
+                          json={"prompt": {"9": {"class_type": "SaveImage", "inputs": {}}}})
+    assert r.status == 401
+    assert worker_app["srv"].prompt_queue.items == []        # nothing ran
+
+
+@pytest.mark.asyncio
+async def test_worker_execute_runs_claimed_job_with_token(worker_app, aiohttp_client):
+    client = await aiohttp_client(worker_app)
+    graph = {"9": {"class_type": "SaveImage", "inputs": {}}}
+    r = await client.post("/v1/worker/execute",
+                          headers={"X-Worker-Token": "s3cret"},
+                          json={"prompt": graph, "prompt_id": "pid-1", "org": "acme"})
+    assert r.status == 200
+    body = await r.json()
+    assert body["prompt_id"] == "pid-1"
+    items = worker_app["srv"].prompt_queue.items
+    assert len(items) == 1                                    # queued for local render
+    _number, pid, _prompt, extra, _outputs, _sensitive = items[0]
+    assert pid == "pid-1"
+    assert extra["org_id"] == "acme"                          # worker-LOCAL scope label (gallery org re-derived from IAM)
+
+
+@pytest.mark.asyncio
+async def test_worker_execute_rejects_non_dict_prompt(worker_app, aiohttp_client):
+    """A non-object prompt (e.g. an injected string) is a clean 400, not a 500."""
+    client = await aiohttp_client(worker_app)
+    r = await client.post("/v1/worker/execute", headers={"X-Worker-Token": "s3cret"},
+                          json={"prompt": "<img src=x onerror=alert(1)>"})
+    assert r.status == 400
+    assert worker_app["srv"].prompt_queue.items == []         # nothing ran
+
+
+@pytest.mark.asyncio
+async def test_worker_execute_rejects_non_numeric_number(worker_app, aiohttp_client):
+    """A non-numeric `number` is a clean 400, not a float() 500."""
+    client = await aiohttp_client(worker_app)
+    r = await client.post("/v1/worker/execute", headers={"X-Worker-Token": "s3cret"},
+                          json={"prompt": {"9": {"class_type": "SaveImage", "inputs": {}}},
+                                "number": "not-a-number"})
+    assert r.status == 400
@@ -0,0 +1,53 @@
+"""Stored-XSS regression for the queue / worklog rows (middleware/studio_home.html).
+
+A malicious job field — j.node (← X-Target-GPU), j.prefix/title (← SaveImage prefix),
+desc (← prompt text), j.id (← client prompt_id) — must render INERT. Two layers:
+  (1) executable: the file's OWN esc() neutralizes an <img onerror> payload (run under
+      node; skipped only if node is unavailable);
+  (2) static: the row builders route every field through esc() and drive row actions via
+      esc'd data-* attributes + one delegated listener — never a raw-concatenated onclick.
+"""
+import re
+from pathlib import Path
+
+HTML = (Path(__file__).resolve().parents[2] / "middleware" / "studio_home.html").read_text()
+PAYLOAD = "<img src=x onerror=alert(1)>"
+
+
+def test_file_esc_map_neutralizes_payload():
+    """Apply the FILE's OWN esc() character map (extracted from studio_home.html) to the
+    payload and assert it renders INERT — every '<'/'>' neutralized, no live tag survives.
+    Pure-Python (no node dep) so it runs in CI; weakening the map (e.g. dropping '<')
+    fails here."""
+    m = re.search(r"const esc=.*?replace\(/\[([^\]]+)\]/g,\s*c=>\((\{[^}]+\})", HTML)
+    assert m, "esc() helper/map not found"
+    charclass, mapsrc = m.group(1), m.group(2)
+    emap = dict(re.findall(r"'(.)':'([^']+)'", mapsrc))
+    assert {"<", ">", "&", '"'} <= set(charclass) and {"<", ">"} <= emap.keys()
+    out = "".join(emap.get(ch, ch) for ch in PAYLOAD)
+    assert "<img" not in out and "<" not in out                    # every '<' neutralized
+    assert emap["<"] in out                                        # escaped entity present, inert text
+
+
+def test_gjRow_escapes_fields_and_delegates_cancel():
+    assert "${esc(j.prefix||'render')}" in HTML                     # SaveImage prefix escaped
+    assert "esc((j.node&&j.node!=='queued')?j.node:'your GPU')" in HTML  # X-Target-GPU/identity escaped
+    assert 'data-cancel="${esc(j.id)}"' in HTML                    # id via esc'd data attr
+
+
+def test_qitem_escapes_title_desc_and_delegates_cancel():
+    assert "${esc(title)}" in HTML
+    assert "${esc(desc)||'(no description)'}" in HTML
+    assert 'data-cancel="${esc(id)}"' in HTML
+
+
+def test_no_raw_id_onclick_sinks_remain():
+    for sink in ("cancelJob('${j.id}')", "cancelJob('${id}')", "cancelJob('${it.id}')",
+                 "openAmend('${it.id}')", "openContext('${it.id}')", "delWork('${it.id}')"):
+        assert sink not in HTML, f"raw client-id onclick sink still present: {sink}"
+
+
+def test_one_delegated_listener_wires_data_actions():
+    assert "t.closest('[data-cancel]')" in HTML
+    assert "cancelJob(el.getAttribute('data-cancel'))" in HTML
+    assert "openAmend(el.getAttribute('data-amend'))" in HTML