from ci.util.global_interrupt_handler import handle_keyboard_interrupt


#!/usr/bin/env python3
"""
WARNING: sys.stdout.flush() causes blocking issues on Windows with QEMU/subprocess processes!
Use conditional flushing: `if os.name != 'nt': sys.stdout.flush()` to avoid Windows blocking
while maintaining real-time output visibility on Unix systems.
"""

import _thread
import os


# Force UTF-8 with replace error handler for child processes (including Meson) on Windows.
# Without this, Meson's mtest.py crashes with UnicodeEncodeError when test output
# contains non-ASCII characters (e.g., ✓) and the console uses cp1252.
if os.name == "nt":
    os.environ.setdefault("PYTHONIOENCODING", "utf-8:replace")

import queue
import re
import subprocess
import sys
import threading
import time
from dataclasses import dataclass
from queue import Queue

# Import for type annotation only
from typing import TYPE_CHECKING, Callable, Optional

from running_process import RunningProcess


if TYPE_CHECKING:
    from typeguard import typechecked
else:

    def typechecked(f):  # type: ignore[no-redef]
        return f


from ci.util.color_output import print_cache_hit


if TYPE_CHECKING:
    from ci.util.fingerprint import FingerprintManager
from ci.util.output_formatter import TimestampFormatter
from ci.util.test_exceptions import (
    TestExecutionFailedException,
    TestFailureInfo,
    TestTimeoutException,
)
from ci.util.test_types import (
    TestArgs,
    TestCategories,
    TestResult,
    TestResultType,
    TestSuiteResult,
    determine_test_categories,
)
from ci.util.timestamp_print import ts_print


@dataclass(slots=True)
class TestCounts:
    unit_test_count: int
    example_count: int
    python_test_count: int


_IS_GITHUB_ACTIONS = os.getenv("GITHUB_ACTIONS") == "true"
_TIMEOUT = 600 if _IS_GITHUB_ACTIONS else 240
_GLOBAL_TIMEOUT = (
    600 if _IS_GITHUB_ACTIONS else 600
)  # Increased to 600s for local uno compilation

# Abort threshold for total failures across all processes (unit + examples)
MAX_FAILURES_BEFORE_ABORT = 3


def get_test_counts() -> TestCounts:
    """
    Get accurate counts of unit tests, examples, and Python tests.

    Returns:
        TestCounts with unit_test_count, example_count, and python_test_count.
    """
    try:
        from pathlib import Path

        from ci.util.smart_selector import discover_all_tests

        # Get unit tests and examples
        unit_tests, examples = discover_all_tests(Path.cwd())
        unit_count = len(unit_tests)
        example_count = len(examples)

        # Count Python tests by running pytest --collect-only
        python_count = 0
        try:
            result = RunningProcess.run(
                ["uv", "run", "pytest", "--collect-only", "-q", "ci/tests"],
                cwd=None,
                check=False,
                timeout=10,
            )
            if result.returncode == 0:
                # Parse output to count tests
                # pytest --collect-only -q outputs lines like "test_foo.py::test_bar"
                lines = result.stdout.strip().split("\n")
                python_count = sum(1 for line in lines if "::" in line)
        except KeyboardInterrupt as ki:
            handle_keyboard_interrupt(ki)
            raise
        except (subprocess.SubprocessError, RuntimeError):
            # If counting fails, use 0 as fallback
            python_count = 0

        return TestCounts(
            unit_test_count=unit_count,
            example_count=example_count,
            python_test_count=python_count,
        )

    except KeyboardInterrupt as ki:
        handle_keyboard_interrupt(ki)
        raise
    except Exception:
        # If discovery fails, return zeros
        return TestCounts(unit_test_count=0, example_count=0, python_test_count=0)


# Pre-compiled patterns for detecting real error lines (not compiler flags).
# These are module-level to avoid re-compilation on every call.
_REAL_ERROR_PATTERNS: list[re.Pattern[str]] = [
    re.compile(
        r"(?<![/-])error\s*:", re.IGNORECASE
    ),  # "error:" not preceded by -W or /
    re.compile(
        r"(?<![/-])fatal\s*:", re.IGNORECASE
    ),  # "fatal:" not preceded by flag chars
    re.compile(r"\bFAILED\s*:", re.IGNORECASE),  # ninja "FAILED:" lines
    re.compile(r"\bundefined reference\b", re.IGNORECASE),  # linker errors
    re.compile(r"\bcannot find\b", re.IGNORECASE),  # missing file/symbol errors
    re.compile(r"\bno such file\b", re.IGNORECASE),  # missing file errors
    re.compile(r"^\s*FAIL\b"),  # test failure lines (start of line)
    re.compile(r"\bninja:\s.*\bfailed\b", re.IGNORECASE),  # ninja summary failures
    re.compile(r"\bSegmentation fault\b", re.IGNORECASE),  # crashes
    re.compile(r"\bAborted\b"),  # ASAN/assertion aborts
    re.compile(r"==\d+==.*\bAddressSanitizer\b"),  # ASAN error reports
    re.compile(r"==\d+==.*\bLeakSanitizer\b"),  # LSAN error reports
    re.compile(r"==\d+==.*\bUndefinedBehaviorSanitizer\b"),  # UBSAN error reports
    re.compile(r"\bERROR SUMMARY:\s*[1-9]"),  # valgrind/sanitizer summary with errors
]


def _is_real_error_line(line: str) -> bool:
    """Check if a line contains a real error indicator, not just compiler flags.

    Returns True for actual error messages like:
      - "error: undeclared identifier 'foo'"
      - "FAILED: examples/example-Blink.so"
      - "fatal error: file not found"
      - "undefined reference to 'bar'"

    Returns False for compiler flag noise like:
      - "-Werror=unused-variable"
      - "--print-errorlogs"
    """
    for pattern in _REAL_ERROR_PATTERNS:
        if pattern.search(line):
            return True
    return False


def extract_error_snippet(accumulated_output: list[str], context_lines: int = 5) -> str:
    """
    Extract relevant error snippets from process output.

    Searches for lines containing actual error messages (not compiler flags
    like -Werror) and extracts context around the first few occurrences.
    Always includes the tail of the output to ensure failure summaries are
    visible.

    Args:
        accumulated_output: List of output lines from the process
        context_lines: Number of lines to capture before/after each error line (default: 5)

    Returns:
        Formatted string containing error snippets with tail context
    """
    if not accumulated_output:
        return "No output captured"

    error_snippets: list[str] = []

    # Find all lines that contain real error indicators
    error_line_indices: list[int] = []
    for i, line in enumerate(accumulated_output):
        if _is_real_error_line(line):
            error_line_indices.append(i)

    # Always capture tail of output (last 30 lines) for failure summaries
    tail_count = 30
    tail_start = max(0, len(accumulated_output) - tail_count)

    if not error_line_indices:
        # No specific errors found, return tail which often has useful info
        max_lines = min(tail_count, len(accumulated_output))
        return (
            "No specific error lines found. Last "
            + str(max_lines)
            + " lines:\n"
            + "\n".join(accumulated_output[-max_lines:])
        )

    # Extract context around first 10 errors
    max_errors_to_show = 10
    shown_lines: set[int] = set()

    for error_idx in error_line_indices[:max_errors_to_show]:
        start_idx = max(0, error_idx - context_lines)
        end_idx = min(len(accumulated_output), error_idx + context_lines + 1)

        snippet_lines: list[str] = []
        for j in range(start_idx, end_idx):
            if j not in shown_lines:
                line_marker = "➤ " if j == error_idx else "  "
                snippet_lines.append(f"{line_marker}{accumulated_output[j]}")
                shown_lines.add(j)

        if snippet_lines:
            error_snippets.append("\n".join(snippet_lines))

    # Summary of remaining errors
    if len(error_line_indices) > max_errors_to_show:
        remaining_count = len(error_line_indices) - max_errors_to_show
        error_snippets.append(f"... and {remaining_count} more error(s) found")

    # Always append tail of output for context (skip lines already shown)
    tail_lines: list[str] = []
    for j in range(tail_start, len(accumulated_output)):
        if j not in shown_lines:
            tail_lines.append(f"  {accumulated_output[j]}")
    if tail_lines:
        error_snippets.append(
            f"--- Last {len(tail_lines)} lines of output ---\n" + "\n".join(tail_lines)
        )

    return "\n\n".join(error_snippets)


@dataclass
class ProcessTiming:
    """Information about a completed process execution for timing summary"""

    name: str
    duration: float
    command: str
    skipped: bool = False
    # Phase timing breakdown for Meson tests (seconds) - None if not applicable
    meson_setup_time: Optional[float] = None
    ninja_maintenance_time: Optional[float] = None
    compile_time: Optional[float] = None
    test_execution_time: Optional[float] = None
    # Compilation sub-phase breakdown (seconds) - None if not applicable
    compile_core_time: Optional[float] = None
    compile_objects_time: Optional[float] = None
    compile_example_link_time: Optional[float] = None
    compile_test_link_time: Optional[float] = None


@dataclass
class ProcessState:
    """Tracks the state of an individual process during parallel execution"""

    process: RunningProcess
    last_activity_time: float
    command: str


@typechecked
class TestOutputFormatter:
    """Formats test output in a consistent way"""

    def __init__(self, verbose: bool = False):
        self.verbose = verbose
        self.current_suite: Optional[TestSuiteResult] = None
        self._start_time = time.time()

    def start_suite(self, name: str) -> None:
        """Start a new test suite"""
        self.current_suite = TestSuiteResult(
            name=name, results=[], start_time=time.time()
        )

    def end_suite(self, passed: bool = True) -> None:
        """End the current test suite"""
        if self.current_suite:
            self.current_suite.end_time = time.time()
            self.current_suite.passed = passed

    def add_result(self, result: TestResult) -> None:
        """Add a test result to the current suite"""
        if self.current_suite:
            self.current_suite.results.append(result)
            self._format_result(result)

    def _format_result(self, result: TestResult) -> None:
        """Format and display a test result"""
        # Only show if verbose or important
        if not (self.verbose or self._should_display(result)):
            return

        # Format timestamp
        delta = result.timestamp - self._start_time

        # Format message based on type
        if result.type == TestResultType.SUCCESS:
            color = "\033[92m"  # Green
        elif result.type == TestResultType.ERROR:
            color = "\033[91m"  # Red
        elif result.type == TestResultType.WARNING:
            color = "\033[93m"  # Yellow
        else:
            color = "\033[0m"  # Reset

        # Build message - only include timestamp, not command name
        msg = f"{delta:.2f} "
        msg += f"{color}{result.message}\033[0m"

        # Print with indentation
        ts_print(f"  {msg}")

    def _should_display(self, result: TestResult) -> bool:
        """Determine if a result should be displayed"""
        # Always show errors
        if result.type == TestResultType.ERROR:
            return True

        # Always show final success/completion messages
        if any(
            marker in result.message
            for marker in ["### SUCCESS", "### ERROR", "Test execution complete:"]
        ):
            return True

        # Show warnings in verbose mode
        if result.type == TestResultType.WARNING and self.verbose:
            return True

        # Show info/debug only in verbose mode
        return self.verbose and result.type in [
            TestResultType.INFO,
            TestResultType.DEBUG,
        ]


@typechecked
@dataclass
class TestProcessConfig:
    """Configuration for a test process"""

    command: str | list[str]
    echo: bool = True
    auto_run: bool = False
    timeout: Optional[int] = None
    enable_stack_trace: bool = True
    description: str = ""
    parallel_safe: bool = True  # Whether this process can run in parallel with others
    output_filter: Callable[[str], bool] | None = (
        None  # Filter function for output lines
    )


class ProcessOutputHandler:
    """Handles capturing and displaying process output with structured results"""

    def __init__(self, verbose: bool = False):
        self.formatter = TestOutputFormatter(verbose=verbose)
        self.current_command: Optional[str] = None
        self.header_printed = False

    def handle_output_line(self, line: str, process_name: str) -> None:
        """Process a line of output and convert it to structured test results"""
        # Start new test suite if command changes
        if process_name != self.current_command:
            if self.current_command:
                self.formatter.end_suite()
            self.formatter.start_suite(process_name)
            self.current_command = process_name
            self.header_printed = True
            self.formatter.add_result(
                TestResult(
                    type=TestResultType.INFO,
                    message=f"=== [{process_name}] ===",
                    test_name=process_name,
                )
            )

        # Convert line to appropriate test result
        result = self._parse_line_to_result(line, process_name)
        if result:
            self.formatter.add_result(result)

    def _parse_line_to_result(
        self, line: str, process_name: str
    ) -> Optional[TestResult]:
        """Parse a line of output into a structured test result"""
        # Skip empty lines
        if not line.strip():
            return None

        # Skip test output noise
        if any(
            noise in line
            for noise in [
                "doctest version is",
                'run with "--help"',
                "assertions:",
                "test cases:",
                "MESSAGE:",
                "TEST CASE:",
                "Test passed",
                "Test execution",
                "Test completed",
                "Running test:",
                "Process completed:",
                "Command completed:",
                "Command output:",
                "Exit code:",
                "All parallel tests",
                "JSON parsing failed",
                "readFrameAt failed",
                "MemoryFileHandle",
                "C:\\Users\\",
                "\\dev\\fastled\\",
                "\\tests\\",
                "\\src\\",
                "test line_simplification.exe",
                "test noise_hires.exe",
                "test mutex.exe",
                "test rbtree.exe",
                "test priority_queue.exe",
                "test json_roundtrip.exe",
                "test malloc_hooks.exe",
                "test rectangular_buffer.exe",
                "test ostream.exe",
                "test active_strip_data_json.exe",
                "test noise_range.exe",
                "test json.exe",
                "test point.exe",
                "test screenmap.exe",
                "test task.exe",
                "test queue.exe",
                "test promise.exe",
                "test shared_ptr.exe",
                "test screenmap_serialization.exe",
                "test strstream.exe",
                "test slice.exe",
                "test hsv_conversion_accuracy.exe",
                "test tile2x2.exe",
                "test tuple.exe",
                "test raster.exe",
                "test transform.exe",
                "test transition_ramp.exe",
                "test thread_local.exe",
                "test set_inlined.exe",
                "test vector.exe",
                "test strip_id_map.exe",
                "test weak_ptr.exe",
                "test variant.exe",
                "test type_traits.exe",
                "test splat.exe",
                "test hsv16.exe",
                "test ui_help.exe",
                "test unordered_set.exe",
                "test traverse_grid.exe",
                "test ui_title_bug.exe",
                "test videofx_wrapper.exe",
                "test ui.exe",
                "test video.exe",
                "test xypath.exe",
            ]
        ):
            return None

        # Success messages
        if "### SUCCESS" in line:
            return TestResult(
                type=TestResultType.SUCCESS, message=line, test_name=process_name
            )

        # Error messages
        if any(
            marker in line
            for marker in [
                "### ERROR",
                "FAILED",
                "ERROR",
                "Crash",
                "Test FAILED",
                "Compilation failed",
                "Build failed",
            ]
        ):
            return TestResult(
                type=TestResultType.ERROR, message=line, test_name=process_name
            )

        # Warning messages
        if any(marker in line.lower() for marker in ["warning:", "note:"]):
            return TestResult(
                type=TestResultType.WARNING, message=line, test_name=process_name
            )

        # Test completion messages
        if "Test execution complete:" in line:
            return TestResult(
                type=TestResultType.INFO, message=line, test_name=process_name
            )

        # Default to info level for other lines
        return TestResult(
            type=TestResultType.INFO, message=line, test_name=process_name
        )


def create_unit_test_process(
    args: TestArgs, enable_stack_trace: bool
) -> RunningProcess:
    """Create a unit test process without starting it"""
    # First compile the tests
    compile_cmd: list[str] = [
        "uv",
        "run",
        "python",
        "-m",
        "ci.compiler.cpp_test_run",
        "--compile-only",
    ]
    if args.test:
        compile_cmd.extend(["--test", args.test])
    if args.clean:
        compile_cmd.append("--clean")
    if args.verbose:
        compile_cmd.append("--verbose")
    if args.show_compile:
        compile_cmd.append("--show-compile")
    if args.show_link:
        compile_cmd.append("--show-link")
    if args.check:
        compile_cmd.append("--check")

    if args.clang:
        compile_cmd.append("--clang")
    if args.gcc:
        compile_cmd.append("--gcc")
    if args.no_pch:
        compile_cmd.append("--no-pch")
    if args.debug:
        compile_cmd.append("--debug")

    # subprocess.run(compile_cmd, check=True)

    # Then run the tests using our new test runner
    test_cmd = ["uv", "run", "python", "-m", "ci.run_tests"]
    if args.test:
        test_cmd.extend(["--test", str(args.test)])
    if args.verbose:
        test_cmd.append("--verbose")
    if enable_stack_trace:
        test_cmd.append("--stack-trace")

    return RunningProcess(
        test_cmd,
        timeout=_TIMEOUT,  # 2 minutes timeout
        auto_run=True,
        output_formatter=TimestampFormatter(),
    )


def create_examples_test_process(
    args: TestArgs, enable_stack_trace: bool
) -> RunningProcess:
    """Create an examples test process using Meson build system"""
    # Use Meson-based example compilation instead of Python compiler
    cmd = ["uv", "run", "python", "-u", "ci/util/meson_example_runner.py"]

    # Add example names if specified
    if args.examples is not None:
        cmd.extend(args.examples)

    # Map command-line arguments to meson_example_runner.py
    if args.clean:
        cmd.append("--clean")
    if args.no_pch:
        cmd.append("--no-pch")  # Ignored by Meson (PCH always enabled)
    if args.no_parallel:
        cmd.append("--no-parallel")
    if args.verbose:
        cmd.append("--verbose")
    if args.debug:
        cmd.append("--debug")
    if args.build_mode:
        cmd.extend(["--build-mode", args.build_mode])

    # Auto-enable full mode for examples to include execution
    if args.full or args.examples is not None:
        cmd.append("--full")
    if args.log_failures is not None:
        cmd.extend(["--log-failures", str(args.log_failures)])

    # Use longer timeout for no-parallel mode since sequential compilation takes much longer
    timeout = (
        1800 if args.no_parallel else 600
    )  # 30 minutes for sequential, 10 minutes for parallel

    return RunningProcess(
        cmd, auto_run=False, timeout=timeout, output_formatter=TimestampFormatter()
    )


def create_python_test_process(
    enable_stack_trace: bool, run_slow: bool = False, verbose: bool = False
) -> RunningProcess:
    """Create a Python test process without starting it"""
    # Use list format for better environment handling
    cmd = [
        "uv",
        "run",
        "pytest",
        "--tb=short",  # Shorter traceback format
        "ci/tests",  # Test directory
    ]

    # Add verbose flags only if requested
    if verbose:
        cmd.extend(["-s", "-v"])  # Don't capture stdout/stderr, verbose output

    # Always show test durations
    cmd.append("--durations=0")

    # If running slow tests, add --runslow flag (handled by conftest.py)
    if run_slow:
        cmd.append("--runslow")

    cmd_str = subprocess.list2cmdline(cmd)

    return RunningProcess(
        cmd_str,
        auto_run=False,  # Don't auto-start - will be started in parallel later
        timeout=_TIMEOUT,  # 2 minute timeout for Python tests
        output_formatter=TimestampFormatter(),
    )


def create_integration_test_process(
    args: TestArgs, enable_stack_trace: bool
) -> RunningProcess:
    """Create an integration test process without starting it"""
    cmd = ["uv", "run", "pytest", "-s", "ci/test_integration", "-xvs", "--durations=0"]
    if args.examples is not None:
        # When --examples --full is specified, only run example-related integration tests
        cmd.extend(["-k", "TestFullProgramLinking"])
    if args.verbose:
        cmd.append("-v")
    return RunningProcess(cmd, auto_run=False, output_formatter=TimestampFormatter())


def create_compile_uno_test_process(enable_stack_trace: bool = True) -> RunningProcess:
    """Create a process to compile the uno tests without starting it"""
    cmd = [
        "uv",
        "run",
        "python",
        "-m",
        "ci.ci-compile",
        "uno",
        "--examples",
        "Blink",
        "--no-interactive",
        "--local",
    ]

    # Force local compilation to avoid Docker detection hangs in test environment
    # Docker auto-detection can timeout if Docker is installed but not running

    # Use 15 minute timeout for platform compilation (as per CLAUDE.md guidelines)
    # Docker compilation may take longer during initial rsync sync without producing output
    return RunningProcess(
        cmd, auto_run=False, timeout=900, output_formatter=TimestampFormatter()
    )


def create_wasm_test_process(enable_stack_trace: bool = True) -> RunningProcess:
    """Create a WASM compilation test process without starting it"""
    cmd = [
        "uv",
        "run",
        "python",
        "ci/wasm_compile.py",
        "examples/wasm",
        "--run",
    ]

    # Use 5 minute timeout for WASM compilation and testing
    return RunningProcess(
        cmd, auto_run=False, timeout=300, output_formatter=TimestampFormatter()
    )


def get_cpp_test_processes(
    args: TestArgs, test_categories: TestCategories, enable_stack_trace: bool
) -> list[RunningProcess]:
    """Return all processes needed for C++ tests"""
    processes: list[RunningProcess] = []

    if test_categories.unit:
        processes.append(create_unit_test_process(args, enable_stack_trace))

    if test_categories.examples:
        processes.append(create_examples_test_process(args, enable_stack_trace))

    return processes


def get_python_test_processes(
    enable_stack_trace: bool, run_slow: bool = False, verbose: bool = False
) -> list[RunningProcess]:
    """Return all processes needed for Python tests"""
    return [
        create_python_test_process(False, run_slow, verbose)
    ]  # Disable stack trace for Python tests


def get_integration_test_processes(
    args: TestArgs, enable_stack_trace: bool
) -> list[RunningProcess]:
    """Return all processes needed for integration tests"""
    return [create_integration_test_process(args, enable_stack_trace)]


def get_all_test_processes(
    args: TestArgs,
    test_categories: TestCategories,
    enable_stack_trace: bool,
    src_code_change: bool,
) -> list[RunningProcess]:
    """Return all processes needed for all tests"""
    processes: list[RunningProcess] = []

    # Add test processes based on categories
    if test_categories.unit:
        processes.append(create_unit_test_process(args, enable_stack_trace))
    if test_categories.examples:
        processes.append(create_examples_test_process(args, enable_stack_trace))
    if test_categories.py:
        processes.append(
            create_python_test_process(False, run_slow=False, verbose=args.verbose)
        )  # Disable stack trace for Python tests
    if test_categories.integration:
        processes.append(create_integration_test_process(args, enable_stack_trace))

    # Add uno test process if source code changed
    if src_code_change:
        processes.append(create_compile_uno_test_process(enable_stack_trace))

    return processes


def _extract_test_name(command: str | list[str]) -> str:
    """Extract a human-readable test name from a command"""
    if isinstance(command, list):
        command = " ".join(command)

    # Extract test name patterns
    if "--test " in command:
        # Extract specific test name after --test flag
        parts = command.split("--test ")
        if len(parts) > 1:
            test_name = parts[1].split()[0]
            return test_name
    elif ".exe" in command:
        # Extract from executable name
        for part in command.split():
            if part.endswith(".exe"):
                return part.replace(".exe", "")
    elif "python" in command and "-m " in command:
        # Extract module name
        parts = command.split("-m ")
        if len(parts) > 1:
            module = parts[1].split()[0]
            return module.replace("ci.compiler.", "").replace("ci.", "")
    elif "python" in command and command.endswith(".py"):
        # Extract script name
        for part in command.split():
            if part.endswith(".py"):
                return part.split("/")[-1].replace(".py", "")

    # Fallback to first meaningful part of command
    parts = command.split()
    for part in parts[1:]:  # Skip 'uv' or 'python'
        if not part.startswith("-") and "python" not in part:
            return part

    return "unknown_test"


def _create_skipped_timing(
    test_name: str, command: str = "", cached_summary: str = ""
) -> ProcessTiming:
    """Create a ProcessTiming entry for a skipped test"""
    # Include cached summary in the name if available
    display_name = f"{test_name} ({cached_summary})" if cached_summary else test_name
    return ProcessTiming(name=display_name, duration=0.0, command=command, skipped=True)


def _format_cache_hit_message(
    fingerprint_manager: Optional["FingerprintManager"],
    fingerprint_name: str,
    default_message: str,
) -> str:
    """Format cache hit message with test metadata if available"""
    if fingerprint_manager is None:
        return default_message

    prev_fp = fingerprint_manager.get_prev_fingerprint(fingerprint_name)
    if prev_fp is None:
        return default_message

    summary = prev_fp.get_cache_summary()
    if summary:
        return f"{default_message} [{summary}]"
    return default_message


def _get_friendly_test_name(command: str | list[str]) -> str:
    """Extract a user-friendly test name for display in summary table"""
    if isinstance(command, list):
        command = " ".join(command)

    # Simplify common command patterns to friendly names
    if "cpp_test_run" in command and "ci.run_tests" in command:
        return "unit_tests"
    elif "meson_example_runner.py" in command:
        # Extract specific example names from command (after the .py script)
        tokens = command.split()
        example_names: list[str] = []
        found_script = False
        for tok in tokens:
            # Find the script first
            if "meson_example_runner" in tok:
                found_script = True
                continue
            if not found_script:
                continue
            # Skip flags
            if tok.startswith("-"):
                continue
            # This is likely an example name (after the script)
            example_names.append(tok)
        if example_names:
            return ", ".join(example_names)
        return "examples"
    elif "test_example_compilation.py" in command:
        # Show script name plus example targets, e.g. "test_example_compilation.py Luminova"
        try:
            import os

            tokens = command.split()
            # Find the script token and collect following non-flag args as examples
            for i, tok in enumerate(tokens):
                normalized = tok.strip('"')
                if normalized.endswith("test_example_compilation.py"):
                    script_name = os.path.basename(normalized)
                    example_parts: list[str] = []
                    for t in tokens[i + 1 :]:
                        if t.startswith("-"):
                            break
                        example_parts.append(t.strip('"'))
                    if example_parts:
                        return f"{script_name} {' '.join(example_parts)}"
                    return script_name
        except KeyboardInterrupt as ki:
            handle_keyboard_interrupt(ki)
            raise
        except Exception:
            # Fall back to generic extraction on any unexpected parsing issue
            pass
        return _extract_test_name(command)
    elif "pytest" in command and "ci/tests" in command:
        return "python_tests"
    elif "pytest" in command and "ci/test_integration" in command:
        return "integration_tests"
    elif "ci-compile" in command and "uno" in command:
        return "uno_compilation"
    elif "wasm_compile.py" in command:
        return "wasm_compilation"
    else:
        # Fallback to the existing extraction logic
        return _extract_test_name(command)


@dataclass
class FailedTestEntry:
    """A test that failed during a test run."""

    name: str
    category: str  # "unit" or "example"

    @property
    def clean_name(self) -> str:
        """Strip suite prefix (e.g., 'fastled:test_name' -> 'test_name')."""
        return self.name.split(":")[-1] if ":" in self.name else self.name

    @property
    def rerun_cmd(self) -> str:
        if self.category == "example":
            return f"bash test {self.clean_name} --examples"
        return f"bash test {self.clean_name} --cpp"

    @property
    def debug_cmd(self) -> str:
        if self.category == "example":
            return f"bash test {self.clean_name} --examples --debug"
        return f"bash test {self.clean_name} --debug"


def _format_failure_summary(
    failed_tests: list[FailedTestEntry],
) -> str:
    """Format a summary table of failed tests with re-run commands.

    Args:
        failed_tests: List of FailedTestEntry objects.

    Returns:
        Formatted string with failure summary table.
    """
    if not failed_tests:
        return ""

    lines: list[str] = []
    lines.append("")
    lines.append("\033[91m" + "=" * 60 + "\033[0m")
    lines.append("\033[91mFAILED TESTS SUMMARY\033[0m")
    lines.append("\033[91m" + "=" * 60 + "\033[0m")

    # Table header
    max_name = max(len(ft.clean_name) for ft in failed_tests)
    max_name = max(max_name, 4)  # min width for "Test"
    header = f"  {'Test':<{max_name}} | {'Category':<8} | Re-run command"
    sep = f"  {'-' * max_name}-+-{'-' * 8}-+-{'-' * 30}"
    lines.append(header)
    lines.append(sep)

    for ft in failed_tests:
        lines.append(
            f"  {ft.clean_name:<{max_name}} | {ft.category:<8} | {ft.rerun_cmd}"
        )

    lines.append("")
    lines.append("To re-run failed tests:")
    for ft in failed_tests:
        lines.append(f"  {ft.rerun_cmd}")
    lines.append("")

    # Loud banner for debug re-run — this is critical for AI agents to notice
    max_debug_show = 3
    lines.append("\033[93m" + "!" * 60 + "\033[0m")
    lines.append(
        "\033[93m!!  AGENT ACTION REQUIRED: RE-RUN FAILED TESTS IN DEBUG  !!\033[0m"
    )
    lines.append("\033[93m" + "!" * 60 + "\033[0m")
    lines.append("")
    lines.append("Quick mode does NOT enable sanitizers (ASAN/LSAN/UBSAN).")
    lines.append("You MUST re-run each failed test in --debug mode to get")
    lines.append("full diagnostics (stack traces, memory errors, UB detection).")
    lines.append("")
    lines.append("\033[93mRun these commands NOW:\033[0m")
    for ft in failed_tests[:max_debug_show]:
        lines.append(f"  \033[1m{ft.debug_cmd}\033[0m")
    if len(failed_tests) > max_debug_show:
        lines.append(
            f"  ... and {len(failed_tests) - max_debug_show} more failed tests"
        )
    lines.append("")
    lines.append("\033[93m" + "!" * 60 + "\033[0m")
    lines.append("")

    return "\n".join(lines)


def _format_timing_summary(process_timings: list[ProcessTiming]) -> str:
    """Format a summary of process execution times with optional phase breakdown.

    Always returns a table format for consistent display.
    """
    if not process_timings:
        return ""

    # Sort by skipped status first (non-skipped first), then by duration (longest first)
    sorted_timings = sorted(process_timings, key=lambda x: (x.skipped, -x.duration))

    # Check if any timing has phase data (for Meson tests)
    has_phases = any(
        t.meson_setup_time and t.meson_setup_time > 0 for t in sorted_timings
    )

    # Build timing rows (main tests + phases if available)
    timing_rows: list[tuple[str, str]] = []

    # First pass: add all main test entries
    for timing in sorted_timings:
        if timing.skipped:
            duration_str = "skipped"
        else:
            duration_str = f"{timing.duration:.2f}s"
        timing_rows.append((timing.name, duration_str))

        # Second pass: add phase breakdown if available
        if has_phases and not timing.skipped:
            phases = []
            if timing.meson_setup_time and timing.meson_setup_time > 0:
                phases.append(("  ├─ Meson Setup", f"{timing.meson_setup_time:.2f}s"))
            if timing.compile_time and timing.compile_time > 0:
                # Check if we have compile sub-phases
                has_sub_phases = any(
                    [
                        timing.compile_core_time and timing.compile_core_time > 0,
                        timing.compile_objects_time and timing.compile_objects_time > 0,
                        timing.compile_example_link_time
                        and timing.compile_example_link_time > 0,
                        timing.compile_test_link_time
                        and timing.compile_test_link_time > 0,
                    ]
                )
                phases.append(("  ├─ Compilation", f"{timing.compile_time:.2f}s"))
                if has_sub_phases:
                    sub = []
                    if timing.compile_core_time and timing.compile_core_time > 0:
                        sub.append(
                            ("  │   ├─ Core + PCH", f"{timing.compile_core_time:.2f}s")
                        )
                    if timing.compile_objects_time and timing.compile_objects_time > 0:
                        sub.append(
                            (
                                "  │   ├─ Object files",
                                f"{timing.compile_objects_time:.2f}s",
                            )
                        )
                    if (
                        timing.compile_example_link_time
                        and timing.compile_example_link_time > 0
                    ):
                        sub.append(
                            (
                                "  │   ├─ Example linking",
                                f"{timing.compile_example_link_time:.2f}s",
                            )
                        )
                    if (
                        timing.compile_test_link_time
                        and timing.compile_test_link_time > 0
                    ):
                        sub.append(
                            (
                                "  │   └─ Test linking",
                                f"{timing.compile_test_link_time:.2f}s",
                            )
                        )
                    # Fix last sub-phase connector
                    if sub:
                        name, dur = sub[-1]
                        sub[-1] = (name.replace("  │   ├─", "  │   └─"), dur)
                    phases.extend(sub)
            if timing.test_execution_time and timing.test_execution_time > 0:
                phases.append(
                    ("  └─ Test Execution", f"{timing.test_execution_time:.2f}s")
                )

            # Add phase rows
            for phase_name, phase_duration in phases:
                timing_rows.append((phase_name, phase_duration))

    # Calculate column widths
    max_name_width = max(len(name) for name, _ in timing_rows)
    max_name_width = max(max_name_width, len("Test"))

    max_duration_width = max(len(duration) for _, duration in timing_rows)
    max_duration_width = max(max_duration_width, len("Duration"))

    # Create header with compact formatting
    header = f"{'Test':<{max_name_width}} | {'Duration':<{max_duration_width}}"
    separator = f"{'-' * max_name_width}-+-{'-' * max_duration_width}"

    # Create rows with compact formatting
    rows: list[str] = []
    for name, duration_str in timing_rows:
        row = f"{name:<{max_name_width}} | {duration_str:<{max_duration_width}}"
        rows.append(row)

    # Combine all parts (header before separator, no bottom border)
    table_lines = [
        "Results:",
        header,
        separator,
    ] + rows

    return "\n".join(table_lines)


def _handle_process_completion(
    proc_state: ProcessState,
    active_processes: list[RunningProcess],
    completed_timings: list[ProcessTiming],
    last_activity_time: dict[RunningProcess, float],
) -> None:
    """
    Handle completion of a single test process

    Args:
        proc_state: State information for the completed process
        active_processes: List of currently active processes
        completed_timings: List to collect timing data
        last_activity_time: Dictionary tracking activity times

    Raises:
        TestExecutionFailedException: If process failed
    """
    proc = proc_state.process
    cmd = proc_state.command
    if isinstance(cmd, list):
        cmd = subprocess.list2cmdline(cmd)

    try:
        returncode = proc.wait()
        if returncode != 0:
            test_name = _extract_test_name(cmd)
            ts_print(f"\nCommand failed: {cmd} with return code {returncode}")
            ts_print("\033[91m###### ERROR ######\033[0m")
            ts_print(f"Test failed: {test_name}")

            # Capture the actual output from the failed process
            try:
                actual_output = proc.stdout
                if actual_output.strip():
                    ts_print("\n=== ACTUAL OUTPUT FROM FAILED PROCESS ===")
                    ts_print(actual_output)
                    ts_print("=== END OF OUTPUT ===")
                else:
                    actual_output = "No output captured from failed process"
            except KeyboardInterrupt as ki:
                handle_keyboard_interrupt(ki)
                raise
            except Exception as e:
                actual_output = f"Error capturing output: {e}"

            # Flush output for real-time visibility (but avoid on Windows due to blocking issues)
            if os.name != "nt":  # Only flush on non-Windows systems
                sys.stdout.flush()
            for p in active_processes:
                if p != proc:
                    p.kill()
            failure = TestFailureInfo(
                test_name=test_name,
                command=str(cmd),
                return_code=returncode,
                output=actual_output,
                error_type="command_failure",
            )
            raise TestExecutionFailedException("Test command failed", [failure])

        active_processes.remove(proc)
        if proc in last_activity_time:
            del last_activity_time[proc]  # Clean up tracking
        # Process completion is shown in the timing summary table, no separate message needed

        # Collect timing data for summary
        if proc.duration is not None:
            timing = ProcessTiming(
                name=_get_friendly_test_name(cmd),
                duration=proc.duration,
                command=str(cmd),
            )
            completed_timings.append(timing)

        # Flush output for real-time visibility (but avoid on Windows due to blocking issues)
        if os.name != "nt":  # Only flush on non-Windows systems
            sys.stdout.flush()

    except KeyboardInterrupt as ki:
        handle_keyboard_interrupt(ki)
        raise
    except Exception as e:
        test_name = _extract_test_name(cmd)
        ts_print(f"\nError waiting for process: {cmd}")
        ts_print(f"Error: {e}")
        ts_print("\033[91m###### ERROR ######\033[0m")
        ts_print(f"Test error: {test_name}")

        # Try to capture any available output
        try:
            actual_output = proc.stdout
            if actual_output.strip():
                ts_print("\n=== PROCESS OUTPUT BEFORE ERROR ===")
                ts_print(actual_output)
                ts_print("=== END OF OUTPUT ===")
        except KeyboardInterrupt as ki:
            handle_keyboard_interrupt(ki)
            raise
        except Exception as output_error:
            ts_print(f"Could not capture process output: {output_error}")

        failures: list[TestFailureInfo] = []
        for p in active_processes:
            p.kill()
            # Try to capture output from this process too
            try:
                process_output = p.stdout if hasattr(p, "stdout") else str(e)
            except KeyboardInterrupt as ki:
                handle_keyboard_interrupt(ki)
                raise
            except Exception:
                process_output = str(e)

            failures.append(
                TestFailureInfo(
                    test_name=_extract_test_name(p.command),
                    command=str(p.command),
                    return_code=1,
                    output=process_output,
                    error_type="process_wait_error",
                )
            )
        raise TestExecutionFailedException("Error waiting for process", failures)


@dataclass
class StuckProcessSignal:
    """Signal from monitoring thread that a process is stuck"""

    process: RunningProcess
    timeout_duration: float


class ProcessStuckMonitor:
    """Manages individual monitoring threads for stuck process detection"""

    def __init__(self, stuck_process_timeout: float):
        self.stuck_process_timeout = stuck_process_timeout
        self.stuck_signals: Queue[StuckProcessSignal] = Queue()
        self.monitoring_threads: dict[RunningProcess, threading.Thread] = {}
        self.shutdown_event = threading.Event()

    def start_monitoring(self, process: RunningProcess) -> None:
        """Start a monitoring thread for the given process"""
        if process in self.monitoring_threads:
            return  # Already monitoring this process

        monitor_thread = threading.Thread(
            target=self._monitor_process,
            args=(process,),
            name=f"StuckMonitor-{_extract_test_name(process.command)}",
            daemon=True,
        )
        self.monitoring_threads[process] = monitor_thread
        monitor_thread.start()

    def stop_monitoring(self, process: RunningProcess) -> None:
        """Stop monitoring a specific process"""
        if process in self.monitoring_threads:
            del self.monitoring_threads[process]

    def shutdown(self) -> None:
        """Shutdown all monitoring threads"""
        self.shutdown_event.set()
        self.monitoring_threads.clear()

    def check_for_stuck_processes(self) -> list[StuckProcessSignal]:
        """Check for any stuck process signals from monitoring threads"""
        stuck_processes: list[StuckProcessSignal] = []
        try:
            while True:
                signal = self.stuck_signals.get_nowait()
                stuck_processes.append(signal)
        except queue.Empty:
            pass
        return stuck_processes

    def _monitor_process(self, process: RunningProcess) -> None:
        """Monitor a single process for being stuck (runs in separate thread)"""
        thread_id = threading.current_thread().ident
        thread_name = threading.current_thread().name

        try:
            last_activity_time = time.time()

            while not self.shutdown_event.is_set():
                if process.finished:
                    # Process completed normally, stop monitoring
                    return

                # Check if we have recent stdout activity
                if process.has_pending_output():
                    # Drain pending output to detect ongoing activity
                    while process.has_pending_output():
                        process.get_next_line_non_blocking()
                    last_activity_time = time.time()

                # Check if process is stuck
                current_time = time.time()
                if current_time - last_activity_time > self.stuck_process_timeout:
                    # Process is stuck, signal the main thread
                    signal = StuckProcessSignal(
                        process=process, timeout_duration=self.stuck_process_timeout
                    )
                    self.stuck_signals.put(signal)
                    return

                # Sleep briefly before next check
                time.sleep(1.0)  # Check every second

        except KeyboardInterrupt as ki:
            handle_keyboard_interrupt(ki)
        except Exception as e:
            # Monitor thread hit an unexpected error (e.g., race with process
            # cleanup). Log it but do NOT interrupt the main thread — this is a
            # monitoring thread, not a critical path. Interrupting main here
            # causes spurious "Interrupt signal received" on normal exits.
            ts_print(f"❌ Thread {thread_id} ({thread_name}) unexpected error: {e}")
            import traceback  # noqa: PLC0415 - lazy: only imported on error path

            traceback.print_exc()


def _handle_stuck_processes(
    stuck_signals: list[StuckProcessSignal],
    active_processes: list[RunningProcess],
    failed_processes: list[str],
    monitor: ProcessStuckMonitor,
) -> None:
    """
    Handle stuck processes reported by monitoring threads

    Args:
        stuck_signals: List of stuck process signals from monitoring threads
        active_processes: List of currently active processes (modified in place)
        failed_processes: List to track failed process commands
        monitor: The process stuck monitor instance
    """
    for signal in stuck_signals:
        proc = signal.process
        if proc in active_processes:
            ts_print(
                f"\nProcess appears stuck (no output for {signal.timeout_duration}s): {proc.command}"
            )
            ts_print("Killing stuck process and its children...")
            proc.kill()  # This now kills the entire process tree

            # Track this as a failure
            failed_processes.append(subprocess.list2cmdline(proc.command))

            active_processes.remove(proc)
            monitor.stop_monitoring(proc)
            ts_print(f"Killed stuck process: {proc.command}")


def run_test_processes(
    processes: list[RunningProcess], parallel: bool = True, verbose: bool = False
) -> list[ProcessTiming]:
    """
    Run multiple test processes using RunningProcessGroup

    Args:
        processes: List of RunningProcess objects to execute
        parallel: Whether to run processes in parallel or sequentially (ignored if NO_PARALLEL is set)
        verbose: Whether to show all output

    Returns:
        List of ProcessTiming objects with execution times
    """
    from ci.util.running_process_group import (
        ExecutionMode,
        ProcessExecutionConfig,
        RunningProcessGroup,
    )

    start_time = time.time()

    # Force sequential execution if NO_PARALLEL is set
    if os.environ.get("NO_PARALLEL"):
        parallel = False

    if not processes:
        ts_print("\n✓ No tests to run")
        return []

    try:
        # Configure execution mode
        execution_mode = (
            ExecutionMode.PARALLEL if parallel else ExecutionMode.SEQUENTIAL
        )

        config = ProcessExecutionConfig(
            execution_mode=execution_mode,
            verbose=verbose,
            max_failures_before_abort=MAX_FAILURES_BEFORE_ABORT,
            enable_stuck_detection=True,
            stuck_timeout_seconds=_GLOBAL_TIMEOUT,
            live_updates=True,  # Enable real-time display
            display_type="auto",  # Auto-detect best display format
        )

        # Create and run process group
        group = RunningProcessGroup(
            processes=processes, config=config, name="TestProcesses"
        )

        # Start real-time display if we have processes and live updates enabled
        display_thread = None
        if len(processes) > 0 and config.live_updates and not verbose:
            # Only show live display if not in verbose mode (verbose already shows all output)
            try:
                from ci.util.process_status_display import display_process_status

                display_thread = display_process_status(
                    group,
                    display_type=config.display_type,
                    update_interval=config.update_interval,
                )
            except ImportError:
                pass  # Fall back to normal execution if display not available
            except KeyboardInterrupt as ki:
                handle_keyboard_interrupt(ki)
                raise
            except Exception:
                pass  # Fall back to normal execution on any display error

        timings = group.run()

        # Stop display thread if it was started
        if display_thread:
            # Give it a moment to show final status
            time.sleep(0.5)

        # Success message for sequential execution
        if not parallel:
            elapsed = time.time() - start_time
            ts_print(f"\n✓ Tests completed ({elapsed:.2f}s)")

        return timings

    except (TestExecutionFailedException, TestTimeoutException) as e:
        # Tests failed - print detailed info and re-raise for proper handling
        ts_print("\n" + "=" * 60)
        ts_print("FASTLED TEST RUNNER FAILURE DETAILS")
        ts_print("=" * 60)
        ts_print(e.get_detailed_failure_info())
        ts_print("=" * 60)
        raise
    except SystemExit as e:
        # Tests failed - extract command name from the error
        if e.code != 0:
            ts_print("\033[91m###### ERROR ######\033[0m")
            ts_print(f"Tests failed with exit code {e.code}")
        raise


def runner(
    args: TestArgs,
    src_code_change: bool = True,
    cpp_test_change: bool = True,
    examples_change: bool = True,
    python_test_change: bool = True,
    wasm_change: bool = True,
    fingerprint_manager: Optional["FingerprintManager"] = None,
) -> None:
    """
    Main test runner function that determines what to run and executes tests

    Args:
        args: Parsed command line arguments
        src_code_change: Whether source code has changed since last run
        cpp_test_change: Whether C++ test-related files (src/ or tests/) have changed
        examples_change: Whether example-related files have changed
        python_test_change: Whether Python test-related files have changed
        wasm_change: Whether WASM-related files have changed
        fingerprint_manager: Optional fingerprint manager for test metadata
    """
    # Debug logging - only shown in verbose mode to reduce UI clutter
    if args.verbose:
        # Show only active/non-default flags instead of full TestArgs dump
        active_flags: list[str] = []
        if args.test:
            active_flags.append(f"test={args.test}")
        if args.examples:
            # Format list as comma-separated values instead of Python list syntax
            active_flags.append(f"examples={','.join(args.examples)}")
        if args.debug:
            active_flags.append("debug")
        if args.clean:
            active_flags.append("clean")
        if args.full:
            active_flags.append("full")
        if args.no_parallel:
            active_flags.append("no-parallel")
        if args.build_mode:
            active_flags.append(f"build_mode={args.build_mode}")
        if args.no_fingerprint:
            active_flags.append("no-fingerprint")
        if args.no_pch:
            active_flags.append("no-pch")

        if active_flags:
            ts_print(f"Active flags: {', '.join(active_flags)}")

    # Clear zccache stats at start to show only metrics from this build
    from ci.util.zccache_config import clear_zccache_stats

    clear_zccache_stats()

    # Determine test categories first to check if we should use meson
    test_categories = determine_test_categories(args)

    # Always use Meson build system for unit tests (Python build system has been removed)
    # Only return early if unit tests are the ONLY thing running (unit_only mode)
    # Skip if fingerprint cache indicates no changes
    if test_categories.unit_only:
        if cpp_test_change:
            from ci.meson.runner import run_meson_build_and_test
            from ci.util.paths import PROJECT_ROOT

            build_dir = PROJECT_ROOT / ".build" / "meson"
            test_name = args.test if args.test else None

            # Extract .hpp filename if user requested a specific detector file
            test_file_filter = None
            if args.raw_test_query and args.raw_test_query.endswith(".hpp"):
                # Extract just the filename (e.g., "backbeat.hpp" from "tests/fl/audio/detectors/backbeat.hpp")
                test_file_filter = args.raw_test_query.split("/")[-1]

            result = run_meson_build_and_test(
                source_dir=PROJECT_ROOT,
                build_dir=build_dir,
                test_name=test_name,
                clean=args.clean,
                verbose=args.verbose,
                debug=args.debug,
                build_mode=args.build_mode,
                check=args.check,
                exclude_suites=[
                    "fastled:examples"
                ],  # Exclude examples in unit-test-only mode
                test_file_filter=test_file_filter,
                log_failures=args.log_failures,
            )

            if not result.success:
                if result.failed_test_names:
                    summary = _format_failure_summary(
                        [
                            FailedTestEntry(name=name, category="unit")
                            for name in result.failed_test_names
                        ]
                    )
                    print(summary)
                sys.exit(1)

            # Update fingerprint metadata for cache display on next run
            if fingerprint_manager is not None:
                fingerprint_manager.update_test_metadata(
                    "cpp_test",
                    num_tests_run=result.num_tests_run,
                    num_tests_passed=result.num_tests_passed,
                    duration_seconds=result.duration,
                    test_name="cpp_unit_tests",
                )

            # Print timing summary table for unit-only mode
            # Skip zccache stats when test result came from cache (no compilation occurred)
            if not result.compilation_skipped:
                from ci.util.zccache_config import show_zccache_stats  # noqa: PLC0415

                show_zccache_stats()
            unit_timing = ProcessTiming(
                name=f"cpp_unit_tests ({result.num_tests_passed}/{result.num_tests_run} passed)",
                duration=result.duration,
                command="meson test",
                skipped=False,
                meson_setup_time=result.meson_setup_time,
                ninja_maintenance_time=result.ninja_maintenance_time,
                compile_time=result.compile_time,
                test_execution_time=result.test_execution_time,
                compile_core_time=result.compile_core_time,
                compile_objects_time=result.compile_objects_time,
                compile_example_link_time=result.compile_example_link_time,
                compile_test_link_time=result.compile_test_link_time,
            )
            summary = _format_timing_summary([unit_timing])
            print(summary)
        else:
            # Fingerprint cache hit - skip unit tests
            cache_msg = _format_cache_hit_message(
                fingerprint_manager,
                "cpp_test",
                "Fingerprint cache valid - skipping C++ unit tests (no changes detected)",
            )
            print_cache_hit(cache_msg)

        return

    # For mixed test modes (unit + examples, etc.), run unit tests via Meson but continue to other tests
    # Skip if fingerprint cache indicates no changes
    # Track Meson test timing for summary
    meson_test_timing: Optional[ProcessTiming] = None

    if test_categories.unit and not test_categories.unit_only:
        if cpp_test_change:
            from ci.meson.runner import run_meson_build_and_test
            from ci.util.paths import PROJECT_ROOT

            build_dir = PROJECT_ROOT / ".build" / "meson"
            test_name = args.test if args.test else None

            # Extract .hpp filename if user requested a specific detector file
            test_file_filter = None
            if args.raw_test_query and args.raw_test_query.endswith(".hpp"):
                # Extract just the filename (e.g., "backbeat.hpp" from "tests/fl/audio/detectors/backbeat.hpp")
                test_file_filter = args.raw_test_query.split("/")[-1]

            result = run_meson_build_and_test(
                source_dir=PROJECT_ROOT,
                build_dir=build_dir,
                test_name=test_name,
                clean=args.clean,
                verbose=args.verbose,
                debug=args.debug,
                build_mode=args.build_mode,
                check=args.check,
                test_file_filter=test_file_filter,
                log_failures=args.log_failures,
            )

            # Create timing entry for summary
            meson_test_timing = ProcessTiming(
                name=f"cpp_unit_tests ({result.num_tests_passed}/{result.num_tests_run} passed)",
                duration=result.duration,
                command="meson test",
                skipped=False,
                meson_setup_time=result.meson_setup_time,
                ninja_maintenance_time=result.ninja_maintenance_time,
                compile_time=result.compile_time,
                test_execution_time=result.test_execution_time,
                compile_core_time=result.compile_core_time,
                compile_objects_time=result.compile_objects_time,
                compile_example_link_time=result.compile_example_link_time,
                compile_test_link_time=result.compile_test_link_time,
            )

            # Update fingerprint metadata for cache display on next run
            if fingerprint_manager is not None:
                fingerprint_manager.update_test_metadata(
                    "cpp_test",
                    num_tests_run=result.num_tests_run,
                    num_tests_passed=result.num_tests_passed,
                    duration_seconds=result.duration,
                    test_name="cpp_unit_tests",
                )

            if not result.success:
                if result.failed_test_names:
                    summary = _format_failure_summary(
                        [
                            FailedTestEntry(name=name, category="unit")
                            for name in result.failed_test_names
                        ]
                    )
                    print(summary)
                sys.exit(1)

            # Save full-run cache after successful complete test suite run
            # so the CASE 2 ultra-early exit can fire on next invocation.
            if result.num_tests_run and result.num_tests_run == result.num_tests_passed:
                try:
                    from ci.meson.cache_utils import save_full_run_result
                    from ci.util.paths import PROJECT_ROOT

                    _build_mode_save = getattr(args, "build_mode", None) or (
                        "debug" if getattr(args, "debug", False) else "quick"
                    )
                    _build_dir_save = (
                        PROJECT_ROOT / ".build" / f"meson-{_build_mode_save}"
                    )
                    save_full_run_result(
                        _build_dir_save,
                        result.num_tests_passed,
                        result.num_tests_run,
                        result.duration,
                    )
                except KeyboardInterrupt as ki:
                    handle_keyboard_interrupt(ki)
                except Exception:
                    pass  # Non-critical
        else:
            # Fingerprint cache hit - skip unit tests
            cache_msg = _format_cache_hit_message(
                fingerprint_manager,
                "cpp_test",
                "Fingerprint cache valid - skipping C++ unit tests (no changes detected)",
            )
            print_cache_hit(cache_msg)
            # Get cached summary for the skipped timing display
            cached_summary = ""
            if fingerprint_manager:
                prev_fp = fingerprint_manager.get_prev_fingerprint("cpp_test")
                if prev_fp:
                    cached_summary = prev_fp.get_cache_summary()
            meson_test_timing = _create_skipped_timing(
                "cpp_unit_tests", cached_summary=cached_summary
            )

        # Continue to run other tests (examples, Python, etc.) - don't return

    try:
        # Test categories already determined above (for meson check)
        # Handle stack trace flags: --stack-trace overrides --no-stack-trace, default is enabled
        if args.stack_trace:
            enable_stack_trace = True
        elif args.no_stack_trace:
            enable_stack_trace = False
        else:
            enable_stack_trace = True  # Default: enabled

        # Build up unified list of all processes to run
        processes: list[RunningProcess] = []
        skipped_timings: list[ProcessTiming] = []

        # Note: Unit tests are now exclusively handled by Meson (see lines 1412-1450)
        # This avoids the call to the removed ci.compiler.cpp_test_run module

        # Add integration tests if needed
        if test_categories.integration or test_categories.integration_only:
            processes.append(create_integration_test_process(args, enable_stack_trace))

        # Add example compilation test if examples fingerprint indicates change
        # The examples_change parameter respects fingerprint caching:
        # - False when fingerprint matches (no code changes in examples/ or src/)
        # - True when fingerprint differs OR user explicitly requested examples
        #
        # Skip the separate examples process when run_meson_build_and_test()
        # already compiled and executed examples in a single Ninja invocation
        # (mixed mode: unit + examples with cpp_test_change).
        examples_already_handled = (
            test_categories.unit and not test_categories.unit_only and cpp_test_change
        )
        should_compile_examples = (
            examples_change
            and not test_categories.py_only
            and not examples_already_handled
        )
        if should_compile_examples:
            processes.append(create_examples_test_process(args, enable_stack_trace))

        # Add Python tests if needed and Python test files have changed
        if (test_categories.py or test_categories.py_only) and python_test_change:
            # Pass run_slow=True if we're running integration tests or any form of full tests
            run_slow = (
                test_categories.integration
                or test_categories.integration_only
                or args.full
            )

            processes.append(
                create_python_test_process(False, run_slow, verbose=args.verbose)
            )  # Disable stack trace for Python tests
        elif (test_categories.py or test_categories.py_only) and not python_test_change:
            cache_msg = _format_cache_hit_message(
                fingerprint_manager,
                "python_test",
                "Fingerprint cache valid - skipping Python tests (no changes detected)",
            )
            print_cache_hit(cache_msg)
            # Get cached summary for the skipped timing display
            cached_summary = ""
            if fingerprint_manager:
                prev_fp = fingerprint_manager.get_prev_fingerprint("python_test")
                if prev_fp:
                    cached_summary = prev_fp.get_cache_summary()
            skipped_timings.append(
                _create_skipped_timing("python_tests", cached_summary=cached_summary)
            )

        # Add WASM compilation test if WASM is enabled and files have changed
        # WASM runs by default unless specific test flags are used
        if (test_categories.wasm or test_categories.wasm_only) and wasm_change:
            processes.append(create_wasm_test_process(enable_stack_trace))
        elif (test_categories.wasm or test_categories.wasm_only) and not wasm_change:
            cache_msg = _format_cache_hit_message(
                fingerprint_manager,
                "wasm",
                "Fingerprint cache valid - skipping WASM compilation (no changes detected)",
            )
            print_cache_hit(cache_msg)
            # Get cached summary for the skipped timing display
            cached_summary = ""
            if fingerprint_manager:
                prev_fp = fingerprint_manager.get_prev_fingerprint("wasm")
                if prev_fp:
                    cached_summary = prev_fp.get_cache_summary()
            skipped_timings.append(
                _create_skipped_timing("wasm", cached_summary=cached_summary)
            )

        # Examples are now compiled via uno compilation test (ci-compile.py)
        # The uno compilation test above already covers example compilation
        # So we just track skipped status if needed
        if (
            test_categories.examples or test_categories.examples_only
        ) and not examples_change:
            cache_msg = _format_cache_hit_message(
                fingerprint_manager,
                "examples",
                "Fingerprint cache valid - skipping examples (no changes detected)",
            )
            print_cache_hit(cache_msg)
            # Get cached summary for the skipped timing display
            cached_summary = ""
            if fingerprint_manager:
                prev_fp = fingerprint_manager.get_prev_fingerprint("examples")
                if prev_fp:
                    cached_summary = prev_fp.get_cache_summary()
            skipped_timings.append(
                _create_skipped_timing("examples", cached_summary=cached_summary)
            )

        # Determine if we'll run in parallel
        will_run_parallel = not bool(os.environ.get("NO_PARALLEL"))

        # Print summary of what we're about to run

        # Run processes (parallel unless NO_PARALLEL is set)
        timings = run_test_processes(
            processes,
            parallel=will_run_parallel,
            verbose=args.verbose,
        )

        # Display timing summary
        all_timings = timings + skipped_timings
        # Add Meson test timing if it was run
        if meson_test_timing:
            all_timings.insert(0, meson_test_timing)  # Put unit tests first
        if all_timings:
            # Show zccache statistics only when actual compilation occurred.
            # Skip when all tests were fingerprint-cached (no subprocess compilation ran).
            # timings is non-empty when at least one process (examples, python, wasm) ran.
            # meson_test_timing.skipped=False when meson tests actually compiled+ran.
            _did_compile = bool(timings) or (
                meson_test_timing is not None and not meson_test_timing.skipped
            )
            if _did_compile:
                from ci.util.zccache_config import show_zccache_stats  # noqa: PLC0415

                show_zccache_stats()
            else:
                # All tests were fingerprint-cached — refresh the full run cache so
                # the next invocation's CASE 2 ultra-early exit can fire even if
                # meson has reconfigured (updated build.ninja) since the last full run.
                # We re-save the previous result with the current build state.
                if not getattr(args, "check", False):
                    try:
                        import json as _json

                        from ci.meson.cache_utils import (
                            get_full_run_cache_file,
                            save_full_run_result,
                        )
                        from ci.util.paths import PROJECT_ROOT

                        _build_mode_tr = getattr(args, "build_mode", None) or (
                            "debug" if getattr(args, "debug", False) else "quick"
                        )
                        _build_dir_tr = (
                            PROJECT_ROOT / ".build" / f"meson-{_build_mode_tr}"
                        )
                        _cache_file_tr = get_full_run_cache_file(_build_dir_tr)
                        if _cache_file_tr.exists():
                            _saved_tr = _json.loads(_cache_file_tr.read_text())
                            if _saved_tr.get("result") == "pass":
                                save_full_run_result(
                                    _build_dir_tr,
                                    _saved_tr.get("num_passed", 0),
                                    _saved_tr.get("num_tests", 0),
                                    _saved_tr.get("duration", 0.0),
                                )
                    except KeyboardInterrupt as ki:
                        handle_keyboard_interrupt(ki)
                    except Exception:
                        pass  # Non-critical optimization; fail silently
            summary = _format_timing_summary(all_timings)
            # Use print() instead of ts_print() to avoid orphan timestamps
            # before multi-line content (the summary starts with \n)
            print(summary)
    except (TestExecutionFailedException, TestTimeoutException) as e:
        # Print failure summary table from exception details
        if e.failures:
            failed_tests: list[FailedTestEntry] = []
            for failure in e.failures:
                # Determine category from test name or command
                category = (
                    "example" if "example" in failure.test_name.lower() else "unit"
                )
                failed_tests.append(
                    FailedTestEntry(name=failure.test_name, category=category)
                )
            if failed_tests:
                summary = _format_failure_summary(failed_tests)
                print(summary)

        # Print summary and exit with proper code
        ts_print("\033[91m###### ERROR ######\033[0m")
        ts_print(f"Tests failed with {len(e.failures)} failure(s)")

        # Exit with appropriate code
        if e.failures:
            # Use the return code from the first failure, or 1 if none available
            exit_code = (
                e.failures[0].return_code if e.failures[0].return_code != 0 else 1
            )
        else:
            exit_code = 1
        sys.exit(exit_code)
