quene v2

gradio-web/config.py

@@ -19,9 +19,11 @@ RIGHT_TOP_LOG_SCALE = 1
 UI_GLOBAL_FONT_SIZE = "24px"
 
 # Session / queue 配置
-SESSION_TIMEOUT = 39999  # 30秒无用户主动操作后，交由 gr.State TTL 自动回收 session
-SESSION_CONCURRENCY_ID = "session_slots"
+SESSION_TIMEOUT = 20  # 30秒无用户主动操作后，交由 gr.State TTL 自动回收 session
+SESSION_CONCURRENCY_ID = "session_actions"
 SESSION_CONCURRENCY_LIMIT = 2
+SESSION_INIT_CONCURRENCY_ID = "session_init"
+SESSION_INIT_CONCURRENCY_LIMIT = 2
 
 # 兜底执行次数配置
 EXECUTE_LIMIT_OFFSET = 4  # 兜底执行次数 = non_demonstration_task_length + EXECUTE_LIMIT_OFFSET

gradio-web/gradio_callbacks.py

@@ -26,6 +26,7 @@ from state_manager import (
     set_play_button_clicked,
     set_task_start_time,
     set_ui_phase,
+    try_create_session,
 )
 from image_utils import draw_marker, save_video, concatenate_frames_horizontally
 from user_manager import user_manager
@@ -75,6 +76,13 @@ def touch_session(uid):
     return uid if uid and get_session(uid) is not None else None
 
 
+def touch_session_or_preserve_pending(uid, init_pending=False):
+    """Keep pending init uid alive until a real session is created."""
+    if init_pending:
+        return uid
+    return touch_session(uid)
+
+
 def cleanup_user_session(uid):
     """Unified cleanup entry for gr.State TTL deletion and unload hooks."""
     if not uid:
@@ -687,6 +695,10 @@ def init_session_and_load_task(uid):
     if get_session(uid) is None:
         create_session(uid)
 
+    return _load_initialized_session_task(uid)
+
+
+def _load_initialized_session_task(uid):
     LOGGER.debug("init_session_and_load_task: init_session uid=%s", _uid_for_log(uid))
     success, msg, status = user_manager.init_session(uid)
     LOGGER.debug(
@@ -703,6 +715,42 @@ def init_session_and_load_task(uid):
     return _load_status_task(uid, status)
 
 
+def try_init_session_and_load_task(uid):
+    """Try to initialize the session without blocking on a full slot queue."""
+    if not uid:
+        return {
+            "status": "ready",
+            "load_result": _task_load_failed_response(uid, _session_error_text()),
+        }
+
+    if get_session(uid) is None:
+        ready, queue_position = try_create_session(uid)
+        if not ready:
+            LOGGER.info(
+                "try_init_session_and_load_task pending uid=%s queue_position=%s",
+                _uid_for_log(uid),
+                queue_position,
+            )
+            return {
+                "status": "pending",
+                "uid": uid,
+                "queue_position": queue_position,
+            }
+
+    return {
+        "status": "ready",
+        "load_result": _load_initialized_session_task(uid),
+    }
+
+
+def resume_pending_init(uid, init_pending=False, request: gr.Request | None = None):
+    """Retry a previously pending init attempt."""
+    if not init_pending:
+        return {"status": "skip"}
+    effective_uid = uid or getattr(request, "session_hash", None)
+    return try_init_session_and_load_task(effective_uid)
+
+
 def load_next_task_wrapper(uid):
     """Move to a random episode within the same env and reload task."""
     if not uid or get_session(uid) is None:
@@ -977,13 +1025,20 @@ def init_app(request: gr.Request):
         uid = getattr(request, "session_hash", None)
         LOGGER.info("init_app: session_hash=%s", _uid_for_log(uid))
         LOGGER.info("init_app: created uid=%s", _uid_for_log(uid))
-        result = init_session_and_load_task(uid)
-        LOGGER.debug("init_app: init_session_and_load_task returned %s outputs", len(result))
+        result = try_init_session_and_load_task(uid)
+        if isinstance(result, dict) and result.get("status") == "ready":
+            LOGGER.debug(
+                "init_app: init_session_and_load_task returned %s outputs",
+                len(result.get("load_result", ()) or ()),
+            )
         return result
     except Exception as e:
         LOGGER.exception("init_app exception")
         # Return a safe fallback that hides the loading overlay and shows error
-        return _task_load_failed_response("", _ui_text("errors", "init_failed", error=e))
+        return {
+            "status": "ready",
+            "load_result": _task_load_failed_response("", _ui_text("errors", "init_failed", error=e)),
+        }
 
 
 def precheck_execute_inputs(uid, option_idx, coords_str):

gradio-web/state_manager.py

@@ -16,6 +16,7 @@ LOGGER = logging.getLogger("robomme.state_manager")
 # --- 全局会话存储 ---
 GLOBAL_SESSIONS = {}
 ACTIVE_SESSION_SLOTS = set()
+WAITING_SESSION_QUEUE = []
 
 # --- 任务索引存储（用于进度显示） ---
 TASK_INDEX_MAP = {}  # {uid: {"task_index": int, "total_tasks": int}}
@@ -42,6 +43,88 @@ def get_session(uid):
         return GLOBAL_SESSIONS.get(uid)
 
 
+def _cleanup_waiting_session_queue_locked():
+    if not WAITING_SESSION_QUEUE:
+        return
+
+    cleaned_queue = []
+    seen = set()
+    for queued_uid in WAITING_SESSION_QUEUE:
+        if not queued_uid or queued_uid in seen:
+            continue
+        if queued_uid in ACTIVE_SESSION_SLOTS:
+            continue
+        if queued_uid in GLOBAL_SESSIONS:
+            continue
+        seen.add(queued_uid)
+        cleaned_queue.append(queued_uid)
+
+    WAITING_SESSION_QUEUE[:] = cleaned_queue
+
+
+def _ensure_waiting_session_locked(uid):
+    _cleanup_waiting_session_queue_locked()
+    if uid not in WAITING_SESSION_QUEUE:
+        WAITING_SESSION_QUEUE.append(uid)
+    return WAITING_SESSION_QUEUE.index(uid) + 1
+
+
+def _try_reserve_session_slot_locked(uid, session_concurrency_limit):
+    if uid in ACTIVE_SESSION_SLOTS:
+        return True, 0
+
+    _cleanup_waiting_session_queue_locked()
+
+    if len(ACTIVE_SESSION_SLOTS) < int(session_concurrency_limit):
+        if WAITING_SESSION_QUEUE and WAITING_SESSION_QUEUE[0] != uid:
+            queue_position = _ensure_waiting_session_locked(uid)
+            LOGGER.debug(
+                "try_reserve_session_slot delayed uid=%s queue_position=%s active_slots=%s limit=%s",
+                uid,
+                queue_position,
+                len(ACTIVE_SESSION_SLOTS),
+                session_concurrency_limit,
+            )
+            return False, queue_position
+
+        if WAITING_SESSION_QUEUE and WAITING_SESSION_QUEUE[0] == uid:
+            WAITING_SESSION_QUEUE.pop(0)
+
+        ACTIVE_SESSION_SLOTS.add(uid)
+        LOGGER.info(
+            "try_reserve_session_slot acquired uid=%s active_slots=%s",
+            uid,
+            len(ACTIVE_SESSION_SLOTS),
+        )
+        return True, 0
+
+    queue_position = _ensure_waiting_session_locked(uid)
+    LOGGER.info(
+        "try_reserve_session_slot queued uid=%s queue_position=%s active_slots=%s limit=%s",
+        uid,
+        queue_position,
+        len(ACTIVE_SESSION_SLOTS),
+        session_concurrency_limit,
+    )
+    return False, queue_position
+
+
+def try_reserve_session_slot(uid):
+    """
+    Try to reserve a session slot without blocking.
+
+    Returns:
+        tuple[bool, int]: (acquired, queue_position)
+    """
+    if not uid:
+        raise ValueError("Session uid cannot be empty")
+
+    from config import SESSION_CONCURRENCY_LIMIT
+
+    with _session_slot_condition:
+        return _try_reserve_session_slot_locked(uid, SESSION_CONCURRENCY_LIMIT)
+
+
 def reserve_session_slot(uid):
     """
     Block until a session slot is available for this uid.
@@ -55,22 +138,17 @@ def reserve_session_slot(uid):
     from config import SESSION_CONCURRENCY_LIMIT
 
     with _session_slot_condition:
-        if uid in ACTIVE_SESSION_SLOTS:
-            return
-        while len(ACTIVE_SESSION_SLOTS) >= int(SESSION_CONCURRENCY_LIMIT):
+        while True:
+            acquired, queue_position = _try_reserve_session_slot_locked(uid, SESSION_CONCURRENCY_LIMIT)
+            if acquired:
+                return
             LOGGER.info(
-                "reserve_session_slot waiting uid=%s active_slots=%s limit=%s",
+                "reserve_session_slot waiting uid=%s queue_position=%s active_slots=%s",
                 uid,
+                queue_position,
                 len(ACTIVE_SESSION_SLOTS),
-                SESSION_CONCURRENCY_LIMIT,
             )
             _session_slot_condition.wait(timeout=0.1)
-        ACTIVE_SESSION_SLOTS.add(uid)
-        LOGGER.info(
-            "reserve_session_slot acquired uid=%s active_slots=%s",
-            uid,
-            len(ACTIVE_SESSION_SLOTS),
-        )
 
 
 def release_session_slot(uid):
@@ -80,6 +158,7 @@ def release_session_slot(uid):
     with _session_slot_condition:
         if uid in ACTIVE_SESSION_SLOTS:
             ACTIVE_SESSION_SLOTS.remove(uid)
+            _cleanup_waiting_session_queue_locked()
             LOGGER.info(
                 "release_session_slot uid=%s active_slots=%s",
                 uid,
@@ -88,6 +167,38 @@ def release_session_slot(uid):
             _session_slot_condition.notify_all()
 
 
+def try_create_session(uid):
+    """
+    Try to create a ProcessSessionProxy without blocking on session slot wait.
+
+    Returns:
+        tuple[bool, int]: (ready, queue_position)
+    """
+    if not uid:
+        raise ValueError("Session uid cannot be empty")
+
+    with _state_lock:
+        if GLOBAL_SESSIONS.get(uid) is not None:
+            return True, 0
+
+    acquired, queue_position = try_reserve_session_slot(uid)
+    if not acquired:
+        return False, queue_position
+
+    try:
+        with _state_lock:
+            session = GLOBAL_SESSIONS.get(uid)
+            if session is None:
+                session = ProcessSessionProxy()
+                GLOBAL_SESSIONS[uid] = session
+    except Exception:
+        release_session_slot(uid)
+        raise
+
+    LOGGER.info("try_create_session uid=%s total_sessions=%s", uid, len(GLOBAL_SESSIONS))
+    return True, 0
+
+
 def create_session(uid):
     """
     为指定 session key 创建 ProcessSessionProxy。
@@ -220,6 +331,8 @@ def cleanup_session(uid):
 
     with _state_lock:
         session = GLOBAL_SESSIONS.pop(uid, None)
+        if uid in WAITING_SESSION_QUEUE:
+            WAITING_SESSION_QUEUE[:] = [queued_uid for queued_uid in WAITING_SESSION_QUEUE if queued_uid != uid]
         TASK_INDEX_MAP.pop(uid, None)
         UI_PHASE_MAP.pop(uid, None)
         PLAY_BUTTON_CLICKED.pop(uid, None)

gradio-web/test/test_queue_session_limit_e2e.py

@@ -106,6 +106,46 @@ def _read_progress_overlay_snapshot(page) -> dict[str, float | bool | None]:
     )
 
 
+def _read_session_wait_overlay_snapshot(page) -> dict[str, float | bool | None]:
+    return page.evaluate(
+        """() => {
+            const host = document.getElementById('native_progress_host');
+            if (!host) {
+                return { present: false, visible: false, width: null, height: null, background: null, content: null };
+            }
+            const markdown = host.querySelector('[data-testid="markdown"]');
+            const prose = markdown ? markdown.querySelector('.prose, .md') || markdown : null;
+            const text = prose ? ((prose.innerText || prose.textContent || '').trim()) : '';
+            const rect = host.getBoundingClientRect();
+            const style = getComputedStyle(host);
+            return {
+                present: true,
+                visible: style.display !== 'none' && rect.width > 0 && rect.height > 0 && text.length > 0,
+                width: rect.width,
+                height: rect.height,
+                background: style.backgroundColor || null,
+                content: text || null,
+            };
+        }"""
+    )
+
+
+def _read_log_output_value(page) -> str | None:
+    return page.evaluate(
+        """() => {
+            const root = document.getElementById('log_output');
+            if (!root) return null;
+            const field = root.querySelector('textarea, input');
+            if (field && typeof field.value === 'string') {
+                const value = field.value.trim();
+                return value || null;
+            }
+            const value = (root.textContent || '').trim();
+            return value || null;
+        }"""
+    )
+
+
 def _mount_demo(demo):
     port = _free_port()
     host = "127.0.0.1"
@@ -141,7 +181,7 @@ def test_gradio_queue_respects_configured_limit_on_init_load(monkeypatch):
         with sync_playwright() as p:
             browser = p.chromium.launch(headless=True)
             pages = []
-            total_pages = int(config.SESSION_CONCURRENCY_LIMIT) + 1
+            total_pages = int(config.SESSION_INIT_CONCURRENCY_LIMIT) + 2
             for _ in range(total_pages):
                 page = browser.new_page(viewport={"width": 1280, "height": 900})
                 page.goto(root_url, wait_until="domcontentloaded")
@@ -150,26 +190,35 @@ def test_gradio_queue_respects_configured_limit_on_init_load(monkeypatch):
 
             def _queue_snapshot_ready():
                 progress_texts = [_read_progress_text(page) for page in pages]
-                first_four_ready = all(
+                processing_ready = all(
                     text and config.UI_TEXT["progress"]["episode_loading"] in text
-                    for text in progress_texts[: config.SESSION_CONCURRENCY_LIMIT]
+                    for text in progress_texts[: config.SESSION_INIT_CONCURRENCY_LIMIT]
                 )
-                queued_text = progress_texts[-1] or ""
-                queued_ready = (
-                    config.UI_TEXT["progress"]["queue_wait"] in queued_text
-                    and "queue:" in queued_text.lower()
+                queued_ready = all(
+                    text
+                    and config.UI_TEXT["progress"]["queue_wait"] in text
+                    and "queue:" in text.lower()
+                    for text in progress_texts[config.SESSION_INIT_CONCURRENCY_LIMIT :]
                 )
-                return first_four_ready and queued_ready
+                return processing_ready and queued_ready
 
             _wait_until(_queue_snapshot_ready, timeout_s=10.0)
-            active_pages = [_read_progress_text(page) or "" for page in pages[: config.SESSION_CONCURRENCY_LIMIT]]
-            queued_text = _read_progress_text(pages[-1]) or ""
-
-            assert all(config.UI_TEXT["progress"]["episode_loading"] in text for text in active_pages)
-            assert config.UI_TEXT["progress"]["queue_wait"] in queued_text
-            assert "queue:" in queued_text.lower()
+            processing_pages = [
+                _read_progress_text(page) or ""
+                for page in pages[: config.SESSION_INIT_CONCURRENCY_LIMIT]
+            ]
+            queued_pages = [
+                _read_progress_text(page) or ""
+                for page in pages[config.SESSION_INIT_CONCURRENCY_LIMIT :]
+            ]
+
+            assert all(config.UI_TEXT["progress"]["episode_loading"] in text for text in processing_pages)
+            assert all(
+                config.UI_TEXT["progress"]["queue_wait"] in text and "queue:" in text.lower()
+                for text in queued_pages
+            )
             assert pages[0].evaluate("() => !!document.getElementById('loading_overlay_group')") is False
-            overlay_snapshot = _read_progress_overlay_snapshot(pages[0])
+            overlay_snapshot = _read_progress_overlay_snapshot(pages[-1])
             assert overlay_snapshot["present"] is True
             assert overlay_snapshot["width"] and overlay_snapshot["width"] > 0
             assert overlay_snapshot["height"] and overlay_snapshot["height"] >= 400
@@ -199,6 +248,10 @@ def test_gradio_state_ttl_cleans_up_idle_session(monkeypatch):
         def __init__(self, uid):
             self.uid = uid
 
+        def get_pil_image(self, use_segmented=False):
+            _ = use_segmented
+            return Image.new("RGB", (32, 32), color=(10, 20, 30))
+
         def close(self):
             closed.append(self.uid)
 
@@ -423,18 +476,18 @@ def test_late_user_waits_for_active_session_slot_release(monkeypatch):
             page3.goto(root_url, wait_until="domcontentloaded")
 
             _wait_until(
-                lambda: (_read_progress_text(page3) or "").startswith(config.UI_TEXT["progress"]["episode_loading"]),
+                lambda: _read_session_wait_overlay_snapshot(page3)["visible"] is True,
                 timeout_s=10.0,
             )
             time.sleep(1.0)
             assert len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT
             assert len(state_manager.ACTIVE_SESSION_SLOTS) == config.SESSION_CONCURRENCY_LIMIT
-            assert _read_progress_text(page3) is not None
+            assert _read_session_wait_overlay_snapshot(page3)["visible"] is True
 
             page1.close()
 
             _wait_until(lambda: len(closed) >= 1, timeout_s=10.0)
-            _wait_until(lambda: _read_progress_text(page3) is None, timeout_s=15.0)
+            _wait_until(lambda: _read_session_wait_overlay_snapshot(page3)["visible"] is False, timeout_s=15.0)
             _wait_until(lambda: len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT, timeout_s=10.0)
             _wait_until(lambda: len(state_manager.ACTIVE_SESSION_SLOTS) == config.SESSION_CONCURRENCY_LIMIT, timeout_s=10.0)
 
@@ -443,3 +496,188 @@ def test_late_user_waits_for_active_session_slot_release(monkeypatch):
         server.should_exit = True
         thread.join(timeout=10)
         demo.close()
+
+
+def test_second_and_later_late_users_show_queue_overlay(monkeypatch):
+    config = importlib.reload(importlib.import_module("config"))
+    state_manager = importlib.reload(importlib.import_module("state_manager"))
+    callbacks = importlib.reload(importlib.import_module("gradio_callbacks"))
+    ui_layout = importlib.reload(importlib.import_module("ui_layout"))
+
+    class _FakeProxy:
+        def __init__(self):
+            self.env_id = None
+            self.episode_idx = None
+            self.language_goal = "goal"
+            self.available_options = [("pick", 0)]
+            self.raw_solve_options = [{"label": "a", "action": "pick", "available": False}]
+            self.demonstration_frames = []
+
+        def load_episode(self, env_id, episode_idx):
+            self.env_id = env_id
+            self.episode_idx = episode_idx
+            return Image.new("RGB", (32, 32), color=(10, 20, 30)), "loaded"
+
+        def get_pil_image(self, use_segmented=False):
+            _ = use_segmented
+            return Image.new("RGB", (32, 32), color=(10, 20, 30))
+
+        def close(self):
+            return None
+
+    monkeypatch.setattr(state_manager, "ProcessSessionProxy", _FakeProxy)
+    monkeypatch.setattr(
+        callbacks.user_manager,
+        "init_session",
+        lambda uid: (
+            True,
+            "ok",
+            {"current_task": {"env_id": "BinFill", "episode_idx": 1}, "completed_count": 0},
+        ),
+    )
+    monkeypatch.setattr(callbacks, "get_task_hint", lambda env_id: "")
+    monkeypatch.setattr(callbacks, "should_show_demo_video", lambda env_id: False)
+
+    demo = ui_layout.create_ui_blocks()
+    root_url, demo, server, thread = _mount_demo(demo)
+
+    try:
+        with sync_playwright() as p:
+            browser = p.chromium.launch(headless=True)
+
+            active_pages = []
+            for _ in range(config.SESSION_CONCURRENCY_LIMIT):
+                page = browser.new_page(viewport={"width": 1280, "height": 900})
+                page.goto(root_url, wait_until="domcontentloaded")
+                active_pages.append(page)
+                time.sleep(0.25)
+
+            for page in active_pages:
+                _wait_until(lambda page=page: _read_progress_text(page) is None, timeout_s=15.0)
+
+            waiting_page = browser.new_page(viewport={"width": 1280, "height": 900})
+            waiting_page.goto(root_url, wait_until="domcontentloaded")
+
+            queued_pages = []
+            for _ in range(2):
+                page = browser.new_page(viewport={"width": 1280, "height": 900})
+                page.goto(root_url, wait_until="domcontentloaded")
+                queued_pages.append(page)
+                time.sleep(0.25)
+
+            pending_pages = [waiting_page, *queued_pages]
+
+            def _pending_pages_ready():
+                snapshots = [_read_session_wait_overlay_snapshot(page) for page in pending_pages]
+                return all(snapshot["visible"] is True for snapshot in snapshots)
+
+            _wait_until(_pending_pages_ready, timeout_s=10.0)
+
+            for page in pending_pages:
+                overlay_snapshot = _read_session_wait_overlay_snapshot(page)
+                assert overlay_snapshot["present"] is True
+                assert overlay_snapshot["width"] and overlay_snapshot["width"] > 0
+                assert overlay_snapshot["height"] and overlay_snapshot["height"] >= 400
+                assert config.UI_TEXT["progress"]["queue_wait"] in str(overlay_snapshot["content"] or "")
+
+            browser.close()
+    finally:
+        server.should_exit = True
+        thread.join(timeout=10)
+        demo.close()
+
+
+def test_active_user_execute_is_not_blocked_by_waiting_init_loads(monkeypatch):
+    config = importlib.reload(importlib.import_module("config"))
+    state_manager = importlib.reload(importlib.import_module("state_manager"))
+    callbacks = importlib.reload(importlib.import_module("gradio_callbacks"))
+    ui_layout = importlib.reload(importlib.import_module("ui_layout"))
+
+    class _FakeProxy:
+        def __init__(self):
+            self.env_id = None
+            self.episode_idx = None
+            self.language_goal = "goal"
+            self.available_options = [("pick", 0)]
+            self.raw_solve_options = [{"label": "a", "action": "pick", "available": False}]
+            self.demonstration_frames = []
+            self.base_frames = [Image.new("RGB", (8, 8), color=(1, 2, 3))]
+            self.difficulty = None
+            self.seed = None
+
+        def load_episode(self, env_id, episode_idx):
+            self.env_id = env_id
+            self.episode_idx = episode_idx
+            return Image.new("RGB", (32, 32), color=(10, 20, 30)), "loaded"
+
+        def get_pil_image(self, use_segmented=False):
+            _ = use_segmented
+            return Image.new("RGB", (32, 32), color=(10, 20, 30))
+
+        def execute_action(self, option_idx, click_coords):
+            _ = option_idx, click_coords
+            time.sleep(0.5)
+            return Image.new("RGB", (32, 32), color=(30, 40, 50)), "SUCCESS", False
+
+        def update_observation(self, use_segmentation=False):
+            _ = use_segmentation
+            return self.get_pil_image(), ""
+
+        def close(self):
+            return None
+
+    monkeypatch.setattr(state_manager, "ProcessSessionProxy", _FakeProxy)
+    monkeypatch.setattr(
+        callbacks.user_manager,
+        "init_session",
+        lambda uid: (
+            True,
+            "ok",
+            {"current_task": {"env_id": "BinFill", "episode_idx": 1}, "completed_count": 0},
+        ),
+    )
+    monkeypatch.setattr(callbacks, "get_task_hint", lambda env_id: "")
+    monkeypatch.setattr(callbacks, "should_show_demo_video", lambda env_id: False)
+
+    demo = ui_layout.create_ui_blocks()
+    root_url, demo, server, thread = _mount_demo(demo)
+
+    try:
+        with sync_playwright() as p:
+            browser = p.chromium.launch(headless=True)
+
+            page1 = browser.new_page(viewport={"width": 1280, "height": 900})
+            page1.goto(root_url, wait_until="domcontentloaded")
+            _wait_until(lambda: _read_progress_text(page1) is None, timeout_s=15.0)
+
+            page2 = browser.new_page(viewport={"width": 1280, "height": 900})
+            page2.goto(root_url, wait_until="domcontentloaded")
+            _wait_until(lambda: _read_progress_text(page2) is None, timeout_s=15.0)
+
+            waiting_page = browser.new_page(viewport={"width": 1280, "height": 900})
+            waiting_page.goto(root_url, wait_until="domcontentloaded")
+
+            queued_page = browser.new_page(viewport={"width": 1280, "height": 900})
+            queued_page.goto(root_url, wait_until="domcontentloaded")
+
+            _wait_until(lambda: _read_session_wait_overlay_snapshot(waiting_page)["visible"] is True, timeout_s=10.0)
+            _wait_until(lambda: _read_session_wait_overlay_snapshot(queued_page)["visible"] is True, timeout_s=10.0)
+
+            page1.locator("#action_radio input[type='radio']").first.check()
+            page1.locator("#exec_btn button, button#exec_btn").first.click()
+
+            _wait_until(
+                lambda: "SUCCESS" in (_read_log_output_value(page1) or ""),
+                timeout_s=6.0,
+            )
+            page1_progress = _read_progress_text(page1) or ""
+            assert "queue:" not in page1_progress.lower()
+            assert config.UI_TEXT["progress"]["queue_wait"] not in page1_progress
+            assert _read_session_wait_overlay_snapshot(waiting_page)["visible"] is True
+            assert _read_session_wait_overlay_snapshot(queued_page)["visible"] is True
+
+            browser.close()
+    finally:
+        server.should_exit = True
+        thread.join(timeout=10)
+        demo.close()

gradio-web/ui_layout.py

@@ -15,6 +15,8 @@ from config import (
     LIVE_OBS_POINT_WAIT_CLASS,
     SESSION_CONCURRENCY_ID,
     SESSION_CONCURRENCY_LIMIT,
+    SESSION_INIT_CONCURRENCY_ID,
+    SESSION_INIT_CONCURRENCY_LIMIT,
     SESSION_TIMEOUT,
     LIVE_OBS_REFRESH_HZ,
     POINT_SELECTION_SCALE,
@@ -37,11 +39,13 @@ from gradio_callbacks import (
     on_video_end_transition,
     precheck_execute_inputs,
     refresh_live_obs,
+    resume_pending_init,
     restart_episode_wrapper,
     switch_env_wrapper,
     switch_to_action_phase,
     switch_to_execute_phase,
     touch_session,
+    touch_session_or_preserve_pending,
 )
 from user_manager import user_manager
 
@@ -403,6 +407,91 @@ PROGRESS_TEXT_REWRITE_JS = f"""
         }}
     }};
 
+    const resetManualOverlayStyles = (host, markdown, prose) => {{
+        host.style.setProperty("pointer-events", "none", "important");
+        if (markdown instanceof HTMLElement) {{
+            [
+                "position",
+                "inset",
+                "display",
+                "align-items",
+                "justify-content",
+                "padding",
+            ].forEach((prop) => markdown.style.removeProperty(prop));
+        }}
+        if (prose instanceof HTMLElement) {{
+            [
+                "min-width",
+                "max-width",
+                "margin",
+                "padding",
+                "border-radius",
+                "background",
+                "border",
+                "box-shadow",
+                "text-align",
+                "color",
+                "font-size",
+                "font-weight",
+                "line-height",
+                "white-space",
+            ].forEach((prop) => prose.style.removeProperty(prop));
+        }}
+    }};
+
+    const updateManualWaitOverlay = () => {{
+        const host = document.getElementById("native_progress_host");
+        if (!(host instanceof HTMLElement)) {{
+            return;
+        }}
+
+        const markdown = host.querySelector('[data-testid="markdown"]');
+        const prose =
+            markdown instanceof HTMLElement
+                ? markdown.querySelector(".prose, .md") || markdown
+                : null;
+        const progressNode = host.querySelector(".progress-text");
+        const text =
+            prose instanceof HTMLElement
+                ? (prose.innerText || prose.textContent || "").trim()
+                : "";
+        const manualVisible =
+            Boolean(text) &&
+            !progressNode &&
+            text.toLowerCase().includes(queueWaitText.toLowerCase());
+
+        if (!manualVisible) {{
+            resetManualOverlayStyles(host, markdown, prose);
+            return;
+        }}
+
+        host.style.setProperty("pointer-events", "auto", "important");
+        if (markdown instanceof HTMLElement) {{
+            markdown.style.setProperty("position", "fixed", "important");
+            markdown.style.setProperty("inset", "0", "important");
+            markdown.style.setProperty("display", "flex", "important");
+            markdown.style.setProperty("align-items", "center", "important");
+            markdown.style.setProperty("justify-content", "center", "important");
+            markdown.style.setProperty("padding", "24px", "important");
+        }}
+        if (prose instanceof HTMLElement) {{
+            prose.style.setProperty("min-width", "min(560px, calc(100vw - 48px))", "important");
+            prose.style.setProperty("max-width", "calc(100vw - 48px)", "important");
+            prose.style.setProperty("margin", "0", "important");
+            prose.style.setProperty("padding", "28px 32px", "important");
+            prose.style.setProperty("border-radius", "16px", "important");
+            prose.style.setProperty("background", "rgba(255, 255, 255, 0.96)", "important");
+            prose.style.setProperty("border", "1px solid rgba(15, 23, 42, 0.08)", "important");
+            prose.style.setProperty("box-shadow", "0 24px 60px rgba(15, 23, 42, 0.14)", "important");
+            prose.style.setProperty("text-align", "center", "important");
+            prose.style.setProperty("color", "#0f172a", "important");
+            prose.style.setProperty("font-size", "var(--text-lg)", "important");
+            prose.style.setProperty("font-weight", "600", "important");
+            prose.style.setProperty("line-height", "1.5", "important");
+            prose.style.setProperty("white-space", "pre-line", "important");
+        }}
+    }};
+
     const splitSegments = (text) =>
         text
             .split("|")
@@ -461,6 +550,7 @@ PROGRESS_TEXT_REWRITE_JS = f"""
     const rewriteAll = () => {{
         ensureOverlayStyles();
         document.querySelectorAll(".progress-text").forEach(rewriteNode);
+        updateManualWaitOverlay();
     }};
 
     const scheduleRewrite = () => {{
@@ -534,7 +624,6 @@ CSS = f"""
     min-height: 100vh !important;
 }}
 
-
 #reference_action_btn button:not(:disabled),
 button#reference_action_btn:not(:disabled) {{
     background: #1f8b4c !important;
@@ -677,11 +766,13 @@ def _phase_from_updates(main_interface_update, video_phase_update):
 
 def _with_phase_from_load(load_result):
     phase = _phase_from_updates(load_result[1], load_result[14])
-    return (*load_result, phase)
-
-
-def _skip_load_flow():
-    return tuple(gr.skip() for _ in range(20))
+    return (
+        *load_result,
+        phase,
+        False,
+        gr.update(value=""),
+        gr.update(active=False),
+    )
 
 
 def _phase_visibility_updates(phase):
@@ -750,9 +841,11 @@ def create_ui_blocks():
             delete_callback=cleanup_user_session,
         )
         ui_phase_state = gr.State(value=PHASE_INIT)
+        session_boot_pending_state = gr.State(value=False)
         current_task_env_state = gr.State(value=None)
         suppress_next_option_change_state = gr.State(value=False)
         live_obs_timer = gr.Timer(value=1.0 / LIVE_OBS_REFRESH_HZ, active=True)
+        session_init_retry_timer = gr.Timer(value=0.5, active=False)
 
         task_info_box = gr.Textbox(visible=False, elem_id="task_info_box")
         progress_info_box = gr.Textbox(visible=False)
@@ -900,12 +993,65 @@ def create_ui_blocks():
             task_hint_display,
             reference_action_btn,
             ui_phase_state,
+            session_boot_pending_state,
+            native_progress_host,
+            session_init_retry_timer,
         ]
         phase_visibility_outputs = [
             video_phase_group,
             action_phase_group,
             control_panel_group,
         ]
+        action_queue_kwargs = {
+            "concurrency_id": SESSION_CONCURRENCY_ID,
+            "concurrency_limit": SESSION_CONCURRENCY_LIMIT,
+        }
+        init_queue_kwargs = {
+            "concurrency_id": SESSION_INIT_CONCURRENCY_ID,
+            "concurrency_limit": SESSION_INIT_CONCURRENCY_LIMIT,
+        }
+
+        def _skip_load_flow():
+            return tuple(gr.skip() for _ in range(len(load_flow_outputs)))
+
+        def _pending_init_flow(uid, queue_position):
+            _ = queue_position
+            return (
+                uid,
+                gr.update(visible=True),
+                gr.update(interactive=False),
+                "",
+                gr.update(choices=[], value=None),
+                "",
+                "",
+                gr.update(value=None, visible=False),
+                gr.update(visible=False, interactive=False),
+                "",
+                "",
+                gr.update(interactive=False),
+                gr.update(interactive=False),
+                gr.update(interactive=False),
+                gr.update(visible=False),
+                gr.update(visible=False),
+                gr.update(visible=False),
+                gr.update(value=""),
+                gr.update(interactive=False),
+                PHASE_INIT,
+                True,
+                gr.update(value=f'{UI_TEXT["progress"]["queue_wait"]}\\n\\nqueue: {max(1, int(queue_position or 1))}'),
+                gr.update(active=True),
+            )
+
+        def _coerce_init_load_result(result):
+            if isinstance(result, dict):
+                status = result.get("status")
+                if status == "pending":
+                    return _pending_init_flow(result.get("uid"), result.get("queue_position"))
+                if status == "skip":
+                    return _skip_load_flow()
+                if status == "ready":
+                    return _with_phase_from_load(result.get("load_result"))
+            return _with_phase_from_load(result)
 
         def _normalize_env_choice(env_value, choices):
             if env_value is None:
@@ -949,7 +1095,10 @@ def create_ui_blocks():
             )
 
         def init_app_with_phase(request: gr.Request):
-            return _with_phase_from_load(init_app(request))
+            return _coerce_init_load_result(init_app(request))
+
+        def resume_pending_init_with_phase(uid, init_pending, request: gr.Request):
+            return _coerce_init_load_result(resume_pending_init(uid, init_pending, request))
 
         def load_next_task_with_phase(uid):
             return _with_phase_from_load(load_next_task_wrapper(uid))
@@ -994,11 +1143,10 @@ def create_ui_blocks():
             fn=maybe_switch_env_with_phase,
             inputs=[uid_state, header_task_box, current_task_env_state],
             outputs=load_flow_outputs,
-            concurrency_id=SESSION_CONCURRENCY_ID,
-            concurrency_limit=SESSION_CONCURRENCY_LIMIT,
             show_progress="full",
             js=SET_EPISODE_LOAD_MODE_IF_SWITCH_JS,
             show_progress_on=[native_progress_host],
+            **action_queue_kwargs,
         ).then(
             fn=_phase_visibility_updates,
             inputs=[ui_phase_state],
@@ -1035,11 +1183,10 @@ def create_ui_blocks():
             fn=load_next_task_with_phase,
             inputs=[uid_state],
             outputs=load_flow_outputs,
-            concurrency_id=SESSION_CONCURRENCY_ID,
-            concurrency_limit=SESSION_CONCURRENCY_LIMIT,
             show_progress="full",
             js=SET_EPISODE_LOAD_MODE_JS,
             show_progress_on=[native_progress_host],
+            **action_queue_kwargs,
         ).then(
             fn=_phase_visibility_updates,
             inputs=[ui_phase_state],
@@ -1076,11 +1223,10 @@ def create_ui_blocks():
             fn=restart_episode_with_phase,
             inputs=[uid_state],
             outputs=load_flow_outputs,
-            concurrency_id=SESSION_CONCURRENCY_ID,
-            concurrency_limit=SESSION_CONCURRENCY_LIMIT,
             show_progress="full",
             js=SET_EPISODE_LOAD_MODE_JS,
             show_progress_on=[native_progress_host],
+            **action_queue_kwargs,
         ).then(
             fn=_phase_visibility_updates,
             inputs=[ui_phase_state],
@@ -1208,8 +1354,7 @@ def create_ui_blocks():
             fn=on_reference_action,
             inputs=[uid_state, options_radio],
             outputs=[img_display, options_radio, coords_box, log_output, suppress_next_option_change_state],
-            concurrency_id=SESSION_CONCURRENCY_ID,
-            concurrency_limit=SESSION_CONCURRENCY_LIMIT,
+            **action_queue_kwargs,
         ).then(
             fn=touch_session,
             inputs=[uid_state],
@@ -1252,9 +1397,8 @@ def create_ui_blocks():
             fn=execute_step,
             inputs=[uid_state, options_radio, coords_box],
             outputs=[img_display, log_output, task_info_box, progress_info_box, restart_episode_btn, next_task_btn, exec_btn],
-            concurrency_id=SESSION_CONCURRENCY_ID,
-            concurrency_limit=SESSION_CONCURRENCY_LIMIT,
             show_progress="hidden",
+            **action_queue_kwargs,
         ).then(
             fn=switch_to_action_phase,
             inputs=[uid_state],
@@ -1317,11 +1461,10 @@ def create_ui_blocks():
             fn=init_app_with_phase,
             inputs=[],
             outputs=load_flow_outputs,
-            concurrency_id=SESSION_CONCURRENCY_ID,
-            concurrency_limit=SESSION_CONCURRENCY_LIMIT,
             show_progress="full",
             js=SET_EPISODE_LOAD_MODE_JS,
             show_progress_on=[native_progress_host],
+            **init_queue_kwargs,
         ).then(
             fn=_phase_visibility_updates,
             inputs=[ui_phase_state],
@@ -1335,8 +1478,8 @@ def create_ui_blocks():
             queue=False,
             show_progress="hidden",
         ).then(
-            fn=touch_session,
-            inputs=[uid_state],
+            fn=touch_session_or_preserve_pending,
+            inputs=[uid_state, session_boot_pending_state],
             outputs=[uid_state],
             queue=False,
             show_progress="hidden",
@@ -1354,6 +1497,32 @@ def create_ui_blocks():
             show_progress="hidden",
         )
 
+        session_init_retry_timer.tick(
+            fn=resume_pending_init_with_phase,
+            inputs=[uid_state, session_boot_pending_state],
+            outputs=load_flow_outputs,
+            show_progress="hidden",
+            **init_queue_kwargs,
+        ).then(
+            fn=_phase_visibility_updates,
+            inputs=[ui_phase_state],
+            outputs=phase_visibility_outputs,
+            queue=False,
+            show_progress="hidden",
+        ).then(
+            fn=sync_header_from_task,
+            inputs=[task_info_box, goal_box],
+            outputs=[header_task_box, header_goal_box, current_task_env_state],
+            queue=False,
+            show_progress="hidden",
+        ).then(
+            fn=touch_session_or_preserve_pending,
+            inputs=[uid_state, session_boot_pending_state],
+            outputs=[uid_state],
+            queue=False,
+            show_progress="hidden",
+        )
+
         demo.unload(fn=cleanup_current_request_session)
         demo.queue(max_size=None, default_concurrency_limit=None)