CVE-2024-0964 (Python, published 2024-02-05T23:15Z)
Description: value: A local file include could be remotely triggered in Gradio due to a vulnerable user-supplied JSON value in an API request.
Severity: nan (CVSS v3 base score: 9.4)
Weakness: CWE-22 \u2014 Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal')
The product uses external input to construct a pathname that is intended to identify a file or directory that is located underneath a restricted parent directory, but the product does not properly neutralize special elements within the pathname that can cause the pathname to resolve to a location that is outside of the restricted directory.
Repository: https://github.com/gradio-app/gradio
Files changed:
.changeset/ripe-spiders-love.md
client/js/src/client.ts
client/js/src/types.ts
client/python/gradio_client/client.py
client/python/gradio_client/utils.py
client/python/test/test_client.py
gradio/blocks.py
gradio/queueing.py
gradio/routes.py
gradio/utils.py
js/app/test/blocks_chained_events.spec.ts
scripts/benchmark_queue.py
test/test_helpers.py
test/test_queueing.py
Commit message: Fix api event drops (#6556)

* changes

* changes

* add changeset

* changes

* changes

* changes

* changs

* chagnes

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* changes~git push

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* changes

* change

* changes

* changes

* changes

* changes

---------

Co-authored-by: gradio-pr-bot <gradio-pr-bot@users.noreply.github.com>
Co-authored-by: Ubuntu <ubuntu@ip-172-31-25-241.us-west-2.compute.internal>
Co-authored-by: Abubakar Abid <abubakar@huggingface.co>

Vulnerable code:
				let protocol = config.protocol ?? "sse";
					} else {
	protocol?: "sse" | "ws";
        self.sse_url = urllib.parse.urljoin(self.src, utils.SSE_URL)
        self.sse_data_url = urllib.parse.urljoin(self.src, utils.SSE_DATA_URL)
        if auth is not None:
            self._login(auth)
        self.config = self._get_config()
        protocol = self.config.get("protocol")
        endpoint_class = Endpoint if protocol == "sse" else EndpointV3Compatibility
            endpoint_class(self, fn_index, dependency)
        if self.endpoints[inferred_fn_index].protocol in ("ws", "sse"):
    def __init__(self, client: Client, fn_index: int, dependency: dict):
        self.protocol = "sse"
            result = utils.synchronize_async(self._sse_fn, data, hash_data, helper)
    async def _sse_fn(self, data: dict, hash_data: dict, helper: Communicator):
            return await utils.get_pred_from_sse(
                sse_url=self.client.sse_url,
                sse_data_url=self.client.sse_data_url,
                headers=self.client.headers,
                cookies=self.client.cookies,
    def __init__(self, client: Client, fn_index: int, dependency: dict):
from typing import Any, Callable, Optional
SSE_URL = "queue/join"
SSE_DATA_URL = "queue/data"
async def get_pred_from_sse(
    cookies: dict[str, str] | None = None,
            asyncio.create_task(check_for_cancel(helper, cookies)),
                stream_sse(
                    headers=headers,
                    cookies=cookies,
async def check_for_cancel(helper: Communicator, cookies: dict[str, str] | None):
                helper.reset_url, json={"event_id": helper.event_id}, cookies=cookies
async def stream_sse(
    cookies: dict[str, str] | None = None,
            cookies=cookies,
                            cookies=cookies,
            "protocol": "sse",
                # Normally, queue-related async tasks (e.g. continuous events created by `gr.Blocks.load(..., every=interval)`, whose async tasks are started at the `/queue/join` endpoint function)
from gradio.utils import run_coro_in_background, safe_get_lock, set_task_name
        self.message_queue = ThreadQueue()
    def send_message(
        self,
        message_type: str,
        data: dict | None = None,
        final: bool = False,
    ):
        data = {} if data is None else data
        self.message_queue.put_nowait({"msg": message_type, **data})
        if final:
            self.message_queue.put_nowait(None)

    async def get_data(self, timeout=5) -> bool:
        self.send_message("send_data", {"event_id": self._id})
        sleep_interval = 0.05
        wait_time = 0
        while wait_time < timeout and self.alive:
            if self.data is not None:
                break
            await asyncio.sleep(sleep_interval)
            wait_time += sleep_interval
        return self.data is not None

    def attach_data(self, body: PredictBody):
        event_id = body.event_id
        if event_id in self.awaiting_data_events:
            event = self.awaiting_data_events[event_id]
            event.data = body
        else:
            raise ValueError("Event not found", event_id)
                    event.send_message("progress", event.progress.model_dump())
                event.send_message("log", log_message.model_dump())

    def push(self, event: Event) -> int | None:
        """
        Add event to queue, or return None if Queue is full
        Parameters:
            event: Event to add to Queue
        Returns:
            rank of submitted Event
        """
        queue_len = len(self.event_queue)
        if self.max_size is not None and queue_len >= self.max_size:
            return None
        self.event_queue.append(event)
        return queue_len

    async def clean_event(self, event: Event | str) -> None:
        if isinstance(event, str):
            for job_set in self.active_jobs:
                if job_set:
                    for job in job_set:
                        if job._id == event:
                            event = job
                            break
        if isinstance(event, str):
            raise ValueError("Event not found", event)
        event.alive = False
        if event in self.event_queue:
        event.send_message("estimation", estimation.model_dump())
                if not event.data:
                    self.awaiting_data_events[event._id] = event
                    client_awake = await event.get_data()
                    del self.awaiting_data_events[event._id]
                    if not client_awake:
                        await self.clean_event(event)
                        continue
                event.send_message("process_starts")
                    event.send_message(
                        final=True,
                        event.send_message(
                    event.send_message(
                        final=True,
                    event.send_message(
                        final=True,
from gradio.queueing import Estimation, Event
    cancel_tasks,
    run_coro_in_background,
    set_task_name,
                await app.get_blocks()._queue.clean_event(body.event_id)
        @app.get("/queue/join", dependencies=[Depends(login_check)])
        async def queue_join(
            fn_index: int,
            session_hash: str,
            username: str = Depends(get_current_user),
            data: Optional[str] = None,
            if blocks._queue.server_app is None:
                blocks._queue.set_server_app(app)

            event = Event(session_hash, fn_index, request, username)
            if data is not None:
                input_data = json.loads(data)
                event.data = PredictBody(
                    session_hash=session_hash,
                    fn_index=fn_index,
                    data=input_data,
                    request=request,
                )

            # Continuous events are not put in the queue so that they do not
            # occupy the queue's resource as they are expected to run forever
            if blocks.dependencies[event.fn_index].get("every", 0):
                await cancel_tasks({f"{event.session_hash}_{event.fn_index}"})
                await blocks._queue.reset_iterators(event._id)
                blocks._queue.continuous_tasks.append(event)
                task = run_coro_in_background(
                    blocks._queue.process_events, [event], False
                )
                set_task_name(task, event.session_hash, event.fn_index, batch=False)
                app._asyncio_tasks.append(task)
            else:
                rank = blocks._queue.push(event)
                if rank is None:
                    event.send_message("queue_full", final=True)
                else:
                    estimation = blocks._queue.get_estimation()
                    await blocks._queue.send_estimation(event, estimation, rank)
                            await blocks._queue.clean_event(event)
                        if not event.alive:
                            message = event.message_queue.get_nowait()
                            if message is None:  # end of stream marker
                                return
                    await blocks._queue.clean_event(event)
        @app.post("/queue/data", dependencies=[Depends(login_check)])
        async def queue_data(
            blocks = app.get_blocks()
            blocks._queue.attach_data(body)
	let last_iteration;
	page.on("websocket", (ws) => {
		last_iteration = ws.waitForEvent("framereceived", {
			predicate: (event) => {
				return JSON.parse(event.payload as string).msg === "process_completed";
			}
		});
	});

	await last_iteration;
	let complete;
	page.on("websocket", (ws) => {
		complete = ws.waitForEvent("framereceived", {
			predicate: (event) => {
				return JSON.parse(event.payload as string).msg === "process_completed";
			}
		});
	});

	await complete;
	let complete;
	page.on("websocket", (ws) => {
		complete = ws.waitForEvent("framereceived", {
			predicate: (event) => {
				return JSON.parse(event.payload as string).msg === "process_completed";
			}
		});
	});

	await complete;
    host = f"{demo.local_url.replace('http', 'ws')}queue/join"
import httpx
        progress_updates = []
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "GET",
                f"http://localhost:{demo.server_port}/queue/join",
                params={"fn_index": 0, "session_hash": "shdce"},
            ) as response:
                async for line in response.aiter_text():
                    if line.startswith("data:"):
                        msg = json.loads(line[5:])
                    if msg["msg"] == "send_data":
                        event_id = msg["event_id"]
                        req = await client.post(
                            f"http://localhost:{demo.server_port}/queue/data",
                            json={
                                "event_id": event_id,
                                "data": [0],
                                "fn_index": 0,
                            },
                        )
                        if not req.is_success:
                            raise ValueError(
                                f"Could not send payload to endpoint: {req.text}"
                            )
                    if msg["msg"] == "progress":
                        progress_updates.append(msg["progress_data"])
                    if msg["msg"] == "process_completed":
                        break

        assert progress_updates == [
            [
                {
                    "index": None,
                    "length": None,
                    "unit": "steps",
                    "progress": 0.0,
                    "desc": "start",
                }
            ],
            [{"index": 0, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [{"index": 1, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [{"index": 2, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [{"index": 3, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [{"index": 4, "length": 4, "unit": "iter", "progress": None, "desc": None}],
        progress_updates = []
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "GET",
                f"http://localhost:{demo.server_port}/queue/join",
                params={"fn_index": 0, "session_hash": "shdce"},
            ) as response:
                async for line in response.aiter_text():
                    if line.startswith("data:"):
                        msg = json.loads(line[5:])
                    if msg["msg"] == "send_data":
                        event_id = msg["event_id"]
                        req = await client.post(
                            f"http://localhost:{demo.server_port}/queue/data",
                            json={
                                "event_id": event_id,
                                "data": [0],
                                "fn_index": 0,
                            },
                        )
                        if not req.is_success:
                            raise ValueError(
                                f"Could not send payload to endpoint: {req.text}"
                            )
                    if msg["msg"] == "progress":
                        progress_updates.append(msg["progress_data"])
                    if msg["msg"] == "process_completed":
                        break

        assert progress_updates == [
            [
                {
                    "index": None,
                    "length": None,
                    "unit": "steps",
                    "progress": 0.0,
                    "desc": "start",
                }
            ],
            [{"index": 0, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [{"index": 1, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [{"index": 2, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [{"index": 3, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [{"index": 4, "length": 4, "unit": "iter", "progress": None, "desc": None}],
            [
                {
                    "index": 0,
                    "length": 3,
                    "unit": "steps",
                    "progress": None,
                    "desc": "alphabet",
                }
            ],
            [
                {
                    "index": 1,
                    "length": 3,
                    "unit": "steps",
                    "progress": None,
                    "desc": "alphabet",
                }
            ],
            [
                {
                    "index": 2,
                    "length": 3,
                    "unit": "steps",
                    "progress": None,
                    "desc": "alphabet",
                }
            ],
        progress_updates = []
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "GET",
                f"http://localhost:{demo.server_port}/queue/join",
                params={"fn_index": 0, "session_hash": "shdce"},
            ) as response:
                async for line in response.aiter_text():
                    if line.startswith("data:"):
                        msg = json.loads(line[5:])
                    if msg["msg"] == "send_data":
                        event_id = msg["event_id"]
                        req = await client.post(
                            f"http://localhost:{demo.server_port}/queue/data",
                            json={
                                "event_id": event_id,
                                "data": ["abc"],
                                "fn_index": 0,
                            },
                        )
                        if not req.is_success:
                            raise ValueError(
                                f"Could not send payload to endpoint: {req.text}"
                            )
                    if msg["msg"] == "progress":
                        progress_updates.append(msg["progress_data"])
                    if msg["msg"] == "process_completed":
                        break

        assert progress_updates == [
            [
                {
                    "index": 1,
                    "length": 3,
                    "unit": "steps",
                    "progress": None,
                    "desc": None,
                }
            ],
            [
                {
                    "index": 2,
                    "length": 3,
                    "unit": "steps",
                    "progress": None,
                    "desc": None,
                }
            ],
            [
                {
                    "index": 3,
                    "length": 3,
                    "unit": "steps",
                    "progress": None,
                    "desc": None,
                }
            ],
        log_messages = []
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "GET",
                f"http://localhost:{demo.server_port}/queue/join",
                params={"fn_index": 0, "session_hash": "shdce"},
            ) as response:
                async for line in response.aiter_text():
                    if line.startswith("data:"):
                        msg = json.loads(line[5:])
                    if msg["msg"] == "send_data":
                        event_id = msg["event_id"]
                        req = await client.post(
                            f"http://localhost:{demo.server_port}/queue/data",
                            json={
                                "event_id": event_id,
                                "data": ["abc"],
                                "fn_index": 0,
                            },
                        )
                        if not req.is_success:
                            raise ValueError(
                                f"Could not send payload to endpoint: {req.text}"
                            )
                    if msg["msg"] == "log":
                        log_messages.append([msg["log"], msg["level"]])
                    if msg["msg"] == "process_completed":
                        break

        assert log_messages == [
            ["Letter a", "info"],
            ["Letter b", "info"],
            ["Letter c", "info"],
            ["Too short!", "warning"],
        await asyncio.sleep(delay)

        log_messages = []
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "GET",
                f"http://localhost:{demo.server_port}/queue/join",
                params={"fn_index": 0, "session_hash": name},
            ) as response:
                async for line in response.aiter_text():
                    if line.startswith("data:"):
                        msg = json.loads(line[5:])
                    if msg["msg"] == "send_data":
                        event_id = msg["event_id"]
                        req = await client.post(
                            f"http://localhost:{demo.server_port}/queue/data",
                            json={
                                "event_id": event_id,
                                "data": [name],
                                "fn_index": 0,
                            },
                        )
                        if not req.is_success:
                            raise ValueError(
                                f"Could not send payload to endpoint: {req.text}"
                            )
                    if msg["msg"] == "log":
                        log_messages.append(msg["log"])
                    if msg["msg"] == "process_completed":
                        break
        return log_messages
    assert alice_logs == ["Hello Alice"]
    assert bob_logs == ["Hello Bob"]
        demo.launch(prevent_thread_lock=True)

                time.sleep(2)
        time.sleep(1)
            sub_job_3 = client.submit(1, 1, fn_index=1)
            time.sleep(1)

Fixed code:
---
"@gradio/client": patch
"gradio": patch
"gradio_client": patch
---

fix:Fix api event drops
			let stream_open = false;
			let event_stream: EventSource | null = null;
			const event_callbacks: Record<string, () => Promise<void>> = {};
				let protocol = config.protocol ?? "ws";
					} else if (protocol == "sse") {
					} else if (protocol == "sse_v1") {
						fire_event({
							type: "status",
							stage: "pending",
							queue: true,
							endpoint: _endpoint,
							fn_index,
							time: new Date()
						});

						post_data(
							`${http_protocol}//${resolve_root(
								host,
								config.path,
								true
							)}/queue/join?${url_params}`,
							{
								...payload,
								session_hash
							},
							hf_token
						).then(([response, status]) => {
							if (status !== 200) {
								fire_event({
									type: "status",
									stage: "error",
									message: BROKEN_CONNECTION_MSG,
									queue: true,
									endpoint: _endpoint,
									fn_index,
									time: new Date()
								});
							} else {
								event_id = response.event_id as string;
								if (!stream_open) {
									open_stream();
								}

								let callback = async function (_data: object): void {
									const { type, status, data } = handle_message(
										_data,
										last_status[fn_index]
									);

									if (type === "update" && status && !complete) {
										// call 'status' listeners
										fire_event({
											type: "status",
											endpoint: _endpoint,
											fn_index,
											time: new Date(),
											...status
										});
									} else if (type === "complete") {
										complete = status;
									} else if (type === "log") {
										fire_event({
											type: "log",
											log: data.log,
											level: data.level,
											endpoint: _endpoint,
											fn_index
										});
									} else if (type === "generating") {
										fire_event({
											type: "status",
											time: new Date(),
											...status,
											stage: status?.stage!,
											queue: true,
											endpoint: _endpoint,
											fn_index
										});
									}
									if (data) {
										fire_event({
											type: "data",
											time: new Date(),
											data: transform_files
												? transform_output(
														data.data,
														api_info,
														config.root,
														config.root_url
												  )
												: data.data,
											endpoint: _endpoint,
											fn_index
										});

										if (complete) {
											fire_event({
												type: "status",
												time: new Date(),
												...complete,
												stage: status?.stage!,
												queue: true,
												endpoint: _endpoint,
												fn_index
											});
										}
									}

									if (status.stage === "complete" || status.stage === "error") {
										if (event_callbacks[event_id]) {
											delete event_callbacks[event_id];
											if (Object.keys(event_callbacks).length === 0) {
												close_stream();
											}
										}
									}
								};
								event_callbacks[event_id] = callback;
							}
						});
			function open_stream(): void {
				stream_open = true;
				let params = new URLSearchParams({
					session_hash: session_hash
				}).toString();
				let url = new URL(
					`${http_protocol}//${resolve_root(
						host,
						config.path,
						true
					)}/queue/data?${params}`
				);
				event_stream = new EventSource(url);
				event_stream.onmessage = async function (event) {
					let _data = JSON.parse(event.data);
					await event_callbacks[_data.event_id](_data);
				};
			}

			function close_stream(): void {
				stream_open = false;
				event_stream?.close();
			}

	protocol?: "sse_v1" | "sse" | "ws";
    Message,
        if auth is not None:
            self._login(auth)

        self.config = self._get_config()
        self.protocol: str = self.config.get("protocol", "ws")
        self.sse_url = urllib.parse.urljoin(
            self.src, utils.SSE_URL_V0 if self.protocol == "sse" else utils.SSE_URL
        )
        self.sse_data_url = urllib.parse.urljoin(
            self.src,
            utils.SSE_DATA_URL_V0 if self.protocol == "sse" else utils.SSE_DATA_URL,
        )
        endpoint_class = (
            Endpoint if self.protocol.startswith("sse") else EndpointV3Compatibility
        )
            endpoint_class(self, fn_index, dependency, self.protocol)
        self.stream_open = False
        self.streaming_future: Future | None = None
        self.pending_messages_per_event: dict[str, list[Message | None]] = {}
        self.pending_event_ids: set[str] = set()

    async def stream_messages(self) -> None:
        try:
            async with httpx.AsyncClient(timeout=httpx.Timeout(timeout=None)) as client:
                buffer = ""
                async with client.stream(
                    "GET",
                    self.sse_url,
                    params={"session_hash": self.session_hash},
                    headers=self.headers,
                    cookies=self.cookies,
                ) as response:
                    async for line in response.aiter_text():
                        buffer += line
                        while "\n\n" in buffer:
                            message, buffer = buffer.split("\n\n", 1)
                            if message.startswith("data:"):
                                resp = json.loads(message[5:])
                                if resp["msg"] == "heartbeat":
                                    continue
                                elif resp["msg"] == "server_stopped":
                                    for (
                                        pending_messages
                                    ) in self.pending_messages_per_event.values():
                                        pending_messages.append(resp)
                                    return
                                event_id = resp["event_id"]
                                if event_id not in self.pending_messages_per_event:
                                    self.pending_messages_per_event[event_id] = []
                                self.pending_messages_per_event[event_id].append(resp)
                                if resp["msg"] == "process_completed":
                                    self.pending_event_ids.remove(event_id)
                                if len(self.pending_event_ids) == 0:
                                    self.stream_open = False
                                    return
                            elif message == "":
                                continue
                            else:
                                raise ValueError(f"Unexpected SSE line: '{message}'")
        except BaseException as e:
            import traceback

            traceback.print_exc()
            raise e

    async def send_data(self, data, hash_data):
        async with httpx.AsyncClient() as client:
            req = await client.post(
                self.sse_data_url,
                json={**data, **hash_data},
                headers=self.headers,
                cookies=self.cookies,
            )
        req.raise_for_status()
        resp = req.json()
        event_id = resp["event_id"]

        if not self.stream_open:
            self.stream_open = True

            def open_stream():
                return utils.synchronize_async(self.stream_messages)

            def close_stream(_):
                self.stream_open = False
                for _, pending_messages in self.pending_messages_per_event.items():
                    pending_messages.append(None)

            if self.streaming_future is None or self.streaming_future.done():
                self.streaming_future = self.executor.submit(open_stream)
                self.streaming_future.add_done_callback(close_stream)

        return event_id

        if self.endpoints[inferred_fn_index].protocol in ("ws", "sse", "sse_v1"):
    def __init__(
        self, client: Client, fn_index: int, dependency: dict, protocol: str = "sse_v1"
    ):
        self.protocol = protocol
            if self.protocol == "sse":
                result = utils.synchronize_async(
                    self._sse_fn_v0, data, hash_data, helper
                )
            elif self.protocol == "sse_v1":
                event_id = utils.synchronize_async(
                    self.client.send_data, data, hash_data
                )
                self.client.pending_event_ids.add(event_id)
                self.client.pending_messages_per_event[event_id] = []
                result = utils.synchronize_async(self._sse_fn_v1, helper, event_id)
            else:
                raise ValueError(f"Unsupported protocol: {self.protocol}")

    async def _sse_fn_v0(self, data: dict, hash_data: dict, helper: Communicator):
            return await utils.get_pred_from_sse_v0(
                self.client.sse_url,
                self.client.sse_data_url,
                self.client.headers,
                self.client.cookies,
    async def _sse_fn_v1(self, helper: Communicator, event_id: str):
        return await utils.get_pred_from_sse_v1(
            helper,
            self.client.headers,
            self.client.cookies,
            self.client.pending_messages_per_event,
            event_id,
        )

    def __init__(self, client: Client, fn_index: int, dependency: dict, *args):
from typing import Any, Callable, Optional, TypedDict
SSE_URL_V0 = "queue/join"
SSE_DATA_URL_V0 = "queue/data"
SSE_URL = "queue/data"
SSE_DATA_URL = "queue/join"
class Message(TypedDict, total=False):
    msg: str
    output: dict[str, Any]
    event_id: str
    rank: int
    rank_eta: float
    queue_size: int
    success: bool
    progress_data: list[dict]
    log: str
    level: str


    LOG = "LOG"
            "log": Status.LOG,
    log: tuple[str, str] | None = None
async def get_pred_from_sse_v0(
    cookies: dict[str, str] | None,
            asyncio.create_task(check_for_cancel(helper, headers, cookies)),
                stream_sse_v0(
                    headers,
                    cookies,
                )
            ),
        ],
        return_when=asyncio.FIRST_COMPLETED,
    )

    for task in pending:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

    assert len(done) == 1
    for task in done:
        return task.result()


async def get_pred_from_sse_v1(
    helper: Communicator,
    headers: dict[str, str],
    cookies: dict[str, str] | None,
    pending_messages_per_event: dict[str, list[Message | None]],
    event_id: str,
) -> dict[str, Any] | None:
    done, pending = await asyncio.wait(
        [
            asyncio.create_task(check_for_cancel(helper, headers, cookies)),
            asyncio.create_task(
                stream_sse_v1(
                    helper,
                    pending_messages_per_event,
                    event_id,
async def check_for_cancel(
    helper: Communicator, headers: dict[str, str], cookies: dict[str, str] | None
):
                helper.reset_url,
                json={"event_id": helper.event_id},
                headers=headers,
                cookies=cookies,
async def stream_sse_v0(
    cookies: dict[str, str] | None,
            cookies=cookies,
                            cookies=cookies,
async def stream_sse_v1(
    helper: Communicator,
    pending_messages_per_event: dict[str, list[Message | None]],
    event_id: str,
) -> dict[str, Any]:
    try:
        pending_messages = pending_messages_per_event[event_id]

        while True:
            if len(pending_messages) > 0:
                msg = pending_messages.pop(0)
            else:
                await asyncio.sleep(0.05)
                continue

            if msg is None:
                raise CancelledError()

            with helper.lock:
                log_message = None
                if msg["msg"] == "log":
                    log = msg.get("log")
                    level = msg.get("level")
                    if log and level:
                        log_message = (log, level)
                status_update = StatusUpdate(
                    code=Status.msg_to_status(msg["msg"]),
                    queue_size=msg.get("queue_size"),
                    rank=msg.get("rank", None),
                    success=msg.get("success"),
                    time=datetime.now(),
                    eta=msg.get("rank_eta"),
                    progress_data=ProgressUnit.from_msg(msg["progress_data"])
                    if "progress_data" in msg
                    else None,
                    log=log_message,
                )
                output = msg.get("output", {}).get("data", [])
                if output and status_update.code != Status.FINISHED:
                    try:
                        result = helper.prediction_processor(*output)
                    except Exception as e:
                        result = [e]
                    helper.job.outputs.append(result)
                helper.job.latest_status = status_update

            if msg["msg"] == "queue_full":
                raise QueueError("Queue is full! Please try again.")
            elif msg["msg"] == "process_completed":
                del pending_messages_per_event[event_id]
                return msg["output"]
            elif msg["msg"] == "server_stopped":
                raise ValueError("Server stopped.")

    except asyncio.CancelledError:
        raise


    @pytest.mark.flaky
    def test_private_space_v4_sse_v1(self):
        space_id = "gradio-tests/not-actually-private-spacev4-sse-v1"
        api = huggingface_hub.HfApi()
        assert api.space_info(space_id).private
        client = Client(
            space_id,
        )
        output = client.predict("abc", api_name="/predict")
        assert output == "abc"

        if every and concurrency_limit is not None:
            if concurrency_limit == "default":
                concurrency_limit = None
            else:
                raise ValueError(
                    "Cannot set a value for `concurrency_limit` with `every`."
                )
            "protocol": "sse_v1",
                # Normally, queue-related async tasks (e.g. continuous events created by `gr.Blocks.load(..., every=interval)`, whose async tasks are started at the `/queue/data` endpoint function)
from gradio.utils import LRUCache, run_coro_in_background, safe_get_lock, set_task_name
        self.pending_messages_per_session: LRUCache[str, ThreadQueue] = LRUCache(2000)
        self.pending_event_ids_session: dict[str, set[str]] = {}
        self.pending_message_lock = safe_get_lock()
    def send_message(
        self,
        event: Event,
        message_type: str,
        data: dict | None = None,
    ):
        data = {} if data is None else data
        messages = self.pending_messages_per_session[event.session_hash]
        messages.put_nowait({"msg": message_type, "event_id": event._id, **data})

    async def push(
        self, body: PredictBody, request: fastapi.Request, username: str | None
    ):
        if body.session_hash is None:
            raise ValueError("No session hash provided.")
        if body.fn_index is None:
            raise ValueError("No function index provided.")
        queue_len = len(self.event_queue)
        if self.max_size is not None and queue_len >= self.max_size:
            raise ValueError(
                f"Queue is full. Max size is {self.max_size} and current size is {queue_len}."
            )

        event = Event(body.session_hash, body.fn_index, request, username)
        event.data = body
        async with self.pending_message_lock:
            if body.session_hash not in self.pending_messages_per_session:
                self.pending_messages_per_session[body.session_hash] = ThreadQueue()
            if body.session_hash not in self.pending_event_ids_session:
                self.pending_event_ids_session[body.session_hash] = set()
        self.pending_event_ids_session[body.session_hash].add(event._id)
        self.event_queue.append(event)

        estimation = self.get_estimation()
        await self.send_estimation(event, estimation, queue_len)

        return event._id
                    self.send_message(event, "progress", event.progress.model_dump())
                self.send_message(event, "log", log_message.model_dump())

    async def clean_events(
        self, *, session_hash: str | None = None, event_id: str | None = None
    ) -> None:
        for job_set in self.active_jobs:
            if job_set:
                for job in job_set:
                    if job.session_hash == session_hash or job._id == event_id:
                        job.alive = False

        events_to_remove = []
        for event in self.event_queue:
            if event.session_hash == session_hash or event._id == event_id:
                events_to_remove.append(event)

        for event in events_to_remove:
        self.send_message(event, "estimation", estimation.model_dump())
                self.send_message(event, "process_starts")
                    self.send_message(
                        event,
                        self.send_message(
                            event,
                    self.send_message(
                        event,
                    self.send_message(
                        event,
from gradio.queueing import Estimation
                await app.get_blocks()._queue.clean_events(event_id=body.event_id)
        @app.get("/queue/data", dependencies=[Depends(login_check)])
        async def queue_data(
            session_hash: str,
                            await blocks._queue.clean_events(session_hash=session_hash)
                        if (
                            session_hash
                            not in blocks._queue.pending_messages_per_session
                        ):
                            raise HTTPException(
                                status_code=status.HTTP_404_NOT_FOUND,
                                detail="Session not found.",
                            )

                            messages = blocks._queue.pending_messages_per_session[
                                session_hash
                            ]
                            message = messages.get_nowait()
                        if blocks._queue.stopped:
                            message = {"msg": "server_stopped", "success": False}
                            if message["msg"] == "process_completed":
                                blocks._queue.pending_event_ids_session[
                                    session_hash
                                ].remove(message["event_id"])
                                if message["msg"] == "server_stopped" or (
                                    message["msg"] == "process_completed"
                                    and (
                                        len(
                                            blocks._queue.pending_event_ids_session[
                                                session_hash
                                            ]
                                        )
                                        == 0
                                    )
                                ):
                                    return
                    del blocks._queue.pending_messages_per_session[session_hash]
                    await blocks._queue.clean_events(session_hash=session_hash)
        @app.post("/queue/join", dependencies=[Depends(login_check)])
        async def queue_join(
            if blocks._queue.server_app is None:
                blocks._queue.set_server_app(app)

            event_id = await blocks._queue.push(body, request, username)
            return {"event_id": event_id}
from collections import OrderedDict
    Generic,


K = TypeVar("K")
V = TypeVar("V")


class LRUCache(OrderedDict, Generic[K, V]):
    def __init__(self, max_size: int = 100):
        super().__init__()
        self.max_size: int = max_size

    def __setitem__(self, key: K, value: V) -> None:
        if key in self:
            self.move_to_end(key)
        elif len(self) >= self.max_size:
            self.popitem(last=False)
        super().__setitem__(key, value)
    host = f"{demo.local_url.replace('http', 'ws')}queue/data"
import gradio_client as grc
        client = grc.Client(demo.local_url)
        job = client.submit("Gradio")

        status_updates = []
        while not job.done():
            status = job.status()
            update = (
                status.progress_data[0].index if status.progress_data else None,
                status.progress_data[0].desc if status.progress_data else None,
            )
            if update != (None, None) and (
                len(status_updates) == 0 or status_updates[-1] != update
            ):
                status_updates.append(update)
            time.sleep(0.05)

        assert status_updates == [
            (None, "start"),
            (0, None),
            (1, None),
            (2, None),
            (3, None),
            (4, None),
        client = grc.Client(demo.local_url)
        job = client.submit("Gradio")

        status_updates = []
        while not job.done():
            status = job.status()
            update = (
                status.progress_data[0].index if status.progress_data else None,
                status.progress_data[0].desc if status.progress_data else None,
            )
            if update != (None, None) and (
                len(status_updates) == 0 or status_updates[-1] != update
            ):
                status_updates.append(update)
            time.sleep(0.05)

        assert status_updates == [
            (None, "start"),
            (0, None),
            (1, None),
            (2, None),
            (3, None),
            (4, None),
            (0, "alphabet"),
            (1, "alphabet"),
            (2, "alphabet"),
        client = grc.Client(demo.local_url)
        job = client.submit("Gradio")

        status_updates = []
        while not job.done():
            status = job.status()
            update = (
                status.progress_data[0].index if status.progress_data else None,
                status.progress_data[0].unit if status.progress_data else None,
            )
            if update != (None, None) and (
                len(status_updates) == 0 or status_updates[-1] != update
            ):
                status_updates.append(update)
            time.sleep(0.05)

        assert status_updates == [
            (1, "steps"),
            (2, "steps"),
            (3, "steps"),
            (4, "steps"),
            (5, "steps"),
            (6, "steps"),
                time.sleep(0.15)
        client = grc.Client(demo.local_url)
        job = client.submit("Jon")

        status_updates = []
        while not job.done():
            status = job.status()
            update = status.log
            if update is not None and (
                len(status_updates) == 0 or status_updates[-1] != update
            ):
                status_updates.append(update)
            time.sleep(0.05)

        assert status_updates == [
            ("Letter J", "info"),
            ("Letter o", "info"),
            ("Letter n", "info"),
            ("Too short!", "warning"),
        await asyncio.sleep(1)
        time.sleep(1)
        client = grc.Client(demo.local_url)
        job = client.submit(name)

        status_updates = []
        while not job.done():
            status = job.status()
            update = status.log
            if update is not None and (
                len(status_updates) == 0 or status_updates[-1] != update
            ):
                status_updates.append(update)
            time.sleep(0.05)
        return status_updates[-1][0] if status_updates else None
    assert alice_logs == "Hello Alice"
    assert bob_logs == "Hello Bob"
                time.sleep(4)
        time.sleep(2)
            time.sleep(2)