benchmarks: halve the io figures of pre-fix mini rows in the frozen copy (#555)

benchmarks/experiments/PUBLISHING.md

@@ -26,7 +26,7 @@ The Engines page of the documentation site (humemai/dbbench, `docs/engines.md`)
 
 The publish side renders rows stamped `instrument: 2026-11`, and nothing else. The freeze, the exporter, the gates, `land_stage.py` and `publish_results_asset.py` refuse a row of any other instrument, or a row with no stamp, with a message that says the tooling publishes 2026-11 rows only (`publish_instrument.refuse_other_instruments`; DECISIONS #196). If `BENCH_INSTRUMENT` is set to anything else, they refuse too; unset and `2026-11` mean the same thing here.
 
-There is one freeze (`results/runs_paper.csv`), one payload (`results/web_benchmarks.json`) and one generated set (`results/generated`: `tables/`, `figures/`, `withheld_recall.json`, `GATE_STATUS.txt`). `BENCH_RESULTS_DIR` moves the whole `results/` tree and exists for tests only; a publish never sets it. The proof that the publish side works on 2026-11 rows is the fixture `fixtures/publish_2611` and two tests: `test_publish_2611_e2e.py` freezes and exports it end to end, and `test_publish_gates_2611.py` runs every gate below on the payload that comes out and requires each to pass. The fixture rows are copies of September and October measurements, so `build_fixture.realistic_2611` gives them what a 2026-11 row has and they lack (the override stamps, one pin, the roster of arms, the deep10m campaign rows behind the overlay files); its docstring names each patch and its reason, and `python build_fixture.py --patch` re-applies them to the committed rows. A gate that fails on the fixture is either right (the fixture row is one no 2026-11 run could write: fix the fixture) or still encodes a retired instrument (fix the gate).
+There is one freeze (`results/runs_paper.csv`), one payload (`results/web_benchmarks.json`) and one generated set (`results/generated`: `tables/`, `figures/`, `withheld_recall.json`, `io_stat_correction.json`, `GATE_STATUS.txt`). `BENCH_RESULTS_DIR` moves the whole `results/` tree and exists for tests only; a publish never sets it. The proof that the publish side works on 2026-11 rows is the fixture `fixtures/publish_2611` and two tests: `test_publish_2611_e2e.py` freezes and exports it end to end, and `test_publish_gates_2611.py` runs every gate below on the payload that comes out and requires each to pass. The fixture rows are copies of September and October measurements, so `build_fixture.realistic_2611` gives them what a 2026-11 row has and they lack (the override stamps, one pin, the roster of arms, the deep10m campaign rows behind the overlay files); its docstring names each patch and its reason, and `python build_fixture.py --patch` re-applies them to the committed rows. A gate that fails on the fixture is either right (the fixture row is one no 2026-11 run could write: fix the fixture) or still encodes a retired instrument (fix the gate).
 
 Every ArcadeDB row measured under another pin is superseded, embedded as well as served; no ArcadeDB row is carried forward, and a comparator row carried forward is stamped `carried_forward_from` and `carried_forward_reason`.
 
@@ -117,6 +117,7 @@ The allowlist says WHICH files are tracked; this says why, and what to re-run to
 | `runs_paper.csv` | `make_paper_tables.py` (the freeze) | every gate, the exporter, the release asset |
 | `web_benchmarks.json` | `export_web.py` | `page_check`, `version_consistency_check`, `refresh_web_page` |
 | `generated/tables/*`, `withheld_recall.json` | `make_paper_tables.py` | `page_check`, `provenance_check`, the exporter's withheld-cell notes |
+| `generated/io_stat_correction.json` | `make_paper_tables.py` (the freeze, via `io_stat_correction.py`) | nothing reads it; it records which frozen rows had their io figures halved (the mini's stacked block devices counted twice before #555) and their raw values. A row is corrected unless it carries `io_stat_dedup: 1`, which the fixed runner stamps; the hosts are in `io_stat_fix.json` |
 | `generated/memo_bottlenecks.html` | `memo_bottlenecks.py` | the maintainers' memo, not the page |
 | `sparse_cliff.jsonl` | `sparse_cliff_probe.py`, one shot | `make_paper_figures.f3_sparse_perquery`; f3 is not built (PAGE-SPEC.md section 3) |
 | `e3_q17/*.json`, `ingest_ab/ab_r*.json`, `tentag/tentag_ab.json` | `e3_recovery.py`, `async_ingest_probe.py`, `ts_stride_probe.py`, one shot each | `claims_check.py` (the recovery table is planned, not built) |

benchmarks/experiments/io_stat_correction.py

@@ -0,0 +1,88 @@
+"""Freeze-time correction for the io.stat double count (humemai/arcadedb-embedded-python#555). Pure functions, no file writes.
+
+THE DEFECT. Until the fix, runner.read_io_stat and memory_probe.parse_io_stat summed every device line of a container's cgroup v2
+io.stat. On the mini the file lists 259:0 (NVMe) and 252:0 (a device-mapper device stacked on it) with identical bytes, plus an
+unrelated 8:0 with 0 bytes, so every figure derived from it was recorded at twice its value. The fixed harness counts a stacked
+device once and says so: its runner stamps `io_stat_dedup: 1` into every row it writes.
+
+THE CORRECTION. The raw rows (runs.jsonl) are never edited. The freeze (make_paper_tables.freeze_paper_rows) calls `correct_rows`
+on the rows it writes to runs_paper.csv. A row that ran on a host named in `hosts` (the row's `bench_host`) and does NOT carry
+`io_stat_dedup == 1` predates the fix: every io figure in the frozen copy is halved, and the frozen row's `io_stat_dedup` reads
+"halved" (a stamped row keeps its 1, a row of another host keeps whatever it had, usually blank). `report` names each corrected
+row with its raw values, so the recorded value is recoverable from the freeze report (results/generated/io_stat_correction.json)
+and from runs.jsonl.
+
+WHICH ROWS PREDATE THE FIX. The row says so itself: the stamp is written by the same commit as the fix, so a row without it was
+written without the fix, and no cutoff time has to be set later. A stamp that is present but is not 1 raises: a row the rule
+cannot read is never silently counted either way.
+"""
+import json
+import re
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+FIX_FILE = HERE / "io_stat_fix.json"
+MARKER = "io_stat_dedup"
+HALVED = "halved"
+
+# Every field the harness derives from io.stat: io_read_mib, io_write_mib, io_{read,write}_mib_sum,
+# {client,server,server_container,sidecar_<name>}_io_{read,write}_mib, and the memory probe's io_<phase>_{read,write}_mib.
+IO_FIELD = re.compile(r"^(?:[a-z0-9_]+_)?io_(?:read|write)_mib(?:_sum)?$|^io_[a-z0-9_]+_(?:read|write)_mib$")
+
+
+def is_io_field(name):
+    return bool(IO_FIELD.match(str(name)))
+
+
+def load_fix(path=FIX_FILE):
+    """The fix record: {"hosts": [hosts whose io.stat stacks block devices]}."""
+    with open(path) as fh:
+        fix = json.load(fh)
+    if not fix.get("hosts"):
+        raise ValueError(f"{path}: no hosts")
+    return fix
+
+
+def predates_fix(row):
+    """True when the row was written without the fix (no stamp), False when stamped 1; ValueError for any other stamp."""
+    v = row.get(MARKER)
+    if v is None or v == "":
+        return True
+    if v == 1 and not isinstance(v, (str, bool)):
+        return False
+    raise ValueError(f"{MARKER}={v!r} on run_id {row.get('run_id')!r}: the stamp is 1 or absent")
+
+
+def correct_row(row, hosts):
+    """(row or a corrected copy, {field: raw value} of what was halved). The input dict is never changed."""
+    if row.get("bench_host") not in hosts or not predates_fix(row):
+        return row, {}
+    out, raw = dict(row), {}
+    for k, v in row.items():
+        if is_io_field(k) and isinstance(v, (int, float)) and not isinstance(v, bool):
+            raw[k] = v
+            out[k] = round(v / 2, 2)   # a recorded value is 0.1 MiB-quantised, so its half is exact at 0.05
+    if raw:
+        out[MARKER] = HALVED
+        return out, raw
+    return row, {}
+
+
+def correct_rows(rows, fix=None):
+    """(rows, report) for the freeze: `rows` in input order with the io figures of stacked-device rows halved, `report` the record
+    of what changed, fit for json.dump (rule, counts, and per corrected row its identity and raw values)."""
+    fix = load_fix() if fix is None else fix
+    hosts = tuple(fix["hosts"])
+    out, corrected, stamped = [], [], 0
+    for r in rows:
+        c, raw = correct_row(r, hosts)
+        out.append(c)
+        stamped += r.get("bench_host") in hosts and r.get(MARKER) == 1
+        if raw:
+            corrected.append({"run_id": r.get("run_id"), "lane": r.get("lane"), "backend": r.get("backend"),
+                              "scale": r.get("scale"), "rep": r.get("rep"), "raw": raw})
+    report = {"issue": "humemai/arcadedb-embedded-python#555",
+              "rule": f"io figures halved in the frozen rows of hosts that stack block devices unless the row carries {MARKER} == 1",
+              "hosts": list(hosts), "rows_frozen": len(rows), "rows_stamped_untouched": stamped,
+              "rows_corrected": len(corrected), "corrected": corrected}
+    return out, report

benchmarks/experiments/io_stat_fix.json

@@ -0,0 +1,4 @@
+{
+ "issue": "humemai/arcadedb-embedded-python#555",
+ "hosts": ["mini"]
+}

benchmarks/experiments/make_paper_tables.py

@@ -88,6 +88,7 @@ def _commit_matches(row_commit):
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 import publish_instrument as _PI
+import io_stat_correction as _IOC
 RESULTS = str(_PI.results_dir())
 
 # Every published cell is N=5. A row outside 1..5 is a probe, not a repetition;
@@ -229,6 +230,17 @@ def _write_withheld_recall():
         fh.write("\n")
 
 
+def _write_io_stat_correction(report):
+    """Sidecar beside the freeze: which rows had their io figures halved, with the raw values (io_stat_correction.py)."""
+    out = os.path.join(GENERATED, "io_stat_correction.json")
+    os.makedirs(os.path.dirname(out), exist_ok=True)
+    with open(out, "w") as fh:
+        json.dump(report, fh, indent=1)
+        fh.write("\n")   # the end-of-file hook wants it; see _write_withheld_recall
+    print(f"io.stat correction: {report['rows_corrected']} of {report['rows_frozen']} frozen rows halved, "
+          f"{report['rows_stamped_untouched']} stamped io_stat_dedup=1 and left as recorded")
+
+
 def _assert_lanes_registered():
     """Every lane the runner can run is registered here, or it says which.
 
@@ -1325,6 +1337,10 @@ def freeze_paper_rows(rows):
     recompute every T2/T3 cell from this file alone.
     """
     import csv as _csv
+    # THE IO.STAT DOUBLE COUNT (#555): rows of a host that stacks block devices, written before the fixed harness, carry io figures at
+    # twice their value. They are halved in the FROZEN copy only; runs.jsonl keeps the raw rows and the sidecar the raw values.
+    rows, io_report = _IOC.correct_rows(rows)
+    _write_io_stat_correction(io_report)
     cols = []
     for r in rows:
         for k in r:

benchmarks/experiments/page_check.py

@@ -779,6 +779,10 @@ def main() -> int:
      "which HTTP client an ArcadeDB served arm ran with (CAMPAIGN 7 row 72, lean_http.py): a condition of the served arms, read from "
      "the session that ran, never a column. The e4 decomposition's row lists the two clients it measured (`arcadedb_http_clients`); its table "
      "names the client in each HTTP column instead (export_web._e4_arms)"),
+    (r"^io_stat_dedup$",
+     "provenance of the io figures (#555): 1 when the harness that wrote the row counted a stacked block device once, \"halved\" "
+     "when the freeze halved the recorded io figures of a row that predates the fix (io_stat_correction.py, raw values in "
+     "generated/io_stat_correction.json), blank otherwise; never a column"),
     (r"^jpype_version$",
      "which JPype the process that ran the cell had loaded (CAMPAIGN 7 row 73, bench_common.jpype_version): a condition of the embedded "
      "ArcadeDB arms, read from the process (`jpype.__version__`), empty for every arm that never imports it, never a column"),

benchmarks/experiments/provenance_check.py

@@ -690,7 +690,10 @@ def check_schema_homogeneity(rows):
                      # logged, captured only once the capture worked (BUGS F1-F4).
                      # A family split on these is a split in
                      # instrumentation history, not in what was measured.
-                     "graph_build_cache", "graph_build_access", "driver_")
+                     "graph_build_cache", "graph_build_access", "driver_",
+                     # PROVENANCE, not a measurement: the harness that wrote the row counted a stacked device once (#555),
+                     # so a cell whose reps straddle the fix legitimately splits on it.
+                     "io_stat_dedup")
     def measured(r):
         return {k for k, v in r.items()
                 if isinstance(v, (int, float))

benchmarks/experiments/test_io_stat_correction.py

@@ -0,0 +1,129 @@
+"""The freeze halves the io figures of pre-fix rows from a host that stacks block devices (#555), and changes nothing else.
+
+Run with `python -m pytest test_io_stat_correction.py -q` from this directory. What is held:
+
+  * the field rule takes every io.stat-derived figure (io_*_mib, the _sum pair, the per-side and per-phase ones) and nothing else
+    that merely looks like it (a memory or disk figure, a thread count);
+  * a mini row without the `io_stat_dedup` stamp predates the fix and is halved (and says "halved"), a mini row stamped 1 is never
+    halved, a row of another host is not, and a stamp that is neither absent nor 1 raises;
+  * the input rows are never changed, so the tables built from them and runs.jsonl keep the raw values;
+  * the freeze writes the halved values to runs_paper.csv and the raw ones to the sidecar report, the committed fix file parses,
+    and the stamp is declared in page_check.NOT_PRINTED and ignored by provenance_check's schema-homogeneity check.
+"""
+import copy
+import csv
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+import io_stat_correction as C  # noqa: E402
+
+MINI = {"run_id": "r1", "bench_host": "mini", "lane": "l1", "backend": "arcadedb", "scale": "tiny", "rep": 1,
+        "io_read_mib": 0.2, "io_write_mib": 16590.4, "io_read_mib_sum": 0.2, "io_write_mib_sum": 16590.4,
+        "client_io_write_mib": 8.0, "server_io_read_mib": 1.0, "server_container_io_write_mib": 6.0,
+        "sidecar_x_io_write_mib": 4.0, "io_load_write_mib": 100.0,
+        "peak_mib_sum": 123.0, "disk_mb_sum": 50.0, "redis_io_threads": 4, "io_note": "text", "wall_s": 9.0}
+STAMPED = dict(MINI, run_id="r2", io_stat_dedup=1)
+FIX = {"hosts": ["mini"]}
+
+
+def test_field_rule():
+    for k in ("io_read_mib", "io_write_mib", "io_read_mib_sum", "io_write_mib_sum", "client_io_read_mib", "server_io_write_mib",
+              "server_container_io_write_mib", "sidecar_x_io_read_mib", "io_load_write_mib", "io_query_read_mib"):
+        assert C.is_io_field(k), k
+    for k in ("peak_mib_sum", "disk_mb_sum", "server_disk_rw_mb", "redis_io_threads", "io_note", "ingest_mib", "wall_s",
+              "io_read_mib_extra", "io_stat_dedup"):
+        assert not C.is_io_field(k), k
+
+
+def test_unstamped_mini_row_is_halved_and_the_rest_is_untouched():
+    before = copy.deepcopy(MINI)
+    out, raw = C.correct_row(MINI, ("mini",))
+    assert MINI == before, "the input row must not change"
+    assert out["io_write_mib"] == 8295.2 and out["io_write_mib_sum"] == 8295.2 and out["io_read_mib"] == 0.1
+    assert out["client_io_write_mib"] == 4.0 and out["server_container_io_write_mib"] == 3.0 and out["io_load_write_mib"] == 50.0
+    assert raw["io_write_mib"] == 16590.4 and "peak_mib_sum" not in raw and len(raw) == 9
+    assert out["io_stat_dedup"] == "halved"
+    assert {k: v for k, v in out.items() if not C.is_io_field(k) and k != "io_stat_dedup"} == \
+        {k: v for k, v in MINI.items() if not C.is_io_field(k)}
+    assert C.correct_row(dict(MINI, io_stat_dedup=None), ("mini",))[1] and C.correct_row(dict(MINI, io_stat_dedup=""), ("mini",))[1]
+
+
+def test_stamped_row_is_never_halved():
+    out, raw = C.correct_row(STAMPED, ("mini",))
+    assert raw == {} and out is STAMPED and out["io_write_mib"] == 16590.4 and out["io_stat_dedup"] == 1
+
+
+def test_rows_not_corrected():
+    assert C.correct_row(dict(MINI, bench_host="laptop"), ("mini",))[1] == {}
+    assert C.correct_row({k: v for k, v in MINI.items() if k != "bench_host"}, ("mini",))[1] == {}
+    assert C.correct_row(dict(MINI, bench_host="laptop", io_stat_dedup="junk"), ("mini",))[1] == {}   # not our host: not read
+    # None, text and bool pass through unhalved
+    odd = C.correct_row(dict(MINI, io_write_mib=None, io_read_mib="n/a", io_write_mib_sum=True), ("mini",))[1]
+    assert odd.keys() == {"io_read_mib_sum", "client_io_write_mib", "server_io_read_mib", "server_container_io_write_mib",
+                          "sidecar_x_io_write_mib", "io_load_write_mib"}
+    assert C.correct_row(dict(MINI, io_read_mib=None, io_write_mib=None, io_read_mib_sum=None, io_write_mib_sum=None,
+                              client_io_write_mib=None, server_io_read_mib=None, server_container_io_write_mib=None,
+                              sidecar_x_io_write_mib=None, io_load_write_mib=None), ("mini",))[0]["bench_host"] == "mini"
+
+
+def test_malformed_stamp_raises():
+    for bad in (2, 0, "1", True, "halved", 1.5):
+        try:
+            C.correct_row(dict(MINI, io_stat_dedup=bad), ("mini",))
+        except ValueError:
+            continue
+        raise AssertionError(f"stamp {bad!r} must raise")
+
+
+def test_correct_rows_report_names_each_corrected_row_with_its_raw_values():
+    rows = [MINI, STAMPED, dict(MINI, run_id="r3", bench_host="laptop")]
+    out, report = C.correct_rows(rows, FIX)
+    assert [r["run_id"] for r in out] == ["r1", "r2", "r3"]
+    assert out[0]["io_write_mib"] == 8295.2 and out[1]["io_write_mib"] == 16590.4 and out[2]["io_write_mib"] == 16590.4
+    assert report["rows_frozen"] == 3 and report["rows_corrected"] == 1 and report["rows_stamped_untouched"] == 1
+    assert report["corrected"][0]["run_id"] == "r1" and report["corrected"][0]["raw"]["io_write_mib"] == 16590.4
+    json.dumps(report)
+
+
+def test_committed_fix_file_parses():
+    assert C.load_fix()["hosts"] == ["mini"]
+
+
+def test_the_stamp_is_declared_to_the_gates():
+    import page_check
+    assert page_check._not_printed_reason("io_stat_dedup")
+    src = (HERE / "provenance_check.py").read_text()
+    assert '"io_stat_dedup")' in src.split("IGNORE_PREFIX = (")[1].split("def measured")[0]
+
+
+def test_freeze_writes_halved_rows_and_the_raw_values_to_the_sidecar():
+    import make_paper_tables as M
+    rows = [dict(MINI), dict(STAMPED, rep=2), dict(MINI, run_id="r3", rep=3, bench_host="laptop")]
+    before = copy.deepcopy(rows)
+    with tempfile.TemporaryDirectory() as d:
+        saved = (M.RESULTS, M.GENERATED)
+        M.RESULTS, M.GENERATED = d, str(Path(d) / "generated")
+        try:
+            M.freeze_paper_rows(rows)
+        finally:
+            M.RESULTS, M.GENERATED = saved
+        frozen = {r["run_id"]: r for r in csv.DictReader(open(Path(d) / "runs_paper.csv"))}
+        report = json.load(open(Path(d) / "generated" / "io_stat_correction.json"))
+    assert rows == before, "freeze_paper_rows must not change the rows the tables are built from"
+    assert float(frozen["r1"]["io_write_mib"]) == 8295.2 and frozen["r1"]["io_stat_dedup"] == "halved"
+    assert float(frozen["r2"]["io_write_mib"]) == 16590.4 and frozen["r2"]["io_stat_dedup"] == "1"
+    assert float(frozen["r3"]["io_write_mib"]) == 16590.4 and frozen["r3"]["io_stat_dedup"] == ""
+    assert frozen["r1"]["peak_mib_sum"] == "123.0" and set(frozen["r1"]) == set(frozen["r2"])
+    assert report["rows_corrected"] == 1 and report["corrected"][0]["raw"]["io_write_mib"] == 16590.4
+
+
+if __name__ == "__main__":
+    for name, fn in sorted(globals().items()):
+        if name.startswith("test_") and callable(fn):
+            fn()
+            print("ok", name)