047: "the leaf is the commit" (feedback 009) set up before it runs: today's store measured as the base (baseline.py, 046's shim and journal: list bytes a message, the log and images, at 1 / 16 / 256 messages a commit: entries' appends 65.5 / 15.7 / 12.4, the bank's transfers 122 / 35.6 / 30.1, the tree apart), and a byte-accounting model (model.py) of commits writing the leaves they changed in memory's own bytes (leaf 296 to 1,320 B, inner node 680 B, cow; tail appends in place past every kept snapshot's count; inplace, which may not be, to show why), its bars written before the run (appends at most today's, sparse updates at most twice, no write a kept snapshot sees) and what a pass for appends with a miss for updates would mean (two units, not one); 046's Run takes pure-run's arguments; 045's keyed file says 24 refused, not the 26 its commit said

5 files changed, 288 insertions, 3 deletions

diff --git a/research/experiments/045-proof-compiler/2026-10-19-keyed.jsonl b/research/experiments/045-proof-compiler/2026-10-19-keyed.jsonl
index 0204b0c..d5215de 100644
--- a/research/experiments/045-proof-compiler/2026-10-19-keyed.jsonl
+++ b/research/experiments/045-proof-compiler/2026-10-19-keyed.jsonl
@@ -1,4 +1,4 @@
-{"what": "the app with ranges keyed by last and a range found by a seek (variants/linebreak-keyed2.patch: first(x in ranges where x.last >= code), its item read for class; ranges[key] for pi, pf, east, pictographic): the executor answers all 18,672 lines (cost.c: 28,061 a point keyed, 27,934 keyed2); the hand rungs keyed_* emit what it says (the seek by each level guide keys and the leaf key column, nothing copied); every rung and end answers all lines; with LB18 wrong all 26 hand rungs and ends refused"}
+{"what": "the app with ranges keyed by last and a range found by a seek (variants/linebreak-keyed2.patch: first(x in ranges where x.last >= code), its item read for class; ranges[key] for pi, pf, east, pictographic): the executor answers all 18,672 lines (cost.c: 28,061 a point keyed, 27,934 keyed2); the hand rungs keyed_* emit what it says (the seek by each level guide keys and the leaf key column, nothing copied); every rung and end answers all lines; with LB18 wrong all 24 hand rungs and ends refused (20 rungs, 4 ends; the commit that brought this file said 26)"}
 {"rung":"exact","lines":18672,"code_points":61285,"rounds":5,"instructions_per_point":8658.5,"cycles_per_point":3697.8,"answers":"the test's, and exact's"}
 {"rung":"integer","lines":18672,"code_points":61285,"rounds":5,"instructions_per_point":7908.5,"cycles_per_point":3252.8,"answers":"the test's, and exact's"}
 {"rung":"stored_flat","lines":18672,"code_points":61285,"rounds":5,"instructions_per_point":6261.9,"cycles_per_point":2464.7,"answers":"the test's, and exact's"}
diff --git a/research/experiments/046-commit-replica/run.py b/research/experiments/046-commit-replica/run.py
index e31ff50..38c24f1 100644
--- a/research/experiments/046-commit-replica/run.py
+++ b/research/experiments/046-commit-replica/run.py
@@ -86,7 +86,7 @@ def stores_differ(a, b):
 class Run:
     """A follower, a shipper and an owner, started in that order on fresh stores."""
 
-    def __init__(self, keys, every=0, shim=True, module='entries'):
+    def __init__(self, keys, every=0, shim=True, module='entries', extra=()):
         self.owner_dir, self.follower_dir = WORK / 'owner-store', WORK / 'follower-store'
         for d in (self.owner_dir, self.follower_dir):
             shutil.rmtree(d, ignore_errors=True)
@@ -112,7 +112,7 @@ class Run:
             self.paused = os.open(WORK / 'paused', os.O_RDWR)  # held open both ways: no open waits for the other end
             self.resume = os.open(WORK / 'resume', os.O_RDWR)
             env.update(SHIM_PAUSE_EVERY=str(every), SHIM_PAUSE_DIR=str(WORK))
-        self.owner = subprocess.Popen([str(RUN), str(WORK / f'{module}.pmod'), '--views', '--store', str(self.owner_dir)],
+        self.owner = subprocess.Popen([str(RUN), str(WORK / f'{module}.pmod'), '--views', '--store', str(self.owner_dir), *extra],
                                       stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                                       env=env, bufsize=1)
         self.done = 0  # messages the owner said done (each line comes after its save, so after the follower's ack)
diff --git a/research/experiments/047-leaf-commit/2026-10-19-baseline.jsonl b/research/experiments/047-leaf-commit/2026-10-19-baseline.jsonl
new file mode 100644
index 0000000..60ca0c8
--- /dev/null
+++ b/research/experiments/047-leaf-commit/2026-10-19-baseline.jsonl
@@ -0,0 +1,6 @@
+{"workload": "entries", "messages": 20000, "batch": 1, "done": 20000, "bytes_per_message": {"compaction": 347.8, "image": 0.0, "log": 65.4, "tree": 404.2}, "zero_bytes_per_message": {"compaction": 347.6, "tree": 49.2}, "list_bytes_per_message": 65.5, "syncs_per_message": 2.001}
+{"workload": "entries", "messages": 20000, "batch": 16, "done": 20000, "bytes_per_message": {"compaction": 13.1, "image": 0.0, "log": 15.7, "tree": 41.9}, "zero_bytes_per_message": {"compaction": 13.1, "tree": 19.7}, "list_bytes_per_message": 15.7, "syncs_per_message": 0.126}
+{"workload": "entries", "messages": 20000, "batch": 256, "done": 20000, "bytes_per_message": {"compaction": 13.1, "image": 0.0, "log": 12.4, "tree": 1.4}, "zero_bytes_per_message": {"compaction": 13.1}, "list_bytes_per_message": 12.4, "syncs_per_message": 0.009}
+{"workload": "bank", "messages": 20000, "batch": 1, "done": 30110, "bytes_per_message": {"log": 122.0}, "zero_bytes_per_message": {}, "list_bytes_per_message": 122.0, "syncs_per_message": 1.0}
+{"workload": "bank", "messages": 20000, "batch": 16, "done": 30110, "bytes_per_message": {"log": 35.6}, "zero_bytes_per_message": {}, "list_bytes_per_message": 35.6, "syncs_per_message": 0.062}
+{"workload": "bank", "messages": 20000, "batch": 256, "done": 30110, "bytes_per_message": {"log": 30.1}, "zero_bytes_per_message": {}, "list_bytes_per_message": 30.1, "syncs_per_message": 0.004}
diff --git a/research/experiments/047-leaf-commit/baseline.py b/research/experiments/047-leaf-commit/baseline.py
new file mode 100644
index 0000000..ace86fa
--- /dev/null
+++ b/research/experiments/047-leaf-commit/baseline.py
@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""Research 047: what today's store writes for a workload, by file, as the base the leaf-commit model is held to.
+046's owner (pure-run --store, the shim shipping every cut to a follower, its journal kept) takes a workload's messages
+at a commit grouping (pure-run --batch N: N messages a save); the journal's ops give the bytes written, by file: the
+list log (log-N), the lists' images, the tree's commits (pure.tree) and its compactions (.pending-*), zeros apart. A
+workload with a setup (the bank's branches, tellers and accounts) runs twice, setup alone and setup then the work, and
+the work's bytes are the difference.
+
+  baseline.py entries MESSAGES        appends of one number (046's entries.aup)
+  baseline.py bank ACCOUNTS TRANSFERS the bank (benchmarks/bank/bank.aup): 10 branches, 100 tellers, ACCOUNTS accounts,
+                                      then TRANSFERS transfers (uniform aid, tid, bid; a whole delta in -5000..5000)
+One JSON line a grouping (1, 16, 256)."""
+import collections
+import json
+import random
+import struct
+import subprocess
+import sys
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE.parent / '046-commit-replica'))
+import prefixes  # noqa: E402  (046's journal reader)
+import run as harness  # noqa: E402
+
+PURE = HERE.parents[2]
+
+
+def entries_commands(n):
+    return [harness.message(i) for i in range(n)], 0
+
+
+def bank_commands(accounts, transfers, seed=47):
+    rng = random.Random(seed)
+    setup = [f'message branch {struct.pack("<I", b).hex()}\n' for b in range(10)]
+    setup += [f'message teller {struct.pack("<II", t, t % 10).hex()}\n' for t in range(100)]
+    setup += [f'message account {struct.pack("<II", a, a % 10).hex()}\n' for a in range(accounts)]
+    work = [f'message transfer {struct.pack("<IIId", rng.randrange(accounts), rng.randrange(100), rng.randrange(10), float(rng.randint(-5000, 5000))).hex()}\n'
+            for _ in range(transfers)]
+    return setup + work, len(setup)
+
+
+def written(journal):
+    """Bytes written by file kind (zeros apart), and the number of syncs."""
+    out, zeros, syncs = collections.Counter(), collections.Counter(), 0
+    for kind, path, rest, _ in prefixes.ops_of(journal.read_bytes()):
+        if kind == prefixes.OP_SYNC:
+            syncs += 1
+        if kind != prefixes.OP_WRITE:
+            continue
+        name = ('compaction' if path.startswith('.pending') else 'image' if 'image' in path else
+                'log' if path.startswith('log-') else 'tree' if path == 'pure.tree' else path)
+        data = rest[12:]
+        out[name] += len(data)
+        if not any(data):
+            zeros[name] += len(data)
+    return out, zeros, syncs
+
+
+def one(keys, module, commands, batch):
+    journal = harness.WORK / 'journal'
+    harness.ENV['SHIM_JOURNAL'] = str(journal)
+    r = harness.Run(keys, module=module, extra=('--batch', str(batch)) if batch > 1 else ())
+    try:
+        for c in commands:
+            r.owner.stdin.write(c)
+        r.owner.stdin.close()
+    except BrokenPipeError:
+        pass
+    r.owner.wait()
+    r.reader.join(timeout=30)
+    r.shipper.communicate(timeout=60)
+    r.stop_follower()
+    del harness.ENV['SHIM_JOURNAL']
+    return written(journal), r.done
+
+
+def main():
+    keys = harness.build()
+    what = sys.argv[1]
+    if what == 'entries':
+        module, (commands, setup) = 'entries', entries_commands(int(sys.argv[2]))
+    else:
+        subprocess.run([str(harness.AUP), 'build', str(PURE / 'benchmarks' / 'bank' / 'bank.aup'), '-o',
+                        str(harness.WORK / 'bank.pmod')], check=True)
+        module, (commands, setup) = 'bank', bank_commands(int(sys.argv[2]), int(sys.argv[3]))
+    work = len(commands) - setup
+    for batch in (1, 16, 256):
+        (total, zeros, syncs), done = one(keys, module, commands, batch)
+        if setup:
+            (base, base_zeros, base_syncs), _ = one(keys, module, commands[:setup], batch)
+            total, zeros, syncs = total - base, zeros - base_zeros, syncs - base_syncs
+        print(json.dumps({'workload': what, 'messages': work, 'batch': batch, 'done': done,
+                          'bytes_per_message': {k: round(v / work, 1) for k, v in sorted(total.items())},
+                          'zero_bytes_per_message': {k: round(v / work, 1) for k, v in sorted(zeros.items())},
+                          'list_bytes_per_message': round((total['log'] + total['image']) / work, 1),
+                          'syncs_per_message': round(syncs / work, 3)}), flush=True)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/research/experiments/047-leaf-commit/model.py b/research/experiments/047-leaf-commit/model.py
new file mode 100644
index 0000000..a441166
--- /dev/null
+++ b/research/experiments/047-leaf-commit/model.py
@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+"""Research 047: "the leaf is the commit" (feedback 009) before any store code: what a store whose commit writes the
+list leaves it changed, in the bytes memory keeps them in, would write for a workload, against what today's store
+writes (baseline.py: the log and the images). A byte-accounting model, not a store.
+
+The layout (the memory's, core/module-list.c, as the disk's: one byte shape):
+- a leaf: a 40-byte header (Node) and, per field, 32 cells (8 bytes; a text a 16-byte span, its bytes elsewhere):
+  296 B for one number, 1,320 B for the bank's account (three numbers and a text), 808 a teller, 552 a branch,
+  1,320 a transfer;
+- an inner node: 680 B (the header, 32 children, the guide: 32 counts before and 32 keys);
+- a commit: a 128-byte record (as the tree's today) and 16 bytes a list it changed (its root and count).
+Two ways to write a commit:
+- cow: every leaf the commit changed is written anew, and every inner node above them (the root too); the old copies
+  are garbage until a compaction.
+- inplace: tail, and a change elsewhere written in place too (its cell alone): not a way the store may take (a kept
+  snapshot reads the cell, and a crash between the write and the commit loses the old value): counted to show it.
+- tail: as cow, but an append at a list's end writes only its cells, in place, into its last leaf's free slots (and a
+  new leaf its header, and its parent one child entry, 20 B, in place): what a reader of an older commit cannot see,
+  since it reads no cell past its commit's count. Changes elsewhere are cow.
+Compaction by the tree's own rule: when the file's end past its 8 KiB of slots passes 4 x live + 1 MiB, the live nodes
+are written again (counted). Snapshots: one kept every 64 commits for 16 x 64 commits; every in-place write is checked
+against the counts the kept ones read (a write a kept snapshot could see stops the model).
+
+Workloads (each as baseline.py drives the store):
+- entries: one number appended a message (046's entries.aup);
+- bank: 10 branches, 100 tellers, 10,000 accounts loaded first (leaves full), then a transfer a message: its account's,
+  teller's and branch's balance set (uniform aid, tid, bid; the same generator as baseline.py), a transfer appended.
+
+The bars, set before running (today's list bytes, the log and images, by grouping 1 / 16 / 256 messages a commit):
+- appends: the leaf commit's list bytes a message at most today's (entries: 65.5 / 15.7 / 12.4);
+- sparse updates: at most twice today's (bank: 2 x 122 / 35.6 / 30.1 = 244 / 71.2 / 60.2);
+- snapshot: no in-place write a kept snapshot can see (cow and tail; inplace is there to break it).
+What it means, said before: if appends pass and updates do not, "one unit" fails as stated: the leaf is the unit for
+appends and bulk, and a sparse change needs a delta form, which is what the log is; the honest design is then two
+units (or the log kept for updates alone), not one.
+
+  model.py WORKLOAD MESSAGES     one JSON line a way and grouping"""
+import json
+import random
+import sys
+
+HEADER, INNER, FANOUT, RECORD, PER_LIST, SLOTS, SLACK = 40, 680, 32, 128, 16, 8192, 1 << 20
+ENTRY = 20  # a child entry written into a parent in place: its pointer (8), count before (4) and key (8)
+KEEP_EVERY, KEPT = 64, 16
+
+
+class List:
+    """A list's tree, its leaves full as loaded: leaf i holds items 32 i .. 32 i + 31."""
+
+    def __init__(self, name, cell, items):
+        self.name, self.cell, self.items = name, cell, items
+        self.leaf = HEADER + FANOUT * cell
+        self.version = {}  # node (level, index) -> the commit that wrote it
+
+    def leaves(self):
+        return max(1, -(-self.items // FANOUT))
+
+    def levels(self):
+        """Inner levels above the leaves (0 when one leaf is the root)."""
+        n, levels = self.leaves(), 0
+        while n > 1:
+            n, levels = -(-n // FANOUT), levels + 1
+        return levels
+
+    def path(self, leaf):
+        """The inner nodes above a leaf: (level, index) from the leaves' parents up."""
+        out, index = [], leaf
+        for level in range(1, self.levels() + 1):
+            index //= FANOUT
+            out.append((level, index))
+        return out
+
+    def live(self):
+        n, total, level = self.leaves(), self.leaves() * self.leaf, 0
+        while n > 1:
+            n = -(-n // FANOUT)
+            total += n * INNER
+        return total
+
+
+def workload(name, messages, seed=47):
+    """Each message's changes: (list, 'set', position) or (list, 'append', None)."""
+    if name == 'entries':
+        lists = {'entries': List('entries', 8, 0)}
+        return lists, [[('entries', 'append', None)] for _ in range(messages)]
+    rng = random.Random(seed)
+    accounts = 10000
+    lists = {'accounts': List('accounts', 40, accounts), 'tellers': List('tellers', 24, 100),
+             'branches': List('branches', 16, 10), 'history': List('history', 40, 0)}
+    out = []
+    for _ in range(messages):
+        aid, tid, bid = rng.randrange(accounts), rng.randrange(100), rng.randrange(10)
+        rng.randint(-5000, 5000)  # the delta, drawn as baseline.py draws it (the same stream)
+        out.append([('accounts', 'set', aid), ('tellers', 'set', tid), ('branches', 'set', bid), ('history', 'append', None)])
+    return lists, out
+
+
+def run(name, messages, batch, way):
+    lists, changes = workload(name, messages)
+    written = {'leaves': 0, 'cells': 0, 'inner': 0, 'entries': 0, 'records': 0, 'compaction': 0}
+    end, live0 = SLOTS, sum(l.live() for l in lists.values())
+    end += live0  # what the setup loaded (not counted: baseline.py takes it away too)
+    kept = []  # snapshots kept: {list name: count it reads}
+    violations, commits = 0, 0
+    for start in range(0, messages, batch):
+        group = changes[start:start + batch]
+        touched = {n: set() for n in lists}  # leaves to write anew
+        tail_cells = {n: 0 for n in lists}
+        for message in group:
+            for list_name, op, position in message:
+                l = lists[list_name]
+                if op == 'append':
+                    at = l.items
+                    l.items += 1
+                    leaf = at // FANOUT
+                    if way in ('tail', 'inplace'):
+                        for snapshot in kept:  # a kept snapshot never reads at or past its count
+                            if at < snapshot.get(list_name, 0):
+                                violations += 1
+                        tail_cells[list_name] += 1
+                        if at % FANOUT == 0:  # a new leaf: its header, and an entry in its parent (in place); a
+                            written['leaves'] += HEADER  # parent full: a new one (its header) and an entry above it
+                            index = leaf
+                            while index:
+                                written['entries'] += ENTRY
+                                if index % FANOUT:
+                                    break
+                                index //= FANOUT
+                                written['inner'] += HEADER
+                    else:
+                        touched[list_name].add(leaf)
+                elif way == 'inplace':  # the cell written where it is: what a kept snapshot (and a crash) would see
+                    for snapshot in kept:
+                        if position < snapshot.get(list_name, 0):
+                            violations += 1
+                    tail_cells[list_name] += 1
+                else:
+                    touched[list_name].add(position // FANOUT)
+        size = 0
+        for n, leaves in touched.items():
+            l = lists[n]
+            inner = set()
+            for leaf in leaves:
+                inner.update(l.path(leaf))
+            size += len(leaves) * l.leaf + len(inner) * INNER
+            written['leaves'] += len(leaves) * l.leaf
+            written['inner'] += len(inner) * INNER
+        for n, cells in tail_cells.items():
+            written['cells'] += cells * lists[n].cell
+            size += cells * lists[n].cell
+        changed = sum(1 for n in lists if touched[n] or tail_cells[n])
+        written['records'] += RECORD + PER_LIST * changed
+        end += size + RECORD + PER_LIST * changed
+        commits += 1
+        if commits % KEEP_EVERY == 0:
+            kept.append({n: l.items for n, l in lists.items()})
+            kept = kept[-KEPT:]
+        live = sum(l.live() for l in lists.values())
+        if end - SLOTS > 4 * live + SLACK:
+            written['compaction'] += live
+            end = SLOTS + live
+    total = sum(written.values())
+    lists_bytes = written['leaves'] + written['cells'] + written['inner'] + written['entries']
+    return {'workload': name, 'messages': messages, 'batch': batch, 'way': way, 'commits': commits,
+            'list_bytes_per_message': round((lists_bytes + written['records'] - RECORD * commits) / messages, 1),
+            'by_part_per_message': {k: round(v / messages, 1) for k, v in written.items()},
+            'all_bytes_per_message': round(total / messages, 1), 'snapshot_violations': violations}
+
+
+def main():
+    name, messages = sys.argv[1], int(sys.argv[2])
+    for batch in (1, 16, 256):
+        for way in ('cow', 'tail', 'inplace'):
+            print(json.dumps(run(name, messages, batch, way)), flush=True)
+
+
+if __name__ == '__main__':
+    main()