diff --git a/scripts/ci/characterize_runner_variance.py b/scripts/ci/characterize_runner_variance.py
new file mode 100644
index 0000000000..8260176591
--- /dev/null
+++ b/scripts/ci/characterize_runner_variance.py
@@ -0,0 +1,290 @@
+#!/usr/bin/env python3
+"""Characterize the self-hosted runner duration variance (#15574 items 1-2).
+
+Collects completed jobs of a workflow from the GitHub Actions API (runner
+name, start/end timestamps, conclusion), attributes each job to a physical
+host via the runner-name prefix, and measures how job duration varies with:
+
+  - the runner identity (per-runner median / p90),
+  - the hour of day (nocturnal vs diurnal buckets),
+  - the number of CPU-heavy jobs overlapping on the same host
+    (co-residency, the mechanism suspected by #15574).
+
+This is a measurement instrument, not a gate: exit code 0 means "measure
+produced", 1 means operational failure (API unreadable). Nothing here turns
+a PR red; the fleet governance decision belongs to #15574 item 3.
+
+Usage:
+  python scripts/ci/characterize_runner_variance.py --workflow scripts-tests.yml --days 14
+  python scripts/ci/characterize_runner_variance.py --workflow scripts-tests.yml --json
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from collections import defaultdict
+from datetime import datetime, timedelta, timezone
+from statistics import median
+from typing import Any, Callable
+
+EXIT_OK = 0
+EXIT_BROKEN = 1
+
+DEFAULT_DAYS = 14
+DEFAULT_WORKFLOW = "scripts-tests.yml"
+
+# Familles de slots dont le cap CPU est "lourd" : la co-residence de CES
+# familles est le mecanisme suspecte par #15574 (sur-souscription CPU de la
+# VM hote). Les waiters (1 vCPU, essentiellement oisifs entre deux gates)
+# sont exclus du decompte de charge : ils pollent, ils ne calculent pas.
+HEAVY_RUNNER_TOKENS = ("-linux-docker-", "-lean-docker-")
+
+
+def _default_runner(argv: list[str]) -> str:
+    """gh api runner par defaut ; injectable pour les tests."""
+    import subprocess
+
+    proc = subprocess.run(argv, capture_output=True, text=True, encoding="utf-8", timeout=60)
+    if proc.returncode != 0:
+        raise RuntimeError(f"command failed rc={proc.returncode}: {argv[:3]}")
+    return proc.stdout
+
+
+def host_of(runner_name: str) -> str:
+    """Attribue un job a son hote physique par prefixe de nom de runner.
+
+    ``myia-po-2024-linux-docker-7`` -> ``myia-po-2024`` (les slots d'une
+    meme famille partagent la VM) ; un nom sans tiret remains tel quel.
+    Le suffixe familial (``-linux-docker-N`` / ``-lean-docker-N`` /
+    ``-linux-waiter-N``) est l'unite de slot, pas d'hote (#15152 : le
+    prefixe derive de la machine, pas d'un hote code en dur).
+    """
+    for token in HEAVY_RUNNER_TOKENS + ("-linux-waiter-",):
+        idx = runner_name.find(token)
+        if idx > 0:
+            return runner_name[:idx]
+    return runner_name
+
+
+def is_heavy(runner_name: str) -> bool:
+    """Un runner de la famille CPU-lourde (docker/lean) ou non."""
+    return any(token in runner_name for token in HEAVY_RUNNER_TOKENS)
+
+
+def _parse_ts(value: str) -> datetime | None:
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def collect_jobs(
+    repo: str,
+    workflow: str,
+    since: datetime,
+    runner: Callable[[list[str]], str] = _default_runner,
+    page_limit: int = 40,
+) -> list[dict[str, Any]]:
+    """Recupere les jobs COMPLETES du workflow sur la fenetre.
+
+    Chaque job retenu porte : name, runner_name, host, heavy, conclusion,
+    started_at, completed_at, duration_s. Les jobs sans runner (hosted,
+    annulees avant placement) sont ecartees : elles ne disent rien de la
+    variance du pool self-hosted.
+    """
+    jobs: list[dict[str, Any]] = []
+    url = f"repos/{repo}/actions/workflows/{workflow}/runs"
+    # Le qualificatif >= du filtre created DOIT etre percent-encode : un `>`
+    # brut dans la query rend le filtre invalide cote GitHub, qui repond alors
+    # zero run en silence (mesure : 2500 runs sans filtre, 0 avec `created>=`
+    # brut).
+    params = f"?status=completed&per_page=100&created=%3E%3D{since.date().isoformat()}"
+    # gh api n'injecte PAS next_page_url dans le corps JSON (il vit dans
+    # l'en-tete Link, invisible sans --include) : la pagination se pilote
+    # par&page=N jusqu'a une page courte (mesure : sans cela, seule la
+    # centaine de runs la plus recente est vue -- 165 jobs au lieu de la
+    # fenetre complete).
+    for page in range(1, page_limit + 1):
+        sep = "&" if "?" in params else "?"
+        payload = json.loads(runner(["gh", "api", f"{url}{params}{sep}page={page}"]))
+        for run in payload.get("workflow_runs", []):
+            # Pre-filtre de fenetre sur created_at : le endpoint LIST ne porte
+            # pas completed_at de facon fiable (null mesure sur ce depot), et
+            # le filtre API created>= borne deja la creation. Les horodatages
+            # fins viennent des jobs, pas de cette entete.
+            created = _parse_ts(run.get("created_at") or "")
+            if created is None or created < since:
+                continue
+            run_id = run["id"]
+            jobs_payload = json.loads(
+                runner(["gh", "api", f"repos/{repo}/actions/runs/{run_id}/jobs?per_page=100"])
+            )
+            for job in jobs_payload.get("jobs", []):
+                started = _parse_ts(job.get("started_at") or "")
+                ended = _parse_ts(job.get("completed_at") or "")
+                runner_name = job.get("runner_name") or ""
+                if job.get("status") != "completed" or not runner_name:
+                    continue
+                if started is None or ended is None or ended <= started:
+                    continue
+                jobs.append(
+                    {
+                        "run_id": run_id,
+                        "name": job.get("name") or "",
+                        "runner_name": runner_name,
+                        "host": host_of(runner_name),
+                        "heavy": is_heavy(runner_name),
+                        "conclusion": job.get("conclusion") or "",
+                        "started_at": started.isoformat(),
+                        "completed_at": ended.isoformat(),
+                        "duration_s": (ended - started).total_seconds(),
+                    }
+                )
+        if len(payload.get("workflow_runs", [])) < 100:
+            break
+    return jobs
+
+
+def overlap_per_job(jobs: list[dict[str, Any]]) -> dict[int, int]:
+    """Pour chaque job lourd (indice -> nombre d'AUTRES jobs lourds du meme
+    hote qui le chevauchaient).
+
+    C'est la grandeur que la decision item 3 de #15574 borne : un plafond
+    de concurrence par hote se lit comme une borne sur cette distribution.
+    """
+    heavy_idx = [i for i, job in enumerate(jobs) if job["heavy"]]
+    counts: dict[int, int] = {}
+    for i in heavy_idx:
+        start = _parse_ts(jobs[i]["started_at"])
+        end = _parse_ts(jobs[i]["completed_at"])
+        if start is None or end is None:
+            counts[i] = 0
+            continue
+        overlaps = 0
+        for j in heavy_idx:
+            if j == i:
+                continue
+            o_start = _parse_ts(jobs[j]["started_at"])
+            o_end = _parse_ts(jobs[j]["completed_at"])
+            if o_start is None or o_end is None:
+                continue
+            if o_start < end and start < o_end:
+                overlaps += 1
+        counts[i] = overlaps
+    return counts
+
+
+def overlap_counts(jobs: list[dict[str, Any]]) -> dict[int, int]:
+    """Histogramme du nombre de chevauchements (voir overlap_per_job)."""
+    hist: dict[int, int] = defaultdict(int)
+    for count in overlap_per_job(jobs).values():
+        hist[count] += 1
+    return dict(sorted(hist.items()))
+
+
+def _percentile(values: list[float], pct: float) -> float:
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    idx = min(len(ordered) - 1, max(0, round(pct * (len(ordered) - 1))))
+    return ordered[idx]
+
+
+def analyze(jobs: list[dict[str, Any]]) -> dict[str, Any]:
+    """Agregats : par runner, par hote, par tranche horaire, par co-residence."""
+    by_runner: dict[str, list[float]] = defaultdict(list)
+    by_host: dict[str, list[float]] = defaultdict(list)
+    by_hour: dict[int, list[float]] = defaultdict(list)
+    for job in jobs:
+        started = _parse_ts(job["started_at"])
+        if started is None:
+            continue
+        by_runner[job["runner_name"]].append(job["duration_s"])
+        by_host[job["host"]].append(job["duration_s"])
+        by_hour[started.hour].append(job["duration_s"])
+
+    # Correlation durant : mediane des durees par nombre de chevauchements
+    # lourds -- LE chiffre que l'arbitrage item 3 regarde (combien coute un
+    # voisin CPU-lourd sur le meme hote).
+    per_overlap: dict[str, list[float]] = defaultdict(list)
+    for idx, count in overlap_per_job(jobs).items():
+        bucket = str(count) if count < 3 else "3+"
+        per_overlap[bucket].append(jobs[idx]["duration_s"])
+
+    def _stats(values: list[float]) -> dict[str, float]:
+        return {
+            "n": len(values),
+            "median_s": round(median(values), 1) if values else 0.0,
+            "p90_s": round(_percentile(values, 0.9), 1),
+        }
+
+    order = {"3+": 3}
+    overlap_keys = sorted(per_overlap.keys(), key=lambda k: order.get(k, None) or int(k))
+    return {
+        "jobs_total": len(jobs),
+        "jobs_heavy": sum(1 for job in jobs if job["heavy"]),
+        "per_runner": {k: _stats(v) for k, v in sorted(by_runner.items())},
+        "per_host": {k: _stats(v) for k, v in sorted(by_host.items())},
+        "per_hour": {str(k): _stats(v) for k, v in sorted(by_hour.items())},
+        "per_overlap": {k: _stats(per_overlap[k]) for k in overlap_keys},
+        "overlap_histogram": overlap_counts(jobs),
+        "median_s": round(median([job["duration_s"] for job in jobs]), 1) if jobs else 0.0,
+    }
+
+
+def format_report(analysis: dict[str, Any]) -> str:
+    lines = [
+        f"jobs: {analysis['jobs_total']} (lourds docker/lean: {analysis['jobs_heavy']})",
+        f"duree mediane globale: {analysis['median_s']} s",
+        "",
+        "par runner (n / mediane / p90, en s) :",
+    ]
+    for name, stats in analysis["per_runner"].items():
+        lines.append(f"  {name:38s} {stats['n']:4d}  {stats['median_s']:9.1f}  {stats['p90_s']:9.1f}")
+    lines.append("")
+    lines.append("par hote (n / mediane / p90, en s) :")
+    for name, stats in analysis["per_host"].items():
+        lines.append(f"  {name:38s} {stats['n']:4d}  {stats['median_s']:9.1f}  {stats['p90_s']:9.1f}")
+    lines.append("")
+    lines.append("histogramme de co-residence (jobs lourds chevauches sur le meme hote) :")
+    for count, jobs_n in analysis["overlap_histogram"].items():
+        lines.append(f"  {count} chevauchement(s) : {jobs_n} job(s)")
+    lines.append("")
+    lines.append("duree par co-residence (n / mediane / p90, en s) -- le cout d'un voisin lourd :")
+    for bucket, stats in analysis.get("per_overlap", {}).items():
+        lines.append(f"  {bucket} chevauchement(s) : {stats['n']:4d}  {stats['median_s']:9.1f}  {stats['p90_s']:9.1f}")
+    lines.append("")
+    lines.append("par heure locale de demarrage (mediane, en s) :")
+    for hour, stats in analysis["per_hour"].items():
+        lines.append(f"  {hour}h  {stats['n']:4d}  {stats['median_s']:9.1f}")
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--repo", default="jsboige/CoursIA")
+    parser.add_argument("--workflow", default=DEFAULT_WORKFLOW)
+    parser.add_argument("--days", type=int, default=DEFAULT_DAYS)
+    parser.add_argument("--json", action="store_true", help="sortie JSON brute")
+    args = parser.parse_args(argv)
+
+    since = datetime.now(timezone.utc) - timedelta(days=args.days)
+    try:
+        jobs = collect_jobs(args.repo, args.workflow, since)
+        analysis = analyze(jobs)
+    except (RuntimeError, json.JSONDecodeError) as exc:
+        print(f"[PANNE] collecte impossible : {exc}", file=sys.stderr)
+        return EXIT_BROKEN
+
+    if args.json:
+        print(json.dumps({"workflow": args.workflow, "window_days": args.days, **analysis}, indent=2))
+    else:
+        print(format_report(analysis))
+    return EXIT_OK
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/scripts/tests/test_characterize_runner_variance.py b/scripts/tests/test_characterize_runner_variance.py
new file mode 100644
index 0000000000..bdc40a592d
--- /dev/null
+++ b/scripts/tests/test_characterize_runner_variance.py
@@ -0,0 +1,272 @@
+"""Tests de scripts/ci/characterize_runner_variance.py (#15574 items 1-2).
+
+La collecte est injectable : aucun test ne parle au reseau. Les fixtures
+construisent des reponses d'API gh minimales (workflow_runs + jobs) et
+verifient l'attribution d'hote, le decompte de co-residence et les
+agregats -- les trois grandeurs que la decision item 3 arbore.
+"""
+from __future__ import annotations
+
+import json
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "ci"))
+
+import characterize_runner_variance as crv  # noqa: E402
+
+UTC = timezone.utc
+
+
+def _ts(minutes: float) -> str:
+    return (datetime(2026, 10, 1, 12, 0, tzinfo=UTC) + timedelta(minutes=minutes)).isoformat()
+
+
+def _run(run_id: int, completed_at: str, created_at: str | None = None) -> dict:
+    # Le pre-filtre de fenetre lit created_at (le endpoint LIST ne porte pas
+    # completed_at de facon fiable) ; les deux champs sont donc fournis.
+    return {"id": run_id, "created_at": created_at or completed_at, "completed_at": completed_at}
+
+
+def _job(name: str, runner: str, started: str, completed: str, conclusion: str = "success") -> dict:
+    return {
+        "name": name,
+        "runner_name": runner,
+        "status": "completed",
+        "conclusion": conclusion,
+        "started_at": started,
+        "completed_at": completed,
+    }
+
+
+class FakeRunner:
+    """Renvoie des payloads gh api sequences par motif d'URL."""
+
+    def __init__(self, runs_payload: dict, jobs_by_run: dict[int, dict]):
+        self.runs_payload = runs_payload
+        self.jobs_by_run = jobs_by_run
+        self.calls: list[list[str]] = []
+
+    def __call__(self, argv: list[str]) -> str:
+        self.calls.append(argv)
+        url = argv[2] if len(argv) > 2 else ""
+        if "/jobs" in url:
+            run_id = int(url.split("/runs/")[1].split("/")[0])
+            return json.dumps(self.jobs_by_run.get(run_id, {"jobs": []}))
+        return json.dumps(self.runs_payload)
+
+
+# --- host_of / is_heavy ----------------------------------------------------
+
+
+def test_host_of_strips_slot_suffix_per_family():
+    assert crv.host_of("myia-po-2024-linux-docker-7") == "myia-po-2024"
+    assert crv.host_of("myia-po-2024-lean-docker-2") == "myia-po-2024"
+    assert crv.host_of("myia-po-2024-linux-waiter-11") == "myia-po-2024"
+    assert crv.host_of("myia-ai-01-wsl-7") == "myia-ai-01-wsl-7"
+
+
+def test_is_heavy_distinguishes_families():
+    assert crv.is_heavy("myia-po-2024-linux-docker-1")
+    assert crv.is_heavy("myia-po-2024-lean-docker-2")
+    assert not crv.is_heavy("myia-po-2024-linux-waiter-3")
+    assert not crv.is_heavy("myia-ai-01-wsl-7")
+
+
+# --- collect_jobs -----------------------------------------------------------
+
+
+def test_collect_jobs_keeps_selfhosted_and_drops_unplaced():
+    runs = {"workflow_runs": [_run(1, _ts(60))], "next_page_url": None}
+    jobs = {
+        1: {
+            "jobs": [
+                _job("Scripts Tests (CPU)", "myia-po-2024-linux-docker-4", _ts(0), _ts(30)),
+                _job("Hosted leg", "", _ts(0), _ts(5)),
+                {
+                    "name": "En cours",
+                    "runner_name": "myia-po-2024-linux-docker-5",
+                    "status": "in_progress",
+                    "conclusion": None,
+                    "started_at": _ts(0),
+                    "completed_at": None,
+                },
+            ]
+        }
+    }
+    collected = crv.collect_jobs("jsboige/CoursIA", "scripts-tests.yml", datetime(2026, 9, 30, tzinfo=UTC), runner=FakeRunner(runs, jobs))
+    assert len(collected) == 1
+    job = collected[0]
+    assert job["runner_name"] == "myia-po-2024-linux-docker-4"
+    assert job["host"] == "myia-po-2024"
+    assert job["duration_s"] == 1800.0
+
+
+def test_collect_jobs_paginates_via_page_param():
+    # gh api n'expose pas next_page_url dans le corps JSON : la pagination se
+    # pilote par &page=N jusqu'a une page courte (<100 runs).
+    page1 = {"workflow_runs": [_run(i, _ts(60 + i)) for i in range(100)]}
+    page2 = {"workflow_runs": [_run(101, _ts(200))]}
+    jobs = {1: {"jobs": [_job("A", "myia-po-2024-linux-docker-1", _ts(0), _ts(10))]},
+            101: {"jobs": [_job("B", "myia-po-2024-linux-docker-2", _ts(61), _ts(80))]}}
+
+    calls: list[list[str]] = []
+
+    def runner(argv: list[str]) -> str:
+        calls.append(argv)
+        url = argv[2]
+        if "/jobs" in url:
+            run_id = int(url.split("/runs/")[1].split("/")[0])
+            return json.dumps(jobs.get(run_id, {"jobs": []}))
+        if "page=2" in url:
+            return json.dumps(page2)
+        return json.dumps(page1)
+
+    collected = crv.collect_jobs("jsboige/CoursIA", "scripts-tests.yml", datetime(2026, 9, 30, tzinfo=UTC), runner=runner)
+    assert sorted(job["name"] for job in collected) == ["A", "B"]
+    assert any("page=2" in argv[2] for argv in calls), "page 2 jamais demandee"
+
+
+def test_collect_jobs_drops_runs_before_window():
+    early_done = datetime(2026, 9, 28, 12, 0, tzinfo=UTC).isoformat()
+    early_created = datetime(2026, 9, 28, 10, 0, tzinfo=UTC).isoformat()
+    runs = {"workflow_runs": [_run(1, early_done, created_at=early_created)], "next_page_url": None}
+    jobs = {1: {"jobs": [_job("A", "myia-po-2024-linux-docker-1", early_created, early_done)]}}
+    collected = crv.collect_jobs("jsboige/CoursIA", "scripts-tests.yml", datetime(2026, 9, 30, tzinfo=UTC), runner=FakeRunner(runs, jobs))
+    assert collected == []
+
+
+# --- overlap_counts ---------------------------------------------------------
+
+
+def test_overlap_counts_two_concurrent_heavy_jobs_on_same_host():
+    jobs = [
+        {"run_id": 1, "name": "A", "runner_name": "myia-po-2024-linux-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(30), "duration_s": 1800.0},
+        {"run_id": 2, "name": "B", "runner_name": "myia-po-2024-lean-docker-2", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(10), "completed_at": _ts(40), "duration_s": 1800.0},
+    ]
+    assert crv.overlap_counts(jobs) == {1: 2}
+
+
+def test_overlap_counts_sequential_heavy_jobs_do_not_overlap():
+    jobs = [
+        {"run_id": 1, "name": "A", "runner_name": "myia-po-2024-linux-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(10), "duration_s": 600.0},
+        {"run_id": 2, "name": "B", "runner_name": "myia-po-2024-lean-docker-2", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(10), "completed_at": _ts(20), "duration_s": 600.0},
+    ]
+    assert crv.overlap_counts(jobs) == {0: 2}
+
+
+def test_overlap_counts_ignores_waiters_and_foreign_hosts():
+    jobs = [
+        {"run_id": 1, "name": "A", "runner_name": "myia-po-2024-linux-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(30), "duration_s": 1800.0},
+        {"run_id": 2, "name": "w", "runner_name": "myia-po-2024-linux-waiter-3", "host": "myia-po-2024", "heavy": False,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(30), "duration_s": 1800.0},
+        {"run_id": 3, "name": "C", "runner_name": "myia-ai-01-wsl-7", "host": "myia-ai-01-wsl-7", "heavy": False,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(30), "duration_s": 1800.0},
+    ]
+    assert crv.overlap_counts(jobs) == {0: 1}
+
+
+def test_overlap_counts_boundary_touch_is_not_overlap():
+    # fin de A == debut de B : chevauchement strict requis (o_start < end).
+    jobs = [
+        {"run_id": 1, "name": "A", "runner_name": "myia-po-2024-linux-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(10), "duration_s": 600.0},
+        {"run_id": 2, "name": "B", "runner_name": "myia-po-2024-lean-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(10), "completed_at": _ts(20), "duration_s": 600.0},
+    ]
+    assert crv.overlap_counts(jobs) == {0: 2}
+
+
+# --- analyze ----------------------------------------------------------------
+
+
+def test_analyze_buckets_by_runner_host_and_hour():
+    jobs = [
+        {"run_id": 1, "name": "A", "runner_name": "myia-po-2024-linux-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(10), "duration_s": 600.0},
+        {"run_id": 2, "name": "B", "runner_name": "myia-po-2024-linux-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(60), "completed_at": _ts(75), "duration_s": 900.0},
+    ]
+    analysis = crv.analyze(jobs)
+    assert analysis["jobs_total"] == 2
+    assert analysis["jobs_heavy"] == 2
+    runner_stats = analysis["per_runner"]["myia-po-2024-linux-docker-1"]
+    assert runner_stats["n"] == 2
+    assert runner_stats["median_s"] == 750.0
+    assert analysis["per_host"]["myia-po-2024"]["n"] == 2
+    assert "12" in analysis["per_hour"]
+
+
+def test_analyze_duration_by_overlap_bucket():
+    # Deux jobs lourds simultanes (1 chevauchement chacun, dures differentes
+    # volontairement : A rapide malgre le voisin, B lent) + un job lourd seul.
+    jobs = [
+        {"run_id": 1, "name": "A", "runner_name": "myia-po-2024-linux-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(6), "duration_s": 360.0},
+        {"run_id": 2, "name": "B", "runner_name": "myia-po-2024-lean-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(30), "duration_s": 1800.0},
+        {"run_id": 3, "name": "C", "runner_name": "myia-po-2024-linux-docker-2", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(60), "completed_at": _ts(70), "duration_s": 600.0},
+    ]
+    analysis = crv.analyze(jobs)
+    assert analysis["per_overlap"]["0"]["n"] == 1
+    assert analysis["per_overlap"]["0"]["median_s"] == 600.0
+    assert analysis["per_overlap"]["1"]["n"] == 2
+    assert analysis["per_overlap"]["1"]["median_s"] == 1080.0
+
+
+def test_analyze_collapses_high_overlap_into_bucket():
+    # 4 jobs lourds mutuellement chevauches -> chacun 3 voisins -> bucket 3+.
+    jobs = [
+        {"run_id": i, "name": f"J{i}", "runner_name": f"myia-po-2024-linux-docker-{i}", "host": "myia-po-2024",
+         "heavy": True, "conclusion": "success", "started_at": _ts(0), "completed_at": _ts(10),
+         "duration_s": 100.0 * (i + 1)}
+        for i in range(4)
+    ]
+    analysis = crv.analyze(jobs)
+    assert set(analysis["per_overlap"].keys()) == {"3+"}
+    assert analysis["per_overlap"]["3+"]["n"] == 4
+
+
+def test_analyze_median_overall():
+    jobs = [
+        {"run_id": i, "name": f"J{i}", "runner_name": "myia-po-2024-linux-docker-1", "host": "myia-po-2024", "heavy": True,
+         "conclusion": "success", "started_at": _ts(i), "completed_at": _ts(i + 1), "duration_s": float(100 * (i + 1))}
+        for i in range(5)
+    ]
+    assert crv.analyze(jobs)["median_s"] == 300.0
+
+
+# --- main -------------------------------------------------------------------
+
+
+def test_main_json_shape(capsys, monkeypatch):
+    runs = {"workflow_runs": [_run(1, _ts(60))], "next_page_url": None}
+    jobs = {1: {"jobs": [_job("Scripts Tests (CPU)", "myia-po-2024-linux-docker-4", _ts(0), _ts(30))]}}
+    real_collect = crv.collect_jobs
+    monkeypatch.setattr(
+        crv, "collect_jobs", lambda *a, **k: real_collect(*a, **{**k, "runner": FakeRunner(runs, jobs)})
+    )
+    rc = crv.main(["--json", "--days", "3"])
+    assert rc == crv.EXIT_OK
+    out = json.loads(capsys.readouterr().out)
+    assert out["workflow"] == crv.DEFAULT_WORKFLOW
+    assert out["jobs_total"] == 1
+    assert "overlap_histogram" in out
+
+
+def test_main_broken_api_returns_panne(capsys, monkeypatch):
+    def boom(*a, **k):
+        raise RuntimeError("gh down")
+
+    monkeypatch.setattr(crv, "collect_jobs", boom)
+    rc = crv.main(["--days", "1"])
+    assert rc == crv.EXIT_BROKEN
+    assert "[PANNE]" in capsys.readouterr().err