guidellm.benchmark.analysis.knee

def analyze_knee(benchmarks: Sequence[GenerativeBenchmark]) -> KneeAnalysis:
    """Analyze concurrent benchmarks using throughput and temporal evidence.

    :param benchmarks: Completed concurrent GuideLLM benchmarks
    :return: Throughput fit, detector boundary, and their combined assessment
    """
    concurrencies = [benchmark_concurrency(benchmark) for benchmark in benchmarks]
    throughputs = [
        benchmark.metrics.output_tokens_per_second.successful.mean
        for benchmark in benchmarks
    ]
    throughput_result = find_throughput_knee(concurrencies, throughputs)
    saturation_result = extract_saturation_boundary(benchmarks)
    return KneeAnalysis(
        throughput=throughput_result,
        saturation=saturation_result,
        assessment=assess_saturation(throughput_result, saturation_result),
    )

def assess_saturation(
    throughput: KneeResult,
    saturation: SaturationBoundary,
) -> SaturationAssessment:
    """Combine a throughput knee with temporal over-saturation evidence.

    :param throughput: Result of fitting the throughput curve
    :param saturation: Boundary inferred from the over-saturation detector
    :return: Combined assessment and optional refinement bounds
    """
    if saturation.status == "inconsistent":
        return SaturationAssessment(
            status="disagreement",
            reason="Over-saturation states are not monotonic across concurrency",
        )

    if throughput.status == "ok":
        return _assess_detected_knee(throughput, saturation)

    if saturation.status == "detected":
        return SaturationAssessment(
            status="oversaturation_boundary",
            reason="No throughput knee was found; refine the detector boundary instead",
            refinement_lower=saturation.previous_safe_concurrency,
            refinement_upper=saturation.first_oversaturated_concurrency,
        )

    return SaturationAssessment(
        status="no_saturation",
        reason="Neither detector found saturation in the measured range",
    )

def benchmark_concurrency(benchmark: GenerativeBenchmark) -> int:
    """Read the concurrent stream count from a completed benchmark.

    :param benchmark: Completed GuideLLM benchmark
    :return: Positive integer concurrent stream count
    :raises ValueError: If the benchmark did not use a concurrent strategy
    """
    strategy = benchmark.config.strategy
    if not isinstance(strategy, ConcurrentStrategy):
        raise ValueError(
            "Knee analysis requires concurrent benchmark strategies; "
            f"received {strategy.type_!r}"
        )
    return strategy.streams

def extract_saturation_boundary(
    benchmarks: Sequence[GenerativeBenchmark],
) -> SaturationBoundary:
    """Infer the safe-to-over-saturated boundary from native benchmark state.

    :param benchmarks: Completed concurrent GuideLLM benchmarks
    :return: Ordered detector points and the inferred saturation boundary
    :raises ValueError: If concurrency or detector metadata is malformed
    """
    points: list[SaturationPoint] = []
    seen_concurrencies: set[float] = set()
    for benchmark in benchmarks:
        concurrency = float(benchmark_concurrency(benchmark))
        if concurrency in seen_concurrencies:
            raise ValueError(f"Duplicate benchmark concurrency: {concurrency:g}")
        seen_concurrencies.add(concurrency)

        action = _constraint_action(benchmark)
        if action is None:
            points.append(
                SaturationPoint(concurrency=concurrency, is_over_saturated=None)
            )
            continue

        metadata: dict[str, object] = action.metadata
        is_over_saturated = metadata.get("is_over_saturated")
        if not isinstance(is_over_saturated, bool):
            raise ValueError(
                "Over-saturation metadata must contain boolean 'is_over_saturated'"
            )
        points.append(
            SaturationPoint(
                concurrency=concurrency,
                is_over_saturated=is_over_saturated,
                concurrent_slope=_optional_finite_number(metadata, "concurrent_slope"),
                concurrent_slope_moe=_optional_finite_number(
                    metadata, "concurrent_slope_moe"
                ),
                concurrent_n=_optional_finite_number(metadata, "concurrent_n"),
                ttft_slope=_optional_finite_number(metadata, "ttft_slope"),
                ttft_slope_moe=_optional_finite_number(metadata, "ttft_slope_moe"),
                ttft_n=_optional_finite_number(metadata, "ttft_n"),
                ttft_violations=_optional_finite_number(metadata, "ttft_violations"),
            )
        )

    sorted_points = tuple(sorted(points, key=lambda point: point.concurrency))
    available_points = [
        point for point in sorted_points if point.is_over_saturated is not None
    ]
    if not available_points:
        return SaturationBoundary(
            status="unavailable",
            reason="The benchmarks contain no over-saturation detector metadata",
            points=sorted_points,
        )

    oversaturated_points = [
        point for point in available_points if point.is_over_saturated is True
    ]
    if not oversaturated_points:
        return SaturationBoundary(
            status="not_detected",
            reason="The detector did not report over-saturation at any concurrency",
            points=sorted_points,
        )

    first_oversaturated = oversaturated_points[0].concurrency
    safe_before = [
        point.concurrency
        for point in available_points
        if point.is_over_saturated is False and point.concurrency < first_oversaturated
    ]
    previous_safe = max(safe_before) if safe_before else None
    safe_after = [
        point.concurrency
        for point in available_points
        if point.is_over_saturated is False and point.concurrency > first_oversaturated
    ]
    if safe_after:
        return SaturationBoundary(
            status="inconsistent",
            reason=(
                "The detector reported a safe concurrency above its first "
                "over-saturated concurrency"
            ),
            points=sorted_points,
            previous_safe_concurrency=previous_safe,
            first_oversaturated_concurrency=first_oversaturated,
        )

    return SaturationBoundary(
        status="detected",
        reason="The detector reported an over-saturated concurrency",
        points=sorted_points,
        previous_safe_concurrency=previous_safe,
        first_oversaturated_concurrency=first_oversaturated,
    )

def find_throughput_knee(
    concurrencies: Sequence[int | float],
    output_throughputs: Sequence[int | float | None],
) -> KneeResult:
    """Find where output throughput changes from rising to substantially flatter.

    Duplicate concurrency measurements are averaged. Missing throughput values
    are ignored. A knee is accepted only when segmented regression improves the
    fit, the tail is substantially flatter, and the breakpoint has reached most
    of the observed peak throughput.

    :param concurrencies: Concurrent stream counts for the measurements
    :param output_throughputs: Successful output-token throughput measurements
    :return: Detected knee and fit diagnostics, or a no-knee explanation
    :raises ValueError: If the inputs have different lengths or invalid values
    """
    xs, ys = _prepare_points(concurrencies, output_throughputs)
    if len(xs) < MIN_POINTS:
        return KneeResult(
            status="no_knee",
            reason=(
                f"Need at least {MIN_POINTS} distinct concurrency points; "
                f"found {len(xs)}"
            ),
        )

    x_min, x_max = xs[0], xs[-1]
    y_min, y_max = min(ys), max(ys)
    if x_max == x_min or y_max == y_min:
        return KneeResult(
            status="no_knee", reason="The throughput curve has no usable range"
        )

    normalized_xs = [(x - x_min) / (x_max - x_min) for x in xs]
    normalized_ys = [(y - y_min) / (y_max - y_min) for y in ys]

    _, _, single_line_error = _line_fit(normalized_xs, normalized_ys)
    if single_line_error <= FIT_EPSILON:
        return KneeResult(
            status="no_knee",
            reason="The throughput curve is approximately linear",
        )

    candidates = _fit_candidates(
        normalized_xs,
        normalized_ys,
        ys,
        y_max,
        single_line_error,
    )

    if not candidates:
        return KneeResult(
            status="no_knee",
            reason="No breakpoint satisfied the saturation thresholds",
        )

    (
        _,
        best_index,
        pre_intercept,
        pre_slope,
        tail_intercept,
        tail_slope,
        slope_ratio,
        fit_improvement,
        throughput_fraction,
    ) = min(candidates)

    slope_delta = pre_slope - tail_slope
    knee_normalized = (
        (tail_intercept - pre_intercept) / slope_delta
        if abs(slope_delta) > FIT_EPSILON
        else normalized_xs[best_index]
    )
    if not (
        normalized_xs[best_index - 1]
        <= knee_normalized
        <= normalized_xs[best_index + 1]
    ):
        knee_normalized = normalized_xs[best_index]

    knee = x_min + knee_normalized * (x_max - x_min)
    epsilon = (x_max - x_min) * 1e-9
    saturation_concurrency = next((x for x in xs if x >= knee - epsilon), xs[-1])

    return KneeResult(
        status="ok",
        reason="A rising segment followed by a substantially flatter tail was detected",
        knee=knee,
        saturation_concurrency=saturation_concurrency,
        breakpoint_concurrency=xs[best_index],
        pre_slope=pre_slope,
        tail_slope=tail_slope,
        slope_ratio=slope_ratio,
        fit_improvement=fit_improvement,
        throughput_fraction=throughput_fraction,
    )

def generate_adaptive_concurrency_plan(
    analysis: KneeAnalysis,
    measured_concurrencies: Sequence[int],
    *,
    points_each_side: int = 5,
    max_step: int = 5,
) -> AdaptiveConcurrencyPlan:
    """Select unmeasured integer concurrency points around saturation.

    A clean ``max_step`` grid is preferred. The step is reduced near the bottom
    of the positive integer range so lower points remain valid.

    :param analysis: Knee and over-saturation assessment from the initial run
    :param measured_concurrencies: Concurrent stream counts already executed
    :param points_each_side: Maximum additional grid points on either side
    :param max_step: Largest concurrency grid step to consider
    :return: Ready plan with new concurrencies, or an explanation for skipping
    :raises ValueError: If configuration or measured concurrencies are invalid
    """
    _validate_positive_integer("points_each_side", points_each_side)
    _validate_positive_integer("max_step", max_step)
    measured = _validated_measured_concurrencies(measured_concurrencies)

    assessment = analysis.assessment
    if assessment.status in {"disagreement", "no_saturation"}:
        return AdaptiveConcurrencyPlan(
            status="skipped",
            reason=f"Cannot safely select adaptive concurrencies: {assessment.reason}",
        )

    center = _adaptive_center(assessment)
    if center is None:
        return AdaptiveConcurrencyPlan(
            status="skipped",
            reason="The saturation assessment did not provide a refinement center",
        )
    if not math.isfinite(center) or center <= 0:
        raise ValueError("Adaptive concurrency center must be positive and finite")

    anchor, step, candidates = _adaptive_grid(center, points_each_side, max_step)
    excluded = tuple(
        concurrency for concurrency in candidates if concurrency in measured
    )
    concurrencies = tuple(
        concurrency for concurrency in candidates if concurrency not in measured
    )
    if not concurrencies:
        return AdaptiveConcurrencyPlan(
            status="skipped",
            reason="All adaptive candidate concurrencies were already measured",
            selection_center=center,
            anchor=anchor,
            step=step,
            candidate_concurrencies=candidates,
            excluded_concurrencies=excluded,
        )

    return AdaptiveConcurrencyPlan(
        status="ready",
        reason="Selected unmeasured concurrencies around the saturation region",
        concurrencies=concurrencies,
        selection_center=center,
        anchor=anchor,
        step=step,
        candidate_concurrencies=candidates,
        excluded_concurrencies=excluded,
    )