omnigent-ai--omnigent/tests/e2e_ui/chat/test_queue_steer.py

"""E2E: steering a queued message sends it NOW, mid-turn.

Guards the steer affordance of the client-side queue:

    A first message is sent (and acked), but no ``session.status`` event
    ever follows, so the session's local status stays "streaming"
    (busy). A follow-up typed into the composer is then held in the
    client-side queue — shown in the docked strip, NOT POSTed (the idle
    auto-flush never fires because idle never comes). Clicking the row's
    "Steer" button POSTs it immediately — the only thing that could send
    it here, since the session never went idle.

Why async Playwright (not the sync ``page`` fixture): the route handler
inspects and fulfills every ``/events`` POST to record which messages the
SPA sent and when, across interleaved UI actions (send, queue, steer). It
is a sync test driving the async flow in a fresh thread (see
:func:`_run_in_fresh_loop`) because the suite's many sync
pytest-playwright tests leave the main-thread loop in a state where
pytest-asyncio can't start one.

The route handler fulfills every ``/events`` POST itself, so no real turn
runs and the test needs no working LLM — it asserts purely on when (and
whether) the SPA POSTs the steered message.
"""

from __future__ import annotations

import asyncio
import json
import re
import threading
from collections.abc import Coroutine
from typing import Any

from playwright.async_api import Route, async_playwright

_COMPOSER_PLACEHOLDER = "Ask the agent anything…"
_MSG1 = "sentinel-steer-msg1-2b8d first message, holds the turn open"
_MSG2 = "sentinel-steer-msg2-6f4a queued then steered"

_EVENTS_RE = re.compile(r"/v1/sessions/([^/]+)/events$")


def _run_in_fresh_loop(coro: Coroutine[Any, Any, None]) -> None:
    """Run *coro* to completion in a dedicated thread with its own event loop.

    The e2e_ui suite runs many pytest-playwright **sync** tests in the same
    session; once one has run, pytest-asyncio can't start a loop on the main
    thread. Running the coroutine from a fresh thread via :func:`asyncio.run`
    sidesteps that. Any exception is captured and re-raised on the calling
    thread so the test fails normally.

    :param coro: The coroutine to run to completion.
    :raises BaseException: Whatever the coroutine raised, re-raised here.
    """
    captured: dict[str, BaseException] = {}

    def _worker() -> None:
        try:
            asyncio.run(coro)
        except BaseException as exc:
            captured["error"] = exc

    thread = threading.Thread(target=_worker)
    thread.start()
    thread.join()
    if "error" in captured:
        raise captured["error"]


async def _wait_until(predicate, *, timeout_s: float = 15.0) -> None:
    """Poll ``predicate`` on the event loop until true or timeout.

    :param predicate: Zero-arg callable returning truthy when satisfied.
    :param timeout_s: Max seconds to wait before failing the test.
    :raises AssertionError: If the predicate never becomes truthy.
    """
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout_s
    while loop.time() < deadline:
        if predicate():
            return
        await asyncio.sleep(0.05)
    raise AssertionError(f"condition not met within {timeout_s:.0f}s")


def test_steer_sends_queued_message_while_busy(
    seeded_session: tuple[str, str],
) -> None:
    """Steering a queued message POSTs it immediately, mid-turn.

    Failure mode this catches: steer does nothing (message stays queued),
    or the message is only sent after the turn ends (indistinguishable
    from auto-flush) — either means the steer path is broken.
    """
    base_url, session_id = seeded_session
    _run_in_fresh_loop(_drive_steer(base_url, session_id))


async def _drive_steer(base_url: str, session_id: str) -> None:
    """Async body of the steer test. See the test docstring.

    :param base_url: Spawned server base URL.
    :param session_id: The seeded, runner-bound session.
    """
    async with async_playwright() as pw:
        browser = await pw.chromium.launch()
        page = await browser.new_page()
        try:
            # Every (session_id, text) POSTed to a /events endpoint. Each is
            # acked immediately; no session.status event ever follows, so the
            # session's local status stays "streaming" (busy) after msg1 —
            # which is what makes the follow-up queue instead of send.
            event_posts: list[tuple[str, str]] = []

            async def handle_events(route: Route) -> None:
                request = route.request
                match = _EVENTS_RE.search(request.url)
                assert match is not None, f"unexpected /events url: {request.url}"
                sid = match.group(1)
                body = request.post_data_json
                text = body["data"]["content"][0]["text"]
                event_posts.append((sid, text))
                await route.fulfill(
                    status=200,
                    content_type="application/json",
                    body=json.dumps({"queued": True, "item_id": "ci_e2e"}),
                )

            await page.route("**/v1/sessions/*/events", handle_events)

            await page.goto(f"{base_url}/c/{session_id}")
            composer = page.get_by_label("Message the agent")
            await page.get_by_placeholder(_COMPOSER_PLACEHOLDER).wait_for(
                state="visible", timeout=15_000
            )
            send_button = page.get_by_role("button", name="Send", exact=True)

            # msg1 → POST + acked; the send flips local status to streaming and
            # no idle event arrives, so the session stays busy.
            await composer.fill(_MSG1)
            await send_button.click()
            await _wait_until(lambda: any(text == _MSG1 for _, text in event_posts))

            # msg2 → typed while busy → held in the client-side queue, shown in
            # the docked strip, NOT POSTed (auto-flush only fires on idle, which
            # never comes here).
            await composer.fill(_MSG2)
            await send_button.click()
            await page.get_by_test_id("composer-queued-strip").wait_for(
                state="visible", timeout=15_000
            )
            assert all(text != _MSG2 for _, text in event_posts), (
                f"msg2 was POSTed before steer (should be held client-side): {event_posts}"
            )

            # Steer msg2 → it must POST now, even though the session never went
            # idle. This is what distinguishes steer from the idle auto-flush:
            # the only reason msg2 could POST here is the explicit steer.
            await page.get_by_role("button", name="Send queued message now").click()
            await _wait_until(lambda: any(text == _MSG2 for _, text in event_posts))

            # The steered message left the queue (strip empties).
            await page.get_by_test_id("composer-queued-strip").wait_for(
                state="hidden", timeout=15_000
            )
        finally:
            await browser.close()