pin ci dotnet version

.github/workflows/build.yml

@@ -28,7 +28,7 @@ jobs:
 
       - uses: actions/setup-dotnet@v6
         with:
-          dotnet-version: '8.0.x'
+          dotnet-version: '8.0.427'
 
       - name: Cache NuGet packages
         uses: actions/cache@v5

.github/workflows/recomp-test.yml

@@ -20,7 +20,7 @@ jobs:
 
       - uses: actions/setup-dotnet@v6
         with:
-          dotnet-version: '8.0.x'
+          dotnet-version: '8.0.427'
 
       # Cache downloads only. Preparation still validates pins, and every run
       # compiles current Aurora, runtime, and generated sources from scratch.

Launcher/Directory.Build.props

@@ -1,7 +1,7 @@
 <Project>
   <PropertyGroup>
     <!-- The one product version. Release tags must match it. -->
-    <Version>0.9.2</Version>
+    <Version>0.9.3</Version>
     <IncludeSourceRevisionInInformationalVersion>false</IncludeSourceRevisionInInformationalVersion>
   </PropertyGroup>
 </Project>

runtime/include/perf_stats.h

@@ -4,6 +4,7 @@
 
 // MKW_PERF_STATS=1: logs the game thread's busy time per presented frame, i.e. wall time minus the
 // time it spends blocked on VI pacing, the GX worker or the audio mixer.
+// MKW_FRAME_TRACE=<csv path> also records those per-frame timings in Aurora's trace.
 namespace PerfStats {
 
 bool Enabled() noexcept;

runtime/src/hle/audio/audio.cpp

@@ -503,19 +503,20 @@ int64_t ConsumeAudioPollDeltaMicros()
     using Clock = std::chrono::steady_clock;
     static thread_local Clock::time_point lastPoll = Clock::now();
 
-    const Clock::time_point now = Clock::now();
-    auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(now - lastPoll).count();
-    lastPoll = now;
-
-    if (elapsed < 0) {
-        elapsed = 0;
-    }
-
     // Cap the catch-up interval after a debugger pause or host stall; backlog drains via
     // kMaxBlocksPerTick per pass. Never return early on a zero delta: two scheduler passes
     // can land in the same microsecond and the backlog still needs servicing.
-    constexpr int64_t kMaxPollDeltaMicros = 100'000;
-    return std::min(elapsed, kMaxPollDeltaMicros);
+    constexpr std::chrono::microseconds kMaxPollDelta{100'000};
+
+    const Clock::time_point now = Clock::now();
+    const auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(now - lastPoll);
+    if (elapsed.count() < 0 || elapsed > kMaxPollDelta) {
+        lastPoll = now;
+        return elapsed.count() < 0 ? 0 : kMaxPollDelta.count();
+    }
+    // Keep the sub-microsecond remainder for the next poll.
+    lastPoll += elapsed;
+    return elapsed.count();
 }
 
 } // namespace

runtime/src/hle/boot_race.cpp

@@ -5,6 +5,7 @@
 #include "guest_function_imports.h"
 #include "memory.h"
 #include "runtime_log.h"
+#include <aurora/aurora.h>
 
 #include <array>
 #include <charconv>
@@ -184,6 +185,11 @@ REGISTER_NATIVE_WRAPPER_AS(0x8052F4E8, RacedataScenario_InitScreens_8052f4e8, "R
 extern "C" void KartAIController_IsCPU_8072624c(CpuContext* ctx) {
     MKW_GUEST_FUNC(8072624C)(ctx);
     if (g_cpuDriven) {
+        static bool benchmarkReady = false;
+        if (!benchmarkReady) {
+            benchmarkReady = true;
+            aurora_report_benchmark_ready();
+        }
         ctx->gpr[3] = 1;
     }
 }

runtime/src/main.cpp

@@ -169,6 +169,7 @@ std::atomic_flag g_abortSignalHandled = ATOMIC_FLAG_INIT;
 std::atomic_bool g_fatalErrorReported{false};
 std::atomic_bool g_fatalPopupShown{false};
 bool g_headless = false;
+bool g_hidden = false;
 std::atomic<int> g_lastExitCode{0};
 std::atomic_bool g_exitCodeSet{false};
 
@@ -775,7 +776,7 @@ void MarkFatalErrorReported() {
 }
 
 void ShowRuntimeFatalPopup(std::string_view category, std::string_view details) noexcept {
-    if (g_headless || g_fatalPopupShown.exchange(true, std::memory_order_acq_rel)) {
+    if (g_headless || g_hidden || g_fatalPopupShown.exchange(true, std::memory_order_acq_rel)) {
         return;
     }
 
@@ -1350,16 +1351,19 @@ int RuntimeMain(int argc, char** argv) {
     // `--headless`: with a boot race, run without a window, sound or dialogs.
     g_headless = std::any_of(argv + 1, argv + argc,
                              [](const char* arg) { return std::string_view(arg) == "--headless"; });
+    g_hidden = std::any_of(argv + 1, argv + argc,
+                           [](const char* arg) { return std::string_view(arg) == "--hidden"; });
 
     try {
         for (int i = 1; i < argc; ++i) {
-            if (std::string_view(argv[i]) != "--headless" && !BootRace::ParseArgument(argv[i])) {
+            if (std::string_view(argv[i]) != "--headless" && std::string_view(argv[i]) != "--hidden" &&
+                !BootRace::ParseArgument(argv[i])) {
                 throw std::invalid_argument(std::string("Unknown command-line option '") + argv[i] +
-                                            "'; the game runtime only accepts --race[=<course id>], --race-cpu and --headless, use Config.toml through the installed host.");
+                                            "'; the game runtime only accepts --race[=<course id>], --race-cpu, --hidden and --headless, use Config.toml through the installed host.");
             }
         }
-        if (g_headless && !BootRace::Requested()) {
-            throw std::invalid_argument("--headless needs --race[=<course id>] or --race-cpu");
+        if ((g_headless || g_hidden) && !BootRace::Requested()) {
+            throw std::invalid_argument("--headless and --hidden need --race[=<course id>] or --race-cpu");
         }
         if (g_headless) {
             AudioBackend::Instance().SetSilenced(true);
@@ -1400,7 +1404,7 @@ int RuntimeMain(int argc, char** argv) {
         auroraConfig.hasWindowPosition = RuntimeConfigFile::WindowPosition(
             auroraConfig.windowPosX, auroraConfig.windowPosY);
         auroraConfig.allowJoystickBackgroundEvents = !g_headless;
-        auroraConfig.startHidden = g_headless;
+        auroraConfig.startHidden = g_hidden || g_headless;
         auroraConfig.disableCopyFilter = RuntimeConfigFile::DisableCopyFilter(true);
         // Dolphin-style custom textures. Aurora indexes <userPath>/texture_replacements
         // once during aurora_initialize, so both knobs only take effect on the next launch.

runtime/src/perf_stats.cpp

@@ -1,6 +1,7 @@
 #include "perf_stats.h"
 
 #include "runtime_log.h"
+#include <aurora/aurora.h>
 
 #include <algorithm>
 #include <array>
@@ -24,12 +25,20 @@ struct FrameWindow {
 
 FrameWindow g_window;
 
+bool LogEnabled() noexcept {
+    static const bool enabled = [] {
+        const char* value = std::getenv("MKW_PERF_STATS");
+        return value != nullptr && std::strcmp(value, "0") != 0;
+    }();
+    return enabled;
+}
+
 } // namespace
 
 bool Enabled() noexcept {
     static const bool enabled = [] {
-        const char* value = std::getenv("MKW_PERF_STATS");
-        return value != nullptr && std::strcmp(value, "0") != 0;
+        const char* trace = std::getenv("MKW_FRAME_TRACE");
+        return LogEnabled() || (trace != nullptr && *trace);
     }();
     return enabled;
 }
@@ -48,7 +57,11 @@ void OnFramePresented() noexcept {
         return;
     }
     const auto busy = (now - g_window.last) - wait;
+    aurora_report_producer_timing(
+        std::chrono::duration_cast<std::chrono::nanoseconds>(now - g_window.last).count(),
+        std::chrono::duration_cast<std::chrono::nanoseconds>(wait).count());
     g_window.last = now;
+    if (!LogEnabled()) return;
     g_window.busyMs[g_window.count++] = std::chrono::duration<double, std::milli>(busy).count();
     if (g_window.count < kWindow) {
         return;

runtime/tools/analyze_stutters.py

@@ -0,0 +1,348 @@
+"""Analyze MKW_FRAME_TRACE CSV files. No third-party packages required."""
+import argparse
+import bisect
+import collections
+import csv
+import json
+import math
+import io
+import re
+import statistics
+from pathlib import Path
+
+NS_MS = 1_000_000
+NATIVE_FPS = 60
+COLUMNS = ("event", "timestamp_ns", "frame_id", "slot", "duration_ns", "deadline_ns",
+           "interval_ns", "detail")
+INTEGER_COLUMNS = COLUMNS[1:]
+SIGNED_COLUMNS = {"slot"}
+DROP_SIGNALS = {"drop_late": "late_slot_dropped", "drop_display": "display_slot_dropped",
+                "present_unavailable": "present_unavailable", "present_failed": "present_failed"}
+ANCHORS = ("auto", "first_row", "first_real_present", "first_transform", "benchmark_ready")
+DISPLAY_LIMITED_INVALID_FRACTION = .01
+
+
+class TraceFormatError(ValueError):
+    pass
+
+
+def percentile(values, fraction):
+    if not values:
+        return 0.0
+    values = sorted(values)
+    return values[max(0, math.ceil(len(values) * fraction) - 1)]
+
+
+def distribution(values):
+    return {"count": len(values), "median_ms": percentile(values, .5),
+            "p95_ms": percentile(values, .95), "p99_ms": percentile(values, .99),
+            "max_ms": max(values, default=0)}
+
+
+def parse_trace(path):
+    """Returns (rows, lost_samples, truncated_tail); rows exclude trace_lost markers."""
+    path = Path(path)
+    try:
+        content = path.read_bytes().decode("utf-8-sig")
+    except UnicodeDecodeError as error:
+        raise TraceFormatError(f"{path}: not UTF-8 text ({error})") from None
+    truncated_tail = bool(content) and not content.endswith("\n")
+    if truncated_tail:
+        content = content[:content.rfind("\n") + 1]
+    reader = csv.reader(io.StringIO(content, newline=""))
+    try:
+        return _parse_rows(path, reader, truncated_tail)
+    except csv.Error as error:
+        raise TraceFormatError(f"{path}: line {reader.line_num}: {error}") from None
+
+
+def _parse_rows(path, reader, truncated_tail):
+    header = next(reader, None)
+    if header is None:
+        raise TraceFormatError(f"{path}: no complete header line")
+    duplicate = sorted({name for name in header if header.count(name) > 1})
+    missing = [name for name in COLUMNS if name not in header]
+    foreign = sorted({name for name in header if name not in COLUMNS})
+    if duplicate or missing or foreign:
+        problems = [f"{label} {names}" for label, names in
+                    (("duplicate columns", duplicate), ("missing columns", missing),
+                     ("unexpected columns", foreign)) if names]
+        raise TraceFormatError(f"{path}: line 1: not a frame trace header: {'; '.join(problems)}")
+    index = {name: header.index(name) for name in COLUMNS}
+    rows, lost = [], 0
+    for fields in reader:
+        line = reader.line_num
+        if len(fields) != len(header):
+            raise TraceFormatError(f"{path}: line {line}: expected {len(header)} fields, got {len(fields)}")
+        row = {"event": fields[index["event"]], "line": line}
+        if not row["event"]:
+            raise TraceFormatError(f"{path}: line {line}: empty event name")
+        for name in INTEGER_COLUMNS:
+            text = fields[index[name]]
+            if not re.fullmatch(r"-?[0-9]{1,20}", text):
+                raise TraceFormatError(f"{path}: line {line}: column '{name}' is not an integer: {text!r}")
+            row[name] = int(text)
+            if row[name] < 0 and name not in SIGNED_COLUMNS:
+                raise TraceFormatError(f"{path}: line {line}: column '{name}' must not be negative: {text}")
+        if row["event"] == "trace_lost":
+            lost += row["detail"]
+        else:
+            rows.append(row)
+    return rows, lost, truncated_tail
+
+
+def trace_state(path):
+    """Cheap readiness probe for a trace still being written; None until the header is complete."""
+    if b"\n" not in Path(path).read_bytes():
+        return None
+    rows = parse_trace(path)[0]
+    state = {"benchmark_ready_ns": None, "real_present_ns": None, "transform_ns": None,
+             "display_limited": False}
+    for row in rows:
+        if row["event"] == "benchmark_ready" and state["benchmark_ready_ns"] is None:
+            state["benchmark_ready_ns"] = row["timestamp_ns"]
+        elif row["event"] == "present_real" and state["real_present_ns"] is None:
+            state["real_present_ns"] = row["timestamp_ns"]
+        elif row["event"] == "interp_uniforms" and row["detail"] and state["transform_ns"] is None:
+            state["transform_ns"] = row["timestamp_ns"]
+        elif row["event"] == "display_limited":
+            state["display_limited"] = True
+    return state
+
+
+def native_period_ms(rows):
+    """Scheduled native VI period from the seal events; 60 Hz only when the trace has none."""
+    intervals = [row["interval_ns"] for row in rows if row["event"] == "seal" and row["interval_ns"]]
+    return statistics.median(intervals) / NS_MS if intervals else 1000 / NATIVE_FPS
+
+
+def timing_anchor(rows, requested):
+    if requested == "auto":
+        requested = "benchmark_ready"
+    matches = {"first_row": lambda row: True,
+               "benchmark_ready": lambda row: row["event"] == "benchmark_ready",
+               "first_real_present": lambda row: row["event"] == "present_real",
+               "first_transform": lambda row: row["event"] == "interp_uniforms" and row["detail"] > 0}
+    for used in dict.fromkeys((requested, "first_real_present", "first_row")):
+        row = next((row for row in rows if matches[used](row)), None)
+        if row:
+            return {"requested": requested, "used": used, "start_ns": row["timestamp_ns"],
+                    "found": used == requested}
+    raise AssertionError("first_row always matches")
+
+
+def analyze(path, warmup=15, fps=60, anchor="auto"):
+    if not math.isfinite(warmup) or warmup < 0:
+        raise ValueError(f"warmup must be a finite number of seconds >= 0, got {warmup}")
+    if isinstance(fps, bool) or not isinstance(fps, int) or fps <= 0:
+        raise ValueError(f"fps must be a positive integer, got {fps!r}")
+    if anchor not in ANCHORS:
+        raise ValueError(f"anchor must be one of {ANCHORS}, got {anchor!r}")
+    rows, lost, truncated_tail = parse_trace(path)
+    if not rows:
+        raise TraceFormatError(f"{path}: no timing samples ({lost} trace_lost samples, no events)")
+    rows.sort(key=lambda row: row["timestamp_ns"])
+    start = rows[0]["timestamp_ns"]
+    timing = timing_anchor(rows, anchor)
+    native_ms = native_period_ms(rows)
+    cutoff = timing["start_ns"] + warmup * 1_000_000_000
+    measured = [row for row in rows if row["timestamp_ns"] >= cutoff]
+    frames = collections.defaultdict(list)
+    for row in rows:
+        frames[row["frame_id"]].append(row)
+    counts = collections.Counter(row["event"] for row in measured)
+    stages = collections.defaultdict(list)
+    for row in measured:
+        if row["duration_ns"]:
+            stages[row["event"]].append(row["duration_ns"] / NS_MS)
+    drops = [row for row in measured if row["event"] in DROP_SIGNALS]
+    drop_times = [row["timestamp_ns"] for row in drops]
+    limited_ms = {}
+    for row in rows:
+        if row["event"] == "display_limited" and row["detail"]:
+            limited_ms[row["frame_id"]] = max(limited_ms.get(row["frame_id"], 0),
+                                              row["interval_ns"] / NS_MS or native_ms)
+
+    def evidence(frame_ids, window, attributed):
+        timings = collections.defaultdict(float)
+        pending = 0
+        wake_late = 0
+        signals = []
+        window_drops = drops[bisect.bisect_left(drop_times, window[0]):bisect.bisect_right(drop_times, window[1])]
+        for row in window_drops:
+            attributed.add(id(row))
+            signals.append(DROP_SIGNALS[row["event"]])
+        for frame_id in frame_ids:
+            for row in frames[frame_id]:
+                timings[row["event"]] = max(timings[row["event"]], row["duration_ns"] / NS_MS)
+                if row["event"] == "pipelines":
+                    pending = max(pending, row["detail"])
+                if row["event"] == "pacer_wait" and row["duration_ns"] > NS_MS and row["deadline_ns"]:
+                    wake_late = max(wake_late, (row["timestamp_ns"] - row["deadline_ns"]) / NS_MS)
+        for stage, threshold, label in [
+            ("surface_wait", 2, "surface_or_driver_wait"),
+            ("seal", 2, "interpolation_seal"),
+            ("pipeline_wait", 1, "pipeline_compilation_wait"),
+            ("worker_queue", 2, "render_worker_queue_wait"),
+            ("resource_wait", 2, "frame_resources_wait"),
+            ("replay_encode", 2, "interpolation_replay_encode"),
+            ("present_work", 2, "submit_or_present_cpu_work"),
+            ("produce", native_ms * 1.25, "producer_wall_time"),
+            ("game_busy", native_ms, "game_thread_busy_over_budget"),
+        ]:
+            if timings[stage] > threshold:
+                signals.append(label)
+        if pending:
+            signals.append("pipelines_pending")
+        if wake_late > 1:
+            signals.append("pacer_woke_late")
+        for frame_id in frame_ids:
+            for row in frames[frame_id]:
+                if row["event"] in DROP_SIGNALS:
+                    attributed.add(id(row))
+                    signals.append(DROP_SIGNALS[row["event"]])
+                if row["event"] == "display_limited":
+                    signals.append("display_refresh_limited")
+                if row["event"] == "interp_health" and fps > NATIVE_FPS:
+                    health = row["detail"]
+                    if not health & (1 << 18):
+                        signals.append("producer_missed_retrace")
+                    if not health & (1 << 17):
+                        signals.append("interpolation_replay_unsafe")
+                    if not health & (1 << 16):
+                        signals.append("interpolation_ineligible")
+                if row["event"] == "interp_matches" and fps > NATIVE_FPS:
+                    matchable, matches = row["detail"] >> 32, row["detail"] & 0xffffffff
+                    if matchable and matches * 100 < matchable * 55:
+                        signals.append("low_transform_match_coverage")
+        signals = sorted(set(signals))
+        return {"signals": signals or ["unattributed"], "stage_ms": dict(timings),
+                "queued_pipelines": pending, "pacer_wake_late_ms": wake_late}
+
+    def cadence(real_only):
+        presents = [row for row in rows if row["event"] in (
+            ("present_real",) if real_only else ("present_real", "present_interp"))]
+        budget = native_ms if real_only or fps <= NATIVE_FPS else min(native_ms, 1000 / fps)
+        tolerance = max(1, budget * .25)
+        intervals, spikes, effective_spikes, lateness = [], [], [], []
+        transitions = collections.defaultdict(list)
+        attributed = set()
+        limited_gaps = 0
+        effective_budget = budget
+        for previous, row in zip(presents, presents[1:]):
+            if previous["timestamp_ns"] < cutoff:
+                continue
+            gap = (row["timestamp_ns"] - previous["timestamp_ns"]) / NS_MS
+            intervals.append(gap)
+            transitions[previous["event"] + "->" + row["event"]].append(gap)
+            if row["deadline_ns"]:
+                lateness.append(max(0, row["timestamp_ns"] - row["deadline_ns"]) / NS_MS)
+            limit = max(limited_ms.get(previous["frame_id"], 0), limited_ms.get(row["frame_id"], 0))
+            expected = max(budget, limit)
+            limited_gaps += expected > budget
+            effective_budget = max(effective_budget, expected)
+            if gap > budget + tolerance:
+                nearby = set()
+                spike = {"seconds": (row["timestamp_ns"] - start) / 1e9,
+                         "frame_id": row["frame_id"], "kind": row["event"],
+                         "gap_ms": gap, "excess_ms": gap - budget,
+                         **evidence({previous["frame_id"], row["frame_id"]},
+                                    (previous["timestamp_ns"], row["timestamp_ns"]), nearby)}
+                spikes.append(spike)
+                if gap > expected + max(1, expected * .25):
+                    effective_spikes.append({**spike, "excess_ms": gap - expected})
+                    attributed |= nearby
+        worst = lambda found: sorted(found, key=lambda item: item["gap_ms"], reverse=True)[:20]
+        counted = lambda found: dict(collections.Counter(
+            signal for spike in found for signal in spike["signals"]))
+        return {**distribution(intervals), "budget_ms": budget, "tolerance_ms": tolerance,
+                "stutters": len(spikes), "signals": counted(spikes),
+                "effective_budget_ms": effective_budget, "display_limited_gaps": limited_gaps,
+                "effective_stutters": len(effective_spikes),
+                "effective_signals": counted(effective_spikes),
+                "transitions": {key: distribution(values) for key, values in transitions.items()},
+                "deadline_lateness": distribution(lateness),
+                "worst": worst(spikes), "effective_worst": worst(effective_spikes)}, attributed
+
+    real, _ = cadence(True)
+    all_presentations, attributed = cadence(False)
+    isolated = [row for row in drops if id(row) not in attributed]
+    match_ratios = [100 * (row["detail"] & 0xffffffff) / (row["detail"] >> 32)
+                    for row in measured if row["event"] == "interp_matches" and row["detail"] >> 32]
+    real_presents = counts["present_real"]
+    limited_frames = len({row["frame_id"] for row in measured
+                          if row["event"] == "display_limited" and row["detail"]})
+    limited_fraction = limited_frames / real_presents if real_presents else 0.0
+    if fps <= NATIVE_FPS:
+        status = "not_requested"
+    elif not real_presents:
+        status = "no_real_presents"
+    elif limited_fraction >= DISPLAY_LIMITED_INVALID_FRACTION:
+        status = "display_limited"
+    elif not any(row["event"] == "interp_uniforms" and row["detail"] > 0 for row in measured):
+        status = "no_transform_work"
+    elif not counts["present_interp"]:
+        status = "no_interpolated_presents"
+    else:
+        status = "ok"
+    return {"trace": str(path), "warmup_seconds": warmup, "target_fps": fps,
+            "timing_anchor": timing, "native_period_ms": native_ms,
+            "native_period_source": "seal" if any(row["event"] == "seal" and row["interval_ns"]
+                                                  for row in rows) else "default_60hz",
+            "interpolation": {"status": status, "target_valid": status in ("ok", "not_requested"),
+                              "display_limited": limited_fraction >= DISPLAY_LIMITED_INVALID_FRACTION,
+                              "display_limited_frames": limited_frames,
+                              "display_limited_fraction": limited_fraction,
+                              "interpolated_presents": counts["present_interp"]},
+            "measured_seconds": max(0, (rows[-1]["timestamp_ns"] - cutoff) / 1e9),
+            "lost_trace_samples": lost, "truncated_final_row": truncated_tail, "events": dict(counts),
+            "isolated_drops": len(isolated),
+            "isolated_drop_examples": [{"seconds": (row["timestamp_ns"] - start) / 1e9, "event": row["event"],
+                                        "frame_id": row["frame_id"], "slot": row["slot"]}
+                                       for row in isolated[:5]],
+            "unassociated_game_timing_samples": sum(row["event"] == "game_busy" and row["frame_id"] == 0
+                                                     for row in measured),
+            "real": real, "all_presentations": all_presentations,
+            "stages": {key: distribution(values) for key, values in stages.items()},
+            "transform_matches": {"frames": len(match_ratios),
+                                  "low_match_frames": sum(value < 55 for value in match_ratios),
+                                  "median_percent": percentile(match_ratios, .5),
+                                  "p05_percent": percentile(match_ratios, .05)},
+            "limitations": ["Present timestamps measure CPU surface submission, not physical scanout or GPU execution.",
+                            "Signals correlate slow stages with gaps; they do not prove causation.",
+                            "Producer wall time includes guest waits. Hidden-window results can differ from visible VSync.",
+                            "A forced stop can discard the final 100 ms of trace samples.",
+                            "stutters/signals use the requested fps budget; effective_* also accept deliberate "
+                            "display-limited suppression. Check interpolation.target_valid before comparing targets."]}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("trace", type=Path)
+    parser.add_argument("--warmup", type=float, default=15)
+    parser.add_argument("--fps", type=int, default=60)
+    parser.add_argument("--anchor", choices=ANCHORS, default="auto",
+                        help="warmup start: auto uses the renderer's benchmark_ready event (first CPU racer "
+                             "initialized, independent of --fps), else the first real present")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    try:
+        result = analyze(args.trace, args.warmup, args.fps, args.anchor)
+    except (OSError, ValueError) as error:
+        raise SystemExit(f"error: {error}")
+    if args.output:
+        args.output.write_text(json.dumps(result, indent=2), encoding="utf-8")
+    for key in ("real", "all_presentations"):
+        entry = result[key]
+        print(f"{key}: {entry['count']} intervals, p99 {entry['p99_ms']:.3f} ms, "
+              f"max {entry['max_ms']:.3f} ms, {entry['stutters']} requested / "
+              f"{entry['effective_stutters']} effective stutters; {entry['effective_signals']}")
+    interpolation = result["interpolation"]
+    print(f"interpolation: {interpolation['status']}, anchor {result['timing_anchor']['used']}")
+    print(f"late slots: {result['events'].get('drop_late', 0)}, "
+          f"display drops: {result['events'].get('drop_display', 0)}, "
+          f"isolated drops: {result['isolated_drops']}, trace loss: {result['lost_trace_samples']}")
+
+
+if __name__ == "__main__":
+    main()

runtime/tools/benchmark_stutters.py

@@ -0,0 +1,354 @@
+"""Windows hidden-race benchmark with inherited affinity and isolated portable settings.
+
+Every case gets its own portable directory, seeded from one pristine snapshot of the NAND and
+(optionally) the cache, so a case never inherits another case's changed cache or saves.
+"""
+import argparse
+import ctypes
+from ctypes import wintypes
+import itertools
+import json
+import math
+import os
+import re
+from pathlib import Path
+import shutil
+import subprocess
+import time
+import tomllib
+
+from analyze_stutters import analyze, trace_state
+
+SUPPORTED_RESOLUTIONS = (0, 1, 1.5, 2, 3, 4, 6, 8)
+GRAPHICS_APIS = ("auto", "d3d12", "vulkan")
+FPS_TARGETS = (0, 120, 180)
+HELPER_COUNTS = ("auto", *map(str, range(7)))
+STARTUP_TIMEOUT = 60
+
+
+class ConfigError(ValueError):
+    pass
+
+
+def cpu_topology():
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+    kernel.GetCurrentProcess.restype = wintypes.HANDLE
+    kernel.GetProcessAffinityMask.argtypes = [wintypes.HANDLE, ctypes.POINTER(ctypes.c_size_t),
+                                              ctypes.POINTER(ctypes.c_size_t)]
+    kernel.SetProcessAffinityMask.argtypes = [wintypes.HANDLE, ctypes.c_size_t]
+    process, system = ctypes.c_size_t(), ctypes.c_size_t()
+    handle = kernel.GetCurrentProcess()
+    if not kernel.GetProcessAffinityMask(handle, ctypes.byref(process), ctypes.byref(system)):
+        raise ctypes.WinError(ctypes.get_last_error())
+
+    # SYSTEM_LOGICAL_PROCESSOR_INFORMATION has an affinity mask, relationship enum,
+    # and a 16-byte union. RelationProcessorCore (0) gives each core's SMT mask.
+    class ProcessorInfo(ctypes.Structure):
+        _fields_ = [("mask", ctypes.c_size_t), ("relationship", ctypes.c_int),
+                    ("data", ctypes.c_ulonglong * 2)]
+    size = wintypes.DWORD()
+    kernel.GetLogicalProcessorInformation(None, ctypes.byref(size))
+    buffer = ctypes.create_string_buffer(size.value)
+    if not kernel.GetLogicalProcessorInformation(buffer, ctypes.byref(size)):
+        raise ctypes.WinError(ctypes.get_last_error())
+    cores = []
+    for offset in range(0, size.value, ctypes.sizeof(ProcessorInfo)):
+        info = ProcessorInfo.from_buffer(buffer, offset)
+        available = info.mask & process.value
+        if info.relationship == 0 and available:
+            cores.append(available & -available)  # one logical processor per physical core
+    return kernel, handle, process.value, cores
+
+
+def quoted(value):
+    # JSON basic strings are valid TOML except for raw DEL.
+    return json.dumps(str(value), ensure_ascii=False).replace("\x7f", "\\u007f")
+
+
+def interpolation_helpers(core_count, requested):
+    if requested != "auto":
+        return int(requested)
+    if not core_count:
+        return None
+    return min(core_count - 1, 6) if core_count > 2 else 0
+
+
+def parse_list(text, name, parse, allowed=None):
+    """Comma-separated matrix axis; rejects malformed, unsupported and duplicate entries."""
+    values = []
+    for item in text.split(","):
+        try:
+            values.append(parse(item.strip()))
+        except ValueError:
+            raise ValueError(f"--{name}: invalid entry {item!r}") from None
+    if allowed is not None and (bad := [value for value in values if value not in allowed]):
+        raise ValueError(f"--{name}: unsupported {bad}; allowed {list(allowed)}")
+    if len(set(values)) != len(values):
+        raise ValueError(f"--{name}: duplicate entries in {text!r}")
+    return values
+
+
+def parse_helper(text):
+    return text if text == "auto" else str(int(text))
+
+
+class SourceConfig:
+    def __init__(self, dvd_root, retro_rewind_root, nand_root, graphics_api):
+        self.dvd_root, self.retro_rewind_root = dvd_root, retro_rewind_root
+        self.nand_root, self.graphics_api = nand_root, graphics_api
+
+
+def load_source_config(config_path):
+    """Reads and validates the user's Config.toml before anything is staged."""
+    config_path = Path(config_path)
+    try:
+        with config_path.open("rb") as source:
+            config = tomllib.load(source)
+    except OSError as error:
+        raise ConfigError(f"cannot read config {config_path}: {error}") from None
+    except (tomllib.TOMLDecodeError, UnicodeDecodeError) as error:
+        raise ConfigError(f"{config_path}: invalid TOML: {error}") from None
+
+    def table(name):
+        value = config.get(name, {})
+        if not isinstance(value, dict):
+            raise ConfigError(f"{config_path}: [{name}] must be a table, got {type(value).__name__}")
+        return value
+
+    paths, video = table("paths"), table("video")
+    base = config_path.resolve().parent
+
+    def source_path(key, directory=False):
+        value = paths.get(key)
+        if value is None or value == "":
+            return None
+        if not isinstance(value, str):
+            raise ConfigError(f"{config_path}: paths.{key} must be a string, got {type(value).__name__}")
+        path = Path(value)
+        path = path if path.is_absolute() else (base / path).resolve()
+        if not (path.is_dir() if directory else path.exists()):
+            raise ConfigError(f"{config_path}: paths.{key} {'is not a directory' if directory else 'does not exist'}: {path}")
+        return path
+
+    api = video.get("graphics_api", "auto")
+    if not isinstance(api, str) or api not in GRAPHICS_APIS:
+        raise ConfigError(f"{config_path}: video.graphics_api must be one of {list(GRAPHICS_APIS)}, got {api!r}")
+    return SourceConfig(source_path("dvd_root"), source_path("retro_rewind_root"),
+                        source_path("nand_root", directory=True), api)
+
+
+def isolated_config_text(source, nand_root, fps, resolution, vsync):
+    lines = ["[paths]"]
+    for key in ("dvd_root", "retro_rewind_root"):
+        if path := getattr(source, key):
+            lines.append(f"{key} = {quoted(path)}")
+    lines.append(f"nand_root = {quoted(nand_root)}")
+    # No physical controller scans or presence/network traffic during measurements.
+    # Keep the audio mixer running; --hidden retains normal audio behavior.
+    lines += ["[controller]", "wii_remotes = false", "[network]", "enabled = false",
+              "[discord]", "enabled = false", "[audio]", "volume = 0.0",
+              "attenuate_music_when_media_plays = false",
+              "[video]", "window_width = 854", "window_height = 480", 'display_mode = "windowed"',
+              f"graphics_api = {quoted(source.graphics_api)}",
+              f"resolution_multiplier = {float(resolution)}", f"frame_interpolation_fps = {fps}",
+              f"vsync = {'true' if vsync else 'false'}", "show_fps = false"]
+    return "\n".join(lines) + "\n"
+
+
+def wait_ready(process, trace, label):
+    """Waits for the renderer's benchmark_ready marker (first CPU racer initialized)."""
+    deadline = time.monotonic() + STARTUP_TIMEOUT
+    while True:
+        if process.poll() is not None:
+            raise RuntimeError(f"{label}: game exited during startup ({process.returncode}); see log")
+        state = trace_state(trace) if trace.exists() else None
+        if state and state["benchmark_ready_ns"] is not None:
+            return state
+        if time.monotonic() >= deadline:
+            raise RuntimeError(f"{label}: no benchmark_ready trace event within {STARTUP_TIMEOUT}s of "
+                               "startup; the race did not start or the executable predates the marker; see log")
+        time.sleep(.25)
+
+
+def stop_process(process, label):
+    if process.poll() is not None:
+        return
+    process.terminate()
+    try:
+        process.wait(timeout=10)
+        return
+    except subprocess.TimeoutExpired:
+        process.kill()
+    try:
+        process.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        raise RuntimeError(f"{label}: game did not exit after terminate and kill") from None
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--exe", type=Path, required=True)
+    parser.add_argument("--config", type=Path,
+                        default=Path(os.environ.get("LOCALAPPDATA", ".")) / "WiiCompiled/Config.toml")
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--cpus", default="0,4", help="physical cores without SMT; 0 keeps full inherited affinity")
+    parser.add_argument("--fps", default="0,120")
+    parser.add_argument("--helpers", default="auto", help="auto or comma-separated counts 0..6")
+    parser.add_argument("--seconds", type=float, default=60, help="measurement time, excluding warmup")
+    parser.add_argument("--warmup", type=float, default=15)
+    parser.add_argument("--repeats", type=int, default=1)
+    parser.add_argument("--course", type=lambda value: int(value, 0), default=8)
+    parser.add_argument("--resolution", type=float, default=1)
+    parser.add_argument("--vsync", action="store_true")
+    parser.add_argument("--no-shared-cache-affinity", action="store_true",
+                        help="keep all threads within the process mask, without additional cache-domain pinning")
+    parser.add_argument("--cache-from", type=Path,
+                        help="snapshot this Cache directory once and seed every case from the snapshot; "
+                             "without it every case starts cold")
+    parser.add_argument("--allow-display-limited", action="store_true",
+                        help="exit 0 even when a requested interpolation target was suppressed by the display")
+    args = parser.parse_args()
+    if os.name != "nt":
+        parser.error("The runner requires Windows; the trace analyzer is cross-platform.")
+    try:
+        cpus = parse_list(args.cpus, "cpus", int)
+        fps_targets = parse_list(args.fps, "fps", int, FPS_TARGETS)
+        helpers = parse_list(args.helpers, "helpers", parse_helper, HELPER_COUNTS)
+    except ValueError as error:
+        parser.error(str(error))
+    if not (math.isfinite(args.seconds) and args.seconds > 0 and math.isfinite(args.warmup) and args.warmup >= 0):
+        parser.error("seconds must be finite and > 0; warmup must be finite and >= 0")
+    if args.repeats < 1 or not 0 <= args.course <= 31:
+        parser.error("invalid repeats or course")
+    if not math.isfinite(args.resolution) or args.resolution not in SUPPORTED_RESOLUTIONS:
+        parser.error(f"resolution must be one of {list(SUPPORTED_RESOLUTIONS)}")
+    kernel, handle, original_mask, cores = cpu_topology()
+    if any(value < 0 or value > len(cores) for value in cpus):
+        parser.error(f"cpus must be between 0 and {len(cores)}")
+    try:
+        exe = args.exe.resolve(strict=True)
+        source = load_source_config(args.config)
+    except (OSError, ConfigError) as error:
+        parser.error(str(error))
+    output = args.output.resolve()
+    if source.nand_root and output.is_relative_to(source.nand_root.resolve()):
+        parser.error("--output must not be inside the source nand_root; the NAND snapshot would copy itself")
+    cache_from = args.cache_from.resolve() if args.cache_from else None
+    if cache_from and (not cache_from.is_dir() or output.is_relative_to(cache_from)
+                       or cache_from.is_relative_to(output)):
+        parser.error("--cache-from must be an existing directory outside the output directory")
+    if (output / "portable").exists() or (output / "seeds").exists():
+        parser.error("output already contains portable/ or seeds/; use a fresh output directory")
+
+    output.mkdir(parents=True, exist_ok=True)
+    seeds = output / "seeds"
+    nand_seed, cache_seed = seeds / "NAND", seeds / "Cache"
+    if source.nand_root:
+        shutil.copytree(source.nand_root, nand_seed, ignore=shutil.ignore_patterns("*.nandsafe.tmp"))
+    else:
+        nand_seed.mkdir(parents=True)
+    if cache_from:
+        shutil.copytree(cache_from, cache_seed)
+    cache_mode = "seeded" if cache_from else "cold"
+    print(f"cache: {cache_mode}" + (f" (every case copies the snapshot of {cache_from})" if cache_from
+          else " (every case starts with an empty cache; results include shader/pipeline compilation)"),
+          flush=True)
+    runtime_files = [exe, *sorted(itertools.chain(exe.parent.glob("*.dll"), exe.parent.glob("*.bin"),
+                                                  exe.parent.glob("initial_pipeline_cache.db")))]
+
+    summaries, invalid = [], []
+    for repeat, core_count, fps, helper in itertools.product(range(args.repeats), cpus, fps_targets, helpers):
+        effective_helpers = interpolation_helpers(core_count, helper)
+        label = f"r{repeat}-cpu{core_count}-fps{fps}-helpers{helper}"
+        mask = original_mask if core_count == 0 else sum(cores[:core_count])
+        trace = output / f"{label}.csv"
+        stage = output / "portable" / label
+        user = stage / "UserData"
+        user.mkdir(parents=True)
+        (stage / "portable.txt").write_text("stutter benchmark\n")
+        for runtime_file in dict.fromkeys(runtime_files):
+            shutil.copy2(runtime_file, stage / runtime_file.name)
+        shutil.copytree(nand_seed, user / "NAND")
+        if cache_from:
+            shutil.copytree(cache_seed, user / "Cache")
+        (user / "Config.toml").write_text(
+            isolated_config_text(source, user / "NAND", fps, args.resolution, args.vsync), encoding="utf-8")
+        print(f"running {label}, affinity=0x{mask:x}, cache={cache_mode}", flush=True)
+        env = os.environ.copy()
+        env["MKW_FRAME_TRACE"] = str(trace)
+        env["MKW_PERF_STATS"] = "1"
+        env.pop("MKW_INTERPOLATION_HELPERS", None)
+        env.pop("MKW_SHARED_CACHE_AFFINITY", None)
+        if args.no_shared_cache_affinity:
+            env["MKW_SHARED_CACHE_AFFINITY"] = "0"
+        if effective_helpers is not None:
+            env["MKW_INTERPOLATION_HELPERS"] = str(effective_helpers)
+        with (output / f"{label}.log").open("wb") as log:
+            try:
+                if not kernel.SetProcessAffinityMask(handle, mask):
+                    raise ctypes.WinError(ctypes.get_last_error())
+                # The child inherits affinity at creation, before any worker pool sizes itself.
+                process = subprocess.Popen([str(stage / exe.name), "--hidden", "--race-cpu",
+                                            f"--race={args.course}"], cwd=stage, env=env,
+                                           stdout=log, stderr=subprocess.STDOUT,
+                                           creationflags=subprocess.CREATE_NO_WINDOW)
+            finally:
+                if not kernel.SetProcessAffinityMask(handle, original_mask):
+                    raise ctypes.WinError(ctypes.get_last_error())
+            forced_stop = False
+            try:
+                # The analyzer anchors warmup at the same event for every fps target, so
+                # boot and loading never consume the warmup/measurement interval.
+                started = time.monotonic()
+                ready = wait_ready(process, trace, label)
+                startup_seconds = time.monotonic() - started
+                process.wait(timeout=args.warmup + args.seconds + 1)
+                raise RuntimeError(f"{label}: game exited early ({process.returncode}); see log")
+            except subprocess.TimeoutExpired:
+                forced_stop = True
+            finally:
+                stop_process(process, label)
+        result = analyze(trace, args.warmup, fps or 60)
+        if result["lost_trace_samples"] or result["truncated_final_row"]:
+            raise RuntimeError(f"{label}: incomplete trace; see log")
+        log_text = (output / f"{label}.log").read_text(encoding="utf-8", errors="replace")
+        affinity_counts = {name: int(count) for name, count in re.findall(
+            r"\[aurora::thread\] (.+?): (\d+) available CPUs after affinity", log_text)}
+        if result["real"]["count"] == 0:
+            raise RuntimeError(f"{label}: no real frame presentations after warmup; see log")
+        interpolation = result["interpolation"]
+        if fps and interpolation["status"] == "no_transform_work":
+            raise RuntimeError(f"{label}: no transform interpolation work after warmup; verify the race loaded")
+        if result["measured_seconds"] < args.seconds * .9:
+            raise RuntimeError(f"{label}: measured only {result['measured_seconds']:.1f}s of {args.seconds}s; see log")
+        if fps and not interpolation["target_valid"]:
+            invalid.append(label)
+            print(f"  WARNING {label}: requested {fps} fps target is {interpolation['status']} "
+                  f"({interpolation['display_limited_fraction']:.0%} of frames display-limited); "
+                  "use effective_* statistics and do not compare it with a real 120/180 fps run", flush=True)
+        result.update(label=label, affinity_mask=hex(mask), physical_core_limit=core_count,
+                      helpers=helper, repeat=repeat, forced_stop=forced_stop,
+                      effective_helpers=effective_helpers, host_logical_processors=os.cpu_count(),
+                      resolution=args.resolution, vsync=args.vsync, course=args.course,
+                      shared_cache_affinity=not args.no_shared_cache_affinity,
+                      cache_mode=cache_mode, cache_seed=str(cache_from) if cache_from else None,
+                      startup_seconds=startup_seconds, startup_ready=ready,
+                      thread_affinity_cpu_counts=affinity_counts,
+                      affinity_narrowed=bool(core_count and any(
+                          count < core_count for count in affinity_counts.values())),
+                      executable=str(exe), available_core_masks=[hex(core) for core in cores])
+        (output / f"{label}.json").write_text(json.dumps(result, indent=2), encoding="utf-8")
+        summaries.append(result)
+        (output / "summary.json").write_text(json.dumps(summaries, indent=2), encoding="utf-8")
+        for kind in ("real", "all_presentations"):
+            entry = result[kind]
+            print(f"  {kind}: p99={entry['p99_ms']:.2f} max={entry['max_ms']:.2f} ms "
+                  f"stutters={entry['stutters']} effective={entry['effective_stutters']} "
+                  f"signals={entry['effective_signals']}", flush=True)
+        print(f"  interpolation={interpolation['status']} isolated_drops={result['isolated_drops']}", flush=True)
+    if invalid and not args.allow_display_limited:
+        raise SystemExit(f"display-limited or invalid interpolation targets: {', '.join(invalid)}")
+
+
+if __name__ == "__main__":
+    main()

runtime/tools/test_analyze_stutters.py

@@ -0,0 +1,332 @@
+import csv
+from pathlib import Path
+import tempfile
+import unittest
+
+from analyze_stutters import COLUMNS, TraceFormatError, analyze, trace_state
+
+HEADER = ",".join(COLUMNS) + "\n"
+
+
+class AnalyzeStuttersTests(unittest.TestCase):
+    def analyze_rows(self, rows, fps=120, warmup=0, tail="", **options):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "trace.csv"
+            with path.open("w", newline="") as stream:
+                writer = csv.writer(stream)
+                writer.writerow(["event", "timestamp_ns", "frame_id", "slot", "duration_ns",
+                                 "deadline_ns", "interval_ns", "detail"])
+                writer.writerows(rows)
+                stream.write(tail)
+            return analyze(path, warmup=warmup, fps=fps, **options)
+
+    def analyze_text(self, text, fps=60, warmup=0):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "trace.csv"
+            path.write_bytes(text.encode("utf-8"))
+            return analyze(path, warmup=warmup, fps=fps)
+
+    @staticmethod
+    def present(ms, frame, real=True):
+        at = round(ms * 1_000_000)
+        return ["present_real" if real else "present_interp", at, frame, 0, 0, at, 8_333_333, 0]
+
+    @staticmethod
+    def limited(ms, frame):
+        return ["display_limited", round(ms * 1_000_000), frame, -1, 0, 0, 16_666_667, 1]
+
+    @staticmethod
+    def seal(ms, frame, period_ns):
+        return ["seal", round(ms * 1_000_000), frame, -1, 1_000_000, 0, period_ns, 0]
+
+    @staticmethod
+    def ready(ms):
+        return ["benchmark_ready", round(ms * 1_000_000), 0, -1, 0, 0, 0, 0]
+
+    def pal_rows(self, gaps_ms, start=100):
+        rows, at = [], start
+        for frame, gap in enumerate([0, *gaps_ms], 1):
+            at += gap
+            rows += [self.present(at, frame), self.seal(at - 1, frame, 20_000_000)]
+        return rows
+
+    def display_limited_rows(self, frames=40):
+        rows = []
+        for i in range(frames):
+            rows += [self.present(100 + i * 1000 / 60, i + 1), self.limited(100 + i * 1000 / 60, i + 1)]
+        return rows
+
+    def test_regular_interpolation_has_no_stutters(self):
+        rows = [self.present(100 + i * 1000 / 120, i // 2, i % 2 == 1) for i in range(40)]
+        result = self.analyze_rows(rows)
+        self.assertEqual(result["real"]["stutters"], 0)
+        self.assertEqual(result["all_presentations"]["stutters"], 0)
+        self.assertAlmostEqual(result["all_presentations"]["p99_ms"], 1000 / 120, places=5)
+
+    def test_missing_interpolation_slot_does_not_mean_real_frame_stutter(self):
+        rows = [self.present(100, 1), self.present(108.333333, 2, False),
+                self.present(116.666667, 2), self.present(133.333333, 3)]
+        rows.append(["seal", 130_000_000, 3, -1, 4_000_000, 0, 0, 0])
+        rows.append(["drop_late", 131_000_000, 3, 0, 0, 125_000_000, 8_333_333, 0])
+        result = self.analyze_rows(rows)
+        self.assertEqual(result["real"]["stutters"], 0)
+        self.assertEqual(result["all_presentations"]["stutters"], 1)
+        self.assertEqual(result["all_presentations"]["effective_stutters"], 1)
+        self.assertIn("interpolation_seal", result["all_presentations"]["signals"])
+        self.assertIn("late_slot_dropped", result["all_presentations"]["effective_signals"])
+        self.assertEqual(result["events"]["drop_late"], 1)
+        self.assertEqual(result["isolated_drops"], 0)
+
+    def test_real_frame_gap_and_driver_wait(self):
+        rows = [self.present(100, 1), self.present(140, 2),
+                ["surface_wait", 140_000_001, 2, 0, 20_000_000, 0, 0, 0]]
+        result = self.analyze_rows(rows, fps=60)
+        self.assertEqual(result["real"]["stutters"], 1)
+        self.assertIn("surface_or_driver_wait", result["real"]["signals"])
+        self.assertEqual(result["interpolation"]["status"], "not_requested")
+        self.assertTrue(result["interpolation"]["target_valid"])
+
+    def test_game_work_is_distinct_from_gx_producer_wall_time(self):
+        rows = [self.present(100, 1), self.present(140, 2),
+                ["game_busy", 139_000_000, 2, -1, 30_000_000, 0, 40_000_000, 10_000_000],
+                ["produce", 138_000_000, 2, -1, 5_000_000, 0, 0, 0]]
+        result = self.analyze_rows(rows, fps=60)
+        self.assertIn("game_thread_busy_over_budget", result["real"]["signals"])
+        self.assertNotIn("producer_wall_time", result["real"]["signals"])
+        self.assertEqual(result["unassociated_game_timing_samples"], 0)
+
+    def test_warmup_excludes_cross_boundary_gap(self):
+        rows = [self.present(100, 1), self.present(2100, 2), self.present(2116.666667, 3)]
+        result = self.analyze_rows(rows, fps=60, warmup=1)
+        self.assertEqual(result["real"]["count"], 1)
+        self.assertEqual(result["real"]["stutters"], 0)
+
+    def test_loss_and_incomplete_final_write_are_visible(self):
+        rows = [self.present(100, 1), self.present(116.666667, 2),
+                ["trace_lost", 0, 0, -1, 0, 0, 0, 3]]
+        result = self.analyze_rows(rows, fps=60, tail="present_real,123")
+        self.assertEqual(result["lost_trace_samples"], 3)
+        self.assertTrue(result["truncated_final_row"])
+
+    def test_display_limit_is_reported_separately_from_cpu_signals(self):
+        rows = [self.present(100, 1), self.present(116.666667, 2),
+                ["display_limited", 116_000_000, 2, -1, 0, 0, 16_666_667, 1]]
+        result = self.analyze_rows(rows)
+        self.assertIn("display_refresh_limited", result["all_presentations"]["signals"])
+
+    def test_average_fps_does_not_hide_alternating_intervals(self):
+        rows = [self.present(100, 1), self.present(115, 2, False),
+                self.present(116.666667, 2), self.present(131.666667, 3, False),
+                self.present(133.333333, 3)]
+        result = self.analyze_rows(rows)
+        self.assertEqual(result["real"]["stutters"], 0)
+        self.assertEqual(result["all_presentations"]["stutters"], 2)
+        transitions = result["all_presentations"]["transitions"]
+        self.assertAlmostEqual(transitions["present_real->present_interp"]["median_ms"], 15)
+        self.assertAlmostEqual(transitions["present_interp->present_real"]["median_ms"], 1000 / 600, places=5)
+
+    def test_low_transform_coverage_is_visible_even_with_regular_presents(self):
+        rows = [self.present(100, 1), self.present(108.333333, 2, False),
+                self.present(116.666667, 2),
+                ["interp_matches", 116_000_000, 2, -1, 0, 0, 0, (100 << 32) | 20]]
+        result = self.analyze_rows(rows)
+        self.assertEqual(result["all_presentations"]["stutters"], 0)
+        self.assertEqual(result["transform_matches"]["low_match_frames"], 1)
+        self.assertEqual(result["transform_matches"]["median_percent"], 20)
+
+    def test_healthy_60hz_display_requested_at_120_is_not_a_stutter_regression(self):
+        result = self.analyze_rows(self.display_limited_rows())
+        self.assertEqual(result["real"]["stutters"], 0)
+        self.assertGreater(result["all_presentations"]["stutters"], 0)
+        self.assertEqual(result["all_presentations"]["effective_stutters"], 0)
+        self.assertEqual(result["all_presentations"]["effective_signals"], {})
+        self.assertGreater(result["all_presentations"]["display_limited_gaps"], 0)
+        self.assertEqual(result["interpolation"]["status"], "display_limited")
+        self.assertTrue(result["interpolation"]["display_limited"])
+        self.assertFalse(result["interpolation"]["target_valid"])
+        self.assertEqual(result["isolated_drops"], 0)
+
+    def test_real_gap_on_display_limited_stream_is_still_a_stutter(self):
+        rows = self.display_limited_rows(10)
+        rows += [self.present(100 + 9 * 1000 / 60 + 50, 11), self.limited(100 + 9 * 1000 / 60 + 50, 11)]
+        result = self.analyze_rows(rows)
+        self.assertEqual(result["all_presentations"]["effective_stutters"], 1)
+        self.assertEqual(result["real"]["stutters"], 1)
+
+    def test_isolated_drop_late_survives_healthy_cadence(self):
+        rows = [self.present(100 + i * 1000 / 120, i // 2, i % 2 == 1) for i in range(40)]
+        rows.append(["drop_late", 150_000_000, 5, 0, 0, 149_000_000, 8_333_333, 0])
+        result = self.analyze_rows(rows)
+        self.assertEqual(result["all_presentations"]["effective_stutters"], 0)
+        self.assertEqual(result["isolated_drops"], 1)
+        self.assertEqual(result["isolated_drop_examples"][0]["event"], "drop_late")
+
+    def test_display_and_present_failures_map_to_signals_in_the_gap_window(self):
+        rows = [self.present(100, 1), self.present(150, 4),
+                ["drop_display", 110_000_000, 2, 0, 0, 0, 0, 0],
+                ["present_unavailable", 120_000_000, 3, -1, 0, 0, 0, 0],
+                ["present_failed", 130_000_000, 3, -1, 0, 0, 0, 0]]
+        result = self.analyze_rows(rows, fps=60)
+        signals = result["all_presentations"]["effective_signals"]
+        for name in ("display_slot_dropped", "present_unavailable", "present_failed"):
+            self.assertIn(name, signals)
+        self.assertEqual(result["isolated_drops"], 0)
+
+    def test_first_transform_task_anchors_warmup(self):
+        rows = [self.present(100 + i * 1000 / 60, i) for i in range(600)]
+        rows += [["interp_uniforms", 200_000_000, 0, -1, 1000, 0, 0, 0],
+                 ["interp_uniforms", 5_000_000_000, 0, -1, 1000, 0, 0, 12]]
+        result = self.analyze_rows(rows, warmup=1, anchor="first_transform")
+        self.assertEqual(result["timing_anchor"]["used"], "first_transform")
+        self.assertEqual(result["timing_anchor"]["start_ns"], 5_000_000_000)
+        self.assertLess(result["measured_seconds"], 4.2)
+
+    def test_missing_transform_falls_back_and_is_flagged(self):
+        rows = [self.present(100 + i * 1000 / 120, i // 2, i % 2 == 1) for i in range(40)]
+        result = self.analyze_rows(rows)
+        self.assertEqual(result["timing_anchor"]["used"], "first_real_present")
+        self.assertFalse(result["timing_anchor"]["found"])
+        self.assertEqual(result["interpolation"]["status"], "no_transform_work")
+        self.assertTrue(self.analyze_rows(rows, anchor="first_row")["timing_anchor"]["found"])
+
+    def test_benchmark_ready_anchors_auto_for_every_fps(self):
+        rows = [self.present(100 + i * 1000 / 60, i) for i in range(600)]
+        rows += [self.ready(3000), ["interp_uniforms", 5_000_000_000, 0, -1, 1000, 0, 0, 12]]
+        for fps in (60, 120, 180):
+            with self.subTest(fps=fps):
+                timing = self.analyze_rows(rows, fps=fps, warmup=1)["timing_anchor"]
+                self.assertEqual((timing["used"], timing["start_ns"], timing["found"]),
+                                 ("benchmark_ready", 3_000_000_000, True))
+
+    def test_explicit_anchors_still_select_their_event(self):
+        rows = [self.present(100 + i * 1000 / 60, i) for i in range(600)]
+        rows += [self.ready(3000), ["interp_uniforms", 5_000_000_000, 0, -1, 1000, 0, 0, 12]]
+        for anchor, start in (("first_transform", 5_000_000_000), ("first_row", 100_000_000),
+                              ("first_real_present", 100_000_000), ("benchmark_ready", 3_000_000_000)):
+            with self.subTest(anchor=anchor):
+                timing = self.analyze_rows(rows, warmup=1, anchor=anchor)["timing_anchor"]
+                self.assertEqual((timing["used"], timing["start_ns"], timing["found"]), (anchor, start, True))
+
+    def test_missing_benchmark_ready_falls_back_for_legacy_traces(self):
+        rows = [self.present(100 + i * 1000 / 60, i) for i in range(20)]
+        for anchor in ("auto", "benchmark_ready"):
+            with self.subTest(anchor=anchor):
+                timing = self.analyze_rows(rows, fps=60, anchor=anchor)["timing_anchor"]
+                self.assertEqual((timing["used"], timing["found"]), ("first_real_present", False))
+
+    def test_transform_work_status_does_not_depend_on_the_anchor(self):
+        rows = [self.ready(90), *[self.present(100 + i * 1000 / 120, i // 2, i % 2 == 1) for i in range(40)]]
+        self.assertEqual(self.analyze_rows(rows)["interpolation"]["status"], "no_transform_work")
+        rows.append(["interp_uniforms", 150_000_000, 0, -1, 1000, 0, 0, 12])
+        result = self.analyze_rows(rows)
+        self.assertEqual(result["timing_anchor"]["used"], "benchmark_ready")
+        self.assertEqual(result["interpolation"]["status"], "ok")
+
+    def test_native_period_comes_from_seal_intervals_not_present_gaps(self):
+        result = self.analyze_rows(self.pal_rows([17, 23, 19, 22, 18, 21]), fps=60)
+        self.assertEqual(result["native_period_ms"], 20)
+        self.assertEqual(result["native_period_source"], "seal")
+        self.assertEqual(result["real"]["budget_ms"], 20)
+        self.assertEqual(self.analyze_rows([self.present(100, 1)], fps=60)["native_period_source"], "default_60hz")
+
+    def test_pal_cadence_is_healthy_and_a_late_frame_is_not(self):
+        result = self.analyze_rows(self.pal_rows([20] * 20), fps=60)
+        self.assertEqual(result["real"]["stutters"], 0)
+        rows = self.pal_rows([20] * 10 + [45] + [20] * 10)
+        rows.append(["game_busy", 330_000_000, 12, -1, 19_000_000, 0, 20_000_000, 1_000_000])
+        late = self.analyze_rows(rows, fps=60)["real"]
+        self.assertEqual(late["stutters"], 1)
+        self.assertEqual(late["worst"][0]["gap_ms"], 45)
+        self.assertEqual(late["signals"], {"unattributed": 1})
+
+    def test_pal_budgets_scale_stage_thresholds(self):
+        rows = self.pal_rows([20, 45])
+        rows += [["game_busy", 164_000_000, 3, -1, 30_000_000, 0, 0, 0],
+                 ["produce", 160_000_000, 3, -1, 22_000_000, 0, 0, 0]]
+        signals = self.analyze_rows(rows, fps=60)["real"]["signals"]
+        self.assertIn("game_thread_busy_over_budget", signals)
+        self.assertNotIn("producer_wall_time", signals)
+
+    def test_pal_display_limit_without_interval_uses_the_native_period(self):
+        rows = []
+        for i in range(40):
+            at = 100 + i * 20
+            rows += [self.present(at, i + 1), self.seal(at - 1, i + 1, 20_000_000),
+                     ["display_limited", at * 1_000_000, i + 1, -1, 0, 0, 0, 1]]
+        result = self.analyze_rows(rows, fps=120)
+        self.assertEqual(result["all_presentations"]["budget_ms"], 1000 / 120)
+        self.assertEqual(result["all_presentations"]["effective_budget_ms"], 20)
+        self.assertEqual(result["all_presentations"]["effective_stutters"], 0)
+        self.assertEqual(result["interpolation"]["status"], "display_limited")
+
+    def test_oversized_csv_field_is_a_line_numbered_format_error(self):
+        text = HEADER + "present_real,1,1,0,0,0,0,0\n" + "present_real,2,2,0,0,0,0," + "9" * 200_000 + "\n"
+        with self.assertRaisesRegex(TraceFormatError, r"line 3.*field larger"):
+            self.analyze_text(text)
+
+    def test_overlong_integer_is_a_line_numbered_format_error(self):
+        with self.assertRaisesRegex(TraceFormatError, r"line 2.*'detail' is not an integer"):
+            self.analyze_text(HEADER + "present_real,1,1,0,0,0,0," + "9" * 5000 + "\n")
+
+    def test_invalid_arguments_are_rejected(self):
+        rows = [self.present(100, 1)]
+        for options in ({"warmup": float("nan")}, {"warmup": -1}, {"fps": 0}, {"fps": 60.5}):
+            with self.assertRaises(ValueError, msg=str(options)):
+                self.analyze_rows(rows, **{"fps": 60, **options})
+        with self.assertRaises(ValueError):
+            self.analyze_rows(rows, anchor="bogus")
+
+    def test_header_problems_name_line_one(self):
+        cases = {
+            "duplicate": (HEADER.strip() + ",detail\n", r"line 1.*duplicate columns \['detail'\]"),
+            "missing": (HEADER.replace(",detail", ""), r"line 1.*missing columns \['detail'\]"),
+            "foreign": ("a,b,c\n1,2,3\n", r"line 1.*missing columns.*unexpected columns \['a', 'b', 'c'\]"),
+        }
+        for name, (text, pattern) in cases.items():
+            with self.subTest(name), self.assertRaisesRegex(TraceFormatError, pattern):
+                self.analyze_text(text)
+
+    def test_row_problems_name_the_physical_line(self):
+        cases = {
+            "non-integer": ("present_real,1,1,0,0,0,0,0\npresent_real,x,2,0,0,0,0,0\npresent_real,3,3,0,0,0,0,0\n",
+                            r"line 3.*'timestamp_ns'.*'x'"),
+            "short": ("present_real,1,1\npresent_real,2,2,0,0,0,0,0\n", r"line 2.*expected 8 fields, got 3"),
+            "negative": ("present_real,1,1,0,-5,0,0,0\npresent_real,2,2,0,0,0,0,0\n",
+                         r"line 2.*'duration_ns'.*negative"),
+            "blank": ("present_real,1,1,0,0,0,0,0\n\npresent_real,2,2,0,0,0,0,0\n", r"line 3.*got 0"),
+            "empty event": (",1,1,0,0,0,0,0\n", r"line 2.*empty event"),
+        }
+        for name, (rows, pattern) in cases.items():
+            with self.subTest(name), self.assertRaisesRegex(TraceFormatError, pattern):
+                self.analyze_text(HEADER + rows)
+
+    def test_trace_without_timing_samples_is_rejected(self):
+        for name, text in {"lost only": HEADER + "trace_lost,0,0,-1,0,0,0,3\n", "header only": HEADER}.items():
+            with self.subTest(name), self.assertRaisesRegex(TraceFormatError, "no timing samples"):
+                self.analyze_text(text)
+        for name, text in {"empty": "", "partial header": "event,timestamp_ns"}.items():
+            with self.subTest(name), self.assertRaisesRegex(TraceFormatError, "no complete header"):
+                self.analyze_text(text)
+
+    def test_truncated_final_row_is_ignored_even_when_malformed(self):
+        result = self.analyze_text(HEADER + "present_real,100,1,0,0,0,0,0\npresent_real,2x")
+        self.assertTrue(result["truncated_final_row"])
+        self.assertEqual(result["events"]["present_real"], 1)
+
+    def test_trace_state_reports_readiness(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "trace.csv"
+            path.write_bytes(b"event,timestamp_ns")
+            self.assertIsNone(trace_state(path))
+            path.write_bytes((HEADER + "present_real,5,1,0,0,0,0,0\ninterp_uniforms,9,0,-1,1,0,0,0\n"
+                              "interp_uniforms,11,0,-1,1,0,0,4\ndisplay_limited,12,1,-1,0,0,0,1\n"
+                              "present_real,13").encode())
+            self.assertEqual(trace_state(path), {"benchmark_ready_ns": None, "real_present_ns": 5,
+                                                 "transform_ns": 11, "display_limited": True})
+            path.write_bytes((HEADER + "present_real,5,1,0,0,0,0,0\nbenchmark_ready,7,0,-1,0,0,0,0\n"
+                              "benchmark_ready,8,0,-1,0,0,0,0\n").encode())
+            self.assertEqual(trace_state(path)["benchmark_ready_ns"], 7)
+
+
+if __name__ == "__main__":
+    unittest.main()

runtime/tools/test_benchmark_stutters.py

@@ -0,0 +1,209 @@
+from pathlib import Path
+from contextlib import nullcontext
+import io
+import os
+import subprocess
+import tempfile
+import tomllib
+import unittest
+from unittest import mock
+from types import SimpleNamespace
+
+import benchmark_stutters
+from benchmark_stutters import (ConfigError, SourceConfig, isolated_config_text, load_source_config,
+                                interpolation_helpers, parse_helper, parse_list, quoted, stop_process,
+                                wait_ready)
+from analyze_stutters import COLUMNS
+
+HEADER = ",".join(COLUMNS) + "\n"
+
+
+class QuotedTests(unittest.TestCase):
+    def test_unicode_and_control_characters_roundtrip_through_toml(self):
+        for value in ["C:\\Users\\Jürgen\\日本語\\Wii", 'a"b', "tab\there", "del\x7fchar", "\u2028 end", ""]:
+            with self.subTest(value=value):
+                self.assertEqual(tomllib.loads(f"x = {quoted(value)}")["x"], value)
+
+    def test_isolated_config_roundtrips_paths_and_video_settings(self):
+        source = SourceConfig(Path("C:/Spiele/Mario Kart é"), Path("C:/RR/日本"), None, "vulkan")
+        text = isolated_config_text(source, Path("C:/Out/ünï/NAND"), 120, 1.5, True)
+        config = tomllib.loads(text)
+        self.assertEqual(config["paths"]["dvd_root"], str(source.dvd_root))
+        self.assertEqual(config["paths"]["retro_rewind_root"], str(source.retro_rewind_root))
+        self.assertEqual(config["paths"]["nand_root"], str(Path("C:/Out/ünï/NAND")))
+        self.assertEqual(config["video"]["frame_interpolation_fps"], 120)
+        self.assertEqual(config["video"]["resolution_multiplier"], 1.5)
+        self.assertEqual(config["video"]["graphics_api"], "vulkan")
+        self.assertTrue(config["video"]["vsync"])
+        self.assertEqual(config["network"], {"enabled": False})
+
+    def test_optional_paths_are_omitted(self):
+        config = tomllib.loads(isolated_config_text(SourceConfig(None, None, None, "auto"), Path("N"), 0, 1, False))
+        self.assertEqual(config["paths"], {"nand_root": "N"})
+
+
+class ParseListTests(unittest.TestCase):
+    def test_valid_lists_keep_order(self):
+        self.assertEqual(parse_list("4, 0,2", "cpus", int), [4, 0, 2])
+        self.assertEqual(parse_list("auto,2", "helpers", parse_helper, ("auto", "2")), ["auto", "2"])
+
+    def test_rejects_duplicates_including_equivalent_spellings(self):
+        for text, parse in [("0,0", int), ("1,01", int)]:
+            with self.subTest(text), self.assertRaisesRegex(ValueError, "duplicate"):
+                parse_list(text, "cpus", parse)
+        with self.assertRaisesRegex(ValueError, "duplicate"):
+            parse_list("auto,0,00", "helpers", parse_helper, ("auto", "0"))
+
+    def test_rejects_malformed_and_unsupported_entries(self):
+        for text in ("1,", "", "a"):
+            with self.subTest(text), self.assertRaisesRegex(ValueError, "invalid entry"):
+                parse_list(text, "cpus", int)
+        with self.assertRaisesRegex(ValueError, "unsupported"):
+            parse_list("0,60", "fps", int, (0, 120, 180))
+        with self.assertRaisesRegex(ValueError, "unsupported"):
+            parse_list("7", "helpers", parse_helper, ("auto", "0"))
+
+
+class SourceConfigTests(unittest.TestCase):
+    def load(self, text, create=()):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            for name in create:
+                (root / name).mkdir()
+            path = root / "Config.toml"
+            path.write_text(text.replace("{root}", quoted(root)[1:-1]), encoding="utf-8")
+            return load_source_config(path), root
+
+    def test_relative_paths_resolve_against_the_config_directory(self):
+        source, root = self.load('[paths]\ndvd_root = "game"\nnand_root = "nand"\n', create=("game", "nand"))
+        self.assertEqual(source.dvd_root, (root / "game").resolve())
+        self.assertEqual(source.nand_root, (root / "nand").resolve())
+        self.assertEqual(source.graphics_api, "auto")
+
+    def test_unicode_directory_names_roundtrip(self):
+        name = "Wii é日本"
+        source, root = self.load('[paths]\ndvd_root = "{root}\\\\' + name + '"\n', create=(name,))
+        self.assertEqual(source.dvd_root, root / name)
+
+    def test_empty_and_missing_values_are_unset(self):
+        source, _ = self.load('[paths]\ndvd_root = ""\n')
+        self.assertIsNone(source.dvd_root)
+        self.assertIsNone(source.nand_root)
+
+    def test_odd_config_types_and_values_are_rejected(self):
+        cases = {
+            "paths not table": ('paths = "x"\n', "\\[paths\\] must be a table"),
+            "video not table": ("video = 3\n", "\\[video\\] must be a table"),
+            "path not string": ("[paths]\ndvd_root = 5\n", "paths.dvd_root must be a string"),
+            "path list": ('[paths]\nnand_root = ["a"]\n', "paths.nand_root must be a string"),
+            "missing path": ('[paths]\ndvd_root = "nope"\n', "does not exist"),
+            "api not string": ("[video]\ngraphics_api = 3\n", "graphics_api"),
+            "api unsupported": ('[video]\ngraphics_api = "dx11"\n', "graphics_api"),
+            "invalid toml": ("[paths\n", "invalid TOML"),
+        }
+        for name, (text, pattern) in cases.items():
+            with self.subTest(name), self.assertRaisesRegex(ConfigError, pattern):
+                self.load(text)
+
+    def test_nand_root_must_be_a_directory(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / "nand").write_text("file")
+            (root / "Config.toml").write_text('[paths]\nnand_root = "nand"\n')
+            with self.assertRaisesRegex(ConfigError, "not a directory"):
+                load_source_config(root / "Config.toml")
+
+    def test_missing_config_file_is_a_config_error(self):
+        with tempfile.TemporaryDirectory() as directory, self.assertRaisesRegex(ConfigError, "cannot read config"):
+            load_source_config(Path(directory) / "missing.toml")
+
+
+class HelperCountTests(unittest.TestCase):
+    def test_limited_runs_use_the_requested_cpu_capacity(self):
+        for count, expected in ((1, 0), (2, 0), (3, 2), (4, 3), (12, 6)):
+            with self.subTest(count=count):
+                self.assertEqual(interpolation_helpers(count, "auto"), expected)
+
+    def test_full_cpu_runs_keep_the_runtime_default(self):
+        self.assertIsNone(interpolation_helpers(0, "auto"))
+
+    def test_explicit_helper_controls_are_preserved(self):
+        self.assertEqual(interpolation_helpers(4, "6"), 6)
+        self.assertEqual(interpolation_helpers(4, "0"), 0)
+
+
+class WaitReadyTests(unittest.TestCase):
+    @staticmethod
+    def running():
+        return mock.Mock(**{"poll.return_value": None})
+
+    def test_requires_benchmark_ready_regardless_of_other_events(self):
+        with tempfile.TemporaryDirectory() as directory:
+            trace = Path(directory) / "trace.csv"
+            trace.write_text(HEADER + "present_real,5,1,0,0,0,0,0\ninterp_uniforms,9,0,-1,1,0,0,4\n")
+            with mock.patch.object(benchmark_stutters, "STARTUP_TIMEOUT", 0), \
+                    self.assertRaisesRegex(RuntimeError, "no benchmark_ready"):
+                wait_ready(self.running(), trace, "case")
+            trace.write_text(HEADER + "benchmark_ready,7,0,-1,0,0,0,0\n")
+            self.assertEqual(wait_ready(self.running(), trace, "case")["benchmark_ready_ns"], 7)
+
+
+class StopProcessTests(unittest.TestCase):
+    def test_kills_after_terminate_times_out(self):
+        process = mock.Mock(**{"poll.return_value": None})
+        process.wait.side_effect = [subprocess.TimeoutExpired("game", 10), 0]
+        stop_process(process, "case")
+        process.terminate.assert_called_once()
+        process.kill.assert_called_once()
+
+    def test_unkillable_process_is_a_clear_failure(self):
+        process = mock.Mock(**{"poll.return_value": None})
+        process.wait.side_effect = subprocess.TimeoutExpired("game", 10)
+        with self.assertRaisesRegex(RuntimeError, "did not exit after terminate and kill"):
+            stop_process(process, "case")
+
+
+class TraceCompletenessTests(unittest.TestCase):
+    def run_case(self, tail, incomplete):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            exe, config, output = root / "game.exe", root / "Config.toml", root / "out"
+            exe.write_bytes(b"")
+            config.write_text("")
+            trace = HEADER + "benchmark_ready,1,0,-1,0,0,0,0\n"
+            trace += "".join(f"present_real,{1 + i * 16_666_667},{i + 1},0,0,0,16666667,0\n"
+                             for i in range(80)) + tail
+            process = mock.Mock(**{"poll.return_value": None})
+            process.wait.side_effect = [subprocess.TimeoutExpired("game", 1), 0]
+
+            def launch(*args, env, **kwargs):
+                Path(env["MKW_FRAME_TRACE"]).write_text(trace)
+                return process
+
+            arguments = ["benchmark", "--exe", str(exe), "--config", str(config), "--output", str(output),
+                         "--cpus", "1", "--fps", "0", "--helpers", "0", "--warmup", "0", "--seconds", "1"]
+            platform = SimpleNamespace(name="nt", environ=os.environ, cpu_count=os.cpu_count)
+            expectation = self.assertRaisesRegex(RuntimeError, "incomplete trace") if incomplete else nullcontext()
+            with mock.patch("sys.argv", arguments), mock.patch("sys.stdout", new_callable=io.StringIO), \
+                    mock.patch.object(benchmark_stutters, "os", platform), \
+                    mock.patch.object(benchmark_stutters, "cpu_topology",
+                                      return_value=(mock.Mock(), 1, 1, [1])), \
+                    mock.patch.object(subprocess, "CREATE_NO_WINDOW", 0, create=True), \
+                    mock.patch.object(subprocess, "Popen", side_effect=launch), expectation:
+                benchmark_stutters.main()
+            process.terminate.assert_called_once()
+            self.assertEqual((output / "summary.json").exists(), not incomplete)
+            self.assertEqual((output / "r0-cpu1-fps0-helpers0.json").exists(), not incomplete)
+
+    def test_complete_trace_writes_results(self):
+        self.run_case("", incomplete=False)
+
+    def test_incomplete_traces_do_not_write_results(self):
+        for tail in ("trace_lost,0,0,-1,0,0,0,3\n", "present_real,",
+                     "trace_lost,0,0,-1,0,0,0,3\npresent_real,"):
+            with self.subTest(tail=tail):
+                self.run_case(tail, incomplete=True)
+
+
+if __name__ == "__main__":
+    unittest.main()