#!/usr/bin/env python3
"""Cached share of the prompt by idle gap in the public GitHub Copilot
coding-agent trace (Azure/AzurePublicDataset, release ghcp-coding-agent-2026).

Standard library only, single process, single thread. Archives stay compressed
and are streamed (tar r|gz -> member -> gzip -> JSON lines).

    nice -n 19 python3 analyze.py download --data ~/data/ghcp-coding-agent-2026
    nice -n 19 python3 analyze.py run      --data ~/data/ghcp-coding-agent-2026
    nice -n 19 python3 analyze.py report   --data ~/data/ghcp-coding-agent-2026

Outputs go to <data>/out/. Nothing is written to the repository.
"""

import os

for _var in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS",
             "VECLIB_MAXIMUM_THREADS", "NUMEXPR_NUM_THREADS",
             "POLARS_MAX_THREADS", "RAYON_NUM_THREADS"):
    os.environ[_var] = "1"

import argparse
import bisect
import gzip
import hashlib
import json
import math
import sys
import tarfile
import time
import urllib.request
from collections import Counter
from datetime import datetime

RELEASE = "https://github.com/Azure/AzurePublicDataset/releases/download/ghcp-coding-agent-2026"
ASSETS = {
    "2026-06-01": (594479134, "cacad69f23d3c87ac54d788fec3f8e9c84df82c0cc9638d693340423c14e68e9"),
    "2026-06-02": (621345734, "95b0450041095dbe88181b88841c64b1e42f55a0127aa7abe91c1c9d3b424240"),
    "2026-06-03": (618403924, "622a5b0742eeb196d69d9a5ee5f6457cf82d432ddc435b9eb76eb5bb4363f1d0"),
    "2026-06-04": (572771629, "fd81d45e0149cae10aa193f2c2536ae709a5eba61b5aa886e534587e80473028"),
    "2026-06-05": (548375182, "ababba256966dc8bb005c98fbb4b72faea5301e17dcd949fd5b69e71b8239c50"),
    "2026-06-06": (124247983, "9546e3fd156d379e2420b8a0ada496ff507cf74aab32b9fc97bd1b6b353fa182"),
    "2026-06-07": (139262550, "8b6da125b486aea3eb37c752adf6be14d61dd1e6a584f5967bca7077fafe8c3c"),
}


def archive_path(data, date):
    return os.path.join(data, f"date.{date}.tar.gz")


# ---------------------------------------------------------------------------
# download
# ---------------------------------------------------------------------------

def cmd_download(args):
    """Fetch archives one at a time, capped at --rate bytes/s. A partial file is
    resumed with an HTTP Range request; the SHA-256 always covers the full file."""
    os.makedirs(args.data, exist_ok=True)
    chunk_size = 256 * 1024
    for date in args.dates:
        size, digest = ASSETS[date]
        path = archive_path(args.data, date)
        stamp = path + ".sha256"
        if os.path.exists(path) and os.path.exists(stamp):
            with open(stamp) as fh:
                if fh.read().strip() == digest and os.path.getsize(path) == size:
                    print(f"{date}: present and verified", flush=True)
                    continue
        url = f"{RELEASE}/date.{date}.tar.gz"
        part = path + ".part"
        h = hashlib.sha256()
        have = os.path.getsize(part) if os.path.exists(part) else 0
        if have:
            with open(part, "rb") as fh:
                while True:
                    block = fh.read(1 << 20)
                    if not block:
                        break
                    h.update(block)
        req = urllib.request.Request(url, headers={"Range": f"bytes={have}-"} if have else {})
        t0 = time.time()
        fetched = 0
        with urllib.request.urlopen(req, timeout=120) as resp:
            if have and resp.status != 206:
                print(f"{date}: server ignored Range; restarting", flush=True)
                have = 0
                h = hashlib.sha256()
            with open(part, "ab" if have else "wb") as out:
                while True:
                    chunk = resp.read(chunk_size)
                    if not chunk:
                        break
                    out.write(chunk)
                    h.update(chunk)
                    fetched += len(chunk)
                    ahead = fetched / args.rate - (time.time() - t0)
                    if ahead > 0:
                        time.sleep(ahead)
        got = have + fetched
        if got != size or h.hexdigest() != digest:
            raise SystemExit(f"{date}: size {got} / sha256 {h.hexdigest()} do not match the release")
        os.replace(part, path)
        with open(stamp, "w") as fh:
            fh.write(digest + "\n")
        dt = time.time() - t0
        print(f"{date}: {got / 1e6:.0f} MB verified ({fetched / 1e6:.0f} MB fetched in {dt:.0f} s, "
              f"{fetched / max(dt, 1e-9) / 1e6:.2f} MB/s)", flush=True)


# ---------------------------------------------------------------------------
# definitions
# ---------------------------------------------------------------------------

TOOL_MODEL = "tool-model"          # instant-apply edit model, as labelled in the release
SUCCESS = "Success"
# `timestamp` marks the END of an LLM call or tool batch: with end semantics,
# consecutive events in a turn are sequential; with start semantics they overlap
# (check 2). The authors' loader reads it as the start; their replications below
# keep that reading, every other gap uses start = timestamp - duration.
COMPACTION_MIN_PROMPT_BEFORE = 10000   # authors' rule (trace_metrics.py)
COMPACTION_MIN_DROP_PCT = 30
META_COMP_MIN_SEGMENTS = 10        # metadata rule: segment count falls below half
META_COMP_RATIO = 0.5
TOOLWAIT_MAX_CLIENT_GAP = 30.0     # agent call whose client gap (G2) is under 30 s

# Bin edges in seconds; a gap g falls in bin i when edges[i] <= g < edges[i+1].
TABLE_EDGES = [0, 30, 120, 240, 300, 360, 600, 1800, 3300, 3600, 3900, 7200, 86400]
TABLE_LABELS = ["<30s", "30s-2m", "2-4m", "4-5m", "5-6m", "6-10m", "10-30m",
                "30-55m", "55-60m", "60-65m", "65m-2h", "2-24h", ">24h"]
FINE_EDGES = [0, 1, 5, 15, 30, 60, 120, 180, 240, 270, 300, 330, 360, 420, 600,
              900, 1800, 2700, 3300, 3600, 3900, 4500, 7200, 21600, 86400]
FINE_LABELS = [f"{a}-{b}" for a, b in zip(FINE_EDGES, FINE_EDGES[1:])] + [f">={FINE_EDGES[-1]}"]
PAPER_EDGES = [0, 1, 5, 30, 120, 600, 3600]
PAPER_LABELS = ["<1s", "1-5s", "5-30s", "30s-2m", "2-10m", "10m-1h", ">1h"]
COARSE_EDGES = [0, 30, 240, 300, 1800, 3300, 3900, 7200, 86400]
COARSE_LABELS = ["<30s", "30s-4m", "4-5m", "5-30m", "30-55m", "55-65m", "65m-2h", "2-24h", ">24h"]
BINSETS = {"table": (TABLE_EDGES, TABLE_LABELS), "fine": (FINE_EDGES, FINE_LABELS),
           "paper": (PAPER_EDGES, PAPER_LABELS), "coarse": (COARSE_EDGES, COARSE_LABELS)}
COLD, CONC = "cold", "concurrent"

DELTA_EDGES = [-3600, -600, -60, -10, -5, -2, -1, -0.5, -0.1, 0, 0.1, 0.5, 1, 2, 5,
               10, 60, 600, 3600]


def bin_label(binset, gap):
    edges, labels = BINSETS[binset]
    return labels[bisect.bisect_right(edges, gap) - 1]


def parse_ts(value, tz_counter=None):
    """ISO-8601 -> epoch seconds (UTC). Naive values are read as UTC."""
    if not value:
        return None
    try:
        dt = datetime.fromisoformat(value)
    except (TypeError, ValueError):
        return None
    if dt.tzinfo is None:
        if tz_counter is not None:
            tz_counter["naive"] += 1
        return (dt - _EPOCH_NAIVE).total_seconds()
    if tz_counter is not None:
        tz_counter["aware"] += 1
    return dt.timestamp()


_EPOCH_NAIVE = datetime(1970, 1, 1)


# ---------------------------------------------------------------------------
# accumulators
# ---------------------------------------------------------------------------

class Acc:
    """Token-weighted cached share per key, with per-session sums for a
    session-clustered (delta-method) standard error, and optional per-call
    histograms of cached/prompt in 1-point buckets."""

    def __init__(self):
        self.g = {}       # key -> [n, c, p, sum c_s^2, sum p_s^2, sum c_s p_s, sessions, ret_c, ret_d]
        self.h = {}       # key -> 101 buckets
        self.local = {}   # key -> [n, c, p, ret_c, ret_d] for the current session

    def add(self, key, c, p, ret_c=0, ret_d=0, hist=False):
        loc = self.local.get(key)
        if loc is None:
            loc = self.local[key] = [0, 0, 0, 0, 0]
        loc[0] += 1
        loc[1] += c
        loc[2] += p
        loc[3] += ret_c
        loc[4] += ret_d
        if hist:
            hh = self.h.get(key)
            if hh is None:
                hh = self.h[key] = [0] * 101
            hh[int(100.0 * c / p + 0.5)] += 1

    def flush(self):
        for key, (n, c, p, rc, rd) in self.local.items():
            g = self.g.get(key)
            if g is None:
                g = self.g[key] = [0, 0, 0, 0.0, 0.0, 0.0, 0, 0, 0]
            g[0] += n
            g[1] += c
            g[2] += p
            g[3] += float(c) * c
            g[4] += float(p) * p
            g[5] += float(c) * p
            g[6] += 1
            g[7] += rc
            g[8] += rd
        self.local.clear()


class Sample:
    """Histogram plus a systematic sample (every k-th value) for quantiles."""

    def __init__(self, edges, every=25):
        self.edges = edges
        self.counts = [0] * (len(edges) + 1)
        self.every = every
        self.n = 0
        self.values = []

    def add(self, x):
        self.counts[bisect.bisect_right(self.edges, x)] += 1
        if self.n % self.every == 0:
            self.values.append(x)
        self.n += 1

    def summary(self):
        v = sorted(self.values)
        q = (lambda f: v[min(len(v) - 1, int(f * len(v)))] if v else None)
        below = sum(c for e, c in zip(self.edges, self.counts) if e <= -1)
        return {
            "n": self.n,
            "p01": q(0.01), "p05": q(0.05), "p25": q(0.25), "median": q(0.5),
            "p75": q(0.75), "p95": q(0.95), "p99": q(0.99),
            "share_ge_minus_1s": (1 - below / self.n) if self.n else None,
            "share_abs_lt_1s": (sum(1 for x in v if -1 < x < 1) / len(v)) if v else None,
            "hist_edges_s": self.edges, "hist_counts": self.counts,
        }


# ---------------------------------------------------------------------------
# run: one streaming pass over the archives
# ---------------------------------------------------------------------------

def iter_sessions(args, stats):
    for date in args.dates:
        path = archive_path(args.data, date)
        if not os.path.exists(path):
            raise SystemExit(f"missing {path}; run the download step first")
        taken = 0
        with tarfile.open(path, mode="r|gz") as tf:
            for member in tf:
                if not member.isfile():
                    continue
                name = member.name
                fh = tf.extractfile(member)
                if name.endswith("manifest.json"):
                    stats["manifests"][date] = json.load(fh)
                    continue
                if not name.endswith(".jsonl.gz"):
                    continue
                if args.max_sessions and taken >= args.max_sessions:
                    continue
                stats["shards"] += 1
                with gzip.GzipFile(fileobj=fh) as gz:
                    for line in gz:
                        if not line.strip():
                            continue
                        yield date, json.loads(line)
                        taken += 1
                        if args.max_sessions and taken >= args.max_sessions:
                            break


class Run:
    def __init__(self):
        self.acc = Acc()
        self.c = Counter()               # scalar counters
        self.models = Counter()          # calls per model
        self.results = Counter()
        self.initiators = Counter()
        self.segtypes = Counter()        # segment types, every 20th session
        self.segtypes_compaction = Counter()  # segment types on authors'-rule compaction calls
        self.tz = Counter()
        self.frac_digits = Counter()
        self.ts_examples = []
        self.round10 = Counter()         # timestamps on a 10 s boundary, by initiator and position
        self.d_after_call = Sample(DELTA_EDGES)       # next event start (end sem.) - call timestamp
        self.d_after_call_r10 = Sample(DELTA_EDGES)   # same, call timestamp on a 10 s boundary
        self.d_after_batch = Sample(DELTA_EDGES)      # next call start (end sem.) - batch timestamp
        self.tb = {}                     # authors' turn-boundary categories
        self.pos = {}                    # authors' call-position means
        self.g3_hist = {}                # paper bins -> 1001 buckets (0.1 points)
        self.g3e_hist = {}               # same rows, end semantics

    # -- per session ---------------------------------------------------------

    def session(self, date, sess):
        c = self.c
        c["sessions"] += 1
        sample_segments = c["sessions"] % 20 == 0
        turns = sess.get("turns") or []
        c["turns"] += len(turns)
        calls = []        # dicts for every LLM call
        batches = []      # (start, end, request_id, turn_idx) for non-empty batches
        for t_idx, turn in enumerate(turns):
            tid = turn.get("turn_id")
            mids = set()
            for pos, call in enumerate(turn.get("llm_calls") or []):
                c["calls"] += 1
                raw = call.get("timestamp")
                if len(self.ts_examples) < 5 and raw:
                    self.ts_examples.append(raw)
                r10 = False
                if raw and "." in raw:
                    frac = raw.split(".", 1)[1].rstrip("Z").split("+")[0].split("-")[0]
                    self.frac_digits[len(frac)] += 1
                    r10 = frac.strip("0") == "" and raw[17:19].isdigit() and int(raw[17:19]) % 10 == 0
                self.round10[f"{call.get('initiator_type')}|{'first' if pos == 0 else 'later'}|"
                             f"{'r10' if r10 else 'other'}"] += 1
                mids.add(call.get("message_id"))
                ts = parse_ts(raw, self.tz)
                dur = call.get("duration_ms")
                tok = call.get("tokens") or {}
                prompt = tok.get("prompt")
                cached = tok.get("cached")
                model = call.get("model")
                result = call.get("result")
                init = call.get("initiator_type")
                self.models[model] += 1
                self.results[result] += 1
                self.initiators[init] += 1
                c["prompt_sum"] += prompt or 0
                c["completion_sum"] += tok.get("completion") or 0
                c["cached_sum"] += cached or 0
                if ts is None:
                    c["calls_no_timestamp"] += 1
                if dur is None:
                    c["calls_no_duration"] += 1
                if not prompt:
                    c["calls_prompt_zero_or_null"] += 1
                if cached is None:
                    c["calls_cached_null"] += 1
                pct = tok.get("cached_pct")
                if prompt and cached is not None:
                    if cached > prompt:
                        c["calls_cached_gt_prompt"] += 1
                    if pct is not None:
                        c["cached_pct_present"] += 1
                        if abs(pct - 100.0 * cached / prompt) <= 0.01:
                            c["cached_pct_is_0_100"] += 1
                        if abs(pct - cached / prompt) <= 0.0001:
                            c["cached_pct_is_0_1"] += 1
                meta = call.get("message_metadata") or []
                if sample_segments:
                    for seg in meta:
                        self.segtypes[seg.get("type")] += 1
                        if seg.get("token_len") is not None:
                            c["segments_with_token_len"] += 1
                        c["segments_sampled"] += 1
                d = (dur or 0) / 1000.0
                calls.append({
                    "t": t_idx, "tid": tid, "pos": pos, "r10": r10,
                    "start": None if ts is None else ts - d, "end": ts,        # end semantics
                    "a_start": ts, "a_end": None if ts is None else ts + d,  # authors' reading
                    "model": model, "result": result, "init": init,
                    "prompt": prompt or 0, "cached": cached or 0, "pct": pct,
                    "meta": meta, "nseg": len(meta),
                })
            c["turns_one_message_id" if len(mids) == 1 else "turns_multi_message_id"] += 1
            rids = set()
            for b in turn.get("tool_batches") or []:
                c["batches_all"] += 1
                fcs = b.get("function_calls") or []
                if not fcs:
                    continue
                c["batches"] += 1
                c["tool_calls"] += len(fcs)
                if b.get("auto_continue") is not None:
                    c["batches_auto_continue_" + str(b.get("auto_continue"))] += 1
                rids.add(b.get("request_id"))
                bt = parse_ts(b.get("timestamp"))
                if bt is None:
                    c["batches_no_timestamp"] += 1
                    continue
                batches.append((bt - (b.get("duration_ms") or 0) / 1000.0, bt, t_idx, b.get("auto_continue")))
            if rids:
                c["turns_with_batches"] += 1
                if rids <= mids:
                    c["turns_batch_request_ids_in_call_message_ids"] += 1

        self.check_timestamps(calls, batches)
        self.authors(calls)
        self.gaps(calls, batches)
        self.acc.flush()

    # -- check 2: does `timestamp` mark the start or the end? -----------------

    def check_timestamps(self, calls, batches):
        """Within a turn the agent loop is sequential: call, tool batch, next call.
        Sort a turn's non-tool-model calls and non-empty batches by `timestamp` and
        count consecutive pairs that overlap by more than 0.5 s under each reading."""
        c = self.c
        per_turn = {}
        for x in calls:
            if x["end"] is not None and x["model"] != TOOL_MODEL:
                per_turn.setdefault(x["t"], []).append((x["end"], x["end"] - x["start"], "C", x["r10"]))
        for bs, be, t, _ in batches:
            per_turn.setdefault(t, []).append((be, be - bs, "B", False))
        for ev in per_turn.values():
            ev.sort(key=lambda e: e[0])
            for a, b in zip(ev, ev[1:]):
                kind = a[2] + b[2]
                c["pairs_" + kind] += 1
                if b[0] < a[0] + a[1] - 0.5:      # b's timestamp before a's start-semantics end
                    c["pairs_" + kind + "_overlap_start_sem"] += 1
                b_start_end_sem = b[0] - b[1]
                if b_start_end_sem < a[0] - 0.5:  # b's end-semantics start before a's end
                    c["pairs_" + kind + "_overlap_end_sem"] += 1
                if a[2] == "C":
                    (self.d_after_call_r10 if a[3] else self.d_after_call).add(b_start_end_sem - a[0])
                elif b[2] == "C":
                    self.d_after_batch.add(b_start_end_sem - a[0])

    # -- replications of the authors' code (trace_metrics.py) ----------------

    def authors(self, calls):
        c = self.c
        timed = [x for x in calls if x["a_start"] is not None]
        # _ordered_calls: prompt > 0, by timestamp (stable on file order).
        ordered = sorted((x for x in timed if x["prompt"] > 0), key=lambda x: x["a_start"])
        comp_session = False
        for prev, cur in zip(ordered, ordered[1:]):
            if (cur["model"] == prev["model"] and prev["prompt"] > COMPACTION_MIN_PROMPT_BEFORE):
                drop = round((prev["prompt"] - cur["prompt"]) * 100.0 / prev["prompt"], 1)
                if drop > COMPACTION_MIN_DROP_PCT:
                    c["authors_compaction_events"] += 1
                    comp_session = True
                    for seg in cur["meta"]:
                        self.segtypes_compaction[seg.get("type")] += 1
            if cur["model"] != TOOL_MODEL and prev["model"] != TOOL_MODEL:
                same_turn = cur["tid"] == prev["tid"]
                same_model = cur["model"] == prev["model"]
                if same_turn and same_model:
                    cat = "Intra-turn (same model)"
                elif not same_turn and same_model:
                    cat = "Turn boundary (same model)"
                elif not same_turn and not same_model:
                    cat = "Turn boundary (model switch)"
                else:
                    cat = None
                if cat:
                    r = self.tb.setdefault(cat, [0, 0.0, 0, 0.0, 0])
                    r[0] += 1
                    if prev["pct"] is not None:
                        r[1] += prev["pct"]
                        r[2] += 1
                    if cur["pct"] is not None:
                        r[3] += cur["pct"]
                        r[4] += 1
        c["authors_ordered_calls"] += len(ordered)
        if comp_session:
            c["authors_compaction_sessions"] += 1

        # cache_rate_by_call: per turn, prompt > 0, by timestamp, index <= 30.
        per_turn = {}
        for x in timed:
            if x["prompt"] > 0:
                per_turn.setdefault(x["tid"], []).append(x)
        for lst in per_turn.values():
            lst.sort(key=lambda x: x["a_start"])
            for i, x in enumerate(lst[:30], start=1):
                if x["pct"] is not None:
                    r = self.pos.setdefault(i, [0, 0.0])
                    r[0] += 1
                    r[1] += x["pct"]

        # cache_after_user_idle: sessions with >= 2 user-initiated calls.
        # G3 keeps the authors' start reading; G3e is the same rows under end semantics.
        if sum(1 for x in calls if x["init"] == "user") < 2:
            return
        first_user = {}
        for x in calls:  # first user call of the turn, file order
            if x["init"] == "user" and x["tid"] not in first_user:
                first_user[x["tid"]] = (x["prompt"], x["cached"])
        for name, k0, k1, hist in (("G3", "a_start", "a_end", self.g3_hist),
                                   ("G3e", "start", "end", self.g3e_hist)):
            turns = {}
            for x in timed:
                t = turns.get(x["tid"])
                if t is None:
                    turns[x["tid"]] = [x[k0], x[k1]]
                else:
                    t[0] = min(t[0], x[k0])
                    t[1] = max(t[1], x[k1])
            seq = sorted(((v[0], v[1], tid) for tid, v in turns.items()), key=lambda t: t[0])
            for idx in range(1, len(seq)):
                start, _, tid = seq[idx]
                prompt, cached = first_user.get(tid, (None, None))
                if not prompt or prompt <= 0:
                    continue
                idle = start - seq[idx - 1][1]
                if idle < 0:
                    c[name.lower() + "_rows_negative_idle"] += 1
                    continue
                idle = int(idle)
                rate = round(cached * 100.0 / prompt, 2)
                c[name.lower() + "_rows"] += 1
                label = bin_label("paper", idle)
                if 0 <= rate <= 100:
                    hh = hist.get(label)
                    if hh is None:
                        hh = hist[label] = [0] * 1001
                    hh[int(rate * 10 + 0.5)] += 1
                self.acc.add((name, "paper", label, "user"), min(cached, prompt), prompt)

    # -- gap definitions G1, G1m, G2, G1t ------------------------------------

    def gaps(self, calls, batches):
        c = self.c
        acc = self.acc
        timed = sorted((x for x in calls if x["start"] is not None),
                       key=lambda x: (x["start"], x["t"], x["pos"]))
        batches.sort()
        bmax = None
        b_ac = None
        j = 0
        g1_max = None            # max end over earlier non-tool-model calls
        g1t_max = None           # same, tool-model calls included
        same_model_max = {}      # model -> max end of earlier calls to it
        same_model_prompt = {}   # model -> prompt of the latest earlier call to it
        prev = None              # previous non-tool-model call
        for x in timed:
            s = x["start"]
            p = x["prompt"]
            cc = min(x["cached"], p)
            while j < len(batches) and batches[j][0] < s:
                if bmax is None or batches[j][1] >= bmax:
                    bmax, b_ac = batches[j][1], batches[j][3]
                j += 1

            # sensitivity: tool-model calls kept as targets and references
            if p > 0:
                g = COLD if g1t_max is None else (CONC if s < g1t_max else bin_label("table", s - g1t_max))
                acc.add(("G1t", "table", g, "all"), cc, p)
            g1t_max = x["end"] if g1t_max is None else max(g1t_max, x["end"])

            if x["model"] == TOOL_MODEL:
                c["targets_tool_model"] += 1
                continue
            if p <= 0:
                c["targets_prompt_zero"] += 1
                prev = x
                g1_max = x["end"] if g1_max is None else max(g1_max, x["end"])
                same_model_max[x["model"]] = max(same_model_max.get(x["model"], x["end"]), x["end"])
                same_model_prompt[x["model"]] = p
                continue
            c["targets"] += 1

            init = x["init"] if x["init"] in ("user", "agent") else "other"
            success = x["result"] == SUCCESS
            switch = prev is not None and prev["model"] != x["model"]
            comp = (prev is not None and not switch and prev["prompt"] > COMPACTION_MIN_PROMPT_BEFORE
                    and (prev["prompt"] - p) * 100.0 / prev["prompt"] > COMPACTION_MIN_DROP_PCT)
            comp_meta = (prev is not None and not switch and prev["nseg"] >= META_COMP_MIN_SEGMENTS
                         and x["nseg"] < META_COMP_RATIO * prev["nseg"])
            if comp or comp_meta:
                c["compaction_" + ("both" if comp and comp_meta else "prompt_only" if comp else "meta_only")] += 1
            clean = success and not switch and not comp and not comp_meta

            # G1: provider gap
            gap1 = None if g1_max is None else s - g1_max
            # G2: client gap (LLM calls or tool batches)
            ref2 = g1_max if bmax is None else (bmax if g1_max is None else max(g1_max, bmax))
            gap2 = None if ref2 is None else s - ref2
            # G1m: same-model gap
            mref = same_model_max.get(x["model"])
            gapm = None if mref is None else s - mref
            prev_same_prompt = same_model_prompt.get(x["model"])
            if prev_same_prompt:
                rd = min(p, prev_same_prompt)
                rc = min(cc, rd)
            else:
                rd = rc = 0

            for grp in ("all", "init=" + init) + (("clean", "clean|init=" + init) if clean else ()):
                acc.add(("ALL", "-", "-", grp), cc, p)

            toolwait = init == "agent" and gap2 is not None and 0 <= gap2 < TOOLWAIT_MAX_CLIENT_GAP
            groups1 = ["all", "init=" + init]
            if clean:
                groups1 += ["clean", "clean|init=" + init, "clean|model=" + str(x["model"])]
                if toolwait:
                    groups1.append("clean|agent_toolwait")
                    # auto_continue of the tool batch that ended last
                    if bmax is not None and ref2 == bmax:
                        groups1.append("clean|agent_toolwait|auto_continue=" + str(b_ac))
            if switch:
                groups1.append("switch")
            if comp:
                groups1.append("compaction")
            if comp_meta:
                groups1.append("compaction_meta")
            if not success:
                groups1.append("nonsuccess")
            hist_groups = ("all", "init=user", "init=agent", "clean|init=user",
                           "clean|init=agent", "clean|agent_toolwait")
            if gap1 is None:
                for grp in groups1:
                    acc.add(("G1", "table", COLD, grp), cc, p)
            elif gap1 < 0:
                for grp in groups1:
                    acc.add(("G1", "table", CONC, grp), cc, p)
            else:
                lt, lf, lc = bin_label("table", gap1), bin_label("fine", gap1), bin_label("coarse", gap1)
                for grp in groups1:
                    h = grp in hist_groups
                    acc.add(("G1", "table", lt, grp), cc, p, rc, rd, hist=h)
                    acc.add(("G1", "fine", lf, grp), cc, p, rc, rd, hist=h)
                    acc.add(("G1", "coarse", lc, grp), cc, p, rc, rd)

            groupsm = ["all"] + (["switch"] if switch else []) + (["clean"] if clean else [])
            lm = COLD if gapm is None else (CONC if gapm < 0 else bin_label("table", gapm))
            for grp in groupsm:
                acc.add(("G1m", "table", lm, grp), cc, p, rc, rd)

            groups2 = ["all", "init=" + init] + (["clean|init=" + init] if clean else [])
            if gap2 is None:
                for grp in groups2:
                    acc.add(("G2", "table", COLD, grp), cc, p)
            elif gap2 < 0:
                for grp in groups2:
                    acc.add(("G2", "table", CONC, grp), cc, p)
            else:
                lt, lf = bin_label("table", gap2), bin_label("fine", gap2)
                for grp in groups2:
                    h = grp in ("all", "init=agent")
                    acc.add(("G2", "table", lt, grp), cc, p, hist=h)
                    acc.add(("G2", "fine", lf, grp), cc, p, hist=h)

            prev = x
            g1_max = x["end"] if g1_max is None else max(g1_max, x["end"])
            same_model_max[x["model"]] = max(same_model_max.get(x["model"], x["end"]), x["end"])
            same_model_prompt[x["model"]] = p

    # -- output -------------------------------------------------------------

    def dump(self, stats, elapsed):
        def keystr(k):
            return "|".join(k)
        return {
            "meta": {
                "elapsed_s": round(elapsed, 1),
                "dates": stats["dates"], "max_sessions": stats["max_sessions"],
                "shards": stats["shards"], "manifests": stats["manifests"],
                "edges": {k: v[0] for k, v in BINSETS.items()},
                "labels": {k: v[1] for k, v in BINSETS.items()},
            },
            "counters": dict(self.c),
            "models": dict(self.models.most_common()),
            "results": {str(k): v for k, v in self.results.most_common()},
            "initiators": {str(k): v for k, v in self.initiators.most_common()},
            "segment_types_sampled": {str(k): v for k, v in self.segtypes.most_common()},
            "segment_types_on_compaction_calls": {str(k): v for k, v in self.segtypes_compaction.most_common()},
            "timestamp": {
                "examples": self.ts_examples, "tz": dict(self.tz),
                "fraction_digits": dict(self.frac_digits),
                "round10_by_initiator_position": dict(self.round10),
                "next_start_minus_call_ts": self.d_after_call.summary(),
                "next_start_minus_call_ts_round10": self.d_after_call_r10.summary(),
                "next_call_start_minus_batch_ts": self.d_after_batch.summary(),
            },
            "authors_turn_boundary": {k: {"n": v[0], "avgBefore": v[1] / v[2] if v[2] else None,
                                          "avgAfter": v[3] / v[4] if v[4] else None}
                                      for k, v in self.tb.items()},
            "authors_call_position": {str(k): v[1] / v[0] for k, v in sorted(self.pos.items())},
            "g3_hist_0p1": self.g3_hist,
            "g3e_hist_0p1": self.g3e_hist,
            "agg": {keystr(k): v for k, v in self.acc.g.items()},
            "hist": {keystr(k): v for k, v in self.acc.h.items()},
        }


def cmd_run(args):
    import resource
    out = os.path.join(args.data, "out")
    os.makedirs(out, exist_ok=True)
    stats = {"dates": args.dates, "max_sessions": args.max_sessions, "shards": 0, "manifests": {}}
    run = Run()
    t0 = time.time()
    last = t0
    for date, sess in iter_sessions(args, stats):
        run.session(date, sess)
        now = time.time()
        if now - last >= 60:
            ru = resource.getrusage(resource.RUSAGE_SELF)
            cpu = ru.ru_utime + ru.ru_stime
            print(f"[{now - t0:6.0f}s] {date} sessions={run.c['sessions']:,} calls={run.c['calls']:,} "
                  f"own_cpu={100 * cpu / (now - t0):.0f}% maxrss={ru.ru_maxrss / 1e6:.0f}MB", flush=True)
            last = now
    res = run.dump(stats, time.time() - t0)
    path = os.path.join(out, f"results-{args.tag}.json")
    with open(path, "w") as fh:
        json.dump(res, fh)
    print(f"wrote {path} ({run.c['sessions']:,} sessions, {run.c['calls']:,} calls, "
          f"{time.time() - t0:.0f} s)", flush=True)


# ---------------------------------------------------------------------------
# report: tables (markdown) and one figure (SVG) from results-<tag>.json
# ---------------------------------------------------------------------------

CARD = {  # DATASET_CARD.md, release ghcp-coding-agent-2026
    "sessions": 301026, "turns": 1189581, "calls": 9310255, "batches": 7767261,
    "tool_calls": 8681190, "models": 37, "prompt_sum": 631.4e9, "completion_sum": 4.91e9,
    "cached_sum": 540.95e9,
}
NOTEBOOK_G3_ROWS = 818990          # vs_cache_after_user_idle rows in the public notebook
NOTEBOOK_COMPACTION_ROWS = 42424   # vs_compaction_events rows in the public notebook
FIG15_PAPER = {"<1s": ">=95", "1-5s": ">=95", "5-30s": ">=95", "30s-2m": ">=95",
               "2-10m": "~70", "10m-1h": "0-5", ">1h": "0-5"}
THIN_CALLS, THIN_SESSIONS = 1000, 200
ROW_BINS = [COLD, CONC] + TABLE_LABELS


def stat(g):
    """Accumulator row -> share, session-clustered 95% CI, n, sessions, retained share."""
    n, c, p, sc2, sp2, scp, s, rc, rd = g
    r = c / p if p else float("nan")
    se = float("nan")
    if s > 1 and p:
        se = math.sqrt(max(s / (s - 1) * (sc2 - 2 * r * scp + r * r * sp2) / (p * p), 0.0))
    return {"share": r, "lo": r - 1.96 * se, "hi": r + 1.96 * se, "n": n, "sessions": s,
            "retained": rc / rd if rd else None, "thin": n < THIN_CALLS or s < THIN_SESSIONS}


def hist_q(h, q, scale=1.0):
    if not h or not sum(h):
        return None
    target, cum = q * sum(h), 0
    for i, v in enumerate(h):
        cum += v
        if cum >= target:
            return i / scale
    return (len(h) - 1) / scale


def pct(x, d=1):
    return "-" if x is None or (isinstance(x, float) and math.isnan(x)) else f"{100 * x:.{d}f}"


def fshare(st, ci=True):
    if st is None:
        return "-"
    s = pct(st["share"])
    if ci and not math.isnan(st["lo"]):
        s += f" [{pct(st['lo'])}, {pct(st['hi'])}]"
    return f"({s})" if st["thin"] else s


def fn(st):
    return "-" if st is None else f"{st['n']:,}"


def md_table(header, rows):
    out = ["| " + " | ".join(header) + " |", "|" + "|".join("---" for _ in header) + "|"]
    out += ["| " + " | ".join(str(x) for x in r) + " |" for r in rows]
    return "\n".join(out)


class Results:
    def __init__(self, res):
        self.res = res
        self.agg = res["agg"]
        self.hist = res["hist"]

    def st(self, gap, binset, label, group):
        g = self.agg.get(f"{gap}|{binset}|{label}|{group}")
        return stat(g) if g else None

    def med(self, gap, binset, label, group):
        return hist_q(self.hist.get(f"{gap}|{binset}|{label}|{group}"), 0.5)

    def groups(self, gap, binset, prefix):
        out = Counter()
        for k, v in self.agg.items():
            g, b, _, grp = k.split("|", 3)
            if g == gap and b == binset and grp.startswith(prefix):
                out[grp[len(prefix):]] += v[0]
        return out


def model_class(r, group):
    """Classify a model's retention on G1 table bins.
    drop5 = share(5-6m) / share(4-5m): under 0.25 is a 5-min cliff, 0.25-0.8 a partial drop.
    half = first bin after <30s whose share is under half the <30s share; a thin bin counts only
    when its CI upper bound is under half too."""
    base = r.st("G1", "table", "<30s", group)
    s45, s56 = r.st("G1", "table", "4-5m", group), r.st("G1", "table", "5-6m", group)
    if base is None or base["thin"] or not s45 or not s56 or not s45["share"]:
        return {"drop5": None, "half": None, "kind": "-", "label": "-"}
    drop5 = s56["share"] / s45["share"]
    half = None
    for lab in TABLE_LABELS[1:]:
        st = r.st("G1", "table", lab, group)
        if st and st["share"] < 0.5 * base["share"] and (not st["thin"] or st["hi"] < 0.5 * base["share"]):
            half = lab
            break
    kind = "5-min cliff" if drop5 < 0.25 else "partial drop at 5 min" if drop5 < 0.8 else "no 5-min drop"
    label = kind if kind == "5-min cliff" else f"{kind}; half by {half or 'none (holds)'}"
    return {"drop5": drop5, "drop5_thin": s45["thin"] or s56["thin"], "half": half, "kind": kind,
            "label": label}


def build_tables(res):
    r = Results(res)
    c = res["counters"]
    out = []
    w = out.append
    S = max(c.get("sessions", 0), 1)
    T = max(c.get("turns", 0), 1)
    scale = 100.0 if c.get("cached_pct_is_0_1", 0) > c.get("cached_pct_is_0_100", 0) else 1.0
    meta = res["meta"]
    w(f"# Results ({', '.join(meta['dates'])}; max_sessions={meta['max_sessions'] or 'all'}; "
      f"{meta['elapsed_s']:.0f} s)\n")
    w("Shares are token-weighted (sum cached / sum prompt) in percent with a 95% CI clustered by "
      f"session. Parentheses mark thin bins (under {THIN_CALLS:,} calls or {THIN_SESSIONS} sessions).\n")

    # -- checks -----------------------------------------------------------
    w("## Check 1: counts against the dataset card\n")
    rows = []
    for key, name in [("sessions", "Sessions"), ("turns", "Turns"), ("calls", "LLM calls"),
                      ("batches", "Tool batches (non-empty)"), ("tool_calls", "Tool calls")]:
        v = c.get(key, 0)
        rows.append([name, f"{v:,}", f"{CARD[key]:,}", f"{100 * (v / CARD[key] - 1):+.2f}%"])
    rows.append(["Model labels", f"{len(res['models'])} ({len([m for m in res['models'] if m != TOOL_MODEL])} without {TOOL_MODEL})",
                 CARD["models"], ""])
    for key, name in [("prompt_sum", "Prompt tokens"), ("completion_sum", "Completion tokens"),
                      ("cached_sum", "Cached tokens")]:
        v = c.get(key, 0)
        rows.append([name, f"{v / 1e9:.2f}B", f"{CARD[key] / 1e9:.2f}B",
                     f"{100 * (v / CARD[key] - 1):+.2f}%"])
    rows.append(["Cached share of prompt tokens", f"{100 * c.get('cached_sum', 0) / max(c.get('prompt_sum', 1), 1):.1f}%",
                 "85.7%", ""])
    rows.append(["Turns per session", f"{c.get('turns', 0) / S:.2f}", "3.95", ""])
    rows.append(["LLM calls per turn", f"{c.get('calls', 0) / T:.2f}", "7.83", ""])
    rows.append(["Tool calls per turn", f"{c.get('tool_calls', 0) / T:.2f}", "7.3", ""])
    w(md_table(["Metric", "This run", "Card", "Diff"], rows) + "\n")
    w("Vocabularies: results " + json.dumps(res["results"]) + "; initiators "
      + json.dumps(res["initiators"]) + "\n")
    w(f"cached_pct units: 0-100 matches {c.get('cached_pct_is_0_100', 0):,}, 0-1 matches "
      f"{c.get('cached_pct_is_0_1', 0):,} of {c.get('cached_pct_present', 0):,}. "
      f"cached > prompt: {c.get('calls_cached_gt_prompt', 0):,}. Prompt 0 or null: "
      f"{c.get('calls_prompt_zero_or_null', 0):,}. No timestamp: {c.get('calls_no_timestamp', 0):,}.\n")
    w("Manifests: " + "; ".join(
        f"{d}: {m.get('n_sessions', '?'):,} sessions in {m.get('n_shards', '?')} shards, "
        f"complete={m.get('complete')}" if isinstance(m, dict) else f"{d}: {m}"
        for d, m in sorted(meta["manifests"].items())) + "\n")

    ts = res["timestamp"]
    w("## Check 2: `timestamp` marks the start or the end of a call or batch\n")
    w(f"Examples: {ts['examples']}; tz {ts['tz']}; fraction digits {ts['fraction_digits']}.\n")
    w("Within a turn the agent loop is sequential. Consecutive non-tool-model calls and non-empty tool "
      "batches, sorted by `timestamp`, overlap by more than 0.5 s:\n")
    rows = []
    tot_s = tot_e = tot_n = 0
    for kind, name in (("CC", "call -> call"), ("CB", "call -> batch"), ("BC", "batch -> call"),
                       ("BB", "batch -> batch")):
        n = c.get("pairs_" + kind, 0)
        s_ov, e_ov = c.get(f"pairs_{kind}_overlap_start_sem", 0), c.get(f"pairs_{kind}_overlap_end_sem", 0)
        tot_n, tot_s, tot_e = tot_n + n, tot_s + s_ov, tot_e + e_ov
        rows.append([name, f"{n:,}", f"{s_ov:,} ({100 * s_ov / max(n, 1):.2f}%)",
                     f"{e_ov:,} ({100 * e_ov / max(n, 1):.3f}%)"])
    rows.append(["all", f"{tot_n:,}", f"{tot_s:,} ({100 * tot_s / max(tot_n, 1):.2f}%)",
                 f"{tot_e:,} ({100 * tot_e / max(tot_n, 1):.3f}%)"])
    w(md_table(["Consecutive pair", "n", "overlap if timestamp = start", "overlap if timestamp = end"], rows) + "\n")
    rows = []
    f = (lambda v: "-" if v is None else f"{v:.2f}")
    for name, key in (("call timestamp -> next event start", "next_start_minus_call_ts"),
                      ("same, call timestamp on a 10 s boundary", "next_start_minus_call_ts_round10"),
                      ("batch timestamp -> next call start", "next_call_start_minus_batch_ts")):
        s = ts[key]
        rows.append([name, f"{s['n']:,}", f(s["p05"]), f(s["p25"]), f(s["median"]), f(s["p75"]),
                     f(s["p95"]), pct(s["share_ge_minus_1s"])])
    w("Delta from an event's `timestamp` to the end-semantics start of the next event in the turn (s):\n")
    w(md_table(["Delta", "n", "p05", "p25", "median", "p75", "p95", "% >= -1 s"], rows) + "\n")
    a = ts["next_start_minus_call_ts"]
    ok = (a["median"] is not None and 0 <= a["median"] <= 5 and (a["share_ge_minus_1s"] or 0) >= 0.95
          and tot_e < 0.01 * max(tot_n, 1))
    w(f"Rule for end semantics: median call delta in [0, 5] s, at least 95% at -1 s or more, and under 1% "
      f"of pairs overlapping. Result: **{'END semantics confirmed' if ok else 'NOT confirmed'}**.\n")
    r10 = ts["round10_by_initiator_position"]
    rows = []
    for init in ("user", "agent", ""):
        for posn in ("first", "later"):
            k1, k2 = f"{init}|{posn}|r10", f"{init}|{posn}|other"
            tot = r10.get(k1, 0) + r10.get(k2, 0)
            if tot:
                rows.append([init or "(empty)", posn, f"{tot:,}", f"{r10.get(k1, 0):,}",
                             f"{100 * r10.get(k1, 0) / tot:.1f}%"])
    w("Call timestamps on an exact 10 s boundary (chance rate about 0.01%):\n")
    w(md_table(["Initiator", "position in turn", "calls", "on 10 s boundary", "share"], rows) + "\n")
    w(f"`message_id` is per turn: {c.get('turns_one_message_id', 0):,} of "
      f"{c.get('turns_one_message_id', 0) + c.get('turns_multi_message_id', 0):,} turns carry one value; "
      f"batch `request_id`s are a subset of the turn's call `message_id`s in "
      f"{c.get('turns_batch_request_ids_in_call_message_ids', 0):,} of {c.get('turns_with_batches', 0):,} "
      "turns with batches. Batch-to-call joins are therefore per turn, not per call.\n")

    w("## Check 3: the paper's averages, rebuilt with the authors' code paths\n")
    tb = res["authors_turn_boundary"]
    rows = []
    bn = ba = 0.0
    for cat in ("Intra-turn (same model)", "Turn boundary (same model)", "Turn boundary (model switch)"):
        v = tb.get(cat)
        if not v:
            continue
        b4, af = (v["avgBefore"] or 0) * scale, (v["avgAfter"] or 0) * scale
        rows.append([cat, f"{v['n']:,}", f"{b4:.1f}", f"{af:.1f}", f"{af - b4:+.1f}"])
        if cat.startswith("Turn boundary"):
            bn += v["n"]
            ba += af * v["n"]
    if bn:
        rows.append(["Turn boundary (both)", f"{int(bn):,}", "", f"{ba / bn:.1f}", ""])
    w(md_table(["Pair", "n", "mean cached_pct before", "after", "change"], rows) + "\n")
    w("Paper: 90% within a turn, 55% across turn boundaries; Figure 14: same-model boundary -26%, "
      "model switch -67% with 8% after a switch.\n")
    pos = res["authors_call_position"]
    w("Call position (mean cached_pct): " + ", ".join(
        f"#{k}: {pos[k] * scale:.1f}" for k in ("1", "2", "3", "4", "5", "10", "20", "30") if k in pos)
      + ". Paper: 45, 86, then 92-94.\n")
    ev, oc = c.get("authors_compaction_events", 0), max(c.get("authors_ordered_calls", 0), 1)
    w(f"Compaction (authors' rule): {ev:,} events = {100 * ev / oc:.2f}% of calls (paper 0.5%; notebook "
      f"{NOTEBOOK_COMPACTION_ROWS:,} rows); {c.get('authors_compaction_sessions', 0):,} sessions = "
      f"{100 * c.get('authors_compaction_sessions', 0) / S:.1f}% (paper 7.8%).\n")
    w(f"Compaction flags on target calls (G1 pass): prompt rule only {c.get('compaction_prompt_only', 0):,}; "
      f"segment-count rule only {c.get('compaction_meta_only', 0):,}; both {c.get('compaction_both', 0):,}. "
      f"Segments with a non-null token_len: {c.get('segments_with_token_len', 0):,} of "
      f"{c.get('segments_sampled', 0):,} sampled.\n")
    ac = {k[len("batches_auto_continue_"):]: v for k, v in c.items() if k.startswith("batches_auto_continue_")}
    w(f"Tool batch auto_continue values (non-null): {json.dumps(ac)}.\n")
    w("Segment types, every 20th session: " + json.dumps(res["segment_types_sampled"]) + "\n")
    w("Segment types on compaction calls: " + json.dumps(res["segment_types_on_compaction_calls"]) + "\n")

    w("## Check 4: Figure 15 replication (G3, inter-turn idle, first user call of the turn)\n")
    w(f"Rows: {c.get('g3_rows', 0):,} (notebook {NOTEBOOK_G3_ROWS:,}); dropped for negative idle "
      f"{c.get('g3_rows_negative_idle', 0):,}. G3e uses the same rows under end semantics: "
      f"{c.get('g3e_rows', 0):,} rows, {c.get('g3e_rows_negative_idle', 0):,} dropped.\n")
    rows = []
    for lab in PAPER_LABELS:
        h, he = res["g3_hist_0p1"].get(lab), res["g3e_hist_0p1"].get(lab)
        q3 = (lambda hh, q: "-" if hh is None else f"{hist_q(hh, q, 10):.1f}")
        rows.append([lab, f"{sum(h):,}" if h else "-", q3(h, 0.25), q3(h, 0.5), q3(h, 0.75),
                     FIG15_PAPER[lab], fshare(r.st("G3", "paper", lab, "user")),
                     f"{sum(he):,}" if he else "-", q3(he, 0.5), fshare(r.st("G3e", "paper", lab, "user"))])
    w(md_table(["Idle", "G3 rows", "p25", "median", "p75", "paper median", "G3 token-weighted",
                "G3e rows", "G3e median", "G3e token-weighted"], rows) + "\n")

    # -- new measurements ---------------------------------------------------
    w("## A. G1 (idle since the end of the previous LLM call), clean calls\n")
    rows = []
    for lab in ROW_BINS:
        u, ag, tw = (r.st("G1", "table", lab, g) for g in
                     ("clean|init=user", "clean|init=agent", "clean|agent_toolwait"))
        mu, ma = r.med("G1", "table", lab, "clean|init=user"), r.med("G1", "table", lab, "clean|init=agent")
        rows.append([lab, fshare(u), fn(u), "-" if mu is None else mu, fshare(ag), fn(ag),
                     "-" if ma is None else ma, fshare(tw), fn(tw)])
    w(md_table(["G1", "user share", "n", "per-call median", "agent share", "n", "per-call median",
                "agent after tool batch (G2 < 30 s)", "n"], rows) + "\n")

    w("## A2. Agent calls after a tool batch, by that batch's auto_continue (G1, clean)\n")
    w("auto_continue is the schema's \"whether the agent auto-continued after this batch\"; None is unrecorded.\n")
    rows = []
    for lab in ROW_BINS:
        cells = [lab]
        for ac in ("True", "False", "None"):
            st = r.st("G1", "table", lab, "clean|agent_toolwait|auto_continue=" + ac)
            cells += [fshare(st, False), fn(st)]
        rows.append(cells)
    w(md_table(["G1", "auto_continue=True", "n", "False", "n", "None", "n"], rows) + "\n")

    w("## B. G1, all calls and by initiator, with the retained share\n")
    w("Retained share = cached / min(prompt, prompt of the previous call to the same model).\n")
    rows = []
    for lab in ROW_BINS:
        al, u, ag, ns = (r.st("G1", "table", lab, g) for g in ("all", "init=user", "init=agent", "nonsuccess"))
        cl = r.st("G1", "table", lab, "clean")
        rows.append([lab, fshare(al), fn(al), "-" if al is None else f"{al['sessions']:,}",
                     fshare(u, False), fshare(ag, False), fshare(cl, False),
                     "-" if not cl or cl["retained"] is None else pct(cl["retained"]),
                     fshare(ns, False), fn(ns)])
    w(md_table(["G1", "all", "n", "sessions", "user", "agent", "clean", "clean retained",
                "non-success", "n"], rows) + "\n")

    w("## C. G2 (client gap: previous LLM call or tool batch) against G1\n")
    rows = []
    for lab in ROW_BINS:
        cells = [lab]
        for gap in ("G1", "G2"):
            for grp in ("init=user", "init=agent"):
                st = r.st(gap, "table", lab, grp)
                cells += [fshare(st, False), fn(st)]
        rows.append(cells)
    w(md_table(["Gap", "G1 user", "n", "G1 agent", "n", "G2 user", "n", "G2 agent", "n"], rows) + "\n")

    w("## D. By anonymized model (G1 coarse bins, clean calls, user and agent pooled)\n")
    totals = r.groups("G1", "coarse", "clean|model=")
    top = [m for m, n in totals.most_common() if n >= 20000][:12]
    cols = COARSE_LABELS
    rows = []
    for m in top:
        mc = model_class(r, "clean|model=" + m)
        d5 = "-" if mc["drop5"] is None else f"{mc['drop5']:.2f}" + (" (thin)" if mc["drop5_thin"] else "")
        rows.append([m, f"{totals[m]:,}"] + [fshare(r.st("G1", "coarse", b, "clean|model=" + m), False)
                                            for b in cols] + [d5, mc["label"]])
    w(md_table(["Model", "clean calls"] + cols + ["5-6m / 4-5m", "class"], rows) + "\n")
    w(f"Models with 20,000+ clean calls: {len(top)}. `5-6m / 4-5m` is the ratio of the table-bin shares "
      "across 5 min: under 0.25 is a 5-min cliff, 0.25-0.8 a partial drop. `half by` is the first table "
      "bin whose share is under half of the model's <30s share (a thin bin counts only when its CI upper "
      "bound is under half). Parentheses mark thin bins.\n")
    rows = []
    for m in top:
        cells = [m]
        for b in ("30s-4m", "4-5m", "5-30m", "55-65m", "65m-2h", "2-24h"):
            st = r.st("G1", "coarse", b, "clean|model=" + m)
            cells += [fshare(st, True), fn(st)]
        rows.append(cells)
    w(md_table(["Model", "30s-4m [CI]", "n", "4-5m [CI]", "n", "5-30m [CI]", "n", "55-65m [CI]", "n",
                "65m-2h [CI]", "n", "2-24h [CI]", "n"], rows) + "\n")
    if top:
        w("### D2. Top models on the table bins (G1, clean)\n")
        cols2 = ["<30s", "4-5m", "5-6m", "6-10m", "10-30m", "30-55m", "55-60m", "60-65m", "65m-2h", "2-24h"]
        rows = []
        for m in top:
            cells = [m]
            for b in cols2:
                st = r.st("G1", "table", b, "clean|model=" + m)
                cells.append("-" if st is None else f"{fshare(st, False)} / {st['n']:,}")
            rows.append(cells)
        w(md_table(["Model"] + [f"{b} share / n" for b in cols2], rows) + "\n")

    w("## E. Model switch and compaction, under G1 and the same-model gap G1m\n")
    rows = []
    for lab in ROW_BINS:
        cells = [lab]
        for gap, grp in (("G1", "switch"), ("G1", "compaction"), ("G1", "compaction_meta"),
                         ("G1m", "switch"), ("G1m", "clean"), ("G1m", "all")):
            st = r.st(gap, "table", lab, grp)
            cells += [fshare(st, False), fn(st)]
        rows.append(cells)
    w(md_table(["Gap", "G1 switch", "n", "G1 compaction (prompt rule)", "n", "G1 compaction (segment rule)",
                "n", "G1m switch", "n", "G1m clean", "n", "G1m all", "n"], rows) + "\n")
    w("Under G1m, `cold` on a switch means the session had not called that model before.\n")

    w("## F. Sensitivity: tool-model calls kept as targets and references (G1t)\n")
    rows = []
    for lab in ROW_BINS:
        a1, at = r.st("G1", "table", lab, "all"), r.st("G1t", "table", lab, "all")
        rows.append([lab, fshare(a1, False), fn(a1), fshare(at, False), fn(at)])
    w(md_table(["Gap", "G1 all (tool-model excluded)", "n", "G1t all (included)", "n"], rows) + "\n")
    w(f"Targets: {c.get('targets', 0):,}; skipped tool-model calls: {c.get('targets_tool_model', 0):,}; "
      f"skipped prompt-0 calls: {c.get('targets_prompt_zero', 0):,}.\n")

    w("## G. Fine bins (G1 clean; G2 agent)\n")
    rows = []
    for lab in FINE_LABELS:
        cells = [lab]
        for gap, grp in (("G1", "clean|init=user"), ("G1", "clean|init=agent"),
                         ("G1", "clean|agent_toolwait"), ("G2", "init=agent")):
            st = r.st(gap, "fine", lab, grp)
            cells += [fshare(st), fn(st)]
        rows.append(cells)
    w(md_table(["Gap (s)", "G1 user", "n", "G1 agent", "n", "G1 agent after tool batch", "n",
                "G2 agent", "n"], rows) + "\n")

    # -- go/no-go indicators --------------------------------------------------
    w("## Go/no-go indicators (new parts only)\n")
    w("(a) Adjacent-bin drops near 5 min and 1 h. `sep` = both bins non-thin and CIs do not overlap.\n")
    rows = []
    pairs = [("fine", "240-270", "270-300"), ("fine", "270-300", "300-330"), ("fine", "300-330", "330-360"),
             ("fine", "330-360", "360-420"), ("table", "4-5m", "5-6m"), ("fine", "2700-3300", "3300-3600"),
             ("fine", "3300-3600", "3600-3900"), ("fine", "3600-3900", "3900-4500"),
             ("table", "55-60m", "60-65m"), ("table", "60-65m", "65m-2h")]
    for grp in ("clean|init=user", "clean|init=agent"):
        for bs, l1, l2 in pairs:
            s1, s2 = r.st("G1", bs, l1, grp), r.st("G1", bs, l2, grp)
            if not s1 or not s2:
                continue
            sep = (not s1["thin"] and not s2["thin"] and (s2["hi"] < s1["lo"] or s2["lo"] > s1["hi"]))
            rows.append([grp, f"{l1} -> {l2}", fshare(s1), fshare(s2),
                         f"{100 * (s2['share'] - s1['share']):+.1f}", "sep" if sep else ""])
    w(md_table(["Group", "Bins", "before", "after", "change (pts)", ""], rows) + "\n")
    long_tw = sum((r.st("G1", "table", lab, "clean|agent_toolwait") or {"n": 0})["n"]
                  for lab in TABLE_LABELS[TABLE_LABELS.index("5-6m"):])
    w(f"(b) Clean agent calls after a tool batch with G1 >= 5 min: {long_tw:,} (threshold 1,000). "
      "Compare with the user column in table A and with table A2.\n")
    classes = Counter(model_class(r, "clean|model=" + m)["kind"] for m in top)
    w("(c) Models in table D by behaviour across 5 min: "
      + ", ".join(f"{k}: {v}" for k, v in classes.most_common()) + ".\n")
    rows = []
    for grp in ("clean|init=user", "clean|init=agent"):
        for lab in TABLE_LABELS:
            st, md = r.st("G1", "table", lab, grp), r.med("G1", "table", lab, grp)
            if st and not st["thin"] and md is not None:
                rows.append([grp, lab, pct(st["share"]), md, f"{md - 100 * st['share']:+.1f}"])
    w("(d) Token-weighted share against the per-call median:\n")
    w(md_table(["Group", "G1", "token-weighted", "per-call median", "median - share"], rows) + "\n")
    return "\n".join(out)


def svg_escape(s):
    return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


MODEL_COLORS = ["#1f5fa8", "#c0392b", "#2e8b57", "#8e44ad", "#d68910", "#555555"]


def build_svg(res):
    r = Results(res)
    W = 820
    ml, mr = 70, 24
    PH = 300                       # plot height of each panel
    top1 = 118                     # top of panel A plot area
    top2 = top1 + PH + 120         # top of panel B plot area
    H = top2 + PH + 96
    lx0, lx1 = math.log10(0.5), math.log10(86400)
    font = 'font-family="Helvetica, Arial, sans-serif"'

    def X(s):
        return ml + (math.log10(s) - lx0) / (lx1 - lx0) * (W - ml - mr)

    o = [f'<svg xmlns="http://www.w3.org/2000/svg" width="{W}" height="{H}" viewBox="0 0 {W} {H}">',
         f'<rect width="{W}" height="{H}" fill="#ffffff"/>',
         f'<text x="{ml}" y="26" {font} font-size="16" font-weight="bold">'
         'Cached share of prompt tokens by idle time before the call</text>',
         f'<text x="{ml}" y="46" {font} font-size="12" fill="#444">GitHub Copilot coding agent, public trace, '
         'June 1-7, 2026. Clean calls: success, same model as the previous call, no compaction.</text>']

    def panel(top, title, series, binset):
        def Y(v):
            return top + (1 - v) * PH
        o.append(f'<text x="{ml}" y="{top - 46}" {font} font-size="13" font-weight="bold">'
                 f'{svg_escape(title)}</text>')
        lx, ly = ml, top - 26
        for grp, name, col in series:
            if lx + 40 + 6.6 * len(name) > W - mr:
                lx, ly = ml, ly + 16
            o.append(f'<line x1="{lx}" x2="{lx + 22}" y1="{ly - 4}" y2="{ly - 4}" stroke="{col}" '
                     'stroke-width="2.5"/>')
            o.append(f'<text x="{lx + 28}" y="{ly}" {font} font-size="12">{svg_escape(name)}</text>')
            lx += 40 + 6.6 * len(name)
        for v in (0, 0.25, 0.5, 0.75, 1.0):
            o.append(f'<line x1="{ml}" x2="{W - mr}" y1="{Y(v):.1f}" y2="{Y(v):.1f}" stroke="#e5e5e5"/>')
            o.append(f'<text x="{ml - 8}" y="{Y(v) + 4:.1f}" {font} font-size="11" text-anchor="end">'
                     f'{int(v * 100)}%</text>')
        for s, lab in ((1, "1s"), (10, "10s"), (60, "1m"), (300, "5m"), (600, "10m"), (1800, "30m"),
                       (3600, "1h"), (21600, "6h"), (86400, "1d")):
            o.append(f'<line x1="{X(s):.1f}" x2="{X(s):.1f}" y1="{Y(0):.1f}" y2="{Y(0) + 5:.1f}" '
                     'stroke="#333"/>')
            o.append(f'<text x="{X(s):.1f}" y="{Y(0) + 18:.1f}" {font} font-size="11" '
                     f'text-anchor="middle">{lab}</text>')
        for s, lab in ((300, "5 min"), (3600, "1 h")):
            o.append(f'<line x1="{X(s):.1f}" x2="{X(s):.1f}" y1="{Y(1):.1f}" y2="{Y(0):.1f}" stroke="#888" '
                     'stroke-dasharray="4 4"/>')
            o.append(f'<text x="{X(s) + 4:.1f}" y="{Y(1) + 12:.1f}" {font} font-size="11" '
                     f'fill="#666">{lab}</text>')
        o.append(f'<line x1="{ml}" x2="{W - mr}" y1="{Y(0):.1f}" y2="{Y(0):.1f}" stroke="#333"/>')
        o.append(f'<line x1="{ml}" x2="{ml}" y1="{Y(1):.1f}" y2="{Y(0):.1f}" stroke="#333"/>')
        o.append(f'<text x="{(ml + W - mr) / 2:.0f}" y="{Y(0) + 38:.0f}" {font} font-size="12" '
                 'text-anchor="middle">Idle time since the end of the previous LLM call in the session '
                 '(G1, log scale)</text>')
        o.append(f'<text transform="translate(18,{top + PH / 2:.0f}) rotate(-90)" {font} font-size="12" '
                 'text-anchor="middle">Cached tokens / prompt tokens</text>')
        edges, labels = BINSETS[binset]
        for grp, name, col in series:
            pts = []
            for i, lab in enumerate(labels[:-1]):
                st = r.st("G1", binset, lab, grp)
                if st is None or math.isnan(st["share"]):
                    continue
                a, b = edges[i], edges[i + 1]
                pts.append((X(math.sqrt(max(a, 0.5) * b)), st))
            solid = [(x, st) for x, st in pts if not st["thin"]]
            if len(solid) > 1:
                o.append(f'<polyline fill="none" stroke="{col}" stroke-width="2" points="'
                         + " ".join(f"{x:.1f},{Y(st['share']):.1f}" for x, st in solid) + '"/>')
            for x, st in pts:
                if not math.isnan(st["lo"]) and not st["thin"]:
                    o.append(f'<line x1="{x:.1f}" x2="{x:.1f}" y1="{Y(min(st["hi"], 1)):.1f}" '
                             f'y2="{Y(max(st["lo"], 0)):.1f}" stroke="{col}" stroke-width="1.2"/>')
                fill = "#ffffff" if st["thin"] else col
                o.append(f'<circle cx="{x:.1f}" cy="{Y(st["share"]):.1f}" r="3.2" fill="{fill}" '
                         f'stroke="{col}" stroke-width="1.2"/>')

    panel(top1, "A. By who opened the call (fine bins)",
          [("clean|init=user", "User-opened calls", "#1f5fa8"),
           ("clean|init=agent", "Agent follow-up calls", "#c0392b"),
           ("clean|agent_toolwait", "Agent calls, tool batch ended under 30 s before", "#2e8b57")], "fine")
    totals = r.groups("G1", "coarse", "clean|model=")
    top = [m for m, n in totals.most_common() if n >= 20000][:12]
    kinds = {m: model_class(r, "clean|model=" + m)["kind"] for m in top}
    cliff = [m for m in top if kinds[m] == "5-min cliff"]
    models = cliff[:1] + [m for m in top if kinds[m] != "5-min cliff"]
    names = {m: (f"{m} (largest of {len(cliff)} with a 5-min cliff)" if m in cliff else m) for m in models}
    panel(top2, "B. By anonymized model: the largest 5-min-cliff model and every top-12 model without one",
          [("clean|model=" + m, names[m], MODEL_COLORS[i % len(MODEL_COLORS)]) for i, m in enumerate(models)],
          "table")
    for k, line in enumerate((
            "Whiskers: 95% CI clustered by session. Hollow markers: under 1,000 calls or 200 sessions, "
            "not joined. Points sit at the geometric",
            "middle of each bin. Source: Azure/AzurePublicDataset release ghcp-coding-agent-2026 (CC BY); "
            "analysis: analyze.py in this folder.")):
        o.append(f'<text x="{ml}" y="{H - 30 + 14 * k}" {font} font-size="10" fill="#666">'
                 f'{svg_escape(line)}</text>')
    o.append("</svg>")
    return "\n".join(o) + "\n"


def cmd_report(args):
    out = os.path.join(args.data, "out")
    with open(os.path.join(out, f"results-{args.tag}.json")) as fh:
        res = json.load(fh)
    md_path = os.path.join(out, f"tables-{args.tag}.md")
    svg_path = os.path.join(out, f"cache-by-gap-{args.tag}.svg")
    with open(md_path, "w") as fh:
        fh.write(build_tables(res))
    with open(svg_path, "w") as fh:
        fh.write(build_svg(res))
    print(f"wrote {md_path}\nwrote {svg_path}", flush=True)


# ---------------------------------------------------------------------------
# entry point
# ---------------------------------------------------------------------------

def main():
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    sub = ap.add_subparsers(dest="cmd", required=True)
    for name in ("download", "run", "report"):
        p = sub.add_parser(name)
        p.add_argument("--data", default=os.path.expanduser("~/data/ghcp-coding-agent-2026"))
        p.add_argument("--dates", nargs="+", default=sorted(ASSETS))
        p.add_argument("--max-sessions", type=int, default=0,
                       help="stop after this many sessions per date (0 = all)")
        p.add_argument("--tag", default="full", help="output name suffix")
        p.add_argument("--rate", type=float, default=2_000_000,
                       help="download cap in bytes per second (default 2 MB/s)")
    args = ap.parse_args()
    {"download": cmd_download, "run": cmd_run, "report": cmd_report}[args.cmd](args)


if __name__ == "__main__":
    main()