diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 0000000..95acbf9
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1,2 @@
+benchmark/*.png filter=lfs diff=lfs merge=lfs -text
+benchmark/results/bench_all.csv.gz filter=lfs diff=lfs merge=lfs -text
diff --git a/benchmark/.gitignore b/benchmark/.gitignore
new file mode 100644
index 0000000..fbca225
--- /dev/null
+++ b/benchmark/.gitignore
@@ -0,0 +1 @@
+results/
diff --git a/benchmark/bench_go.go b/benchmark/bench_go.go
deleted file mode 100644
index 568797a..0000000
--- a/benchmark/bench_go.go
+++ /dev/null
@@ -1,344 +0,0 @@
-// caio Go baseline benchmark
-// Same sweeps as bench.py — outputs bench_go.csv in the same format.
-//
-// Usage:
-//   go run bench_go.go -data /tmp/caio-bench -results /tmp/results
-package main
-
-import (
-	"encoding/csv"
-	"flag"
-	"fmt"
-	"math/rand"
-	"os"
-	"path/filepath"
-	"sync"
-	"time"
-)
-
-var (
-	dataDir    = flag.String("data", "/tmp/caio-bench", "directory with *.bin files")
-	resultsDir = flag.String("results", "/tmp/results", "output directory")
-	totalOps   = flag.Int("ops", 10000, "ops per cell")
-	warmupOps  = flag.Int("warmup", 500, "warmup ops")
-)
-
-const (
-	concSweepChunk  = 16 * 1024
-	chunkSweepConc  = 64
-	backend         = "go_goroutine"
-)
-
-var concSweep  = []int{1, 4, 16, 64, 256, 512, 1024, 2048, 4096}
-var chunkSweep = []int{4 * 1024, 16 * 1024, 64 * 1024, 256 * 1024, 1024 * 1024, 4 * 1024 * 1024}
-
-// ── file pool ────────────────────────────────────────────────────────────────
-
-type fileEntry struct {
-	f    *os.File
-	size int64
-}
-
-type readPool struct {
-	files []fileEntry
-}
-
-func openPool(dir string) (*readPool, error) {
-	matches, err := filepath.Glob(filepath.Join(dir, "*.bin"))
-	if err != nil || len(matches) == 0 {
-		return nil, fmt.Errorf("no .bin files in %s", dir)
-	}
-	p := &readPool{}
-	for _, path := range matches {
-		f, err := os.Open(path)
-		if err != nil {
-			return nil, err
-		}
-		st, _ := f.Stat()
-		p.files = append(p.files, fileEntry{f, st.Size()})
-	}
-	return p, nil
-}
-
-func (p *readPool) randArgs(chunk int) (f *os.File, offset int64) {
-	e := p.files[rand.Intn(len(p.files))]
-	hi := e.size - int64(chunk)
-	if hi < 0 {
-		hi = 0
-	}
-	off := rand.Int63n(hi+1) &^ 0xFFF // 4 KB aligned
-	return e.f, off
-}
-
-func (p *readPool) seqArgs(chunk int) []struct {
-	f      *os.File
-	offset int64
-} {
-	var out []struct {
-		f      *os.File
-		offset int64
-	}
-	for _, e := range p.files {
-		for off := int64(0); off+int64(chunk) <= e.size; off += int64(chunk) {
-			out = append(out, struct {
-				f      *os.File
-				offset int64
-			}{e.f, off})
-		}
-	}
-	return out
-}
-
-func (p *readPool) close() {
-	for _, e := range p.files {
-		e.f.Close()
-	}
-}
-
-// ── engine ───────────────────────────────────────────────────────────────────
-
-// runRead runs n pread ops at `concurrency` goroutines in parallel.
-// Returns (per-op latencies in µs, wall-clock seconds for the whole batch).
-func runRead(pool *readPool, concurrency, n, chunk int, sequential bool) ([]float64, float64) {
-	sem := make(chan struct{}, concurrency)
-	lats := make([]float64, n)
-
-	var seqSlice []struct {
-		f      *os.File
-		offset int64
-	}
-	var seqIdx int
-	var seqMu sync.Mutex
-	if sequential {
-		seqSlice = pool.seqArgs(chunk)
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(n)
-
-	t0 := time.Now()
-	for i := 0; i < n; i++ {
-		sem <- struct{}{}
-		idx := i
-		go func() {
-			defer func() {
-				<-sem
-				wg.Done()
-			}()
-
-			var f *os.File
-			var offset int64
-			if sequential {
-				seqMu.Lock()
-				entry := seqSlice[seqIdx%len(seqSlice)]
-				seqIdx++
-				seqMu.Unlock()
-				f, offset = entry.f, entry.offset
-			} else {
-				f, offset = pool.randArgs(chunk)
-			}
-
-			b := make([]byte, chunk)
-			opT0 := time.Now()
-			_, _ = f.ReadAt(b, offset)
-			lats[idx] = float64(time.Since(opT0).Nanoseconds()) / 1000.0
-		}()
-	}
-	wg.Wait()
-	wall := time.Since(t0).Seconds()
-	return lats, wall
-}
-
-func runWrite(concurrency, n, chunk int, sequential bool) ([]float64, float64, func()) {
-	f, _ := os.CreateTemp("", "caio-bench-write-*")
-	size := int64(1024 * 1024 * 1024)
-	f.Seek(size-1, 0)
-	f.Write([]byte{0})
-
-	buf := make([]byte, chunk)
-	rand.Read(buf)
-
-	sem := make(chan struct{}, concurrency)
-	lats := make([]float64, n)
-	var wg sync.WaitGroup
-	wg.Add(n)
-
-	seqIdx := 0
-	var seqMu sync.Mutex
-
-	t0 := time.Now()
-	for i := 0; i < n; i++ {
-		sem <- struct{}{}
-		idx := i
-		go func() {
-			defer func() {
-				<-sem
-				wg.Done()
-			}()
-
-			var offset int64
-			if sequential {
-				seqMu.Lock()
-				offset = int64(seqIdx%int(size/int64(chunk))) * int64(chunk)
-				seqIdx++
-				seqMu.Unlock()
-			} else {
-				hi := size - int64(chunk)
-				offset = rand.Int63n(hi+1) &^ 0xFFF
-			}
-
-			opT0 := time.Now()
-			f.WriteAt(buf, offset)
-			lats[idx] = float64(time.Since(opT0).Nanoseconds()) / 1000.0
-		}()
-	}
-	wg.Wait()
-	wall := time.Since(t0).Seconds()
-
-	cleanup := func() {
-		name := f.Name()
-		f.Close()
-		os.Remove(name)
-	}
-	return lats, wall, cleanup
-}
-
-// ── stats ────────────────────────────────────────────────────────────────────
-
-func mean(xs []float64) float64 {
-	s := 0.0
-	for _, x := range xs {
-		s += x
-	}
-	return s / float64(len(xs))
-}
-
-func pct(xs []float64, p float64) float64 {
-	cp := make([]float64, len(xs))
-	copy(cp, xs)
-	// simple insertion sort is fine for small slices; use sort for large
-	for i := 1; i < len(cp); i++ {
-		for j := i; j > 0 && cp[j] < cp[j-1]; j-- {
-			cp[j], cp[j-1] = cp[j-1], cp[j]
-		}
-	}
-	idx := int(float64(len(cp)) * p)
-	if idx >= len(cp) {
-		idx = len(cp) - 1
-	}
-	return cp[idx]
-}
-
-// ── main ─────────────────────────────────────────────────────────────────────
-
-func main() {
-	flag.Parse()
-
-	pool, err := openPool(*dataDir)
-	if err != nil {
-		fmt.Fprintln(os.Stderr, err)
-		os.Exit(1)
-	}
-	defer pool.close()
-
-	os.MkdirAll(*resultsDir, 0755)
-	outPath := filepath.Join(*resultsDir, "bench_go.csv")
-	outF, _ := os.Create(outPath)
-	w := csv.NewWriter(outF)
-	w.Write([]string{"backend", "sweep", "op", "concurrency", "chunk_bytes", "latency_us"})
-
-	hdr := fmt.Sprintf("%-14s %-5s %6s  %8s  %6s  %7s  %7s  %7s  %7s  %8s  (ms)",
-		"backend", "op", "conc", "ops/s", "MB/s", "mean", "p50", "p95", "p99", "p999")
-
-	sep := "────────────────────────────────────────────────────────────────────────────────────────────────────────"
-
-	// ── concurrency sweeps (fixed chunk, vary concurrency) ───────────────────
-	type concSweepSpec struct {
-		tag   string
-		label string
-		seq   bool
-	}
-	concSweeps := []concSweepSpec{
-		{"conc_sweep_rand", "SWEEP 1 — concurrency / random", false},
-		{"conc_sweep_seq",  "SWEEP 3 — concurrency / sequential", true},
-	}
-	for _, sw := range concSweeps {
-		fmt.Printf("\n%s\n%s\n%s\n", sw.label, hdr, sep)
-		for _, conc := range concSweep {
-			runRead(pool, conc, *warmupOps, concSweepChunk, sw.seq)
-			lats, wallR := runRead(pool, conc, *totalOps, concSweepChunk, sw.seq)
-			opsS := float64(*totalOps) / wallR
-			mbS := opsS * float64(concSweepChunk) / 1e6
-			fmt.Printf("%-14s %-5s %6d  %8.0f  %6.1f  %7.3f  %7.3f  %7.3f  %7.3f  %8.3f\n",
-				backend, "read", conc, opsS, mbS,
-				mean(lats)/1000, pct(lats, 0.5)/1000, pct(lats, 0.95)/1000,
-				pct(lats, 0.99)/1000, pct(lats, 0.999)/1000)
-			for _, l := range lats {
-				w.Write([]string{backend, sw.tag, "read", fmt.Sprint(conc), fmt.Sprint(concSweepChunk), fmt.Sprintf("%.1f", l)})
-			}
-			_, _, wc := runWrite(conc, *warmupOps, concSweepChunk, sw.seq)
-			wc()
-			wlats, wallW, cleanup := runWrite(conc, *totalOps, concSweepChunk, sw.seq)
-			opsW := float64(*totalOps) / wallW
-			mbW := opsW * float64(concSweepChunk) / 1e6
-			fmt.Printf("%-14s %-5s %6d  %8.0f  %6.1f  %7.3f  %7.3f  %7.3f  %7.3f  %8.3f\n",
-				backend, "write", conc, opsW, mbW,
-				mean(wlats)/1000, pct(wlats, 0.5)/1000, pct(wlats, 0.95)/1000,
-				pct(wlats, 0.99)/1000, pct(wlats, 0.999)/1000)
-			for _, l := range wlats {
-				w.Write([]string{backend, sw.tag, "write", fmt.Sprint(conc), fmt.Sprint(concSweepChunk), fmt.Sprintf("%.1f", l)})
-			}
-			cleanup()
-		}
-	}
-
-	// ── chunk sweeps (fixed concurrency=64, vary chunk) ──────────────────────
-	type chunkSweepSpec struct {
-		tag   string
-		label string
-		seq   bool
-	}
-	chunkSweeps := []chunkSweepSpec{
-		{"chunk_sweep_rand", "SWEEP 2 — chunk size / random", false},
-		{"chunk_sweep_seq",  "SWEEP 4 — chunk size / sequential", true},
-	}
-	hdrChunk := fmt.Sprintf("%-14s %-5s %6s  %8s  %6s  %7s  %7s  %7s  %7s  %8s  (ms)",
-		"backend", "op", "chunk", "ops/s", "MB/s", "mean", "p50", "p95", "p99", "p999")
-	for _, sw := range chunkSweeps {
-		fmt.Printf("\n%s\n%s\n%s\n", sw.label, hdrChunk, sep)
-		for _, chunk := range chunkSweep {
-			label := fmt.Sprintf("%dK", chunk/1024)
-			if chunk >= 1024*1024 {
-				label = fmt.Sprintf("%dM", chunk/(1024*1024))
-			}
-			runRead(pool, chunkSweepConc, *warmupOps, chunk, sw.seq)
-			lats, wallR := runRead(pool, chunkSweepConc, *totalOps, chunk, sw.seq)
-			opsS := float64(*totalOps) / wallR
-			mbS := opsS * float64(chunk) / 1e6
-			fmt.Printf("%-14s %-5s %6s  %8.0f  %6.1f  %7.3f  %7.3f  %7.3f  %7.3f  %8.3f\n",
-				backend, "read", label, opsS, mbS,
-				mean(lats)/1000, pct(lats, 0.5)/1000, pct(lats, 0.95)/1000,
-				pct(lats, 0.99)/1000, pct(lats, 0.999)/1000)
-			for _, l := range lats {
-				w.Write([]string{backend, sw.tag, "read", fmt.Sprint(chunkSweepConc), fmt.Sprint(chunk), fmt.Sprintf("%.1f", l)})
-			}
-			_, _, wc := runWrite(chunkSweepConc, *warmupOps, chunk, sw.seq)
-			wc()
-			wlats, wallW, cleanup := runWrite(chunkSweepConc, *totalOps, chunk, sw.seq)
-			opsW := float64(*totalOps) / wallW
-			mbW := opsW * float64(chunk) / 1e6
-			fmt.Printf("%-14s %-5s %6s  %8.0f  %6.1f  %7.3f  %7.3f  %7.3f  %7.3f  %8.3f\n",
-				backend, "write", label, opsW, mbW,
-				mean(wlats)/1000, pct(wlats, 0.5)/1000, pct(wlats, 0.95)/1000,
-				pct(wlats, 0.99)/1000, pct(wlats, 0.999)/1000)
-			for _, l := range wlats {
-				w.Write([]string{backend, sw.tag, "write", fmt.Sprint(chunkSweepConc), fmt.Sprint(chunk), fmt.Sprintf("%.1f", l)})
-			}
-			cleanup()
-		}
-	}
-
-	w.Flush()
-	outF.Close()
-	fmt.Printf("\nCSV → %s\n", outPath)
-}
diff --git a/benchmark/bench_runner.py b/benchmark/bench_runner.py
index db5ad0a..ea28920 100644
--- a/benchmark/bench_runner.py
+++ b/benchmark/bench_runner.py
@@ -2,14 +2,16 @@
 """
 Run each caio backend in a separate subprocess so thread pools from one
 backend cannot influence measurements of the next.
-Also runs the Go goroutine baseline if `go` is available.
+
+Every per-backend CSV uses the same schema. The merge step validates that
+schema instead of silently attaching an outdated header to wider rows.
 
 Usage:
   CAIO_BENCH_DATA=/tmp/caio-bench CAIO_RESULTS=/tmp/results uv run python bench_runner.py
 """
+import csv
 import os
 import pathlib
-import shutil
 import subprocess
 import sys
 
@@ -18,10 +20,20 @@
 RESULTS_DIR = pathlib.Path(os.environ.get("CAIO_RESULTS", "/tmp/results"))
 DATA_DIR    = pathlib.Path(os.environ.get("CAIO_BENCH_DATA", "/tmp/caio-bench"))
 BENCH       = pathlib.Path(__file__).parent / "bench.py"
-BENCH_GO    = pathlib.Path(__file__).parent / "bench_go.go"
 
 env = os.environ.copy()
 
+CSV_COLUMNS = [
+    "backend",
+    "sweep",
+    "op",
+    "concurrency",
+    "chunk_bytes",
+    "latency_us",
+    "wall_s",
+    "n_ops",
+]
+
 
 def run_backend(name: str):
     print(f"\n{'━' * 80}", flush=True)
@@ -38,24 +50,34 @@ def run_backend(name: str):
         print(f"[!] {name} exited with code {proc.returncode}", flush=True)
 
 
-def run_go():
-    go_bin = shutil.which("go")
-    if not go_bin:
-        print("\n[!] go not found in PATH, skipping go_goroutine baseline", flush=True)
-        return
+def merge_results():
+    # Merge all per-backend CSVs into one, rejecting incompatible files
+    # rather than producing a syntactically valid but column-shifted result.
+    rows: list[dict[str, str]] = []
+
+    def collect(path: pathlib.Path):
+        if not path.exists():
+            return
+        with path.open(newline="") as fp:
+            reader = csv.DictReader(fp)
+            if reader.fieldnames != CSV_COLUMNS:
+                print(
+                    f"[!] {path}: incompatible CSV header "
+                    f"{reader.fieldnames!r}; expected {CSV_COLUMNS!r}",
+                    flush=True,
+                )
+                return
+            rows.extend(reader)
 
-    print(f"\n{'━' * 80}", flush=True)
-    print(f"  backend: go_goroutine", flush=True)
-    print(f"{'━' * 80}", flush=True)
+    for name in BACKENDS:
+        collect(RESULTS_DIR / f"bench_{name}.csv")
 
-    proc = subprocess.run(
-        [go_bin, "run", str(BENCH_GO),
-         "-data", str(DATA_DIR),
-         "-results", str(RESULTS_DIR)],
-        cwd=BENCH_GO.parent,
-    )
-    if proc.returncode != 0:
-        print(f"[!] go_goroutine exited with code {proc.returncode}", flush=True)
+    merged = RESULTS_DIR / "bench_all.csv"
+    with merged.open("w", newline="") as fp:
+        writer = csv.DictWriter(fp, fieldnames=CSV_COLUMNS)
+        writer.writeheader()
+        writer.writerows(rows)
+    print(f"\nMerged CSV → {merged}")
 
 
 def main():
@@ -64,26 +86,7 @@ def main():
     for name in BACKENDS:
         run_backend(name)
 
-    run_go()
-
-    # merge all per-backend CSVs into one
-    header = "backend,sweep,op,concurrency,chunk_bytes,latency_us\n"
-    rows: list[str] = []
-    for name in BACKENDS:
-        f = RESULTS_DIR / f"bench_{name}.csv"
-        if not f.exists():
-            continue
-        lines = f.read_text().splitlines()
-        rows.extend(lines[1:])   # skip header
-
-    go_csv = RESULTS_DIR / "bench_go.csv"
-    if go_csv.exists():
-        lines = go_csv.read_text().splitlines()
-        rows.extend(lines[1:])   # skip header
-
-    merged = RESULTS_DIR / "bench_all.csv"
-    merged.write_text(header + "\n".join(rows) + "\n")
-    print(f"\nMerged CSV → {merged}")
+    merge_results()
 
 
 if __name__ == "__main__":
diff --git a/benchmark/chunk_sweep_latency_rand.png b/benchmark/chunk_sweep_latency_rand.png
index 85cef21..a7899ee 100644
--- a/benchmark/chunk_sweep_latency_rand.png
+++ b/benchmark/chunk_sweep_latency_rand.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:7790bf8bbd767b39656fd1ffa7fb734bfe6572d2cf432844befce33af1a8de95
-size 266023
+oid sha256:6750b9152b457231248b1531afa8043eb6ef095182840f9fc783e7891d74ec4e
+size 297775
diff --git a/benchmark/chunk_sweep_latency_seq.png b/benchmark/chunk_sweep_latency_seq.png
index 321d379..9b893d7 100644
--- a/benchmark/chunk_sweep_latency_seq.png
+++ b/benchmark/chunk_sweep_latency_seq.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:cdbdb2336073843bc9aa97089eaa70647b570143a8dd9df93aaf0500c3439975
-size 273119
+oid sha256:f860d1b86fdd9d2860a44832056d454c111041f3f19395377bf72c365ac893f9
+size 294864
diff --git a/benchmark/chunk_sweep_throughput_rand.png b/benchmark/chunk_sweep_throughput_rand.png
index 8a27543..7b095d5 100644
--- a/benchmark/chunk_sweep_throughput_rand.png
+++ b/benchmark/chunk_sweep_throughput_rand.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:f573f3c7f7c5ceb4add5ddc84d42a9abe8255471f7475715a0a620b04c69139f
-size 148848
+oid sha256:b378e33d1510b9fa620b7f03c477a79be8de5fb4ce1a39966b449ae86b64945e
+size 148823
diff --git a/benchmark/chunk_sweep_throughput_seq.png b/benchmark/chunk_sweep_throughput_seq.png
index a96c3cf..29800f6 100644
--- a/benchmark/chunk_sweep_throughput_seq.png
+++ b/benchmark/chunk_sweep_throughput_seq.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:50e48b1c0bc277fc86f70a47bf8f449566b1498f915196acfaafd1917cb54aee
-size 145776
+oid sha256:3c8636aa9446cca65c1bce9dc02d2e76b88f035a681c1eac1d64511be67e5371
+size 149952
diff --git a/benchmark/concurrency_sweep_latency_rand.png b/benchmark/concurrency_sweep_latency_rand.png
index 2855b1b..8f5c145 100644
--- a/benchmark/concurrency_sweep_latency_rand.png
+++ b/benchmark/concurrency_sweep_latency_rand.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:e23185c056e7f8aadff30999f34030170c1e265538b4daf41a29553edfe6d88d
-size 356655
+oid sha256:4d985e01dd20b692382e164e597097b0cca5ccf73aeff49dd618c97c08d17d75
+size 350573
diff --git a/benchmark/concurrency_sweep_latency_seq.png b/benchmark/concurrency_sweep_latency_seq.png
index 8432a4e..09d62fb 100644
--- a/benchmark/concurrency_sweep_latency_seq.png
+++ b/benchmark/concurrency_sweep_latency_seq.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:9906605b254604e6bb7620a3c32c0397133f47ffb34837009c1b788c622f560e
-size 351726
+oid sha256:1617d9604cbf92fbf451d6a06531a64e71199cb83e333d2101f11396c17e821e
+size 345652
diff --git a/benchmark/concurrency_sweep_throughput_rand.png b/benchmark/concurrency_sweep_throughput_rand.png
index 5f4d6bf..058f7f5 100644
--- a/benchmark/concurrency_sweep_throughput_rand.png
+++ b/benchmark/concurrency_sweep_throughput_rand.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:a6e412927b423a78d237d558e54d6e3a166770e58d5001de559ea24b181b3d6f
-size 143094
+oid sha256:7dccef412d17b0d5f06b51524a0cc82375e9fb0475da7ba26e75ad3a7ee9ccf8
+size 166242
diff --git a/benchmark/concurrency_sweep_throughput_seq.png b/benchmark/concurrency_sweep_throughput_seq.png
index 4539097..bafcafa 100644
--- a/benchmark/concurrency_sweep_throughput_seq.png
+++ b/benchmark/concurrency_sweep_throughput_seq.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:b506d1a757d4c8b24bb5ab015e70259cfa3936560afe5d3f3fed8e065e390ad6
-size 158177
+oid sha256:046b41fb77df09f37b57a4959781d51c2051b93633d0b65ccbfebeac20a9fb36
+size 165142
diff --git a/benchmark/latency_histograms.png b/benchmark/latency_histograms.png
index f4a9aac..bc22447 100644
--- a/benchmark/latency_histograms.png
+++ b/benchmark/latency_histograms.png
@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:13d9d8696a67122d1ec70fd6a980f3ba893786d615611a7c41d44e0f89ee6a4e
-size 131087
+oid sha256:c65fe6aed023eaa54599cf432b500a1a8690cdb4686f64c97c7494b9197efd5e
+size 126937
diff --git a/benchmark/plot_results.py b/benchmark/plot_results.py
index 383d856..c88bd38 100644
--- a/benchmark/plot_results.py
+++ b/benchmark/plot_results.py
@@ -10,7 +10,8 @@
   latency_histograms.png  (rand only)
 
 Usage:
-  CAIO_RESULTS=/tmp/results uv run python plot_results.py
+  CAIO_RESULTS=/tmp/results PLOT_RESULTS=/tmp/results \
+    uv run --with matplotlib --with numpy python plot_results.py
 """
 import csv
 import os
@@ -28,7 +29,12 @@
 RESULTS_DIR = pathlib.Path(os.environ.get("CAIO_RESULTS", "/tmp/results"))
 CSV_PATH    = RESULTS_DIR / "bench_all.csv"
 
-BACKENDS = ["linux_uring", "linux_aio", "thread_aio", "python_aio"]
+BACKENDS = [
+    "linux_uring",
+    "linux_aio",
+    "thread_aio",
+    "python_aio",
+]
 
 OP_COLORS  = {"read": "#3498db", "write": "#e74c3c"}
 OP_MARKERS = {"read": "o", "write": "s"}
@@ -122,6 +128,17 @@ def _chunk_xticks(ax, chunks: List[int]):
     ax.set_xticklabels([fmt_chunk(c) for c in chunks])
 
 
+def _conc_xticks(ax, values: List[int]):
+    """Keep logarithmic concurrency labels readable in wide backend grids."""
+    ticks = values
+    if len(ticks) > 6:
+        ticks = ticks[::2]
+        if ticks[-1] != values[-1]:
+            ticks.append(values[-1])
+    ax.set_xticks(ticks)
+    ax.tick_params(axis="x", labelsize=8)
+
+
 def _available(rows: List[Row]) -> List[str]:
     """Backends that actually appear in the CSV data."""
     present = {r["backend"] for r in rows}
@@ -167,7 +184,7 @@ def plot_conc_throughput(rows: List[Row], access: str = "rand"):
 
         apply_style(ax, "Concurrency", "kops/s", backend, xscale="log")
         ax.xaxis.set_major_formatter(ticker.ScalarFormatter())
-        ax.set_xticks(xs_all)
+        _conc_xticks(ax, xs_all)
         ax.legend(fontsize=8, framealpha=0.7)
 
     fig.tight_layout()
@@ -220,7 +237,7 @@ def plot_conc_latency(rows: List[Row], access: str = "rand"):
                     xscale="log", yscale="log")
         ax.xaxis.set_major_formatter(ticker.ScalarFormatter())
         if pivots:
-            ax.set_xticks(pivots)
+            _conc_xticks(ax, pivots)
         _op_legend(ax)
 
     fig.tight_layout()
diff --git a/benchmark/results/bench_all.csv.gz b/benchmark/results/bench_all.csv.gz
new file mode 100644
index 0000000..4fad60f
--- /dev/null
+++ b/benchmark/results/bench_all.csv.gz
@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:38da57086d60b8d2b9e73ad56f9d47b1a7a548ee448de26448454603f79eb260
+size 836960
diff --git a/caio/asyncio_base.py b/caio/asyncio_base.py
index 1a211f3..9720d69 100644
--- a/caio/asyncio_base.py
+++ b/caio/asyncio_base.py
@@ -14,9 +14,14 @@ class AsyncioContextBase(abc.ABC):
     CONTEXT_CLASS: ContextType
     OPERATION_CLASS: OperationType
 
-    def __init__(self, max_requests=None, loop=None, **kwargs):
+    def __init__(self, max_requests=None, loop=None, deferred=False, **kwargs):
         max_requests = max_requests or self.MAX_REQUESTS_DEFAULT
         self.loop = loop or asyncio.get_event_loop()
+        # Opt-in: batch multiple submissions into one syscall instead of
+        # flushing/submitting eagerly after each one. Only linux_aio's and
+        # linux_uring's adapters actually act on this (see their own
+        # _submit_op/_on_submitted overrides) - it's a no-op elsewhere.
+        self.deferred = deferred
         self.semaphore = asyncio.BoundedSemaphore(max_requests)
         self.context = self._create_context(max_requests, **kwargs)
 
@@ -43,9 +48,7 @@ async def submit(self, op):
         op.set_callback(partial(self._on_done, future))
 
         async with self.semaphore:
-            if self.context.submit(op) != 1:
-                raise OSError("Operation was not submitted")
-
+            self._submit_op(op, future)
             self._on_submitted()
 
             try:
@@ -58,6 +61,13 @@ async def submit(self, op):
                 raise
             return op.get_value()
 
+    def _submit_op(self, op, future):
+        """Default: submit immediately and raise on rejection right away.
+        Subclasses may defer submission (batching) - in that case they must
+        resolve `future` with an exception instead if it's later rejected."""
+        if self.context.submit(op) != 1:
+            raise OSError("Operation was not submitted")
+
     def _on_submitted(self):
         """Hook called after each op is placed in the context's queue.
         Subclasses can override to implement batched submission (deferred flush).
diff --git a/caio/linux_aio_asyncio.py b/caio/linux_aio_asyncio.py
index 7aed215..05a7cb7 100644
--- a/caio/linux_aio_asyncio.py
+++ b/caio/linux_aio_asyncio.py
@@ -9,8 +9,51 @@ class AsyncioContext(AsyncioContextBase):
     def _create_context(self, max_requests):
         context = super()._create_context(max_requests)
         self.loop.add_reader(context.fileno, self._on_read_event)
+        self._pending = []
+        self._submit_scheduled = False
         return context
 
+    def _submit_op(self, op, future):
+        if not self.deferred:
+            super()._submit_op(op, future)
+            return
+        # Context.submit() already accepts *ops and builds one iocbpp array
+        # for the whole batch - call_soon() defers to the next _run_once()
+        # pass, after every currently-ready submit() has queued its op, so
+        # one submit() call handles all of them together.
+        self._pending.append((op, future))
+        if not self._submit_scheduled:
+            self._submit_scheduled = True
+            self.loop.call_soon(self._deferred_submit)
+
+    def _deferred_submit(self):
+        self._submit_scheduled = False
+        pending, self._pending = self._pending, []
+        # Already cancelled while waiting here - never reached the kernel,
+        # nothing to submit or cancel for these.
+        pending = [(op, fut) for op, fut in pending if not fut.done()]
+        if not pending:
+            return
+        ops = [op for op, _ in pending]
+        try:
+            accepted = self.context.submit(*ops)
+        except BaseException as exc:  # noqa: BLE001 (must isolate any submit()-time exception, including SystemExit/KeyboardInterrupt, since this runs as a bare call_soon callback)
+            # submit() itself can raise synchronously (e.g. a bad fd in the
+            # batch) - this runs as a bare call_soon callback, not inside
+            # any of the pending coroutines, so letting it escape here
+            # would just get logged by asyncio's default handler and leave
+            # every future in the batch unresolved forever. Every op in
+            # this batch failed together (io_submit() rejects the whole
+            # call on this kind of error, not just a prefix).
+            for op, fut in pending:
+                if not fut.done():
+                    fut.set_exception(exc)
+            return
+        # io_submit() accepts a prefix; anything past `accepted` failed.
+        for op, fut in pending[accepted:]:
+            if not fut.done():
+                fut.set_exception(OSError("Operation was not submitted"))
+
     def _on_done(self, future, result):
         """
         Allow to set result directly.
diff --git a/caio/linux_uring.c b/caio/linux_uring.c
index 29c87e4..f50dbd3 100644
--- a/caio/linux_uring.c
+++ b/caio/linux_uring.c
@@ -657,7 +657,8 @@ static int AIOContext_init(AIOContext *self, PyObject *args, PyObject *kwds) {
         params.flags = flag_table[i];
         if (params.flags & IORING_SETUP_SQPOLL)
             params.sq_thread_idle = 100;  /* ms; let kthread sleep when idle */
-        self->uring_fd = io_uring_setup(self->max_requests, &params);
+        /* +1: SQPOLL dequeue lag headroom */
+        self->uring_fd = io_uring_setup(self->max_requests + 1, &params);
         if (self->uring_fd >= 0) {
             flags_used = params.flags;
             break;
@@ -1413,10 +1414,29 @@ PyMODINIT_FUNC PyInit_linux_uring(void) {
     }
     close(probe_fd);
 
+    /* Probe whether SQPOLL is actually usable (kernel + capabilities) -
+     * exposed as SQPOLL_ALLOWED so callers can pick a default without
+     * constructing a real Context just to find out. Context(sqpoll=True)
+     * already falls back gracefully on its own either way (see
+     * flag_table_sqpoll) - this is purely informational. */
+    struct io_uring_params sqpoll_probe;
+    memset(&sqpoll_probe, 0, sizeof(sqpoll_probe));
+    sqpoll_probe.flags = IORING_SETUP_SQPOLL;
+    sqpoll_probe.sq_thread_idle = 100;
+    int sqpoll_probe_fd = io_uring_setup(1, &sqpoll_probe);
+    int sqpoll_allowed = sqpoll_probe_fd >= 0;
+    if (sqpoll_probe_fd >= 0)
+        close(sqpoll_probe_fd);
+
     PyObject *m = PyModule_Create(&linux_uring_module);
     if (m == NULL)
         return NULL;
 
+    if (PyModule_AddObject(m, "SQPOLL_ALLOWED", PyBool_FromLong(sqpoll_allowed)) < 0) {
+        Py_DECREF(m);
+        return NULL;
+    }
+
     if (PyType_Ready(&AIOContextType)   < 0) return NULL;
     if (PyType_Ready(&AIOOperationType) < 0) return NULL;
 
diff --git a/caio/linux_uring.pyi b/caio/linux_uring.pyi
index f54a06a..347b380 100644
--- a/caio/linux_uring.pyi
+++ b/caio/linux_uring.pyi
@@ -3,6 +3,10 @@ from typing import Any
 
 from .abstract import AbstractContext, AbstractOperation
 
+# True if this kernel/process can actually negotiate IORING_SETUP_SQPOLL -
+# probed once at import time (see linux_uring.c's PyInit_linux_uring).
+SQPOLL_ALLOWED: bool
+
 # noinspection PyPropertyDefinition
 class Context(AbstractContext):
     def __init__(self, max_requests: int = 32, sqpoll: bool = False): ...
diff --git a/caio/linux_uring_asyncio.py b/caio/linux_uring_asyncio.py
index b04525d..abd8949 100644
--- a/caio/linux_uring_asyncio.py
+++ b/caio/linux_uring_asyncio.py
@@ -1,14 +1,19 @@
 from .asyncio_base import AsyncioContextBase
-from .linux_uring import Context, Operation
+from .linux_uring import SQPOLL_ALLOWED, Context, Operation
 
 
 class AsyncioContext(AsyncioContextBase):
     OPERATION_CLASS = Operation
     CONTEXT_CLASS = Context
 
-    def _create_context(self, max_requests):
-        context = super()._create_context(max_requests)
+    def _create_context(self, max_requests, **kwargs):
+        # SQPOLL_ALLOWED reflects a real kernel/capability probe done once
+        # at import time (see linux_uring.c) - default to it rather than
+        # to sqpoll=False, but let an explicit caller kwarg win either way.
+        kwargs.setdefault("sqpoll", SQPOLL_ALLOWED)
+        context = super()._create_context(max_requests, **kwargs)
         self.loop.add_reader(context.fileno, self._on_read_event)
+        self._flush_scheduled = False
         return context
 
     def _on_done(self, future, result):
@@ -20,16 +25,31 @@ def _destroy_context(self):
         self.loop.remove_reader(self.context.fileno)
 
     def _on_submitted(self):
-        # Flush immediately after every submit.
-        #
-        # Non-SQPOLL (default): flush() calls io_uring_enter() which completes
-        # page-cache ops inline, then drain_cq() fires futures *before* the
-        # caller reaches `await future` — so the coroutine never suspends for
-        # those ops.  Truly async ops (real disk) leave the future unset; the
-        # coroutine suspends and the eventfd wakes it when the kernel is done.
-        #
-        # SQPOLL: flush() wakes the kernel thread if sleeping, then drain_cq().
-        # Same "fast path" benefit when the thread has already completed the op.
+        # Non-SQPOLL flush() completes page-cache ops inline (drain_cq fires
+        # futures before `await future` even suspends); SQPOLL just wakes
+        # the kernel thread if it's asleep - and only when it's actually
+        # gone idle, so eager per-op flush() is already close to syscall-
+        # free there. Batching doesn't reduce a cost that's mostly already
+        # gone - measured no meaningful difference either way - so ignore
+        # deferred whenever the kernel actually negotiated SQPOLL (not
+        # just what was requested - EPERM/EINVAL can silently fall back to
+        # a plain ring, see linux_uring.c's flag_table_sqpoll).
+        if not self.deferred or self.context.sqpoll:
+            self.context.flush()
+            return
+        # Nothing between submit()'s isinstance check and here ever awaits,
+        # so N concurrently-scheduled submits used to each flush their own
+        # single SQE back to back - no batching despite N being ready at
+        # once. call_soon() defers to the next _run_once() pass, after all
+        # of them have written their SQE, so one flush() covers the whole
+        # batch - at the cost of one extra event-loop round-trip for a lone
+        # unbatched op.
+        if not self._flush_scheduled:
+            self._flush_scheduled = True
+            self.loop.call_soon(self._deferred_flush)
+
+    def _deferred_flush(self):
+        self._flush_scheduled = False
         self.context.flush()
 
     def _on_read_event(self):
diff --git a/pyproject.toml b/pyproject.toml
index 251bf7f..8e2f163 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -38,6 +38,7 @@ develop = [
   "coveralls",
   "pytest",
   "pytest-cov",
+  "pytest-rerunfailures",
   "setuptools",
 ]
 
diff --git a/tests/conftest.py b/tests/conftest.py
index f385a28..e4cc32e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,19 +1,66 @@
+import functools
 import time
+import types
 
 import pytest
 
-from caio import python_aio, thread_aio, variants, variants_asyncio
-
-# thread_aio and python_aio only - the two backends actually backed by a
-# bounded worker-thread pool (multiprocessing.pool.ThreadPool for both),
-# where max_requests (acceptance capacity) and pool_size (worker count) are
-# genuinely independent, so submitting more than pool_size at once leaves
-# some operations sitting in the pool's own internal queue rather than
-# already dispatched. linux_aio/linux_uring have no such queue - max_requests
-# there bounds the kernel-visible I/O context directly, and neither Context
-# accepts a pool_size argument at all. thread_aio may be None (unavailable
-# on this platform/build) - filtering against variants (already None-free)
-# handles that the same way as everywhere else.
+from caio import (
+    linux_aio,
+    linux_aio_asyncio,
+    linux_uring,
+    linux_uring_asyncio,
+    python_aio,
+    thread_aio,
+    variants,
+    variants_asyncio,
+)
+
+
+def named_variant(name, **attrs):
+    ns = types.SimpleNamespace(__name__=name, **attrs)
+    return ns
+
+
+# Test-only sqpoll/deferred variants - not part of caio's public API.
+extra_context_variants = []
+extra_asyncio_variants = []
+
+if linux_uring is not None:
+    extra_context_variants.append(named_variant(
+        "linux_uring[sqpoll=True]",
+        Context=functools.partial(linux_uring.Context, sqpoll=True),
+        Operation=linux_uring.Operation,
+    ))
+    extra_asyncio_variants.append(named_variant(
+        "linux_uring[sqpoll=True]",
+        AsyncioContext=functools.partial(linux_uring_asyncio.AsyncioContext, sqpoll=True),
+    ))
+    # sqpoll=False forced: SQPOLL_ALLOWED's default would otherwise mask
+    # the deferred-batching path this variant exists to cover.
+    extra_asyncio_variants.append(named_variant(
+        "linux_uring[deferred=True]",
+        AsyncioContext=functools.partial(
+            linux_uring_asyncio.AsyncioContext, sqpoll=False, deferred=True,
+        ),
+    ))
+
+if linux_aio is not None:
+    extra_asyncio_variants.append(named_variant(
+        "linux_aio[deferred=True]",
+        AsyncioContext=functools.partial(linux_aio_asyncio.AsyncioContext, deferred=True),
+    ))
+
+all_variants = variants + tuple(extra_context_variants)
+all_variants_asyncio = variants_asyncio + tuple(extra_asyncio_variants)
+
+# extra_context_variants are always linux_uring, always polling-capable -
+# added directly rather than via hasattr() (doesn't see through partial).
+polling_variants = [
+    v for v in variants if hasattr(v.Context, "process_events")
+] + extra_context_variants
+
+# Only thread_aio/python_aio have a worker-pool queue distinct from
+# "accepted" - max_requests/pool_size are independent there.
 pooled_variants = tuple(v for v in variants if v in (thread_aio, python_aio))
 
 
@@ -39,10 +86,6 @@ def drain(ctx, want, timeout=5.0):
     return total
 
 
-def has_polling_api(backend):
-    return hasattr(backend.Context, "process_events")
-
-
 def wait_until(ctx, predicate, timeout=5.0):
     """Like drain(), but works across every backend uniformly, including
     the two purely callback-driven ones (thread_aio/python_aio, whose
@@ -63,33 +106,36 @@ def wait_until(ctx, predicate, timeout=5.0):
     return True
 
 
-@pytest.fixture(params=variants)
+@pytest.fixture(params=all_variants)
 def context_maker(request):
     return request.param.Context
 
 
-@pytest.fixture(params=variants)
+@pytest.fixture(params=all_variants)
 def operation_maker(request):
     return request.param.Operation
 
 
-@pytest.fixture(params=variants)
+@pytest.fixture(params=all_variants)
 def backend(request):
-    # Unlike using context_maker/operation_maker together (which would
-    # produce the cross product of both fixtures' independent
-    # parametrization, mismatching e.g. thread_aio's Context with
-    # linux_aio's Operation), this keeps Context/Operation from the same
-    # backend module paired together.
+    # Keeps Context/Operation paired from the same backend, unlike
+    # context_maker+operation_maker's independent cross product.
     return request.param
 
 
-@pytest.fixture(params=variants_asyncio)
+@pytest.fixture(params=all_variants_asyncio)
 def async_context_maker(request):
     return request.param.AsyncioContext
 
 
-@pytest.fixture
-def polling_backend(backend):
-    if not has_polling_api(backend):
-        pytest.skip(f"{backend.__name__} has no process_events()/poll() API")
-    return backend
+@pytest.fixture(params=all_variants_asyncio)
+async def async_context(request):
+    """Ready, already-entered AsyncioContext, default args. Use
+    async_context_maker instead for a non-default constructor kwarg."""
+    async with request.param.AsyncioContext() as context:
+        yield context
+
+
+@pytest.fixture(params=polling_variants)
+def polling_backend(request):
+    return request.param
diff --git a/tests/test_asyncio_adapter.py b/tests/test_asyncio_adapter.py
index 020aa25..2ae3dfd 100644
--- a/tests/test_asyncio_adapter.py
+++ b/tests/test_asyncio_adapter.py
@@ -8,47 +8,64 @@
 
 
 @aiomisc.timeout(5)
-async def test_adapter(tmp_path, async_context_maker):
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
+async def test_linux_uring_asyncio_forwards_context_kwargs():
+    uring_asyncio = pytest.importorskip("caio.linux_uring_asyncio")
+
+    async with uring_asyncio.AsyncioContext(
+        max_requests=8,
+        sqpoll=True,
+        deferred=True,
+    ) as context:
+        # Unsupported kernels/permissions may make the C context fall back
+        # to a regular ring. Reaching this point proves that the asyncio
+        # adapter forwarded the backend-specific option instead of rejecting
+        # it in _create_context().
+        assert context.context.sqpoll in (0, 1)
+        assert context.deferred is True
+
 
-            assert await context.read(32, fd, 0) == b""
-            s = b"Hello world"
-            assert await context.write(s, fd, 0) == len(s)
-            assert await context.read(32, fd, 0) == s
+@aiomisc.timeout(5)
+async def test_adapter(tmp_path, async_context):
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
 
-            s = b"Hello real world"
-            assert await context.write(s, fd, 0) == len(s)
-            assert await context.read(32, fd, 0) == s
+        assert await context.read(32, fd, 0) == b""
+        s = b"Hello world"
+        assert await context.write(s, fd, 0) == len(s)
+        assert await context.read(32, fd, 0) == s
 
-            part = b"\x00\x01\x02\x03"
-            limit = 32
-            expected_hash = hashlib.md5(part * limit).hexdigest()
+        s = b"Hello real world"
+        assert await context.write(s, fd, 0) == len(s)
+        assert await context.read(32, fd, 0) == s
 
-            await asyncio.gather(
-                *[context.write(part, fd, len(part) * i) for i in range(limit)]
-            )
+        part = b"\x00\x01\x02\x03"
+        limit = 32
+        expected_hash = hashlib.md5(part * limit).hexdigest()
 
-            await context.fdsync(fd)
+        await asyncio.gather(
+            *[context.write(part, fd, len(part) * i) for i in range(limit)]
+        )
 
-            data = await context.read(limit * len(part), fd, 0)
-            assert data == part * limit
+        await context.fdsync(fd)
 
-            assert hashlib.md5(bytes(data)).hexdigest() == expected_hash
+        data = await context.read(limit * len(part), fd, 0)
+        assert data == part * limit
+
+        assert hashlib.md5(bytes(data)).hexdigest() == expected_hash
 
 
 @aiomisc.timeout(3)
-async def test_bad_file_descritor(tmp_path, async_context_maker):
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
+async def test_bad_file_descritor(tmp_path, async_context):
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
 
-        with pytest.raises((SystemError, OSError, AssertionError, ValueError)):
-            assert await context.read(1, fd, 0) == b""
+    with pytest.raises((SystemError, OSError, AssertionError, ValueError)):
+        assert await context.read(1, fd, 0) == b""
 
-        with pytest.raises((SystemError, OSError, AssertionError, ValueError)):
-            assert await context.write(b"hello", fd, 0)
+    with pytest.raises((SystemError, OSError, AssertionError, ValueError)):
+        assert await context.write(b"hello", fd, 0)
 
 
 @pytest.fixture
@@ -65,129 +82,144 @@ async def asyncio_exception_handler():
 
 @aiomisc.timeout(3)
 async def test_operations_cancel_cleanly(
-    tmp_path, async_context_maker, asyncio_exception_handler
+    tmp_path, async_context, asyncio_exception_handler
 ):
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
 
-            await context.write(b"\x00", fd, 1024**2 - 1)
-            assert os.stat(fd).st_size == 1024**2
+        await context.write(b"\x00", fd, 1024**2 - 1)
+        assert os.stat(fd).st_size == 1024**2
 
-            for _ in range(50):
-                reads = [
-                    asyncio.create_task(context.read(2**16, fd, 2**16 * i))
-                    for i in range(16)
-                ]
-                _, pending = await asyncio.wait(
-                    reads, return_when=asyncio.FIRST_COMPLETED
-                )
-                for read in pending:
-                    read.cancel()
-                if pending:
-                    await asyncio.wait(pending)
-                asyncio_exception_handler.assert_not_called()
+        for _ in range(50):
+            reads = [
+                asyncio.create_task(context.read(2**16, fd, 2**16 * i))
+                for i in range(16)
+            ]
+            _, pending = await asyncio.wait(
+                reads, return_when=asyncio.FIRST_COMPLETED
+            )
+            for read in pending:
+                read.cancel()
+            if pending:
+                await asyncio.wait(pending)
+            asyncio_exception_handler.assert_not_called()
 
 
 @aiomisc.timeout(3)
 async def test_write_operations_cancel_cleanly(
-    tmp_path, async_context_maker, asyncio_exception_handler
+    tmp_path, async_context, asyncio_exception_handler
 ):
     """Mirrors test_operations_cancel_cleanly but for writes - cancellation
     handling shouldn't be read-specific."""
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
-
-            for _ in range(50):
-                writes = [
-                    asyncio.create_task(
-                        context.write(b"\x01" * 2**16, fd, 2**16 * i),
-                    )
-                    for i in range(16)
-                ]
-                _, pending = await asyncio.wait(
-                    writes, return_when=asyncio.FIRST_COMPLETED
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
+
+        for _ in range(50):
+            writes = [
+                asyncio.create_task(
+                    context.write(b"\x01" * 2**16, fd, 2**16 * i),
                 )
-                for write in pending:
-                    write.cancel()
-                if pending:
-                    await asyncio.wait(pending)
-                asyncio_exception_handler.assert_not_called()
+                for i in range(16)
+            ]
+            _, pending = await asyncio.wait(
+                writes, return_when=asyncio.FIRST_COMPLETED
+            )
+            for write in pending:
+                write.cancel()
+            if pending:
+                await asyncio.wait(pending)
+            asyncio_exception_handler.assert_not_called()
+
+
+@aiomisc.timeout(3)
+async def test_cancel_before_first_step_runs(tmp_path, async_context, asyncio_exception_handler):
+    """Cancelling right after the op's own first step (submit queued, still
+    suspended at `await future`) - covers context.cancel() raising ValueError
+    for an op the backend never actually got to submit to the kernel yet."""
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230
+        fd = fp.fileno()
+        task = asyncio.ensure_future(context.write(b"x", fd, 0))
+        await asyncio.sleep(0)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        asyncio_exception_handler.assert_not_called()
 
 
 @aiomisc.timeout(5)
-async def test_zero_byte_read_and_write(tmp_path, async_context_maker):
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
+async def test_zero_byte_read_and_write(tmp_path, async_context):
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
 
-            assert await context.write(b"", fd, 0) == 0
-            assert await context.read(0, fd, 0) == b""
+        assert await context.write(b"", fd, 0) == 0
+        assert await context.read(0, fd, 0) == b""
 
 
 @aiomisc.timeout(5)
-async def test_partial_read_at_eof(tmp_path, async_context_maker):
+async def test_partial_read_at_eof(tmp_path, async_context):
     """Requesting more bytes than the file actually has must return exactly
     what's there, not garbage/padding out to the requested size - this
     exercises the "kernel filled less than the whole buffer" slow path
     that the fast, no-copy path (used when the buffer is filled
     completely) doesn't."""
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
 
-            payload = b"hello"
-            assert await context.write(payload, fd, 0) == len(payload)
+        payload = b"hello"
+        assert await context.write(payload, fd, 0) == len(payload)
 
-            data = await context.read(len(payload) * 20, fd, 0)
-            assert data == payload
-            assert len(data) == len(payload)
+        data = await context.read(len(payload) * 20, fd, 0)
+        assert data == payload
+        assert len(data) == len(payload)
 
 
 @aiomisc.timeout(5)
-async def test_fsync_and_fdsync(tmp_path, async_context_maker):
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
+async def test_fsync_and_fdsync(tmp_path, async_context):
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
 
-            await context.write(b"data", fd, 0)
-            # Neither call should raise; the return value is deliberately
-            # not asserted beyond that - it's None on some backends and
-            # b"" on others (python_aio), both meaningless here.
-            await context.fsync(fd)
-            await context.fdsync(fd)
+        await context.write(b"data", fd, 0)
+        # Return value not asserted - None on some backends, b"" on others
+        # (python_aio), both meaningless here.
+        await context.fsync(fd)
+        await context.fdsync(fd)
 
 
 @aiomisc.timeout(15)
-async def test_large_transfer(tmp_path, async_context_maker):
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
+async def test_large_transfer(tmp_path, async_context):
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
 
-            payload = os.urandom(4 * 1024 * 1024)
-            expected_hash = hashlib.sha256(payload).hexdigest()
+        payload = os.urandom(4 * 1024 * 1024)
+        expected_hash = hashlib.sha256(payload).hexdigest()
 
-            written = await context.write(payload, fd, 0)
-            assert written == len(payload)
+        written = await context.write(payload, fd, 0)
+        assert written == len(payload)
 
-            data = await context.read(len(payload), fd, 0)
-            assert len(data) == len(payload)
-            assert hashlib.sha256(bytes(data)).hexdigest() == expected_hash
+        data = await context.read(len(payload), fd, 0)
+        assert len(data) == len(payload)
+        assert hashlib.sha256(bytes(data)).hexdigest() == expected_hash
 
 
 @aiomisc.timeout(5)
-async def test_write_extends_file_sparsely(tmp_path, async_context_maker):
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
+async def test_write_extends_file_sparsely(tmp_path, async_context):
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
 
-            hole_size = 4 * 1024 * 1024
-            await context.write(b"\x01", fd, hole_size)
-            assert os.stat(fd).st_size == hole_size + 1
+        hole_size = 4 * 1024 * 1024
+        await context.write(b"\x01", fd, hole_size)
+        assert os.stat(fd).st_size == hole_size + 1
 
-            hole = await context.read(hole_size, fd, 0)
-            assert hole == b"\x00" * hole_size
+        hole = await context.read(hole_size, fd, 0)
+        assert hole == b"\x00" * hole_size
 
 
 @aiomisc.timeout(10)
@@ -219,32 +251,32 @@ async def test_max_requests_backpressure(tmp_path, async_context_maker):
 
 
 @aiomisc.timeout(10)
-async def test_concurrent_non_overlapping_chunks(tmp_path, async_context_maker):
+async def test_concurrent_non_overlapping_chunks(tmp_path, async_context):
     """Writes distinct, non-overlapping regions concurrently, then reads
     them back concurrently - if any backend's buffer handling ever aliased
     two in-flight operations' memory (e.g. a zero-copy fast path reused
     somewhere it shouldn't), this would show up as data from the wrong
     chunk appearing in the wrong place."""
-    async with async_context_maker() as context:
-        with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
-            fd = fp.fileno()
-
-            chunk = 8192
-            count = 32
-            expected = [
-                bytes([(i * 7 + 3) % 256]) * chunk for i in range(count)
+    context = async_context
+    with open(str(tmp_path / "temp.bin"), "wb+") as fp:  # noqa: ASYNC230 (brief sync setup, not the operation under test)
+        fd = fp.fileno()
+
+        chunk = 8192
+        count = 32
+        expected = [
+            bytes([(i * 7 + 3) % 256]) * chunk for i in range(count)
+        ]
+
+        await asyncio.gather(
+            *[
+                context.write(expected[i], fd, i * chunk)
+                for i in range(count)
             ]
+        )
 
-            await asyncio.gather(
-                *[
-                    context.write(expected[i], fd, i * chunk)
-                    for i in range(count)
-                ]
-            )
-
-            results = await asyncio.gather(
-                *[context.read(chunk, fd, i * chunk) for i in range(count)]
-            )
+        results = await asyncio.gather(
+            *[context.read(chunk, fd, i * chunk) for i in range(count)]
+        )
 
-            for i, (got, want) in enumerate(zip(results, expected)):
-                assert got == want, f"chunk {i} mismatch"
+        for i, (got, want) in enumerate(zip(results, expected)):
+            assert got == want, f"chunk {i} mismatch"
diff --git a/tests/test_raw_low_level.py b/tests/test_raw_low_level.py
index 0ff4641..905e48d 100644
--- a/tests/test_raw_low_level.py
+++ b/tests/test_raw_low_level.py
@@ -80,6 +80,9 @@ def test_process_events_respects_timeout_and_releases_gil(polling_backend):
     proving the GIL was actually released, not just that the call happened
     to return quickly for some other reason.
     """
+    if not polling_backend.__name__.startswith("caio."):
+        pytest.skip("synthetic test-only variant, not importable by name in a subprocess")
+
     code = (
         "import threading, time\n"
         f"import {polling_backend.__name__} as m\n"
@@ -119,14 +122,11 @@ def test_process_events_respects_timeout_and_releases_gil(polling_backend):
     )
 
 
-def test_raw_poll_reflects_completions(tmp_path, backend):
-    if not hasattr(backend.Context, "poll"):
-        pytest.skip(f"{backend.__name__} has no poll() API")
-
+def test_raw_poll_reflects_completions(tmp_path, polling_backend):
     with open(str(tmp_path / "temp.bin"), "wb+") as f:
         fd = f.fileno()
-        ctx = backend.Context(max_requests=8)
-        op = backend.Operation.write(b"hi", fd, 0)
+        ctx = polling_backend.Context(max_requests=8)
+        op = polling_backend.Operation.write(b"hi", fd, 0)
         ctx.submit(op)
 
         # poll() drains the eventfd counter, so it must not be called
@@ -146,6 +146,9 @@ def test_poll_does_not_block_when_nothing_pending(polling_backend):
     would hold the GIL hostage forever regardless of what the test's own
     main thread does.
     """
+    if not polling_backend.__name__.startswith("caio."):
+        pytest.skip("synthetic test-only variant, not importable by name in a subprocess")
+
     code = (
         f"import {polling_backend.__name__} as m\n"
         f"ctx = m.Context(max_requests=8)\n"
@@ -269,6 +272,8 @@ def test_read_with_absurd_nbytes_raises_cleanly(backend):
     """
     if backend.__name__ == "caio.python_aio":
         pytest.skip("python_aio doesn't preallocate at construction time")
+    if not backend.__name__.startswith("caio."):
+        pytest.skip("synthetic test-only variant, not importable by name in a subprocess")
 
     code = (
         f"import {backend.__name__} as m\n"
@@ -982,6 +987,7 @@ def test_linux_aio_process_events_max_requests_is_bounded(backend):
     )
 
 
+@pytest.mark.flaky(reruns=3)
 def test_process_events_negative_timeout_waits_indefinitely(tmp_path, polling_backend):
     """process_events(timeout<0) means "wait indefinitely" - matching the
     native convention each backend's own blocking primitive already uses
diff --git a/uv.lock b/uv.lock
index 57107d1..a4f0d70 100644
--- a/uv.lock
+++ b/uv.lock
@@ -14,9 +14,9 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 dependencies = [
-    { name = "colorlog", marker = "python_full_version < '3.11'" },
-    { name = "logging-journald", marker = "python_full_version < '3.11' and sys_platform == 'linux'" },
-    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
+    { name = "colorlog" },
+    { name = "logging-journald", marker = "sys_platform == 'linux'" },
+    { name = "typing-extensions" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/ce/c1/c3f0838d23e60eb7695a0054ac9a964db6d2b5f0649d96068e0515f5bb0b/aiomisc-17.10.3.tar.gz", hash = "sha256:db891af7cd96c78064d93956bb0ea9bdf44b878ee5f1026f1c3fac73e2cec556", size = 466598, upload-time = "2026-01-13T10:04:21.684Z" }
 wheels = [
@@ -31,9 +31,9 @@ resolution-markers = [
     "python_full_version >= '3.11'",
 ]
 dependencies = [
-    { name = "aiothreads", marker = "python_full_version >= '3.11'" },
-    { name = "colorlog", marker = "python_full_version >= '3.11'" },
-    { name = "logging-journald", marker = "python_full_version >= '3.11' and sys_platform == 'linux'" },
+    { name = "aiothreads" },
+    { name = "colorlog" },
+    { name = "logging-journald", marker = "sys_platform == 'linux'" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/2d/3f/8533fe0b8483f04c2289cec3da83f48ddbebfd5bd8618d006770e68bfe3d/aiomisc-18.0.6.tar.gz", hash = "sha256:bd282a1e5606010c787a2adecb37e7641115a8e23c5ec4a509de6d8800cef38d", size = 459177, upload-time = "2026-02-09T00:34:22.433Z" }
 wheels = [
@@ -48,8 +48,8 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 dependencies = [
-    { name = "aiomisc", version = "17.10.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
-    { name = "pytest", version = "8.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
+    { name = "aiomisc", version = "17.10.3", source = { registry = "https://pypi.org/simple" } },
+    { name = "pytest", version = "8.4.2", source = { registry = "https://pypi.org/simple" } },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/31/29/961940714a4c747a0454e511427959e114a819602250e1e934905f76d1f9/aiomisc_pytest-1.3.4.tar.gz", hash = "sha256:f4e7f5a5251c322221933ece5b97ab178a0395ffc006469a98304f83bf966c11", size = 11528, upload-time = "2025-07-30T09:42:58.711Z" }
 wheels = [
@@ -64,8 +64,8 @@ resolution-markers = [
     "python_full_version >= '3.11'",
 ]
 dependencies = [
-    { name = "aiomisc", version = "18.0.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
-    { name = "pytest", version = "9.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
+    { name = "aiomisc", version = "18.0.6", source = { registry = "https://pypi.org/simple" } },
+    { name = "pytest", version = "9.0.2", source = { registry = "https://pypi.org/simple" } },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/5b/41/083ee3de2aafdf20895bac41a535f3425852098a5130ed1681ce2a57ef9e/aiomisc_pytest-2.0.0.tar.gz", hash = "sha256:71b62b90ee984523c57e18e0757cee540db063947212659c1be9b2199bb54c65", size = 6032, upload-time = "2026-02-09T00:09:06.732Z" }
 wheels = [
@@ -103,6 +103,7 @@ develop = [
     { name = "pytest", version = "8.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
     { name = "pytest", version = "9.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
     { name = "pytest-cov" },
+    { name = "pytest-rerunfailures" },
     { name = "setuptools" },
 ]
 
@@ -117,6 +118,7 @@ requires-dist = [
     { name = "coveralls", marker = "extra == 'develop'" },
     { name = "pytest", marker = "extra == 'develop'" },
     { name = "pytest-cov", marker = "extra == 'develop'" },
+    { name = "pytest-rerunfailures", marker = "extra == 'develop'" },
     { name = "setuptools", marker = "extra == 'develop'" },
 ]
 provides-extras = ["develop"]
@@ -392,7 +394,7 @@ name = "exceptiongroup"
 version = "1.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
+    { name = "typing-extensions" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" }
 wheels = [
@@ -482,13 +484,13 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 dependencies = [
-    { name = "colorama", marker = "python_full_version < '3.11' and sys_platform == 'win32'" },
-    { name = "exceptiongroup", marker = "python_full_version < '3.11'" },
-    { name = "iniconfig", marker = "python_full_version < '3.11'" },
-    { name = "packaging", marker = "python_full_version < '3.11'" },
-    { name = "pluggy", marker = "python_full_version < '3.11'" },
-    { name = "pygments", marker = "python_full_version < '3.11'" },
-    { name = "tomli", marker = "python_full_version < '3.11'" },
+    { name = "colorama", marker = "sys_platform == 'win32'" },
+    { name = "exceptiongroup" },
+    { name = "iniconfig" },
+    { name = "packaging" },
+    { name = "pluggy" },
+    { name = "pygments" },
+    { name = "tomli" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/a3/5c/00a0e072241553e1a7496d638deababa67c5058571567b92a7eaa258397c/pytest-8.4.2.tar.gz", hash = "sha256:86c0d0b93306b961d58d62a4db4879f27fe25513d4b969df351abdddb3c30e01", size = 1519618, upload-time = "2025-09-04T14:34:22.711Z" }
 wheels = [
@@ -503,11 +505,11 @@ resolution-markers = [
     "python_full_version >= '3.11'",
 ]
 dependencies = [
-    { name = "colorama", marker = "python_full_version >= '3.11' and sys_platform == 'win32'" },
-    { name = "iniconfig", marker = "python_full_version >= '3.11'" },
-    { name = "packaging", marker = "python_full_version >= '3.11'" },
-    { name = "pluggy", marker = "python_full_version >= '3.11'" },
-    { name = "pygments", marker = "python_full_version >= '3.11'" },
+    { name = "colorama", marker = "sys_platform == 'win32'" },
+    { name = "iniconfig" },
+    { name = "packaging" },
+    { name = "pluggy" },
+    { name = "pygments" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" }
 wheels = [
@@ -529,6 +531,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ee/49/1377b49de7d0c1ce41292161ea0f721913fa8722c19fb9c1e3aa0367eecb/pytest_cov-7.0.0-py3-none-any.whl", hash = "sha256:3b8e9558b16cc1479da72058bdecf8073661c7f57f7d3c5f22a1c23507f2d861", size = 22424, upload-time = "2025-09-09T10:57:00.695Z" },
 ]
 
+[[package]]
+name = "pytest-rerunfailures"
+version = "16.4"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "packaging" },
+    { name = "pytest", version = "8.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
+    { name = "pytest", version = "9.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/47/60/a90ca1cc6cffcb97b4260ed0ad2b7934b999d7c48abe4ea0840344862a3b/pytest_rerunfailures-16.4.tar.gz", hash = "sha256:8222d17c37eb7b9e4d6fc96a3c724ff4e1a5c97a5cc7cbb2c19e9282cfd21a11", size = 36635, upload-time = "2026-07-01T06:30:56.813Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ce/93/3cdcc4033444e822e01b573414b03fd37fd5533070c750477b8f5fa5224b/pytest_rerunfailures-16.4-py3-none-any.whl", hash = "sha256:f69b5beb39622c90d1e44bd945d826eff6db545dcf0b68f52b7e4ad15eaf6d6c", size = 16955, upload-time = "2026-07-01T06:30:55.333Z" },
+]
+
 [[package]]
 name = "requests"
 version = "2.33.0"