#!/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.

    Reports NOT INSTRUMENTED rather than zero when the archive predates the BEGIN line: an archive
    with no BEGINs cannot distinguish "no kills" from "no instrument", and the whole point of this
    requirement is that those two look identical. [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)
    rate = (100 * len(killed) / started) if started else 0.0
    stamp = lambda ms: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ms / 1000))
    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)