tests_python/test_gradio_app.py

from arena.codebase_archive import LocalCodebaseArchiveStore
from arena.gradio_app import GradioArena, build_app
from arena.model_catalog import DEFAULT_LOCAL_MODEL, ProviderModel
from arena.run_store import InMemoryRunStore
from arena.service import ArenaService
from arena.swarm_runtime import SwarmRuntime
from arena.validator_graph import DeterministicValidatorExecutor, ValidatorGraph

from conftest import make_local_session


def make_service():
    store = LocalCodebaseArchiveStore()
    return ArenaService(
        session_factory=make_local_session,
        run_store=InMemoryRunStore(),
        swarm_runtime=SwarmRuntime(
            session_factory=make_local_session,
            archive_store=store,
        ),
        validator_graph=ValidatorGraph(
            executor=DeterministicValidatorExecutor(),
            rubric_enabled=False,
        ),
        archive_store=store,
    )


def create_local_run(arena: GradioArena, prompt: str, criteria: str = "", tests: str = ""):
    return arena.create_run(prompt, criteria, tests, "local", DEFAULT_LOCAL_MODEL, "", "")


def test_gradio_local_provider_runs_inside_create_callback() -> None:
    service = make_service()
    calls: list[str] = []
    original_create_run = service.create_run
    original_start_run = service.start_run

    def create_run(request):
        calls.append("create")
        return original_create_run(request)

    def start_run(request):
        calls.append("start")
        return original_start_run(request)

    service.create_run = create_run
    service.start_run = start_run
    arena = GradioArena(service)

    create_local_run(arena, "Build a CLI", tests="echo ok")

    assert calls == ["create"]


def test_gradio_create_run_returns_state_and_summary() -> None:
    arena = GradioArena(make_service())

    state, status_html, run_json, tasks, events, token, download, _, chat = create_local_run(
        arena,
        "Build a small FastAPI app",
        "pytest must pass\ninclude README",
        "echo ok",
    )

    assert state.role == "run"
    assert state.run_id == run_json["id"]
    assert isinstance(status_html, str)
    assert run_json["prompt"] == "Build a small FastAPI app"
    assert [item["text"] for item in run_json["criteria"]] == ["pytest must pass", "include README"]
    assert run_json["user_tests"] == ["echo ok"]
    assert run_json["status"] in ("working", "validating", "completed")
    assert isinstance(run_json.get("codebase_zip_url", ""), str)
    assert token == ""
    assert state.token == ""
    assert "Build a small FastAPI app" in chat


def test_gradio_create_run_accepts_empty_model_dropdown_value() -> None:
    arena = GradioArena(make_service())

    state, _, run_json, *_ = create_local_run(
        arena,
        "Build a small Gradio demo",
        "Works as a Gradio app",
        "python -m py_compile app.py",
    )

    assert state.role == "run"
    assert run_json["prompt"] == "Build a small Gradio demo"


def test_gradio_hosted_provider_requires_user_api_key() -> None:
    arena = GradioArena(make_service())

    state, status_html, run_json, *rest = arena.create_run(
        "Build a small Gradio demo",
        "Works as a Gradio app",
        "python -m py_compile app.py",
        "openrouter",
        "qwen/qwen3.5-9b",
        "",
        "",
    )

    assert state.role == ""
    assert "enter your own API key" in status_html
    assert "enter your own API key" in rest[4]
    assert run_json == {}


def test_gradio_refresh_models_accepts_empty_optional_inputs() -> None:
    class FakeService:
        def refresh_provider_models(self, config):
            assert config.provider == "openrouter"
            assert config.key_value() == ""
            assert config.base_url == ""
            return [
                ProviderModel(
                    id="nvidia/nemotron-3.5-content-safety:free",
                    label="NVIDIA: Nemotron 3.5 Content Safety",
                    provider="openrouter",
                    parameter_count_b=4.0,
                    eligible=True,
                ),
                ProviderModel(
                    id="too-large/model-70b",
                    label="Too Large 70B",
                    provider="openrouter",
                    parameter_count_b=70.0,
                    eligible=False,
                    status="too_large",
                ),
            ]

    arena = GradioArena(FakeService())

    dropdown, models = arena.refresh_models("openrouter", None, None)

    assert dropdown.choices == [
        ("nvidia/nemotron-3.5-content-safety:free", "nvidia/nemotron-3.5-content-safety:free")
    ]
    assert dropdown.value == "nvidia/nemotron-3.5-content-safety:free"
    assert [model["id"] for model in models] == ["nvidia/nemotron-3.5-content-safety:free"]
    assert models[0]["parameter_count_b"] == 4.0


def test_gradio_send_message_records_followup_in_chat() -> None:
    arena = GradioArena(make_service())
    state, *_ = create_local_run(arena, "Build a CLI")

    _, _, run_json, _, events, _, _, _, chat = arena.send_message(state, "Can you keep working on this?")

    assert run_json["events"][-1]["message"] == "User: Can you keep working on this?"
    assert events[-1][1] == "User: Can you keep working on this?"
    assert "Can you keep working on this?" in chat
    assert 'class="chat-bubble is-user"' in chat


def test_gradio_feedback_records_mom_test_event() -> None:
    arena = GradioArena(make_service())
    state, *_ = create_local_run(arena, "Build a CLI")

    _, _, run_json, _, events, _, _, _, chat = arena.record_feedback(
        state,
        "Mom",
        "I would use this with buyers.",
        "scale",
        "Needs Vietnamese copy.",
    )

    assert "Mom Test feedback" in run_json["events"][-1]["message"]
    assert "Decision: scale" in run_json["events"][-1]["message"]
    assert "I would use this with buyers." in events[-1][1]
    assert "Needs Vietnamese copy." in chat


def test_gradio_refresh_shows_current_status() -> None:
    arena = GradioArena(make_service())
    state, *_ = create_local_run(arena, "Build a CLI", tests="echo ok")

    _, status_html, run_json, *_ = arena.refresh(state)

    assert run_json["id"] == state.run_id
    assert run_json["status"] in ("working", "validating", "completed")


def test_gradio_refresh_shows_completed_tasks_and_events() -> None:
    """Bug E: After a run completes, the tasks and events dataframes must show the run's data."""
    import time

    service = make_service()
    arena = GradioArena(service)
    state, *_ = create_local_run(arena, "Build a CLI", tests="test -f README.md")

    for _ in range(60):
        run = service.get_run(state.run_id)
        if run.status == "completed":
            break
        time.sleep(0.5)

    _, _, _, tasks, events, *_ = arena.refresh(state)

    assert len(tasks) >= 1, f"Expected at least 1 task, got {tasks}"
    assert len(events) >= 1, f"Expected at least 1 event, got {events}"


def test_gradio_status_badge_renders_html() -> None:
    arena = GradioArena(make_service())
    state, status_html, *_ = create_local_run(arena, "Build a CLI", tests="echo ok")

    assert "<span" in status_html
    assert _extract_status_from_html(status_html) in ("working", "validating", "completed")


def test_gradio_app_constructs_without_crashing() -> None:
    app = build_app(make_service())

    assert type(app).__name__ == "Blocks"
    assert app.title == "Backyard Demo Builder"


def test_gradio_defaults_target_backyard_demo_builder() -> None:
    app = build_app(make_service())
    values = "\n".join(
        str(getattr(block, "value", "") or "")
        for block in app.blocks.values()
        if hasattr(block, "value")
    )

    assert "real-estate agent" in values
    assert "handoff_spec.md" in values
    assert "field_notes.md" in values
    assert "Backyard Demo Builder" in values
    assert "Record feedback" in values or any(
        getattr(block, "value", None) == "Record feedback"
        for block in app.blocks.values()
    )


def test_gradio_two_consecutive_runs_both_complete() -> None:
    """Bug D: A second run on the same service must complete, not hang."""
    import time

    service = make_service()
    arena = GradioArena(service)

    state1, *_ = create_local_run(arena, "Build a CLI", tests="test -f README.md")
    for _ in range(60):
        if service.get_run(state1.run_id).status == "completed":
            break
        time.sleep(0.5)
    assert service.get_run(state1.run_id).status == "completed", "First run did not complete"

    state2, *_ = create_local_run(arena, "Build another CLI", tests="test -f README.md")
    for _ in range(60):
        if service.get_run(state2.run_id).status == "completed":
            break
        time.sleep(0.5)
    assert service.get_run(state2.run_id).status == "completed", "Second run did not complete"


def test_gradio_outputs_tuple_has_nine_distinct_outputs() -> None:
    """Bug F: The 9-tuple outputs list must bind to 9 distinct Gradio components.

    Before the fix, position 1 (status bar) and position 7 (status panel with summary)
    were both bound to the same gr.HTML element. The status bar was dead code because
    the full status panel overwrote it.
    """
    app = build_app(make_service())

    for fn in app.fns.values():
        if not fn.outputs or len(fn.outputs) != 9:
            continue
        ids = [o.elem_id if hasattr(o, "elem_id") and o.elem_id else id(o) for o in fn.outputs]
        assert len(ids) == len(set(ids)), (
            f"Duplicate output binding detected in {fn.name}: {ids}. "
            "Each output slot must bind to a distinct Gradio component."
        )


def test_gradio_does_not_poll_refresh_while_idle() -> None:
    """Idle timer refreshes compete with manual model refresh in Gradio's queue."""
    app = build_app(make_service())

    assert all(fn.name != "refresh_1" for fn in app.fns.values())


def test_css_has_mobile_breakpoint_with_hero_scaling() -> None:
    """Bug G/H: The mobile media query must scale the hero H1 and rebalance the top bar."""
    from arena.gradio_app import CSS

    assert "@media" in CSS, "CSS must include responsive media queries for mobile"
    assert "max-width: 880px" in CSS, "CSS must include the 880px mobile breakpoint"

    start = CSS.index("@media (max-width: 880px)")
    rest = CSS[start + 1:]
    next_media = rest.find("@media")
    mobile_block = rest[:next_media] if next_media != -1 else rest

    assert "hero-title" in mobile_block, (
        "Mobile media query must include hero-title rules to prevent H1 wrap on small viewports"
    )
    assert "wordmark" in mobile_block, (
        "Mobile media query must rebalance the top bar wordmark so it fits on one line"
    )


def test_gradio_workbench_children_live_inside_workbench_wrapper() -> None:
    """Bug I: The workbench Row must render inside the .workbench wrapper, not as a sibling.

    Gradio 6 renders gr.Row as a sibling of raw HTML elements, not nested.
    If we wrap a gr.Row in custom <div class="workbench">…</div>, the wrapper
    collapses to 1px and the form/dispatch land outside it. That 1px wrapper
    shows up as a stray horizontal line in the middle of the page.
    """
    app = build_app(make_service())

    workbench = None
    for block in app.blocks.values():
        if getattr(block, "elem_id", None) is None and "workbench" in (getattr(block, "elem_classes", []) or []):
            workbench = block
            break
        if getattr(block, "elem_id", None) == "run-shell":
            pass

    app_html = app.fns
    assert app_html, "App must have registered event handlers"

    blocks_list = list(app.blocks.values())
    workbench_blocks = [b for b in blocks_list if "workbench" in (getattr(b, "elem_classes", []) or [])]
    assert workbench_blocks, "App must have at least one workbench-classed block"

    workbench_block = workbench_blocks[0]
    assert workbench_block is not None
    assert workbench_block.elem_classes and "workbench" in workbench_block.elem_classes
    assert workbench_block.__class__.__name__ in ("Row", "Group", "Column"), (
        f"Workbench wrapper must be a Gradio Row/Group/Column so the form and dispatch "
        f"land inside it, not a raw HTML block. Got type={workbench_block.__class__.__name__!r}."
    )


def test_gradio_run_state_rail_uses_dynamic_status_panel() -> None:
    """The rail must not keep stale static 'No active run' text after a run starts."""
    app = build_app(make_service())

    html = "\n".join(
        getattr(block, "value", "") or ""
        for block in app.blocks.values()
        if block.__class__.__name__ == "HTML"
    )
    status_panels = [
        block for block in app.blocks.values()
        if getattr(block, "elem_id", None) == "status-panel"
    ]

    assert "No active run. Start one to populate telemetry." not in html
    assert len(status_panels) == 1


def test_css_fills_full_viewport_no_centering_black_hole() -> None:
    """Bug J: The .gradio-container must not cap the page width below 100%.

    At 1920x1200 the page should fill the frame. A max-width: 1280px
    rule leaves a 320px black bar on each side that the user calls
    'a black hole on the right panel' — uncomfortable to look at.
    """
    from arena.gradio_app import CSS

    container_rule = next(
        (rule for rule in CSS.split("}") if ".gradio-container" in rule and "max-width" in rule),
        None,
    )
    assert container_rule is not None, "Expected an explicit .gradio-container max-width rule"

    import re
    px_match = re.search(r"max-width:\s*(\d+)px", container_rule)
    pct_match = re.search(r"max-width:\s*100%", container_rule)

    if pct_match:
        return

    assert px_match, f"Could not parse max-width from rule: {container_rule!r}"

    px = int(px_match.group(1))
    assert px >= 1800, (
        f"gradio-container max-width={px}px is too narrow for 1920x1200 frames. "
        "Bump to >= 1800 or remove the constraint so the page fills the viewport."
    )


def test_hero_splits_into_main_and_side_rail() -> None:
    """The hero must split into a main column (manifesto) and a side rail
    (numbered marks) on wide viewports so the 1920x1200 frame has visual
    density on the right side, not a black hole.
    """
    app = build_app(make_service())
    html_blocks = [
        b for b in app.blocks.values()
        if b.__class__.__name__ == "HTML" and "hero" in (getattr(b, "value", "") or "")
    ]
    assert html_blocks, "App must include at least one hero HTML block"
    html = html_blocks[0].value
    assert "hero-main" in html, "Hero must have a .hero-main column for the manifesto"
    assert "hero-side" in html, "Hero must have a .hero-side column with numbered marks"
    assert "mark-num" in html, "Hero side rail must include numbered marks (01/02/03/04)"
    assert "mark-text" in html, "Hero side rail must include mark descriptions"


def test_workbench_has_three_columns_at_wide_viewport() -> None:
    """On >= 1100px, the workbench must render a 3-column grid: form / dispatch / rail.

    A 2-column layout leaves the right half of a 1920x1200 frame empty
    once the form finishes growing — the black-hole symptom. The third
    column (telemetry, keyboard hints, run state) is the fix.
    """
    from arena.gradio_app import CSS

    root_block = next(
        (block for block in CSS.split("}") if ":root" in block and "--workbench-cols:" in block),
        None,
    )
    assert root_block is not None, "Expected --workbench-cols variable in :root"

    import re
    match = re.search(r"--workbench-cols:\s*([^;]+);", root_block)
    assert match, f"Could not parse --workbench-cols from: {root_block!r}"
    cols = match.group(1).strip()
    track_count = len([t for t in re.split(r"\s+(?![^()]*\))", cols) if t.strip()])
    assert track_count == 3, (
        f"--workbench-cols must define 3 tracks, got {track_count}: {cols!r}"
    )


def _extract_status_from_html(html: str) -> str:
    import re
    match = re.search(r">(\w+)</span>", html)
    return match.group(1) if match else ""