gradio-web/config.py

@@ -22,8 +22,6 @@ UI_GLOBAL_FONT_SIZE = "24px"
 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
@@ -89,6 +87,7 @@ UI_TEXT = {
     "progress": {
         "episode_loading": "The episode is loading...",
         "queue_wait": "Lots of people are playing! Please wait...",
+        "entry_rejected": "too many people playing",
     },
     "errors": {
         "load_missing_task": "Error loading task: missing current_task",

gradio-web/gradio_callbacks.py

@@ -16,7 +16,6 @@ from PIL import Image
 
 from state_manager import (
     cleanup_session,
-    create_session,
     get_execute_count,
     get_play_button_clicked,
     get_session,
@@ -71,18 +70,15 @@ def _session_error_text():
     return _ui_text("log", "session_error")
 
 
+def _entry_rejected_text():
+    return _ui_text("progress", "entry_rejected")
+
+
 def touch_session(uid):
     """Re-emit the current session key to refresh gr.State TTL."""
     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:
@@ -692,8 +688,8 @@ def init_session_and_load_task(uid):
     if not uid:
         return _task_load_failed_response(uid, _session_error_text())
 
-    if get_session(uid) is None:
-        create_session(uid)
+    if get_session(uid) is None and not try_create_session(uid):
+        return _task_load_failed_response(uid, _entry_rejected_text())
 
     return _load_initialized_session_task(uid)
 
@@ -716,7 +712,7 @@ def _load_initialized_session_task(uid):
 
 
 def try_init_session_and_load_task(uid):
-    """Try to initialize the session without blocking on a full slot queue."""
+    """Try to initialize the session and reject immediately when full."""
     if not uid:
         return {
             "status": "ready",
@@ -724,17 +720,18 @@ def try_init_session_and_load_task(uid):
         }
 
     if get_session(uid) is None:
-        ready, queue_position = try_create_session(uid)
+        ready = try_create_session(uid)
         if not ready:
+            message = _entry_rejected_text()
             LOGGER.info(
-                "try_init_session_and_load_task pending uid=%s queue_position=%s",
+                "try_init_session_and_load_task rejected uid=%s",
                 _uid_for_log(uid),
-                queue_position,
             )
             return {
-                "status": "pending",
+                "status": "rejected",
                 "uid": uid,
-                "queue_position": queue_position,
+                "message": message,
+                "load_result": _task_load_failed_response(uid, message),
             }
 
     return {
@@ -743,14 +740,6 @@ def try_init_session_and_load_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:

gradio-web/state_manager.py

@@ -16,7 +16,6 @@ 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}}
@@ -34,7 +33,6 @@ TASK_START_TIMES = {}  # {"{uid}:{env_id}:{episode_idx}": iso_timestamp}
 PLAY_BUTTON_CLICKED = {}  # {uid: bool}
 
 _state_lock = threading.Lock()
-_session_slot_condition = threading.Condition(_state_lock)
 
 
 def get_session(uid):
@@ -43,70 +41,26 @@ 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
+        return True
 
-    _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)
+    if len(ACTIVE_SESSION_SLOTS) >= int(session_concurrency_limit):
         LOGGER.info(
-            "try_reserve_session_slot acquired uid=%s active_slots=%s",
+            "try_reserve_session_slot rejected uid=%s active_slots=%s limit=%s",
             uid,
             len(ACTIVE_SESSION_SLOTS),
+            session_concurrency_limit,
         )
-        return True, 0
+        return False
 
-    queue_position = _ensure_waiting_session_locked(uid)
+    ACTIVE_SESSION_SLOTS.add(uid)
     LOGGER.info(
-        "try_reserve_session_slot queued uid=%s queue_position=%s active_slots=%s limit=%s",
+        "try_reserve_session_slot acquired uid=%s active_slots=%s",
         uid,
-        queue_position,
         len(ACTIVE_SESSION_SLOTS),
-        session_concurrency_limit,
     )
-    return False, queue_position
+    return True
 
 
 def try_reserve_session_slot(uid):
@@ -114,57 +68,29 @@ def try_reserve_session_slot(uid):
     Try to reserve a session slot without blocking.
 
     Returns:
-        tuple[bool, int]: (acquired, queue_position)
+        bool: whether the slot was acquired
     """
     if not uid:
         raise ValueError("Session uid cannot be empty")
 
     from config import SESSION_CONCURRENCY_LIMIT
 
-    with _session_slot_condition:
+    with _state_lock:
         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.
-
-    Slots are held for the full lifetime of a ProcessSessionProxy and released
-    only after cleanup closes the worker process.
-    """
-    if not uid:
-        raise ValueError("Session uid cannot be empty")
-
-    from config import SESSION_CONCURRENCY_LIMIT
-
-    with _session_slot_condition:
-        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 queue_position=%s active_slots=%s",
-                uid,
-                queue_position,
-                len(ACTIVE_SESSION_SLOTS),
-            )
-            _session_slot_condition.wait(timeout=0.1)
-
-
 def release_session_slot(uid):
     if not uid:
         return
 
-    with _session_slot_condition:
+    with _state_lock:
         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,
                 len(ACTIVE_SESSION_SLOTS),
             )
-            _session_slot_condition.notify_all()
 
 
 def try_create_session(uid):
@@ -172,56 +98,48 @@ def try_create_session(uid):
     Try to create a ProcessSessionProxy without blocking on session slot wait.
 
     Returns:
-        tuple[bool, int]: (ready, queue_position)
+        bool: whether the session is ready
     """
     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
+            return True
+        if not _try_reserve_session_slot_locked(uid, _get_session_concurrency_limit()):
+            return False
+        try:
+            GLOBAL_SESSIONS[uid] = ProcessSessionProxy()
+        except Exception:
+            ACTIVE_SESSION_SLOTS.discard(uid)
+            raise
 
     LOGGER.info("try_create_session uid=%s total_sessions=%s", uid, len(GLOBAL_SESSIONS))
-    return True, 0
+    return True
 
 
 def create_session(uid):
     """
     为指定 session key 创建 ProcessSessionProxy。
 
-    只在初始页面 load 时调用，uid 由 request.session_hash 提供。
+    超出并发上限时立即失败，不执行排队等待。
     """
     if not uid:
         raise ValueError("Session uid cannot be empty")
 
-    reserve_session_slot(uid)
-    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
+    ready = try_create_session(uid)
+    if not ready:
+        raise RuntimeError("No session slots available")
     LOGGER.info("create_session uid=%s total_sessions=%s", uid, len(GLOBAL_SESSIONS))
     return uid
 
 
+def _get_session_concurrency_limit():
+    from config import SESSION_CONCURRENCY_LIMIT
+
+    return SESSION_CONCURRENCY_LIMIT
+
+
 def get_task_index(uid):
     """获取任务索引信息。"""
     with _state_lock:
@@ -331,8 +249,6 @@ 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

@@ -185,17 +185,46 @@ def _mount_demo(demo):
     return root_url, demo, server, thread
 
 
-def test_gradio_queue_respects_configured_limit_on_init_load(monkeypatch):
+def test_entry_rejects_immediately_when_session_limit_is_full(monkeypatch):
     config = importlib.reload(importlib.import_module("config"))
-    importlib.reload(importlib.import_module("gradio_callbacks"))
+    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"))
 
-    def fake_init_app(request):
-        uid = str(getattr(request, "session_hash", "missing"))
+    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
+
+    def fake_init_session(uid):
+        _ = uid
         time.sleep(6.0)
-        return _minimal_load_result(uid, log_text=f"ready:{uid}")
+        return (
+            True,
+            "ok",
+            {"current_task": {"env_id": "BinFill", "episode_idx": 1}, "completed_count": 0},
+        )
 
-    monkeypatch.setattr(ui_layout, "init_app", fake_init_app)
+    monkeypatch.setattr(state_manager, "ProcessSessionProxy", _FakeProxy)
+    monkeypatch.setattr(callbacks.user_manager, "init_session", fake_init_session)
+    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)
@@ -204,53 +233,45 @@ 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_INIT_CONCURRENCY_LIMIT) + 2
+            total_pages = int(config.SESSION_CONCURRENCY_LIMIT) + 2
             for _ in range(total_pages):
                 page = browser.new_page(viewport={"width": 1280, "height": 900})
                 page.goto(root_url, wait_until="domcontentloaded")
                 pages.append(page)
                 time.sleep(0.25)
 
-            def _queue_snapshot_ready():
-                progress_texts = [_read_unified_overlay_text(page) for page in pages]
-                processing_ready = all(
-                    text and config.UI_TEXT["progress"]["episode_loading"] in text
-                    for text in progress_texts[: config.SESSION_INIT_CONCURRENCY_LIMIT]
+            def _overlay_snapshot_ready():
+                overlay_texts = [_read_unified_overlay_text(page) for page in pages]
+                rejected_ready = all(
+                    text == config.UI_TEXT["progress"]["entry_rejected"]
+                    for text in overlay_texts[config.SESSION_CONCURRENCY_LIMIT :]
                 )
-                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 processing_ready and queued_ready
+                return rejected_ready
 
-            _wait_until(_queue_snapshot_ready, timeout_s=10.0)
+            _wait_until(_overlay_snapshot_ready, timeout_s=10.0)
             processing_pages = [
                 _read_unified_overlay_text(page) or ""
-                for page in pages[: config.SESSION_INIT_CONCURRENCY_LIMIT]
+                for page in pages[: config.SESSION_CONCURRENCY_LIMIT]
             ]
-            queued_pages = [
+            rejected_pages = [
                 _read_unified_overlay_text(page) or ""
-                for page in pages[config.SESSION_INIT_CONCURRENCY_LIMIT :]
+                for page in pages[config.SESSION_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 all(text != config.UI_TEXT["progress"]["entry_rejected"] for text in processing_pages)
+            assert all(text == config.UI_TEXT["progress"]["entry_rejected"] for text in rejected_pages)
+            assert len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT
+            assert len(state_manager.ACTIVE_SESSION_SLOTS) == config.SESSION_CONCURRENCY_LIMIT
             assert pages[0].evaluate("() => !!document.getElementById('loading_overlay_group')") is False
             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
             assert overlay_snapshot["background"] == "rgba(255, 255, 255, 0.92)"
-            session_wait_snapshot = _read_session_wait_overlay_snapshot(pages[-1])
-            assert session_wait_snapshot["content"] in (None, "")
-
-            _wait_until(lambda: _read_progress_text(pages[0]) is None, timeout_s=15.0)
-            _wait_until(lambda: _read_progress_text(pages[-1]) is None, timeout_s=25.0)
+            rejection_snapshot = _read_session_wait_overlay_snapshot(pages[-1])
+            assert rejection_snapshot["content"] == config.UI_TEXT["progress"]["entry_rejected"]
+            assert _read_unified_overlay_text(pages[-1]) == config.UI_TEXT["progress"]["entry_rejected"]
+            assert len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT
 
             browser.close()
     finally:
@@ -302,6 +323,7 @@ def test_gradio_state_ttl_cleans_up_idle_session(monkeypatch):
             page.goto(root_url, wait_until="domcontentloaded")
             page.wait_for_selector("#main_interface_root", state="visible", timeout=15000)
 
+            _wait_until(lambda: len(state_manager.GLOBAL_SESSIONS) == 1, timeout_s=8.0)
             uid = next(iter(state_manager.GLOBAL_SESSIONS))
             assert uid in user_manager_mod.user_manager.session_progress
 
@@ -341,7 +363,7 @@ def test_single_load_uses_native_episode_loading_copy(monkeypatch):
             page.goto(root_url, wait_until="domcontentloaded")
 
             _wait_until(
-                lambda: (_read_progress_text(page) or "").startswith(config.UI_TEXT["progress"]["episode_loading"]),
+                lambda: _read_progress_overlay_snapshot(page)["present"] is True,
                 timeout_s=8.0,
             )
             assert page.evaluate("() => !!document.getElementById('loading_overlay_group')") is False
@@ -352,7 +374,6 @@ def test_single_load_uses_native_episode_loading_copy(monkeypatch):
             assert overlay_snapshot["background"] == "rgba(255, 255, 255, 0.92)"
 
             page.wait_for_selector("#main_interface_root", state="visible", timeout=15000)
-            _wait_until(lambda: _read_progress_text(page) is None, timeout_s=8.0)
 
             browser.close()
     finally:
@@ -426,7 +447,7 @@ def test_execute_does_not_use_episode_loading_copy(monkeypatch):
 
             body_text = page.evaluate("() => document.body.innerText")
             assert "The episode is loading..." not in body_text
-            assert "Lots of people are playing! Please wait..." not in body_text
+            assert "too many people playing" not in body_text
 
             browser.close()
     finally:
@@ -435,7 +456,7 @@ def test_execute_does_not_use_episode_loading_copy(monkeypatch):
         demo.close()
 
 
-def test_late_user_waits_for_active_session_slot_release(monkeypatch):
+def test_rejected_user_stays_rejected_after_active_session_slot_release(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"))
@@ -508,18 +529,20 @@ def test_late_user_waits_for_active_session_slot_release(monkeypatch):
             assert len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT
             assert len(state_manager.ACTIVE_SESSION_SLOTS) == config.SESSION_CONCURRENCY_LIMIT
             assert _read_session_wait_overlay_snapshot(page3)["visible"] is True
-            wait_snapshot = _read_session_wait_overlay_snapshot(page3)
-            assert wait_snapshot["content"] == f'{config.UI_TEXT["progress"]["queue_wait"]} | queue: 1'
-            assert wait_snapshot["proseBackground"] == "rgba(0, 0, 0, 0)"
-            assert wait_snapshot["proseBorderRadius"] == "0px"
-            assert wait_snapshot["proseBoxShadow"] == "none"
+            reject_snapshot = _read_session_wait_overlay_snapshot(page3)
+            assert reject_snapshot["content"] == config.UI_TEXT["progress"]["entry_rejected"]
+            assert reject_snapshot["proseBackground"] == "rgba(0, 0, 0, 0)"
+            assert reject_snapshot["proseBorderRadius"] == "0px"
+            assert reject_snapshot["proseBoxShadow"] == "none"
 
             page1.close()
 
             _wait_until(lambda: len(closed) >= 1, timeout_s=10.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)
+            _wait_until(lambda: len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT - 1, timeout_s=10.0)
+            _wait_until(lambda: len(state_manager.ACTIVE_SESSION_SLOTS) == config.SESSION_CONCURRENCY_LIMIT - 1, timeout_s=10.0)
+            time.sleep(1.0)
+            assert _read_session_wait_overlay_snapshot(page3)["visible"] is True
+            assert _read_unified_overlay_text(page3) == config.UI_TEXT["progress"]["entry_rejected"]
 
             browser.close()
     finally:
@@ -528,7 +551,7 @@ def test_late_user_waits_for_active_session_slot_release(monkeypatch):
         demo.close()
 
 
-def test_second_and_later_late_users_show_queue_overlay(monkeypatch):
+def test_second_and_later_late_users_show_rejected_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"))
@@ -582,8 +605,11 @@ def test_second_and_later_late_users_show_queue_overlay(monkeypatch):
                 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)
+            _wait_until(
+                lambda: len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT
+                and len(state_manager.ACTIVE_SESSION_SLOTS) == config.SESSION_CONCURRENCY_LIMIT,
+                timeout_s=15.0,
+            )
 
             waiting_page = browser.new_page(viewport={"width": 1280, "height": 900})
             waiting_page.goto(root_url, wait_until="domcontentloaded")
@@ -595,21 +621,21 @@ def test_second_and_later_late_users_show_queue_overlay(monkeypatch):
                 queued_pages.append(page)
                 time.sleep(0.25)
 
-            pending_pages = [waiting_page, *queued_pages]
+                rejected_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)
+                def _rejected_pages_ready():
+                    texts = [_read_unified_overlay_text(page) for page in rejected_pages]
+                    return all(text == config.UI_TEXT["progress"]["entry_rejected"] for text in texts)
 
-            _wait_until(_pending_pages_ready, timeout_s=10.0)
+            _wait_until(_rejected_pages_ready, timeout_s=10.0)
+            assert len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT
 
-            for page in pending_pages:
+            for page in rejected_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 "")
-                assert "queue:" in str(overlay_snapshot["content"] or "").lower()
+                assert str(overlay_snapshot["content"] or "") == config.UI_TEXT["progress"]["entry_rejected"]
                 assert overlay_snapshot["proseBackground"] == "rgba(0, 0, 0, 0)"
                 assert overlay_snapshot["proseBorderRadius"] == "0px"
                 assert overlay_snapshot["proseBoxShadow"] == "none"
@@ -621,7 +647,7 @@ def test_second_and_later_late_users_show_queue_overlay(monkeypatch):
         demo.close()
 
 
-def test_active_user_execute_is_not_blocked_by_waiting_init_loads(monkeypatch):
+def test_active_user_execute_is_not_blocked_by_rejected_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"))
@@ -682,11 +708,17 @@ def test_active_user_execute_is_not_blocked_by_waiting_init_loads(monkeypatch):
 
             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)
+            _wait_until(
+                lambda: (_read_log_output_value(page1) or "").startswith("Please select the action."),
+                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)
+            _wait_until(
+                lambda: (_read_log_output_value(page2) or "").startswith("Please select the action."),
+                timeout_s=15.0,
+            )
 
             waiting_page = browser.new_page(viewport={"width": 1280, "height": 900})
             waiting_page.goto(root_url, wait_until="domcontentloaded")
@@ -705,10 +737,11 @@ def test_active_user_execute_is_not_blocked_by_waiting_init_loads(monkeypatch):
                 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 config.UI_TEXT["progress"]["entry_rejected"] 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
+            assert _read_unified_overlay_text(waiting_page) == config.UI_TEXT["progress"]["entry_rejected"]
+            assert _read_unified_overlay_text(queued_page) == config.UI_TEXT["progress"]["entry_rejected"]
 
             browser.close()
     finally:
@@ -717,7 +750,7 @@ def test_active_user_execute_is_not_blocked_by_waiting_init_loads(monkeypatch):
         demo.close()
 
 
-def test_waiting_page_enters_task_after_idle_session_ttl_release(monkeypatch):
+def test_rejected_page_requires_refresh_after_idle_session_ttl_release(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"))
@@ -773,29 +806,33 @@ def test_waiting_page_enters_task_after_idle_session_ttl_release(monkeypatch):
                 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)
+            _wait_until(
+                lambda: len(state_manager.GLOBAL_SESSIONS) == config.SESSION_CONCURRENCY_LIMIT
+                and len(state_manager.ACTIVE_SESSION_SLOTS) == config.SESSION_CONCURRENCY_LIMIT,
+                timeout_s=15.0,
+            )
 
-            waiting_page = browser.new_page(viewport={"width": 1280, "height": 900})
-            waiting_page.goto(root_url, wait_until="domcontentloaded")
+            rejected_page = browser.new_page(viewport={"width": 1280, "height": 900})
+            rejected_page.goto(root_url, wait_until="domcontentloaded")
 
             _wait_until(
-                lambda: _read_session_wait_overlay_snapshot(waiting_page)["visible"] is True,
+                lambda: _read_session_wait_overlay_snapshot(rejected_page)["visible"] is True,
                 timeout_s=10.0,
             )
             _wait_until(
                 lambda: len(state_manager.ACTIVE_SESSION_SLOTS) < config.SESSION_CONCURRENCY_LIMIT,
                 timeout_s=10.0,
             )
+            time.sleep(1.0)
+            assert _read_session_wait_overlay_snapshot(rejected_page)["visible"] is True
+            assert _read_unified_overlay_text(rejected_page) == config.UI_TEXT["progress"]["entry_rejected"]
+
+            rejected_page.reload(wait_until="domcontentloaded")
             _wait_until(
-                lambda: _read_session_wait_overlay_snapshot(waiting_page)["visible"] is False,
-                timeout_s=15.0,
-            )
-            _wait_until(
-                lambda: (_read_log_output_value(waiting_page) or "").strip() == "Please select the action.\nActions with 🎯 need to select a point on the image as input",
+                lambda: (_read_log_output_value(rejected_page) or "").strip() == "Please select the action.\nActions with 🎯 need to select a point on the image as input",
                 timeout_s=15.0,
             )
-            assert _read_progress_text(waiting_page) is None
+            assert _read_session_wait_overlay_snapshot(rejected_page)["visible"] is False
 
             browser.close()
     finally:

gradio-web/test/test_ui_native_layout_contract.py

@@ -169,7 +169,7 @@ def test_native_ui_config_contains_phase_machine_and_precheck_chain(reload_modul
         )
         assert all("Loading environment, please wait..." not in str(v) for v in values)
         assert "The episode is loading..." in ui_layout.PROGRESS_TEXT_REWRITE_JS
-        assert "Lots of people are playing! Please wait..." in ui_layout.PROGRESS_TEXT_REWRITE_JS
+        assert ui_layout.UI_TEXT["progress"]["entry_rejected"] == "too many people playing"
 
         log_output_comp = next(
             comp

gradio-web/ui_layout.py

@@ -15,8 +15,6 @@ 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,
@@ -39,13 +37,11 @@ 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
 
@@ -715,9 +711,15 @@ def _with_phase_from_load(load_result):
     return (
         *load_result,
         phase,
-        False,
         gr.update(value=""),
-        gr.update(active=False),
+    )
+
+
+def _with_rejected_init(load_result, message):
+    return (
+        *load_result,
+        PHASE_INIT,
+        gr.update(value=message),
     )
 
 
@@ -787,11 +789,9 @@ 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)
@@ -799,7 +799,7 @@ def create_ui_blocks():
 
         with gr.Column(visible=True, elem_id="main_interface_root") as main_interface:
             native_progress_host = gr.Markdown(
-                value="",
+                value=UI_TEXT["progress"]["episode_loading"],
                 visible=True,
                 container=False,
                 elem_id="native_progress_host",
@@ -939,9 +939,7 @@ 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,
@@ -952,51 +950,20 @@ def create_ui_blocks():
             "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_label = f'queue: {max(1, int(queue_position or 1))}'
-            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"]} | {queue_label}'),
-                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"))
+                if status == "rejected":
+                    return _with_rejected_init(
+                        result.get("load_result"),
+                        result.get("message", UI_TEXT["progress"]["entry_rejected"]),
+                    )
             return _with_phase_from_load(result)
 
         def _normalize_env_choice(env_value, choices):
@@ -1043,9 +1010,6 @@ def create_ui_blocks():
         def init_app_with_phase(request: gr.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))
 
@@ -1410,7 +1374,7 @@ def create_ui_blocks():
             show_progress="full",
             js=SET_EPISODE_LOAD_MODE_JS,
             show_progress_on=[native_progress_host],
-            **init_queue_kwargs,
+            queue=False,
         ).then(
             fn=_phase_visibility_updates,
             inputs=[ui_phase_state],
@@ -1424,8 +1388,8 @@ def create_ui_blocks():
             queue=False,
             show_progress="hidden",
         ).then(
-            fn=touch_session_or_preserve_pending,
-            inputs=[uid_state, session_boot_pending_state],
+            fn=touch_session,
+            inputs=[uid_state],
             outputs=[uid_state],
             queue=False,
             show_progress="hidden",
@@ -1443,32 +1407,6 @@ 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)