#!/usr/bin/env python3
"""Hands-on P15 — the integrated system: earlier blocks, IMPORTED and wired together.

Nothing here is reimplemented. Every mechanism is loaded from the hands-on file
that built it, which is the only honest test of whether they compose.
"""
import importlib.util, io, math, os, sys, time, contextlib
import numpy as np
from _harness import block, run_all

def load(mod, path):
    """Import a hands-on file and run its blocks silently to recover their state."""
    spec = importlib.util.spec_from_file_location(mod, path)
    m = importlib.util.module_from_spec(spec)
    import _harness
    saved = list(_harness._BLOCKS)
    _harness._BLOCKS.clear()
    real = _harness.run_all
    _harness.run_all = lambda *a, **k: None
    try:
        with contextlib.redirect_stdout(io.StringIO()):
            spec.loader.exec_module(m)
            st = {}
            for _, _, _, fn in _harness._BLOCKS:
                st.update(fn(st, False) or {})
    finally:
        _harness._BLOCKS.clear(); _harness._BLOCKS.extend(saved)
        _harness.run_all = real
    return m, st

@block(1, "Load the parts", "if they cannot be imported, they were never components")
def b1(s, show):
    here = os.path.dirname(os.path.abspath(__file__))
    parts = {}
    for name, f in (("ann", "h02_ann.py"), ("lsm", "h04_lsm.py"),
                    ("stream", "h07_streaming.py"), ("ab", "h10_abtest.py"),
                    ("hw", "h14_hardware.py")):
        t0 = time.perf_counter()
        parts[name] = load(name, os.path.join(here, f))
        if show:
            print(f"  {f:<22}-> {len(parts[name][1]):>2} exports "
                  f"({time.perf_counter()-t0:>5.2f}s)")
    if show:
        print("  Each file's blocks were re-run with output suppressed, so what we")
        print("  hold now are the actual functions those projects built -- not copies.")
        print("  This is the moment a curriculum of exercises becomes a system: the")
        print("  interfaces either line up or they do not, and no amount of prose")
        print("  about 'composability' substitutes for the import statement.")
    return {"parts": parts}

@block(2, "A service: ANN retrieval over an LSM store", "two projects, one request path")
def b2(s, show):
    ann = s["parts"]["ann"][1]
    lsm = s["parts"]["lsm"][1]
    n = ann["n"]
    # h04's Run is a sorted immutable segment with a Bloom filter. Build one
    # per 500 documents, exactly as its own make_db does, but keyed to OUR ids.
    runs = [lsm["Run"]([(f"doc{i}".encode(), f"payload for document {i}".encode())
                        for i in range(base, min(base+500, n))], bpk=10)
            for base in range(0, n, 500)]
    def query(qv, ef=32, k=5):
        ids, ndist = ann["greedy"](qv, ann["graph"], ann["entry"], ef)
        out = []
        for i in ids[:k]:
            v, st = lsm["get"](runs, f"doc{i}".encode())
            out.append((i, v, st))
        return out, ndist
    if show:
        hits, ndist = query(ann["q"][0])
        print(f"  index: {n} vectors of dim {ann['d']}, NSW graph with M={ann['M']}")
        print(f"  store: {len(runs)} LSM runs of 500 docs, 10-bit Bloom filters")
        print(f"  one query touched {ndist} vectors, returned {len(hits)} docs, "
              f"all found: {all(v is not None for _, v, _ in hits)}")
        for i, v, st in hits[:3]:
            print(f"    doc{i:<6} {v.decode()[:32]:<34} "
                  f"{st['block_reads']} block read, {st['skipped']} runs skipped")
        tot = sum(st["skipped"] for _, _, st in hits)
        print(f"  Bloom filters skipped {tot} of {len(runs)*len(hits)} possible run")
        print(f"  probes -- {tot/(len(runs)*len(hits)):.0%} of the store never touched.")
        print("  The vector index answers WHICH documents; the LSM answers WHAT they")
        print("  contain. Neither project knew the other existed. The first attempt")
        print("  at this seam failed on a KeyError -- I assumed the ANN module")
        print("  exported build_nsw/search_nsw and it exports greedy/graph/entry.")
        print("  That is the normal cost of integration and the reason this block")
        print("  exists: an interface you have never called is a guess.")
    return {"query": query, "runs": runs}

@block(3, "Measure the request path", "a latency budget, decomposed")
def b3(s, show):
    ann = s["parts"]["ann"][1]; lsm = s["parts"]["lsm"][1]; runs = s["runs"]
    qs = list(ann["q"])[:120]
    if show:
        def pct(fn, xs):
            ts = []
            for x in xs:
                t0 = time.perf_counter(); fn(x); ts.append((time.perf_counter()-t0)*1e6)
            ts.sort(); return ts[len(ts)//2], ts[int(.99*(len(ts)-1))]
        ids = ann["greedy"](qs[0], ann["graph"], ann["entry"], 32)[0][:5]
        stages = [("ANN search (ef=32)",
                   pct(lambda q: ann["greedy"](q, ann["graph"], ann["entry"], 32), qs)),
                  ("LSM fetch x5",
                   pct(lambda _: [lsm["get"](runs, f"doc{i}".encode()) for i in ids], qs)),
                  ("end to end", pct(lambda q: s["query"](q), qs))]
        print(f"  {'stage':<26}{'p50':>10}{'p99':>10}{'share of p50':>15}")
        total = stages[-1][1][0]
        for lbl, (p50, p99) in stages:
            print(f"  {lbl:<26}{p50:>8.0f}us{p99:>8.0f}us{p50/total:>14.0%}")
        print(f"  p99/p50 end to end: {stages[-1][1][1]/total:.2f}x")
        print("  Search dominates and the store is nearly free -- which is only true")
        print("  because the Bloom filters made most runs untouchable. Turn them off")
        print("  and the fetch stage grows by the number of runs, which is the")
        print("  experiment P04 already ran. Components carry their measurements with")
        print("  them; that is what makes a budget like this cheap to build.")
        print("  Note the tail: a request is slow if EITHER stage is slow, so the")
        print("  composed p99/p50 is worse than either component's own. That is the")
        print("  tail-at-scale arithmetic of proofs.md P14 appearing in a two-stage")
        print("  pipeline on a single machine -- it does not need a cluster to bite.")
    return {}

@block(4, "Capacity planning before deployment", "P14's roofline applied to P02's index")
def b4(s, show):
    hw = s["parts"]["hw"][1]; ann = s["parts"]["ann"][1]
    if show:
        d = ann["d"]
        print(f"  measured: {hw['bw']/1e9:.1f} GB/s, fp64 peak {hw['flops']/1e9:.0f} "
              f"GFLOP/s, ridge {hw['ridge']:.1f} FLOP/byte")
        print(f"  {'strategy':<20}{'vectors read':>14}{'bytes':>10}{'intensity':>11}"
              f"{'bound':>9}{'ceiling QPS':>13}")
        rows = [("brute force", ann["n"])]
        for ef in (8, 32, 64):
            nd = ann["greedy"](ann["q"][0], ann["graph"], ann["entry"], ef)[1]
            rows.append((f"NSW ef={ef}", nd))
        for lbl, nv in rows:
            byts = nv * d * 8; flops = 2 * nv * d
            ai = flops / byts
            qps = min(hw["flops"], ai * hw["bw"]) / flops
            print(f"  {lbl:<20}{nv:>14,}{byts/1e3:>8.1f}K{ai:>11.2f}"
                  f"{'MEMORY':>9}{qps:>13,.0f}")
        print(f"  Intensity is {2/16:.2f} FLOP/byte for every row and cannot be")
        print("  otherwise: a dot product does two flops per eight-byte coordinate.")
        print("  So the ONLY lever on throughput is touching fewer vectors -- which is")
        print("  precisely what the index does, and precisely why quantisation (fewer")
        print("  BYTES per vector) is the other half of every production ANN system.")
        print("  These ceilings are far above the measured p50 in block 3, and the")
        print("  gap is not hardware: it is Python walking a graph one node at a")
        print("  time. The roofline prices data movement, so the difference between")
        print("  it and reality is exactly the implementation's overhead -- which")
        print("  makes it a budget for a rewrite, not a criticism of the design.")
    return {}

@block(5, "Ship it behind an experiment", "P10 decides whether the change was real")
def b5(s, show):
    ab = s["parts"]["ab"][1]
    rng = np.random.default_rng(17)
    if show:
        print("  Proposed change: raise ef from 8 to 32 -- better recall, slower.")
        n = 40_000
        arm = np.array([ab["assign"](f"u{i}", "ef-32-rollout") for i in range(n)])
        chi, p = ab["srm"]([int((arm == 0).sum()), int((arm == 1).sum())])
        print(f"  assignment {int((arm==0).sum()):,}/{int((arm==1).sum()):,}   "
              f"SRM chi2={chi:.2f} p={p:.3f} -> {'PASS' if p > 0.001 else 'FAIL'}")
        need = ab["n_per_arm"](0.05, 0.03)
        ok = (arm == 0).sum() >= need
        print(f"  to detect a 3% relative lift at 80% power: {need:,} per arm; "
              f"we have {int((arm==0).sum()):,} -> {'POWERED' if ok else 'UNDERPOWERED'}")
        conv = rng.binomial(1, np.where(arm == 1, 0.05*1.06, 0.05)).astype(float)
        a, b = conv[arm == 0], conv[arm == 1]
        t, pv = ab["welch"](a, b)
        print(f"  conversion A={a.mean():.4f} B={b.mean():.4f} "
              f"lift={(b.mean()-a.mean())/a.mean():+.2%} p={pv:.4f} -> "
              f"{'SHIP' if pv < 0.05 and b.mean() > a.mean() else 'DO NOT SHIP'}")
        print("  Recall improved in an offline benchmark; that is not a reason to")
        print("  ship. The A/B platform converts an engineering improvement into a")
        print("  claim about users, and the SRM and power checks run BEFORE anyone is")
        print("  allowed to read the conversion number.")
    return {}

@block(6, "Watch it in production", "P07 turns the request log into a live metric")
def b6(s, show):
    st = s["parts"]["stream"][1]
    rng = np.random.default_rng(18)
    if show:
        events = []
        for _ in range(6000):
            et = rng.uniform(0, 300_000)
            delay = (rng.exponential(1500) if rng.random() > .06
                     else rng.uniform(0, 60_000))
            events.append((et, et + delay, "err" if rng.random() < .02 else "ok"))
        events.sort(key=lambda e: e[1])
        W = st["W"]
        truth = {}
        for et, _, k in events:
            truth[(int(et // W), k)] = truth.get((int(et // W), k), 0) + 1
        terr = sum(v for (_, k), v in truth.items() if k == "err")
        ttot = sum(truth.values())
        print(f"  {len(events)} request events, {W//1000}s tumbling windows, "
              f"6% arrive late")
        print(f"  true error rate = {terr/ttot:.4%}")
        print(f"  {'policy':<24}{'emit delay':>12}{'measured rate':>15}"
              f"{'error':>10}{'events lost':>13}")
        for lbl, lag, grace in (("live dashboard", 2_000, 0),
                                ("alerting", 10_000, 30_000),
                                ("weekly report", 10_000, 90_000)):
            got, dropped, _ = st["run"](events, st["fixed"](lag), W, grace)
            errs = sum(v for (_, k), v in got.items() if k == "err")
            tot = max(sum(got.values()), 1)
            print(f"  {lbl:<24}{(lag+grace)/1000:>10.0f}s{errs/tot:>15.4%}"
                  f"{errs/tot - terr/ttot:>+10.4%}{ttot-tot:>13,}")
        print("  The dashboard drops a couple of hundred events at a 2-second emit")
        print("  delay and still lands within 0.04pp of the true rate -- because")
        print("  lateness here is very nearly independent of whether a request")
        print("  failed, so it loses from numerator and denominator alike. That is a")
        print("  property of THIS workload, not a general licence: correlate the")
        print("  delay with the metric (slow requests are the failing ones) and the")
        print("  same pipeline becomes systematically optimistic. Measure the bias")
        print("  before trusting a fast estimate, then set the threshold against it.")
    return {}

def assembly(s):
    print("\nSix blocks, five imported projects, one system. The whole path:\n")
    ann = s["parts"]["ann"][1]; hw = s["parts"]["hw"][1]
    d = ann["d"]
    qs = list(ann["q"]); truth = list(ann["truth"])
    print(f"  {'ef':>5}{'recall@10':>12}{'vectors read':>14}{'p50':>10}"
          f"{'QPS':>9}{'ceiling QPS':>13}{'efficiency':>12}")
    for ef in (8, 16, 32, 64):
        recs, ts, nds = [], [], []
        for q, tr in zip(qs, truth):
            t0 = time.perf_counter()
            got, nd = ann["greedy"](q, ann["graph"], ann["entry"], ef)
            ts.append(time.perf_counter() - t0)
            recs.append(ann["recall_at_k"](got, tr, 10)); nds.append(nd)
        ts.sort(); p50 = ts[len(ts)//2]; nd = sum(nds)/len(nds)
        flops = 2 * nd * d
        ceil = min(hw["flops"], (flops/(nd*d*8)) * hw["bw"]) / flops
        print(f"  {ef:>5}{sum(recs)/len(recs):>12.4f}{nd:>14,.0f}{p50*1e6:>8.0f}us"
              f"{1/p50:>9,.0f}{ceil:>13,.0f}{1/p50/ceil:>11.3%}")
    print("\n  Read that as a product decision, because that is what it is. Each row")
    print("  is a different service: ef=8 is fast and wrong, ef=64 is accurate and")
    print("  slow, and recall is bought at a steeply rising price in vectors read.")
    print("  The efficiency column says every row leaves over 99% of the hardware")
    print("  unused -- this is Python chasing pointers through a graph. That is not")
    print("  a criticism of the index; it is a measured budget for a rewrite, and it")
    print("  says a C implementation has roughly three orders of magnitude of")
    print("  headroom before the memory system becomes the constraint.")
    print("\n  And then the honest part. Choosing a row from this table is an offline")
    print("  decision made on synthetic queries with a synthetic notion of relevance.")
    print("  Block 5 is what makes it real: hash assignment, SRM, power, and a")
    print("  conversion metric that decides. Block 6 is what keeps it real: a")
    print("  windowed error rate with a stated emit delay and a measured bias.")
    print("  Offline recall is a hypothesis, the experiment is the test, the stream")
    print("  is the monitor. A system is not the sum of its components; it is those")
    print("  three loops closed around them.")
    print("\n  What this file proves and the other fourteen cannot: the parts IMPORT.")
    print("  Every function used here was written for a different project with no")
    print("  knowledge of this one. The integration cost was a few lines per seam")
    print("  and one real failure -- a KeyError from assuming an API that did not")
    print("  exist. Days of component work, an hour of integration, one genuine")
    print("  interface bug. That ratio is the thing to expect and to budget for.")
    print("\n  Built from: P02 (NSW index, greedy search, recall@k), P04 (LSM runs,")
    print("  Bloom-filtered get), P07 (windowing, watermarks, late data), P10 (hash")
    print("  assignment, SRM, power, Welch), P14 (measured roofline). Missing, on the")
    print("  project page: a serving layer with real concurrency (m3-m4), the")
    print("  distributed control plane from P05 (m6), P06-style batch reindexing")
    print("  (m8), and the capstone report that puts one end-to-end number against a")
    print("  stated SLO.")

if __name__ == "__main__":
    run_all(assembly, "HANDS-ON P15 — The integrated system, block by block")