#!/usr/bin/env python3

"""Fleet-scale report over contention-lane row JSONL files.

Reads parallel_contention rows (canonical or .partial.jsonl — same encoding)

and renders a markdown report with three sections, each computed by a pure

function the CLI merely renders:

1. Throughput saturation: commits/sec per worker tier and the saturation

   knee. KNEE RULE: per-worker throughput at tier w is throughput(w)/w; the

   baseline is per-worker throughput at the w2 tier (falling back to the

   smallest observed tier when w2 is absent); the knee is the LARGEST tier

   whose per-worker throughput is >= 70% of that baseline. Above the knee,

   adding workers buys less than 70 cents on the baseline dollar.

2. Fairness per scenario: spread between the p99 and p50 of per-worker

   elapsed_ms (worker_p99_spread = p99/p50), plus starvation — workers whose

   measured completed-op count is exactly 0 (null/absent counts are

   unmeasured, never starved; ADR-0002).

3. Server-error detection per tier: rows whose stderr-ish fields match

   5xx/throttle patterns ("HTTP 5xx", "throttl", "rate limit", "503", "502",

   "500"), plus nonzero-returncode counts (skips excluded) as the fallback

   signal when no stderr is present.

Usage:

    python3 scripts/fleet_report.py results/parallel-contention/latest.jsonl

    python3 scripts/fleet_report.py a.jsonl b.partial.jsonl --out report.md

"""

from __future__ import annotations

import argparse

import json

import math

import re

from pathlib import Path

from typing import Any, Optional

SKIP_RETURNCODE = 77

KNEE_THRESHOLD = 0.7

KNEE_BASELINE_TIER = 2

WORKER_OPERATIONS = ("parallel.worker", "parallel.reader")

# Completed-op count per worker row: the first present, non-null counter wins.

COMPLETED_OP_KEYS = ("commits_succeeded", "lands_succeeded", "polls_total", "pushes_succeeded")

SERVER_ERROR_PATTERNS = (

    re.compile(r"HTTP[ /]?5\d\d", re.IGNORECASE),

    re.compile(r"\b50[023]\b"),

    re.compile(r"throttl", re.IGNORECASE),

    re.compile(r"rate limit", re.IGNORECASE),

)

def load_rows(paths: list[Path]) -> list[dict[str, Any]]:

    """Parse JSONL rows from every path, skipping blank/malformed lines."""

    rows: list[dict[str, Any]] = []

    for path in paths:

        for line in Path(path).read_text(encoding="utf-8").splitlines():

            stripped = line.strip()

            if not stripped:

                continue

            try:

                row = json.loads(stripped)

            except json.JSONDecodeError:

                continue

            if isinstance(row, dict):

                rows.append(row)

    return rows

def percentile_nearest(values: list[float], pct: float) -> Optional[float]:

    """Nearest-rank percentile: sorted[ceil(p/100 * n) - 1]. None when empty."""

    if not values:

        return None

    ordered = sorted(values)

    rank = max(1, math.ceil(pct / 100.0 * len(ordered)))

    return float(ordered[rank - 1])

def _number(value: Any) -> Optional[float]:

    if isinstance(value, bool) or not isinstance(value, (int, float)):

        return None

    return float(value)

def total_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:

    return [

        row

        for row in rows

        if row.get("operation") == "parallel.total" and not row.get("skipped")

    ]

def worker_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:

    return [row for row in rows if row.get("operation") in WORKER_OPERATIONS]

def throughput_by_tier(rows: list[dict[str, Any]]) -> dict[tuple[str, str], dict[int, float]]:

    """Mean commit_throughput_per_s per worker tier, grouped by

    (subject, contention_mode). Rows without a measured throughput are

    excluded (null is unmeasured, never zero)."""

    samples: dict[tuple[str, str], dict[int, list[float]]] = {}

    for row in total_rows(rows):

        metrics = row.get("parallel_metrics") or {}

        throughput = _number(metrics.get("commit_throughput_per_s"))

        workers = row.get("workers")

        if throughput is None or not isinstance(workers, int):

            continue

        key = (str(row.get("subject")), str(row.get("contention_mode")))

        samples.setdefault(key, {}).setdefault(workers, []).append(throughput)

    return {

        key: {tier: sum(values) / len(values) for tier, values in tiers.items()}

        for key, tiers in samples.items()

    }

def saturation_knee(

    tier_throughput: dict[int, float],

    baseline_tier: int = KNEE_BASELINE_TIER,

    threshold: float = KNEE_THRESHOLD,

) -> Optional[int]:

    """The saturation knee for one (subject, mode) throughput curve.

    Rule (documented in the module docstring): baseline per-worker throughput

    is throughput(w2)/2 (smallest observed tier if w2 is absent). The knee is

    the largest tier whose throughput/worker >= threshold * baseline. Returns

    None when no baseline is measurable.

    """

    if not tier_throughput:

        return None

    base = baseline_tier if baseline_tier in tier_throughput else min(tier_throughput)

    baseline = tier_throughput[base] / base

    if baseline <= 0:

        return None

    qualifying = [

        tier

        for tier, throughput in tier_throughput.items()

        if (throughput / tier) >= threshold * baseline

    ]

    return max(qualifying) if qualifying else None

def completed_ops(row: dict[str, Any]) -> Optional[int]:

    """Measured completed-op count for a worker row; None means unmeasured."""

    for key in COMPLETED_OP_KEYS:

        value = row.get(key)

        if isinstance(value, bool):

            continue

        if isinstance(value, int):

            return value

    return None

def fairness_by_scenario(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:

    """Per (scenario, subject): worker p99/p50 elapsed spread and starvation.

    worker_p50_ms/worker_p99_ms are nearest-rank percentiles of per-worker

    elapsed_ms; worker_p99_spread = p99/p50 (None when p50 is 0 or missing).

    starvation_count counts workers whose measured completed-op count is 0;

    unmeasured workers are never counted as starved.

    """

    groups: dict[tuple[str, str], list[dict[str, Any]]] = {}

    for row in worker_rows(rows):

        key = (str(row.get("scenario")), str(row.get("subject")))

        groups.setdefault(key, []).append(row)

    report: list[dict[str, Any]] = []

    for (scenario, subject), members in sorted(groups.items()):

        elapsed = [value for value in (_number(row.get("elapsed_ms")) for row in members) if value is not None]

        p50 = percentile_nearest(elapsed, 50.0)

        p99 = percentile_nearest(elapsed, 99.0)

        spread = round(p99 / p50, 3) if p50 not in (None, 0) and p99 is not None else None

        starved = [

            row.get("worker")

            for row in members

            if completed_ops(row) == 0

        ]

        report.append(

            {

                "scenario": scenario,

                "subject": subject,

                "workers_observed": len(members),

                "worker_p50_ms": p50,

                "worker_p99_ms": p99,

                "worker_p99_spread": spread,

                "starvation_count": len(starved),

                "starved_workers": starved,

            }

        )

    return report

def _row_error_text(row: dict[str, Any]) -> str:

    parts: list[str] = []

    stderr = row.get("stderr")

    if isinstance(stderr, str):

        parts.append(stderr)

    examples = row.get("error_examples")

    if isinstance(examples, list):

        parts.extend(str(item) for item in examples)

    return "\n".join(parts)

def server_error_summary(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:

    """Per worker tier: 5xx/throttle pattern hits in stderr-ish fields, and

    nonzero returncodes (skip rows excluded) as the fallback signal."""

    tiers: dict[int, dict[str, int]] = {}

    for row in rows:

        workers = row.get("workers")

        if not isinstance(workers, int):

            continue

        bucket = tiers.setdefault(

            workers, {"rows": 0, "server_error_pattern_hits": 0, "nonzero_returncodes": 0}

        )

        bucket["rows"] += 1

        text = _row_error_text(row)

        if text and any(pattern.search(text) for pattern in SERVER_ERROR_PATTERNS):

            bucket["server_error_pattern_hits"] += 1

        returncode = row.get("returncode")

        if isinstance(returncode, int) and returncode not in (0, SKIP_RETURNCODE):

            bucket["nonzero_returncodes"] += 1

    return [

        {"workers": tier, **counts} for tier, counts in sorted(tiers.items())

    ]

def _fmt(value: Any) -> str:

    if value is None:

        return "unmeasured"

    if isinstance(value, float):

        return f"{value:.3f}"

    return str(value)

def render_markdown(rows: list[dict[str, Any]]) -> str:

    lines: list[str] = ["# Fleet Report", ""]

    lines += [

        "## Throughput vs worker tier",

        "",

        "commits/sec is the mean `parallel_metrics.commit_throughput_per_s` of total",

        "rows per tier. Saturation knee rule: per-worker throughput at tier w is",

        "throughput(w)/w; baseline is the w2 tier (smallest observed tier when w2 is",

        f"absent); the knee is the largest tier with throughput/worker >= {KNEE_THRESHOLD:.0%}",

        "of the baseline.",

        "",

        "| Subject | Mode | Tier (workers) | Commits/s | Commits/s/worker | Knee |",

        "| --- | --- | ---: | ---: | ---: | --- |",

    ]

    curves = throughput_by_tier(rows)

    if not curves:

        lines.append("| (no measured throughput rows) | | | | | |")

    for (subject, mode), tiers in sorted(curves.items()):

        knee = saturation_knee(tiers)

        for tier in sorted(tiers):

            throughput = tiers[tier]

            marker = f"knee (w{knee})" if knee == tier else ""

            lines.append(

                f"| `{subject}` | `{mode}` | {tier} | {throughput:.3f} | {throughput / tier:.3f} | {marker} |"

            )

        if knee is None:

            lines.append(f"| `{subject}` | `{mode}` | — | — | — | knee: unmeasured |")

    lines.append("")

    lines += [

        "## Fairness per scenario",

        "",

        "Spread is p99/p50 of per-worker elapsed_ms (nearest-rank percentiles over the",

        "observed workers). Starvation counts workers whose measured completed-op count",

        "is exactly 0; unmeasured workers are never counted as starved (ADR-0002).",

        "",

        "| Scenario | Subject | Workers observed | p50 ms | p99 ms | p99/p50 spread | Starved |",

        "| --- | --- | ---: | ---: | ---: | ---: | ---: |",

    ]

    fairness = fairness_by_scenario(rows)

    if not fairness:

        lines.append("| (no worker rows) | | | | | | |")

    for entry in fairness:

        lines.append(

            f"| `{entry['scenario']}` | `{entry['subject']}` | {entry['workers_observed']} | "

            f"{_fmt(entry['worker_p50_ms'])} | {_fmt(entry['worker_p99_ms'])} | "

            f"{_fmt(entry['worker_p99_spread'])} | {entry['starvation_count']} |"

        )

    lines.append("")

    lines += [

        "## Server errors / throttling per tier",

        "",

        "Pattern hits scan `stderr` and `error_examples` for HTTP 5xx / throttle /",

        "rate-limit signatures. Nonzero returncodes (skips excluded) are the fallback",

        "signal when no stderr text is present.",

        "",

        "| Tier (workers) | Rows | 5xx/throttle pattern hits | Nonzero returncodes |",

        "| ---: | ---: | ---: | ---: |",

    ]

    errors = server_error_summary(rows)

    if not errors:

        lines.append("| (no tiered rows) | | | |")

    for entry in errors:

        lines.append(

            f"| {entry['workers']} | {entry['rows']} | {entry['server_error_pattern_hits']} | "

            f"{entry['nonzero_returncodes']} |"

        )

    lines.append("")

    return "\n".join(lines)

def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:

    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)

    parser.add_argument("paths", nargs="+", type=Path, help="Row JSONL paths (canonical or .partial.jsonl)")

    parser.add_argument("--out", type=Path, help="Write the markdown report here (default: stdout)")

    return parser.parse_args(argv)

def main(argv: Optional[list[str]] = None) -> int:

    args = parse_args(argv)

    rows = load_rows(args.paths)

    report = render_markdown(rows)

    if args.out:

        args.out.parent.mkdir(parents=True, exist_ok=True)

        args.out.write_text(report)

        print(f"[report] {args.out}")

    else:

        print(report)

    return 0

if __name__ == "__main__":

    raise SystemExit(main())