fix: sync streaming and PE fixes while preserving ZeroGPU integration

Align shared modules with JoyAI-Video-Edit d88456a, including RV2V prompt
enhancement, streaming cleanup, complete file recordings and API defaults.
Honor explicit static-KV overrides while defaulting file input to off.

Run the shared session handler through the process-queue adapter instead of
maintaining a separate session loop. Keep ZeroGPU acquisition, leases, AOTI,
HF dependency choices, quota UI and team gallery intact.

requirements.txt

@@ -18,7 +18,7 @@ opencv-python-headless==4.13.0.92
 av==13.1.0
 imageio-ffmpeg==0.6.0
 websockets==16.0
-openai==2.41.0
+httpx==0.28.1
 loguru==0.7.3
 
 # Prebuilt custom CUDA kernels (cp310, torch 2.9.1 ABI, sm_120) are NOT listed

static/index.html

@@ -417,12 +417,14 @@ const I18N = {
     gate_settling: "保持不动，正在对焦",
     gate_no_face_wait: "没有检测到人脸，请正对摄像头 😊",
     toast_done: "✨ 编辑完成，点右下角气泡下载",
+    download_retry: "重试下载",
+    download_failed: "下载失败：",
     toast_finalizing: "✨ 编辑完成，正在生成可下载视频",
     edit_processing: "✨ 正在编辑",
     q_downgrade_auto: " · 已自动降到低画质", q_downgrade_hint: " · 建议降低画质",
     flow_drop_tail: " · 带宽不足，正在丢帧保延迟",
     up_auto_tail: " · 上行已自动降至中档",
-    res_notice: "本演示运行在 {res}，而非模型原生的 1248×720 分辨率，并且没有提示词增强，画质与编辑效果会有所下降。",
+    res_notice: "本演示运行在 {res}，而非模型原生的 1248×720 分辨率，画质与编辑效果会有所下降。",
   },
   en: {
     tagline: "Streaming video editing",
@@ -495,12 +497,14 @@ const I18N = {
     gate_settling: "Hold still, focusing",
     gate_no_face_wait: "No face detected, please face the camera 😊",
     toast_done: "✨ Done — click the bubble at bottom-right to download",
+    download_retry: "Retry download",
+    download_failed: "Download failed: ",
     toast_finalizing: "✨ Done — generating the downloadable video",
     edit_processing: "✨ Editing",
     q_downgrade_auto: " · auto-lowered to low quality", q_downgrade_hint: " · consider lowering quality",
     flow_drop_tail: " · bandwidth limited, dropping frames",
     up_auto_tail: " · uplink auto-lowered to mid",
-    res_notice: "This demo runs at {res}, not the model's native 1248×720 resolution, and without prompt enhancement; visual quality and edit fidelity are reduced.",
+    res_notice: "This demo runs at {res}, not the model's native 1248×720 resolution; visual quality and edit fidelity are reduced.",
   },
 };
 let currentLang = localStorage.getItem("joyomni_lang") || "en";
@@ -736,6 +740,8 @@ let tickInFlight = false;
 let pePausedSend = false;
 let peDeferThisSend = false;
 let videoPullActive = false;
+let videoPullGeneration = 0;
+let videoPullTimer = null;
 let videoPullIdx = 0;
 const videoPullFps = 24;
 let videoPullTotal = 0;
@@ -752,6 +758,8 @@ function stopRecordDrain() {
 let startingRun = false;
 let sessionGranted = false;
 let peCache = null;
+let sourceRevision = 0;
+let activeSessionId = null;
 let pendingPeCacheKey = null;
 let outputPlaybackTimer = null;
 let outputPlaybackIntervalMs = null;
@@ -781,10 +789,8 @@ let lastObjectUrl = null;
 let alignMaxAgeMs = 0;
 let lastShownCaptureMs = 0;
 let pendingRevokeUrl = null;
-let lastSourceObjectUrl = null;
 let refImagePreviewDataUrl = null;
 let injectedRefImageDataUrl = null;
-let suppressPeThisSend = false;
 let outputWaitingFace = false;
 let waitHintReason = null;
 let waitHintAt = 0;
@@ -838,6 +844,7 @@ function backendPendingFrames() {
 }
 
 function noteBackendAck(msg) {
+  if (msg && msg.session_id && msg.session_id !== activeSessionId) return;
   const framesIn = Number(msg && msg.frames_in);
   if (Number.isFinite(framesIn)) {
     backendAckedFrames = Math.max(backendAckedFrames, framesIn);
@@ -872,6 +879,7 @@ function orientDims(o) {
   return o === "portrait" ? { width: short, height: long } : { width: long, height: short };
 }
 function applyOrientation(o) {
+  invalidatePeSource();
   const dims = orientDims(o);
   document.getElementById("width").value = dims.width;
   document.getElementById("height").value = dims.height;
@@ -892,7 +900,14 @@ function maxTemporalIdsValue() {
 }
 
 function peCacheKey(rawPrompt, refImage) {
-  return JSON.stringify({ prompt: rawPrompt || "", refImage: refImage || "" });
+  return JSON.stringify({ prompt: rawPrompt || "", refImage: refImage || "", sourceRevision,
+    width: document.getElementById("width").value, height: document.getElementById("height").value });
+}
+
+function invalidatePeSource() {
+  sourceRevision += 1;
+  peCache = null;
+  pendingPeCacheKey = null;
 }
 
 function startFrameTimer() {
@@ -913,7 +928,7 @@ function startNetPingTimer() {
   if (netPingTimer) clearInterval(netPingTimer);
   netPingTimer = setInterval(() => {
     if (ws && ws.readyState === WebSocket.OPEN) {
-      try { ws.send(JSON.stringify({ type: "ping", t: Date.now(), recv: receivedFrames, rtt: netRttLastMs })); } catch (err) {}
+      try { ws.send(JSON.stringify({ session_id: activeSessionId, type: "ping", t: Date.now(), recv: receivedFrames, rtt: netRttLastMs })); } catch (err) {}
     }
     autoQualityTick();
   }, 1000);
@@ -941,40 +956,18 @@ function readRefImageDataUrl() {
   });
 }
 
-function applyRefImagePreview() {
-  if (!refImagePreviewDataUrl) return false;
-  if (lastSourceObjectUrl) URL.revokeObjectURL(lastSourceObjectUrl);
-  lastSourceObjectUrl = null;
-  return true;
-}
-
-function updateSourceFrameVisibility() {
-  const enabled = refImageSelected() || !!refImagePreviewDataUrl;
-  if (!enabled) {
-    if (lastSourceObjectUrl) URL.revokeObjectURL(lastSourceObjectUrl);
-    lastSourceObjectUrl = null;
-    return;
-  }
-  applyRefImagePreview();
-}
-
 async function handleRefImageChange() {
   injectedRefImageDataUrl = null;
   if (!refImageSelected()) {
     refImagePreviewDataUrl = null;
-    if (lastSourceObjectUrl) URL.revokeObjectURL(lastSourceObjectUrl);
-    lastSourceObjectUrl = null;
-    updateSourceFrameVisibility();
     syncRefThumb();
     updateMetrics();
     return;
   }
   try {
     await readRefImageDataUrl();
-    applyRefImagePreview();
   } catch (err) {
   }
-  updateSourceFrameVisibility();
   syncRefThumb();
   updateMetrics();
 }
@@ -1246,10 +1239,7 @@ function resetRunMetrics() {
   if (pendingRevokeUrl) { URL.revokeObjectURL(pendingRevokeUrl); pendingRevokeUrl = null; }
   outputImg.onload = outputImg.onerror = null;
   outputImg.removeAttribute("src");
-  if (lastSourceObjectUrl) URL.revokeObjectURL(lastSourceObjectUrl);
-  lastSourceObjectUrl = null;
   hidePeResult();
-  updateSourceFrameVisibility();
   sessionProfileTimings = false;
   stageTimings.innerHTML = "";
   stageTimingsPanel.style.display = "none";
@@ -1347,8 +1337,10 @@ function resetUplinkEncoder() {
 
 function ensureUplinkEncoder(width, height) {
   if (upEncoder && upEncoder.state === "configured") return;
+  const sessionId = activeSessionId;
   upEncoder = new VideoEncoder({
     output: (chunk) => {
+      if (sessionId !== activeSessionId) return;
       const info = upPendingByTs.get(chunk.timestamp);
       upPendingByTs.delete(chunk.timestamp);
       if (!info || !ws || ws.readyState !== WebSocket.OPEN) return;
@@ -1359,7 +1351,7 @@ function ensureUplinkEncoder(width, height) {
       const buf = new ArrayBuffer(chunk.byteLength);
       chunk.copyTo(buf);
       try {
-        ws.send(JSON.stringify({ type: "frame_meta", seq: info.seq, t_capture_ms: info.t_capture_ms }));
+        ws.send(JSON.stringify({ session_id: activeSessionId, type: "frame_meta", seq: info.seq, t_capture_ms: info.t_capture_ms }));
         ws.send(buf);
       } catch (e) {
         return;
@@ -1399,12 +1391,15 @@ function resetOutputDecoder() {
 
 function feedOutputH264(buf, meta) {
   if (!outDecoder || outDecoder.state === "closed") {
+    const sessionId = activeSessionId;
     outDecoder = new VideoDecoder({
       output: (frame) => {
+        if (sessionId !== activeSessionId) { frame.close(); return; }
         const m = outMetaByTs.get(frame.timestamp);
         outMetaByTs.delete(frame.timestamp);
         createImageBitmap(frame).then((bmp) => {
           frame.close();
+          if (sessionId !== activeSessionId) { bmp.close(); return; }
           enqueueOutputFrame(bmp, m);
         }, () => { frame.close(); });
       },
@@ -1422,6 +1417,10 @@ function feedOutputH264(buf, meta) {
 }
 
 function enqueueOutputFrame(frame, meta) {
+  if (meta && meta.session_id && meta.session_id !== activeSessionId) {
+    if (frame instanceof ImageBitmap) frame.close();
+    return;
+  }
   const item = frame instanceof ImageBitmap
     ? { bmp: frame, meta, receivedAtMs: Date.now() }
     : { url: URL.createObjectURL(frame), meta, receivedAtMs: Date.now() };
@@ -1581,6 +1580,7 @@ function frameBlob(width, height, quality) {
 }
 
 async function tick() {
+  const generation = videoPullGeneration;
   if (!ws || ws.readyState !== WebSocket.OPEN) return;
   if (pePausedSend) return;
   if (tickInFlight) {
@@ -1603,11 +1603,11 @@ async function tick() {
       return;
     }
     const blob = await frameBlob(width, height, effectiveUpQuality());
-    if (!blob || ws.readyState !== WebSocket.OPEN) return;
+    if (generation !== videoPullGeneration || !blob || !ws || ws.readyState !== WebSocket.OPEN) return;
     const seq = sentFrames + 1;
     const tCaptureMs = Date.now();
     try {
-      ws.send(JSON.stringify({ type: "frame_meta", seq, t_capture_ms: tCaptureMs }));
+      ws.send(JSON.stringify({ session_id: activeSessionId, type: "frame_meta", seq, t_capture_ms: tCaptureMs }));
       if (ws.readyState !== WebSocket.OPEN) return;
       ws.send(blob);
     } catch (err) {
@@ -1617,7 +1617,7 @@ async function tick() {
     sentFramesWindow += 1;
     updateMetrics();
   } finally {
-    tickInFlight = false;
+    if (generation === videoPullGeneration) tickInFlight = false;
   }
 }
 
@@ -1630,17 +1630,17 @@ function seekVideoTo(timeSec) {
     setTimeout(finish, 400);
   });
 }
-async function sendCurrentVideoFrame(mediaTimeSec) {
+async function sendCurrentVideoFrame(mediaTimeSec, generation = videoPullGeneration) {
   if (!ws || ws.readyState !== WebSocket.OPEN) return false;
   const width = Number(document.getElementById("width").value);
   const height = Number(document.getElementById("height").value);
   const quality = effectiveUpQuality();
   const blob = await frameBlob(width, height, quality);
-  if (!blob || ws.readyState !== WebSocket.OPEN) return false;
+  if (generation !== videoPullGeneration || !blob || !ws || ws.readyState !== WebSocket.OPEN) return false;
   const seq = sentFrames + 1;
   const tCaptureMs = Math.round(mediaTimeSec * 1000);
   try {
-    ws.send(JSON.stringify({ type: "frame_meta", seq, t_capture_ms: tCaptureMs }));
+    ws.send(JSON.stringify({ session_id: activeSessionId, type: "frame_meta", seq, t_capture_ms: tCaptureMs }));
     if (ws.readyState !== WebSocket.OPEN) return false;
     ws.send(blob);
   } catch (err) {
@@ -1651,8 +1651,8 @@ async function sendCurrentVideoFrame(mediaTimeSec) {
   updateMetrics();
   return true;
 }
-async function pullNextVideoFrame() {
-  if (!videoPullActive) return;
+async function pullNextVideoFrame(generation) {
+  if (generation !== videoPullGeneration || !videoPullActive) return;
   if (!ws || ws.readyState !== WebSocket.OPEN) { videoPullActive = false; return; }
   if (pePausedSend) return;
   if (videoPullIdx >= videoPullTotal) {
@@ -1663,30 +1663,42 @@ async function pullNextVideoFrame() {
   const frameStartMs = Date.now();
   const intervalMs = 1000 / Math.max(1, videoPullFps);
   if (backendPendingFrames() >= MAX_BACKEND_PENDING_FRAMES) {
-    setTimeout(() => { pullNextVideoFrame(); }, 30);
+    videoPullTimer = setTimeout(() => pullNextVideoFrame(generation), 30);
     return;
   }
   const t = videoPullIdx / videoPullFps;
   await seekVideoTo(t);
-  if (!videoPullActive || pePausedSend) return;
-  const ok = await sendCurrentVideoFrame(t);
-  if (!videoPullActive) return;
+  if (generation !== videoPullGeneration || !videoPullActive || pePausedSend) return;
+  const ok = await sendCurrentVideoFrame(t, generation);
+  if (generation !== videoPullGeneration || !videoPullActive) return;
   if (ok) videoPullIdx += 1;
   const spent = Date.now() - frameStartMs;
   const wait = ok ? Math.max(0, intervalMs - spent) : 30;
-  setTimeout(() => { pullNextVideoFrame(); }, wait);
+  videoPullTimer = setTimeout(() => pullNextVideoFrame(generation), wait);
 }
 function startVideoPull() {
+  if (videoPullActive) return;
+  stopVideoPull();
   const dur = (Number.isFinite(camera.duration) && camera.duration > 0) ? camera.duration : 0;
   videoPullTotal = dur > 0 ? Math.max(1, Math.ceil(dur * videoPullFps)) : Number.MAX_SAFE_INTEGER;
   videoPullIdx = 0;
   videoPullActive = true;
-  pullNextVideoFrame();
+  pullNextVideoFrame(videoPullGeneration);
+}
+function stopVideoPull() {
+  videoPullActive = false;
+  videoPullGeneration += 1;
+  if (videoPullTimer !== null) clearTimeout(videoPullTimer);
+  videoPullTimer = null;
 }
-function stopVideoPull() { videoPullActive = false; }
 
 async function beginSession() {
   if (!ws || ws.readyState !== WebSocket.OPEN) return false;
+  stopVideoPull();
+  activeSessionId = String(Number(activeSessionId) + 1);
+  const sessionId = activeSessionId;
+  const socket = ws;
+  if (!usingVideoFile) invalidatePeSource();
   resetRunMetrics();
   pePausedSend = false;
   clearResultVideo();
@@ -1695,17 +1707,14 @@ async function beginSession() {
     try { camera.pause(); } catch (err) {}
     try { camera.currentTime = 0; } catch (err) {}
   }
-  updateSourceFrameVisibility();
   let refImage = null;
   try {
     refImage = await readRefImageDataUrl();
-    if (refImage) applyRefImagePreview();
   } catch (err) {
   }
+  if (sessionId !== activeSessionId || socket !== ws) return false;
   const rawPrompt = document.getElementById("prompt").value;
-  const peSuppressed = suppressPeThisSend;
-  suppressPeThisSend = false;
-  const usePe = !peSuppressed && !refImage && document.getElementById("usePe").checked;
+  const usePe = document.getElementById("usePe").checked;
   const width = Number(document.getElementById("width").value);
   const height = Number(document.getElementById("height").value);
   const cacheKey = peCacheKey(rawPrompt, refImage);
@@ -1731,6 +1740,7 @@ async function beginSession() {
   }
   const startPayload = {
     type: "start",
+    session_id: activeSessionId,
     prompt: rawPrompt,
     width,
     height,
@@ -1750,15 +1760,14 @@ async function beginSession() {
   }
   if (cachedEnhancedPrompt) startPayload.enhanced_prompt = cachedEnhancedPrompt;
   if (refImage) startPayload.ref_image = refImage;
-  const maxTemporalIds = maxTemporalIdsValue();
-  if (maxTemporalIds !== null) startPayload.max_temporal_ids = maxTemporalIds;
-  startPayload.freeze_kv_on_static = document.getElementById("freezeKvOnStatic").checked;
+  startPayload.max_temporal_ids = maxTemporalIdsValue();
+  startPayload.freeze_kv_on_static = !usingVideoFile && document.getElementById("freezeKvOnStatic").checked;
   const sdt = parseFloat(document.getElementById("staticDiffThresh").value);
   startPayload.static_diff_thresh = Number.isFinite(sdt) ? sdt : 0.5;
   startPayload.profile_timings = document.getElementById("profileTimings").checked;
   sessionProfileTimings = startPayload.profile_timings;
   stageTimingsPanel.style.display = sessionProfileTimings ? "" : "none";
-  if (!ws || ws.readyState !== WebSocket.OPEN) return false;
+  if (sessionId !== activeSessionId || socket !== ws || !ws || ws.readyState !== WebSocket.OPEN) return false;
   ws.send(JSON.stringify(startPayload));
   return true;
 }
@@ -1807,8 +1816,10 @@ async function start() {
   socket.onopen = async () => {
   };
   socket.onmessage = async (event) => {
+    if (ws !== socket) return;
     if (typeof event.data === "string") {
       const msg = JSON.parse(event.data);
+      if (msg.session_id && msg.session_id !== activeSessionId) return;
       noteBackendAck(msg);
       if (msg.type === "queue_position") {
         const ahead = Number(msg.ahead != null ? msg.ahead : msg.position) || 0;
@@ -1821,6 +1832,7 @@ async function start() {
         return;
       }
       if (msg.type === "started") {
+        peDeferThisSend = msg.pe_deferred ?? (peDeferThisSend && msg.use_pe !== false);
         if (Number(msg.width) > 0 && Number(msg.height) > 0) {
           document.getElementById("width").value = msg.width;
           document.getElementById("height").value = msg.height;
@@ -1849,8 +1861,9 @@ async function start() {
         if (usingVideoFile && peDeferThisSend) {
           ensureOutputPlaybackTimer(true);
           startNetPingTimer();
+          const generation = videoPullGeneration;
           await seekVideoTo(0);
-          await sendCurrentVideoFrame(0);
+          if (generation === videoPullGeneration) await sendCurrentVideoFrame(0, generation);
         } else {
           startFrameTimer();
         }
@@ -1899,7 +1912,8 @@ async function start() {
         clearOutputQueue();
         lastShownCaptureMs = Date.now();
         alignMaxAgeMs = 0;
-        if (usingVideoFile) {
+        if (usingVideoFile && peDeferThisSend && !msg.cached) {
+          peDeferThisSend = false;
           try { camera.pause(); } catch (err) {}
           startVideoPull();
         }
@@ -1982,7 +1996,7 @@ async function start() {
         return;
       }
       if (msg.type === "chunk_done") {
-        try { ws.send(JSON.stringify({ type: "ack", recv: receivedFrames })); } catch (err) {}
+        try { ws.send(JSON.stringify({ session_id: activeSessionId, type: "ack", recv: receivedFrames })); } catch (err) {}
         if (
           chunkProfileLast &&
           (msg.ws_send_s !== undefined || msg.server_residence_s !== undefined) &&
@@ -2070,12 +2084,13 @@ async function start() {
       }
       return;
     }
-    if (startingRun) return;
+    if (startingRun || !pendingOutputMeta ||
+        (pendingOutputMeta.session_id && pendingOutputMeta.session_id !== activeSessionId)) return;
     const meta = pendingOutputMeta;
     pendingOutputMeta = null;
     receivedFrames += 1;
     if (receivedFrames % 4 === 0) {
-      try { ws.send(JSON.stringify({ type: "ack", recv: receivedFrames })); } catch (err) {}
+      try { ws.send(JSON.stringify({ session_id: activeSessionId, type: "ack", recv: receivedFrames })); } catch (err) {}
     }
     receivedFramesWindow += 1;
     if (meta) {
@@ -2107,11 +2122,11 @@ async function start() {
     clearOutputQueue();
     stopRecordDrain();
     finalizeIsForResult = false;
-    resultEpoch += 1;
+    if (!lastRecId) resultEpoch += 1;
     ws = null;
     startingRun = false;
     sessionGranted = false;
-    if (!terminalHold) {
+    if (!terminalHold && !lastRecId) {
       setSendBusy(false, t("busy_disconnected"));
       showOutputIdle(true);
     }
@@ -2141,7 +2156,7 @@ function stopActiveRun() {
   ws = null;
   if (currentWs) {
     try {
-      if (currentWs.readyState === WebSocket.OPEN) currentWs.send(JSON.stringify({ type: "stop" }));
+      if (currentWs.readyState === WebSocket.OPEN) currentWs.send(JSON.stringify({ session_id: activeSessionId, type: "stop" }));
     } catch (err) {}
     currentWs.onopen = null;
     currentWs.onmessage = null;
@@ -2172,7 +2187,7 @@ async function send() {
   await start();
 }
 
-sendBtn.onclick = () => { suppressPeThisSend = false; send(); };
+sendBtn.onclick = send;
 
 const PROMPT_GROUPS = [
   {
@@ -2265,16 +2280,13 @@ function applyCaseSelection(item, group) {
     if (el) el.value = "";
     injectedRefImageDataUrl = REF_IMAGES[item.ref];
     refImagePreviewDataUrl = injectedRefImageDataUrl;
-    applyRefImagePreview();
   } else if (!refImageSelected()) {
     injectedRefImageDataUrl = null;
     refImagePreviewDataUrl = null;
   }
-  updateSourceFrameVisibility();
   syncRefThumb();
   closeCardsPop();
   updateMetrics();
-  suppressPeThisSend = false;
   send();
 }
 
@@ -2293,14 +2305,10 @@ function syncRefThumb() {
   syncPeAvailability();
 }
 
-function refImageActive() {
-  return refImageSelected() || !!injectedRefImageDataUrl || !!refImagePreviewDataUrl;
-}
-
 function syncPeAvailability() {
   const usePeEl = document.getElementById("usePe");
   const peAvailable = !(SERVER_DEFAULTS && SERVER_DEFAULTS.pe_available === false);
-  const locked = refImageActive() || !peAvailable;
+  const locked = !peAvailable;
   if (usePeEl) {
     usePeEl.disabled = locked;
     if (!peAvailable) usePeEl.checked = false;
@@ -2368,6 +2376,7 @@ syncPeAvailability();
 function handleVideoFileChange(event) {
   const file = event.target.files && event.target.files[0];
   if (!file) return;
+  invalidatePeSource();
   stopActiveRun();
   camera.onended = null;
   camera.onloadedmetadata = null;
@@ -2431,12 +2440,13 @@ function handleVideoEnded() {
       showOutputIdle(true);
       setSendBusy(false, "");
     }, 30000);
-    try { ws.send(JSON.stringify({ type: "finalize_recording" })); }
+    try { ws.send(JSON.stringify({ session_id: activeSessionId, type: "finalize_recording" })); }
     catch (err) {}
   }
 }
 
 function switchToCamera() {
+  invalidatePeSource();
   stopActiveRun();
   startingRun = false;
   sessionGranted = false;
@@ -2547,6 +2557,7 @@ function stopResultSync() {
   try { if (usingVideoFile) camera.pause(); } catch (e) {}
 }
 function clearResultVideo() {
+  if (downloadBubble) downloadBubble.disabled = false;
   stopResultSync();
   finalizeIsForResult = false;
   lastRecId = null;
@@ -2578,17 +2589,19 @@ function triggerBrowserDownload(url, filename) {
 }
 function fetchAndSaveRecording() {
   if (resultBlobUrl) triggerBrowserDownload(resultBlobUrl, "joyomni_output.mp4");
+  else if (lastRecId) fetchAndShowResult();
 }
 async function fetchAndShowResult() {
   const myEpoch = resultEpoch;
+  setSendBusy(true, t("toast_finalizing"));
+  if (downloadBubble) downloadBubble.disabled = true;
   try {
     const resp = await fetch("/download_last?rec=" + encodeURIComponent(lastRecId), { cache: "no-store" });
     if (myEpoch !== resultEpoch) return;
     if (!resp.ok) {
       let detail = "";
       try { const j = await resp.json(); detail = j.error || ""; } catch (e) {}
-      setSendBusy(false, "");
-      return;
+      throw new Error(detail || `HTTP ${resp.status}`);
     }
     const blob = await resp.blob();
     if (myEpoch !== resultEpoch) return;
@@ -2597,20 +2610,33 @@ async function fetchAndShowResult() {
     showResultVideo(resultBlobUrl);
     if (usingVideoFile) startResultSync();
     showDownloadBubble(true);
+    if (downloadBubble) downloadBubble.textContent = t("download_done");
     showOutputToast(t("toast_done"), 4000);
-    setSendBusy(false, "");
   } catch (err) {
+    if (myEpoch !== resultEpoch) return;
+    if (outputWaitOverlay) outputWaitOverlay.style.display = "none";
+    showOutputToast(t("download_failed") + String(err.message || err), 8000);
+    if (lastRecId) {
+      showDownloadBubble(true);
+      if (downloadBubble) downloadBubble.textContent = t("download_retry");
+    }
+  } finally {
+    if (myEpoch === resultEpoch) {
+      setSendBusy(false, "");
+      if (downloadBubble) downloadBubble.disabled = false;
+    }
   }
 }
 function onRecordingFinalized(msg) {
   if (finalizeWatchdog) { clearTimeout(finalizeWatchdog); finalizeWatchdog = null; }
-  lastRecId = msg.rec || null;
+  lastRecId = msg.ok === false ? null : (msg.rec || null);
   if (!lastRecId) {
     finalizeIsForResult = false;
     stopResultSync();
     if (outputWaitOverlay) outputWaitOverlay.style.display = "none";
     showOutputIdle(true);
     setSendBusy(false, "");
+    if (msg.message) showOutputToast(msg.message, 8000);
     return;
   }
   finalizeIsForResult = false;
@@ -2752,7 +2778,7 @@ let autoQLowStreak = 0;
 let autoQHoldTicks = 0;
 function sendOutputQuality(v) {
   if (ws && ws.readyState === WebSocket.OPEN && sessionGranted) {
-    try { ws.send(JSON.stringify({ type: "set_output_quality", value: v })); } catch (err) {}
+    try { ws.send(JSON.stringify({ session_id: activeSessionId, type: "set_output_quality", value: v })); } catch (err) {}
   }
 }
 function setDownQuality(v, opts) {
@@ -2858,9 +2884,9 @@ function applyServerDefaults() {
     if (v !== null && v !== undefined) document.getElementById(id).checked = !!v;
   }
   const maxIds = d.max_temporal_ids;
-  if (maxIds !== null && maxIds !== undefined) {
-    document.getElementById("useMaxTemporalIds").checked = true;
-    document.getElementById("maxTemporalIds").value = maxIds;
+  if (maxIds !== undefined) {
+    document.getElementById("useMaxTemporalIds").checked = maxIds !== null;
+    if (maxIds !== null) document.getElementById("maxTemporalIds").value = maxIds;
   }
   if (d.output_quality !== undefined && d.output_quality !== "auto") {
     autoQuality = false;
@@ -2884,7 +2910,6 @@ window.addEventListener("load", () => {
   applyLang();
   applyOrientation(defaultCameraOrientation());
   updateMaxTemporalIdsVisibility();
-  updateSourceFrameVisibility();
   syncRefThumb();
   showOutputIdle(true);
   updateMetrics();

xvideo/models/pipeline.py

@@ -314,38 +314,39 @@ class Pipeline(DiffusionPipeline):
             return Pipeline._KV_CACHE_ID_REF_IMAGE
         raise ValueError(f"Unsupported cache kind: {kind!r}")
 
-    @staticmethod
+    @classmethod
     def _get_chunk_windows(
+        cls,
         total_latent_frames: int,
         chunk_size: int,
         window_size: int,
         global_sink_chunk: bool,
     ) -> List[Dict[str, Any]]:
+        num_chunks = (total_latent_frames + chunk_size - 1) // chunk_size
+        return [cls._get_chunk_window(i, total_latent_frames, chunk_size, window_size, global_sink_chunk)
+                for i in range(num_chunks)]
+
+    @staticmethod
+    def _get_chunk_window(
+        chunk_idx: int,
+        total_latent_frames: int,
+        chunk_size: int,
+        window_size: int,
+        global_sink_chunk: bool,
+    ) -> Dict[str, Any]:
         if window_size <= 0:
             raise ValueError(f"`window_size` must be positive, got {window_size}.")
-
-        windows = []
-        num_chunks = (total_latent_frames + chunk_size - 1) // chunk_size
-        for chunk_idx in range(num_chunks):
-            chunk_start = chunk_idx * chunk_size
-            chunk_end = min(total_latent_frames, chunk_start + chunk_size)
-            if global_sink_chunk and chunk_idx > 0:
-                tail_window_size = max(window_size - 1, 1)
-                tail_chunk_start = max(1, chunk_idx - tail_window_size + 1)
-                selected_chunk_ids = [0] + list(range(tail_chunk_start, chunk_idx + 1))
-            else:
-                window_chunk_start = max(0, chunk_idx - window_size + 1)
-                selected_chunk_ids = list(range(window_chunk_start, chunk_idx + 1))
-
-            windows.append(
-                {
-                    "chunk_idx": chunk_idx,
-                    "chunk_start": chunk_start,
-                    "chunk_end": chunk_end,
-                    "selected_chunk_ids": selected_chunk_ids,
-                }
-            )
-        return windows
+        if global_sink_chunk and chunk_idx > 0:
+            tail_start = max(1, chunk_idx - max(window_size - 1, 1) + 1)
+            selected_chunk_ids = [0] + list(range(tail_start, chunk_idx + 1))
+        else:
+            selected_chunk_ids = list(range(max(0, chunk_idx - window_size + 1), chunk_idx + 1))
+        return {
+            "chunk_idx": chunk_idx,
+            "chunk_start": chunk_idx * chunk_size,
+            "chunk_end": min(total_latent_frames, (chunk_idx + 1) * chunk_size),
+            "selected_chunk_ids": selected_chunk_ids,
+        }
 
     @staticmethod
     def _chunk_frame_bounds(chunk_id: int, chunk_size: int, total_latent_frames: int) -> Tuple[int, int]:

xvideo/serving/joyomni_streaming.py

@@ -352,15 +352,11 @@ class JoyOmniRuntime:
             postprocess_device=postprocess_device_obj,
         )
 
-        try:
-            if os.environ.get("JOYOMNI_SKIP_LOAD_WARMUP", "0").lower() in {"1", "true", "yes", "on"}:
-                print("#####[STREAM] load-time warmup skipped (JOYOMNI_SKIP_LOAD_WARMUP); "
-                      "warm up later on a real GPU")
-            else:
-                for (_wh, _ww) in _orientations:
-                    runtime.warmup_full_pipeline(height=_wh, width=_ww)
-        except Exception as _wexc:
-            print(f"#####[STREAM] full-pipeline warmup error (non-fatal): {_wexc!r}")
+        if os.environ.get("JOYOMNI_SKIP_LOAD_WARMUP", "0").lower() in {"1", "true", "yes", "on"}:
+            print("#####[STREAM] load-time warmup skipped (JOYOMNI_SKIP_LOAD_WARMUP)")
+        else:
+            for (_wh, _ww) in _orientations:
+                runtime.warmup_full_pipeline(height=_wh, width=_ww)
 
         if device_obj.type == "cuda":
             _free_b, _total_b = torch.cuda.mem_get_info(device_obj)
@@ -388,20 +384,6 @@ class JoyOmniRuntime:
             ref_image=ref_image,
         )
 
-    def prebake_graph(
-        self,
-        prompt: str,
-        *,
-        settings: StreamingSettings,
-        ref_image: Image.Image | None = None,
-    ) -> None:
-        session = self.create_v2v_session(prompt, settings=settings, ref_image=ref_image)
-        arr = np.zeros((settings.height, settings.width, 3), dtype=np.uint8)
-        try:
-            session._initialize(Image.fromarray(arr, mode="RGB"))
-        finally:
-            session.close()
-
     def warmup_full_pipeline(
         self,
         *,
@@ -445,10 +427,7 @@ class JoyOmniRuntime:
             print(f"#####[STREAM] full-pipeline warmup skipped/failed: {exc!r}")
         finally:
             if session is not None:
-                try:
-                    session.close()
-                except Exception:
-                    pass
+                session.close()
 
 class JoyOmniV2VStreamingSession:
     _session_serial_counter = 0
@@ -528,6 +507,7 @@ class JoyOmniV2VStreamingSession:
         self._result_queue: queue.Queue[StreamingChunkResult] | None = None
         self._workers: list[threading.Thread] = []
         self._worker_error: BaseException | None = None
+        self._stop_event = threading.Event()
         self._worker_error_lock = threading.Lock()
         self._pseudo_latent_condition = threading.Condition()
         self._pseudo_latent_chunk_idx: int | None = None
@@ -599,6 +579,8 @@ class JoyOmniV2VStreamingSession:
         self,
         frame: Image.Image,
         frame_meta: dict[str, Any] | None = None,
+        *,
+        drain_results: bool = True,
     ) -> list[StreamingChunkResult]:
         self._raise_worker_error_if_needed()
         frame = self._resize_frame(frame)
@@ -607,20 +589,20 @@ class JoyOmniV2VStreamingSession:
             self._initialize(frame)
         self.pending_frames.append(frame)
         self.pending_metas.append(meta)
-        results: list[StreamingChunkResult] = []
         while len(self.pending_frames) >= self.frames_per_next_chunk:
             n = self.frames_per_next_chunk
             chunk_frames = self.pending_frames[:n]
             chunk_metas = self.pending_metas[:n]
             del self.pending_frames[:n]
             del self.pending_metas[:n]
-            results.extend(self._submit_or_process_chunk(chunk_frames, chunk_metas))
-        results.extend(self._drain_async_results())
+            self._submit_async_chunk(chunk_frames, chunk_metas)
+        results = self._drain_async_results() if drain_results else []
         self._raise_worker_error_if_needed()
         return results
 
     @torch.no_grad()
     def flush_pending(self) -> None:
+        self._raise_worker_error_if_needed()
         if not self.initialized or not self.pending_frames:
             return
         valid = len(self.pending_frames)
@@ -629,18 +611,25 @@ class JoyOmniV2VStreamingSession:
         chunk_metas = self.pending_metas + [self.pending_metas[-1]] * pad
         self.pending_frames = []
         self.pending_metas = []
-        self._submit_or_process_chunk(chunk_frames, chunk_metas, valid_count=valid)
+        self._submit_async_chunk(chunk_frames, chunk_metas, valid_count=valid)
         self._raise_worker_error_if_needed()
 
-    def close(self) -> None:
+    def close(self, timeout: float = 5.0) -> None:
+        self._stop_async_workers(timeout)
+        if torch.cuda.is_available():
+            devices = {str(self.device), str(self.postprocess_device)}
+            devices.update(str(_module_device(vae)) for vae in (
+                self.pipeline.vae, self.decode_vae, self.pseudo_encode_vae,
+            ) if vae is not None)
+            for device in devices:
+                if torch.device(device).type == "cuda":
+                    torch.cuda.synchronize(device)
         self._clear_vae_feature_caches()
         self.pipeline.transformer.reset_inference_kv_cache()
-        self._stop_async_workers()
-        # No empty_cache: keep the freed KV/graph blocks cached in the allocator
-        # so the next session's pools reuse them instead of fresh cudaMallocs.
 
     @torch.no_grad()
     def _initialize(self, first_frame: Image.Image) -> None:
+        self._raise_worker_error_if_needed()
         if self.initialized:
             return
 
@@ -732,8 +721,10 @@ class JoyOmniV2VStreamingSession:
         return np.asarray(frame, dtype=np.float32)
 
     def _update_static_anchor(self, chunk_idx: int, source_frames: list[Image.Image]) -> int | None:
+        if not self.settings.freeze_kv_on_static:
+            return None
         gray = self._chunk_last_frame_gray(source_frames)
-        if not self.settings.freeze_kv_on_static or chunk_idx == 0 or gray is None:
+        if chunk_idx == 0 or gray is None:
             self._prev_static_gray = gray if gray is not None else self._prev_static_gray
             self._static_anchor_id = None
             return None
@@ -764,15 +755,6 @@ class JoyOmniV2VStreamingSession:
         self._prev_static_gray = gray
         return anchor_id
 
-    def _submit_or_process_chunk(
-        self,
-        source_frames: list[Image.Image],
-        source_metas: list[dict[str, Any]],
-        valid_count: int | None = None,
-    ) -> list[StreamingChunkResult]:
-        self._submit_async_chunk(source_frames, source_metas, valid_count)
-        return []
-
     def _new_profile(self, chunk_idx: int, input_frames: int) -> dict[str, Any]:
         return {
             "chunk_idx": chunk_idx,
@@ -973,13 +955,13 @@ class JoyOmniV2VStreamingSession:
         )
 
         total_latent_frames = chunk_idx + 1
-        chunk_windows = self.pipeline._get_chunk_windows(
+        chunk_window = self.pipeline._get_chunk_window(
+            chunk_idx=chunk_idx,
             total_latent_frames=total_latent_frames,
             chunk_size=self.chunk_size,
             window_size=self.local_window_size,
             global_sink_chunk=self.global_sink_chunk,
         )
-        chunk_window = chunk_windows[-1]
         selected_chunk_ids = chunk_window["selected_chunk_ids"]
         history_chunk_ids = selected_chunk_ids[:-1]
         active_chunk_id = selected_chunk_ids[-1]
@@ -1233,16 +1215,35 @@ class JoyOmniV2VStreamingSession:
         for worker in self._workers:
             worker.start()
 
-    def _stop_async_workers(self) -> None:
-        if self._encode_queue is None:
-            return
-        try:
-            self._encode_queue.put(None, timeout=0.1)
-        except queue.Full:
-            if self._worker_error is None:
-                self._encode_queue.put(None)
+    def request_stop(self) -> None:
+        self._stop_event.set()
+        with self._pseudo_latent_condition:
+            self._pseudo_latent_condition.notify_all()
+
+    def _queue_get(self, q):
+        while not self._stop_event.is_set():
+            try:
+                return q.get(timeout=0.05)
+            except queue.Empty:
+                pass
+        return None
+
+    def _queue_put(self, q, item) -> None:
+        while not self._stop_event.is_set():
+            try:
+                q.put(item, timeout=0.05)
+                return
+            except queue.Full:
+                pass
+
+    def _stop_async_workers(self, timeout: float = 5.0) -> None:
+        self.request_stop()
+        deadline = time.monotonic() + timeout
         for worker in self._workers:
-            worker.join()
+            worker.join(timeout=max(0.0, deadline - time.monotonic()))
+        alive = [worker.name for worker in self._workers if worker.is_alive()]
+        if alive:
+            raise TimeoutError(f"streaming workers did not stop: {', '.join(alive)}")
         self._encode_queue = None
         self._dit_queue = None
         self._decode_queue = None
@@ -1252,21 +1253,23 @@ class JoyOmniV2VStreamingSession:
         self._workers = []
 
     def _set_worker_error(self, exc: BaseException) -> None:
+        if self._stop_event.is_set():
+            return
         import traceback as _tb
         print(f"#####[STREAM-WORKER-ERROR] {exc!r}", flush=True)
         _tb.print_exc()
         with self._worker_error_lock:
             if self._worker_error is None:
                 self._worker_error = exc
-        with self._pseudo_latent_condition:
-            self._pseudo_latent_condition.notify_all()
+        self.request_stop()
 
     def _raise_worker_error_if_needed(self) -> None:
         with self._worker_error_lock:
             exc = self._worker_error
-            self._worker_error = None
         if exc is not None:
             raise RuntimeError(f"async streaming worker failed: {exc!r}") from exc
+        if self._stop_event.is_set():
+            raise RuntimeError("streaming session stopped")
 
     @staticmethod
     def _queue_size(q: queue.Queue | None) -> int | None:
@@ -1351,6 +1354,7 @@ class JoyOmniV2VStreamingSession:
         source_metas: list[dict[str, Any]],
         valid_count: int | None = None,
     ) -> None:
+        self._raise_worker_error_if_needed()
         if self._encode_queue is None:
             self._start_async_workers()
         assert self._encode_queue is not None
@@ -1368,13 +1372,8 @@ class JoyOmniV2VStreamingSession:
         )
         self._set_debug_state("submit", "put_encode", chunk_idx)
 
-        while True:
-            self._raise_worker_error_if_needed()
-            try:
-                self._encode_queue.put(job, timeout=1.0)
-                break
-            except queue.Full:
-                continue
+        self._queue_put(self._encode_queue, job)
+        self._raise_worker_error_if_needed()
         self._inc_debug_counter("submitted_chunks")
         self.chunk_idx += 1
 
@@ -1416,10 +1415,9 @@ class JoyOmniV2VStreamingSession:
         while True:
             self._set_debug_state("vae-encode", "wait_encode_queue")
             _q_started = time.perf_counter()
-            job = self._encode_queue.get()
+            job = self._queue_get(self._encode_queue)
             if job is None:
                 self._set_debug_state("vae-encode", "stop")
-                self._dit_queue.put(None)
                 return
             self._record_queue_wait(job.profile, "q_wait_encode_s", _q_started)
             try:
@@ -1438,12 +1436,11 @@ class JoyOmniV2VStreamingSession:
                     ready = _record_ready_event()
                 self._timer_record(job.profile, "reference_prepare_s", started)
                 self._set_debug_state("vae-encode", "put_dit_queue", job.chunk_idx)
-                self._dit_queue.put(_EncodedChunk(job=job, ref_chunk_latent=ref_chunk_latent, ready_event=ready))
+                self._queue_put(self._dit_queue, _EncodedChunk(job=job, ref_chunk_latent=ref_chunk_latent, ready_event=ready))
                 self._inc_debug_counter("encoded_chunks")
             except BaseException as exc:
                 self._set_debug_state("vae-encode", "error", job.chunk_idx)
                 self._set_worker_error(exc)
-                self._dit_queue.put(None)
                 return
 
     def _dit_worker(self) -> None:
@@ -1452,10 +1449,9 @@ class JoyOmniV2VStreamingSession:
         while True:
             self._set_debug_state("dit-denoise", "wait_dit_queue")
             _q_started = time.perf_counter()
-            encoded = self._dit_queue.get()
+            encoded = self._queue_get(self._dit_queue)
             if encoded is None:
                 self._set_debug_state("dit-denoise", "stop")
-                self._decode_queue.put(None)
                 return
             self._record_queue_wait(encoded.job.profile, "q_wait_dit_s", _q_started)
             try:
@@ -1479,14 +1475,13 @@ class JoyOmniV2VStreamingSession:
                         )
                         _dit_ready = _record_ready_event()
                 self._set_debug_state("dit-denoise", "put_decode_queue", encoded.job.chunk_idx)
-                self._decode_queue.put(
+                self._queue_put(self._decode_queue,
                     _DenoisedChunk(job=encoded.job, current_chunk_latents=current_chunk_latents, ready_event=_dit_ready)
                 )
                 self._inc_debug_counter("denoised_chunks")
             except BaseException as exc:
                 self._set_debug_state("dit-denoise", "error", encoded.job.chunk_idx)
                 self._set_worker_error(exc)
-                self._decode_queue.put(None)
                 return
 
     def _decode_worker(self) -> None:
@@ -1496,11 +1491,9 @@ class JoyOmniV2VStreamingSession:
         while True:
             self._set_debug_state("vae-decode", "wait_decode_queue")
             _q_started = time.perf_counter()
-            denoised = self._decode_queue.get()
+            denoised = self._queue_get(self._decode_queue)
             if denoised is None:
                 self._set_debug_state("vae-decode", "stop")
-                self._pseudo_queue.put(None)
-                self._postprocess_queue.put(None)
                 return
             self._record_queue_wait(denoised.job.profile, "q_wait_decode_s", _q_started)
             try:
@@ -1519,19 +1512,17 @@ class JoyOmniV2VStreamingSession:
                     _dec_ready = _record_ready_event()
 
                 self._set_debug_state("vae-decode", "put_pseudo_queue", denoised.job.chunk_idx)
-                self._pseudo_queue.put(
+                self._queue_put(self._pseudo_queue,
                     _DecodedPixelsChunk(job=denoised.job, decoded_pixels=decoded_pixels, ready_event=_dec_ready)
                 )
                 self._set_debug_state("vae-decode", "put_postprocess_queue", denoised.job.chunk_idx)
-                self._postprocess_queue.put(
+                self._queue_put(self._postprocess_queue,
                     _DecodedPixelsChunk(job=denoised.job, decoded_pixels=decoded_pixels, ready_event=_dec_ready)
                 )
                 self._inc_debug_counter("decoded_pixel_chunks")
             except BaseException as exc:
                 self._set_debug_state("vae-decode", "error", denoised.job.chunk_idx)
                 self._set_worker_error(exc)
-                self._pseudo_queue.put(None)
-                self._postprocess_queue.put(None)
                 return
 
     def _pseudo_worker(self) -> None:
@@ -1539,7 +1530,7 @@ class JoyOmniV2VStreamingSession:
         while True:
             self._set_debug_state("pseudo-encode", "wait_pseudo_queue")
             _q_started = time.perf_counter()
-            decoded = self._pseudo_queue.get()
+            decoded = self._queue_get(self._pseudo_queue)
             if decoded is None:
                 self._set_debug_state("pseudo-encode", "stop")
                 return
@@ -1570,7 +1561,7 @@ class JoyOmniV2VStreamingSession:
         while True:
             self._set_debug_state("postprocess", "wait_postprocess_queue")
             _q_started = time.perf_counter()
-            decoded = self._postprocess_queue.get()
+            decoded = self._queue_get(self._postprocess_queue)
             if decoded is None:
                 self._set_debug_state("postprocess", "stop")
                 return
@@ -1600,7 +1591,7 @@ class JoyOmniV2VStreamingSession:
                     n_out,
                 )
                 self._set_debug_state("postprocess", "put_result_queue", decoded.job.chunk_idx)
-                self._result_queue.put(
+                self._queue_put(self._result_queue,
                     StreamingChunkResult(
                         jpegs=packed,
                         profile=decoded.job.profile,
@@ -1830,14 +1821,15 @@ class JoyOmniV2VStreamingSession:
 
     def _next_selected_chunk_ids(self, chunk_idx: int | None = None) -> list[int]:
         chunk_idx = self.chunk_idx if chunk_idx is None else chunk_idx
-        next_total = chunk_idx + 2
-        windows = self.pipeline._get_chunk_windows(
+        next_total = (chunk_idx + 2) * self.chunk_size
+        window = self.pipeline._get_chunk_window(
+            chunk_idx=chunk_idx + 1,
             total_latent_frames=next_total,
             chunk_size=self.chunk_size,
             window_size=self.local_window_size,
             global_sink_chunk=self.global_sink_chunk,
         )
-        return list(windows[-1]["selected_chunk_ids"])
+        return window["selected_chunk_ids"]
 
     def _resize_frame(self, frame: Image.Image) -> Image.Image:
         frame = frame.convert("RGB")

xvideo/serving/pe.py

@@ -1,14 +1,13 @@
+import asyncio
 import base64
-import json
 import logging
 import os
 import re
-import time
-import urllib.request
 from io import BytesIO
 from typing import List, Optional
 
 from PIL import Image
+import httpx
 
 logger = logging.getLogger("joyomni.pe")
 
@@ -56,10 +55,10 @@ V2V_TEMPLATE = """# INPUT DATA
    background/scene replacement, no art-style or medium change (anime / painting / cartoon look and
    similar), no relighting or color grading beyond what a requested edit needs for integration, no
    camera motion, speed changes, depth-of-field effects, or extra elements on your own initiative.
-   Unless the Target Objective names an art style, the edited video stays photorealistic
-   live-action — never add a painting / ink / anime / cartoon style because the scene's culture or
-   era suggests one. Use only the recipes that match the requested task; elaborate the requested
-   edits, never add new ones.
+   Preserve the source video's existing visual style and medium unless the Target Objective
+   explicitly requests changing them. Do not introduce a new style merely because the scene's
+   culture or era suggests one. Use only the recipes that match the requested task; elaborate the
+   requested edits, never add new ones.
 2. Priority: Describe the edits in the Target Objective's order of importance — the primary subject
    edit comes FIRST and carries the most detail (for a well-known character or person, spell out
    their canonical visual features); secondary edits such as a background swap get one concise
@@ -98,10 +97,17 @@ FOUNDATION of your prompt, and seamlessly expand them into a highly detailed, co
 - Background Replacement: "Replace the original background with [highly detailed description of the
   new environment], ensuring the foreground elements are seamlessly integrated with matching global
   illumination, reflections, and realistic cast shadows."
-  Whenever this recipe is used, append this clause verbatim right after it: "the area directly
-  behind the subject's head and shoulders shows only the new environment — the original chair and
-  its headrest are gone." In an output that does not replace the background, that clause and any
-  mention of removing the chair, headrest, or other scene objects are FORBIDDEN.
+  Inspect the original background directly behind the subject, especially visible chairs, chair
+  backs and headrests closely bordering the subject's visible outline. These objects are part of
+  the background even when they touch the subject's outline. When present within the requested
+  background replacement, their removal is MANDATORY unless the Target Objective explicitly asks
+  to keep them. Explicitly name each visible object or component in this region and state that it
+  is completely removed, so the exposed area directly behind the subject shows the new environment;
+  a generic "replace the background" instruction is not enough. Name only components actually
+  visible in the source: a visible chair back does not imply a visible headrest. When none is
+  visible, do not mention these objects at all, even in a removal or preservation clause. Make this
+  decision from the image before writing; never output "if present" or "if visible" conditions.
+  Do not add these removals to unrelated edits unless the Target Objective requests them.
 - Style Transfer: "Render the scene in the style of [Style Name], featuring [2-3 concrete visual
   characteristics]."
 - Whole-frame Style Coverage: When the objective converts the entire video to an art style, the
@@ -116,9 +122,11 @@ FOUNDATION of your prompt, and seamlessly expand them into a highly detailed, co
   cyberpunk) is paired with "keep everything unchanged", realize the style as bold, clearly visible
   lighting and color grading (e.g. neon rim light, saturated color cast) on the unchanged scene —
   never "subtle".
-- Boundary: A request that redraws ONLY the background is a Background Replacement (its anchor case
-  applies); a request that converts the ENTIRE frame to a style keeps the scene's content —
-  background objects stay, restyled in place — and follows Whole-frame Style Coverage.
+- Boundary: Replacing the background with a different environment is Background Replacement.
+  Restyling the existing background preserves its scene content and layout unless the Target
+  Objective requests content changes; changing its appearance alone does not trigger the
+  background furniture-removal rule. A whole-frame style conversion restyles both the subject
+  and background in place and follows Whole-frame Style Coverage.
 - Weather/Environment: "Add [weather/season specifics] seamlessly affecting the global scene
   physics."
 - Lighting & Color Grading: "Apply cinematic relighting and color grading: [detailed description of
@@ -138,12 +146,24 @@ FOUNDATION of your prompt, and seamlessly expand them into a highly detailed, co
 Anchor ONLY what the Target Objective leaves untouched — an anchor must never contradict the
 requested edit, and preservation statements must never conflict with the Target Objective. Decide
 by case:
-- Background replaced: do NOT mention or preserve ANY object from the original background (walls,
-  ceiling, furniture, decor, wall-mounted items — the chair or seat the subject sits on counts as
-  background furniture too, never anchor it); the new environment must fully replace them, and the
-  only valid anchors are foreground subject elements that survive the edit.
-- Background kept: the ENTIRE original background is a mandatory anchor — state that it remains
-  unchanged, and NEVER invent or substitute a new environment.
+- Background replaced: do not preserve original background objects within the replacement scope
+  unless the Target Objective explicitly keeps them. Original objects directly behind the subject,
+  including visible chairs, chair backs and headrests along the subject's outline, remain part of
+  the background; explicitly remove them under the Background Replacement rule. The new
+  environment must fill all replaced regions, including visible gaps around the subject and frame
+  edges. Anchor only surviving foreground elements and objects the user explicitly keeps, and
+  describe only source objects actually visible in the provided frame.
+- Background untouched: when no requested edit affects the background, preserve its scene content,
+  spatial layout and visual appearance. State that the background remains unchanged; do not invent
+  or substitute a new environment.
+- Background appearance edited without scene replacement: for requested full-frame or
+  background-only stylization, weather/season changes, lighting, color grading or depth-of-field
+  changes, preserve the original scene's objects and spatial layout except for requested content
+  changes. Apply the requested visual changes to all affected background regions. Do not call the
+  background "unchanged" or preserve its original rendering, lighting, colors or focus when those
+  attributes are being edited. For whole-frame art-style conversion, use the preservation wording
+  in Whole-frame Style Coverage; otherwise state which scene content/layout stays and which visual
+  attributes change. Apply effects only within the requested scope; respect explicit exclusions.
 - Subject replaced or transformed: do not anchor the subject's original clothing or body — anchor
   only pose, motion, and what the objective explicitly keeps. When the subject turns into a
   different material or character, express likeness as part of the transformation ("an ice
@@ -156,6 +176,76 @@ by case:
   Objective names them.
 """
 
+RV2V_SYSTEM_PROMPT = """You write precise English instructions for reference-image-guided video
+editing (RV2V). Understand the source video and the separate reference image, then describe only
+the requested transfer. The source video determines the subject, pose, motion, framing and scene;
+the reference supplies only the visual attributes requested by the user."""
+
+RV2V_TEMPLATE = """# INPUT
+User request: {user_prompt}
+Image 0 is the REFERENCE IMAGE: the donor of the requested appearance, garment, object or scene.
+Any subsequent images are SOURCE VIDEO FRAMES: the video to edit.
+
+# OUTPUT
+Return ONLY one cohesive English editing prompt, about 80-140 words. No explanation or labels.
+Start with a direct edit instruction, naming the target in the video and the requested item from
+the reference image. Refer to it as "the reference image", never by image number in the output.
+Inspect the reference and describe 3-5 distinctive visible features relevant to the request:
+color, shape, material, pattern and construction. Be concrete and accurate; do not invent details.
+Keep the entire request, including any explicit exceptions, extra edits and exact quoted text.
+
+# TRANSFER SCOPE
+Explicit requested edits and exceptions override the default preservation rules below. Determine
+all requested changes first; preserve only attributes outside that combined edit scope.
+- Clothes: identify the visible source clothing layers and the requested replacement scope before
+  choosing what to preserve. A request explicitly targeting an inner top, outer garment or trousers
+  changes only that layer; a complete outfit transfers the visible outfit. For an unqualified
+  request to replace the top/upper-body clothing, make the reference top the visible upper-body
+  garment: replace existing layers that would cover or conceal it, including outerwear or
+  shoulder-draped coverings, unless the user explicitly keeps them. Do not classify these clothing
+  layers as accessories to preserve. Top-only edits keep lower-body clothing, and trousers-only
+  edits keep upper-body clothing. Name the source garments/layers being replaced in the final
+  prompt, then describe the reference garment's cut, silhouette, length, color and visible details.
+  Preserve the source person's identity, face, hair, body proportions and pose within the unchanged
+  scope.
+  The reference model's face, body, pose and background are not part of a clothing transfer.
+  Fit the garment naturally to the source body, with correct overlap and contact at hands/arms.
+- Background: replace the original environment with the environment in the reference, describing
+  its main visible structures. Keep the source foreground person and their clothing. Remove old
+  background furniture only when it is actually visible. Do not add people from the reference.
+- Person/character replacement: describe the reference face/head AND the requested body/outfit.
+  Preserve source pose and motion, without preserving identity attributes that must change.
+- Object/accessory: transfer only the requested object, at its functional position and scale.
+- Style: apply only the requested visual style from the reference to the specified content.
+
+Close with a concise preservation clause for what stays unchanged. Keep the original background
+unless the user requests changing it. Do not copy reference framing, pose, lighting, typography,
+logos, watermarks, props or other garments unless requested. Do not describe invisible body parts
+or force the camera to reveal the whole garment. Avoid vague quality words and excessive negative
+instructions. The edit is already fully present in the first output frame and follows source
+motion consistently; do not describe a gradual transformation.
+
+# SOURCE FRAMING — CLOTHING EDITS
+First identify the source person's actual visible crop and pose. In a seated chest-up or
+head-and-shoulders view, begin the final editing prompt by anchoring the edit to that existing
+close-up and seated pose. Describe only the reference garment surfaces that can appear inside
+this source crop. For a dress, name the visible upper part of the dress and describe its neckline,
+shoulder fabric, sleeves and chest texture. For an outfit, describe only its visible jacket/top
+and inner layer. Completely omit descriptions of offscreen garments, waist/hip shaping, skirt
+length, trouser legs, hemlines, feet and full-body silhouettes from the final prompt, including
+negative mentions of those details. This is a visibility rule, not a change in the user's outfit
+choice. Full-body source shots still receive the full requested outfit. At each output frame,
+match the person's head, shoulder and arm positions, pose, subject scale and framing to the
+corresponding source-video frame; let the clothing follow the source motion with natural
+deformation. Do not freeze the person in the provided frame's pose or suppress source movement.
+Preserve all visible source accessories that fall outside the requested edit scope. Explicitly
+apply any requested accessory removal, replacement or modification; never also describe that
+accessory as unchanged. Retained accessories follow the source body's motion
+and placement, with the new fabric layered naturally around them. State the actual visible crop
+directly; do not output an "if visible" or "if cropped" condition. Never enlarge the visible body
+region to show a garment.
+"""
+
 
 def _downscale(image: Image.Image, max_side: int = PE_IMAGE_MAX_SIDE) -> Image.Image:
     if max_side and max_side > 0:
@@ -220,10 +310,17 @@ def _message_content_to_text(content) -> str:
 def _sanitize_enhanced(text: str, fallback: str) -> str:
     if not text:
         return fallback
-    cleaned = text
-    cleaned = re.sub(r"!\[[^\]]*\]\([^)]*\)", "", cleaned)
-    cleaned = re.sub(r"\[([^\]]*)\]\([^)]*\)", r"\1", cleaned)
-    cleaned = re.sub(r"https?://\S+", "", cleaned)
+    def clean_link(match):
+        if match.group() in fallback or match.group(2) in fallback:
+            return match.group()
+        return "" if match.group().startswith("!") else match.group(1)
+
+    def clean_url(match):
+        url = match.group().rstrip(".,;:!?，。；：！？)")
+        return match.group() if url in fallback else match.group()[len(url):]
+
+    cleaned = re.sub(r"!?\[([^\]]*)\]\(([^)]*)\)", clean_link, text)
+    cleaned = re.sub(r'https?://[^\s<>"\'\[\]（）“”]+', clean_url, cleaned)
     cleaned = re.sub(r"[ \t]+\n", "\n", cleaned)
     cleaned = re.sub(r"\n{3,}", "\n\n", cleaned).strip()
     if len(cleaned) < 10:
@@ -258,54 +355,65 @@ class PromptEnhancer:
         self.base_url = base_url or os.environ.get("OPENAI_BASE_URL") or DEFAULT_BASE_URL
         self.model = model or os.environ.get("PE_MODEL") or DEFAULT_MODEL
         self.anthropic = "/anthropic" in self.base_url
-        if not self.anthropic:
-            from openai import OpenAI
+        self.max_retries = max(1, max_retries)
 
-            self.client = OpenAI(api_key=self.api_key, base_url=self.base_url)
-        self.max_retries = max_retries
-
-    def _anthropic_complete(self, system_prompt, user_text, images_b64) -> str:
+    def _request(self, system_prompt, user_text, images_b64):
+        headers = {"Authorization": f"Bearer {self.api_key}"}
+        if not self.anthropic:
+            return f"{self.base_url.rstrip('/')}/chat/completions", headers, {
+                "model": self.model, "max_completion_tokens": 8192,
+                "messages": _build_messages(system_prompt, user_text, images_b64),
+            }
         content = [{"type": "text", "text": user_text}]
         for i, b64 in enumerate(images_b64):
             content.append({"type": "text", "text": f"\n[Image {i}]:"})
             content.append({"type": "image", "source": {
                 "type": "base64", "media_type": "image/png", "data": b64}})
-        body = json.dumps({
+        body = {
             "model": self.model, "max_tokens": 4096, "system": system_prompt,
             "messages": [{"role": "user", "content": content}],
-        }).encode()
-        req = urllib.request.Request(
-            f"{self.base_url.rstrip('/')}/v1/messages", data=body,
-            headers={"Authorization": f"Bearer {self.api_key}",
-                     "anthropic-version": "2023-06-01",
-                     "Content-Type": "application/json"})
-        resp = json.load(urllib.request.urlopen(req, timeout=90))
-        return "".join(b.get("text", "") for b in resp.get("content", [])
-                       if b.get("type") == "text")
-
-    def _chat(self, system_prompt, user_text, images_b64, raw_fallback="") -> Optional[str]:
-        messages = None if self.anthropic else _build_messages(system_prompt, user_text, images_b64)
-        last_err = None
-        for attempt in range(1, self.max_retries + 1):
-            try:
-                if self.anthropic:
-                    text = self._anthropic_complete(system_prompt, user_text, images_b64)
-                else:
-                    resp = self.client.chat.completions.create(
-                        model=self.model, messages=messages, max_completion_tokens=8192
+        }
+        headers["anthropic-version"] = "2023-06-01"
+        return f"{self.base_url.rstrip('/')}/v1/messages", headers, body
+
+    async def _chat(self, system_prompt, user_text, images_b64, raw_fallback):
+        url, headers, body = self._request(system_prompt, user_text, images_b64)
+        async with httpx.AsyncClient(timeout=90.0) as client:
+            for attempt in range(1, self.max_retries + 1):
+                try:
+                    response = await client.post(url, headers=headers, json=body)
+                    response.raise_for_status()
+                    data = response.json()
+                    content = data["content"] if self.anthropic else data["choices"][0]["message"]["content"]
+                    return _sanitize_enhanced(_message_content_to_text(content).strip(), raw_fallback)
+                except (httpx.TransportError, httpx.HTTPStatusError) as exc:
+                    retryable = not isinstance(exc, httpx.HTTPStatusError) or (
+                        exc.response.status_code in (408, 409, 429) or exc.response.status_code >= 500
                     )
-                    text = _message_content_to_text(resp.choices[0].message.content)
-                return _sanitize_enhanced(text.strip(), raw_fallback or text.strip())
-            except Exception as e:  # noqa: BLE001
-                last_err = e
-                logger.warning("PE attempt %d/%d failed: %s", attempt, self.max_retries, e)
-                time.sleep(min(attempt, 5))
-        logger.error("PE failed after %d attempts: %s", self.max_retries, last_err)
-        return None
+                    if not retryable or attempt == self.max_retries:
+                        raise
+                    logger.warning("PE attempt %d/%d failed: %s", attempt, self.max_retries, type(exc).__name__)
+                    await asyncio.sleep(min(attempt, 5))
 
-    def __call__(self, task_type, user_prompt, video=None) -> Optional[str]:
+    async def enhance(self, task_type, user_prompt, video=None, ref_image=None, *, timeout=90.0):
         if not user_prompt or not user_prompt.strip():
             return user_prompt
         video_frames = _video_frames_to_b64(video)
-        text = V2V_TEMPLATE.format(user_prompt=user_prompt)
-        return self._chat(SYSTEM_PROMPT, text, video_frames, raw_fallback=user_prompt) or user_prompt
+        if task_type == "rv2v":
+            reference = _img_to_b64(ref_image)
+            if reference is None:
+                logger.warning("RV2V PE needs a reference image; using raw prompt")
+                return user_prompt
+            text = RV2V_TEMPLATE.format(user_prompt=user_prompt)
+            system, images = RV2V_SYSTEM_PROMPT, [reference, *video_frames]
+        else:
+            text = V2V_TEMPLATE.format(user_prompt=user_prompt)
+            system, images = SYSTEM_PROMPT, video_frames
+        return await asyncio.wait_for(self._chat(system, text, images, user_prompt), timeout=timeout)
+
+    def __call__(self, task_type, user_prompt, video=None, ref_image=None, *, timeout=90.0) -> Optional[str]:
+        try:
+            return asyncio.run(self.enhance(task_type, user_prompt, video, ref_image, timeout=timeout))
+        except Exception as exc:
+            logger.warning("PE failed: %s; using raw prompt", type(exc).__name__)
+            return user_prompt

xvideo/serving/serve_joyomni_streaming.py

@@ -13,6 +13,7 @@ import sys
 import tempfile
 import threading
 import time
+import uuid
 from contextlib import asynccontextmanager
 from fractions import Fraction
 from pathlib import Path
@@ -371,14 +372,16 @@ def _person_present(image: Image.Image, *, onnx_path: str, conf: float) -> bool:
     out = net.forward()
     return float(out[0, 4, :].max()) >= float(conf)
 
-def _enhance_prompt_sync(
+async def _enhance_prompt(
     *,
     raw_prompt: str,
     pe_frame: Image.Image | None,
     pe_model: str | None,
+    timeout: float,
+    ref_image: Image.Image | None = None,
 ) -> dict[str, Any]:
     started = time.time()
-    task_type = "v2v"
+    task_type = "rv2v" if ref_image is not None else "v2v"
     enhanced_prompt = raw_prompt
     error = None
     model = pe_model or DEFAULT_PE_MODEL
@@ -387,10 +390,12 @@ def _enhance_prompt_sync(
 
         enhancer = PromptEnhancer(model=pe_model)
         model = enhancer.model or model
-        enhanced = enhancer(
+        enhanced = await enhancer.enhance(
             task_type,
             raw_prompt,
             video=[pe_frame] if pe_frame is not None else None,
+            ref_image=ref_image,
+            timeout=timeout,
         )
         if isinstance(enhanced, str) and enhanced.strip():
             enhanced_prompt = enhanced.strip()
@@ -420,11 +425,14 @@ class _SegmentedRecorder:
         segment_seconds: int,
         queue_max: int = 64,
         lossless: bool = False,
+        reliable: bool = False,
     ) -> None:
         self._prefix = prefix
         self._codec = codec
         self._bitrate = int(bitrate)
         self._lossless = bool(lossless)
+        self._reliable = bool(reliable)
+        self._error: BaseException | None = None
         self._segment_ms = max(1, int(segment_seconds)) * 1000
         self._q: "queue.Queue[Any]" = queue.Queue(maxsize=max(1, queue_max))
         self._stop = threading.Event()
@@ -442,25 +450,37 @@ class _SegmentedRecorder:
         self._started = True
 
     def submit(self, item: Any, t_capture_ms: float) -> None:
-        if not self._started or self._stop.is_set():
-            return
-        try:
-            self._q.put_nowait((item, float(t_capture_ms)))
-        except queue.Full:
-            self.frames_dropped_recording += 1
+        deadline = time.monotonic() + 20.0
+        while True:
+            self._raise_error()
+            if not self._started or self._stop.is_set():
+                raise RuntimeError("recorder is not accepting frames")
+            try:
+                self._q.put((item, float(t_capture_ms)), timeout=0.05 if self._reliable else 0)
+                return
+            except queue.Full:
+                if not self._reliable:
+                    self.frames_dropped_recording += 1
+                    return
+                if time.monotonic() >= deadline:
+                    raise TimeoutError("recording queue remained full")
+
+    def _raise_error(self) -> None:
+        if self._error is not None:
+            raise RuntimeError(f"recording failed: {self._error!r}") from self._error
 
     def stop(self, timeout: float = 5.0) -> None:
         if not self._started:
             return
-        if self._stop.is_set():
-            self._thread.join(timeout=timeout)
-            return
         self._stop.set()
         try:
             self._q.put_nowait(None)
         except queue.Full:
             pass
         self._thread.join(timeout=timeout)
+        if self._thread.is_alive():
+            raise TimeoutError("recorder did not finish encoding")
+        self._raise_error()
 
     def _to_image(self, item: Any) -> "Image.Image | None":
         if isinstance(item, Image.Image):
@@ -513,23 +533,16 @@ class _SegmentedRecorder:
         try:
             for packet in stream.encode():
                 output.mux(packet)
-        except Exception:
-            pass
-        try:
+        finally:
             output.close()
-        except Exception:
-            pass
 
     def _run(self) -> None:
-        try:
-            import av
-        except Exception:
-            return
         output = stream = None
         time_base = Fraction(1, 1000)
         seg_t0 = 0.0
         last_pts = -1
         try:
+            import av
             while True:
                 try:
                     got = self._q.get(timeout=0.2)
@@ -542,7 +555,7 @@ class _SegmentedRecorder:
                 item, t_capture_ms = got
                 image = self._to_image(item)
                 if image is None:
-                    continue
+                    raise ValueError("invalid recording frame")
                 if self._width is None:
                     self._width, self._height = int(image.width), int(image.height)
                 if output is None:
@@ -552,21 +565,24 @@ class _SegmentedRecorder:
                 pts = round(t_capture_ms - seg_t0)
                 if pts <= last_pts:
                     pts = last_pts + 1
-                try:
-                    frame = av.VideoFrame.from_image(image).reformat(format="yuv420p")
-                    frame.pts = pts
-                    frame.time_base = time_base
-                    for packet in stream.encode(frame):
-                        output.mux(packet)
-                    self.frames_written += 1
-                    last_pts = pts
-                except Exception:
-                    pass
+                frame = av.VideoFrame.from_image(image).reformat(format="yuv420p")
+                frame.pts = pts
+                frame.time_base = time_base
+                for packet in stream.encode(frame):
+                    output.mux(packet)
+                self.frames_written += 1
+                last_pts = pts
                 if pts >= self._segment_ms:
-                    self._close_segment(output, stream)
+                    previous_output, previous_stream = output, stream
                     output = stream = None
+                    self._close_segment(previous_output, previous_stream)
+        except BaseException as exc:
+            self._error = exc
         finally:
-            self._close_segment(output, stream)
+            try:
+                self._close_segment(output, stream)
+            except BaseException as exc:
+                self._error = exc
 
 def _optional_positive_int(value: Any, *, name: str) -> int | None:
     if value is None or value == "":
@@ -630,6 +646,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
     app.state.runtime_lock = threading.Lock()
     app.state.inference_lock = threading.Lock()
     app.state.active_session = None
+    app.state.runtime_error = None
     app.state.ws_debug = {}
 
     app.state.session_gate = SessionGate()
@@ -664,7 +681,8 @@ def create_app(args: argparse.Namespace) -> FastAPI:
     def health() -> JSONResponse:
         return JSONResponse(
             {
-                "ok": True,
+                "ok": app.state.runtime_error is None,
+                "error": app.state.runtime_error,
                 "runtime_loaded": app.state.runtime is not None,
                 "dit_ckpt": args.dit_ckpt,
                 "device": str(app.state.runtime.device) if app.state.runtime is not None else args.device,
@@ -783,6 +801,12 @@ def create_app(args: argparse.Namespace) -> FastAPI:
         ticket = None
         frames_in = 0
         frames_out = 0
+        session_id = None
+        edit_frames = 0
+        completed_frames = 0
+        pending_work: set[asyncio.Task] = set()
+        pe_report = None
+        raw_session_prompt = args.prompt
         session_prompt = args.prompt
         session_settings: StreamingSettings | None = None
         ref_image: Image.Image | None = None
@@ -819,7 +843,6 @@ def create_app(args: argparse.Namespace) -> FastAPI:
 
         rec_input: _SegmentedRecorder | None = None
         rec_output: _SegmentedRecorder | None = None
-        rec_seq = 0
         rec_base: Path | None = None
         lossless_mode = False
         ws_debug: dict[str, Any] = {
@@ -842,7 +865,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
 
         async def _ws_send_json(payload: dict[str, Any]) -> None:
             try:
-                await asyncio.wait_for(websocket.send_json(payload), timeout=WS_SEND_TIMEOUT_S)
+                await asyncio.wait_for(websocket.send_json({**payload, "session_id": session_id}), timeout=WS_SEND_TIMEOUT_S)
             except asyncio.TimeoutError:
                 raise WebSocketDisconnect()
 
@@ -869,7 +892,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
             if not encoded_frames:
                 return 0
 
-            if gate_state.get("absent_hold"):
+            if not lossless_mode and gate_state.get("absent_hold"):
                 return 0
             count = len(encoded_frames)
             if not source_metas:
@@ -989,7 +1012,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                     ws_debug["frames_out"] = frames_out
                     _rec_o = rec_output
                     if _rec_o is not None:
-                        _rec_o.submit(encoded, source_meta.get("t_capture_ms"))
+                        await asyncio.to_thread(_rec_o.submit, encoded, source_meta.get("t_capture_ms"))
                         ws_debug["rec_out_written"] = _rec_o.frames_written
                         ws_debug["rec_out_dropped"] = _rec_o.frames_dropped_recording
                     continue
@@ -1067,49 +1090,28 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                 )
             return count
 
-        async def _send_chunk_result(
-            result,
-            *,
-            fallback_meta: dict[str, Any] | None = None,
-            fallback_elapsed: float = 0.0,
-        ) -> int:
-            if not result.jpegs:
-                return 0
+        async def _send_chunk_result(result) -> int:
             jpegs = result.jpegs
-            if result.source_metas:
-                source_metas = result.source_metas
-            elif fallback_meta is not None:
-                source_metas = [fallback_meta] * len(jpegs)
-            else:
-                source_metas = [{} for _ in jpegs]
+            source_metas = result.source_metas or [{} for _ in jpegs]
             if result.valid_count is not None:
                 jpegs = jpegs[: result.valid_count]
                 source_metas = source_metas[: result.valid_count]
-            server_elapsed = float(result.elapsed or fallback_elapsed)
             return await _send_encoded_frames(
-                jpegs, source_metas, result.profile, server_elapsed,
+                jpegs, source_metas, result.profile, float(result.elapsed or 0.0),
             )
 
         async def _output_pump(session_ref) -> None:
+            nonlocal completed_frames
             while not stop_output_pump.is_set():
-                try:
-                    result = await asyncio.to_thread(session_ref.wait_async_result, 0.05)
-                except Exception as exc:
-                    await _send_json({"type": "error", "message": repr(exc)})
-                    break
+                result = await asyncio.to_thread(session_ref.wait_async_result, 0.05)
                 if result is None:
                     continue
-                if result.jpegs:
-                    jpegs = result.jpegs
-                    source_metas = result.source_metas or [{} for _ in jpegs]
-                    if result.valid_count is not None:
-                        jpegs = jpegs[: result.valid_count]
-                        source_metas = source_metas[: result.valid_count]
-                    await _send_encoded_frames(
-                        jpegs, source_metas, result.profile, float(result.elapsed or 0.0)
-                    )
+                await _send_chunk_result(result)
+                completed_frames += min(len(result.jpegs), result.valid_count) if result.valid_count is not None else len(result.jpegs)
 
         def _create_session():
+            if app.state.runtime_error:
+                raise RuntimeError(app.state.runtime_error)
             if session_settings is None:
                 raise RuntimeError("streaming settings are not initialized")
             return runtime.create_v2v_session(
@@ -1118,65 +1120,37 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                 ref_image=ref_image,
             )
 
-        def _session_health_error(session_ref) -> str | None:
-            if session_ref is None:
-                return None
+        async def _run_work(func, *, timeout=None):
+            task = asyncio.create_task(asyncio.to_thread(func))
+            pending_work.add(task)
             try:
-                snapshot = session_ref.debug_snapshot()
-            except Exception:
-                return None
-
-            workers = snapshot.get("workers") or {}
-            errored = []
-            dead = []
-            for name, info in workers.items():
-                state = info.get("state") or {}
-                state_name = state.get("state")
-                if state_name == "error":
-                    errored.append(f"{name}@chunk={state.get('chunk_idx')}")
-                if info.get("alive") is False:
-                    dead.append(f"{name}:{state_name or 'unknown'}")
-            if errored:
-                return "streaming pipeline worker error: " + ", ".join(errored)
-
-            queues = snapshot.get("queues") or {}
-            queue_maxsize = snapshot.get("queue_maxsize") or {}
-            encode_depth = queues.get("encode")
-            encode_max = queue_maxsize.get("encode")
-            encode_full = (
-                isinstance(encode_depth, int) and
-                isinstance(encode_max, int) and
-                encode_max > 0 and
-                encode_depth >= encode_max
-            )
-            if encode_full and dead:
-                return (
-                    f"streaming pipeline stuck: encode queue full "
-                    f"({encode_depth}/{encode_max}) and workers not alive: " +
-                    ", ".join(dead)
-                )
-            return None
-
-        def _close_session_sync(session_ref) -> None:
-            with app.state.inference_lock:
-                session_ref.close()
+                return await asyncio.wait_for(asyncio.shield(task), timeout=timeout)
+            finally:
+                if task.done():
+                    pending_work.discard(task)
 
         async def _close_session_safely(session_ref, reason: str) -> None:
+            if app.state.runtime_error:
+                return
+            session_ref.request_stop()
+            deadline = time.monotonic() + max(0.1, float(args.session_close_timeout_s))
             try:
-                await asyncio.wait_for(
-                    asyncio.to_thread(_close_session_sync, session_ref),
-                    timeout=max(0.1, float(args.session_close_timeout_s)),
-                )
-            except asyncio.TimeoutError:
-                ws_debug["close_timeout"] = reason
-                print(
-                    f"#####[WS-GUARD] session close timed out after "
-                    f"{args.session_close_timeout_s:.1f}s reason={reason}",
-                    flush=True,
-                )
+                if pending_work:
+                    done, waiting = await asyncio.wait(pending_work, timeout=max(0.0, deadline - time.monotonic()))
+                    for task in done:
+                        if not task.cancelled():
+                            task.exception()
+                    pending_work.difference_update(done)
+                    if waiting:
+                        raise TimeoutError("session operations have not stopped")
+                remaining = max(0.01, deadline - time.monotonic())
+                await _run_work(lambda: session_ref.close(timeout=remaining), timeout=remaining)
             except Exception as exc:
                 ws_debug["close_error"] = repr(exc)
-                print(f"#####[WS-GUARD] session close failed reason={reason}: {exc!r}", flush=True)
+                app.state.runtime_error = f"Unsafe runtime after {reason}: {exc!r}; restart required"
+                for task in pending_work:
+                    task.add_done_callback(lambda done: done.exception() if not done.cancelled() else None)
+                raise RuntimeError(app.state.runtime_error) from exc
 
         async def _stop_output_task() -> None:
             nonlocal output_task
@@ -1186,23 +1160,24 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                     await asyncio.wait_for(output_task, timeout=1.0)
                 except asyncio.TimeoutError:
                     output_task.cancel()
+                    await asyncio.gather(output_task, return_exceptions=True)
                 except Exception:
                     pass
             output_task = None
 
         def _start_recorders() -> None:
-            nonlocal rec_input, rec_output, rec_seq, rec_base
+            nonlocal rec_input, rec_output, rec_base
             if args.record_dir is None:
                 return
             _stop_recorders()
             try:
-                rec_seq += 1
-                base = Path(args.record_dir) / f"{int(time.time())}_{rec_seq}"
-                base.mkdir(parents=True, exist_ok=True)
+                base = Path(args.record_dir) / f"{int(time.time())}_{uuid.uuid4().int}"
+                base.mkdir(parents=True)
                 common = dict(
                     codec=str(args.record_codec),
                     bitrate=int(args.record_bitrate),
                     segment_seconds=int(args.record_segment_seconds),
+                    reliable=lossless_mode,
                 )
                 rec_input = _SegmentedRecorder(prefix=base / "input", **common)
                 rec_output = _SegmentedRecorder(prefix=base / "output", lossless=lossless_mode, **common)
@@ -1221,21 +1196,25 @@ def create_app(args: argparse.Namespace) -> FastAPI:
             except Exception as exc:
                 ws_debug["rec_error"] = repr(exc)
                 print(f"#####[REC] start failed: {exc!r} -> recording OFF", flush=True)
-                rec_input = None
-                rec_output = None
-                rec_base = None
+                _stop_recorders()
+                if lossless_mode:
+                    raise
 
         def _stop_recorders() -> None:
             nonlocal rec_input, rec_output, rec_base
+            error = None
             for _rec in (rec_input, rec_output):
                 if _rec is not None:
                     try:
-                        _rec.stop()
+                        _rec.stop(timeout=20.0 if lossless_mode else 5.0)
                     except Exception as exc:
                         ws_debug["rec_error"] = f"stop: {exc!r}"
+                        error = exc
             rec_input = None
             rec_output = None
             rec_base = None
+            if error is not None:
+                raise error
 
         def _write_prompt_sidecar() -> None:
             if rec_base is None:
@@ -1259,7 +1238,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                 pe_task.cancel()
                 try:
                     await pe_task
-                except BaseException:
+                except asyncio.CancelledError:
                     pass
                 pe_task = None
 
@@ -1269,19 +1248,37 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                 stop_output_pump.clear()
                 output_task = asyncio.create_task(_output_pump(session_ref))
 
+        def _check_output() -> None:
+            if output_task is not None and output_task.done():
+                output_task.result()
+                raise RuntimeError("output processing stopped before completion")
+            session._raise_worker_error_if_needed()
+
+        async def _drain_session() -> None:
+            await _run_work(session.flush_pending, timeout=max(0.1, float(args.push_frame_timeout_s)))
+            deadline = time.monotonic() + 20.0
+            while completed_frames < edit_frames:
+                _check_output()
+                if time.monotonic() >= deadline:
+                    raise TimeoutError(f"incomplete output: {completed_frames}/{edit_frames} frames")
+                await asyncio.sleep(0.01)
+            _check_output()
 
         async def _reset_session(reason: str) -> None:
-            nonlocal session, frames_since_session_reset, reset_count
+            nonlocal session, frames_since_session_reset, reset_count, edit_frames, completed_frames
             if session is None:
                 return
             print(f"#####[STREAM] session reset ({reason})", flush=True)
 
+            if lossless_mode:
+                await _drain_session()
             await _stop_output_task()
             await _close_session_safely(session, "kv_reset")
             reset_count += 1
             session = _create_session()
             app.state.active_session = session
             frames_since_session_reset = 0
+            edit_frames = completed_frames = 0
             ws_debug["kv_reset_count"] = reset_count
             ws_debug["frames_since_session_reset"] = frames_since_session_reset
             await _send_json(
@@ -1317,6 +1314,10 @@ def create_app(args: argparse.Namespace) -> FastAPI:
             last_activity = time.monotonic()
             last_frames_out = frames_out
             while True:
+                if app.state.runtime_error:
+                    raise RuntimeError(app.state.runtime_error)
+                if output_task is not None and output_task.done():
+                    output_task.result()
                 if frames_out != last_frames_out or pe_task is not None:
                     last_frames_out = frames_out
                     last_activity = time.monotonic()
@@ -1338,6 +1339,8 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                 if "text" in message and message["text"] is not None:
                     payload = json.loads(message["text"])
                     msg_type = payload.get("type")
+                    if msg_type != "start" and payload.get("session_id", session_id) != session_id:
+                        continue
                     if msg_type == "start":
                         print(f"#####[RESTART] 'start' received (session {'live' if session is not None else 'none'})", flush=True)
                         last_activity = time.monotonic()
@@ -1352,6 +1355,11 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                                 app.state.active_session = None
                             session = None
 
+                        session_id = str(payload.get("session_id") or uuid.uuid4().hex)
+                        frames_in = frames_out = edit_frames = completed_frames = last_frames_out = 0
+                        next_frame_meta = None
+                        ws_debug.update(frames_in=0, frames_out=0, output_bytes=0, chunk_results_sent=0,
+                                        frames_dropped_backpressure=0, session_id=session_id)
                         raw_session_prompt = str(payload.get("prompt", args.prompt))
                         session_prompt = raw_session_prompt
                         try:
@@ -1382,7 +1390,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                             continue
 
                         freeze_kv_on_static = bool(
-                            payload.get("freeze_kv_on_static", args.freeze_kv_on_static)
+                            payload.get("freeze_kv_on_static", False if lossless_mode else args.freeze_kv_on_static)
                         )
                         static_diff_thresh = float(
                             payload.get("static_diff_thresh", args.static_diff_thresh)
@@ -1392,7 +1400,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                         ))
                         use_pe = bool(payload.get("use_pe", args.use_pe)) and bool(os.environ.get("OPENAI_API_KEY"))
 
-                        entry_gate = bool(payload.get("gate_enabled", True))
+                        entry_gate = bool(payload.get("gate_enabled", True)) and not lossless_mode
                         face_gate_pending = entry_gate
                         flow["recv"] = None
                         flow["at"] = 0.0
@@ -1401,6 +1409,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                         flow["clamped"] = False
                         flow["consec"] = 0
                         flow["base"] = frames_out
+                        flow.update(has_ack=False, dropped=0, skew_min=None, skew_at=0.0, up_ms=0.0)
 
                         gate_on = bool(args.online_gate) and entry_gate
 
@@ -1506,25 +1515,20 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                                 "ref_image": ref_image is not None,
                                 "kv_reset_frames": kv_reset_frames,
                                 "use_pe": use_pe,
+                                "pe_deferred": pe_defer,
                                 "pe_model": args.pe_model or DEFAULT_PE_MODEL,
                                 "max_temporal_ids": max_temporal_ids,
                                 "freeze_kv_on_static": freeze_kv_on_static,
                                 "static_diff_thresh": static_diff_thresh,
                             }
                         )
-                        _start_recorders()
+                        await asyncio.to_thread(_start_recorders)
                         _start_output_task(session)
-                        if not pe_defer:
-                            def _prebake(_p=session_prompt, _s=session_settings, _r=ref_image):
-                                with app.state.inference_lock:
-                                    runtime.prebake_graph(_p, settings=_s, ref_image=_r)
-                            asyncio.create_task(asyncio.to_thread(_prebake))
                     elif msg_type == "stop":
                         ws_debug["last_message_type"] = "stop"
 
                         await _cancel_pe()
                         await _stop_output_task()
-                        await asyncio.to_thread(_stop_recorders)
                         break
                     elif msg_type == "finalize_recording":
                         ws_debug["last_message_type"] = "finalize_recording"
@@ -1533,15 +1537,18 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                                               "message": "Recording is not enabled (--record-dir is unset)."})
                             continue
 
-                        if session is not None:
-                            await asyncio.to_thread(session.flush_pending)
-                            _flush_deadline = time.monotonic() + 20.0
-                            while frames_out < frames_in and time.monotonic() < _flush_deadline:
-                                await asyncio.sleep(0.05)
+                        try:
+                            if pe_task is not None:
+                                await pe_task
+                            if session is not None:
+                                await _drain_session()
+                            await _stop_output_task()
+                            finalized = rec_base
+                            await asyncio.to_thread(_stop_recorders)
+                        except Exception as exc:
+                            await _send_json({"type": "recording_finalized", "ok": False, "message": str(exc)})
+                            break
                         last_activity = time.monotonic()
-                        await _stop_output_task()
-                        finalized = rec_base
-                        await asyncio.to_thread(_stop_recorders)
                         await _send_json({
                             "type": "recording_finalized",
                             "ok": finalized is not None,
@@ -1616,6 +1623,8 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                     continue
 
                 if "bytes" not in message or message["bytes"] is None:
+                    if message.get("type") == "websocket.disconnect":
+                        break
                     continue
                 if session is None:
                     await _send_json({"type": "error", "message": "send start JSON before frames"})
@@ -1670,11 +1679,11 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                     if pe_report and pe_report.get("enhanced_prompt"):
                         session_prompt = pe_report["enhanced_prompt"]
                     await _reset_session("kv_reset_frames")
-                    continue
 
                 _prof_on = session_settings is not None and session_settings.profile_timings
 
                 def _run_frame():
+                    nonlocal edit_frames
                     if uplink_frame is not None:
                         frame = uplink_frame
                     elif _prof_on:
@@ -1828,9 +1837,7 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                         gate_state["pe_anchor"] = frame
                         return "__gate_pe__"
 
-                    health_error = _session_health_error(session)
-                    if health_error:
-                        raise RuntimeError(health_error)
+                    session._raise_worker_error_if_needed()
 
                     acquired = app.state.inference_lock.acquire(
                         timeout=max(0.1, float(args.inference_lock_timeout_s))
@@ -1840,29 +1847,32 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                             f"inference lock timeout after {args.inference_lock_timeout_s:.1f}s"
                         )
                     try:
-                        health_error = _session_health_error(session)
-                        if health_error:
-                            raise RuntimeError(health_error)
-
-                        if session_max_inflight and session.inflight_chunks() >= session_max_inflight:
+                        if not lossless_mode and session_max_inflight and session.inflight_chunks() >= session_max_inflight:
                             ws_debug["frames_dropped_backpressure"] = (
                                 int(ws_debug.get("frames_dropped_backpressure", 0)) + 1
                             )
-                            return session._drain_async_results()
+                            return []
 
                         _rec_i = rec_input
                         if _rec_i is not None:
                             _rec_i.submit(frame, frame_meta.get("t_capture_ms"))
                             ws_debug["rec_in_written"] = _rec_i.frames_written
                             ws_debug["rec_in_dropped"] = _rec_i.frames_dropped_recording
-                        return session.push_frame(frame, frame_meta=frame_meta)
+                        session.push_frame(frame, frame_meta=frame_meta, drain_results=False)
+                        edit_frames += 1
+                        return []
                     finally:
                         app.state.inference_lock.release()
 
-                started = time.time()
                 try:
-                    chunk_results = await asyncio.wait_for(
-                        asyncio.to_thread(_run_frame),
+                    capacity_deadline = time.monotonic() + max(0.1, float(args.push_frame_timeout_s))
+                    while lossless_mode and session_max_inflight and session.inflight_chunks() >= session_max_inflight:
+                        _check_output()
+                        if time.monotonic() >= capacity_deadline:
+                            raise TimeoutError("file input timed out waiting for inference capacity")
+                        await asyncio.sleep(0.01)
+                    chunk_results = await _run_work(
+                        _run_frame,
                         timeout=max(0.1, float(args.push_frame_timeout_s)),
                     )
                 except asyncio.TimeoutError:
@@ -1901,15 +1911,13 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                         gate_state["pe_anchor"] = None
                         await _send_json({"type": "pe_running", "frames_in": frames_in})
 
-                        async def _run_pe(_sess=session, _anchor=anchor, _raw=raw_session_prompt):
+                        async def _run_pe(_sess=session, _anchor=anchor, _raw=raw_session_prompt, _ref=ref_image):
                             nonlocal pe_defer, pe_report, pe_task
                             try:
-                                report = await asyncio.wait_for(
-                                    asyncio.to_thread(
-                                        _enhance_prompt_sync,
-                                        raw_prompt=_raw,
-                                        pe_frame=_anchor, pe_model=args.pe_model,
-                                    ),
+                                report = await _enhance_prompt(
+                                    raw_prompt=_raw,
+                                    pe_frame=_anchor, pe_model=args.pe_model,
+                                    ref_image=_ref,
                                     timeout=max(1.0, float(args.pe_timeout_s)),
                                 )
                                 txt = str(report.get("enhanced_prompt") or _raw)
@@ -1918,18 +1926,13 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                                     with app.state.inference_lock:
                                         _sess.prompt = txt
                                         _sess._initialize(_anchor)
-                                await asyncio.to_thread(_swap)
+                                await _run_work(_swap, timeout=max(0.1, float(args.push_frame_timeout_s)))
                                 pe_report = report
                                 ws_debug["pe_report"] = report
                                 await _send_json({"type": "prompt_enhanced", **report})
-                            except Exception:
-                                def _swap_raw():
-                                    with app.state.inference_lock:
-                                        _sess._initialize(_anchor)
-                                await asyncio.to_thread(_swap_raw)
-                                await _send_json({"type": "prompt_enhanced", "enabled": True,
-                                                  "raw_prompt": _raw, "enhanced_prompt": _raw,
-                                                  "fallback": True, "error": True, "elapsed_s": 0.0})
+                            except Exception as exc:
+                                _sess.request_stop()
+                                await _send_json({"type": "error", "message": str(exc)})
                             finally:
                                 pe_defer = False
                                 pe_task = None
@@ -1954,44 +1957,21 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                         session_prompt = pe_report["enhanced_prompt"]
                     await _reset_session("person_count_changed")
                     continue
-                elapsed = time.time() - started
-                if not chunk_results:
-                    await _send_json(
-                        {
-                            "type": "accepted",
-                            "frames_in": frames_in,
-                            "frames_out": frames_out,
-                            "next_chunk_needs": session.frames_per_next_chunk - len(session.pending_frames),
-                        }
-                    )
-                    continue
-
-                last_count = 0
-                for result in chunk_results:
-                    last_count = await _send_chunk_result(
-                        result,
-                        fallback_meta=frame_meta,
-                        fallback_elapsed=elapsed,
-                    )
-                if last_count == 0:
-                    await _send_json(
-                        {
-                            "type": "accepted",
-                            "frames_in": frames_in,
-                            "frames_out": frames_out,
-                            "next_chunk_needs": session.frames_per_next_chunk - len(session.pending_frames),
-                        }
-                    )
+                await _send_json({
+                    "type": "accepted", "frames_in": frames_in, "frames_out": frames_out,
+                    "next_chunk_needs": session.frames_per_next_chunk - len(session.pending_frames),
+                })
         except WebSocketDisconnect:
             pass
-        except RuntimeError as exc:
-            if "disconnect" not in str(exc).lower():
-                raise
+        except Exception as exc:
+            try:
+                await _send_json({"type": "error", "message": str(exc)})
+            except (WebSocketDisconnect, RuntimeError):
+                pass
         finally:
             try:
                 await _cancel_pe()
                 await _stop_output_task()
-                await asyncio.to_thread(_stop_recorders)
                 if session is not None:
                     await _close_session_safely(session, "finally")
                 if getattr(app.state, "active_session", None) is session:
@@ -1999,6 +1979,10 @@ def create_app(args: argparse.Namespace) -> FastAPI:
                 ws_debug["closed_at"] = time.time()
                 ws_debug["send_state"] = "closed"
             finally:
+                try:
+                    await asyncio.to_thread(_stop_recorders)
+                except Exception as exc:
+                    ws_debug["rec_error"] = repr(exc)
                 if ticket is not None:
                     app.state.session_gate.release(ticket)
 
@@ -2056,13 +2040,13 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--prompt", type=str, default="Keep the person and scene temporally consistent while applying the requested edit.")
     parser.add_argument("--profile-timings", action=argparse.BooleanOptionalAction, default=False)
     parser.add_argument("--kv-reset-frames", type=int, default=600)
-    parser.add_argument("--max-temporal-ids", type=int, default=None)
+    parser.add_argument("--max-temporal-ids", type=int, default=8)
     parser.add_argument("--freeze-kv-on-static", action=argparse.BooleanOptionalAction, default=True)
     parser.add_argument("--static-diff-thresh", type=float, default=0.5)
     parser.add_argument("--preload", action=argparse.BooleanOptionalAction, default=True)
-    parser.add_argument("--push-frame-timeout-s", type=float, default=15.0, help="Max seconds a single frame submission may block before releasing the WS session gate.")
-    parser.add_argument("--max-inflight-chunks", type=int, default=2, help="Drop incoming frames while this many chunks are already in the pipeline (latency governor; pins display latency to ~N chunk periods). 2 keeps glass-to-glass latency low (~1 chunk) and prevents the session-start init stall from building a frame backlog (the 'ramp-up'); higher values buffer more (smoother under hiccups) at the cost of latency and a startup ramp. 0 disables. Overridable per session via the start payload's max_inflight_chunks.")
-    parser.add_argument("--session-close-timeout-s", type=float, default=5.0, help="Best-effort session cleanup timeout during WS teardown.")
+    parser.add_argument("--push-frame-timeout-s", type=float, default=15.0, help="Timeout for frame submission or file-input backpressure; teardown must finish before model reuse.")
+    parser.add_argument("--max-inflight-chunks", type=int, default=2, help="Limit chunks in flight: camera input drops excess frames, file input waits for capacity. 0 disables; overridable per session.")
+    parser.add_argument("--session-close-timeout-s", type=float, default=5.0, help="Session cleanup timeout. If workers or GPU work remain active, reject new sessions until restart.")
     parser.add_argument("--inference-lock-timeout-s", type=float, default=5.0, help="Max seconds to wait for the process-wide inference lock.")
 
     parser.add_argument("--record-dir", type=str, default=None, help="Directory to record input/output mp4s into (per-session subfolder). Off if unset.")

xvideo/serving/zerogpu_engine.py

@@ -1,57 +1,18 @@
-"""ZeroGPU session engine for JoyAI-Video-Edit.
-
-NEW FILE — does not modify the existing standalone server
-(serve_joyomni_streaming.py). It runs INSIDE the @spaces.GPU fork and reuses the
-vendored gate / prompt-enhancement / streaming logic to drive one browser session
-whose WebSocket transport has been replaced by two multiprocessing.Queues.
-
-The WS-facing loop from the standalone server is deeply coupled to the live
-`session` object (pending_frames, inference_lock, _encode_streaming_prompt, ...).
-Rather than bridge those across processes, the whole loop runs here, in the fork,
-where the real session is local. The only substitution vs. the standalone server
-is the transport: a `_QueueWebSocket` adapter presents the same
-`receive()/send_json()/send_bytes()` surface the original loop expects, backed by
-the in/out queues instead of a socket. Gate/PE/codec helpers are imported from
-serve_joyomni_streaming verbatim (see the import block).
-"""
+"""Run the shared streaming server inside a ZeroGPU fork over process queues."""
 from __future__ import annotations
 
 import argparse
 import asyncio
-import json
 import os
-import queue as _queue
-import threading
-import time
+import queue
 import traceback
 from pathlib import Path
-from typing import Any
 
-from PIL import Image
+from fastapi import WebSocketDisconnect
 
-from xvideo.serving.serve_joyomni_streaming import (
-    _check_face_gate,
-    build_parser,
-    _decode_image,
-    _decode_ref_image,
-    _enhance_prompt_sync,
-    _count_faces_from,
-    _detect_gate_faces,
-    _face_present_from,
-    _optional_positive_int,
-    _person_present,
-    _H264Stream,
-    _H264Ingest,
-    _SegmentedRecorder,
-    _snap_to_align,
-)
-from xvideo.serving.joyomni_streaming import StreamingSettings
+from xvideo.serving.serve_joyomni_streaming import build_parser, create_app
 
 
-# Server "args" the vendored gate logic reads: serve_joyomni_streaming's own
-# argparse defaults (= run_server.sh effective config), overridden only with
-# Space wiring. Recording persists to /data like the local server records to
-# deploy/recordings (owner's decision, 2026-08-19).
 def _default_args(paths: dict) -> argparse.Namespace:
     args = build_parser().parse_args([])
     args.face_detector_onnx = paths.get("face_onnx", "")
@@ -67,29 +28,29 @@ def _default_args(paths: dict) -> argparse.Namespace:
     return args
 
 
-# Transport adapter: presents a WebSocket-like surface over the in/out queues so
-# the ported session loop reads/writes exactly as it did against a real socket.
 class _QueueWebSocket:
     def __init__(self, in_q, out_q):
         self._in = in_q
         self._out = out_q
         self._closed = False
 
+    async def accept(self) -> None:
+        pass
+
     async def receive(self) -> dict:
-        loop = asyncio.get_event_loop()
-        while True:
+        while not self._closed:
             try:
-                item = await loop.run_in_executor(None, lambda: self._in.get(timeout=1.0))
-            except _queue.Empty:
+                item = self._in.get_nowait()
+            except queue.Empty:
+                await asyncio.sleep(0.005)
                 continue
             kind = item.get("kind")
             if kind == "close":
                 self._closed = True
-                return {"type": "websocket.disconnect"}
-            if kind == "text":
-                return {"text": item["data"], "bytes": None, "type": "websocket.receive"}
-            if kind == "bytes":
-                return {"text": None, "bytes": item["data"], "type": "websocket.receive"}
+                break
+            if kind in {"text", "bytes"}:
+                return {"type": "websocket.receive", kind: item["data"]}
+        return {"type": "websocket.disconnect"}
 
     async def send_json(self, payload: dict) -> None:
         await self._put(("json", payload))
@@ -98,25 +59,15 @@ class _QueueWebSocket:
         await self._put(("bin", data))
 
     async def _put(self, item) -> None:
-        loop = asyncio.get_event_loop()
-        if item[0] == "json":
-            def _put_blocking():
-                deadline = time.monotonic() + 60.0
-                while not self._closed and time.monotonic() < deadline:
-                    try:
-                        self._out.put(item, timeout=1.0)
-                        return
-                    except _queue.Full:
-                        continue
-            await loop.run_in_executor(None, _put_blocking)
-            return
-        try:
-            await loop.run_in_executor(None, lambda: self._out.put(item, timeout=2.0))
-        except _queue.Full:
-            pass  # drop frames under backpressure; control messages never drop
+        while not self._closed:
+            try:
+                self._out.put_nowait(item)
+                return
+            except queue.Full:
+                await asyncio.sleep(0.005)
+        raise WebSocketDisconnect()
 
 
-# Entry point called from the fork (app.py::_run_session_in_fork).
 _VAE_AOTI = {"done": False}
 
 
@@ -231,550 +182,8 @@ def run_session_blocking(runtime, paths: dict, in_q, out_q) -> None:
             pass
 
 
-# The ported session loop. Structurally mirrors websocket_endpoint() in
-# serve_joyomni_streaming.py, minus the SessionGate and with the transport
-# swapped for _QueueWebSocket. Kept compact: gate/PE helpers are imported;
-# the per-frame gate STATE MACHINE is preserved.
 async def _session_loop(runtime, args, websocket: _QueueWebSocket) -> None:
-    session = None
-    frames_in = 0
-    frames_out = 0
-    session_prompt = args.prompt
-    raw_session_prompt = args.prompt
-    session_settings: StreamingSettings | None = None
-    ref_image: Image.Image | None = None
-    output_quality = 60 if args.output_quality == "auto" else int(args.output_quality)
-    output_codec = "mjpeg"
-    h264_stream: _H264Stream | None = None
-    input_codec = "mjpeg"
-    h264_ingest: _H264Ingest | None = None
-    frames_since_session_reset = 0
-    next_frame_meta: dict[str, Any] | None = None
-
-    gate_state: dict[str, Any] = {}
-
-    inference_lock = threading.Lock()
-
-    def _locked_close(_s):
-        with inference_lock:
-            _s.close()
-
-    async def _send_json(payload: dict) -> None:
-        await websocket.send_json(payload)
-
-    def _create_session():
-        return runtime.create_v2v_session(
-            prompt=session_prompt, settings=session_settings, ref_image=ref_image,
-        )
-
-    # ---- output pump: drains decoded chunks and streams them out ----
-    stop_output = threading.Event()
-    output_task: asyncio.Task | None = None
-
-    async def _send_chunk_result(result) -> int:
-        if not result.jpegs:
-            return 0
-        frames = result.jpegs
-        metas = result.source_metas or [{} for _ in frames]
-        if result.valid_count is not None:
-            frames = frames[: result.valid_count]
-            metas = metas[: result.valid_count]
-        nonlocal frames_out
-        packets = None
-        if h264_stream is not None and frames and not isinstance(frames[0], bytes):
-            packets = await asyncio.to_thread(h264_stream.encode, frames)
-        await _send_json({"type": "chunk_start", "count": len(frames),
-                          "elapsed": float(result.elapsed or 0.0), "frames_in": frames_in})
-        for idx, item in enumerate(frames):
-            meta = metas[min(idx, len(metas) - 1)]
-            if rec_output is not None:
-                rec_output.submit(item, meta.get("t_capture_ms"))
-            wire, is_key = (packets[idx] if packets is not None else (item, False))
-            await _send_json({"type": "output_frame", "index": idx, "count": len(frames),
-                              "source_seq": meta.get("seq"), "t_capture_ms": meta.get("t_capture_ms"),
-                              "profile": result.profile, "key": is_key})
-            await websocket.send_bytes(wire)
-            frames_out += 1
-        await _send_json({"type": "chunk_done", "count": len(frames),
-                          "frames_in": frames_in, "frames_out": frames_out})
-        return len(frames)
-
-    async def _output_pump(sess) -> None:
-        while not stop_output.is_set():
-            try:
-                result = await asyncio.to_thread(sess.wait_async_result, 0.05)
-            except Exception as exc:  # noqa: BLE001
-                await _send_json({"type": "error", "message": repr(exc)})
-                break
-            if result is not None and result.jpegs:
-                await _send_chunk_result(result)
-        # Stop requested: flush any chunk still in flight in the pipeline so the
-        # user's final frames are not dropped on stop/reset/disconnect.
-        try:
-            for result in await asyncio.to_thread(sess._drain_async_results):
-                if result is not None and result.jpegs:
-                    await _send_chunk_result(result)
-        except Exception:  # noqa: BLE001
-            pass
-
-    def _start_output(sess):
-        nonlocal output_task
-        stop_output.clear()
-        output_task = asyncio.create_task(_output_pump(sess))
-
-    async def _stop_output():
-        nonlocal output_task
-        stop_output.set()
-        if output_task is not None:
-            try:
-                await asyncio.wait_for(output_task, timeout=1.0)
-            except Exception:  # noqa: BLE001
-                output_task.cancel()
-        output_task = None
-
-    async def _reset_session(reason: str):
-        nonlocal session, frames_since_session_reset
-        if session is None:
-            return
-        await _stop_output()
-        await asyncio.to_thread(_locked_close, session)
-        session = _create_session()
-        frames_since_session_reset = 0
-        await _send_json({"type": "session_reset", "reason": reason,
-                          "frames_per_next_chunk": session.frames_per_next_chunk})
-        _start_output(session)
-
-    # ---- recording: same shape as the local server (input/output mp4 segments
-    # per session dir + prompts.json sidecar); recorders span session resets ----
-    rec_input: _SegmentedRecorder | None = None
-    rec_output: _SegmentedRecorder | None = None
-    rec_base: Path | None = None
-    rec_seq = 0
-
-    def _write_prompt_sidecar() -> None:
-        if rec_base is None:
-            return
-        try:
-            doc = {"raw_prompt": raw_session_prompt}
-            if session_prompt and session_prompt != raw_session_prompt:
-                doc["enhanced_prompt"] = session_prompt
-            (rec_base / "prompts.json").write_text(
-                json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
-        except Exception:  # noqa: BLE001
-            pass
-
-    def _stop_recorders() -> None:
-        nonlocal rec_input, rec_output, rec_base
-        for _rec in (rec_input, rec_output):
-            if _rec is not None:
-                try:
-                    _rec.stop()
-                except Exception:  # noqa: BLE001
-                    pass
-        rec_input = rec_output = None
-        rec_base = None
-
-    def _start_recorders() -> None:
-        nonlocal rec_input, rec_output, rec_seq, rec_base
-        if args.record_dir is None:
-            return
-        _stop_recorders()
-        try:
-            rec_seq += 1
-            base = Path(args.record_dir) / f"{int(time.time())}_{rec_seq}"
-            base.mkdir(parents=True, exist_ok=True)
-            common = dict(codec=str(args.record_codec), bitrate=int(args.record_bitrate),
-                          segment_seconds=int(args.record_segment_seconds))
-            rec_input = _SegmentedRecorder(prefix=base / "input", **common)
-            rec_output = _SegmentedRecorder(prefix=base / "output", **common)
-            rec_input.start()
-            rec_output.start()
-            rec_base = base
-            print(f"#####[REC] recording -> {base}", flush=True)
-            _write_prompt_sidecar()
-            if ref_image is not None:
-                try:
-                    ref_image.save(base / "ref.png")
-                except Exception:  # noqa: BLE001
-                    pass
-        except Exception as exc:  # noqa: BLE001
-            print(f"#####[REC] start failed: {exc!r} -> recording OFF", flush=True)
-            rec_input = rec_output = None
-            rec_base = None
-
-    await _send_json({"type": "session_granted"})
-
-    # ---- main receive loop ----
-    while True:
-        message = await websocket.receive()
-        if message.get("type") == "websocket.disconnect":
-            break
-        text = message.get("text")
-        if text is not None:
-            payload = json.loads(text)
-            msg_type = payload.get("type")
-
-            if msg_type == "start":
-                if session is not None:
-                    await _stop_output()
-                    await asyncio.to_thread(_locked_close, session)
-                    session = None
-                raw_session_prompt = str(payload.get("prompt", args.prompt))
-                session_prompt = raw_session_prompt
-                try:
-                    ref_image = _decode_ref_image(payload.get("ref_image"))
-                except Exception as exc:  # noqa: BLE001
-                    await _send_json({"type": "error", "message": f"ref image decode: {exc!r}"})
-                    continue
-                kv_reset_frames = max(0, int(payload.get("kv_reset_frames", args.kv_reset_frames)))
-                output_quality = max(1, min(100, int(payload.get("output_quality", output_quality))))
-                runtime.output_quality = output_quality
-                runtime.lossless_output = False
-                _up_allow = args.uplink_codec == "auto"
-                _dn_allow = args.downlink_codec == "auto"
-                output_codec = "h264" if payload.get("output_codec") == "h264" and _dn_allow else "mjpeg"
-                h264_stream = _H264Stream(output_quality) if output_codec == "h264" else None
-                input_codec = "h264" if payload.get("input_codec") == "h264" and _up_allow else "mjpeg"
-                h264_ingest = _H264Ingest() if input_codec == "h264" else None
-                try:
-                    max_temporal_ids = _optional_positive_int(
-                        payload.get("max_temporal_ids", args.max_temporal_ids), name="max_temporal_ids")
-                except Exception as exc:  # noqa: BLE001
-                    await _send_json({"type": "error", "message": f"bad max_temporal_ids: {exc!r}"})
-                    continue
-                freeze_kv_on_static = bool(payload.get("freeze_kv_on_static", args.freeze_kv_on_static))
-                static_diff_thresh = float(payload.get("static_diff_thresh", args.static_diff_thresh))
-                session_max_inflight = max(0, int(
-                    payload.get("max_inflight_chunks", args.max_inflight_chunks) or 0))
-                use_pe = bool(payload.get("use_pe", args.use_pe)) and bool(os.environ.get("OPENAI_API_KEY"))
-                entry_gate = bool(payload.get("gate_enabled", True))
-                face_gate_pending = entry_gate
-                gate_on = bool(args.online_gate) and entry_gate
-                fg_score = float(payload.get("fg_score", args.face_gate_score))
-                fg_min_below = float(payload.get("fg_min_below_ratio", args.face_gate_min_below_ratio))
-                fg_stable = max(1, int(payload.get("fg_stable_frames", args.face_gate_stable_frames)))
-                fg_absent = max(1, int(args.presence_absent_frames))
-                fg_return = max(1, int(args.presence_return_frames))
-                _gate_fps = float(payload.get("fps") or args.fps or 24.0)
-                _fscale = _gate_fps / 24.0
-                fg_stable = max(1, int(round(fg_stable * _fscale)))
-                fg_absent = max(1, int(round(fg_absent * _fscale)))
-                fg_return = max(1, int(round(fg_return * _fscale)))
-                count_change_frames = max(1, int(round(int(args.person_count_change_frames) * _fscale)))
-                body_flip_frames = max(1, int(round(int(args.person_body_flip_frames) * _fscale)))
-                person_stride = max(1, int(round(int(args.person_check_stride) * _fscale)))
-                gate_move_eps = float(args.face_gate_move_eps) / _fscale
-                _reset_gate_state(gate_state)
-                pe_defer = False
-                if use_pe:
-                    cached = str(payload.get("enhanced_prompt") or "").strip()
-                    if cached:
-                        session_prompt = cached
-                    else:
-                        pe_defer = True
-                _align = runtime.pipeline.vae.stem.stride * 8
-                session_settings = StreamingSettings(
-                    height=_snap_to_align(int(payload.get("height", args.height)), _align),
-                    width=_snap_to_align(int(payload.get("width", args.width)), _align),
-                    num_inference_steps=int(payload.get("num_inference_steps", args.num_inference_steps)),
-                    seed=int(payload.get("seed", args.seed)),
-                    max_temporal_ids=max_temporal_ids,
-                    freeze_kv_on_static=freeze_kv_on_static,
-                    static_diff_thresh=static_diff_thresh,
-                    profile_timings=bool(payload.get("profile_timings", args.profile_timings)),
-                    output_codec=output_codec,
-                )
-                session = _create_session()
-                frames_since_session_reset = 0
-                if not pe_defer:
-                    # Bake the graph before "started" so early frames don't age in queue.
-                    def _prebake(_p=session_prompt, _s=session_settings, _r=ref_image):
-                        with inference_lock:
-                            runtime.prebake_graph(_p, settings=_s, ref_image=_r)
-                    try:
-                        await asyncio.to_thread(_prebake)
-                    except Exception:  # noqa: BLE001
-                        pass
-                await _send_json({"type": "started",
-                                  "frames_per_next_chunk": session.frames_per_next_chunk,
-                                  "height": session_settings.height, "width": session_settings.width,
-                                  "output_codec": output_codec, "input_codec": input_codec,
-                                  "ref_image": ref_image is not None, "use_pe": use_pe})
-                _start_output(session)
-                await asyncio.to_thread(_start_recorders)
-
-            elif msg_type == "stop":
-                await _stop_output()
-                await asyncio.to_thread(_stop_recorders)
-                break
-
-            elif msg_type == "finalize_recording":
-                if args.record_dir is None:
-                    await _send_json({"type": "recording_finalized", "ok": False,
-                                      "message": "Recording is not enabled (--record-dir is unset)."})
-                    continue
-                if session is not None:
-                    await asyncio.to_thread(session.flush_pending)
-                    _deadline = time.monotonic() + 20.0
-                    while frames_out < frames_in and time.monotonic() < _deadline:
-                        await asyncio.sleep(0.05)
-                await _stop_output()
-                finalized = rec_base
-                await asyncio.to_thread(_stop_recorders)
-                await _send_json({"type": "recording_finalized", "ok": finalized is not None,
-                                  "rec": finalized.name if finalized is not None else None,
-                                  "message": None if finalized is not None
-                                  else "No downloadable result yet. Send an edit first."})
-            elif msg_type == "frame_meta":
-                next_frame_meta = {"seq": int(payload.get("seq", frames_in + 1)),
-                                   "t_capture_ms": float(payload.get("t_capture_ms", time.time() * 1000.0))}
-            elif msg_type == "ping":
-                await _send_json({"type": "pong", "t": payload.get("t")})
-            elif msg_type == "set_output_quality":
-                try:
-                    output_quality = max(1, min(100, int(payload.get("value", output_quality))))
-                    runtime.output_quality = output_quality
-                    if h264_stream is not None:
-                        h264_stream.set_quality(output_quality)
-                except (TypeError, ValueError):
-                    pass
-            continue
-
-        # ---- binary frame ----
-        frame_bytes = message.get("bytes")
-        if frame_bytes is None:
-            continue
-        if session is None:
-            await _send_json({"type": "error", "message": "send start before frames"})
-            continue
-
-        frames_in += 1
-        frame_meta = next_frame_meta or {"seq": frames_in, "t_capture_ms": time.time() * 1000.0}
-        next_frame_meta = None
-
-        uplink_frame: Image.Image | None = None
-        if h264_ingest is not None:
-            uplink_frame = await asyncio.to_thread(h264_ingest.decode_one, frame_bytes)
-            if uplink_frame is None:
-                continue
-
-        if rec_input is not None:
-            rec_input.submit(uplink_frame if uplink_frame is not None else frame_bytes,
-                             frame_meta.get("t_capture_ms"))
-
-        if kv_reset_frames > 0 and frames_since_session_reset >= kv_reset_frames:
-            face_gate_pending = entry_gate
-            _reset_gate_state(gate_state)
-            await _reset_session("kv_reset_frames")
-            continue
-
-        def _run_frame():
-            frame = uplink_frame if uplink_frame is not None else _decode_image(frame_bytes)
-            sentinel = _apply_gate(
-                frame, gate_state, args,
-                face_gate_pending=face_gate_pending, gate_on=gate_on, pe_defer=pe_defer,
-                fg_score=fg_score, fg_min_below=fg_min_below, fg_stable=fg_stable,
-                fg_absent=fg_absent, fg_return=fg_return, person_stride=person_stride,
-                body_flip_frames=body_flip_frames, count_change_frames=count_change_frames,
-                gate_move_eps=gate_move_eps, session=session,
-            )
-            if sentinel is not None:
-                return sentinel
-            if session_max_inflight and session.inflight_chunks() >= session_max_inflight:
-                return session._drain_async_results()
-            with inference_lock:
-                return session.push_frame(frame, frame_meta=frame_meta)
-
-        chunk_results = await asyncio.to_thread(_run_frame)
-
-        # ---- sentinels from the gate ----
-        if isinstance(chunk_results, tuple) and chunk_results and isinstance(chunk_results[0], str):
-            s = chunk_results[0]
-            if s == "__no_person__":
-                await _send_json({"type": "no_person",
-                                  "reason": chunk_results[1] if len(chunk_results) > 1 else "no_person",
-                                  "frames_in": frames_in})
-                continue
-            if s == "__person_returned__":
-                face_gate_pending = True
-                _reset_gate_state(gate_state)
-                await _reset_session("person_returned")
-                continue
-        if chunk_results == "__gate_pe__":
-            anchor = gate_state.pop("pe_anchor", None)
-            await _send_json({"type": "pe_running", "frames_in": frames_in})
-            try:
-                pe_report = await asyncio.wait_for(
-                    asyncio.to_thread(
-                        _enhance_prompt_sync, raw_prompt=raw_session_prompt,
-                        ref_image=ref_image, pe_frame=anchor, pe_model=args.pe_model),
-                    timeout=max(1.0, float(args.pe_timeout_s)),
-                )
-                enhanced = str(pe_report.get("enhanced_prompt") or raw_session_prompt)
-                session_prompt = enhanced
-
-                def _swap(_s=session, _t=enhanced, _a=anchor):
-                    with inference_lock:
-                        _s.prompt = _t
-                        _s._initialize(_a)
-                await asyncio.to_thread(_swap)
-                print(f"#####[PE] enhanced ok in {float(pe_report.get('elapsed_s', 0.0)):.1f}s "
-                      f"(model={pe_report.get('model')}) -> session initialized", flush=True)
-                _write_prompt_sidecar()
-                await _send_json({"type": "prompt_enhanced", **pe_report})
-            except Exception as exc:  # noqa: BLE001
-                _why = "timeout" if isinstance(exc, asyncio.TimeoutError) else repr(exc)
-                print(f"#####[PE] deferred enhance failed: {_why} -> raw prompt", flush=True)
-
-                def _init_raw(_s=session, _a=anchor):
-                    with inference_lock:
-                        _s._initialize(_a)
-                await asyncio.to_thread(_init_raw)
-                await _send_json({"type": "prompt_enhanced", "enabled": True,
-                                  "raw_prompt": raw_session_prompt, "enhanced_prompt": raw_session_prompt,
-                                  "fallback": True, "error": True, "elapsed_s": 0.0})
-            pe_defer = False
-            continue
-        if isinstance(chunk_results, str):
-            await _send_json({"type": "waiting_face", "reason": chunk_results, "frames_in": frames_in})
-            continue
-
-        if face_gate_pending:
-            face_gate_pending = False
-        frames_since_session_reset += 1
-
-        if gate_state.get("recount"):
-            gate_state["recount"] = False
-            await _reset_session("person_count_changed")
-            continue
-
-        if not chunk_results:
-            await _send_json({"type": "accepted", "frames_in": frames_in, "frames_out": frames_out,
-                              "next_chunk_needs": session.frames_per_next_chunk - len(session.pending_frames)})
-            continue
-        # output frames are streamed by the pump; push_frame's direct returns
-        # (init chunk) are also flushed here for parity.
-        for result in chunk_results:
-            await _send_chunk_result(result)
-
-    await _stop_output()
-    await asyncio.to_thread(_stop_recorders)
-    if session is not None:
-        try:
-            await asyncio.to_thread(_locked_close, session)
-        except Exception:  # noqa: BLE001
-            pass
-
-
-# Gate state machine — extracted from serve_joyomni_streaming's _run_frame.
-# Returns None to proceed to push_frame, or a sentinel (str / tuple) the caller
-# translates into a WS message. Kept faithful to the original transitions.
-def _reset_gate_state(gs: dict) -> None:
-    gs.clear()
-    gs.update({"count": 0, "cx": None, "cy": None, "csz": None, "absent": 0,
-               "absent_hold": False, "present": 0,
-               "person_check_i": 0, "person_last": True, "face_last": True,
-               "hold_reason": "no_person", "body_miss": 0, "subject_count": None,
-               "cand": None, "cand_n": 0, "recount": False, "settle_ax": None,
-               "settle_ay": None, "settle_asz": None, "pe_anchor": None})
-
-
-def _apply_gate(frame, gs, args, *, face_gate_pending, gate_on, pe_defer, fg_score,
-                fg_min_below, fg_stable, fg_absent, fg_return, person_stride,
-                body_flip_frames, count_change_frames, gate_move_eps, session):
-    if face_gate_pending:
-        reason, center, nf = _check_face_gate(
-            frame, onnx_path=args.face_detector_onnx,
-            score_thresh=fg_score, min_below_ratio=fg_min_below)
-        if reason is not None:
-            gs["count"] = 0; gs["settle_ax"] = None; gs["settle_ay"] = None
-            return reason
-        if center is not None:
-            cx, cy, csz = center
-            if nf < 2 and abs(cx - 0.5) > float(args.face_gate_center_margin):
-                gs["count"] = 0; gs["settle_ax"] = None; gs["settle_ay"] = None
-                gs["settle_asz"] = None; gs["cx"] = cx; gs["cy"] = cy; gs["csz"] = csz
-                return "off_center"
-            pcx, pcy, pcsz = gs.get("cx"), gs.get("cy"), gs.get("csz")
-            eps = gate_move_eps; cap = float(args.face_gate_settle_drift)
-            szeps = eps * 0.5
-            still = (pcx is not None and abs(cx - pcx) <= eps and abs(cy - pcy) <= eps
-                     and pcsz is not None and abs(csz - pcsz) <= szeps)
-            ax, ay, asz = gs.get("settle_ax"), gs.get("settle_ay"), gs.get("settle_asz")
-            if (still and ax is not None and abs(cx - ax) <= cap and abs(cy - ay) <= cap
-                    and asz is not None and abs(csz - asz) <= cap):
-                gs["count"] += 1
-            else:
-                gs["count"] = 1; gs["settle_ax"] = cx; gs["settle_ay"] = cy; gs["settle_asz"] = csz
-            gs["cx"] = cx; gs["cy"] = cy; gs["csz"] = csz
-            if gs["count"] < fg_stable:
-                return "settling"
-        if pe_defer:
-            gs["pe_anchor"] = frame
-            return "__gate_pe__"
-
-    if gate_on and not face_gate_pending:
-        gfaces, gfw, gfh = _detect_gate_faces(
-            frame, onnx_path=args.face_detector_onnx, score_thresh=fg_score)
-        tick = gs.get("person_check_i", 0)
-        if tick == 0 or gs.get("absent_hold"):
-            gs["person_last"] = _person_present(
-                frame, onnx_path=args.person_detector_onnx, conf=float(args.person_gate_conf))
-            if gs["person_last"]:
-                gs["face_last"] = _face_present_from(
-                    gfaces, gfw, gfh,
-                    min_ratio=float(args.face_present_min_ratio),
-                    edge_margin=float(args.face_present_edge_margin))
-            else:
-                gs["face_last"] = True
-        gs["person_check_i"] = (tick + 1) % person_stride
-        body_here = bool(gs["person_last"]); face_here = bool(gs.get("face_last", True))
-        present = body_here and face_here
-        reason_now = "no_person" if not body_here else ("no_face" if not face_here else "")
-        body_flip = body_flip_frames
-        if body_here:
-            gs["body_miss"] = 0
-        else:
-            gs["body_miss"] = gs.get("body_miss", 0) + 1
-        if present:
-            gs["absent"] = 0
-            if gs.get("absent_hold"):
-                gs["present"] = gs.get("present", 0) + 1
-                if gs["present"] >= fg_return:
-                    gs["absent_hold"] = False; gs["present"] = 0
-                    return ("__person_returned__",)
-        else:
-            gs["present"] = 0; gs["absent"] += 1
-            if reason_now == "no_person" and gs.get("body_miss", 0) < body_flip:
-                reason_now = "no_face"
-            gs["hold_reason"] = reason_now or gs.get("hold_reason") or "no_person"
-            if not gs.get("absent_hold") and gs["absent"] >= fg_absent:
-                gs["absent_hold"] = True
-                try:
-                    if session is not None:
-                        session.pending_frames.clear(); session.pending_metas.clear()
-                except Exception:  # noqa: BLE001
-                    pass
-        if gs.get("absent_hold"):
-            return ("__no_person__", gs.get("hold_reason", "no_person"))
-
-        n = _count_faces_from(
-            gfaces, gfw, gfh, count_min_ratio=float(args.count_face_min_ratio))
-        if gs["subject_count"] is None:
-            gs["subject_count"] = n; gs["cand"] = None; gs["cand_n"] = 0
-        elif n != gs["subject_count"]:
-            if n == gs["cand"]:
-                gs["cand_n"] += 1
-            else:
-                gs["cand"] = n; gs["cand_n"] = 1
-            if gs["cand_n"] >= count_change_frames:
-                if n > gs["subject_count"]:
-                    gs["recount"] = True
-                gs["subject_count"] = n; gs["cand"] = None; gs["cand_n"] = 0
-        else:
-            gs["cand"] = None; gs["cand_n"] = 0
-
-    if pe_defer and not face_gate_pending:
-        gs["pe_anchor"] = frame
-        return "__gate_pe__"
-    return None
+    app = create_app(args)
+    app.state.runtime = runtime
+    endpoint = next(route.endpoint for route in app.routes if route.path == "/ws")
+    await endpoint(websocket)