Measure API-wait stalls across a whole Claude Code transcript corpus (stdlib only, read-only)

api_wait_gaps.py

#!/usr/bin/env python3
"""api_wait_gaps.py - measure API-wait stalls across a whole Claude Code transcript corpus.

A "stall" here = wall-clock time in which the RUNTIME owed the next event and no bytes
arrived. Only two record pairs qualify:

    tool_result -> assistant     (tool came back, model has not spoken yet)
    assistant   -> assistant     (mid-stream continuation)

Why the restriction matters: a naive "every gap >= 60s" scan over a corpus is dominated by
session RESUMES, not stalls. On the box this was written for, the naive version reported a
top "stall" of 32 days - it was a session paused overnight and resumed. Any number produced
without this filter is measuring human idle time, not the API.

Usage:  python api_wait_gaps.py [days]        (default 7)
Reads:  ~/.claude/projects/*/*.jsonl          (read-only, nothing is written)
Stdlib only, no network.
"""
import collections
import datetime
import glob
import json
import os
import sys

DAYS = int(sys.argv[1]) if len(sys.argv) > 1 else 7
BACKSTOP_MS = int(os.environ.get("API_TIMEOUT_MS", "900000"))
BACKSTOP_S = BACKSTOP_MS / 1000.0
ROOT = os.path.expanduser("~/.claude/projects")


def parse_ts(rec):
    t = rec.get("timestamp")
    if not t:
        return None
    try:
        return datetime.datetime.fromisoformat(t.replace("Z", "+00:00"))
    except ValueError:
        return None


def kind(rec):
    ty = rec.get("type")
    if ty == "assistant":
        return "assistant"
    if ty == "user":
        content = rec.get("message", {}).get("content")
        if isinstance(content, list) and any(
            isinstance(b, dict) and b.get("type") == "tool_result" for b in content
        ):
            return "tool_result"
        return "human"
    return ty or "other"


def main():
    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=DAYS)
    gaps, scanned, timed_out = [], 0, 0

    for path in glob.glob(os.path.join(ROOT, "*", "*.jsonl")):
        try:
            mtime = datetime.datetime.fromtimestamp(
                os.path.getmtime(path), datetime.timezone.utc
            )
        except OSError:
            continue
        if mtime < cutoff:
            continue
        scanned += 1

        seq = []
        try:
            with open(path, encoding="utf-8", errors="replace") as fh:
                for line in fh:
                    line = line.strip()
                    if not line:
                        continue
                    if "Request timed out" in line:
                        timed_out += 1
                    try:
                        rec = json.loads(line)
                    except ValueError:
                        continue
                    ts = parse_ts(rec)
                    if ts is not None:
                        seq.append((ts, kind(rec)))
        except OSError:
            continue

        for (t0, k0), (t1, k1) in zip(seq, seq[1:]):
            delta = (t1 - t0).total_seconds()
            if delta < 60:
                continue
            if (k0 == "tool_result" and k1 == "assistant") or (k0 == k1 == "assistant"):
                gaps.append((delta, k0, k1, os.path.basename(path), t0.isoformat()))

    gaps.sort(reverse=True)
    lo, hi = BACKSTOP_S - 40, BACKSTOP_S + 40

    print(f"transcripts touched in last {DAYS}d : {scanned}")
    print(f"records with 'Request timed out'  : {timed_out}")
    print(f"API-wait gaps >= 60s              : {len(gaps)}")

    buckets = collections.Counter()
    for delta, *_ in gaps:
        if delta < 120:
            buckets["  60-120s"] += 1
        elif delta < 300:
            buckets[" 120-300s"] += 1
        elif delta < 600:
            buckets[" 300-600s"] += 1
        elif delta < lo:
            buckets[f" 600-{int(lo)}s"] += 1
        elif delta <= hi:
            buckets[f" {int(lo)}-{int(hi)}s  <-- backstop band"] += 1
        elif delta <= 3600:
            buckets[f" {int(hi)}-3600s"] += 1
        else:
            buckets["    >1h  (resume, not a stall)"] += 1
    for label in sorted(buckets):
        print(f"  {label}: {buckets[label]}")

    band = [g for g in gaps if lo <= g[0] <= hi]
    print(f"\nbackstop-band hits ({int(lo)}-{int(hi)}s): {len(band)}")
    for delta, k0, k1, fname, ts in band[:20]:
        print(f"  {delta:7.1f}s  {k0} -> {k1}  {ts}  {fname[:18]}")

    print("\nlongest real stalls (<= 1h):")
    for delta, k0, k1, fname, ts in [g for g in gaps if g[0] <= 3600][:10]:
        print(f"  {delta:7.1f}s  {k0} -> {k1}  {ts}  {fname[:18]}")


if __name__ == "__main__":
    main()