Function _handle_streaming_failure

ci/meson/streaming_runner.py:200–250

Print failure details and write logs after the streaming path failed.

(
    ctx: StreamingContext,
    sr,
    failed_outputs: dict[str, str],
    failed_outputs_lock: threading.Lock,
    duration: float,
    num_tests_run: int,
)

def _handle_streaming_failure(
    ctx: StreamingContext,
    sr,
    failed_outputs: dict[str, str],
    failed_outputs_lock: threading.Lock,
    duration: float,
    num_tests_run: int,
) -> MesonTestResult:
    """Print failure details and write logs after the streaming path failed."""
    print_error(
        f"[MESON] ❌ Some tests failed ({sr.num_passed}/{num_tests_run} tests in {duration:.2f}s)"
    )
    # Snapshot under lock — workers may still be running after shutdown(wait=False).
    with failed_outputs_lock:
        failed_snapshot = dict(failed_outputs)

    if failed_snapshot:
        print_error(f"\n{'=' * 80}")
        print_error(f"[MESON] Test failure details ({len(failed_snapshot)} tests):")
        print_error(f"{'=' * 80}")
        for tname, output in failed_snapshot.items():
            print_error(f"\n--- {tname} ---")
            for line in output.splitlines()[-30:]:
                print_error(f"  {line}")
        print_error(f"{'=' * 80}")

    if ctx.log_failures is not None:
        if num_tests_run == 0:
            per_target = _extract_compile_failures(sr.compile_output)
            if per_target:
                for tname, output in per_target.items():
                    _write_failure_log(ctx.log_failures, tname, "compile", output)
            else:
                _write_failure_log(
                    ctx.log_failures,
                    "compile",
                    "compile",
                    sr.compile_output,
                )
        else:
            for tname, output in failed_snapshot.items():
                _write_failure_log(ctx.log_failures, tname, "run", output)

    return MesonTestResult(
        success=False,
        duration=duration,
        num_tests_run=num_tests_run,
        num_tests_passed=sr.num_passed,
        num_tests_failed=sr.num_failed,
        failed_test_names=sr.failed_names,
    )


def run_streaming_path(ctx: StreamingContext) -> MesonTestResult: