"""Run a module's async tests concurrently on one event loop, with correct per-test isolation.

A thin layer over the `pytest-asyncio-concurrent` plugin, which supplies the hard part:
collecting a *group*'s async tests and running them together via `asyncio.gather` (so a slow,
I/O-bound module like `test_agent_session.py` finishes in a fraction of the wall time). What the
plugin does *not* give us, and this module adds:

1. Per-task output capture. pytest's stdout/stderr/log/caplog capture is process-global, so under
   concurrency output leaks or is misattributed to whichever test finishes first. We key capture
   on a `contextvars.ContextVar` instead: each gathered test runs in its own task/context, so the
   stream proxies, root-log handler, and `caplog` records route to the right test's buffer and
   surface only when that test fails -- as normal.

2. Per-test reporting + a live progress line. The plugin reports nothing until the whole group is
   done (and double-prints names under `-v`); we report each test the moment it finishes, and on
   a non-verbose color TTY draw a transient one-cell-per-test line with a blue head orbiting the
   still-running cells, each cell freezing to its status letter on completion.

3. Per-test leaked-task attribution. On a shared loop, diffing `asyncio.all_tasks()` can't tell
   whose task is whose. A task factory tags each task with its creating test's nodeid (via a
   contextvar, inherited by spawned tasks); `fail_on_leaked_tasks` then inspects only its own
   still-pending tasks. See :func:`owned_pending_tasks`.

4. Concurrency control, via two markers and a switch:

   - `@pytest.mark.concurrent`    -- always run this async test concurrently (unless off).
   - `@pytest.mark.no_concurrent` -- never (process-global state, signal handlers, capsys, ...);
     honored even under `--concurrent`.
   - `--concurrent` (or `LK_TEST_CONCURRENCY=all`) runs *every* async test concurrently except
     `no_concurrent` ones; the default runs only `concurrent` ones; `--no-concurrent` (or
     `LK_TEST_CONCURRENCY=0`) forces everything sequential.

   Markers apply at test, class, or module level (`pytestmark = ...`); `capsys`/`capfd` users
   are auto-excluded (process-global fixtures).

5. The fixes that make pytest-asyncio's `auto` mode cooperate:

   - :func:`repromote_collected` restores the concurrent members after auto mode claims (and would
     sequentialize) every async test.
   - One concurrent group per *module*, so module-level tests and every class in the file share a
     loop (the plugin otherwise pins a group to its first member's parent).
   - The gather's loop is captured *after* per-member setup: in auto mode pytest-asyncio builds its
     own loop during fixture setup, and capturing before it crashes with "future belongs to a
     different loop".

Loaded as a *global* plugin via `-p tests.concurrency` (see pyproject.toml), not conftest.py: the
plugin drives `pytest_runtest_protocol_async_group` through the rootdir-scoped `session.ihook`,
which would not see hooks defined in a subdirectory conftest.

Limitations:
- Module-scoped groups break *class*-scoped fixtures on concurrent members (the finalizer needs a
  Class node that module grouping drops); function/module/session scopes are fine.
- The shared loop couples timing: a test that hogs it can perturb another's. If a concurrent run
  gets flaky, force it sequential with `--no-concurrent`.
- pytest-xdist distributes by item and would split a group across workers, so `-n` disables
  concurrency entirely: :func:`pytest_configure` unregisters the underlying plugin (whose
  runtestloop wrapper and collection regrouping otherwise desync xdist's per-worker item indexing,
  crashing the worker with an IndexError), and the default mode falls back to sequential with a
  one-line notice at session start (see :func:`pytest_report_header`); an explicit `--concurrent`
  under `-n` is rejected.
"""

from __future__ import annotations

import asyncio
import contextlib
import contextvars
import inspect
import io
import logging
import math
import os
import sys
import threading
import warnings
import weakref
from typing import TYPE_CHECKING, Any, Literal

import pytest

try:
    from pytest_asyncio_concurrent.grouping import (
        AsyncioConcurrentGroup,
        AsyncioConcurrentGroupMember,
        PytestAsyncioConcurrentGroupingWarning,
    )
    from pytest_asyncio_concurrent.plugin import (
        _call_and_report,
        _call_runtest_async,
        _check_interactive_exception,
        _setup_child,
        _teardown_child,
    )

    PYTEST_ASYNCIO_CONCURRENT_INSTALLED = True
except ImportError:
    PYTEST_ASYNCIO_CONCURRENT_INSTALLED = False

if TYPE_CHECKING:
    from _pytest.config.argparsing import Parser
    from pluggy import Result

# `caplog` reads its handler/records out of these item-stash keys. We populate them per
# test ourselves (pytest's logging plugin never runs for concurrent members). Guarded so a
# pytest internals change degrades gracefully to "no caplog support" instead of a hard error.
try:
    from _pytest.logging import (
        LogCaptureHandler,
        caplog_handler_key,
        caplog_records_key,
    )

    _CAPLOG_SUPPORTED = True
except Exception:  # pragma: no cover - defensive
    LogCaptureHandler = None  # type: ignore[assignment,misc]
    _CAPLOG_SUPPORTED = False

_ENV_TOGGLE = "LK_TEST_CONCURRENCY"
_DEFAULT_LOG_FORMAT = "%(levelname)-8s %(name)s:%(filename)s:%(lineno)s %(message)s"
_CAPTURE_FIXTURES = frozenset({"capsys", "capfd", "capsysbinary", "capfdbinary", "capteesys"})


def register_markers(config: pytest.Config) -> None:
    config.addinivalue_line(
        "markers", "concurrent: run this async test concurrently with its peers."
    )
    config.addinivalue_line(
        "markers",
        "no_concurrent: never run this async test concurrently (it relies on process-global "
        "state, signal handlers, capsys, etc.).",
    )


def add_toggle_option(parser: Parser) -> None:
    group = parser.getgroup("concurrency", "concurrent async test execution")
    group.addoption(
        "--concurrent",
        action="store_true",
        default=False,
        help="Run every async test concurrently, except @pytest.mark.no_concurrent ones.",
    )
    group.addoption(
        "--no-concurrent",
        action="store_true",
        default=False,
        help="Run all tests sequentially, even @pytest.mark.concurrent ones.",
    )


def _xdist_active(config: pytest.Config) -> bool:
    """True when pytest-xdist is in play -- on the controller *and* inside each worker.

    xdist splits a run by *item*, which would scatter a concurrent group (one shared event loop)
    across workers. The controller carries `-n` / `--dist`; a worker process carries neither
    (xdist clears them so the worker doesn't itself re-distribute) but does carry `workerinput`.
    The concurrent plugin must be disabled on the worker too -- that's where its runtestloop
    wrapper fights xdist's remote loop (`items[nextitem_index]` -> IndexError). `-n0` / no
    xdist leaves this False everywhere.
    """
    if hasattr(config, "workerinput"):
        return True
    if getattr(config.option, "dist", "no") not in ("no", None):
        return True
    return bool(getattr(config.option, "numprocesses", None))


def _requested_mode(config: pytest.Config) -> Literal["default", "all", "off"]:
    """Run mode the user asked for via flags/env, before the xdist override: off/default/all."""
    env = os.environ.get(_ENV_TOGGLE, "").strip().lower()
    if config.getoption("--no-concurrent", default=False) or env in ("0", "off", "false", "no"):
        return "off"
    if config.getoption("--concurrent", default=False) or env in ("all", "1", "on", "yes"):
        return "all"
    return "default"


def concurrency_mode(config: pytest.Config) -> Literal["default", "all", "off"]:
    """Resolve the effective run mode.

    Under pytest-xdist (`-n`) concurrency can't work -- a group shares one event loop and xdist
    would split it across workers -- so any non-off request is forced to `"off"` here. The
    incompatible `--concurrent` + `-n` case is rejected outright in :func:`pytest_configure`;
    the default mode is downgraded quietly (notice in :func:`pytest_report_header`).
    """
    requested = _requested_mode(config)
    if requested != "off" and _xdist_active(config):
        return "off"
    return requested


def concurrency_enabled(config: pytest.Config) -> bool:
    """Whether any concurrent execution happens at all."""
    return concurrency_mode(config) != "off"


def should_run_concurrently(item: pytest.Function, mode: Literal["default", "all", "off"]) -> bool:
    """Whether this async test should run as a concurrent group member (see marker table above)."""
    if mode == "off":
        return False
    if item.get_closest_marker("no_concurrent") is not None:
        return False
    if _CAPTURE_FIXTURES.intersection(getattr(item._fixtureinfo, "names_closure", ())):
        return False
    if mode == "all":
        return True
    return item.get_closest_marker("concurrent") is not None


def is_concurrent_member(node: object) -> bool:
    """True if `node` is a test running as part of a concurrent group.

    Used by fixtures that can't behave correctly when tests share one event loop
    (e.g. per-test leaked-task detection). When concurrency is toggled off the tests
    are ordinary items and this returns False, so such fixtures run normally.
    """
    return PYTEST_ASYNCIO_CONCURRENT_INSTALLED and isinstance(node, AsyncioConcurrentGroupMember)


# --------------------------------------------------------------------------- #
# Per-task capture machinery
# --------------------------------------------------------------------------- #
_active: contextvars.ContextVar[_TestCapture | None] = contextvars.ContextVar(
    "lk_concurrent_capture", default=None
)


class _TestCapture:
    """Output sink for a single concurrently-running test."""

    def __init__(self) -> None:
        self.stdout = io.StringIO()
        self.stderr = io.StringIO()
        # pytest's own handler type so `caplog.text` / `caplog.records` keep working.
        self.log_handler = LogCaptureHandler() if _CAPLOG_SUPPORTED else None
        if self.log_handler is not None:
            self.log_handler.setLevel(0)


class _DispatchStream:
    """A `sys.stdout`/`sys.stderr` stand-in that routes writes to the active test.

    With no active test (e.g. pytest's own machinery between tests) it falls through to the
    real stream, so nothing is silently swallowed.
    """

    def __init__(self, wrapped: Any, attr: str) -> None:
        self._wrapped = wrapped
        self._attr = attr

    def write(self, data: str) -> int:
        cap = _active.get()
        if cap is None:
            return self._wrapped.write(data)
        return getattr(cap, self._attr).write(data)

    def flush(self) -> None:
        if _active.get() is None:
            self._wrapped.flush()

    def isatty(self) -> bool:
        return False

    def __getattr__(self, name: str) -> Any:
        return getattr(self._wrapped, name)


class _RoutingLogHandler(logging.Handler):
    """Root-logger handler that routes each record to the active test's capture."""

    def emit(self, record: logging.LogRecord) -> None:
        cap = _active.get()
        if cap is None or cap.log_handler is None:
            return
        # Replicate the level gating Logger.callHandlers would normally do, so
        # `caplog.set_level(...)` keeps working.
        if record.levelno >= cap.log_handler.level:
            cap.log_handler.handle(record)


class _CaptureController:
    """Installs the dispatch streams + routing handler once, around a concurrent gather.

    The group's members all run on one event loop in one thread. Each member installs on
    entry and uninstalls on exit; a refcount keeps the proxies in place for the whole gather
    window (every member starts -- and so increments -- before any of them awaits to
    completion, so the count only returns to zero once the last member is done).
    """

    def __init__(self) -> None:
        self._depth = 0
        self._saved: tuple[Any, Any] | None = None
        self._handler: logging.Handler | None = None

    def __enter__(self) -> _CaptureController:
        if self._depth == 0:
            self._saved = (sys.stdout, sys.stderr)
            sys.stdout = _DispatchStream(sys.stdout, "stdout")  # type: ignore[assignment]
            sys.stderr = _DispatchStream(sys.stderr, "stderr")  # type: ignore[assignment]
            self._handler = _RoutingLogHandler()
            self._handler.setLevel(0)
            logging.getLogger().addHandler(self._handler)
        self._depth += 1
        return self

    def __exit__(self, *exc: object) -> None:
        self._depth -= 1
        if self._depth == 0:
            assert self._saved is not None
            sys.stdout, sys.stderr = self._saved
            if self._handler is not None:
                logging.getLogger().removeHandler(self._handler)
            self._saved = None
            self._handler = None


_controller = _CaptureController()


@contextlib.contextmanager
def _capture_subprocess_fds(terminalreporter: Any) -> Any:
    """Capture file-descriptor-level output for the duration of a concurrent gather.

    The per-task :class:`_CaptureController` only swaps `sys.stdout`/`sys.stderr`, so it sees
    in-process `print`s but not writes a *subprocess* (or C extension) makes straight to the
    inherited stdout/stderr file descriptors -- e.g. a spawned child's `print()`. With pytest's
    own fd capture suspended while tests run, those writes hit the real terminal and corrupt the
    live progress line (and leak into CI logs).

    So for the gather window we point fds 1/2 at a pipe and drain it on a background thread, while
    rerouting the *parent's* terminal writer to the preserved fds -- so dots, the live line and
    reports still render; only out-of-process writes are diverted. The buffer is shared by every
    member (they share the fds) and so can't be attributed to one test; it's yielded back for the
    caller to surface only if the group failed, matching pytest's "show captured output on failure
    only" behaviour. A no-op (`b""`) is yielded when there are no real fds to capture.
    """
    tw = terminalreporter._tw
    original_file = tw._file
    try:
        original_file.flush()
        sys.stdout.flush()
        sys.stderr.flush()
    except Exception:  # pragma: no cover - best effort flush
        pass

    try:
        saved_out, saved_err = os.dup(1), os.dup(2)
    except OSError:  # pragma: no cover - no real fds (e.g. fully in-memory streams)
        yield lambda: b""
        return

    pipe_r, pipe_w = os.pipe()
    os.dup2(pipe_w, 1)
    os.dup2(pipe_w, 2)
    os.close(pipe_w)

    # Parent-side terminal output bypasses the captured fds via the preserved stdout.
    preserved = os.fdopen(saved_out, "w", buffering=1, closefd=True)
    tw._file = preserved

    buffer = bytearray()

    def _pump() -> None:
        while True:
            try:
                chunk = os.read(pipe_r, 65536)
            except OSError:
                break
            if not chunk:  # EOF: every write end (fds 1/2) has been restored/closed
                break
            buffer.extend(chunk)

    pump = threading.Thread(target=_pump, name="lk-concurrent-fd-capture", daemon=True)
    pump.start()

    try:
        yield lambda: bytes(buffer)
    finally:
        try:
            preserved.flush()
        except Exception:  # pragma: no cover
            pass
        # Restore the real fds; this drops the last refs to the pipe's write end, so the pump
        # sees EOF and exits.
        os.dup2(saved_out, 1)
        os.dup2(saved_err, 2)
        os.close(saved_err)
        pump.join(timeout=2.0)
        if not pump.is_alive():
            os.close(pipe_r)
        tw._file = original_file
        preserved.close()  # closes saved_out


def _emit_captured_subprocess_output(
    terminalreporter: Any,
    results: list[tuple[Any, pytest.CallInfo, pytest.TestReport]],
    data: bytes,
) -> None:
    """Surface a concurrent group's captured fd output -- but only if a member failed.

    The output can't be attributed to a single member (they shared the fds), so it's shown as a
    group-level section rather than per test, mirroring pytest's failure-only capture display.
    """
    if not data or not any(report.failed for _, _, report in results):
        return
    text = data.decode("utf-8", "replace")
    terminalreporter.write_sep("-", "Captured subprocess stdout/stderr (concurrent group)")
    terminalreporter._tw.write(text if text.endswith("\n") else text + "\n")


def _format_log(records: list[logging.LogRecord], config: pytest.Config) -> str:
    fmt = config.getini("log_format") or _DEFAULT_LOG_FORMAT
    formatter = logging.Formatter(fmt)
    return "\n".join(formatter.format(record) for record in records)


# --------------------------------------------------------------------------- #
# Hook bodies (called from conftest.py)
# --------------------------------------------------------------------------- #
def repromote_collected(outcome: Result, config: pytest.Config) -> None:
    """Turn the async tests that should run concurrently into concurrent group members.

    In auto mode pytest-asyncio claims every async test (and would run it sequentially); after it
    has had its say we promote the ones :func:`should_run_concurrently` selects, grouping each by
    its parent (module or class) so a group's members share a parent as the plugin requires. The
    user-facing markers are `concurrent` / `no_concurrent`; `asyncio_concurrent` (which the
    underlying plugin keys on) is an internal detail we attach here.

    Meant to be called from a `pytest_pycollect_makeitem` *hookwrapper* registered
    `tryfirst` (outermost), so it runs after pytest-asyncio has had its say.
    """
    mode = concurrency_mode(config)
    if mode == "off":
        return
    result = outcome.get_result()
    if not result:
        return
    items = result if isinstance(result, list) else [result]
    changed = False
    for index, item in enumerate(items):
        if (
            isinstance(item, pytest.Function)
            and not isinstance(item, AsyncioConcurrentGroupMember)
            and inspect.iscoroutinefunction(item.obj)
            and should_run_concurrently(item, mode)
        ):
            member = AsyncioConcurrentGroupMember.promote_from_function(item)
            # Group by module (possible thanks to _enable_cross_parent_grouping)
            group_name = item.getparent(pytest.Module).nodeid
            member.add_marker(pytest.mark.asyncio_concurrent(group=group_name))
            items[index] = member
            changed = True
    if changed:
        outcome.force_result(items)


# --------------------------------------------------------------------------- #
# Task ownership (per-test leaked-task attribution)
# --------------------------------------------------------------------------- #
# Concurrent group members share one event loop, so the usual "diff asyncio.all_tasks() around
# each test" can't tell whose task is whose -- a still-running task from test B looks like a
# leak from test A. Instead we tag every task with the test that created it: a contextvar
# carries the owning test's nodeid, and a task factory on the group's loop records it. The
# factory runs synchronously in the creating task's context, and asyncio tasks inherit their
# parent's context, so background tasks a test spawns (and their descendants) are attributed to
# that test too. fail_on_leaked_tasks then asks only about *its own* still-pending tasks.
_task_owner: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "lk_task_owner", default=None
)
_owner_by_task: weakref.WeakKeyDictionary[asyncio.Task[Any], str] = weakref.WeakKeyDictionary()
_group_loop: asyncio.AbstractEventLoop | None = None


def _tagging_task_factory(
    loop: asyncio.AbstractEventLoop, coro: Any, **kwargs: Any
) -> asyncio.Task[Any]:
    task: asyncio.Task[Any] = asyncio.Task(coro, loop=loop, **kwargs)
    owner = _task_owner.get()
    if owner is not None:
        _owner_by_task[task] = owner
    return task


@contextlib.contextmanager
def _task_ownership(loop: asyncio.AbstractEventLoop) -> Any:
    """Install the tagging task factory + expose the loop for the duration of a group."""
    global _group_loop
    previous_factory = loop.get_task_factory()
    previous_loop = _group_loop
    loop.set_task_factory(_tagging_task_factory)
    _group_loop = loop
    try:
        yield
    finally:
        loop.set_task_factory(previous_factory)
        _group_loop = previous_loop


def owned_pending_tasks(owner: str) -> list[asyncio.Task[Any]]:
    """Still-pending tasks created by test `owner` (see module note on task ownership).

    Lets fail_on_leaked_tasks attribute leaks per test even though the whole group shares one
    event loop. Returns `[]` when no group is running (e.g. sequential mode), so callers fall
    back to their normal whole-loop check.
    """
    loop = _group_loop
    if loop is None:
        return []
    # asyncio.all_tasks() already excludes finished tasks.
    return [task for task in asyncio.all_tasks(loop) if _owner_by_task.get(task) == owner]


# --------------------------------------------------------------------------- #
# Live progress line
# --------------------------------------------------------------------------- #
_ORBIT_INTERVAL = 0.01  # seconds between frames; the only thing that wakes the shared loop
_ORBIT_TRAIL = 10  # base comet length (lit dots behind the head); widened to cover fast steps
_ORBIT_SPEED = 1.0  # default dots advanced per frame (speed); 2.0 == twice as fast, same frame rate
_RUNNING_MARKUP = {"blue": True}


# Cell sentinel: this test is in flight -- render the orbiting head for it (blank braille otherwise).
_RUNNING = object()


def _emit(child: pytest.Function, callinfo: pytest.CallInfo, report: pytest.TestReport) -> None:
    # logstart + logreport back-to-back => one clean line per test (no -v double-print).
    child.ihook.pytest_runtest_logstart(nodeid=child.nodeid, location=child.location)
    if _check_interactive_exception(call=callinfo, report=report):
        child.ihook.pytest_exception_interact(node=child, call=callinfo, report=report)
    child.ihook.pytest_runtest_logreport(report=report)


def _orbit_path(cells: list[int]) -> list[tuple[int, int]]:
    """Braille-dot positions tracing the rectangle around `cells` clockwise from the top-left.

    `cells` are the running columns, left to right; finished columns are simply absent, so the
    head jumps over them and the walls sit on the first/last running column. Per cell the dots are
    1,4 across the top, 7,8 across the bottom, 1,2,3,7 down the left and 4,5,6,8 down the right.
    Top dots run left->right, down the right wall of the last cell, bottom dots right->left, then
    up the left wall of the first cell. Shared corners are not repeated.
    """
    if not cells:
        return []
    first, last = cells[0], cells[-1]
    top = [(cell, bit) for cell in cells for bit in (0x01, 0x08)]
    right = [(last, bit) for bit in (0x10, 0x20, 0x80)]
    bottom = [(cell, bit) for cell in reversed(cells) for bit in (0x80, 0x40)][1:]
    left = [(first, 0x04), (first, 0x02)]
    return top + right + bottom + left


class _GroupProgress:
    """Owns how one concurrent group reports -- both strategies behind one interface.

    The mode is decided once in `__init__` and never leaks to the caller:

    * live (non-verbose colour TTY, capture on): a transient in-place line, one cell per test.
      While a test runs its cell is blank braille; collectively the running cells show a colored
      head orbiting just the running cells -- finished columns are dropped from the loop, so the
      head jumps over them (see :func:`_orbit_path`). Test starts and completions only mutate
      cell state; the line is redrawn on the animation timer (see :meth:`ticking`), so a cell
      freezes to its final status letter/colour on the next frame and a burst of completions
      costs one redraw rather than one per test. Per-test reports are withheld until the line is
      frozen, then replayed through a silenced terminal writer so stats/failures/other plugins
      still see them via the hooks without a second dot row; the cumulative `[ XX%]` is stamped
      on afterwards (see :meth:`_finish_line`).
    * streaming (everything else): no line; each report is emitted the moment its test finishes,
      so pytest prints dots/names live as usual.

    `run_group` drives this purely through `begin` / `on_start` / `on_finish` /
    `ticking` / `finalize` and never branches on the mode.
    """

    def __init__(self, terminalreporter: Any, config: pytest.Config) -> None:
        self._terminalreporter = terminalreporter
        self._tw = terminalreporter._tw
        self._config = config
        verbose = config.get_verbosity() > 0
        capture_off = config.option.capture == "no"
        self._live: bool = self._tw.hasmarkup and not verbose and not capture_off
        self._cell: dict[Any, Any] = {}
        self._done: set[Any] = set()
        self._order: list[Any] = []
        self._lit: dict[int, int] = {}
        self._label = ""
        self._width = 0
        self._last_plain = ""
        self._phase = 0.0
        self._step = _ORBIT_SPEED
        self._trail = max(_ORBIT_TRAIL, math.ceil(self._step))
        self._marked: dict[Any, str] = {}
        self._glyph_cache: dict[str, str] = {}
        self._count_width = 0
        self._fullwidth = 0

    # -- driven by run_group ------------------------------------------------- #
    def begin(self, children: list[Any]) -> None:
        self._order = list(children)
        self._cell = dict.fromkeys(children, _RUNNING)
        self._marked = {}
        self._count_width = len(str(len(self._order)))
        self._fullwidth = self._tw.fullwidth
        if self._order:
            self._label = self._order[0].nodeid.split("::", 1)[0] + " "
        self._render()

    def on_start(self, child: Any) -> None:
        self._cell[child] = _RUNNING

    def on_finish(
        self, child: pytest.Function, callinfo: pytest.CallInfo, report: pytest.TestReport
    ) -> None:
        letter, markup = self._letter_and_markup(report)
        if self._live:
            self._marked[child] = self._tw.markup(letter, **markup)
        self._cell[child] = (letter, markup)
        self._done.add(child)
        if not self._live:
            _emit(child, callinfo, report)

    @contextlib.asynccontextmanager
    async def ticking(self) -> Any:
        """Step the orbiting head on a timer while the group's gather runs (live only).

        Also the only place the live line is redrawn while the gather runs: on_start/on_finish
        merely mutate cell state, and the final redraw here (after the head is cancelled) freezes
        the end state that _finish_line then stamps the percentage onto.
        """
        if not self._live:
            yield
            return
        # No owner tag (created in the gather's own task context), so it can't look like a
        # leaked test task; cancelled and awaited here.
        orbit = asyncio.ensure_future(self._drive_orbit())
        try:
            yield
        finally:
            orbit.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await orbit
            self._render()

    @contextlib.contextmanager
    def finalize(
        self,
        results: list[tuple[AsyncioConcurrentGroupMember, pytest.CallInfo, pytest.TestReport]],
        *,
        is_last_group: bool,
    ) -> Any:
        """Wrap teardown: replay withheld reports silently, then stamp the line's progress (live only).

        Streaming mode already emitted each report in :meth:`on_finish`, so it just runs the body.
        The reports replay through a throwaway writer so stats/failures/other plugins still see
        them via the hooks without a second dot row -- but that throwaway writer is also where
        pytest's own `[ XX%]` would land, so the cumulative percentage is stamped onto the real
        writer afterwards (see :meth:`_finish_line`).
        """
        if not self._live:
            yield
            return
        from _pytest.config import create_terminal_writer

        original_tw = self._terminalreporter._tw
        self._terminalreporter._tw = create_terminal_writer(
            self._terminalreporter.config, io.StringIO()
        )
        try:
            for child, callinfo, report in results:
                _emit(child, callinfo, report)
            yield
        finally:
            self._terminalreporter._tw = original_tw
            self._finish_line(is_last_group=is_last_group)

    # -- rendering ----------------------------------------------------------- #
    async def _drive_orbit(self) -> None:
        while True:
            await asyncio.sleep(_ORBIT_INTERVAL)
            self._advance_orbit()
            self._render()

    def _advance_orbit(self) -> None:
        """Advance the head by `self._step` dots along the loop of still-running cells.

        Distance per frame is the speed knob: raising it moves the head faster without touching
        the frame rate (`_ORBIT_INTERVAL`), so no extra event-loop wakeups and no added load on
        the gathered tests. The head is a phase in [0, 1) so that when a finished cell drops out
        and the loop shortens it stays at the same fraction of the way round rather than
        teleporting (which flickered when a burst of completions landed between two frames). The
        trail widens to `self._trail` >= step so a fast head stays a continuous comet instead of
        a dot hopping with gaps.
        """
        running = [
            index for index, child in enumerate(self._order) if self._cell[child] is _RUNNING
        ]
        path = _orbit_path(running)
        if not path:
            self._lit = {}
            return
        period = len(path)
        self._phase = (self._phase + self._step / period) % 1.0
        head = int(self._phase * period)
        lit: dict[int, int] = {}
        for offset in range(self._trail):
            cell, dot = path[(head - offset) % period]
            lit[cell] = lit.get(cell, 0) | dot
        self._lit = lit

    def _letter_and_markup(self, report: pytest.TestReport) -> tuple[str, dict[str, bool]]:
        """The exact status letter + colour pytest would use for this report."""
        _, letter, word = self._config.hook.pytest_report_teststatus(
            report=report, config=self._config
        )
        if isinstance(word, tuple):  # a plugin already chose the markup
            return letter, word[1]
        was_xfail = hasattr(report, "wasxfail")
        if report.passed:
            markup = {"yellow": True} if was_xfail else {"green": True}  # xpass vs pass
        elif report.failed:
            markup = {"red": True}
        elif report.skipped:
            markup = {"yellow": True}  # skipped / xfailed
        else:
            markup = {}
        return letter, markup

    def _finish_line(self, is_last_group: bool) -> None:
        """Stamp the cumulative `[ XX%]` onto the finished colour line and keep it as the record.

        The colour line carries embedded markup, so its raw width is wrong -- reset
        `_current_line` to the plain text first, then let pytest fill the same right-aligned
        progress message it writes after an ordinary dot row, so concurrent groups read exactly
        like the sequential lines. The final group is left untouched: pytest's own end-of-loop
        `[100%]` lands on it, and writing it here too would double.
        """
        if not (self._live and self._width):
            return
        self._tw._current_line = self._last_plain
        if is_last_group:
            return
        self._terminalreporter._write_progress_information_filling_space()
        self._tw.line()

    def _render(self) -> None:
        if not self._live:
            return
        counter = f" [{len(self._done):>{self._count_width}}/{len(self._order)}]"
        glyphs: list[str] = []
        marked: list[str] = []
        for index, child in enumerate(self._order):
            cell = self._cell[child]
            if cell is _RUNNING:
                glyph = chr(0x2800 + self._lit.get(index, 0))
                glyphs.append(glyph)
                marked.append(self._render_glyph(glyph))
            else:
                glyphs.append(cell[0])
                marked.append(self._marked[child])
        plain = self._label + "".join(glyphs) + counter
        if len(plain) >= self._fullwidth:  # too wide for one cell per test
            body = plain = f"{self._label}running {len(self._done)}/{len(self._order)}"
        else:
            body = self._label + "".join(marked) + counter
        pad = " " * max(0, self._width - len(plain))  # erase a previously longer line
        self._tw.write("\r" + body + pad, flush=True)
        self._width = len(plain)
        self._last_plain = plain

    def _render_glyph(self, glyph: str) -> str:
        cached = self._glyph_cache.get(glyph)
        if cached is None:
            cached = self._tw.markup(glyph, **_RUNNING_MARKUP)
            self._glyph_cache[glyph] = cached
        return cached


def run_group(group: AsyncioConcurrentGroup, nextgroup: AsyncioConcurrentGroup | None) -> object:
    """Run one concurrent group, fixing the plugin's event-loop handling and reporting.

    This mirrors `pytest_asyncio_concurrent.plugin.pytest_runtest_protocol_async_group` but
    differs in two ways:

    * It captures the event loop **after** per-member setup instead of before. In auto mode,
      pytest-asyncio sets up its own loop while the members' fixtures are created, so the
      plugin's pre-setup capture no longer matches the loop `asyncio.gather` actually binds
      to -- which crashes with "future belongs to a different loop". Capturing after setup keeps
      the gather, `run_until_complete` and the test coroutines all on the same loop.

    * It reports each test **as it finishes** instead of all at once after the gather, via the
      :class:`_GroupProgress` reporter -- which also drives a colour-coded live line for a
      non-verbose TTY. The plugin instead emits `logstart` for every test up front and
      `logreport` for every test at the end, which shows no progress until everything is done
      and prints every name twice under `-v`.

    Registered `tryfirst` for the plugin's `firstresult` hook, so this wins and the plugin's
    own implementation does not run. We reuse the plugin's (private) helpers for the individual
    setup/call/teardown steps so behaviour otherwise stays identical.
    """
    if not group.children_have_same_parent:
        for child in group.children:
            child.add_marker("skip")
        warnings.warn(
            PytestAsyncioConcurrentGroupingWarning(
                f"Asyncio Concurrent Group [{group.name}] has children from different parents, "
                "skipping all of its children."
            ),
            stacklevel=2,
        )

    terminalreporter = group.config.pluginmanager.get_plugin("terminalreporter")
    progress = _GroupProgress(terminalreporter, group.config)

    item_passed_setup: list[AsyncioConcurrentGroupMember] = []
    for child in group.children:
        report = _call_and_report(_setup_child(child), child, "setup")
        if report.passed:
            item_passed_setup.append(child)

    progress.begin(item_passed_setup)

    async def run_one(
        child: AsyncioConcurrentGroupMember,
    ) -> tuple[AsyncioConcurrentGroupMember, pytest.CallInfo, pytest.TestReport]:
        progress.on_start(child)
        callinfo = await _call_runtest_async(child)
        report = child.ihook.pytest_runtest_makereport(item=child, call=callinfo)
        progress.on_finish(child, callinfo, report)
        return child, callinfo, report

    async def _run_all() -> list[
        tuple[AsyncioConcurrentGroupMember, pytest.CallInfo, pytest.TestReport]
    ]:
        async with progress.ticking():
            return await asyncio.gather(*[run_one(c) for c in item_passed_setup])

    loop = _current_event_loop()
    # Under `-s` the user asked to see output live, so don't divert the subprocess fds.
    capture_subprocess = group.config.option.capture != "no"
    # Tag tasks with their owning test (for leaked-task attribution) for the whole group,
    # including teardown -- that's when fail_on_leaked_tasks inspects the loop.
    with _task_ownership(loop):
        if capture_subprocess:
            with _capture_subprocess_fds(terminalreporter) as captured:
                results = loop.run_until_complete(_run_all())
            captured_subprocess_output = captured()
        else:
            results = loop.run_until_complete(_run_all())
            captured_subprocess_output = b""
        with progress.finalize(results, is_last_group=nextgroup is None):
            for child in group.children:
                _call_and_report(_teardown_child(child, nextgroup=nextgroup), child, "teardown")
                child.ihook.pytest_runtest_logfinish(nodeid=child.nodeid, location=child.location)
        _emit_captured_subprocess_output(terminalreporter, results, captured_subprocess_output)

    return True


def _current_event_loop() -> asyncio.AbstractEventLoop:
    try:
        return asyncio.get_event_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        return loop


async def run_member_capturing(item: pytest.Function) -> object:
    """Run one concurrent test with per-task output capture.

    Replaces the plugin's `pytest_runtest_call_async` (registered `tryfirst` so this wins
    the `firstresult` race). We deliberately skip the plugin's `hook_wrapper_entered` --
    that is what re-enters pytest's broken-under-concurrency global capture -- and capture here.
    """
    if not inspect.iscoroutinefunction(item.obj):
        pytest.skip("Marking a sync function with @asyncio_concurrent is invalid.")

    testfunction = item.obj
    testargs = {arg: item.funcargs[arg] for arg in item._fixtureinfo.argnames}

    # Tag tasks this test spawns with its nodeid so fail_on_leaked_tasks can attribute leaks
    # to it even though the whole group shares one event loop.
    owner_token = _task_owner.set(item.nodeid)
    try:
        # `-s` / `--capture=no`: the user asked to see output live -- don't buffer it.
        if item.config.option.capture == "no":
            return await testfunction(**testargs)

        cap = _TestCapture()
        # Wire caplog to *this* test's records so caplog-using tests stay correct (and isolated).
        if _CAPLOG_SUPPORTED and cap.log_handler is not None:
            item.stash[caplog_handler_key] = cap.log_handler
            item.stash[caplog_records_key] = {
                "setup": [],
                "call": cap.log_handler.records,
                "teardown": [],
            }

        with _controller:
            token = _active.set(cap)
            try:
                return await testfunction(**testargs)
            finally:
                _active.reset(token)
                if cap.stdout.getvalue():
                    item.add_report_section("call", "stdout", cap.stdout.getvalue())
                if cap.stderr.getvalue():
                    item.add_report_section("call", "stderr", cap.stderr.getvalue())
                if cap.log_handler is not None and cap.log_handler.records:
                    item.add_report_section(
                        "call", "log", _format_log(cap.log_handler.records, item.config)
                    )
    finally:
        _task_owner.reset(owner_token)


# HACK: some callers invoke pytest without installing dev dependencies,
# so we might end up here without the underlying plugin installed.
# Keeping the previous behavior intact here.
if PYTEST_ASYNCIO_CONCURRENT_INSTALLED:
    # --------------------------------------------------------------------------- #
    # pytest hooks
    #
    # This module is loaded as a *global* plugin (`-p tests.concurrency` in pyproject), not via
    # conftest.py. That matters: the underlying plugin invokes `pytest_runtest_protocol_async_group`
    # through `session.ihook`, which is scoped to the rootdir and would NOT see hooks defined in a
    # subdirectory conftest like tests/conftest.py.
    # --------------------------------------------------------------------------- #
    def pytest_addoption(parser: Parser) -> None:
        add_toggle_option(parser)

    def _disable_concurrent_plugin(config: pytest.Config) -> None:
        """Unregister pytest-asyncio-concurrent so it doesn't fight xdist.

        Its runtestloop wrapper and collection regrouping assume a single in-process run; under xdist
        they desync the worker's positional item indexing -- `items[nextitem_index]` runs off the
        end and the worker dies with an IndexError (surfaced as an INTERNALERROR on the controller).
        With the plugin gone, plain pytest-asyncio (auto mode) handles the async tests and xdist
        distributes them by item as usual. Matching on the module name covers both the package and its
        `.plugin` submodule, and does not match regular `pytest_asyncio`.
        """
        pm = config.pluginmanager
        for plugin in list(pm.get_plugins()):
            module = getattr(plugin, "__name__", "") or getattr(type(plugin), "__module__", "")
            if isinstance(module, str) and module.startswith("pytest_asyncio_concurrent"):
                pm.unregister(plugin)

    def pytest_configure(config: pytest.Config) -> None:
        if not PYTEST_ASYNCIO_CONCURRENT_INSTALLED:
            return

        register_markers(config)
        if _xdist_active(config):
            # xdist distributes by item; a concurrent group can't span workers. Reject an explicit
            # request, otherwise disable concurrency (the underlying plugin included) and go sequential.
            if _requested_mode(config) == "all":
                raise pytest.UsageError(
                    "--concurrent (or LK_TEST_CONCURRENCY=all) is incompatible with pytest-xdist "
                    "(-n): a concurrent group shares a single event loop, which xdist would split "
                    "across worker processes. Drop -n to run concurrently, or drop --concurrent."
                )
            _disable_concurrent_plugin(config)
            return
        _enable_cross_parent_grouping()

    def pytest_report_header(config: pytest.Config) -> str | None:
        # Once per session, on the controller, at the top of the run: announce that -n forced the
        # default concurrent mode to sequential. (--concurrent under -n already errored in configure.)
        if hasattr(config, "workerinput"):
            return None
        if _xdist_active(config) and _requested_mode(config) != "off":
            return "concurrency: disabled under pytest-xdist (-n); async tests run sequentially."
        return None

    @pytest.hookimpl(hookwrapper=True, tryfirst=True)
    def pytest_pycollect_makeitem(collector: pytest.Collector, name: str, obj: object) -> Any:
        # Outermost wrapper: runs after pytest-asyncio's auto-mode conversion, so we can restore
        # the concurrent group members it converted back to ordinary sequential items.
        outcome = yield
        repromote_collected(outcome, collector.config)

    @pytest.hookimpl(specname="pytest_runtest_protocol_async_group", tryfirst=True)
    def pytest_runtest_protocol_async_group(
        group: AsyncioConcurrentGroup, nextgroup: AsyncioConcurrentGroup | None
    ) -> object:
        # Wins the plugin's firstresult hook; fixes its event-loop handling (see run_group).
        return run_group(group, nextgroup)

    @pytest.hookimpl(specname="pytest_runtest_call_async", tryfirst=True)
    async def pytest_runtest_call_async(item: pytest.Function) -> object:
        # Wins the plugin's firstresult hook; adds per-task output capture.
        return await run_member_capturing(item)

    def _enable_cross_parent_grouping() -> None:
        """One concurrent group per module: module-level tests plus every class in the file.

        The plugin pins a group to the parent of its first member and skips any group whose
        members don't all share it. Here the group sits at the Module and mixed parents are
        allowed. A class-scoped fixture on a concurrent member fails at setup as a result --
        its finalizer needs the Class node on SetupState, which module grouping removes.
        Function-, module-, and session-scoped fixtures are unaffected.
        """
        if getattr(AsyncioConcurrentGroup, "_lk_cross_parent_grouping", False):
            return

        original_init = AsyncioConcurrentGroup.__init__

        def init_at_module(self, parent, originalname):
            original_init(self, parent.getparent(pytest.Module) or parent, originalname)

        def add_child_keep_group(self, item):
            item.group = self
            self.children.append(item)
            self.children_finalizer[item] = []

        AsyncioConcurrentGroup.__init__ = init_at_module
        AsyncioConcurrentGroup.add_child = add_child_keep_group
        AsyncioConcurrentGroup._lk_cross_parent_grouping = True