diff --git a/.gitignore b/.gitignore
index 82c0ba33..8e52b11c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,6 +10,9 @@ clients/ruby/pkg/
 clients/ruby/.bundle/
 clients/ruby/*.gem
 
+# Benchmark harness deps (bun.lock is committed; deps are not)
+benchmark/partition-keys/node_modules/
+
 # Claude Code agent worktrees (ephemeral isolation per agent run)
 .claude/worktrees/
 
diff --git a/benchmark/partition-keys/README.md b/benchmark/partition-keys/README.md
new file mode 100644
index 00000000..73d25ef2
--- /dev/null
+++ b/benchmark/partition-keys/README.md
@@ -0,0 +1,103 @@
+# Partition-keys read-amplification benchmark
+
+Exercises the pgque v0.8 **partition keys** feature (`sql/pgque-api/partition_keys.sql`)
+at a high-volume multi-tenant scale — the read-amplification scenario in
+`blueprints/partition-keys/SPEC.md` §14 (S4). It uses **only the real installed
+pgque API** (`send` keyed / `subscribe_slot` / `claim_slot` /
+`receive_partitioned` / `ack_partitioned` / `release_slot` and the
+`pgque.partition_slot_status` view) — no demo schema.
+
+## What it measures
+
+The slot mechanism is N independent slot consumers, each scanning the full
+event stream and filtering server-side to its hash class. That gives two
+properties this bench quantifies:
+
+- **R2 — read amplification ~N×.** Each produced event is scanned by all N
+  slots, so buffers touched per produced event scale ~linearly with N. Measured
+  from `pg_stat_statements` (under `track = top`, the event-table scan buffers
+  roll up into the top-level `receive_partitioned` call), at **N=16 vs N=32**.
+- **R7 — a stalled slot pins rotation for the whole queue.** One slot's worker
+  is SIGSTOPped mid-run; its subscription cursor freezes, so the engine cannot
+  drop old `event_N_M` tables. The bench tracks per-slot lag growth, the
+  rotation floor (queue table count), and the catch-up slope after resume.
+
+## Target profile (Fabrizio: >400M events/day)
+
+- **Producer:** 5,000 ev/s sustained keyed `send`, Zipfian tenant skew (s=1.1)
+  over 2,000 tenants, ~200-byte JSON payload — see `producer.sql`. Rate-limited
+  pgbench (`-R 5000 -c 16 -j 8`).
+- **Consumers:** N slot workers (one per slot) on the lease loop — `slot_worker.ts`
+  (bun + node-postgres). Each worker claims its slot, sticky-drains it in
+  batches of 500, acks, releases at drain, re-polls after 200 ms idle. One ack
+  log line per batch: `ts,worker,slot,events,max_ev_id`.
+- **Ticker:** `pk_ticker.py` — `pgque.ticker()` every 250 ms, `pgque.maint()`
+  every 60 s.
+
+## Phases
+
+`run_bench.sh` runs, all output under `/tmp/bench/pk/<phase>/`:
+
+1. **steady-16** (30 min): `bench_q`, consumer `w16`, 16 slots, 5k ev/s.
+2. **steady-32** (30 min): fresh `bench_q32`, consumer `w32`, 32 slots, 5k ev/s.
+3. **stalled-16** (15 min): reuse `bench_q`; slot 7's worker is SIGSTOPped at
+   minute 2 and resumed at minute 10.
+
+Each phase runs, at 5 s / 10 s / 30 s cadences: `slot_status_sampler.sh`
+(per-slot lease + lag from `partition_slot_status`, and queue-level throughput
++ table count from `get_queue_info`), `sys_metrics_sampler.py`,
+`pg_stat_statements_snapshot.py`, and `bloat_sampler.py`. `pg_stat_statements`
+is reset at each phase start and snapshotted at the boundary for the read-amp
+measurement.
+
+Then `summarize.py` parses the CSVs into `summary.md` — producer/consume
+throughput, per-slot pending percentiles, CPU/mem, the N-scaling read-amp
+table, the stalled-slot timeline, and a headline table to paste into a PR.
+
+## Running
+
+On a fresh Hetzner CCX43 (16 dedicated cores, 64 GiB, local NVMe, Ubuntu 24.04),
+Postgres 18 from PGDG:
+
+```bash
+# from the operator machine: ship the repo, then bootstrap
+rsync -a --exclude .git ./ root@VM:/root/pgque/
+ssh root@VM 'bash /root/pgque/benchmark/partition-keys/setup_vm.sh'
+
+# on the VM: install driver deps and run the full ~90 min of measured phases
+ssh root@VM 'cd /root/pgque/benchmark/partition-keys && bun install && \
+  PGUSER=postgres DEVICE=sda bash run_bench.sh'
+```
+
+`DEVICE` is the block device name in `/proc/diskstats` (Hetzner cloud volumes
+usually show as `sda`; adjust if the NVMe is named differently).
+
+### Knobs (env)
+
+`RATE` (5000), `N_SLOTS` (16; steady-32 uses 2×), `DURATION_MIN` (30),
+`STALL_MIN` (15), `STALL_ON_MIN`/`STALL_OFF_MIN`/`STALL_SLOT`, `ROTATION`
+(`30 seconds`), `BATCH` (500), `TTL_S` (30), `PGB_C`/`PGB_J`, `PHASES`
+(`1,2,3`), `OUT` (`/tmp/bench/pk`), and libpq `PGHOST`/`PGUSER`/`PGDATABASE`.
+
+### Smoke run
+
+A 60-second micro-run of phase 1 at 200 ev/s with 4 slots — just env vars:
+
+```bash
+PGHOST=/tmp PGUSER="$(id -un)" PGDATABASE=bench \
+  RATE=200 N_SLOTS=4 DURATION_MIN=1 PHASES=1 DEVICE=none \
+  bash run_bench.sh
+```
+
+## Files
+
+| File | Role |
+|---|---|
+| `setup_vm.sh` | VM bootstrap: PG18 + tuning + bun + pgque install |
+| `producer.sql` | pgbench keyed-send script (`@QUEUE@` rendered at runtime) |
+| `slot_worker.ts` | bun slot-worker lease-loop driver |
+| `package.json` | driver deps (`pg`) |
+| `pk_ticker.py` | ticker + maint loop |
+| `slot_status_sampler.sh` | per-slot lease/lag + queue-rate sampler |
+| `run_bench.sh` | phase orchestrator |
+| `summarize.py` | CSV → markdown report |
diff --git a/benchmark/partition-keys/bun.lock b/benchmark/partition-keys/bun.lock
new file mode 100644
index 00000000..bd2700f0
--- /dev/null
+++ b/benchmark/partition-keys/bun.lock
@@ -0,0 +1,50 @@
+{
+  "lockfileVersion": 1,
+  "configVersion": 1,
+  "workspaces": {
+    "": {
+      "name": "pgque-partition-keys-bench",
+      "dependencies": {
+        "pg": "^8.16.0",
+      },
+      "devDependencies": {
+        "@types/pg": "^8.15.2",
+      },
+    },
+  },
+  "packages": {
+    "@types/node": ["@types/node@26.1.0", "", { "dependencies": { "undici-types": "~8.3.0" } }, "sha512-O0A1G3xPGy4w7AgQdAQYUlQ+BKk2Oovw8eRpofyp5KdBZULnbe+WqaOVNrm705SHphCiG4XHsACrSmPu1f+Kgw=="],
+
+    "@types/pg": ["@types/pg@8.20.0", "", { "dependencies": { "@types/node": "*", "pg-protocol": "*", "pg-types": "^2.2.0" } }, "sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow=="],
+
+    "pg": ["pg@8.22.0", "", { "dependencies": { "pg-connection-string": "^2.14.0", "pg-pool": "^3.14.0", "pg-protocol": "^1.15.0", "pg-types": "2.2.0", "pgpass": "1.0.5" }, "optionalDependencies": { "pg-cloudflare": "^1.4.0" }, "peerDependencies": { "pg-native": ">=3.0.1" }, "optionalPeers": ["pg-native"] }, "sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA=="],
+
+    "pg-cloudflare": ["pg-cloudflare@1.4.0", "", {}, "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A=="],
+
+    "pg-connection-string": ["pg-connection-string@2.14.0", "", {}, "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg=="],
+
+    "pg-int8": ["pg-int8@1.0.1", "", {}, "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw=="],
+
+    "pg-pool": ["pg-pool@3.14.0", "", { "peerDependencies": { "pg": ">=8.0" } }, "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw=="],
+
+    "pg-protocol": ["pg-protocol@1.15.0", "", {}, "sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ=="],
+
+    "pg-types": ["pg-types@2.2.0", "", { "dependencies": { "pg-int8": "1.0.1", "postgres-array": "~2.0.0", "postgres-bytea": "~1.0.0", "postgres-date": "~1.0.4", "postgres-interval": "^1.1.0" } }, "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA=="],
+
+    "pgpass": ["pgpass@1.0.5", "", { "dependencies": { "split2": "^4.1.0" } }, "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug=="],
+
+    "postgres-array": ["postgres-array@2.0.0", "", {}, "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA=="],
+
+    "postgres-bytea": ["postgres-bytea@1.0.1", "", {}, "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ=="],
+
+    "postgres-date": ["postgres-date@1.0.7", "", {}, "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q=="],
+
+    "postgres-interval": ["postgres-interval@1.2.0", "", { "dependencies": { "xtend": "^4.0.0" } }, "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ=="],
+
+    "split2": ["split2@4.2.0", "", {}, "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg=="],
+
+    "undici-types": ["undici-types@8.3.0", "", {}, "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ=="],
+
+    "xtend": ["xtend@4.0.2", "", {}, "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ=="],
+  }
+}
diff --git a/benchmark/partition-keys/package.json b/benchmark/partition-keys/package.json
new file mode 100644
index 00000000..b0ea7a93
--- /dev/null
+++ b/benchmark/partition-keys/package.json
@@ -0,0 +1,15 @@
+{
+  "name": "pgque-partition-keys-bench",
+  "private": true,
+  "type": "module",
+  "description": "Slot-worker driver for the partition-keys read-amplification benchmark (bun + node-postgres)",
+  "scripts": {
+    "worker": "bun slot_worker.ts"
+  },
+  "dependencies": {
+    "pg": "^8.16.0"
+  },
+  "devDependencies": {
+    "@types/pg": "^8.15.2"
+  }
+}
diff --git a/benchmark/partition-keys/pk_ticker.py b/benchmark/partition-keys/pk_ticker.py
new file mode 100644
index 00000000..5636a090
--- /dev/null
+++ b/benchmark/partition-keys/pk_ticker.py
@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""pk_ticker.py -- tight ticker + maintenance loop for the partition-keys bench.
+
+Calls pgque.ticker() every TICK_MS (default 250 ms) so slot cursors advance
+promptly, and pgque.maint() every MAINT_S (default 60 s) so event tables rotate
+and vacuum on the pgque cadence. Persistent autocommit connection, mirroring
+tooling/pgq_ticker_daemon.py.
+
+Env: PGHOST/PGDATABASE/PGUSER (libpq), TICK_MS, MAINT_S, RUN_S (0 = forever).
+"""
+import os
+import signal
+import sys
+import time
+
+import psycopg2
+
+DSN = (
+    f"host={os.environ.get('PGHOST', '127.0.0.1')} "
+    f"dbname={os.environ.get('PGDATABASE', 'bench')} "
+    f"user={os.environ.get('PGUSER', 'postgres')} "
+    "application_name=pk_ticker"
+)
+TICK_S = float(os.environ.get("TICK_MS", "250")) / 1000.0
+MAINT_S = float(os.environ.get("MAINT_S", "60"))
+RUN_S = float(os.environ.get("RUN_S", "0"))
+
+conn = psycopg2.connect(DSN)
+conn.autocommit = True
+cur = conn.cursor()
+
+
+def shutdown(signum, frame):
+    try:
+        conn.close()
+    except Exception:
+        pass
+    sys.exit(0)
+
+
+signal.signal(signal.SIGTERM, shutdown)
+signal.signal(signal.SIGINT, shutdown)
+
+print(f"pk_ticker: tick={TICK_S}s maint={MAINT_S}s run={RUN_S or 'forever'}", flush=True)
+t_start = time.monotonic()
+last_maint = 0.0
+while True:
+    now = time.monotonic()
+    if RUN_S and now - t_start >= RUN_S:
+        break
+    try:
+        cur.execute("select pgque.ticker()")
+        if now - last_maint >= MAINT_S:
+            cur.execute("select pgque.maint()")
+            last_maint = now
+    except Exception as e:  # noqa: BLE001
+        print(f"pk_ticker err: {e}", file=sys.stderr, flush=True)
+        time.sleep(1)
+    time.sleep(TICK_S)
+
+conn.close()
diff --git a/benchmark/partition-keys/producer.sql b/benchmark/partition-keys/producer.sql
new file mode 100644
index 00000000..d78a85a4
--- /dev/null
+++ b/benchmark/partition-keys/producer.sql
@@ -0,0 +1,24 @@
+-- producer.sql -- keyed pgbench producer for the partition-keys read-amp bench.
+--
+-- Zipfian tenant skew (s=1.1) over 2000 tenants models a high-volume multi-tenant
+-- profile: a heavy head of hot buckets plus a long tail. Each send carries a
+-- ~200-byte JSON payload and a partition key of 'tenant-<n>'. The key rides
+-- ev_extra1 (SPEC D1) so every slot's server-side hash filter routes it.
+--
+-- @QUEUE@ is rendered to the live queue name by run_bench.sh (pgbench has no
+-- string-literal variable substitution, so the runner seds this placeholder).
+\set tenant random_zipfian(1, 2000, 1.1)
+select pgque.send(
+    '@QUEUE@',
+    'StorageObjectCreated',
+    json_build_object(
+        'tenant', 'tenant-' || :tenant,
+        'bucket', 'bkt-' || (:tenant % 32),
+        'object', md5(random()::text) || '/' || md5(random()::text) || '.bin',
+        'size_bytes', (random() * 10485760)::bigint,
+        'content_type', 'application/octet-stream',
+        'etag', md5(random()::text),
+        'created_at', clock_timestamp()
+    )::text,
+    'tenant-' || :tenant
+);
diff --git a/benchmark/partition-keys/results/r1-ccx43-summary.md b/benchmark/partition-keys/results/r1-ccx43-summary.md
new file mode 100644
index 00000000..c18d4f96
--- /dev/null
+++ b/benchmark/partition-keys/results/r1-ccx43-summary.md
@@ -0,0 +1,108 @@
+# Partition-keys read-amplification benchmark
+
+High-volume multi-tenant profile: keyed producer + N slot workers on the real
+pgque v0.8 lease API. Maps to SPEC R2 (read amplification ~N x) and R7
+(a stalled slot pins rotation for the whole queue).
+
+## Phase `steady-16`  (N=16, target 5000 ev/s, 30 min)
+
+- Producer: **5,000 ev/s** achieved, 8,996,791 events sent
+- Consumers: **4,999 ev/s** end-to-end, 8,998,023 events acked
+- CPU: 25% avg / 26% peak | RAM used: 3.5 / 5.4 GiB
+- Pending events across slots: p50 0, p99 1,320, max 1,332
+- Read-amp (receive_partitioned): 39,458,647 buffers (39,458,647 hit / 0 read) over 388,732 calls = **4.39 buffers/event**
+- Per-slot consumed: { w0:288,446, w1:303,241, w2:267,539, w3:896,810, w4:499,753, w5:257,939, w6:530,210, w7:327,033, w8:787,695, w9:1,843,421, w10:845,659, w11:862,379, w12:283,429, w13:353,890, w14:298,668, w15:351,911 }
+
+**Bloat & dead tuples**
+
+- Metadata tables by peak dead tuples (end size):
+    - `pgque.subscription`: peak 7,586 dead (end 7,572 dead, 704.00 KiB)
+    - `pgque.tick`: peak 1,608 dead (end 0 dead, 888.00 KiB)
+    - `pgque.partition_slot`: peak 557 dead (end 363 dead, 96.00 KiB)
+    - `pgque.partition_consumer`: peak 35 dead (end 35 dead, 32.00 KiB)
+    - `pgque.consumer`: peak 20 dead (end 20 dead, 48.00 KiB)
+- `pgque.partition_slot` (v0.8 lease, 2 UPDATEs/batch): peak **557** dead, end 363 dead, 96.00 KiB — HOT updates should keep this tiny
+- `pgque.tick` peak dead tuples: 1,608 (rotation-metadata churn)
+- `pgque.subscription` peak dead tuples: 7,586 (rotation-metadata churn)
+- Event tables: 4 present at phase start -> 4 at end (rotation dropping old `event_N_M`)
+- Event-table bytes: start 56.00 KiB, end 3.57 GiB, peak 3.57 GiB
+- Event-table peak dead tuples: 0 (append-only — expect ~0; nonzero means retries)
+- Total DB footprint: end 3.57 GiB, peak 3.57 GiB
+
+## Phase `steady-32`  (N=32, target 5000 ev/s, 30 min)
+
+- Producer: **5,001 ev/s** achieved, 8,997,460 events sent
+- Consumers: **5,000 ev/s** end-to-end, 9,000,840 events acked
+- CPU: 30% avg / 31% peak | RAM used: 7.3 / 9.3 GiB
+- Pending events across slots: p50 0, p99 1,318, max 1,337
+- Read-amp (receive_partitioned): 63,166,134 buffers (63,166,134 hit / 0 read) over 775,996 calls = **7.02 buffers/event**
+- Per-slot consumed: { w0:145,553, w1:142,039, w2:185,695, w3:451,545, w4:95,106, w5:116,060, w6:196,430, w7:166,868, w8:331,859, w9:1,707,562, w10:118,269, w11:782,539, w12:121,190, w13:219,230, w14:122,795, w15:211,793, w16:143,608, w17:160,798, w18:82,634, w19:447,463, w20:403,448, w21:141,426, w22:335,823, w23:159,336, w24:453,227, w25:136,178, w26:728,446, w27:80,067, w28:161,406, w29:135,008, w30:176,196, w31:141,243 }
+
+**Bloat & dead tuples**
+
+- Metadata tables by peak dead tuples (end size):
+    - `pgque.subscription`: peak 9,502 dead (end 1,878 dead, 1.26 MiB)
+    - `pgque.partition_slot`: peak 704 dead (end 578 dead, 96.00 KiB)
+    - `pgque.partition_consumer`: peak 66 dead (end 0 dead, 64.00 KiB)
+    - `pgque.consumer`: peak 20 dead (end 20 dead, 48.00 KiB)
+    - `pgque.queue`: peak 8 dead (end 8 dead, 48.00 KiB)
+- `pgque.partition_slot` (v0.8 lease, 2 UPDATEs/batch): peak **704** dead, end 578 dead, 96.00 KiB — HOT updates should keep this tiny
+- `pgque.tick` peak dead tuples: 0 (rotation-metadata churn)
+- `pgque.subscription` peak dead tuples: 9,502 (rotation-metadata churn)
+- Event tables: 8 present at phase start -> 8 at end (rotation dropping old `event_N_M`)
+- Event-table bytes: start 3.62 GiB, end 7.19 GiB, peak 7.19 GiB
+- Event-table peak dead tuples: 0 (append-only — expect ~0; nonzero means retries)
+- Total DB footprint: end 7.19 GiB, peak 7.19 GiB
+
+## Phase `stalled-16`  (N=16, target 5000 ev/s, 15 min)
+
+- Producer: **4,997 ev/s** achieved, 4,495,477 events sent
+- Consumers: **4,996 ev/s** end-to-end, 4,496,137 events acked
+- CPU: 25% avg / 30% peak | RAM used: 8.9 / 9.0 GiB
+- Pending events across slots: p50 0, p99 1,665,413, max 2,386,485
+- Read-amp (receive_partitioned): 13,361,585 buffers (13,361,585 hit / 0 read) over 189,979 calls = **2.97 buffers/event**
+- Per-slot consumed: { w0:143,742, w1:150,971, w2:133,571, w3:449,575, w4:249,337, w5:128,238, w6:265,385, w7:163,156, w8:392,808, w9:921,401, w10:423,363, w11:431,755, w12:141,882, w13:176,922, w14:148,272, w15:175,759 }
+
+**Bloat & dead tuples**
+
+- Metadata tables by peak dead tuples (end size):
+    - `pgque.subscription`: peak 8,912 dead (end 1,672 dead, 1.05 MiB)
+    - `pgque.tick`: peak 7,205 dead (end 0 dead, 1.70 MiB)
+    - `pgque.partition_slot`: peak 640 dead (end 329 dead, 96.00 KiB)
+    - `pgque.consumer`: peak 36 dead (end 36 dead, 48.00 KiB)
+    - `pgque.partition_consumer`: peak 16 dead (end 16 dead, 64.00 KiB)
+- `pgque.partition_slot` (v0.8 lease, 2 UPDATEs/batch): peak **640** dead, end 329 dead, 96.00 KiB — HOT updates should keep this tiny
+- `pgque.tick` peak dead tuples: 7,205 (rotation-metadata churn)
+- `pgque.subscription` peak dead tuples: 8,912 (rotation-metadata churn)
+- Event tables: 8 present at phase start -> 8 at end (rotation dropping old `event_N_M`)
+- Event-table bytes: start 3.62 GiB, end 5.38 GiB, peak 5.38 GiB
+- Event-table peak dead tuples: 0 (append-only — expect ~0; nonzero means retries)
+- Total DB footprint: end 5.38 GiB, peak 5.38 GiB
+- R7 pin/release — event-table bytes: stall start 3.62 GiB -> stall end 5.38 GiB -> phase end 5.38 GiB (grows while the stalled cursor pins rotation, drops after resume)
+
+## Read amplification: N-scaling (SPEC R2)
+
+| N | buffers/event | read/event | ratio vs smallest N |
+|--:|--:|--:|--:|
+| 16 | 4.39 | 0.00 | 1.00x |
+| 32 | 7.02 | 0.00 | 1.60x |
+
+Every slot scans the full stream and filters server-side, so buffers touched per produced event scale ~linearly with N. Observed 32/16 = 1.60x (ideal 2x).
+
+## Stalled-slot: rotation pinning (SPEC R7)
+
+Stalled slot (by peak lag): **slot 7**.
+- Lease-expiry stall window: 896s (no live owner)
+- Pending events: baseline 0 -> peak 2,386,485
+- Lag growth during stall: **2,663 events/s**
+- Catch-up after resume: did not return to baseline within the phase
+- Rotation floor: queue held 3..3 event tables (the stalled slot pins the drop floor)
+
+## Headline
+
+| Phase | N | Producer ev/s | Consume ev/s | Pending p99 | Buffers/event | CPU peak | Peak dead tup (meta) | Event tbl GiB end |
+|---|--:|--:|--:|--:|--:|--:|--:|--:|
+| steady-16 | 16 | 5,000 | 4,999 | 1,320 | 4.39 | 26% | 7,586 | 3.57 |
+| steady-32 | 32 | 5,001 | 5,000 | 1,318 | 7.02 | 31% | 9,502 | 7.19 |
+| stalled-16 | 16 | 4,997 | 4,996 | 1,665,413 | 2.97 | 30% | 8,912 | 5.38 |
+
diff --git a/benchmark/partition-keys/run_bench.sh b/benchmark/partition-keys/run_bench.sh
new file mode 100644
index 00000000..04ac3d3d
--- /dev/null
+++ b/benchmark/partition-keys/run_bench.sh
@@ -0,0 +1,229 @@
+#!/usr/bin/env bash
+# run_bench.sh -- partition-keys read-amplification benchmark (SPEC section 14
+# S4) at a high-volume multi-tenant profile: sustained keyed producer + N slot
+# workers driving the real pgque v0.8 lease API, measured at N=16 and N=32,
+# plus a stalled-slot phase that exercises the R7 rotation-pinning failure mode.
+#
+# Assumes pgque is already installed into $PGDATABASE (setup_vm.sh does this).
+# Everything lands under $OUT (default /tmp/bench/pk); a phases.csv manifest
+# ties each phase directory to its metadata for summarize.py.
+#
+# Knobs (env): RATE, N_SLOTS, DURATION_MIN drive the whole run; the smoke run
+# is just those three plus PHASES=1. See README.md.
+set -Eeuo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+TOOLING="${TOOLING:-$SCRIPT_DIR/../tooling}"
+
+# --- connection ------------------------------------------------------------
+export PGHOST="${PGHOST:-127.0.0.1}"
+export PGUSER="${PGUSER:-postgres}"
+export PGDATABASE="${PGDATABASE:-bench}"
+
+# --- profile ---------------------------------------------------------------
+RATE="${RATE:-5000}"                 # producer sends/s
+N_SLOTS="${N_SLOTS:-16}"             # base slot count (steady-16); steady-32 = 2x
+DURATION_MIN="${DURATION_MIN:-30}"   # steady-phase minutes
+STALL_MIN="${STALL_MIN:-15}"         # stalled-slot phase minutes
+STALL_ON_MIN="${STALL_ON_MIN:-2}"    # minute the target slot is SIGSTOPped
+STALL_OFF_MIN="${STALL_OFF_MIN:-10}" # minute the target slot is resumed
+STALL_SLOT="${STALL_SLOT:-7}"        # which slot to stall
+ROTATION="${ROTATION:-30 seconds}"   # queue rotation period (fast => visible floor)
+BATCH="${BATCH:-500}"                # receive_partitioned max
+TTL_S="${TTL_S:-30}"                 # lease ttl
+PGB_C="${PGB_C:-16}"                 # pgbench clients
+PGB_J="${PGB_J:-8}"                  # pgbench threads
+DEVICE="${DEVICE:-sda}"              # block device for sys_metrics_sampler
+PHASES="${PHASES:-1,2,3}"            # which measured phases to run
+OUT="${OUT:-/tmp/bench/pk}"
+
+BUN="${BUN:-bun}"
+PYTHON="${PYTHON:-python3}"
+
+N1="$N_SLOTS"
+N2=$(( N_SLOTS * 2 ))
+DUR_S=$(( DURATION_MIN * 60 ))
+STALL_DUR_S=$(( STALL_MIN * 60 ))
+
+mkdir -p "$OUT"
+MANIFEST="$OUT/phases.csv"
+echo "label,queue,consumer,n,rate,dur_s,start_iso,end_iso,dir" > "$MANIFEST"
+
+log() { echo "[$(date -u +%FT%TZ)] $*"; }
+
+psql_q() { psql -X -q -v ON_ERROR_STOP=1 -c "$1"; }
+
+# Kill background jobs of this run on exit; -CONT any stopped worker first.
+CLEAN_PIDS=()
+cleanup() {
+  for p in "${CLEAN_PIDS[@]:-}"; do
+    [[ -n "$p" ]] && kill -CONT "$p" 2>/dev/null || true
+    [[ -n "$p" ]] && kill "$p" 2>/dev/null || true
+  done
+}
+trap cleanup EXIT
+
+ensure_partitioned() {
+  # ensure_partitioned <queue> <consumer> <n>
+  local queue="$1" consumer="$2" n="$3" k
+  # Force-drop any pre-existing queue: a leftover subscription (e.g. from a
+  # smoke run) has a frozen cursor and pins rotation for the WHOLE queue
+  # (SPEC R7), silently distorting every measured phase.
+  psql -X -q -c "select pgque.drop_queue('$queue', true)" >/dev/null 2>&1 || true
+  psql -X -q -c "select pgque.create_queue('$queue')" >/dev/null 2>&1 || true
+  psql_q "update pgque.queue set queue_rotation_period = '$ROTATION'::interval where queue_name = '$queue'"
+  for (( k = 0; k < n; k++ )); do
+    psql_q "select pgque.subscribe_slot('$queue', '$consumer', $k, $n)" >/dev/null
+  done
+  log "queue '$queue' ready: consumer '$consumer', $n slots, rotation '$ROTATION'"
+}
+
+rest_slots_spec() {
+  # rest_slots_spec <stall_slot> <n>  -> prints slot spec excluding stall_slot
+  local s="$1" n="$2"
+  if (( s == 0 )); then
+    echo "1-$(( n - 1 ))"
+  elif (( s == n - 1 )); then
+    echo "0-$(( n - 2 ))"
+  else
+    echo "0-$(( s - 1 )),$(( s + 1 ))-$(( n - 1 ))"
+  fi
+}
+
+run_phase() {
+  # run_phase <label> <queue> <consumer> <n> <dur_s> <stall_slot|-1>
+  local label="$1" queue="$2" consumer="$3" n="$4" dur="$5" stall="$6"
+  local dir="$OUT/$label"
+  mkdir -p "$dir"
+
+  # Render the producer script with the live queue name.
+  sed "s/@QUEUE@/$queue/g" "$SCRIPT_DIR/producer.sql" > "$dir/producer.sql"
+
+  psql_q "select pg_stat_statements_reset()" >/dev/null
+  local start_iso
+  start_iso="$(date -u +%FT%TZ)"
+  log "PHASE $label start: queue=$queue consumer=$consumer n=$n rate=$RATE dur=${dur}s stall_slot=$stall"
+
+  # --- samplers ------------------------------------------------------------
+  QUEUE="$queue" CONSUMER="$consumer" INTERVAL=5 DURATION="$dur" OUTDIR="$dir" \
+    bash "$SCRIPT_DIR/slot_status_sampler.sh" >"$dir/slot_status_sampler.log" 2>&1 &
+  CLEAN_PIDS+=($!)
+  "$PYTHON" "$TOOLING/sys_metrics_sampler.py" --interval 10 --duration "$dur" \
+    --device "$DEVICE" --out "$dir/sys_metrics.csv" >"$dir/sys_metrics.log" 2>&1 &
+  CLEAN_PIDS+=($!)
+  "$PYTHON" "$TOOLING/pg_stat_statements_snapshot.py" --interval 10 --duration "$dur" \
+    --dsn "host=$PGHOST dbname=$PGDATABASE user=$PGUSER" \
+    --out "$dir/pgss_timeseries.csv" >"$dir/pgss_ts.log" 2>&1 &
+  CLEAN_PIDS+=($!)
+  DSN="host=$PGHOST dbname=$PGDATABASE user=$PGUSER" \
+    "$PYTHON" "$TOOLING/bloat_sampler.py" --system pgque --interval 30 --duration "$dur" \
+    >"$dir/bloat.csv" 2>"$dir/bloat.log" &
+  CLEAN_PIDS+=($!)
+
+  # --- ticker --------------------------------------------------------------
+  RUN_S="$dur" TICK_MS=250 MAINT_S=60 \
+    "$PYTHON" "$SCRIPT_DIR/pk_ticker.py" >"$dir/ticker.log" 2>&1 &
+  CLEAN_PIDS+=($!)
+
+  # --- producer ------------------------------------------------------------
+  pgbench -h "$PGHOST" -U "$PGUSER" -d "$PGDATABASE" -n -f "$dir/producer.sql" \
+    -c "$PGB_C" -j "$PGB_J" -R "$RATE" -T "$dur" -P 30 \
+    --aggregate-interval=10 --log --log-prefix="$dir/producer_agg" \
+    >"$dir/producer.log" 2>&1 &
+  local prod_pid=$!
+  CLEAN_PIDS+=("$prod_pid")
+
+  # --- consumers -----------------------------------------------------------
+  local acks="$dir/acks.log"
+  : > "$acks"
+  if (( stall < 0 )); then
+    QUEUE="$queue" CONSUMER="$consumer" N_SLOTS="$n" SLOTS="0-$(( n - 1 ))" \
+      BATCH="$BATCH" TTL_S="$TTL_S" LOG="$acks" RUN_S="$dur" \
+      "$BUN" "$SCRIPT_DIR/slot_worker.ts" >"$dir/worker.log" 2>&1 &
+    CLEAN_PIDS+=($!)
+  else
+    # Rest of the slots as one process; the target slot as its own process so
+    # it can be SIGSTOPped (a genuine stalled worker -- SPEC R7).
+    local rest
+    rest="$(rest_slots_spec "$stall" "$n")"
+    QUEUE="$queue" CONSUMER="$consumer" N_SLOTS="$n" SLOTS="$rest" \
+      BATCH="$BATCH" TTL_S="$TTL_S" LOG="$acks" RUN_S="$dur" \
+      "$BUN" "$SCRIPT_DIR/slot_worker.ts" >"$dir/worker_rest.log" 2>&1 &
+    CLEAN_PIDS+=($!)
+    QUEUE="$queue" CONSUMER="$consumer" N_SLOTS="$n" SLOTS="$stall" \
+      BATCH="$BATCH" TTL_S="$TTL_S" LOG="$acks" RUN_S="$dur" \
+      "$BUN" "$SCRIPT_DIR/slot_worker.ts" >"$dir/worker_stall.log" 2>&1 &
+    local stall_pid=$!
+    CLEAN_PIDS+=("$stall_pid")
+    # Schedule the stall window: STOP at STALL_ON_MIN, CONT at STALL_OFF_MIN.
+    (
+      sleep $(( STALL_ON_MIN * 60 ))
+      echo "[$(date -u +%FT%TZ)] SIGSTOP slot $stall (pid $stall_pid)" >>"$dir/stall.log"
+      kill -STOP "$stall_pid" 2>/dev/null || true
+      sleep $(( (STALL_OFF_MIN - STALL_ON_MIN) * 60 ))
+      echo "[$(date -u +%FT%TZ)] SIGCONT slot $stall (pid $stall_pid)" >>"$dir/stall.log"
+      kill -CONT "$stall_pid" 2>/dev/null || true
+    ) &
+    CLEAN_PIDS+=($!)
+  fi
+
+  # Producer bounds the phase (-T). Consumers/samplers self-terminate at RUN_S.
+  wait "$prod_pid" 2>/dev/null || true
+  log "PHASE $label producer done; letting consumers/samplers flush"
+  # The consumers and samplers self-terminate at RUN_S/--duration; give them a
+  # moment to flush their last writes before the boundary snapshot.
+  sleep 5
+
+  # --- read-amp boundary snapshot -----------------------------------------
+  # Under pg_stat_statements.track=top the event-table scan buffers roll up
+  # into the top-level receive_partitioned call, so these normalized rows are
+  # the read-amp measurement (no per-batch cursor eviction).
+  psql -X -q -c "\copy (
+      select left(regexp_replace(query, '\s+', ' ', 'g'), 80) as query_head,
+             calls, rows, shared_blks_hit, shared_blks_read,
+             round(total_exec_time::numeric, 1) as total_exec_time_ms
+        from pg_stat_statements
+       where query ~ 'receive_partitioned|ack_partitioned|claim_slot|release_slot|pgque\.ticker|pgque\.send|pgque\.maint'
+       order by shared_blks_hit + shared_blks_read desc
+    ) to '$dir/pgss_end.csv' csv header" >/dev/null 2>&1 || true
+
+  local end_iso
+  end_iso="$(date -u +%FT%TZ)"
+  echo "$label,$queue,$consumer,$n,$RATE,$dur,$start_iso,$end_iso,$dir" >> "$MANIFEST"
+  log "PHASE $label end"
+
+  # Reap this phase's background jobs before the next phase.
+  cleanup
+  CLEAN_PIDS=()
+  wait 2>/dev/null || true
+}
+
+# ---------------------------------------------------------------------------
+# Phase dispatch
+# ---------------------------------------------------------------------------
+has_phase() { [[ ",$PHASES," == *",$1,"* ]]; }
+
+log "=== partition-keys bench: RATE=$RATE N1=$N1 N2=$N2 DURATION_MIN=$DURATION_MIN PHASES=$PHASES ==="
+
+if has_phase 1; then
+  ensure_partitioned "bench_q" "w$N1" "$N1"
+  run_phase "steady-$N1" "bench_q" "w$N1" "$N1" "$DUR_S" -1
+fi
+
+if has_phase 2; then
+  ensure_partitioned "bench_q32" "w$N2" "$N2"
+  run_phase "steady-$N2" "bench_q32" "w$N2" "$N2" "$DUR_S" -1
+fi
+
+if has_phase 3; then
+  # Reuse (or create) the N1 queue; stall one slot mid-phase.
+  ensure_partitioned "bench_q" "w$N1" "$N1"
+  run_phase "stalled-$N1" "bench_q" "w$N1" "$N1" "$STALL_DUR_S" "$STALL_SLOT"
+fi
+
+log "=== all phases done; summarizing ==="
+"$PYTHON" "$SCRIPT_DIR/summarize.py" --out "$OUT" > "$OUT/summary.md" 2>"$OUT/summarize.log" || {
+  log "summarize failed; see $OUT/summarize.log"
+}
+log "summary written to $OUT/summary.md"
+cat "$OUT/summary.md" 2>/dev/null || true
diff --git a/benchmark/partition-keys/setup_vm.sh b/benchmark/partition-keys/setup_vm.sh
new file mode 100644
index 00000000..27362a1e
--- /dev/null
+++ b/benchmark/partition-keys/setup_vm.sh
@@ -0,0 +1,90 @@
+#!/usr/bin/env bash
+# setup_vm.sh -- minimal bootstrap for the partition-keys bench on a fresh
+# Hetzner CCX43 (16 dedicated cores, 64 GiB, local NVMe, Ubuntu 24.04).
+#
+# Derived from ../install/bootstrap.sh but stripped of AWS-isms: no NVMe
+# re-mount, no pg_ash/pgfr. Installs PostgreSQL 18 (PGDG), tunes it for the
+# bench, installs bun, and loads pgque from the LOCAL repo checkout into db
+# 'bench'. The repo (including sql/pgque.sql and benchmark/) is expected to be
+# rsynced to $REPO_DIR first; run this as root over ssh.
+#
+#   rsync -a --exclude .git ./ root@VM:/root/pgque/
+#   ssh root@VM 'bash /root/pgque/benchmark/partition-keys/setup_vm.sh'
+set -Eeuo pipefail
+export DEBIAN_FRONTEND=noninteractive
+
+REPO_DIR="${REPO_DIR:-/root/pgque}"
+PGQUE_SQL="${PGQUE_SQL:-$REPO_DIR/sql/pgque.sql}"
+PG_MAJOR="${PG_MAJOR:-18}"
+PGDATABASE="${PGDATABASE:-bench}"
+CONF="/etc/postgresql/$PG_MAJOR/main/postgresql.conf"
+HBA="/etc/postgresql/$PG_MAJOR/main/pg_hba.conf"
+
+echo "=== [$(hostname)] partition-keys setup_vm start $(date -u +%FT%TZ) ==="
+[[ -f "$PGQUE_SQL" ]] || { echo "cannot find $PGQUE_SQL -- rsync the repo to $REPO_DIR first" >&2; exit 1; }
+
+# Wait for any cloud-init apt locks to clear.
+for _ in $(seq 1 60); do
+  if ! fuser /var/lib/dpkg/lock-frontend >/dev/null 2>&1; then break; fi
+  echo "  waiting for dpkg lock..."
+  sleep 5
+done
+
+apt-get update -qq
+apt-get install -y -qq curl gnupg lsb-release git python3 python3-psycopg2 unzip
+
+# PGDG
+curl -fsSL https://www.postgresql.org/media/keys/ACCC4CF8.asc \
+  | gpg --batch --yes --dearmor -o /usr/share/keyrings/postgresql.gpg
+echo "deb [signed-by=/usr/share/keyrings/postgresql.gpg] https://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" \
+  | tee /etc/apt/sources.list.d/postgresql.list >/dev/null
+apt-get update -qq
+apt-get install -y -qq "postgresql-$PG_MAJOR"
+
+# bun (for the slot-worker driver)
+if [[ ! -x "$HOME/.bun/bin/bun" ]]; then
+  curl -fsSL https://bun.sh/install | bash
+fi
+export PATH="$HOME/.bun/bin:$PATH"
+
+# --- postgresql.conf tuning ------------------------------------------------
+tee -a "$CONF" >/dev/null <<'CONF'
+
+# ── partition-keys bench tuning ──────────────────────────────────────────────
+shared_preload_libraries = 'pg_stat_statements'
+shared_buffers = 16GB
+effective_cache_size = 48GB
+max_connections = 200
+
+synchronous_commit = off
+wal_level = minimal
+max_wal_senders = 0
+max_wal_size = 16GB
+checkpoint_timeout = 15min
+wal_compression = on
+
+random_page_cost = 1.1
+effective_io_concurrency = 200
+jit = off
+listen_addresses = 'localhost'
+CONF
+
+# trust local for bench workers (postgres over 127.0.0.1)
+sed -i '/^host.*127.0.0.1.*scram-sha-256/i host all postgres 127.0.0.1/32 trust' "$HBA" || true
+
+systemctl restart "postgresql@$PG_MAJOR-main"
+
+# --- database + pgque ------------------------------------------------------
+sudo -u postgres psql -tAc "select 1 from pg_database where datname='$PGDATABASE'" | grep -q 1 \
+  || sudo -u postgres psql -c "create database $PGDATABASE;"
+sudo -u postgres psql -d "$PGDATABASE" -c "create extension if not exists pg_stat_statements;"
+
+echo "==> installing pgque from $PGQUE_SQL"
+# copy to a postgres-readable path: $PGQUE_SQL may live under /root
+install -m 0644 "$PGQUE_SQL" /tmp/pgque_install.sql
+sudo -u postgres psql -v ON_ERROR_STOP=1 -q -d "$PGDATABASE" -f /tmp/pgque_install.sql >/dev/null
+rm -f /tmp/pgque_install.sql
+
+echo "=== [$(hostname)] partition-keys setup_vm DONE $(date -u +%FT%TZ) ==="
+echo "Next: cd $REPO_DIR/benchmark/partition-keys && bun install && \\"
+echo "      PGUSER=postgres bash run_bench.sh"
diff --git a/benchmark/partition-keys/slot_status_sampler.sh b/benchmark/partition-keys/slot_status_sampler.sh
new file mode 100644
index 00000000..1db03177
--- /dev/null
+++ b/benchmark/partition-keys/slot_status_sampler.sh
@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+# slot_status_sampler.sh -- per-slot lease + lag sampler for the partition-keys
+# bench. Every INTERVAL seconds it appends two CSVs:
+#
+#   $OUTDIR/slot_status.csv   ts,queue,consumer,slot,lease_owner,epoch,pending_events
+#       one row per slot of the consumer, from pgque.partition_slot_status --
+#       this is the R7 rotation-pinning signal (a stalled slot shows a growing
+#       pending_events and, after its lease expires, a null lease_owner).
+#
+#   $OUTDIR/queue_rate.csv    ts,queue,ev_per_sec,ev_new,last_tick_id,ntables,cur_table
+#       queue-level throughput + table count from pgque.get_queue_info(queue).
+#       ntables/cur_table are the rotation-floor evidence: a pinned slot keeps
+#       the engine from dropping old event_N_M tables, so ntables climbs.
+set -Eeuo pipefail
+
+QUEUE="${QUEUE:-bench_q}"
+CONSUMER="${CONSUMER:-w16}"
+INTERVAL="${INTERVAL:-5}"
+DURATION="${DURATION:-3600}"
+OUTDIR="${OUTDIR:-/tmp/bench/pk}"
+PGHOST="${PGHOST:-127.0.0.1}"
+PGDATABASE="${PGDATABASE:-bench}"
+PGUSER="${PGUSER:-postgres}"
+export PGHOST PGDATABASE PGUSER
+
+mkdir -p "$OUTDIR"
+slot_csv="$OUTDIR/slot_status.csv"
+rate_csv="$OUTDIR/queue_rate.csv"
+[[ -f "$slot_csv" ]] || echo "ts,queue,consumer,slot,lease_owner,epoch,pending_events" > "$slot_csv"
+[[ -f "$rate_csv" ]] || echo "ts,queue,ev_per_sec,ev_new,last_tick_id,ntables,cur_table" > "$rate_csv"
+
+end=$(( $(date +%s) + DURATION ))
+while [[ $(date +%s) -lt $end ]]; do
+  ts=$(date -u +%FT%TZ)
+
+  psql -X -q -At -F',' -d "$PGDATABASE" \
+    -c "select '${ts}', queue_name, consumer, slot, coalesce(lease_owner,''), epoch, pending_events
+          from pgque.partition_slot_status
+         where queue_name = '${QUEUE}' and consumer = '${CONSUMER}'
+         order by slot" \
+    >> "$slot_csv" 2>/dev/null || true
+
+  psql -X -q -At -F',' -d "$PGDATABASE" \
+    -c "select '${ts}', queue_name, round(ev_per_sec::numeric, 1), ev_new, last_tick_id,
+               queue_ntables, queue_cur_table
+          from pgque.get_queue_info('${QUEUE}')" \
+    >> "$rate_csv" 2>/dev/null || true
+
+  sleep "$INTERVAL"
+done
diff --git a/benchmark/partition-keys/slot_worker.ts b/benchmark/partition-keys/slot_worker.ts
new file mode 100644
index 00000000..f6e3be62
--- /dev/null
+++ b/benchmark/partition-keys/slot_worker.ts
@@ -0,0 +1,149 @@
+/**
+ * slot_worker.ts -- partition-keys slot-worker driver (bun + node-postgres).
+ *
+ * Drives the LEASE consume loop against the REAL installed pgque v0.8 API
+ * (sql/pgque-api/partition_keys.sql) -- claim_slot / receive_partitioned /
+ * ack_partitioned / release_slot. No demo schema; every call is a plain
+ * autocommit query, so it works exactly as a transaction-mode-pooled worker
+ * would.
+ *
+ * One async loop per slot, each on its own connection, each with a stable
+ * worker id `${WORKER_PREFIX}${slot}`. A loop:
+ *   1. claim_slot(queue, consumer, slot, worker, ttl); null => back off, retry.
+ *   2. sticky drain: receive_partitioned(max=BATCH); while it returns events,
+ *      ack_partitioned and log one line per ack; renewal rides receive/ack.
+ *   3. after IDLE_ROUNDS empty polls, release_slot and sleep IDLE_MS, then
+ *      re-claim and poll again.
+ *
+ * The runner runs steady phases as ONE process covering all slots
+ * (SLOTS=0-<N-1>) and the stalled-slot phase as two processes -- the target
+ * slot alone (so it can be SIGSTOPped) plus the rest -- via the SLOTS spec.
+ *
+ * Env (all optional except where noted):
+ *   PGHOST / PGDATABASE / PGUSER   libpq connection (defaults: bench db)
+ *   QUEUE          queue name                       (default bench_q)
+ *   CONSUMER       partitioned consumer name         (default w16)
+ *   N_SLOTS        pinned slot count N               (default 16)
+ *   SLOTS          slot spec, e.g. "0-15" | "7" | "0-6,8-15" (default 0..N-1)
+ *   BATCH          receive_partitioned max           (default 500)
+ *   TTL_S          lease ttl seconds                 (default 30)
+ *   IDLE_MS        sleep between empty drain cycles   (default 200)
+ *   IDLE_ROUNDS    empty polls before release        (default 2)
+ *   WORKER_PREFIX  worker-id prefix                  (default w)
+ *   LOG            ack-log file to append; '' = stdout (default stdout)
+ *   RUN_S          run seconds then exit; 0 = until SIGTERM (default 0)
+ */
+import pg from "pg";
+import { appendFileSync } from "node:fs";
+const { Client } = pg;
+
+const QUEUE = process.env.QUEUE || "bench_q";
+const CONSUMER = process.env.CONSUMER || "w16";
+const N_SLOTS = Number(process.env.N_SLOTS || 16);
+const BATCH = Number(process.env.BATCH || 500);
+const TTL_S = Number(process.env.TTL_S || 30);
+const IDLE_MS = Number(process.env.IDLE_MS || 200);
+const IDLE_ROUNDS = Number(process.env.IDLE_ROUNDS || 2);
+const WORKER_PREFIX = process.env.WORKER_PREFIX || "w";
+const LOG = process.env.LOG || "";
+const RUN_S = Number(process.env.RUN_S || 0);
+
+let stopped = false;
+process.on("SIGTERM", () => (stopped = true));
+process.on("SIGINT", () => (stopped = true));
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+function parseSlots(spec: string | undefined, n: number): number[] {
+  if (!spec) return Array.from({ length: n }, (_, i) => i);
+  const out: number[] = [];
+  for (const part of spec.split(",")) {
+    const t = part.trim();
+    if (t === "") continue;
+    if (t.includes("-")) {
+      const [a, b] = t.split("-").map((x) => Number(x));
+      for (let i = a; i <= b; i++) out.push(i);
+    } else {
+      out.push(Number(t));
+    }
+  }
+  return out;
+}
+
+function logAck(line: string) {
+  if (LOG) appendFileSync(LOG, line + "\n");
+  else console.log(line);
+}
+
+function newClient() {
+  return new Client({
+    host: process.env.PGHOST,
+    database: process.env.PGDATABASE || "bench",
+    user: process.env.PGUSER || process.env.USER || "postgres",
+  });
+}
+
+const deadline = () => (RUN_S > 0 ? Date.now() + RUN_S * 1000 : Infinity);
+
+async function slotLoop(slot: number, endAt: number): Promise<void> {
+  const worker = `${WORKER_PREFIX}${slot}`;
+  const c = newClient();
+  await c.connect();
+  try {
+    while (!stopped && Date.now() < endAt) {
+      const claim = await c.query(
+        "select pgque.claim_slot($1,$2,$3,$4, make_interval(secs => $5)) as epoch",
+        [QUEUE, CONSUMER, slot, worker, TTL_S],
+      );
+      if (claim.rows[0].epoch === null) {
+        // Held by another live worker (only happens across processes) -- wait.
+        await sleep(IDLE_MS);
+        continue;
+      }
+
+      let idle = 0;
+      while (!stopped && Date.now() < endAt) {
+        const r = await c.query(
+          "select msg_id from pgque.receive_partitioned($1,$2,$3,$4,$5,$6)",
+          [QUEUE, CONSUMER, slot, N_SLOTS, worker, BATCH],
+        );
+        const got = r.rows.length;
+        // ack finishes the whole batch (finish_batch); a no-op when the empty
+        // filtered batch was already auto-finished inside receive_partitioned.
+        await c.query("select pgque.ack_partitioned($1,$2,$3,$4,$5)", [
+          QUEUE,
+          CONSUMER,
+          slot,
+          N_SLOTS,
+          worker,
+        ]);
+        if (got > 0) {
+          let maxId = 0;
+          for (const row of r.rows) if (+row.msg_id > maxId) maxId = +row.msg_id;
+          logAck(`${new Date().toISOString()},${worker},${slot},${got},${maxId}`);
+          idle = 0;
+        } else if (++idle >= IDLE_ROUNDS) {
+          break;
+        }
+      }
+
+      await c.query("select pgque.release_slot($1,$2,$3,$4)", [
+        QUEUE,
+        CONSUMER,
+        slot,
+        worker,
+      ]);
+      await sleep(IDLE_MS);
+    }
+  } finally {
+    await c.end().catch(() => {});
+  }
+}
+
+const slots = parseSlots(process.env.SLOTS, N_SLOTS);
+const endAt = deadline();
+console.error(
+  `slot_worker: queue=${QUEUE} consumer=${CONSUMER} N=${N_SLOTS} slots=[${slots.join(",")}] batch=${BATCH} ttl=${TTL_S}s run=${RUN_S || "until-SIGTERM"}`,
+);
+await Promise.all(slots.map((s) => slotLoop(s, endAt)));
+console.error("slot_worker: all slot loops exited");
diff --git a/benchmark/partition-keys/summarize.py b/benchmark/partition-keys/summarize.py
new file mode 100644
index 00000000..ee49a192
--- /dev/null
+++ b/benchmark/partition-keys/summarize.py
@@ -0,0 +1,512 @@
+#!/usr/bin/env python3
+"""summarize.py -- turn the partition-keys bench CSVs into a markdown report.
+
+Reads $OUT/phases.csv (the manifest run_bench.sh writes) and, per phase
+directory, parses:
+  producer_agg.*      pgbench aggregate log -> produced events, achieved TPS
+  acks.log            slot-worker ack log   -> consumed events, throughput
+  slot_status.csv     per-slot lease + lag  -> pending_events percentiles,
+                                               stalled-slot timeline
+  queue_rate.csv      get_queue_info         -> rotation-floor (ntables) evidence
+  sys_metrics.csv     CPU/mem sampler        -> peak+avg CPU and memory
+  pgss_end.csv        pg_stat_statements      -> read-amp (buffers per event)
+
+Output: a markdown summary ending in a headline table ready to paste into a
+GitHub PR comment. Maps directly to SPEC R2 (read amplification ~N x) and R7
+(a stalled slot pins rotation for the whole queue).
+
+Stdlib only.
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import glob
+import os
+import re
+import statistics
+from datetime import datetime
+
+# Rotating event tables: parents (event_N) and children (event_N_M). Everything
+# else in bloat.csv is a metadata table.
+EVENT_TBL_RE = re.compile(r"^pgque\.event_\d+(_\d+)?$")
+
+
+def read_csv(path: str) -> list[dict]:
+    if not os.path.exists(path):
+        return []
+    with open(path, newline="") as f:
+        return list(csv.DictReader(f))
+
+
+def parse_iso(s: str) -> datetime | None:
+    try:
+        return datetime.fromisoformat(s.replace("Z", "+00:00"))
+    except Exception:
+        return None
+
+
+def pctile(xs: list[float], p: float) -> float:
+    if not xs:
+        return 0.0
+    xs = sorted(xs)
+    if len(xs) == 1:
+        return xs[0]
+    k = (len(xs) - 1) * p
+    lo = int(k)
+    hi = min(lo + 1, len(xs) - 1)
+    return xs[lo] + (xs[hi] - xs[lo]) * (k - lo)
+
+
+def fmt(n: float) -> str:
+    return f"{n:,.0f}"
+
+
+def fmt_bytes(n: float) -> str:
+    """Binary-unit size (CLAUDE.md: always KiB/MiB/GiB/TiB)."""
+    n = float(n)
+    for unit in ("B", "KiB", "MiB", "GiB"):
+        if abs(n) < 1024 or unit == "GiB":
+            return f"{n:.0f} B" if unit == "B" else f"{n:.2f} {unit}"
+        n /= 1024
+    return f"{n:.2f} TiB"
+
+
+def gib(n: float) -> float:
+    return float(n) / (1024 ** 3)
+
+
+def _bloat_int(v) -> int:
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _nearest_ts(all_ts: list[datetime], target: datetime) -> datetime:
+    return min(all_ts, key=lambda t: abs((t - target).total_seconds()))
+
+
+def bloat_stats(phase_dir: str) -> dict | None:
+    """Parse bloat.csv into dead-tuple / footprint stats.
+
+    Self-contained: depends on bloat.csv alone, so the bloat section renders
+    even when the other phase CSVs are absent. Returns None if there is no
+    bloat.csv (caller notes the section was skipped).
+    """
+    rows = read_csv(os.path.join(phase_dir, "bloat.csv"))
+    if not rows:
+        return None
+    per_table: dict[str, list[dict]] = {}
+    parsed: list[dict] = []
+    ts_set: set[datetime] = set()
+    for r in rows:
+        ts = parse_iso(r.get("ts", ""))
+        if ts is None:
+            continue
+        rec = dict(
+            ts=ts,
+            table=r.get("table", ""),
+            n_live=_bloat_int(r.get("n_live_tup")),
+            n_dead=_bloat_int(r.get("n_dead_tup")),
+            total=_bloat_int(r.get("total_bytes")),
+        )
+        parsed.append(rec)
+        per_table.setdefault(rec["table"], []).append(rec)
+        ts_set.add(ts)
+    if not parsed:
+        return None
+    all_ts = sorted(ts_set)
+    first_ts, last_ts = all_ts[0], all_ts[-1]
+    for recs in per_table.values():
+        recs.sort(key=lambda x: x["ts"])
+
+    # Split metadata vs rotating event tables.
+    meta: dict[str, dict] = {}
+    event_tables: set[str] = set()
+    for tbl, recs in per_table.items():
+        if EVENT_TBL_RE.match(tbl):
+            event_tables.add(tbl)
+            continue
+        last = recs[-1]
+        meta[tbl] = dict(
+            peak_dead=max(x["n_dead"] for x in recs),
+            end_dead=last["n_dead"],
+            end_bytes=last["total"],
+        )
+
+    # Event-table aggregates per timestamp: summed bytes, present-count, and the
+    # peak dead-tuple count across all event tables (append-only -> want ~0).
+    ev_bytes_by_ts = {t: 0 for t in all_ts}
+    ev_present_by_ts = {t: 0 for t in all_ts}
+    ev_peak_dead = 0
+    for tbl in event_tables:
+        for rec in per_table[tbl]:
+            ev_bytes_by_ts[rec["ts"]] += rec["total"]
+            ev_present_by_ts[rec["ts"]] += 1
+            ev_peak_dead = max(ev_peak_dead, rec["n_dead"])
+
+    total_by_ts = {t: 0 for t in all_ts}
+    for rec in parsed:
+        total_by_ts[rec["ts"]] += rec["total"]
+
+    return dict(
+        all_ts=all_ts,
+        meta=meta,
+        ev_bytes_by_ts=ev_bytes_by_ts,
+        ev_peak_dead=ev_peak_dead,
+        ev_bytes_start=ev_bytes_by_ts[first_ts],
+        ev_bytes_end=ev_bytes_by_ts[last_ts],
+        ev_bytes_peak=max(ev_bytes_by_ts.values()),
+        ev_present_start=ev_present_by_ts[first_ts],
+        ev_present_end=ev_present_by_ts[last_ts],
+        total_end=total_by_ts[last_ts],
+        total_peak=max(total_by_ts.values()),
+        meta_peak_dead=max((v["peak_dead"] for v in meta.values()), default=0),
+    )
+
+
+# ---------------------------------------------------------------------------
+def produced_events(phase_dir: str) -> int:
+    """Sum pgbench aggregate-log transaction counts (field 2)."""
+    total = 0
+    for path in glob.glob(os.path.join(phase_dir, "producer_agg.*")):
+        with open(path) as f:
+            for ln in f:
+                parts = ln.split()
+                if len(parts) >= 2:
+                    try:
+                        total += int(parts[1])
+                    except ValueError:
+                        pass
+    return total
+
+
+def achieved_tps(phase_dir: str) -> float:
+    for path in glob.glob(os.path.join(phase_dir, "producer.log")):
+        with open(path) as f:
+            for ln in f:
+                if ln.strip().startswith("tps ="):
+                    try:
+                        return float(ln.split("=")[1].split("(")[0].strip())
+                    except (IndexError, ValueError):
+                        pass
+    return 0.0
+
+
+def consumed(phase_dir: str) -> tuple[int, int, dict[str, int]]:
+    """(total_events, ack_count, per_worker_events) from acks.log."""
+    total = 0
+    acks = 0
+    per_worker: dict[str, int] = {}
+    path = os.path.join(phase_dir, "acks.log")
+    if not os.path.exists(path):
+        return 0, 0, {}
+    with open(path) as f:
+        for ln in f:
+            parts = ln.strip().split(",")
+            if len(parts) < 4:
+                continue
+            worker, ev = parts[1], parts[3]
+            try:
+                e = int(ev)
+            except ValueError:
+                continue
+            total += e
+            acks += 1
+            per_worker[worker] = per_worker.get(worker, 0) + e
+    return total, acks, per_worker
+
+
+def pending_by_slot(rows: list[dict]) -> dict[int, list[float]]:
+    out: dict[int, list[float]] = {}
+    for r in rows:
+        try:
+            slot = int(r["slot"])
+            pe = float(r["pending_events"])
+        except (KeyError, ValueError):
+            continue
+        out.setdefault(slot, []).append(pe)
+    return out
+
+
+def cpu_mem(phase_dir: str) -> dict:
+    rows = read_csv(os.path.join(phase_dir, "sys_metrics.csv"))
+    busy, mem_used = [], []
+    for r in rows:
+        try:
+            busy.append(float(r["cpu_user_pct"]) + float(r["cpu_system_pct"]))
+            mem_used.append(float(r["mem_used_mb"]))
+        except (KeyError, ValueError):
+            continue
+    if not busy:
+        return {}
+    return {
+        "cpu_avg": statistics.mean(busy),
+        "cpu_peak": max(busy),
+        "mem_avg_gib": statistics.mean(mem_used) / 1024,
+        "mem_peak_gib": max(mem_used) / 1024,
+    }
+
+
+def read_amp(phase_dir: str, produced: int) -> dict:
+    rows = read_csv(os.path.join(phase_dir, "pgss_end.csv"))
+    recv = {"calls": 0, "rows": 0, "hit": 0, "read": 0, "ms": 0.0}
+    for r in rows:
+        if "receive_partitioned" not in r.get("query_head", ""):
+            continue
+        try:
+            recv["calls"] += int(r["calls"])
+            recv["rows"] += int(r["rows"])
+            recv["hit"] += int(r["shared_blks_hit"])
+            recv["read"] += int(r["shared_blks_read"])
+            recv["ms"] += float(r["total_exec_time_ms"])
+        except (KeyError, ValueError):
+            continue
+    blks = recv["hit"] + recv["read"]
+    recv["blks_total"] = blks
+    recv["blks_per_event"] = blks / produced if produced else 0.0
+    recv["read_per_event"] = recv["read"] / produced if produced else 0.0
+    return recv
+
+
+def stall_timeline(phase_dir: str, stall_slot: int) -> dict | None:
+    rows = read_csv(os.path.join(phase_dir, "slot_status.csv"))
+    if not rows:
+        return None
+    # Series for the stalled slot, ordered by ts.
+    series = []
+    for r in rows:
+        try:
+            if int(r["slot"]) != stall_slot:
+                continue
+            ts = parse_iso(r["ts"])
+            pe = float(r["pending_events"])
+            owner = r.get("lease_owner", "")
+            series.append((ts, pe, owner))
+        except (KeyError, ValueError):
+            continue
+    series = [s for s in series if s[0] is not None]
+    series.sort(key=lambda x: x[0])
+    if len(series) < 3:
+        return None
+
+    # The stall window = the contiguous span with no live lease owner (the
+    # SIGSTOPped worker stops renewing, so its lease expires).
+    stalled = [i for i, s in enumerate(series) if not s[2]]
+    if not stalled:
+        return {"detected": False, "peak_pending": max(s[1] for s in series)}
+    i0, i1 = stalled[0], stalled[-1]
+    t0, t1 = series[i0][0], series[i1][0]
+    stall_secs = (t1 - t0).total_seconds()
+    baseline = min(s[1] for s in series[:i0]) if i0 > 0 else series[0][1]
+    peak = max(s[1] for s in series[i0 : i1 + 1])
+    growth_rate = (peak - baseline) / stall_secs if stall_secs > 0 else 0.0
+
+    # Catch-up: after resume, time until pending falls back near baseline.
+    catchup_secs = None
+    thresh = baseline + max(1.0, 0.1 * (peak - baseline))
+    for ts, pe, _ in series[i1 + 1 :]:
+        if pe <= thresh:
+            catchup_secs = (ts - t1).total_seconds()
+            break
+
+    # Rotation floor: queue table count over the phase.
+    qrows = read_csv(os.path.join(phase_dir, "queue_rate.csv"))
+    ntables = []
+    for r in qrows:
+        try:
+            ntables.append(int(r["ntables"]))
+        except (KeyError, ValueError):
+            continue
+    return {
+        "detected": True,
+        "t0": t0,
+        "t1": t1,
+        "stall_secs": stall_secs,
+        "baseline": baseline,
+        "peak_pending": peak,
+        "growth_rate": growth_rate,
+        "catchup_secs": catchup_secs,
+        "ntables_min": min(ntables) if ntables else None,
+        "ntables_max": max(ntables) if ntables else None,
+    }
+
+
+# ---------------------------------------------------------------------------
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--out", default="/tmp/bench/pk")
+    args = ap.parse_args()
+
+    manifest = read_csv(os.path.join(args.out, "phases.csv"))
+    if not manifest:
+        print(f"# partition-keys bench\n\nNo phases.csv under {args.out}.")
+        return 0
+
+    L: list[str] = []
+    L.append("# Partition-keys read-amplification benchmark")
+    L.append("")
+    L.append("High-volume multi-tenant profile: keyed producer + N slot workers on the real")
+    L.append("pgque v0.8 lease API. Maps to SPEC R2 (read amplification ~N x) and R7")
+    L.append("(a stalled slot pins rotation for the whole queue).")
+    L.append("")
+
+    phase_metrics = []
+    for ph in manifest:
+        d = ph["dir"]
+        n = int(ph["n"])
+        dur = int(ph["dur_s"])
+        prod = produced_events(d)
+        tps = achieved_tps(d)
+        ctot, acks, per_worker = consumed(d)
+        cthru = ctot / dur if dur else 0.0
+        pend = pending_by_slot(read_csv(os.path.join(d, "slot_status.csv")))
+        all_pending = [v for vs in pend.values() for v in vs]
+        cm = cpu_mem(d)
+        ra = read_amp(d, prod)
+        bloat = bloat_stats(d)
+        # Stall phases (R7): recover the stalled slot from its dominant lag peak,
+        # then its lease-expiry window. Computed once here, reused by the bloat
+        # block and the stalled-slot section.
+        is_stall = ph["label"].startswith("stalled") or "stall" in os.path.basename(d.rstrip("/"))
+        stall_slot, stall = None, None
+        if is_stall:
+            peaks = {s: max(v) for s, v in pend.items() if v}
+            stall_slot = max(peaks, key=peaks.get) if peaks else 0
+            stall = stall_timeline(d, stall_slot)
+        phase_metrics.append(
+            dict(label=ph["label"], n=n, rate=int(ph["rate"]), dur=dur,
+                 produced=prod, tps=tps, consumed=ctot, cthru=cthru,
+                 per_worker=per_worker, pend=pend, all_pending=all_pending,
+                 cm=cm, ra=ra, dir=d, queue=ph["queue"], consumer=ph["consumer"],
+                 bloat=bloat, is_stall=is_stall, stall_slot=stall_slot, stall=stall))
+
+    # --- per-phase detail ---------------------------------------------------
+    for m in phase_metrics:
+        L.append(f"## Phase `{m['label']}`  (N={m['n']}, target {m['rate']} ev/s, {m['dur']//60} min)")
+        L.append("")
+        L.append(f"- Producer: **{fmt(m['tps'])} ev/s** achieved, {fmt(m['produced'])} events sent")
+        L.append(f"- Consumers: **{fmt(m['cthru'])} ev/s** end-to-end, {fmt(m['consumed'])} events acked")
+        if m["cm"]:
+            L.append(f"- CPU: {m['cm']['cpu_avg']:.0f}% avg / {m['cm']['cpu_peak']:.0f}% peak"
+                     f" | RAM used: {m['cm']['mem_avg_gib']:.1f} / {m['cm']['mem_peak_gib']:.1f} GiB")
+        if m["all_pending"]:
+            L.append(f"- Pending events across slots: p50 {fmt(pctile(m['all_pending'],0.5))},"
+                     f" p99 {fmt(pctile(m['all_pending'],0.99))}, max {fmt(max(m['all_pending']))}")
+        ra = m["ra"]
+        if ra["calls"]:
+            L.append(f"- Read-amp (receive_partitioned): {fmt(ra['blks_total'])} buffers"
+                     f" ({fmt(ra['hit'])} hit / {fmt(ra['read'])} read) over {fmt(ra['calls'])} calls"
+                     f" = **{ra['blks_per_event']:.2f} buffers/event**")
+        if m["per_worker"]:
+            pw = ", ".join(f"{w}:{fmt(c)}" for w, c in sorted(m["per_worker"].items(),
+                          key=lambda kv: int(kv[0].lstrip("abcdefghijklmnopqrstuvwxyz") or 0)))
+            L.append(f"- Per-slot consumed: {{ {pw} }}")
+        b = m["bloat"]
+        if b is None:
+            L.append("- Bloat & dead tuples: _no bloat.csv in this phase dir; section skipped_")
+        else:
+            L.append("")
+            L.append("**Bloat & dead tuples**")
+            L.append("")
+            top = sorted(b["meta"].items(), key=lambda kv: kv[1]["peak_dead"], reverse=True)[:5]
+            L.append("- Metadata tables by peak dead tuples (end size):")
+            for tbl, s in top:
+                L.append(f"    - `{tbl}`: peak {fmt(s['peak_dead'])} dead"
+                         f" (end {fmt(s['end_dead'])} dead, {fmt_bytes(s['end_bytes'])})")
+            ps = b["meta"].get("pgque.partition_slot")
+            if ps:
+                L.append(f"- `pgque.partition_slot` (v0.8 lease, 2 UPDATEs/batch):"
+                         f" peak **{fmt(ps['peak_dead'])}** dead, end {fmt(ps['end_dead'])} dead,"
+                         f" {fmt_bytes(ps['end_bytes'])} — HOT updates should keep this tiny")
+            for tbl in ("pgque.tick", "pgque.subscription"):
+                s = b["meta"].get(tbl)
+                if s:
+                    L.append(f"- `{tbl}` peak dead tuples: {fmt(s['peak_dead'])} (rotation-metadata churn)")
+            L.append(f"- Event tables: {b['ev_present_start']} present at phase start ->"
+                     f" {b['ev_present_end']} at end (rotation dropping old `event_N_M`)")
+            L.append(f"- Event-table bytes: start {fmt_bytes(b['ev_bytes_start'])},"
+                     f" end {fmt_bytes(b['ev_bytes_end'])}, peak {fmt_bytes(b['ev_bytes_peak'])}")
+            L.append(f"- Event-table peak dead tuples: {fmt(b['ev_peak_dead'])}"
+                     f" (append-only — expect ~0; nonzero means retries)")
+            L.append(f"- Total DB footprint: end {fmt_bytes(b['total_end'])}, peak {fmt_bytes(b['total_peak'])}")
+            if m["is_stall"]:
+                st = m["stall"]
+                if st and st.get("detected") and st.get("t0") and st.get("t1"):
+                    at = lambda tgt: b["ev_bytes_by_ts"][_nearest_ts(b["all_ts"], tgt)]
+                    L.append(f"- R7 pin/release — event-table bytes: stall start {fmt_bytes(at(st['t0']))}"
+                             f" -> stall end {fmt_bytes(at(st['t1']))} -> phase end {fmt_bytes(b['ev_bytes_end'])}"
+                             f" (grows while the stalled cursor pins rotation, drops after resume)")
+                else:
+                    L.append("- R7 pin/release: stall window not resolvable (needs slot_status.csv)")
+        L.append("")
+
+    # --- read-amp comparison (R2) ------------------------------------------
+    steady = [m for m in phase_metrics if m["label"].startswith("steady")]
+    if len(steady) >= 2:
+        steady.sort(key=lambda m: m["n"])
+        lo, hi = steady[0], steady[-1]
+        L.append("## Read amplification: N-scaling (SPEC R2)")
+        L.append("")
+        L.append("| N | buffers/event | read/event | ratio vs smallest N |")
+        L.append("|--:|--:|--:|--:|")
+        base = lo["ra"]["blks_per_event"] or 1.0
+        for m in steady:
+            r = m["ra"]["blks_per_event"] / base if base else 0.0
+            L.append(f"| {m['n']} | {m['ra']['blks_per_event']:.2f} |"
+                     f" {m['ra']['read_per_event']:.2f} | {r:.2f}x |")
+        L.append("")
+        L.append(f"Every slot scans the full stream and filters server-side, so buffers"
+                 f" touched per produced event scale ~linearly with N. Observed"
+                 f" {hi['n']}/{lo['n']} = {(hi['ra']['blks_per_event']/base):.2f}x"
+                 f" (ideal {hi['n']/lo['n']:.0f}x).")
+        L.append("")
+
+    # --- stalled-slot timeline (R7) ----------------------------------------
+    stall_phases = [m for m in phase_metrics if m["label"].startswith("stalled")]
+    if stall_phases:
+        m = stall_phases[0]
+        stall_slot = m["stall_slot"] or 0
+        st = m["stall"]
+        L.append("## Stalled-slot: rotation pinning (SPEC R7)")
+        L.append("")
+        L.append(f"Stalled slot (by peak lag): **slot {stall_slot}**.")
+        if st and st.get("detected"):
+            L.append(f"- Lease-expiry stall window: {st['stall_secs']:.0f}s (no live owner)")
+            L.append(f"- Pending events: baseline {fmt(st['baseline'])} -> peak {fmt(st['peak_pending'])}")
+            L.append(f"- Lag growth during stall: **{fmt(st['growth_rate'])} events/s**")
+            if st["catchup_secs"] is not None:
+                L.append(f"- Catch-up after resume: **{st['catchup_secs']:.0f}s** back to baseline")
+            else:
+                L.append("- Catch-up after resume: did not return to baseline within the phase")
+            if st["ntables_max"] is not None:
+                L.append(f"- Rotation floor: queue held {st['ntables_min']}..{st['ntables_max']}"
+                         f" event tables (the stalled slot pins the drop floor)")
+        else:
+            L.append("- Stall window not clearly detected in slot_status.csv"
+                     f" (peak pending {fmt(st['peak_pending']) if st else 'n/a'}).")
+        L.append("")
+
+    # --- headline table -----------------------------------------------------
+    L.append("## Headline")
+    L.append("")
+    L.append("| Phase | N | Producer ev/s | Consume ev/s | Pending p99 | Buffers/event | CPU peak | Peak dead tup (meta) | Event tbl GiB end |")
+    L.append("|---|--:|--:|--:|--:|--:|--:|--:|--:|")
+    for m in phase_metrics:
+        p99 = fmt(pctile(m["all_pending"], 0.99)) if m["all_pending"] else "-"
+        bpe = f"{m['ra']['blks_per_event']:.2f}" if m["ra"]["calls"] else "-"
+        cpu = f"{m['cm']['cpu_peak']:.0f}%" if m["cm"] else "-"
+        b = m["bloat"]
+        meta_dead = fmt(b["meta_peak_dead"]) if b else "-"
+        ev_gib = f"{gib(b['ev_bytes_end']):.2f}" if b else "-"
+        L.append(f"| {m['label']} | {m['n']} | {fmt(m['tps'])} |"
+                 f" {fmt(m['cthru'])} | {p99} | {bpe} | {cpu} | {meta_dead} | {ev_gib} |")
+    L.append("")
+
+    print("\n".join(L))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/blueprints/partition-keys/repro/.gitignore b/blueprints/partition-keys/repro/.gitignore
new file mode 100644
index 00000000..c2658d7d
--- /dev/null
+++ b/blueprints/partition-keys/repro/.gitignore
@@ -0,0 +1 @@
+node_modules/
diff --git a/blueprints/partition-keys/repro/README.md b/blueprints/partition-keys/repro/README.md
new file mode 100644
index 00000000..113d0f52
--- /dev/null
+++ b/blueprints/partition-keys/repro/README.md
@@ -0,0 +1,150 @@
+# Partition-keys reproduction
+
+A runnable spike for the partition-keys design
+(`blueprints/partition-keys/SPEC.md`), aligned to the **two cases** from the
+original thread. It installs pgque on a throwaway Postgres, drives each case
+with concurrent producers/workers, **measures**, and **checks the guarantees
+empirically** — so we test the design instead of arguing about it.
+
+TypeScript on **bun** + `pg` (node-postgres), matching `clients/typescript/`.
+Everything consumer/producer-side is a thin recipe in `schema.sql`; the engine
+is unchanged.
+
+## The two cases (and what each one actually needs)
+
+These are **two different features**, not two tiers of one. Case 2 is the
+partition-keys feature; Case 1 is a *plain-queue* recipe that doesn't use
+partition slots at all.
+
+**Case 1 — migrations: producer idempotency + consumer mutual exclusion**
+(`--tier a`, `--tier hazard`). On a **plain queue** (no slots, no fixed N).
+Fabrizio's ask has **two distinct guarantees**, *complementary layers*:
+
+| Layer | Guarantee | Mechanism | Prevents |
+|-------|-----------|-----------|----------|
+| **L1 producer** | idempotency | TTL dedup window (`demo.send_idem`) | duplicate **insert** (log bloat, no-op enqueues) |
+| **L2 consumer** | one job at a time per key | cooperative consumers + per-key `pg_try_advisory_xact_lock` | duplicate **work** |
+
+L1 is the producer-side TTL dedup (the SQS/NATS model — its own send-layer
+feature, spec'd in `blueprints/idempotency/`). L2 is mutual exclusion on the
+consume side. **They are not substitutes** — L1 keeps the log small; L2
+guarantees correctness even if a duplicate slips in. Neither needs the
+partition-keys (slot) feature: the "key" here is a *lock key*, not a partition.
+
+> **Key-scope footgun (`--tier hazard`).** A TTL dedup keyed on the *entity*
+> (`migrate:tenant`) silently drops a *needed* job when the desired effect
+> changes inside the window — ship migration v1, then v2 within the TTL, and v2
+> is suppressed as a "duplicate". The idempotency key must represent the desired
+> **effect**, not just the entity: `migrate:tenant:vN`. The guardrail test
+> reproduces both.
+
+**Case 2 — lifecycle events: ordered per key** (`--tier b`). *This* is the
+partition-keys feature. Ordered per tenant, parallel across tenants: N
+hash-routed slot subscriptions filtering via `get_batch_cursor` `extra_where`.
+Checks G1 (per-key FIFO + single-slot affinity) + exactly-once, and measures
+read amplification (~N) — the honest cost, plus rotation-pin risk (the slowest
+slot lowers the rotation floor; see SPEC R7).
+
+## Run it on a fresh Linux VM
+
+```bash
+# Debian/Ubuntu VM, from inside the pgque repo:
+sudo bash blueprints/partition-keys/repro/run.sh
+```
+
+`run.sh` installs Postgres + bun, creates `pgque_repro` and a peer-auth role,
+installs `sql/pgque.sql` + `schema.sql`, `bun install`s, then runs Case 1
+(both with and without producer dedup) and Case 2, each ending in `PASS`/`FAIL`
+invariant lines.
+
+### Run pieces directly / change scale
+
+```bash
+cd blueprints/partition-keys/repro
+export PGHOST=/var/run/postgresql PGDATABASE=pgque_repro PGUSER=$(id -un)
+
+# Case 1 with producer dedup ON, 8 producers racing 5000 tenants:
+bun driver.ts --tier a --tenants 5000 --producers 8 --dups 4 --dedup-ttl 60 --workers 16
+
+# Case 2, 16 slots:
+bun driver.ts --tier b --tenants 1000 --events-per-tenant 30 --slots 16
+
+# Dedup key-scope guardrail (entity-only vs effect-scoped key):
+bun driver.ts --tier hazard --tenants 500 --dedup-ttl 300
+```
+
+Knobs (env for `run.sh`, flags for `driver.ts`):
+
+- Case 1: `A_TENANTS`/`--tenants`, `A_PRODUCERS`/`--producers` (concurrent
+  producers racing the same tenants), `A_DUPS`/`--dups` (attempts each),
+  `A_DEDUP_TTL`/`--dedup-ttl` (seconds; `0` = producer dedup off),
+  `A_WORKERS`/`--workers`, `A_WORK_MS`/`--work-ms`.
+- Case 2: `B_TENANTS`/`--tenants`, `B_EPT`/`--events-per-tenant`,
+  `B_SLOTS`/`--slots` (= N = worker count, static assignment).
+
+## What the reports show
+
+**Case 1**
+- `[L1 producer] attempts` vs `INSERTED` — with dedup ON, thousands of
+  concurrent "migrate tenant T" attempts collapse to one insert per tenant
+  (the literal idempotency ask). With dedup OFF, all are inserted.
+- `[L2 consumer] jobs RUN` vs `ack-dropped` — exactly one run per tenant either
+  way; residual duplicates are dropped at consume.
+- Invariants: producer dedup → inserted == distinct tenants; no two workers ran
+  the same key with overlapping windows; every tenant migrated exactly once.
+
+**Case 2**
+- `read amplification` = measured `scanned/delivered`, ~`N` (every slot scans
+  the full window) — the cost of strict ordering, quantified.
+- Invariants: each key on exactly one slot (affinity), non-decreasing `ev_id`
+  per key (FIFO), nothing lost or duplicated.
+
+## Benchmark: bloat-under-backlog + throughput (`bench.ts`)
+
+The question that started this: *when workers fall behind, does the store bloat?*
+`bench.ts` runs PgQue (append + rotation) against a **pg-boss-style mutable job
+table** (`insert → update(active) → delete(complete)`) in the same database, and
+measures footprint, **dead tuples**, vacuum activity, and throughput.
+
+```bash
+cd blueprints/partition-keys/repro
+export PGHOST=/var/run/postgresql PGDATABASE=pgque_repro PGUSER=$(id -un)
+bun bench.ts --mode throughput --dur 8
+bun bench.ts --mode bloat --build-sec 16 --produce-rate 20000 --consume-rate 3000
+```
+
+**Measured (PG16, this box).** The differentiator is **churn**, not transient
+size — both engines store a backlog, but only the mutable one rots:
+
+| | PgQue (append+rotation) | jobq (mutable) |
+|---|---|---|
+| consume throughput | **~208k ev/s** | ~40k ev/s |
+| dead tuples while building a backlog | **0** | climbs (≈2× processed) |
+| vacuums needed | **0** | autovacuum must chase the churn |
+| after draining a ~150k backlog | 0 dead tuples (clean append) | **~300k dead tuples, table grew to ~34 MiB** |
+
+The punchline: a mutable queue creates ~2 dead tuples per processed job
+(`update` + `delete`), so draining a backlog leaves the table **larger and full
+of dead space** that vacuum must reclaim — exactly the pg-boss bloat. PgQue's
+events are append-only (**zero** dead tuples, zero vacuum on the event tables);
+space is reclaimed by **rotation = `TRUNCATE` of whole tables**, not row deletes.
+
+**Honest gaps (see caveats):** the produce side is round-trip-bound, not a tuned
+ingest benchmark; and the end-to-end "PgQue reclaims to ~0 via rotation" wasn't
+captured as a clean number here — PgQ rotation is a multi-period state machine
+(truncates the next ring table each `rotation_period` once consumers are past),
+which this short harness + a reap-happy sandbox didn't drive through a full
+cycle. The *mechanism* (TRUNCATE vs DELETE) and the *zero-dead-tuple* property
+are what's measured; the reclaim-curve is a follow-up.
+
+## Caveats (it's a spike)
+
+- One-shot: produce-then-drain, not steady-state. Throughput is indicative, not
+  a tuned benchmark; producer throughput is per-row round trips, not bulk.
+- Case 1 processes a whole cooperative batch in one transaction (the per-key
+  advisory lock is `xact`-scoped), so the visible processing window equals the
+  batch — fine for checking exclusion, not for latency.
+- Case 2 uses static `worker k → slot k`; the dynamic claim variants
+  (advisory / `SKIP LOCKED`) from SPEC §15 are not exercised here.
+- The L1 TTL table is GC'd by expiry; a real install would tie the window to
+  rotation (see the idempotency design note).
diff --git a/blueprints/partition-keys/repro/bench.ts b/blueprints/partition-keys/repro/bench.ts
new file mode 100644
index 00000000..91b70b81
--- /dev/null
+++ b/blueprints/partition-keys/repro/bench.ts
@@ -0,0 +1,298 @@
+/**
+ * PgQue vs pg-boss-style mutable queue — bloat-under-backlog + throughput.
+ *
+ * The question that actually triggered Fabrizio: when workers fall behind, does
+ * the store bloat? PgQue is append-only + rotation (TRUNCATE whole tables);
+ * a mutable job table is insert->update->delete (dead tuples, vacuum, bloat).
+ *
+ *   bun bench.ts --mode bloat       --build-sec 30 --consume-rate 4000
+ *   bun bench.ts --mode throughput  --dur 15
+ *
+ * Connection: libpq env (PGHOST/PGDATABASE/PGUSER). Default db pgque_repro.
+ */
+import pg from "pg";
+const { Client } = pg;
+
+function newClient() {
+  return new Client({
+    host: process.env.PGHOST,
+    database: process.env.PGDATABASE || "pgque_repro",
+    user: process.env.PGUSER || process.env.USER || "postgres",
+    connectionTimeoutMillis: 5000,
+    // fail fast instead of hanging if the server is reaped mid-run
+    options: "-c statement_timeout=30000",
+  });
+}
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+const now = () => performance.now();
+const MiB = (b: number) => (b / 1024 / 1024).toFixed(1) + " MiB";
+const k = (n: number) => Math.round(n).toLocaleString("en-US");
+
+interface Sample { t: number; bytes: number; live: number; dead: number; vacs: number; }
+
+interface Engine {
+  name: string;
+  reset(c: pg.Client): Promise<void>;
+  produce(c: pg.Client, n: number): Promise<void>; // append n events/jobs (fast)
+  postProduce?(c: pg.Client): Promise<void>;        // pgque: tick to make consumable
+  consume(c: pg.Client, max: number): Promise<number>;
+  maint?(c: pg.Client): Promise<void>;              // pgque: rotate/reclaim
+  sample(c: pg.Client): Promise<Sample>;
+}
+
+// --- PgQue: append-only events + rotation -----------------------------------
+function pgqueEngine(): Engine {
+  const queue = "bench_pgque";
+  const consumer = "c";
+  let qid = 0;
+  return {
+    name: "pgque (append + rotation)",
+    async reset(c) {
+      try { await c.query("select pgque.drop_queue($1,true)", [queue]); } catch {}
+      await c.query("select pgque.create_queue($1)", [queue]);
+      await c.query("select pgque.register_consumer($1,$2)", [queue, consumer]);
+      // aggressive rotation so reclaim is visible within the run
+      await c.query("select pgque.set_queue_config($1,'rotation_period','1 second')", [queue]);
+      qid = (await c.query("select queue_id from pgque.queue where queue_name=$1", [queue])).rows[0].queue_id;
+    },
+    async produce(c, n) {
+      // set-based bulk insert (the fair fast path, mirrors jobq's bulk INSERT)
+      await c.query(
+        "select pgque.insert_event_bulk($1,'e', array(select '{\"x\":1}' from generate_series(1,$2)))",
+        [queue, n],
+      );
+    },
+    async postProduce(c) {
+      await c.query("select pgque.force_tick($1)", [queue]);
+      await c.query("select pgque.ticker($1)", [queue]);
+    },
+    async consume(c, max) {
+      const r = await c.query("select batch_id from pgque.receive($1,$2,$3)", [queue, consumer, max]);
+      if (r.rows.length === 0) return 0;
+      await c.query("select pgque.ack($1)", [r.rows[0].batch_id]);
+      return r.rows.length;
+    },
+    async maint(c) {
+      await c.query("select pgque.ticker($1)", [queue]);
+      await c.query("select pgque.maint()");
+    },
+    async sample(c) {
+      const r = await c.query(
+        `select coalesce(sum(pg_total_relation_size(cl.oid)),0)::bigint bytes,
+                coalesce(sum(st.n_live_tup),0)::bigint live,
+                coalesce(sum(st.n_dead_tup),0)::bigint dead,
+                coalesce(sum(st.vacuum_count+st.autovacuum_count),0)::bigint vacs
+           from pg_class cl
+           join pg_namespace ns on ns.oid=cl.relnamespace
+           left join pg_stat_user_tables st on st.relid=cl.oid
+          where ns.nspname='pgque' and cl.relkind='r' and cl.relname like 'event_'||$1||'_%'`,
+        [qid],
+      );
+      const x = r.rows[0];
+      return { t: 0, bytes: +x.bytes, live: +x.live, dead: +x.dead, vacs: +x.vacs };
+    },
+  };
+}
+
+// --- pg-boss-style: mutable rows, insert->update->delete --------------------
+function jobqEngine(): Engine {
+  return {
+    name: "jobq (mutable: insert/update/delete)",
+    async reset(c) {
+      await c.query("truncate demo.jobq");
+      // realistic: leave autovacuum ON (so we measure how hard it must work)
+      await c.query("select pg_stat_reset_single_table_counters('demo.jobq'::regclass)");
+    },
+    async produce(c, n) {
+      await c.query(
+        "insert into demo.jobq(key,state,payload) select null,'created','{\"x\":1}' from generate_series(1,$1)",
+        [n],
+      );
+    },
+    async consume(c, max) {
+      // claim (update -> active), then complete (delete): pg-boss-shaped churn,
+      // two statements so we don't update+delete the same row in one (UB).
+      const claimed = await c.query(
+        `with cl as (
+           select id from demo.jobq where state='created' order by id
+           limit $1 for update skip locked)
+         update demo.jobq j set state='active' from cl
+          where j.id = cl.id returning j.id`,
+        [max],
+      );
+      const ids = claimed.rows.map((r) => r.id);
+      if (ids.length) await c.query("delete from demo.jobq where id = any($1::bigint[])", [ids]);
+      return ids.length;
+    },
+    async sample(c) {
+      const r = await c.query(
+        `select pg_total_relation_size('demo.jobq')::bigint bytes,
+                n_live_tup live, n_dead_tup dead, (vacuum_count+autovacuum_count) vacs
+           from pg_stat_user_tables where relid='demo.jobq'::regclass`,
+      );
+      const x = r.rows[0] || { bytes: 0, live: 0, dead: 0, vacs: 0 };
+      return { t: 0, bytes: +x.bytes, live: +x.live, dead: +x.dead, vacs: +x.vacs };
+    },
+  };
+}
+
+// ---------------------------------------------------------------------------
+async function runBloat(eng: Engine, a: Args) {
+  await usingClient((c) => eng.reset(c));
+  const series: Sample[] = [];
+  const start = now();
+  let producing = true;
+  let produced = 0;
+  let consumed = 0;
+  console.log(`\n  === ${eng.name} ===`);
+  console.log(`  build ${a.buildSec}s: produce ~${k(a.produceRate)}/s, consume ~${k(a.consumeRate)}/s  (backlog grows)`);
+  console.log(`  ${"phase".padEnd(6)} ${"t(s)".padStart(6)} ${"size".padStart(11)} ${"live".padStart(10)} ${"dead".padStart(10)} ${"vac".padStart(5)}`);
+
+  // PRODUCER: throttled to --produce-rate events/sec (its own connection)
+  const producer = usingClient(async (c) => {
+    while (producing) {
+      const t = now();
+      let allow = Math.ceil(a.produceRate * 0.1); // per 100ms window
+      while (allow > 0) {
+        const n = Math.min(a.batch, allow);
+        await eng.produce(c, n);
+        produced += n;
+        allow -= n;
+      }
+      if (eng.postProduce) await eng.postProduce(c);
+      const dt = now() - t;
+      if (dt < 100) await sleep(100 - dt);
+    }
+  });
+
+  // CONSUMER: throttled to --consume-rate events/sec (its own connection)
+  const consumer = usingClient(async (c) => {
+    while (producing) {
+      const t = now();
+      let allow = Math.ceil(a.consumeRate * 0.1); // per 100ms window
+      while (allow > 0) {
+        const got = await eng.consume(c, Math.min(a.batch, allow));
+        if (got === 0) break;
+        consumed += got;
+        allow -= got;
+      }
+      const dt = now() - t;
+      if (dt < 100) await sleep(100 - dt);
+    }
+  });
+
+  // SAMPLER (streams each line live)
+  const line = (phase: string, s: Sample) =>
+    console.log(`  ${phase.padEnd(6)} ${s.t.toFixed(1).padStart(6)} ${MiB(s.bytes).padStart(11)} ${k(s.live).padStart(10)} ${k(s.dead).padStart(10)} ${String(s.vacs).padStart(5)}`);
+  const sampler = usingClient(async (c) => {
+    while (producing) {
+      const s = await eng.sample(c);
+      s.t = (now() - start) / 1000;
+      series.push(s);
+      line("build", s);
+      await sleep(a.sampleMs);
+    }
+  });
+
+  await sleep(a.buildSec * 1000);
+  producing = false;
+  await Promise.all([producer, consumer, sampler]);
+
+  // DRAIN: stop producing, consume flat out + reclaim, then final sample
+  const drainStart = now();
+  await usingClient(async (c) => {
+    if (eng.postProduce) await eng.postProduce(c);
+    while ((now() - drainStart) / 1000 < a.drainTimeout) {
+      const got = await eng.consume(c, a.batch);
+      consumed += got;
+      if (got === 0) {
+        if (eng.maint) { await eng.maint(c); await sleep(1100); } // let rotation_period elapse
+        const more = await eng.consume(c, a.batch);
+        consumed += more;
+        if (more === 0) break;
+      }
+    }
+    if (eng.maint) for (let i = 0; i < 4; i++) { await eng.maint(c); await sleep(1100); }
+  });
+  const afterDrain = await usingClient((c) => eng.sample(c));
+  const drainS = (now() - drainStart) / 1000;
+
+  afterDrain.t = (now() - start) / 1000;
+  line("DRAIN", afterDrain);
+  const peak = series.reduce((m, s) => (s.bytes > m.bytes ? s : m), series[0] || afterDrain);
+  const peakDead = series.reduce((m, s) => Math.max(m, s.dead), 0);
+  console.log(`  produced ${k(produced)}, consumed ${k(consumed)}; peak size ${MiB(peak.bytes)}, peak dead ${k(peakDead)}`);
+  console.log(`  >> AFTER DRAIN (${drainS.toFixed(1)}s): size ${MiB(afterDrain.bytes)} · dead ${k(afterDrain.dead)} · vacuums ${afterDrain.vacs}`);
+  return { peak: peak.bytes, peakDead, afterDrain };
+}
+
+async function runThroughput(eng: Engine, a: Args) {
+  await usingClient((c) => eng.reset(c));
+  // PRODUCE flat out for --dur seconds
+  let produced = 0;
+  const pStart = now();
+  await usingClient(async (c) => {
+    while ((now() - pStart) / 1000 < a.dur) {
+      await eng.produce(c, a.batch);
+      produced += a.batch;
+      if (eng.postProduce) await eng.postProduce(c);
+    }
+  });
+  const produceS = (now() - pStart) / 1000;
+  // CONSUME flat out until drained
+  let consumed = 0;
+  const cStart = now();
+  await usingClient(async (c) => {
+    if (eng.postProduce) await eng.postProduce(c);
+    let empty = 0;
+    while (empty < 3 && (now() - cStart) / 1000 < a.drainTimeout) {
+      const got = await eng.consume(c, a.batch);
+      consumed += got;
+      empty = got === 0 ? empty + 1 : 0;
+    }
+  });
+  const consumeS = (now() - cStart) / 1000;
+  console.log(`\n  === ${eng.name} ===`);
+  console.log(`  produce: ${k(produced)} in ${produceS.toFixed(1)}s = ${k(produced / produceS)} ev/s`);
+  console.log(`  consume: ${k(consumed)} in ${consumeS.toFixed(1)}s = ${k(consumed / consumeS)} ev/s`);
+}
+
+// ---------------------------------------------------------------------------
+async function usingClient<T>(fn: (c: pg.Client) => Promise<T>): Promise<T> {
+  const c = newClient();
+  await c.connect();
+  try { return await fn(c); } finally { await c.end(); }
+}
+
+interface Args {
+  mode: string; engine: string; batch: number; buildSec: number;
+  produceRate: number; consumeRate: number; sampleMs: number; drainTimeout: number; dur: number;
+}
+function parseArgs(): Args {
+  const a: any = {
+    mode: "bloat", engine: "both", batch: 2000, buildSec: 20,
+    produceRate: 20000, consumeRate: 4000, sampleMs: 3000, drainTimeout: 120, dur: 15,
+  };
+  const map: Record<string, keyof Args> = {
+    "--mode": "mode", "--engine": "engine", "--batch": "batch", "--build-sec": "buildSec",
+    "--produce-rate": "produceRate", "--consume-rate": "consumeRate", "--sample-ms": "sampleMs",
+    "--drain-timeout": "drainTimeout", "--dur": "dur",
+  };
+  const argv = process.argv.slice(2);
+  for (let i = 0; i < argv.length; i += 2) {
+    const key = map[argv[i]];
+    if (!key) throw new Error(`unknown arg ${argv[i]}`);
+    (a as any)[key] = ["mode", "engine"].includes(key) ? argv[i + 1] : Number(argv[i + 1]);
+  }
+  return a as Args;
+}
+
+const args = parseArgs();
+const engines = args.engine === "pgque" ? [pgqueEngine()]
+  : args.engine === "jobq" ? [jobqEngine()]
+  : [pgqueEngine(), jobqEngine()];
+console.log(`# ${args.mode} — PgQue vs pg-boss-style mutable queue`);
+for (const eng of engines) {
+  if (args.mode === "bloat") await runBloat(eng, args);
+  else await runThroughput(eng, args);
+}
diff --git a/blueprints/partition-keys/repro/bun.lock b/blueprints/partition-keys/repro/bun.lock
new file mode 100644
index 00000000..8d20b6a7
--- /dev/null
+++ b/blueprints/partition-keys/repro/bun.lock
@@ -0,0 +1,50 @@
+{
+  "lockfileVersion": 1,
+  "configVersion": 1,
+  "workspaces": {
+    "": {
+      "name": "pgque-partition-keys-repro",
+      "dependencies": {
+        "pg": "^8.16.0",
+      },
+      "devDependencies": {
+        "@types/pg": "^8.15.2",
+      },
+    },
+  },
+  "packages": {
+    "@types/node": ["@types/node@26.0.1", "", { "dependencies": { "undici-types": "~8.3.0" } }, "sha512-fc3KiUoBt6kie0N9bIW3E47vZsuaMf0PM2AaUpLCLT0s/LvX1nxAim6Fc049cNxODPpGm6qRAuUOB86SkRuPQw=="],
+
+    "@types/pg": ["@types/pg@8.20.0", "", { "dependencies": { "@types/node": "*", "pg-protocol": "*", "pg-types": "^2.2.0" } }, "sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow=="],
+
+    "pg": ["pg@8.22.0", "", { "dependencies": { "pg-connection-string": "^2.14.0", "pg-pool": "^3.14.0", "pg-protocol": "^1.15.0", "pg-types": "2.2.0", "pgpass": "1.0.5" }, "optionalDependencies": { "pg-cloudflare": "^1.4.0" }, "peerDependencies": { "pg-native": ">=3.0.1" }, "optionalPeers": ["pg-native"] }, "sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA=="],
+
+    "pg-cloudflare": ["pg-cloudflare@1.4.0", "", {}, "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A=="],
+
+    "pg-connection-string": ["pg-connection-string@2.14.0", "", {}, "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg=="],
+
+    "pg-int8": ["pg-int8@1.0.1", "", {}, "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw=="],
+
+    "pg-pool": ["pg-pool@3.14.0", "", { "peerDependencies": { "pg": ">=8.0" } }, "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw=="],
+
+    "pg-protocol": ["pg-protocol@1.15.0", "", {}, "sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ=="],
+
+    "pg-types": ["pg-types@2.2.0", "", { "dependencies": { "pg-int8": "1.0.1", "postgres-array": "~2.0.0", "postgres-bytea": "~1.0.0", "postgres-date": "~1.0.4", "postgres-interval": "^1.1.0" } }, "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA=="],
+
+    "pgpass": ["pgpass@1.0.5", "", { "dependencies": { "split2": "^4.1.0" } }, "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug=="],
+
+    "postgres-array": ["postgres-array@2.0.0", "", {}, "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA=="],
+
+    "postgres-bytea": ["postgres-bytea@1.0.1", "", {}, "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ=="],
+
+    "postgres-date": ["postgres-date@1.0.7", "", {}, "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q=="],
+
+    "postgres-interval": ["postgres-interval@1.2.0", "", { "dependencies": { "xtend": "^4.0.0" } }, "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ=="],
+
+    "split2": ["split2@4.2.0", "", {}, "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg=="],
+
+    "undici-types": ["undici-types@8.3.0", "", {}, "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ=="],
+
+    "xtend": ["xtend@4.0.2", "", {}, "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ=="],
+  }
+}
diff --git a/blueprints/partition-keys/repro/driver.ts b/blueprints/partition-keys/repro/driver.ts
new file mode 100644
index 00000000..0266b11b
--- /dev/null
+++ b/blueprints/partition-keys/repro/driver.ts
@@ -0,0 +1,420 @@
+/**
+ * Partition-keys reproduction driver (TypeScript / bun + node-postgres).
+ *
+ * Produces a keyed workload, drains it with N concurrent workers, measures
+ * throughput, and checks the design's guarantees empirically against a real
+ * pgque install.
+ *
+ *   Tier A (mutual exclusion):  cooperative consumers + per-key advisory lock.
+ *   Tier B (ordered per key):   N hash-routed slot subscriptions.
+ *
+ * Run:
+ *   PGDATABASE=pgque_repro bun driver.ts --tier a --tenants 2000 --dups 4 --workers 8 --work-ms 3
+ *   PGDATABASE=pgque_repro bun driver.ts --tier b --tenants 500 --events-per-tenant 20 --slots 8
+ *
+ * Connection: standard libpq env vars (PGDATABASE/PGHOST/PGUSER/...). Defaults
+ * to database "pgque_repro".
+ */
+import pg from "pg";
+const { Client } = pg;
+
+const SEED = 1234;
+
+// ---- tiny seeded RNG (mulberry32) for a reproducible shuffle ---------------
+function rng(seed: number): () => number {
+  let a = seed >>> 0;
+  return () => {
+    a |= 0;
+    a = (a + 0x6d2b79f5) | 0;
+    let t = Math.imul(a ^ (a >>> 15), 1 | a);
+    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
+    return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
+  };
+}
+function shuffle<T>(arr: T[], rand: () => number): T[] {
+  for (let i = arr.length - 1; i > 0; i--) {
+    const j = Math.floor(rand() * (i + 1));
+    [arr[i], arr[j]] = [arr[j], arr[i]];
+  }
+  return arr;
+}
+
+function newClient() {
+  // Honor libpq env. bun does not populate process.env.USER and its
+  // os.userInfo() returns "unknown", so for peer auth the role must come from
+  // PGUSER/USER (run.sh sets PGUSER); fall back to "postgres".
+  return new Client({
+    host: process.env.PGHOST,
+    database: process.env.PGDATABASE || "pgque_repro",
+    user: process.env.PGUSER || process.env.USER || "postgres",
+  });
+}
+async function withClient<T>(fn: (c: pg.Client) => Promise<T>): Promise<T> {
+  const c = newClient();
+  await c.connect();
+  try {
+    return await fn(c);
+  } finally {
+    await c.end();
+  }
+}
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+async function resetQueue(c: pg.Client, queue: string) {
+  // A prior run may have left cooperative members registered; drop_queue
+  // refuses a queue that still has coop_members (pgque.sql), so tear them down
+  // explicitly first (i_batch_handling=1 forces any leftover open batch).
+  const members = await c
+    .query(
+      `select co.co_name
+         from pgque.subscription s
+         join pgque.queue q on q.queue_id = s.sub_queue
+         join pgque.consumer co on co.co_id = s.sub_consumer
+        where q.queue_name = $1 and s.sub_role = 'coop_member'`,
+      [queue],
+    )
+    .catch(() => ({ rows: [] as { co_name: string }[] }));
+  for (const m of members.rows) {
+    const dot = m.co_name.indexOf(".");
+    const main = m.co_name.slice(0, dot);
+    const sub = m.co_name.slice(dot + 1);
+    await c
+      .query("select pgque.unregister_subconsumer($1,$2,$3,1)", [queue, main, sub])
+      .catch(() => {});
+  }
+  try {
+    await c.query("select pgque.drop_queue($1, true)", [queue]);
+  } catch (e: any) {
+    if (!String(e.message).includes("No such event queue")) throw e;
+  }
+  await c.query("select pgque.create_queue($1)", [queue]);
+}
+async function tick(c: pg.Client, queue: string) {
+  await c.query("select pgque.force_tick($1)", [queue]);
+  await c.query("select pgque.ticker($1)", [queue]);
+}
+const fmt = (n: number) => Math.round(n).toLocaleString("en-US");
+function check(label: string, passed: boolean, detail = ""): boolean {
+  const mark = passed ? "PASS" : "FAIL";
+  console.log(`    [${mark}] ${label}${passed ? "" : `  <-- ${detail}`}`);
+  return passed;
+}
+
+interface Args {
+  tier: string;
+  tenants: number;
+  workers: number;
+  dups: number;
+  workMs: number;
+  producers: number;
+  dedupTtl: number; // seconds; 0 = producer dedup OFF
+  slots: number;
+  eventsPerTenant: number;
+  chunk: number;
+  maxBatch: number;
+}
+
+// ---------------------------------------------------------------------------
+// Tier A — mutual exclusion (migration-style workload)
+// ---------------------------------------------------------------------------
+async function runTierA(a: Args): Promise<boolean> {
+  const queue = "migrations";
+  const main = "mig";
+  const workers = Array.from({ length: a.workers }, (_, i) => `w${i}`);
+  const dedupOn = a.dedupTtl > 0;
+
+  await withClient(async (c) => {
+    await c.query("select demo.reset()");
+    await c.query("truncate demo.idem");
+    await resetQueue(c, queue);
+    for (const w of workers)
+      await c.query("select pgque.register_subconsumer($1,$2,$3)", [queue, main, w]);
+  });
+
+  // --- producer phase: P concurrent producers, each attempting every tenant
+  // `dups` times (simulating many app instances racing on the same tenants).
+  // A background ticker runs continuously (as pg_cron would), creating many
+  // tick windows so the cooperative consumer can spread the load.
+  const attempts = a.producers * a.tenants * a.dups;
+  let inserted = 0; // counted producer-side from send_idem / produce
+  let producing = true;
+  const tickerTask = withClient(async (c) => {
+    while (producing) {
+      await tick(c, queue);
+      await sleep(25);
+    }
+    await tick(c, queue);
+  });
+
+  const t0 = performance.now();
+  await Promise.all(
+    Array.from({ length: a.producers }, (_, p) =>
+      withClient(async (c) => {
+        const rand = rng(SEED + p);
+        let local = 0;
+        for (let d = 0; d < a.dups; d++) {
+          const order = shuffle(Array.from({ length: a.tenants }, (_, i) => i), rand);
+          for (const t of order) {
+            const key = `tenant-${t}`;
+            if (dedupOn) {
+              const r = await c.query(
+                "select deduped from demo.send_idem($1,$2,$3,$4,$5, make_interval(secs => $6))",
+                [queue, "migrate", '{"op":"ensure_latest"}', key, `migrate:${key}`, a.dedupTtl],
+              );
+              if (!r.rows[0].deduped) local++;
+            } else {
+              await c.query("select demo.produce($1,$2,$3,$4)", [
+                queue, "migrate", '{"op":"ensure_latest"}', key,
+              ]);
+              local++;
+            }
+          }
+        }
+        inserted += local;
+      }),
+    ),
+  );
+  producing = false;
+  await tickerTask;
+  const produceS = (performance.now() - t0) / 1000;
+  const total = inserted; // events actually appended to the log
+
+  // drain with N concurrent workers (each its own connection)
+  const stats = new Map(workers.map((w) => [w, { got: 0, ran: 0, dropped: 0 }]));
+  const t1 = performance.now();
+  await Promise.all(
+    workers.map((w) =>
+      withClient(async (c) => {
+        let empty = 0;
+        const s = stats.get(w)!;
+        while (empty < 2) {
+          const r = await c.query(
+            "select got, ran, dropped from demo.tier_a_consume($1,$2,$3,$4,$5)",
+            [queue, main, w, a.maxBatch, a.workMs],
+          );
+          const { got, ran, dropped } = r.rows[0];
+          s.got += +got;
+          s.ran += +ran;
+          s.dropped += +dropped;
+          empty = +got === 0 ? empty + 1 : 0;
+          if (+got === 0) await sleep(10);
+        }
+      }),
+    ),
+  );
+  const drainS = (performance.now() - t1) / 1000;
+
+  return withClient(async (c) => {
+    const got = [...stats.values()].reduce((x, s) => x + s.got, 0);
+    const ran = [...stats.values()].reduce((x, s) => x + s.ran, 0);
+    const dropped = [...stats.values()].reduce((x, s) => x + s.dropped, 0);
+    const migrated = +(await c.query("select count(*) from demo.tenant_migrated")).rows[0].count;
+    const doubleRun = +(
+      await c.query("select count(*) from demo.tenant_migrated where runs > 1")
+    ).rows[0].count;
+    const overlaps = +(
+      await c.query(`
+        select count(*) from demo.mutex_log a join demo.mutex_log b
+          on a.part_key = b.part_key and a.worker <> b.worker and a.id < b.id
+         and a.started_at < b.ended_at and b.started_at < a.ended_at`)
+    ).rows[0].count;
+    const unfinished = +(
+      await c.query("select count(*) from demo.mutex_log where ended_at is null")
+    ).rows[0].count;
+
+    const perWorker = workers.map((w) => `${w}:${stats.get(w)!.ran}`).join(", ");
+    console.log(`  setup                   : ${a.tenants} tenants, ${a.producers} concurrent producers, ${a.dups} attempts each`);
+    console.log(`  producer dedup          : ${dedupOn ? `ON  (TTL window ${a.dedupTtl}s)` : "OFF"}`);
+    console.log(`  [L1 producer] attempts  : ${attempts}`);
+    console.log(`  [L1 producer] INSERTED  : ${total}   (deduped at the door: ${attempts - total})`);
+    console.log(`  produce time            : ${produceS.toFixed(2)}s`);
+    console.log(`  drain time              : ${drainS.toFixed(2)}s   (${a.workers} workers, work-ms=${a.workMs})`);
+    console.log(`  consume throughput      : ${fmt(got / drainS)} events/s`);
+    console.log(`  [L2 consumer] jobs RUN  : ${ran}`);
+    console.log(`  [L2 consumer] ack-dropped: ${dropped}   (duplicate/contended, collapsed at consume)`);
+    console.log(`  per-worker ran          : { ${perWorker} }`);
+    console.log("  ---- invariants ----");
+    let ok = true;
+    if (dedupOn)
+      ok = check("L1 producer dedup: inserted == distinct tenants", total === a.tenants, `inserted=${total}, tenants=${a.tenants}`) && ok;
+    ok = check("L2 G2 mutual exclusion: no overlapping runs per key", overlaps === 0, `${overlaps} overlaps`) && ok;
+    ok = check("L2 no double-run per tenant (runs==1)", doubleRun === 0, `${doubleRun} tenants ran >1x`) && ok;
+    ok = check("every tenant migrated exactly once", ran === a.tenants && migrated === a.tenants, `ran=${ran}, migrated=${migrated}, expected=${a.tenants}`) && ok;
+    ok = check("every inserted event consumed once", got === total, `got=${got}, inserted=${total}`) && ok;
+    ok = check("all processing windows closed", unfinished === 0, `${unfinished} unfinished`) && ok;
+    return ok;
+  });
+}
+
+// ---------------------------------------------------------------------------
+// Tier B — ordered per key (lifecycle workload)
+// ---------------------------------------------------------------------------
+async function runTierB(a: Args): Promise<boolean> {
+  const queue = "lifecycle";
+  const n = a.slots;
+  const types = ["FileCreated", "FileOverwritten", "FileDeleted"];
+
+  const produceS = await withClient(async (c) => {
+    await c.query("select demo.reset()");
+    await resetQueue(c, queue);
+    for (let k = 0; k < n; k++)
+      await c.query("select pgque.register_consumer($1,$2)", [queue, `life_slot_${k}`]);
+
+    // ordered lifecycle events, interleaved across tenants so each tick window
+    // holds a mix of keys; per-tenant emission order is preserved (round j emits
+    // each tenant's j-th event) → ev_id monotonic per key.
+    const t0 = performance.now();
+    let pending = 0;
+    for (let j = 0; j < a.eventsPerTenant; j++) {
+      const etype = j === 0 ? types[0] : j === a.eventsPerTenant - 1 ? types[2] : types[1];
+      for (let t = 0; t < a.tenants; t++) {
+        await c.query("select demo.produce($1,$2,$3,$4)", [
+          queue,
+          etype,
+          `{"seq":${j}}`,
+          `tenant-${t}`,
+        ]);
+        if (++pending >= a.chunk) {
+          await tick(c, queue);
+          pending = 0;
+        }
+      }
+    }
+    if (pending) await tick(c, queue);
+    return (performance.now() - t0) / 1000;
+  });
+  const total = a.tenants * a.eventsPerTenant;
+
+  // drain: static assignment — worker k owns slot k
+  const agg = { scanned: 0, delivered: 0 };
+  const t1 = performance.now();
+  await Promise.all(
+    Array.from({ length: n }, (_, k) =>
+      withClient(async (c) => {
+        let empty = 0;
+        while (empty < 2) {
+          const r = await c.query(
+            "select scanned, delivered from demo.tier_b_consume($1,$2,$3,$4,$5)",
+            [queue, `life_slot_${k}`, k, n, a.maxBatch],
+          );
+          const sc = +r.rows[0].scanned;
+          agg.scanned += sc;
+          agg.delivered += +r.rows[0].delivered;
+          empty = sc === 0 ? empty + 1 : 0;
+          if (sc === 0) await sleep(10);
+        }
+      }),
+    ),
+  );
+  const drainS = (performance.now() - t1) / 1000;
+
+  return withClient(async (c) => {
+    const consumed = +(await c.query("select count(*) from demo.consume_log")).rows[0].count;
+    const distinct = +(await c.query("select count(distinct msg_id) from demo.consume_log")).rows[0].count;
+    const multislot = +(
+      await c.query(`select count(*) from (
+        select part_key from demo.consume_log group by part_key
+        having count(distinct slot) > 1) z`)
+    ).rows[0].count;
+    const outoforder = +(
+      await c.query(`select count(*) from (
+        select msg_id, lag(msg_id) over (partition by part_key order by seq) as prev
+        from demo.consume_log) t
+        where prev is not null and msg_id < prev`)
+    ).rows[0].count;
+
+    const amp = agg.delivered ? agg.scanned / agg.delivered : 0;
+    console.log(`  produced events         : ${total}  (${a.tenants} tenants x ${a.eventsPerTenant} events)`);
+    console.log(`  produce time            : ${produceS.toFixed(2)}s`);
+    console.log(`  drain time              : ${drainS.toFixed(2)}s   (${n} slots = ${n} workers)`);
+    console.log(`  consume throughput      : ${fmt(consumed / drainS)} events/s`);
+    console.log(`  read amplification      : ${amp.toFixed(2)}x   (scanned ${fmt(agg.scanned)} / delivered ${fmt(agg.delivered)}; ideal = N = ${n})`);
+    console.log("  ---- invariants ----");
+    let ok = true;
+    ok = check("delivered exactly once (no loss, no dup)", consumed === total && distinct === total, `consumed=${consumed}, distinct=${distinct}, produced=${total}`) && ok;
+    ok = check("G1 affinity: each key on exactly one slot", multislot === 0, `${multislot} keys on >1 slot`) && ok;
+    ok = check("G1 FIFO: per-key msg_id non-decreasing", outoforder === 0, `${outoforder} out-of-order`) && ok;
+    return ok;
+  });
+}
+
+// ---------------------------------------------------------------------------
+// Dedup hazard guardrail — the version-suppression footgun.
+//
+// TTL dedup keyed on the ENTITY alone (tenant) silently drops a *needed* job
+// when the desired effect changes inside the window: ship migration v1, then
+// v2 within the TTL, and the v2 job is suppressed as a "duplicate". The fix is
+// to key on the desired EFFECT (tenant + target version), so a new version is
+// a new key. This runs both schemes back to back and proves it.
+// ---------------------------------------------------------------------------
+async function runDedupHazard(a: Args): Promise<boolean> {
+  const queue = "migrations";
+  const ttl = a.dedupTtl > 0 ? a.dedupTtl : 300;
+  const T = a.tenants;
+
+  async function wave(c: pg.Client, mode: string, ver: number): Promise<number> {
+    let inserted = 0;
+    for (let t = 0; t < T; t++) {
+      const idem = mode === "tenant" ? `migrate:tenant-${t}` : `migrate:tenant-${t}:v${ver}`;
+      const r = await c.query(
+        "select deduped from demo.send_idem($1,$2,$3,$4,$5, make_interval(secs => $6))",
+        [queue, "migrate", `{"v":${ver}}`, `tenant-${t}`, idem, ttl],
+      );
+      if (!r.rows[0].deduped) inserted++;
+    }
+    return inserted;
+  }
+
+  const res: Record<string, { w1: number; w2: number }> = {};
+  for (const mode of ["tenant", "version"]) {
+    await withClient(async (c) => {
+      await c.query("truncate demo.idem");
+      await resetQueue(c, queue);
+      const w1 = await wave(c, mode, 1); // ship migration v1
+      const w2 = await wave(c, mode, 2); // ship v2 INSIDE the TTL window
+      res[mode] = { w1, w2 };
+    });
+  }
+
+  console.log(`  scenario: ${T} tenants, ship migration v1 then v2 within TTL=${ttl}s`);
+  console.log(`  ${"idempotency key".padEnd(20)} ${"v1 inserted".padStart(12)} ${"v2 inserted".padStart(12)}   outcome`);
+  console.log(`  ${"migrate:tenant".padEnd(20)} ${String(res.tenant.w1).padStart(12)} ${String(res.tenant.w2).padStart(12)}   <- v2 SUPPRESSED (hazard)`);
+  console.log(`  ${"migrate:tenant:vN".padEnd(20)} ${String(res.version.w1).padStart(12)} ${String(res.version.w2).padStart(12)}   <- v2 delivered (correct)`);
+  console.log("  ---- guardrail ----");
+  let ok = true;
+  ok = check("hazard is real: entity-only key DROPS the v2 migration", res.tenant.w2 === 0, `v2 inserted=${res.tenant.w2}, expected 0`) && ok;
+  ok = check("fix: effect-scoped key (tenant+version) delivers BOTH waves", res.version.w1 === T && res.version.w2 === T, `v1=${res.version.w1}, v2=${res.version.w2}, expected ${T}`) && ok;
+  return ok;
+}
+
+// ---------------------------------------------------------------------------
+function parseArgs(): Args {
+  const a: any = {
+    tier: "", tenants: 1000, workers: 8, dups: 4, workMs: 3,
+    producers: 4, dedupTtl: 0,
+    slots: 8, eventsPerTenant: 20, chunk: 500, maxBatch: 100000,
+  };
+  const map: Record<string, keyof Args> = {
+    "--tier": "tier", "--tenants": "tenants", "--workers": "workers",
+    "--dups": "dups", "--work-ms": "workMs", "--producers": "producers",
+    "--dedup-ttl": "dedupTtl", "--slots": "slots",
+    "--events-per-tenant": "eventsPerTenant", "--chunk": "chunk", "--max-batch": "maxBatch",
+  };
+  const argv = process.argv.slice(2);
+  for (let i = 0; i < argv.length; i += 2) {
+    const k = map[argv[i]];
+    if (!k) throw new Error(`unknown arg ${argv[i]}`);
+    (a as any)[k] = k === "tier" ? argv[i + 1] : Number(argv[i + 1]);
+  }
+  if (!["a", "b", "hazard"].includes(a.tier))
+    throw new Error("--tier must be a, b, or hazard");
+  return a as Args;
+}
+
+const args = parseArgs();
+const ok = await (args.tier === "a"
+  ? runTierA(args)
+  : args.tier === "b"
+    ? runTierB(args)
+    : runDedupHazard(args));
+process.exit(ok ? 0 : 1);
diff --git a/blueprints/partition-keys/repro/package.json b/blueprints/partition-keys/repro/package.json
new file mode 100644
index 00000000..81209da6
--- /dev/null
+++ b/blueprints/partition-keys/repro/package.json
@@ -0,0 +1,16 @@
+{
+  "name": "pgque-partition-keys-repro",
+  "private": true,
+  "type": "module",
+  "description": "Two-tier partition-keys reproduction (bun + node-postgres)",
+  "scripts": {
+    "tier-a": "bun driver.ts --tier a",
+    "tier-b": "bun driver.ts --tier b"
+  },
+  "dependencies": {
+    "pg": "^8.16.0"
+  },
+  "devDependencies": {
+    "@types/pg": "^8.15.2"
+  }
+}
diff --git a/blueprints/partition-keys/repro/run.sh b/blueprints/partition-keys/repro/run.sh
new file mode 100755
index 00000000..dae71e53
--- /dev/null
+++ b/blueprints/partition-keys/repro/run.sh
@@ -0,0 +1,37 @@
+#!/usr/bin/env bash
+# End-to-end: provision, then run both Fabrizio cases and print reports.
+# Override any knob via env, e.g.  A_WORKERS=16 B_SLOTS=16 bash run.sh
+set -Eeuo pipefail
+
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+DB="${PGQUE_DB:-pgque_repro}"
+
+PGQUE_DB="$DB" bash "$HERE/setup.sh"
+
+export PATH="$HOME/.bun/bin:$PATH"
+export PGHOST="${PGHOST:-/var/run/postgresql}"
+export PGDATABASE="$DB"
+export PGUSER="${PGUSER:-$(id -un)}"
+
+drive() { ( cd "$HERE" && bun driver.ts "$@" ); }
+
+echo
+echo "############ CASE 1 — migrations: producer idempotency + mutual exclusion ############"
+echo "--- 1a: producer dedup OFF — duplicates collapse at consume (advisory lock) ---"
+drive --tier a --tenants "${A_TENANTS:-1000}" --producers "${A_PRODUCERS:-4}" \
+  --dups "${A_DUPS:-3}" --dedup-ttl 0 --workers "${A_WORKERS:-8}" --work-ms "${A_WORK_MS:-1}"
+echo
+echo "--- 1b: producer dedup ON (TTL window) — duplicates never inserted ---"
+drive --tier a --tenants "${A_TENANTS:-1000}" --producers "${A_PRODUCERS:-4}" \
+  --dups "${A_DUPS:-3}" --dedup-ttl "${A_DEDUP_TTL:-60}" --workers "${A_WORKERS:-8}" --work-ms "${A_WORK_MS:-1}"
+echo
+echo "--- 1c: dedup KEY-SCOPE guardrail — entity-only key vs effect-scoped key ---"
+drive --tier hazard --tenants "${A_TENANTS:-1000}" --dedup-ttl 300
+
+echo
+echo "############ CASE 2 — lifecycle: ordered per tenant, parallel across tenants ############"
+drive --tier b --tenants "${B_TENANTS:-500}" --events-per-tenant "${B_EPT:-20}" \
+  --slots "${B_SLOTS:-8}"
+
+echo
+echo "Done. Tweak scale with A_*/B_* env vars (see README.md)."
diff --git a/blueprints/partition-keys/repro/schema.sql b/blueprints/partition-keys/repro/schema.sql
new file mode 100644
index 00000000..e9fc9e59
--- /dev/null
+++ b/blueprints/partition-keys/repro/schema.sql
@@ -0,0 +1,219 @@
+-- Partition-keys reproduction — demo schema (NOT part of pgque core).
+-- Installs alongside sql/pgque.sql into a throwaway database. Everything here
+-- is a thin recipe over existing engine primitives, to validate the two-tier
+-- design from blueprints/partition-keys/SPEC.md empirically.
+--
+-- Copyright 2026 Nikolay Samokhvalov. Apache-2.0 license.
+
+create schema if not exists demo;
+
+-- ---------------------------------------------------------------------------
+-- Invariant-checking logs
+-- ---------------------------------------------------------------------------
+
+-- Tier A (mutual exclusion / G2): each time a worker *runs* a job for a key it
+-- records the processing window. G2 holds iff no two windows for the same key,
+-- owned by different workers, overlap in time.
+create table if not exists demo.mutex_log (
+    id         bigserial primary key,
+    part_key   text not null,
+    worker     text not null,
+    started_at timestamptz not null default clock_timestamp(),
+    ended_at   timestamptz
+);
+
+-- Tier A idempotency target: "tenant is on latest migration". A real handler
+-- runs the migration; here we record that it ran. `runs` > 1 for any tenant
+-- would itself be a mutual-exclusion failure (two workers both ran it).
+create table if not exists demo.tenant_migrated (
+    tenant      text primary key,
+    migrated_at timestamptz not null default clock_timestamp(),
+    runs        int not null default 1
+);
+
+-- Tier B (ordered per key / G1): every delivered event, in consume order, with
+-- the slot that owned it. G1 holds iff, per key, msg_id is non-decreasing in
+-- seq order and every row for a key carries the same slot.
+create table if not exists demo.consume_log (
+    seq         bigserial primary key,
+    part_key    text not null,
+    msg_id      bigint not null,
+    slot        int not null,
+    worker      text not null,
+    consumed_at timestamptz not null default clock_timestamp()
+);
+
+create or replace function demo.reset() returns void language plpgsql as $$
+begin
+  truncate demo.mutex_log, demo.consume_log;
+  delete from demo.tenant_migrated;
+end $$;
+
+-- ---------------------------------------------------------------------------
+-- BENCH ONLY: a minimal pg-boss-style mutable job table, used by bench.ts to
+-- contrast against PgQue's append+rotation model. A job's life is
+-- insert(created) -> update(active) -> delete(complete): every processed job
+-- leaves dead tuples that vacuum must reclaim. This is the row churn PgQue
+-- avoids (events are append-only; rotation TRUNCATEs whole tables).
+-- Not part of any pgque feature.
+-- ---------------------------------------------------------------------------
+create table if not exists demo.jobq (
+    id         bigserial primary key,
+    key        text,
+    state      text not null default 'created',
+    payload    text,
+    created_at timestamptz not null default now()
+);
+create index if not exists jobq_state_id on demo.jobq (state, id);
+
+-- ---------------------------------------------------------------------------
+-- Keyed producer (decision D1/D6): the partition key rides in ev_extra1.
+-- This is the only "new" producer surface the design needs; here it is just a
+-- one-line wrapper over the existing 7-arg insert_event.
+-- ---------------------------------------------------------------------------
+create or replace function demo.produce(
+    i_queue text, i_type text, i_payload text, i_key text)
+returns bigint language plpgsql as $$
+begin
+  return pgque.insert_event(i_queue, i_type, i_payload, i_key, null, null, null);
+end $$;
+
+-- ---------------------------------------------------------------------------
+-- PRODUCER-SIDE IDEMPOTENCY (Case 1, the literal ask): a TTL dedup window
+-- (nik's option 1; the SQS/NATS model). This is what prevents duplicate
+-- *inserts* — distinct from the consumer-side mutual exclusion below, which
+-- only prevents duplicate *work*. The two are complementary layers.
+--
+-- Multi-producer-safe: the unique key + atomic upsert serialize concurrent
+-- producers, so exactly one caller wins per idempotency key per window. The
+-- table is append-only-ish (one row per live key) and GC'd by expiry — in a
+-- real install this maps onto a window GC'd by rotation (see the idempotency
+-- design note).
+-- ---------------------------------------------------------------------------
+create table if not exists demo.idem (
+    idem_key   text primary key,
+    expires_at timestamptz not null
+);
+
+create or replace function demo.send_idem(
+    i_queue text, i_type text, i_payload text, i_key text,
+    i_idem text, i_ttl interval,
+    out event_id bigint, out deduped boolean)
+language plpgsql as $$
+declare
+  v_claimed boolean;
+begin
+  -- Claim the idempotency key for this window. Fresh key -> insert wins;
+  -- expired key -> the WHERE lets the update win; live key -> 0 rows.
+  insert into demo.idem(idem_key, expires_at)
+    values (i_idem, now() + i_ttl)
+  on conflict (idem_key) do update
+    set expires_at = excluded.expires_at
+    where demo.idem.expires_at <= now()
+  returning true into v_claimed;
+
+  if v_claimed then
+    event_id := pgque.insert_event(i_queue, i_type, i_payload, i_key, i_idem, null, null);
+    deduped := false;
+  else
+    event_id := null;        -- duplicate within the window: NOT inserted
+    deduped := true;
+  end if;
+end $$;
+
+-- ---------------------------------------------------------------------------
+-- TIER A — mutual exclusion via cooperative consumers + per-key advisory lock.
+--
+-- Consume one cooperative batch (the engine spreads batches across workers),
+-- serialize per key with a NON-BLOCKING per-key advisory lock, and ack-drop
+-- contended or duplicate events. No new slots, no N, no assignment.
+--
+--   i_work_ms : simulated work while holding the key (widens the window so
+--               concurrent collisions are observable / measurable).
+-- returns: got = events seen, ran = jobs actually executed, dropped = ack-drop
+--          (either contended on the lock, or duplicate of an already-done key)
+-- ---------------------------------------------------------------------------
+create or replace function demo.tier_a_consume(
+    i_queue text, i_main text, i_worker text, i_max int, i_work_ms int default 0,
+    out got int, out ran int, out dropped int)
+language plpgsql as $$
+declare
+  m         pgque.message;
+  v_batch   bigint := null;
+  v_log_id  bigint;
+  v_already boolean;
+begin
+  got := 0; ran := 0; dropped := 0;
+  for m in select * from pgque.receive_coop(i_queue, i_main, i_worker, i_max) loop
+    v_batch := m.batch_id;
+    got := got + 1;
+
+    -- per-key mutual exclusion: try, never block (the non-blocking claim).
+    if pg_try_advisory_xact_lock(hashtextextended(m.extra1, 0)) then
+      -- idempotency: a duplicate for an already-migrated tenant collapses to a
+      -- no-op (free-once-done, without any producer-side dedup).
+      select true into v_already from demo.tenant_migrated where tenant = m.extra1;
+      if v_already then
+        dropped := dropped + 1;
+      else
+        insert into demo.mutex_log(part_key, worker) values (m.extra1, i_worker)
+          returning id into v_log_id;
+        if i_work_ms > 0 then perform pg_sleep(i_work_ms / 1000.0); end if;
+        insert into demo.tenant_migrated(tenant) values (m.extra1)
+          on conflict (tenant) do update set runs = demo.tenant_migrated.runs + 1;
+        update demo.mutex_log set ended_at = clock_timestamp() where id = v_log_id;
+        ran := ran + 1;
+      end if;
+    else
+      dropped := dropped + 1;  -- another worker holds this key right now
+    end if;
+  end loop;
+
+  if v_batch is not null then
+    perform pgque.ack(v_batch);
+  end if;
+end $$;
+
+-- ---------------------------------------------------------------------------
+-- TIER B — ordered per key via N hash-routed slot subscriptions.
+--
+-- Each slot is its own consumer (own cursor). It reads the full tick window and
+-- server-side-filters to its hash class through the existing get_batch_cursor
+-- extra_where hook. The engine is untouched.
+--
+-- returns: scanned   = events in the tick window this slot read (read-amp num)
+--          delivered = events that passed the hash filter (read-amp denom)
+-- ---------------------------------------------------------------------------
+create or replace function demo.tier_b_consume(
+    i_queue text, i_slot_consumer text, i_k int, i_n int, i_max int,
+    out scanned int, out delivered int)
+language plpgsql as $$
+declare
+  v_batch  bigint;
+  e        record;
+  v_cursor text;
+begin
+  scanned := 0; delivered := 0;
+  v_batch := pgque.next_batch(i_queue, i_slot_consumer);
+  if v_batch is null then
+    return;
+  end if;
+
+  -- read amplification: the whole tick window is scanned by every slot.
+  select count(*) into scanned from pgque.get_batch_events(v_batch);
+
+  v_cursor := 'pk_cur_' || i_slot_consumer || '_' || v_batch::text;
+  for e in
+    select *
+    from pgque.get_batch_cursor(
+      v_batch, v_cursor, i_max,
+      format('(hashtextextended(ev_extra1, 0) %% %s + %s) %% %s = %s',
+             i_n, i_n, i_n, i_k))
+  loop
+    insert into demo.consume_log(part_key, msg_id, slot, worker)
+      values (e.ev_extra1, e.ev_id, i_k, i_slot_consumer);
+    delivered := delivered + 1;
+  end loop;
+
+  perform pgque.ack(v_batch);
+end $$;
diff --git a/blueprints/partition-keys/repro/setup.sh b/blueprints/partition-keys/repro/setup.sh
new file mode 100755
index 00000000..14c1d32f
--- /dev/null
+++ b/blueprints/partition-keys/repro/setup.sh
@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+# Provision a throwaway Postgres + bun project, install pgque core + demo
+# schema. Idempotent. Targets Debian/Ubuntu (the common fresh-VM case).
+# Run with sudo on a fresh VM.
+set -Eeuo pipefail
+
+DB="${PGQUE_DB:-pgque_repro}"
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT="$(cd "$HERE/../../.." && pwd)"          # repo root
+PGQUE_SQL="$ROOT/sql/pgque.sql"
+OSUSER="$(id -un)"                            # role for peer auth
+
+[[ -f "$PGQUE_SQL" ]] || { echo "cannot find $PGQUE_SQL — run from inside the pgque repo" >&2; exit 1; }
+
+if ! command -v psql >/dev/null 2>&1; then
+  echo "==> installing PostgreSQL (apt)"
+  export DEBIAN_FRONTEND=noninteractive
+  apt-get update -qq
+  apt-get install -y -qq postgresql postgresql-contrib
+fi
+
+if ! command -v bun >/dev/null 2>&1 && [[ ! -x "$HOME/.bun/bin/bun" ]]; then
+  echo "==> installing bun"
+  curl -fsSL https://bun.sh/install | bash
+fi
+export PATH="$HOME/.bun/bin:$PATH"
+
+# make sure a cluster is running
+pg_lsclusters -h 2>/dev/null | grep -q online || pg_ctlcluster "$(pg_lsclusters -h | awk 'NR==1{print $1}')" main start || service postgresql start || true
+
+echo "==> (re)creating database '$DB' and a superuser role for '$OSUSER'"
+sudo -u postgres psql -v ON_ERROR_STOP=1 -q <<SQL
+select 'create database ${DB}'
+ where not exists (select from pg_database where datname = '${DB}')\gexec
+do \$\$ begin
+  if not exists (select from pg_roles where rolname = '${OSUSER}') then
+    execute format('create role %I login superuser', '${OSUSER}');
+  end if;
+end \$\$;
+SQL
+
+echo "==> installing pgque core + demo schema"
+sudo -u postgres psql -v ON_ERROR_STOP=1 -q -d "$DB" -f "$PGQUE_SQL" >/dev/null
+sudo -u postgres psql -v ON_ERROR_STOP=1 -q -d "$DB" -f "$HERE/schema.sql"
+
+echo "==> bun install"
+( cd "$HERE" && bun install >/dev/null )
+
+echo "==> done. database='$DB', role='$OSUSER'. Run: bash $HERE/run.sh"