D2 + D3: measured autotune loop + arbiter fallback log (sm_120)

docs/audit/compiler/COMPILER_REFACTOR_PLAN.md

@@ -80,7 +80,7 @@ table is the single skim surface. `✅` done · `🟡` partial · `⬜` not star
 | **5** | C3 tail — drive WMMA/MFMA `Generate*` passes through the loop | ✅ WMMA host-free | ✅ gfx1151 fused WMMA F4-gated (MFMA=CDNA-gated) | — |
 | **3.5** | ROCm shipped-kernel → F4 gate (flash-attn, f16 budget) + shared scalar body | ✅ | ✅ gfx1151 attn | — |
 | **6** | D1 candidate registry + F4-gate + tier-priority arbiter + `force` (E3) | ✅ (`emit/candidate.py`) | ✅ gfx1151 enumerate+select | — |
-| **6** | D2 measured autotune loop · D3 fallback log | ⬜ | ⬜ | ⬜ |
+| **6** | D2 measured autotune loop · D3 fallback log | ✅ (`emit/autotune.py` + arbiter log) | ⬜ | ✅ sm_120 (matmul measure+cache) |
 
 **Gate reality (softens §4/§9.2):** "Phase 0 gates everything" holds only for the
 *lead-execution* proofs. The Mac-side E1 gate is green and gfx1151 E2 is recorded,
@@ -376,13 +376,26 @@ chains, small attention). Crown-jewel GEMM stays Tier 2/3.
   (generic C / AOCL-DLP). **Still open:** the shape-bucket key on selection (today
   keyed per `(target, op)`; `bucket_key` exists and threads in when D2 lands) and
   generalizing Apple's `select_variant` + `best_record` into it.
-- **D2 · Measured autotune loop** — `[AMD]` on gfx1151, `[NV]` on sm_120 run
-  live; CDNA/sm_90/sm_100 fall back to analytical roofline + `MmaDescriptor` cost
-  model until silicon. Measure-at-first-miss + cache keyed by
-  `device+shape-bucket+accuracy-margin`.
-- **D3 · Fallback log everywhere** `[MAC]` — generalize
-  `dispatch_fallback_log`/`fallback_histogram` so "did the compiled path win or
-  silently degrade?" is answerable per backend.
+- **D2 · Measured autotune loop** — **core landed 2026-07-07** (`emit/autotune.py`).
+  `measured_arbitrate()` layers on the D1 arbiter's `measure` seam: it F4-gates the
+  candidates, times each survivor on-device (`measure_latency`, median of N after
+  warmup), and caches the fastest in a `MeasureCache` keyed by `(device, target, op,
+  shape-bucket, dtype)` — **measure-at-first-miss** (a re-query hits the cache, no
+  re-timing). Lead-safety holds: only in-budget F4-passing candidates are timed, so a
+  faster-but-wrong kernel can't win. **Live on sm_120** (RTX 5070 Ti): times the
+  shipped vs emitted GEMM lanes and caches the winner per bucket
+  (`test_nvidia_plugin.py`); `_nvidia_device_name()` supplies the `sm_<cc>` device
+  tag. **Still open:** persisting the cache as the committed *fleet-shared autotune
+  corpus* (Theory §7.5 — hangs off `MeasureCache.to_dict`); `[AMD]` gfx1151 wiring;
+  CDNA/sm_90/sm_100 analytical-roofline + `MmaDescriptor` fallback until silicon.
+- **D3 · Fallback log everywhere** — **landed 2026-07-07.** The arbiter records every
+  dispatch as `(target, op, selected, tag)` (`candidate._note_arbiter_dispatch`, wired
+  into `run_arbitrated` + `run_measured_arbitrated`); `arbiter_dispatch_histogram()`
+  answers **"did the compiled path win, silently degrade, or was there no candidate?"**
+  per `(target, op)` — a selection that ran but returned a reference tag is the silent
+  degrade (the arbiter-layer analog of `runtime.dispatch_fallback_log`). Proven both
+  host-free and live (the emitted lane forced on a ragged shape it can't run logs a
+  `degraded`).
 
 ### Workstream E — Regression guardrails (continuous, not last)
 

python/tessera/compiler/emit/autotune.py

@@ -0,0 +1,170 @@
+"""Workstream D2 — measured autotune loop for the D1 arbiter.
+
+D1 (:mod:`emit.candidate`) selects by **tier priority** (crown-jewel first —
+lead-safe by construction). D2 replaces that with **real on-device latency**: for
+a given ``(device, target, op, shape-bucket, dtype)`` it times each F4-passing
+candidate once, caches the fastest (**measure-at-first-miss**), and reuses that
+verdict thereafter. Lead-safety is preserved end-to-end — only candidates that
+already pass the universal F4 oracle *within their accuracy budget* are timed, so
+a faster-but-wrong (or out-of-budget) kernel can never win.
+
+This layers on the arbiter's existing ``measure`` seam
+(:func:`emit.candidate.arbitrate` picks ``min(cands, key=measure)``): D2 supplies
+the latency callback + the cache. The cache is process-local here; persisting it
+as the committed *fleet-shared autotune corpus* (Theory §7.5 — a config proven on
+one box warm-starts the others) is the follow-on that hangs off :meth:`MeasureCache.to_dict`.
+"""
+from __future__ import annotations
+
+import statistics
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+from tessera.compiler.emit.candidate import (
+    Candidate,
+    _note_arbiter_dispatch,
+    arbitrate,
+    candidates_for,
+)
+from tessera.compiler.emit.kernel_emitter import SpecPolicy, bucket_key
+
+
+@dataclass(frozen=True)
+class MeasureRecord:
+    """The measured verdict for one ``(device, target, op, bucket, dtype)`` key:
+    the fastest candidate, its median latency (ms), and every timed candidate's
+    latency (for a fallback log / the fleet corpus)."""
+
+    winner: str
+    latency_ms: float
+    candidates: dict[str, float] = field(default_factory=dict)
+
+
+class MeasureCache:
+    """Content-keyed cache of :class:`MeasureRecord` — measure-at-first-miss. Key =
+    ``(device, target, op, shape-bucket, dtype)`` so nearby shapes share a verdict
+    (the bucket) while distinct devices/dtypes stay separate."""
+
+    def __init__(self) -> None:
+        self._store: dict[tuple[Any, ...], MeasureRecord] = {}
+        self.hits = 0
+        self.misses = 0
+
+    def get(self, key: tuple[Any, ...]) -> MeasureRecord | None:
+        rec = self._store.get(key)
+        if rec is not None:
+            self.hits += 1
+        else:
+            self.misses += 1
+        return rec
+
+    def put(self, key: tuple[Any, ...], rec: MeasureRecord) -> None:
+        self._store[key] = rec
+
+    def clear(self) -> None:
+        self._store.clear()
+        self.hits = 0
+        self.misses = 0
+
+    @property
+    def size(self) -> int:
+        return len(self._store)
+
+    def to_dict(self) -> dict[str, MeasureRecord]:
+        """A JSON-friendly view (string keys) — the seam the fleet-shared corpus
+        persists. Follow-on; not wired to disk here."""
+        return {repr(k): v for k, v in self._store.items()}
+
+
+#: Process-wide default cache (the arbiter/runtime share one).
+_DEFAULT_CACHE = MeasureCache()
+
+
+def default_cache() -> MeasureCache:
+    return _DEFAULT_CACHE
+
+
+def measure_latency(run_fn: Any, *, reps: int = 20, warmup: int = 3) -> float:
+    """Median wall-clock latency (ms) of ``run_fn`` over ``reps`` calls after
+    ``warmup`` untimed calls. ``run_fn`` runs the candidate end-to-end (H2D /
+    launch / D2H) so the comparison reflects what a caller actually pays."""
+    for _ in range(warmup):
+        run_fn()
+    samples = []
+    for _ in range(reps):
+        t0 = time.perf_counter()
+        run_fn()
+        samples.append((time.perf_counter() - t0) * 1e3)
+    return statistics.median(samples)
+
+
+def _device_id(target: str) -> str:
+    """A stable per-device tag for the cache key. Probes the live device name where
+    cheap (NVIDIA), else falls back to the target id — so a config measured on one
+    device is never reused on another."""
+    if target == "nvidia":
+        try:
+            from tessera import runtime as rt
+            name = rt._nvidia_device_name()
+            if name:
+                return f"nvidia:{name}"
+        except Exception:
+            pass
+    return target
+
+
+def measured_arbitrate(region: Any, op: str, target: str, *inputs: Any,
+                       dims: tuple[int, ...] | None = None, dtype: str = "f32",
+                       cache: MeasureCache | None = None, reps: int = 20,
+                       warmup: int = 3, device: str | None = None) -> Candidate | None:
+    """Pick the winning candidate by **measured latency** (measure-at-first-miss),
+    or ``None`` if none apply/verify (caller uses the reference).
+
+    On a cache hit for ``(device, target, op, bucket(dims), dtype)`` the recorded
+    winner is returned if it is still applicable/available (no re-timing). On a
+    miss, the arbiter F4-gates the candidates and times the survivors on ``inputs``
+    (median of ``reps`` after ``warmup``); the fastest is cached and returned."""
+    cache = cache if cache is not None else _DEFAULT_CACHE
+    dev = device or _device_id(target)
+    bucket = bucket_key(dims, SpecPolicy.BUCKET) if dims is not None else None
+    key = (dev, target, op, bucket, dtype)
+
+    rec = cache.get(key)
+    if rec is not None:
+        for c in candidates_for(target, op):
+            if c.name == rec.winner and c.applies_to(region) and c.available():
+                return c
+        # cached winner is gone/unavailable — fall through and re-measure.
+
+    latencies: dict[str, float] = {}
+
+    def _measure(cand: Candidate) -> float:
+        t = measure_latency(lambda: cand.run(region, *inputs), reps=reps, warmup=warmup)
+        latencies[cand.name] = t
+        return t
+
+    winner = arbitrate(region, op, target, verify=True, measure=_measure)
+    if winner is not None:
+        cache.put(key, MeasureRecord(
+            winner=winner.name,
+            latency_ms=latencies.get(winner.name, float("nan")),
+            candidates=dict(latencies)))
+    return winner
+
+
+def run_measured_arbitrated(region: Any, op: str, target: str, *inputs: Any,
+                            dims: tuple[int, ...] | None = None, dtype: str = "f32",
+                            cache: MeasureCache | None = None, reps: int = 20,
+                            warmup: int = 3) -> tuple[Any, str]:
+    """:func:`measured_arbitrate` then execute the winner on ``inputs`` →
+    ``(output, tag)``. Falls back to ``region.reference(*inputs)`` tagged
+    ``"reference"`` when no candidate wins (Decision #21: honest)."""
+    winner = measured_arbitrate(region, op, target, *inputs, dims=dims, dtype=dtype,
+                                cache=cache, reps=reps, warmup=warmup)
+    if winner is None:
+        _note_arbiter_dispatch(target, op, None, "reference")
+        return region.reference(*inputs), "reference"
+    out, tag = winner.run(region, *inputs)
+    _note_arbiter_dispatch(target, op, winner.name, tag)
+    return out, tag

python/tessera/compiler/emit/candidate.py

@@ -33,6 +33,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from collections import deque
 from enum import IntEnum
 from typing import Any, Callable
 
@@ -280,9 +281,63 @@ def run_arbitrated(region: Any, op: str, target: str, *inputs: Any,
     """Arbitrate then execute: pick the winning candidate and run it on
     ``inputs`` → ``(output, tag)``. When no candidate applies/verifies, fall back
     to ``region.reference(*inputs)`` tagged ``"reference"`` (Decision #21: honest —
-    never a mislabeled kernel)."""
+    never a mislabeled kernel). Every dispatch is recorded in the D3 arbiter log."""
     winner = arbitrate(region, op, target, verify=verify, force=force,
                        measure=measure)
     if winner is None:
+        _note_arbiter_dispatch(target, op, None, "reference")
         return region.reference(*inputs), "reference"
-    return winner.run(region, *inputs)
+    out, tag = winner.run(region, *inputs)
+    _note_arbiter_dispatch(target, op, winner.name, tag)
+    return out, tag
+
+
+# --- D3: arbiter dispatch log (did the compiled path win or silently degrade?) --
+#
+# Every arbitrated dispatch records ``(target, op, selected_candidate | None,
+# execution_tag)``. The arbiter *selects* a candidate, but that candidate's
+# ``run`` may still decline to the numpy reference at execution time (a device
+# error, an unsupported shape) — a **silent degrade** the tag reveals: the
+# selection is non-None but the tag is a reference tag. This is the observability
+# the plan's D3 asks for, generalized across every backend the arbiter serves
+# (the analog of ``runtime.dispatch_fallback_log`` for the arbiter layer).
+
+_ARBITER_LOG: "deque[tuple[str, str, str | None, str]]" = deque(maxlen=4096)
+
+
+def _note_arbiter_dispatch(target: str, op: str, selected: str | None,
+                           tag: str) -> None:
+    _ARBITER_LOG.append((target, op, selected, tag))
+
+
+def arbiter_dispatch_log() -> list[tuple[str, str, str | None, str]]:
+    """The ``(target, op, selected, tag)`` of each arbitrated dispatch this process
+    (most recent 4096). A ``selected`` that ran but whose ``tag`` is a reference
+    tag is a silent degrade."""
+    return list(_ARBITER_LOG)
+
+
+def reset_arbiter_dispatch_log() -> None:
+    _ARBITER_LOG.clear()
+
+
+def arbiter_dispatch_histogram(target: str | None = None, op: str | None = None
+                               ) -> dict[tuple[str, str], dict[str, int]]:
+    """Per ``(target, op)`` counts of ``{won, degraded, no_candidate}`` — did the
+    compiled path win, silently degrade to the reference, or was no candidate
+    available? Optionally filtered to a ``target`` / ``op``."""
+    from tessera.compiler.emit.kernel_emitter import REFERENCE_EXECUTIONS
+    hist: dict[tuple[str, str], dict[str, int]] = {}
+    for (t, o, selected, tag) in _ARBITER_LOG:
+        if target is not None and t != target:
+            continue
+        if op is not None and o != op:
+            continue
+        bucket = hist.setdefault((t, o), {"won": 0, "degraded": 0, "no_candidate": 0})
+        if selected is None:
+            bucket["no_candidate"] += 1
+        elif tag in REFERENCE_EXECUTIONS:
+            bucket["degraded"] += 1
+        else:
+            bucket["won"] += 1
+    return hist

python/tessera/runtime.py

@@ -2000,6 +2000,42 @@ def _nvidia_ptx_gemm_2d(A: Any, B: Any, dtype: str = "bfloat16") -> Any:
     return D
 
 
+_nvidia_device_name_probe: Any = False       # False = unprobed; None/str after
+
+
+def _nvidia_device_name() -> str | None:
+    """Best-effort stable per-device tag (``"sm_<cc>"``, e.g. ``"sm_120"``) for
+    autotune-cache keying — so a config measured on one device is never reused on
+    another. Cached; returns None (never raises) with no usable CUDA driver."""
+    global _nvidia_device_name_probe
+    if _nvidia_device_name_probe is not False:
+        return _nvidia_device_name_probe
+    _nvidia_device_name_probe = None
+    try:
+        cu = None
+        for cand in ("libcuda.so.1", "libcuda.so"):
+            try:
+                cu = ctypes.CDLL(cand)
+                break
+            except OSError:
+                continue
+        if cu is None:
+            return None
+        if cu.cuInit(0) != 0:
+            return None
+        dev = ctypes.c_int(0)
+        if cu.cuDeviceGet(ctypes.byref(dev), 0) != 0:
+            return None
+        major, minor = ctypes.c_int(), ctypes.c_int()
+        # CU_DEVICE_ATTRIBUTE_COMPUTE_CAPABILITY_{MAJOR,MINOR} = 75, 76
+        cu.cuDeviceGetAttribute(ctypes.byref(major), 75, dev)
+        cu.cuDeviceGetAttribute(ctypes.byref(minor), 76, dev)
+        _nvidia_device_name_probe = f"sm_{major.value}{minor.value}"
+    except Exception:
+        _nvidia_device_name_probe = None
+    return _nvidia_device_name_probe
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Stage L4 — the COMPILED GEMM lane (the default rocm matmul execution path).
 #