#!/usr/bin/env python3
"""Harvest hook-trace.log into a durable archive, then report the UPS busy distribution.

Why this exists: `hook-trace.log` rolls one generation at 512 KB. Post-v0.39.0 the node
writes it at ~35 B/s, so a generation lasts hours — never the multi-day window the busy-race
measurement (#32, BUSY-RACE-PLAN.md step 3) has to count over. Running this on every session
carries the evidence forward across rolls; the archive is append-only and de-duplicated on the
raw line, so re-running it any number of times is safe.

The trace is UTF-8-lossy: it is read as bytes and decoded with errors="replace", never grepped.

    python ci/measure/trace-harvest.py            # harvest + report
    python ci/measure/trace-harvest.py --report   # report the archive only, harvest nothing
"""

import json
import os
import re
import statistics
import subprocess
import sys
import time
from collections import Counter
from pathlib import Path

SOURCE_DIR = Path(os.environ["LOCALAPPDATA"]) / "spt-core/adapters/_github/SaberMage-claude-spt"
EVIDENCE = Path(os.environ["LOCALAPPDATA"]) / "spt-claude-code/evidence"
ARCHIVE = EVIDENCE / "hook-trace-archive.log"
INBOUND = EVIDENCE / "inbound-arrivals.tsv"

LINE = re.compile(r"\[(\d+) pid (\d+)\] claude-spt hook: (.*)")


def read_lossy(path):
    return path.read_bytes().decode("utf-8", "replace").split("\n")


def harvest():
    """Append every unseen line of every live generation to the archive. Returns (new, total)."""
    ARCHIVE.parent.mkdir(parents=True, exist_ok=True)
    seen = set(read_lossy(ARCHIVE)) if ARCHIVE.exists() else set()
    fresh = []
    # Oldest generation first so the archive stays roughly time-ordered.
    for name in ("hook-trace.log.1", "hook-trace.log"):
        p = SOURCE_DIR / name
        if not p.exists():
            continue
        for line in read_lossy(p):
            if line.strip() and line not in seen:
                seen.add(line)
                fresh.append(line)
    if fresh:
        # newline="\n" is load-bearing: the default on Windows writes \r\n, the source lines end
        # \n, and the de-dup set then matches NOTHING — every run re-appends the whole log and
        # silently doubles every count the report makes.
        with ARCHIVE.open("a", encoding="utf-8", newline="\n") as f:
            f.write("\n".join(fresh) + "\n")
    return len(fresh), len(seen)


def harvest_inbound(endpoint):
    """Bank this endpoint's MSG_IN arrival times from the io funnel (step 2's other half).

    `--after` carries our own cursor and writes no session cursor, so paging the funnel disturbs
    nothing. Only the arrival stamp and the peer are kept — never the payload, which is the
    endpoint occupant's message text and has no business in an evidence file.
    """
    EVIDENCE.mkdir(parents=True, exist_ok=True)
    seen = set()
    if INBOUND.exists():
        seen = {l.split("\t")[0] for l in INBOUND.read_text(encoding="utf-8").split("\n") if l}
    rows, after = [], 0
    while True:
        out = subprocess.run(
            ["spt", "api", "io-events", endpoint, "--after", str(after), "--limit", "200", "--json"],
            capture_output=True, text=True, encoding="utf-8", errors="replace",
        )
        if out.returncode != 0:
            print(f"funnel poll failed at seq {after}: {out.stderr.strip()[:200]}")
            break
        answer = json.loads(out.stdout)
        for e in answer["events"]:
            if e["kind"] == "MSG_IN" and str(e["at_ms"]) not in seen:
                seen.add(str(e["at_ms"]))
                rows.append(f"{e['at_ms']}\t{endpoint}\t{e.get('peer', '?')}")
        if not answer["more"]:
            break
        after = answer["cursor"]
    if rows:
        with INBOUND.open("a", encoding="utf-8", newline="\n") as f:
            f.write("\n".join(rows) + "\n")
    return len(rows)


def report_windows(endpoint):
    """Step 2: did an inbound land inside the gap between Enter and the busy mark?

    A window nothing ever lands in is a window, not a defect — so this reports the count, and
    reports it as insufficient rather than as zero when there is not yet enough of either input.
    """
    if not INBOUND.exists():
        return
    arrivals = []
    for line in INBOUND.read_text(encoding="utf-8").split("\n"):
        if line.strip():
            at, ep, peer = line.split("\t")
            if ep == endpoint:
                arrivals.append((int(at), peer))
    windows = []
    for line in read_lossy(ARCHIVE) if ARCHIVE.exists() else []:
        m = LINE.match(line)
        if not m or f"TRACE UserPromptSubmit id={endpoint} " not in m.group(3):
            continue
        b = re.search(r"busy=(\d+)ms", m.group(3))
        t = re.search(r"total=(\d+)ms", m.group(3))
        if not (b and t):
            continue
        # The trace stamp is written when the hook FINISHES, so the window opens total ms back
        # and closes once the busy stage has run.
        end = int(m.group(1))
        start = end - int(t.group(1))
        windows.append((start, start + int(b.group(1))))
    hits = [(a, p, w) for a, p in arrivals for w in windows if w[0] <= a <= w[1]]
    print(f"step 2 — {endpoint}: {len(arrivals)} inbound, {len(windows)} busy windows, "
          f"{len(hits)} arrival(s) inside a window")
    for a, p, w in hits[:10]:
        print(f"  {time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(a / 1000))} from {p} "
              f"({a - w[0]}ms into a {w[1] - w[0]}ms window)")


def report_deadline_kills(grace_ms=60_000):
    """How many UserPromptSubmit hooks were killed at Claude Code's deadline.

    A killed hook reaches neither `finish` nor a `Drop`, so it writes no TRACE line — which is why
    every distribution below covers COMPLETED hooks only and reads the timeout rate as near-zero.
    The BEGIN line (REQ-HOOK-DEADLINE-VISIBLE) closes that: each hook invocation is its own
    process, so a BEGIN whose pid never produced a TRACE IS a kill.

    Pids are reused, so pairing walks chronologically and treats a second BEGIN on a still-pending
    pid as evidence the first one died. BEGINs inside `grace_ms` of the archive's end are dropped —
    those hooks may simply still be running, and counting an in-flight hook as a kill would
    manufacture the number this exists to measure.

    TWO refusals to print a zero, because a zero here is indistinguishable from a measurement:
    NOT INSTRUMENTED when the archive holds no BEGIN lines at all (cannot tell "no kills" from "no
    instrument"), and INSUFFICIENT when BEGINs exist but none are older than the grace window
    (a 0/0 denominator is not a rate of zero — it reads as "measured, no kills" and is not).
    [impl->REQ-HOOK-DEADLINE-VISIBLE]
    """
    begins, traces = [], []
    for line in read_lossy(ARCHIVE) if ARCHIVE.exists() else []:
        m = LINE.match(line)
        if not m:
            continue
        at, pid, body = int(m.group(1)), m.group(2), m.group(3)
        if body.startswith("BEGIN UserPromptSubmit"):
            begins.append((at, pid, body))
        elif body.startswith("TRACE UserPromptSubmit"):
            traces.append((at, pid))
    if not begins:
        print("deadline kills: NOT INSTRUMENTED — no BEGIN lines archived yet (an archive without "
              "them cannot tell 'no kills' from 'no instrument'; needs a build carrying the BEGIN "
              "marker to have run)")
        return
    horizon = max([b[0] for b in begins] + [t[0] for t in traces]) - grace_ms
    events = sorted([(a, p, "B", b) for a, p, b in begins] + [(a, p, "T", "") for a, p in traces])
    pending, killed = {}, []
    for at, pid, kind, body in events:
        if kind == "B":
            if pid in pending:
                killed.append(pending[pid])
            pending[pid] = (at, pid, body)
        elif pid in pending:
            del pending[pid]
    killed += [v for v in pending.values() if v[0] <= horizon]
    started = len(begins) - sum(1 for a, _, _ in begins if a > horizon)
    stamp = lambda ms: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ms / 1000))
    # A denominator of zero is NOT a rate of zero. Every BEGIN can be inside the grace window on
    # the first harvest after the marker reaches the field, and printing "0/0 = 0.00%" there reads
    # exactly like "measured, no kills" — the same confusion the NOT INSTRUMENTED branch above
    # exists to prevent, one case narrower. Guard the DENOMINATOR, not just the presence of lines.
    if started == 0:
        print(f"deadline kills: INSUFFICIENT — {len(begins)} BEGIN line(s) archived, all inside the "
              f"{grace_ms // 1000}s grace window (they may still be running). No rate is claimed.")
        return
    rate = 100 * len(killed) / started
    print(f"deadline kills: {len(killed)}/{started} started = {rate:.2f}% (BEGIN with no TRACE)")
    for at, pid, body in sorted(killed)[-10:]:
        who = re.search(r"id=(\S+)", body)
        print(f"  {stamp(at)} pid {pid} id={who.group(1) if who else '?'}")


def report():
    """The two numbers BUSY-RACE-PLAN.md asks for: the busy window, and candidate (b)'s count."""
    rows = []
    for line in read_lossy(ARCHIVE) if ARCHIVE.exists() else []:
        m = LINE.match(line)
        if m and m.group(3).startswith("TRACE "):
            rows.append((int(m.group(1)), m.group(3)))
    ups = [r for r in rows if r[1].startswith("TRACE UserPromptSubmit")]
    if not ups:
        print("no UserPromptSubmit TRACE lines archived yet")
        return
    span_h = (ups[-1][0] - ups[0][0]) / 3_600_000
    stamp = lambda ms: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ms / 1000))
    print(f"archive: {ARCHIVE}")
    print(f"UPS traces: {len(ups)} over {span_h:.2f}h ({stamp(ups[0][0])} .. {stamp(ups[-1][0])})")

    # Candidate (b): an identity that resolves empty marks nothing, so the line carries no busy=.
    nobusy = [r for r in ups if "busy=" not in r[1]]
    print(f"candidate (b) — UPS with NO busy= stage: {len(nobusy)}")
    for r in nobusy[:10]:
        print(f"  {stamp(r[0])} {r[1][:150]}")

    # Candidate (a): the width of the window between Enter and the busy mark landing.
    busy = [int(m.group(1)) for m in (re.search(r"busy=(\d+)ms", r[1]) for r in ups) if m]
    ordered = sorted(busy)
    pct = lambda q: ordered[min(len(ordered) - 1, int(len(ordered) * q))]
    print(
        f"candidate (a) — busy ms: n={len(busy)} min={ordered[0]} p50={pct(0.5)} "
        f"p90={pct(0.9)} max={ordered[-1]} mean={statistics.mean(busy):.0f}"
    )
    # The deadline number: total wall time against CC's UserPromptSubmit budget. SURVIVORS ONLY —
    # a hook killed at the deadline never wrote a TRACE, so this tail is biased low by construction
    # and `deadline kills` below is the honest count. [impl->REQ-HOOK-DEADLINE-VISIBLE]
    tot = sorted(int(m.group(1)) for m in (re.search(r"total=(\d+)ms", r[1]) for r in ups) if m)
    tpct = lambda q: tot[min(len(tot) - 1, int(len(tot) * q))]
    print(
        f"total ms (COMPLETED hooks only): n={len(tot)} p50={tpct(0.5)} p90={tpct(0.9)} "
        f"p99={tpct(0.99)} max={tot[-1]}"
    )
    for thr in (5000, 10000, 20000):
        print(f"  over {thr}ms: {sum(1 for t in tot if t > thr)}/{len(tot)}")

    ids = Counter(m.group(1) for m in (re.search(r"id=(\S+)", r[1]) for r in ups) if m)
    print(f"endpoints: {ids.most_common()}")


if __name__ == "__main__":
    endpoint = os.environ.get("SPT_ENDPOINT_ID", "perri")
    if "--report" not in sys.argv:
        new, total = harvest()
        print(f"harvested {new} new line(s); archive holds {total}")
        print(f"banked {harvest_inbound(endpoint)} new inbound arrival(s) for {endpoint}")
    report()
    report_deadline_kills()
    report_windows(endpoint)