litellm/tests/e2e/perf/setup/observe.py

"""Turns what a run left behind (the client records, what each mock received, the guardrail's
calls, the workers' memory) into the case's metrics, and the case's limits into a verdict."""

from __future__ import annotations

import hashlib
import json
import statistics
from collections import Counter, defaultdict
from collections.abc import Iterable, Mapping
from dataclasses import asdict, dataclass
from itertools import chain, groupby
from types import MappingProxyType
from typing import Final, Literal

from e2e_http import NoBody, get_external, post_external, unwrap
from memory_readings import RssCapture, RssReading, WorkerKey
from pydantic import BaseModel
from typing_extensions import ReadOnly, TypedDict

from setup.datastore import (
    DbCounters,
    RedisCommand,
    RedisCounters,
    StatementCalls,
    active_sessions,
    calls_in,
    slowest_statement,
    statement_deltas,
)
from setup.mocks.reply_common import Provider
from setup.performance_test_case import Limit, Metric, MockLatency, PerformanceTestCase, Ramp
from setup.ramp import StepReading, render_steps, sustained_rps
from setup.series import Series
from setup.telemetry import GatewayTelemetry
from setup.wire import WireRecord


class MockRequestSeen(BaseModel):
    provider: Provider
    marker: str | None
    route: str
    stream: bool
    output_tokens: int
    text_sha1: str
    delay_s: float


class MockRequestsSeen(BaseModel):
    requests: list[MockRequestSeen]


class GuardrailCallSeen(BaseModel):
    marker: str | None
    input_type: str
    action: str


class GuardrailCallsSeen(BaseModel):
    calls: list[GuardrailCallSeen]


class Cleared(BaseModel):
    cleared: int


class LatencyBody(BaseModel):
    seconds: float
    jitter_seconds: float


def read_mock_requests(url: str) -> list[MockRequestSeen]:
    return unwrap(get_external(f"{url}/mock/requests", response_type=MockRequestsSeen, timeout=60.0)).requests


def read_guardrail_calls(url: str) -> list[GuardrailCallSeen]:
    return unwrap(get_external(f"{url}/mock/calls", response_type=GuardrailCallsSeen, timeout=60.0)).calls


def reset_mock(url: str) -> int:
    return unwrap(post_external(f"{url}/mock/reset", json=NoBody(), response_type=Cleared)).cleared


def set_mock_latency(url: str, latency: MockLatency) -> LatencyBody:
    body: Final = LatencyBody(seconds=latency.seconds, jitter_seconds=latency.jitter_seconds)
    return unwrap(post_external(f"{url}/mock/latency", json=body, response_type=LatencyBody))


def percentile(values: Iterable[float], p: int) -> float | None:
    ordered: Final = sorted(values)
    if not ordered:
        return None
    index: Final = min(len(ordered) - 1, max(0, round(p / 100 * (len(ordered) - 1))))
    return ordered[index]


def worst_worker_growth_mb(before: RssCapture, after: RssCapture) -> float | None:
    """Worst RSS growth across workers. A worker missing from `before` (a pod replaced during the
    run) is measured against the median of the old workers."""
    old: Final[dict[WorkerKey, RssReading]] = {reading.worker: reading for reading in before.readings}
    if not old or not after.readings:
        return None
    median_before: Final = statistics.median(reading.ram_usage_mb for reading in old.values())
    growth: list[float] = []
    for reading in after.readings:
        baseline = old[reading.worker].ram_usage_mb if reading.worker in old else median_before
        growth.append(reading.ram_usage_mb - baseline)
    return max(growth)


@dataclass(frozen=True, slots=True)
class Observation:
    records: tuple[WireRecord, ...]
    mock_requests: tuple[MockRequestSeen, ...]
    guardrail_calls: tuple[GuardrailCallSeen, ...]
    rss_before: RssCapture
    rss_after: RssCapture
    wall_seconds: float
    spend_logged_share: float | None = None
    db_before: DbCounters | None = None
    db_after: DbCounters | None = None
    redis_before: RedisCounters | None = None
    redis_after: RedisCounters | None = None
    redis_commands: tuple[RedisCommand, ...] | None = None
    gateway_cpu_seconds: float | None = None
    telemetry_before: GatewayTelemetry | None = None
    telemetry_after: GatewayTelemetry | None = None
    peak_connections: int | None = None
    ramp: Ramp | None = None
    cpu_series: Series[float] | None = None
    connection_series: Series[int] | None = None
    telemetry_series: Series[GatewayTelemetry] | None = None

    @property
    def telemetry(self) -> GatewayTelemetry | None:
        """What the gateway's `/metrics` counted during the run."""
        if self.telemetry_before is None or self.telemetry_after is None:
            return None
        return self.telemetry_after.minus(self.telemetry_before)

    def per_request(self, before: int | None, after: int | None) -> float | None:
        if before is None or after is None or not self.records:
            return None
        return (after - before) / len(self.records)

    def count_per_request(self, count: int | None) -> float | None:
        return None if count is None or not self.records else count / len(self.records)

    @property
    def statements(self) -> tuple[StatementCalls, ...] | None:
        return statement_deltas(self.db_before, self.db_after)

    @property
    def busiest_awaited_statement(self) -> StatementCalls | None:
        """The awaited statement with the most calls: a lookup added to every request shows up here at
        1.0 per request while cache refreshes stay fractional, which the awaited total cannot tell apart."""
        awaited: Final = tuple(item for item in self.statements or () if item.bucket == "awaited")
        return max(awaited, key=_item_count, default=None)

    def busiest_background_statement(self, exempt_tables: frozenset[str]) -> StatementCalls | None:
        """The background statement with the most calls, skipping the case's per-flush exemptions: a rollup
        written once per request reads 1.0 here while batched writers stay below it."""
        background: Final = tuple(
            item for item in self.statements or () if item.bucket == "background" and item.table not in exempt_tables
        )
        return max(background, key=_item_count, default=None)

    @property
    def retried_markers(self) -> frozenset[str]:
        counts: Final = Counter(seen.marker for seen in self.mock_requests if seen.marker is not None)
        return frozenset(marker for marker, count in counts.items() if count > 1)

    @property
    def served(self) -> tuple[WireRecord, ...]:
        return tuple(record for record in self.records if record.ok and not record.blocked)

    @property
    def _delay_by_marker(self) -> dict[str, float]:
        return {seen.marker: seen.delay_s for seen in self.mock_requests if seen.marker is not None}

    def _clean(self, *, stream: bool) -> list[tuple[WireRecord, float]]:
        """Served requests of one kind that were answered on the first attempt, with the mock's delay."""
        delays: Final = self._delay_by_marker
        retried: Final = self.retried_markers
        pairs: list[tuple[WireRecord, float]] = []
        for record in self.served:
            delay = record.mock_delay_s if record.mock_delay_s is not None else delays.get(record.marker)
            if record.stream == stream and delay is not None and record.marker not in retried:
                pairs.append((record, delay))
        return pairs

    def unexpected_status_count(self) -> int:
        """Requests the plan expected to succeed that came back with any other status (0 is a client timeout)."""
        return sum(1 for record in self.records if not record.blocked and record.status != 200)

    def latency_pairs(self, metric: LatencyMetric) -> list[tuple[WireRecord, float]]:
        """Each first-attempt request with its value of one latency metric. `gateway_added_seconds` is the
        seconds the gateway added (total time minus the mock's own delay for a non-streamed request, time to the
        first token minus the mock's delay for a stream), `time_to_first_token_seconds` is the client's wait for
        the first token with the mock delay included, and the two `litellm_*` metrics are the gateway's own
        `x-litellm-overhead-duration-ms` and `x-litellm-response-duration-ms` headers in seconds."""
        if metric == "gateway_added_seconds":
            non_stream: Final = [(record, record.total_s - delay) for record, delay in self._clean(stream=False)]
            stream: Final = [
                (record, record.first_token_s - delay)
                for record, delay in self._clean(stream=True)
                if record.first_token_s is not None
            ]
            return non_stream + stream
        if metric == "time_to_first_token_seconds":
            return [(r, r.first_token_s) for r, _ in self._clean(stream=True) if r.first_token_s is not None]
        if metric == "litellm_overhead_seconds":
            return [
                (r, r.litellm_overhead_ms / 1000) for r in self._first_attempts() if r.litellm_overhead_ms is not None
            ]
        return [(r, r.litellm_response_ms / 1000) for r in self._first_attempts() if r.litellm_response_ms is not None]

    def _first_attempts(self) -> list[WireRecord]:
        return [record for record, _ in self._clean(stream=False)] + [record for record, _ in self._clean(stream=True)]

    def gateway_added_seconds(self) -> list[float]:
        return [value for _, value in self.latency_pairs("gateway_added_seconds")]

    def litellm_overhead_seconds(self) -> list[float]:
        return [value for _, value in self.latency_pairs("litellm_overhead_seconds")]

    def litellm_response_seconds(self) -> list[float]:
        return [value for _, value in self.latency_pairs("litellm_response_seconds")]

    def time_to_first_token_seconds(self) -> list[float]:
        return [value for _, value in self.latency_pairs("time_to_first_token_seconds")]

    def bodies_off_reply(self) -> int:
        """Served requests whose text is not the reply the mock generated for them (plus the post-call rewrite
        for requests in the intervene slice), a translation or streaming fault somewhere in the gateway."""
        seen_by_marker: Final = {seen.marker: seen for seen in self.mock_requests if seen.marker is not None}
        return sum(
            1
            for record in self.served
            if (seen := seen_by_marker.get(record.marker)) is not None
            and _text_sha1(record.text, intervened=record.marker.endswith("-i")) != seen.text_sha1
        )

    def guardrail_missed(self) -> int:
        """Served requests that did not get exactly one pre-call and one post-call guardrail call."""
        pre: Final = Counter(call.marker for call in self.guardrail_calls if call.input_type == "request")
        post: Final = Counter(call.marker for call in self.guardrail_calls if call.input_type == "response")
        return sum(1 for record in self.served if pre[record.marker] != 1 or post[record.marker] != 1)

    def steps(self) -> tuple[StepReading, ...]:
        """One reading per step of a ramped run: the gateway's added latency over the step's requests, the rate
        at which requests completed inside the step's window, the step's unexpected statuses, and what the pod
        was doing in that window (CPU per completed request, Redis latency, Postgres connections)."""
        ramp: Final = self.ramp
        if ramp is None or not self.records:
            return ()
        started: Final = min(record.started_at for record in self.records)
        added: Final = _by_step(ramp, self.latency_pairs("gateway_added_seconds"))
        first_token: Final = _by_step(ramp, self.latency_pairs("time_to_first_token_seconds"))
        completed: Final = Counter(
            int((record.started_at + record.total_s - started) // ramp.step_seconds) for record in self.records
        )
        unexpected: Final = Counter(
            ramp.step_of(record.seq) for record in self.records if not record.blocked and record.status != 200
        )
        return tuple(
            StepReading(
                rps=rate,
                planned=count,
                completed_rps=completed[index] / ramp.step_seconds,
                gateway_added_p99=percentile(added.get(index, ()), 99),
                unexpected=unexpected[index],
                held_p99_seconds=ramp.held_p99_seconds,
                gateway_added_p50=percentile(added.get(index, ()), 50),
                time_to_first_token_p99=percentile(first_token.get(index, ()), 99),
                **self._window_readings(started + index * ramp.step_seconds, ramp.step_seconds, completed[index]),
            )
            for index, (rate, count) in enumerate(zip(ramp.rates, ramp.counts, strict=True))
        )

    def _window_readings(self, start: float, seconds: float, completed: int) -> WindowReadings:
        end: Final = start + seconds
        cpu: Final = None if self.cpu_series is None else sum(self.cpu_series.between(start, end))
        conns: Final = None if self.connection_series is None else self.connection_series.between(start, end)
        before: Final = None if self.telemetry_series is None else self.telemetry_series.last_before(start)
        after: Final = None if self.telemetry_series is None else self.telemetry_series.last_before(end)
        return WindowReadings(
            cpu_seconds_per_request=None if cpu is None or completed == 0 else cpu / completed,
            redis_latency_mean_seconds=None
            if before is None or after is None
            else after.minus(before).redis_latency.mean,
            postgres_peak_connections=max(conns, default=None) if conns else None,
        )


class WindowReadings(TypedDict):
    cpu_seconds_per_request: ReadOnly[float | None]
    redis_latency_mean_seconds: ReadOnly[float | None]
    postgres_peak_connections: ReadOnly[int | None]


def _by_step(ramp: Ramp, pairs: Iterable[tuple[WireRecord, float]]) -> Mapping[int, tuple[float, ...]]:
    ordered: Final = sorted(pairs, key=lambda pair: pair[0].seq)
    return MappingProxyType(
        {
            step: tuple(value for _, value in group)
            for step, group in groupby(ordered, key=lambda pair: ramp.step_of(pair[0].seq))
        }
    )


def _text_sha1(text: str, *, intervened: bool) -> str:
    original: Final = text.removesuffix(" [post]") if intervened else text
    return hashlib.sha1(original.encode()).hexdigest()


LatencyMetric = Literal[
    "gateway_added_seconds", "time_to_first_token_seconds", "litellm_overhead_seconds", "litellm_response_seconds"
]
TOP_CONTRIBUTORS: Final = 5


@dataclass(frozen=True, slots=True)
class Contributor:
    """One slice of a metric: a statement, a Redis command, an endpoint, a worker or a status, with its own
    value in the metric's unit and, where the metric is a total, its share of that total."""

    label: str
    value: float
    share: float | None = None


def _top(items: Iterable[Contributor]) -> tuple[Contributor, ...]:
    return tuple(sorted(items, key=lambda item: item.value, reverse=True)[:TOP_CONTRIBUTORS])


def _endpoint_label(record: WireRecord) -> str:
    return f"{record.endpoint}{' stream' if record.stream else ''}"


def _latency_breakdown(metric: LatencyMetric, rank: int, observation: Observation) -> tuple[Contributor, ...]:
    pairs: Final = observation.latency_pairs(metric)
    by_endpoint: dict[str, list[float]] = defaultdict(list)
    for record, value in pairs:
        by_endpoint[_endpoint_label(record)].append(value)
    return _top(
        Contributor(label, percentile(values, rank) or 0.0, len(values) / len(pairs))
        for label, values in by_endpoint.items()
    )


def _statements_for(
    metric: Metric, statements: tuple[StatementCalls, ...], exempt_tables: frozenset[str]
) -> tuple[StatementCalls, ...]:
    if metric in ("postgres_calls_per_request", "postgres_max_statement_per_request"):
        return tuple(item for item in statements if item.bucket == "awaited")
    if metric == "postgres_background_calls_per_request":
        return tuple(item for item in statements if item.bucket == "background")
    if metric == "postgres_max_background_statement_per_request":
        return tuple(item for item in statements if item.bucket == "background" and item.table not in exempt_tables)
    return tuple(item for item in statements if item.calls > 0)


def _statement_breakdown(
    metric: Metric, observation: Observation, exempt_tables: frozenset[str]
) -> tuple[Contributor, ...]:
    chosen: Final = _statements_for(metric, observation.statements or (), exempt_tables)
    requests: Final = max(1, len(observation.records))
    total_calls: Final = sum(item.calls for item in chosen)
    total_ms: Final = sum(item.total_exec_ms for item in chosen)
    if metric == "postgres_slowest_statement_seconds":
        return _top(
            Contributor(
                item.summary, item.mean_exec_seconds or 0.0, item.total_exec_ms / total_ms if total_ms else None
            )
            for item in chosen
        )
    if metric == "postgres_active_sessions":
        return _top(
            Contributor(
                item.summary, item.total_exec_ms / 1000 / observation.wall_seconds, item.total_exec_ms / total_ms
            )
            for item in chosen
            if total_ms
        )
    return _top(
        Contributor(item.summary, item.calls / requests, item.calls / total_calls) for item in chosen if total_calls
    )


def _redis_breakdown(metric: Metric, observation: Observation) -> tuple[Contributor, ...]:
    bucket: Final = "awaited" if metric == "redis_calls_per_request" else "background"
    chosen: Final = tuple(item for item in observation.redis_commands or () if item.bucket == bucket)
    requests: Final = max(1, len(observation.records))
    total: Final = sum(item.count for item in chosen)
    return _top(
        Contributor(f"{item.command} {item.key}".strip(), item.count / requests, item.count / total)
        for item in chosen
        if total
    )


def _memory_breakdown(observation: Observation) -> tuple[Contributor, ...]:
    before: Final = {reading.worker: reading.ram_usage_mb for reading in observation.rss_before.readings}
    return _top(
        Contributor(f"{reading.replica} pid {reading.worker_pid}", reading.ram_usage_mb - before[reading.worker])
        for reading in observation.rss_after.readings
        if reading.worker in before
    )


def _status_breakdown(observation: Observation) -> tuple[Contributor, ...]:
    unexpected: Final = [r for r in observation.records if not r.blocked and r.status != 200]
    counts: Final = Counter(f"{_endpoint_label(record)} status {record.status}" for record in unexpected)
    return _top(Contributor(label, float(count), count / len(unexpected)) for label, count in counts.items())


def breakdown(
    metric: Metric, percentile_rank: int | None, observation: Observation, exempt_tables: frozenset[str] = frozenset()
) -> tuple[Contributor, ...]:
    """The top contributors to one metric, so a spike on the chart names its statement, command, endpoint or
    worker. Gauges with no slices (`cpu_seconds_per_request`, `prometheus_redis_latency_mean_seconds`,
    `postgres_peak_connections`) have none."""
    if metric in (
        "gateway_added_seconds",
        "time_to_first_token_seconds",
        "litellm_overhead_seconds",
        "litellm_response_seconds",
    ):
        return _latency_breakdown(metric, 50 if percentile_rank is None else percentile_rank, observation)
    if metric.startswith("postgres_") and metric != "postgres_peak_connections":
        return _statement_breakdown(metric, observation, exempt_tables)
    if metric in {"redis_calls_per_request", "redis_background_calls_per_request"}:
        return _redis_breakdown(metric, observation)
    if metric == "memory_growth_megabytes":
        return _memory_breakdown(observation)
    if metric == "unexpected_status_count":
        return _status_breakdown(observation)
    return ()


def observe(
    metric: Metric, percentile_rank: int | None, observation: Observation, exempt_tables: frozenset[str] = frozenset()
) -> float | None:
    """The observed value of one metric, at a percentile when the metric is per request."""
    per_request: dict[str, list[float]] = {
        "gateway_added_seconds": observation.gateway_added_seconds(),
        "time_to_first_token_seconds": observation.time_to_first_token_seconds(),
        "litellm_overhead_seconds": observation.litellm_overhead_seconds(),
        "litellm_response_seconds": observation.litellm_response_seconds(),
    }
    if metric in per_request:
        return percentile(per_request[metric], 50 if percentile_rank is None else percentile_rank)
    if metric == "unexpected_status_count":
        return float(observation.unexpected_status_count())
    if metric == "prometheus_redis_latency_mean_seconds":
        counted: Final = observation.telemetry
        return None if counted is None else counted.redis_latency.mean
    if metric == "postgres_active_sessions":
        return active_sessions(observation.statements, observation.wall_seconds)
    if metric == "postgres_slowest_statement_seconds":
        slowest: Final = slowest_statement(observation.statements)
        return None if slowest is None else slowest.mean_exec_seconds
    if metric == "postgres_peak_connections":
        return None if observation.peak_connections is None else float(observation.peak_connections)
    if metric == "postgres_calls_per_request":
        return observation.count_per_request(calls_in("awaited", observation.statements))
    if metric == "postgres_background_calls_per_request":
        return observation.count_per_request(calls_in("background", observation.statements))
    if metric == "postgres_max_statement_per_request":
        if observation.statements is None:
            return None
        busiest: Final = observation.busiest_awaited_statement
        return observation.count_per_request(0 if busiest is None else busiest.calls)
    if metric == "postgres_max_background_statement_per_request":
        if observation.statements is None:
            return None
        busiest_background: Final = observation.busiest_background_statement(exempt_tables)
        return observation.count_per_request(0 if busiest_background is None else busiest_background.calls)
    if metric == "redis_calls_per_request":
        return observation.count_per_request(calls_in("awaited", observation.redis_commands))
    if metric == "redis_background_calls_per_request":
        return observation.count_per_request(calls_in("background", observation.redis_commands))
    if metric == "cpu_seconds_per_request":
        cpu: Final = observation.gateway_cpu_seconds
        return None if cpu is None or not observation.records else cpu / len(observation.records)
    if metric == "sustained_rps":
        return None if observation.ramp is None else sustained_rps(observation.steps())
    return worst_worker_growth_mb(observation.rss_before, observation.rss_after)


@dataclass(frozen=True, slots=True)
class Row:
    metric: Metric
    percentile: int | None
    observed: float | None
    limit: float | None
    floor: float | None = None
    detail: str = ""
    breakdown: tuple[Contributor, ...] = ()

    @property
    def label(self) -> str:
        return f"{self.metric} percentile {self.percentile}" if self.percentile is not None else self.metric

    @property
    def property_name(self) -> str:
        return f"perf_{self.metric}" if self.percentile is None else f"perf_{self.metric}_p{self.percentile}"

    @property
    def over(self) -> bool:
        """Past the limit: above a max, or below a floor."""
        if self.observed is None:
            return False
        return (self.limit is not None and self.observed > self.limit) or (
            self.floor is not None and self.observed < self.floor
        )

    @property
    def missing(self) -> bool:
        return (self.limit is not None or self.floor is not None) and self.observed is None

    @property
    def bound(self) -> str:
        if self.limit is not None:
            return f"{self.limit:.4f}"
        return "" if self.floor is None else f">= {self.floor:.4f}"


REPORTED: Final[tuple[tuple[Metric, int | None], ...]] = (
    ("unexpected_status_count", None),
    ("gateway_added_seconds", 50),
    ("gateway_added_seconds", 99),
    ("litellm_overhead_seconds", 50),
    ("litellm_overhead_seconds", 99),
    ("litellm_response_seconds", 50),
    ("litellm_response_seconds", 99),
    ("prometheus_redis_latency_mean_seconds", None),
    ("postgres_active_sessions", None),
    ("postgres_slowest_statement_seconds", None),
    ("postgres_peak_connections", None),
    ("time_to_first_token_seconds", 50),
    ("time_to_first_token_seconds", 99),
    ("memory_growth_megabytes", None),
    ("postgres_calls_per_request", None),
    ("postgres_background_calls_per_request", None),
    ("postgres_max_statement_per_request", None),
    ("postgres_max_background_statement_per_request", None),
    ("redis_calls_per_request", None),
    ("redis_background_calls_per_request", None),
    ("cpu_seconds_per_request", None),
)


def evaluate(case: PerformanceTestCase, observation: Observation) -> tuple[list[Row], list[str]]:
    """Every reported metric with its observed value and the case's limit if it has one, plus one line per
    limit that was exceeded or that the run could not observe."""
    limits: Final[dict[tuple[Metric, int | None], Limit]] = {
        (limit.metric, limit.percentile): limit for limit in case.limits
    }
    keys: list[tuple[Metric, int | None]] = list(REPORTED)
    if observation.ramp is not None:
        keys.append(("sustained_rps", None))
    keys.extend(key for key in limits if key not in keys)
    rows: list[Row] = []
    for metric, rank in keys:
        limit = limits.get((metric, rank))
        rows.append(
            Row(
                metric=metric,
                percentile=rank,
                observed=observe(metric, rank, observation, case.background_cap_exempt_tables),
                limit=limit.max if limit else None,
                floor=limit.min if limit else None,
                detail=_detail(metric, observation, case.background_cap_exempt_tables),
                breakdown=breakdown(metric, rank, observation, case.background_cap_exempt_tables),
            )
        )
    exceeded: Final = [_breach(row) for row in rows if row.over or row.missing]
    return rows, exceeded


def _breach(row: Row) -> str:
    if row.observed is None:
        return f"{row.label}: limit set but nothing observed"
    if row.floor is not None:
        return f"{row.label}: observed {row.observed:.4f} < floor {row.floor:.4f}"
    return f"{row.label}: observed {row.observed:.4f} > limit {row.bound}{f' ({row.detail})' if row.detail else ''}"


def _detail(metric: Metric, observation: Observation, exempt_tables: frozenset[str]) -> str:
    if metric == "postgres_max_statement_per_request":
        busiest: Final = observation.busiest_awaited_statement
        return "" if busiest is None else busiest.summary
    if metric == "postgres_max_background_statement_per_request":
        busiest_background: Final = observation.busiest_background_statement(exempt_tables)
        return "" if busiest_background is None else busiest_background.summary
    if metric == "postgres_slowest_statement_seconds":
        slowest: Final = slowest_statement(observation.statements)
        return "" if slowest is None else slowest.summary
    return ""


def _row_properties(row: Row) -> tuple[tuple[str, float | str], ...]:
    observed: Final = ((row.property_name, row.observed),) if row.observed is not None else ()
    limit: Final = ((f"{row.property_name}_limit", row.limit),) if row.limit is not None else ()
    floor: Final = ((f"{row.property_name}_floor", row.floor),) if row.floor is not None else ()
    contributors: Final = (
        ((f"{row.property_name}_breakdown", json.dumps([asdict(item) for item in row.breakdown])),)
        if row.breakdown
        else ()
    )
    return observed + limit + floor + contributors


def junit_properties(rows: Iterable[Row]) -> tuple[tuple[str, float | str], ...]:
    """One numeric property per observed metric and per limit, plus a `_breakdown` JSON list of the top
    contributors for every metric that has slices, all under the `perf_` prefix the dashboard ingests."""
    return tuple(chain.from_iterable(map(_row_properties, rows)))


def render(case: PerformanceTestCase, observation: Observation, rows: list[Row]) -> str:
    by_endpoint: Final = Counter((record.endpoint, record.stream) for record in observation.records)
    failures: Final = [record for record in observation.records if not record.ok][:10]
    lines: list[str] = [
        f"perf case {case.name}: {len(observation.records)} requests, {len(observation.served)} served, "
        f"{sum(1 for r in observation.records if r.blocked and r.ok)} blocked as planned, "
        f"{sum(1 for r in observation.records if not r.ok)} wrong, wall {observation.wall_seconds:.1f}s",
        "  "
        + ", ".join(
            f"{endpoint}{' stream' if stream else ''}={count}"
            for (endpoint, stream), count in sorted(by_endpoint.items())
        ),
        f"  {'metric':32} {'observed':>12} {'limit':>12}",
    ]
    for row in rows:
        observed = "n/a" if row.observed is None else f"{row.observed:.4f}"
        flag = (
            "  UNDER"
            if row.over and row.floor is not None
            else "  OVER"
            if row.over
            else "  MISSING"
            if row.missing
            else ""
        )
        lines.append(f"  {row.label:32} {observed:>12} {row.bound:>12}{flag}")
    lines.extend(render_steps(observation.steps()))
    lines.extend(_diagnostics(observation))
    lines.extend(_datastore_lines(observation))
    for record in failures:
        lines.append(
            f"  wrong: seq {record.seq} {record.caller} {record.endpoint}{' stream' if record.stream else ''} status {record.status}: {record.error}"
        )
    return "\n".join(lines)


def _diagnostics(observation: Observation) -> list[str]:
    """Functional side-channels printed for whoever reads a red run, never limits of a case."""
    count: Final = max(1, len(observation.records))
    share: Final = observation.spend_logged_share
    return [
        "  diagnostics (not limits):",
        f"    bodies off reply {observation.bodies_off_reply()}",
        f"    mock routes {_mock_routes(observation)}",
        f"    guardrail calls per request {len(observation.guardrail_calls) / count:.3f}",
        f"    guardrail missed {observation.guardrail_missed() if observation.guardrail_calls else 'n/a'}",
        f"    retried share {len(observation.retried_markers) / count:.4f}",
        f"    spend logged share {'n/a' if share is None else f'{share:.3f}'}",
        f"    postgres transactions per request {_fmt(observation.per_request(_commits(observation.db_before), _commits(observation.db_after)))}",
        f"    redis commands per request {_fmt(observation.per_request(_commands(observation.redis_before), _commands(observation.redis_after)))}",
        f"    redis connected clients {_clients(observation.redis_before)} -> {_clients(observation.redis_after)}",
    ]


def _mock_routes(observation: Observation) -> str:
    """Which provider route each request landed on, so a direct-vs-bridge case shows its split."""
    counts: Final = Counter(f"{seen.provider}/{seen.route}" for seen in observation.mock_requests)
    return ", ".join(f"{route} {count}" for route, count in sorted(counts.items())) or "n/a"


def _datastore_lines(observation: Observation) -> list[str]:
    """What the per-request Postgres and Redis counts are made of, so a red run names the statement or
    command that grew. `plumbing` (BEGIN, COMMIT, session settings) and `unclassified` count in neither limit."""
    count: Final = max(1, len(observation.records))
    lines: list[str] = []
    for title, items in (
        ("postgres statements", observation.statements),
        ("redis commands", observation.redis_commands),
    ):
        if items is None:
            lines.append(f"  {title}: n/a")
            continue
        for bucket in ("awaited", "background", "plumbing", "unclassified"):
            in_bucket = sorted((item for item in items if item.bucket == bucket), key=_item_count, reverse=True)
            total = sum(_item_count(item) for item in in_bucket)
            if not in_bucket:
                continue
            lines.append(f"  {title} {bucket}: {total / count:.3f} per request, {len(in_bucket)} distinct")
            lines.extend(f"    {_item_count(item) / count:8.3f}  {item.summary}" for item in in_bucket[:8])
    return lines


def _item_count(item: StatementCalls | RedisCommand) -> int:
    return item.calls if isinstance(item, StatementCalls) else item.count


def _commits(counters: DbCounters | None) -> int | None:
    return None if counters is None else counters.commits


def _commands(counters: RedisCounters | None) -> int | None:
    return None if counters is None else counters.commands


def _fmt(value: float | None) -> str:
    return "n/a" if value is None else f"{value:.3f}"


def _clients(counters: RedisCounters | None) -> str:
    return "n/a" if counters is None else str(counters.connected_clients)