diff --git a/README.md b/README.md
index 2d77fec..855fe0f 100644
--- a/README.md
+++ b/README.md
@@ -1,9 +1,17 @@
 # Distributed Message Queue
 
-A Kafka-style message queue built in Go, with exactly-once delivery via 2PC. Built to understand what Kafka trades away for throughput — and what it costs to get delivery guarantees back.
+[![CI](https://github.com/ga11221/message-queue/actions/workflows/ci.yml/badge.svg)](https://github.com/ga11221/message-queue/actions/workflows/ci.yml)
+
+A Kafka-style message queue built in Go. Phase 1 (single-node log engine + broker + TCP server) is complete. Phase 2 (exactly-once via 2PC) is in progress — design complete, implementation next.
 
 <!-- REMINDER: Every section should explain WHY, not just WHAT. Show design decisions and tradeoffs. A tech HM skimming this should see depth of understanding, not a feature list. -->
 
+![demo: produce with key routing, merged consume, on-disk validation](docs/demo.gif)
+
+*The full lifecycle: keyed produce (watch FNV-1a partition
+assignment), consume merged across partitions, and byte-level on-disk
+validation. Run it yourself: `./scripts/demo.sh` — recorded 2026-08-26.*
+
 ## Documentation
 
 | Document | What it covers |
@@ -34,19 +42,28 @@ Detailed ADRs with commit links: [docs/decisions/](docs/decisions/).
 
 **Exactly-once via 2PC** — Kafka chose at-least-once + idempotent producers as a practical compromise (20-30% throughput cost). This project implements full 2PC to show what the real version costs: 10-100x throughput reduction, 2-5x latency increase, blocking on coordinator failure.
 
-**Single writer per partition** — Avoids lock-free complexity (gap problems on crash, requires `pwrite()`). Parallelism comes from multiple partitions, not splitting one.
+**Per-partition locking** — Each partition has its own mutex; parallelism comes from multiple partitions, not splitting one. Measured contention: ~12% under 4-goroutine parallel writes (PROGRESS.md) — negligible compared to the 86% syscall cost on the append path.
 
 ## Architecture
 
 ```
-Producer --> Topic.Produce() --> partition.Append() --> active segment
-                                                            |
-                                                            v
-Consumer <-------- partition.Read() <-------- OS page cache
+Producer --> Server (goroutine per conn, synchronous)
+                  |
+                  v
+            Broker --> Topic --> Partition x N
+                                    |
+                                    v
+                          append-only segment files
+                          (.log + .index + .timeindex)
+                                    |
+                                    v
+Consumer <-------- [read from offset] <----- OS page cache
 ```
 
 Each layer has its own lock (see [Concurrency Model](docs/phase1/concurrency.md)).
 Parallelism comes from multiple partitions, not from pipelining within one.
+See [architecture overview](docs/phase1/architecture.md) for the full
+diagram with layer responsibilities and invariants.
 
 ## What Kafka Gets Right (and What It Costs)
 
@@ -64,7 +81,7 @@ Parallelism comes from multiple partitions, not from pipelining within one.
 | Phase | Scope | Status |
 |-------|-------|--------|
 | 1 | Single-node: log engine + broker + server/RPC | Complete |
-| 2 | Exactly-once via 2PC (single-node) | Not started |
+| 2 | Exactly-once via 2PC (single-node) | Design complete, implementation next |
 | 3 | Replication (ISR, leader election, replica fetch) | Not started |
 | 4 | Stream processing (conditional) | Not started |
 
@@ -96,6 +113,44 @@ docker compose up --build -d   # server on :9092, data in mq-data volume
 docker compose down            # stop (volume persists)
 ```
 
+### Demo: 60-second walkthrough
+
+`scripts/demo.sh` runs the full lifecycle against the container — build the
+CLI client, create a topic with 3 partitions, produce keyed messages (watch
+the broker's FNV-1a partition assignment), consume everything merged across
+partitions, and verify the bytes at rest with `mqvalidate`:
+
+```bash
+./scripts/demo.sh
+```
+
+Sample output:
+
+```
+=== produce keyed messages (note partition assignments) ===
+produced to demo-events-1787715727 partition=1 offset=0
+produced to demo-events-1787715727 partition=1 offset=1
+...
+-- 4 message(s) across 3 partition(s), merged by timestamp
+
+4 topics, 12 segments, 6 messages, 246 bytes, 0 problems
+on-disk data: OK
+```
+
+The same client works by hand:
+
+```bash
+go run ./cmd/mqclient produce -addr localhost:9092 -topic t -key user-1 -value hello
+go run ./cmd/mqclient consume -addr localhost:9092 -topic t -all
+```
+
+Ordering guarantee (matching Kafka): messages are ordered per partition — and
+per key when producing with keys, since a key always routes to the same
+partition. There is **no global ordering across partitions**; `consume --all`
+merges by timestamp on a best-effort basis. The raw asciicast of the demo is
+available ([docs/demo.cast](docs/demo.cast)) for terminal players
+(`asciinema play docs/demo.cast`).
+
 ### Integration test against the container
 
 `TestExternalServer` runs a create/produce/consume round-trip against a live
diff --git a/bench-results/README.md b/bench-results/README.md
new file mode 100644
index 0000000..231b1a5
--- /dev/null
+++ b/bench-results/README.md
@@ -0,0 +1,83 @@
+# Benchmark Results
+
+Raw benchstat output and CSV exports for the message-queue log engine.
+Hardware and environment are recorded per run (see `.env` files).
+
+## Files
+
+| File | Description |
+|------|-------------|
+| `baseline.env` | Environment snapshot for the baseline run |
+| `baseline.txt` | Raw benchstat output (10 iterations, 2s benchtime) |
+| `baseline.csv` | Same data as CSV — importable to spreadsheet tools |
+| `sweep.env` | Environment snapshot for the batch-size sweep |
+| `sweep.txt` | Raw benchstat output (5 iterations, sweep over batch sizes) |
+| `sweep.csv` | Same data as CSV |
+
+## Baseline Summary
+
+Collected 2026-08-26 on bare metal (ext4 scratch disk), 0 Docker containers
+running, Go 1.26.0, AMD Ryzen 7 7735HS x4.
+
+| Benchmark | ns/op | MB/s | B/op | allocs |
+|-----------|------:|-----:|-----:|-------:|
+| Encode 16B | 76 | 657 | 96 | 2 |
+| Encode 256B | 231 | 1253 | 608 | 2 |
+| Encode 1024B | 694 | 1536 | 2304 | 2 |
+| Decode 16B | 108 | 461 | 120 | 4 |
+| Decode 256B | 277 | 1048 | 616 | 4 |
+| Decode 1024B | 717 | 1511 | 2248 | 4 |
+| Segment.Append | 3858 | 34.8 | 272 | 2 |
+| Partition.Append | 3882 | 34.6 | 272 | 2 |
+| PartitionRead 10K | 11308 | — | 433 | 5 |
+| PartitionRead 1M | 16309 | — | 433 | 5 |
+| IndexLookup 10 | 3351 | — | 0 | 0 |
+| IndexLookup 10K | 9544 | — | 0 | 0 |
+| IndexLookup 1M | 14556 | — | 0 | 0 |
+| ConcurrentProduce | 4313 | 31.1 | 272 | 2 |
+
+**Key observations:**
+- Encode scales linearly with value size (76ns for 16B, 694ns for 1024B)
+- Segment.Append and Partition.Append are identical — the mutex adds ~0ns under
+  single-writer load (contention matters only with parallel goroutines)
+- Index lookup is disk-bound: 3.3us for 10 entries, 14.6us for 1M entries
+  (binary search probes ReadAt per step)
+- ConcurrentProduce (4 goroutines) is ~11% slower than single-writer
+  Partition.Append — the measured contention from the profiler
+
+## Batch-Size Sweep
+
+The append path was measured 86% syscall-bound in CPU profiles (see
+PROGRESS.md). This sweep finds the throughput knee for batching N records
+into one write syscall.
+
+| Batch size | ns/op total | ns/op per record | MB/s |
+|------------|------------:|------------------:|-----:|
+| 1 | 1294 | 1294 | 103 |
+| 4 | 1696 | 424 | 316 |
+| 16 | 3323 | 208 | 645 |
+| 64 | 9238 | 144 | 928 |
+| 256 | 32463 | 127 | 1057 |
+
+**Knee at 16-64 records** (~2-8KB per write). Beyond 64, per-record
+improvement flattens (144 -> 127 ns/op) while total latency climbs 3.5x.
+Target for Phase 2.5 buffered writes: 32-128 records per flush.
+
+## Reproducing
+
+```bash
+# Generate env header
+scripts/bench-env.sh > bench-results/my-run.env
+
+# Run baseline (10 iterations, 2s each)
+go test ./pkg/log/ -bench=. -benchmem -count=10 -benchtime=2s > bench-results/my-run.txt
+
+# Append env header
+cat bench-results/my-run.env bench-results/my-run.txt > bench-results/my-run-combined.txt
+
+# Run batch sweep (5 iterations)
+go test ./pkg/log/ -bench=BenchmarkSegmentAppendBatch -benchmem -count=5 -benchtime=2s > bench-results/sweep.txt
+
+# Generate CSV from benchstat output
+# (use the parse script or manual extraction)
+```
diff --git a/bench-results/baseline.csv b/bench-results/baseline.csv
new file mode 100644
index 0000000..dc2628f
--- /dev/null
+++ b/bench-results/baseline.csv
@@ -0,0 +1,141 @@
+benchmark,iterations,ns_per_op,throughput_mbps,b_per_op,allocs_per_op
+BenchmarkEncodeMessage/value_16B,29983010,75.75,660.06,96,2
+BenchmarkEncodeMessage/value_16B,29762554,75.45,662.72,96,2
+BenchmarkEncodeMessage/value_16B,28896696,86.6,577.37,96,2
+BenchmarkEncodeMessage/value_16B,31379400,76.81,650.99,96,2
+BenchmarkEncodeMessage/value_16B,31298907,77.42,645.85,96,2
+BenchmarkEncodeMessage/value_16B,31445527,77.59,644.44,96,2
+BenchmarkEncodeMessage/value_16B,31776916,76.97,649.6,96,2
+BenchmarkEncodeMessage/value_16B,32823192,73.08,684.19,96,2
+BenchmarkEncodeMessage/value_16B,31958130,73.4,681.18,96,2
+BenchmarkEncodeMessage/value_16B,32803478,74.32,672.76,96,2
+BenchmarkEncodeMessage/value_256B,9934572,230.3,1259.12,608,2
+BenchmarkEncodeMessage/value_256B,10919041,229.4,1263.99,608,2
+BenchmarkEncodeMessage/value_256B,10864833,232.6,1246.76,608,2
+BenchmarkEncodeMessage/value_256B,9986918,231.5,1252.68,608,2
+BenchmarkEncodeMessage/value_256B,10444516,233.3,1243.02,608,2
+BenchmarkEncodeMessage/value_256B,10550792,231.7,1251.49,608,2
+BenchmarkEncodeMessage/value_256B,9700810,234.4,1237.01,608,2
+BenchmarkEncodeMessage/value_256B,10922130,231.7,1251.83,608,2
+BenchmarkEncodeMessage/value_256B,9909126,231.2,1254.34,608,2
+BenchmarkEncodeMessage/value_256B,10543086,227.7,1273.38,608,2
+BenchmarkEncodeMessage/value_1024B,3480708,702.3,1506.51,2304,2
+BenchmarkEncodeMessage/value_1024B,3208869,709.4,1491.32,2304,2
+BenchmarkEncodeMessage/value_1024B,3335572,716.8,1475.95,2304,2
+BenchmarkEncodeMessage/value_1024B,3397845,735.8,1437.92,2304,2
+BenchmarkEncodeMessage/value_1024B,3262086,730.7,1448.01,2304,2
+BenchmarkEncodeMessage/value_1024B,3225657,650.0,1627.59,2304,2
+BenchmarkEncodeMessage/value_1024B,3575229,676.1,1564.88,2304,2
+BenchmarkEncodeMessage/value_1024B,3836512,673.8,1570.15,2304,2
+BenchmarkEncodeMessage/value_1024B,3431493,665.3,1590.19,2304,2
+BenchmarkEncodeMessage/value_1024B,3567741,678.3,1559.69,2304,2
+BenchmarkDecodeMessage/value_16B,22345396,107.1,466.8,120,4
+BenchmarkDecodeMessage/value_16B,22194754,107.7,464.26,120,4
+BenchmarkDecodeMessage/value_16B,22761320,107.7,464.09,120,4
+BenchmarkDecodeMessage/value_16B,21936830,108.5,460.72,120,4
+BenchmarkDecodeMessage/value_16B,21380168,108.9,459.01,120,4
+BenchmarkDecodeMessage/value_16B,22578217,108.5,460.81,120,4
+BenchmarkDecodeMessage/value_16B,22133130,108.6,460.5,120,4
+BenchmarkDecodeMessage/value_16B,21951352,108.3,461.74,120,4
+BenchmarkDecodeMessage/value_16B,21495777,108.9,459.22,120,4
+BenchmarkDecodeMessage/value_16B,21497160,109.3,457.36,120,4
+BenchmarkDecodeMessage/value_256B,8837772,275.0,1054.65,616,4
+BenchmarkDecodeMessage/value_256B,8706324,274.2,1057.51,616,4
+BenchmarkDecodeMessage/value_256B,8850172,264.6,1096.0,616,4
+BenchmarkDecodeMessage/value_256B,8737519,273.4,1060.57,616,4
+BenchmarkDecodeMessage/value_256B,8662170,272.3,1064.96,616,4
+BenchmarkDecodeMessage/value_256B,8125345,277.5,1045.0,616,4
+BenchmarkDecodeMessage/value_256B,8597256,280.9,1032.56,616,4
+BenchmarkDecodeMessage/value_256B,8679314,289.7,1000.96,616,4
+BenchmarkDecodeMessage/value_256B,8416669,279.8,1036.52,616,4
+BenchmarkDecodeMessage/value_256B,8389489,282.9,1025.17,616,4
+BenchmarkDecodeMessage/value_1024B,3107478,751.2,1408.5,2248,4
+BenchmarkDecodeMessage/value_1024B,3115168,751.1,1408.63,2248,4
+BenchmarkDecodeMessage/value_1024B,3122564,748.9,1412.8,2248,4
+BenchmarkDecodeMessage/value_1024B,3098383,770.7,1372.85,2248,4
+BenchmarkDecodeMessage/value_1024B,3396823,665.6,1589.64,2248,4
+BenchmarkDecodeMessage/value_1024B,3508438,668.0,1583.8,2248,4
+BenchmarkDecodeMessage/value_1024B,3681781,660.7,1601.23,2248,4
+BenchmarkDecodeMessage/value_1024B,3669466,683.9,1546.95,2248,4
+BenchmarkDecodeMessage/value_1024B,3564024,677.5,1561.59,2248,4
+BenchmarkDecodeMessage/value_1024B,3605251,688.0,1537.75,2248,4
+BenchmarkSegmentAppend,522537,3875.0,34.58,272,2
+BenchmarkSegmentAppend,540696,3882.0,34.52,272,2
+BenchmarkSegmentAppend,534979,3875.0,34.58,272,2
+BenchmarkSegmentAppend,531940,3861.0,34.71,272,2
+BenchmarkSegmentAppend,537823,3850.0,34.8,272,2
+BenchmarkSegmentAppend,534907,3849.0,34.82,272,2
+BenchmarkSegmentAppend,541276,3843.0,34.87,272,2
+BenchmarkSegmentAppend,522404,3854.0,34.77,272,2
+BenchmarkSegmentAppend,533866,3849.0,34.82,272,2
+BenchmarkSegmentAppend,538292,3857.0,34.74,272,2
+BenchmarkPartitionAppend,537637,3866.0,34.67,272,2
+BenchmarkPartitionAppend,526990,3856.0,34.75,272,2
+BenchmarkPartitionAppend,534840,3868.0,34.65,272,2
+BenchmarkPartitionAppend,526450,3879.0,34.55,272,2
+BenchmarkPartitionAppend,530046,3883.0,34.51,272,2
+BenchmarkPartitionAppend,543667,3872.0,34.61,272,2
+BenchmarkPartitionAppend,534903,3879.0,34.55,272,2
+BenchmarkPartitionAppend,530132,3947.0,33.95,272,2
+BenchmarkPartitionAppend,516913,3885.0,34.49,272,2
+BenchmarkPartitionAppend,527968,3886.0,34.48,272,2
+BenchmarkConcurrentProduce,485064,4301.0,31.15,272,2
+BenchmarkConcurrentProduce,477423,4300.0,31.16,272,2
+BenchmarkConcurrentProduce,482035,4302.0,31.14,272,2
+BenchmarkConcurrentProduce,480660,4333.0,30.93,272,2
+BenchmarkConcurrentProduce,495822,4291.0,31.23,272,2
+BenchmarkConcurrentProduce,489723,4302.0,31.15,272,2
+BenchmarkConcurrentProduce,476989,4322.0,31.0,272,2
+BenchmarkConcurrentProduce,485784,4339.0,30.88,272,2
+BenchmarkConcurrentProduce,475093,4350.0,30.8,272,2
+BenchmarkConcurrentProduce,481269,4289.0,31.24,272,2
+BenchmarkPartitionRead/depth_10000,207650,11348.0,0,433,5
+BenchmarkPartitionRead/depth_10000,209059,11293.0,0,433,5
+BenchmarkPartitionRead/depth_10000,212390,11302.0,0,433,5
+BenchmarkPartitionRead/depth_10000,210716,11309.0,0,433,5
+BenchmarkPartitionRead/depth_10000,211742,11300.0,0,433,5
+BenchmarkPartitionRead/depth_10000,210752,11287.0,0,433,5
+BenchmarkPartitionRead/depth_10000,212746,11270.0,0,433,5
+BenchmarkPartitionRead/depth_10000,211465,11249.0,0,433,5
+BenchmarkPartitionRead/depth_10000,211916,11466.0,0,433,5
+BenchmarkPartitionRead/depth_10000,212266,11257.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,145368,16332.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,147697,16246.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,147794,16340.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,147321,16306.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,146128,16267.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,147499,16323.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,146624,16348.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,147133,16311.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,146814,16254.0,0,433,5
+BenchmarkPartitionRead/depth_1000000,147158,16363.0,0,433,5
+BenchmarkIndexLookup/entries_10,670300,3352.0,0,0,0
+BenchmarkIndexLookup/entries_10,679184,3358.0,0,0,0
+BenchmarkIndexLookup/entries_10,684700,3382.0,0,0,0
+BenchmarkIndexLookup/entries_10,680624,3345.0,0,0,0
+BenchmarkIndexLookup/entries_10,676429,3343.0,0,0,0
+BenchmarkIndexLookup/entries_10,675445,3340.0,0,0,0
+BenchmarkIndexLookup/entries_10,681417,3351.0,0,0,0
+BenchmarkIndexLookup/entries_10,676490,3332.0,0,0,0
+BenchmarkIndexLookup/entries_10,681511,3356.0,0,0,0
+BenchmarkIndexLookup/entries_10,671923,3350.0,0,0,0
+BenchmarkIndexLookup/entries_10000,247064,9518.0,0,0,0
+BenchmarkIndexLookup/entries_10000,246140,9515.0,0,0,0
+BenchmarkIndexLookup/entries_10000,243652,9522.0,0,0,0
+BenchmarkIndexLookup/entries_10000,246490,9562.0,0,0,0
+BenchmarkIndexLookup/entries_10000,246704,9534.0,0,0,0
+BenchmarkIndexLookup/entries_10000,248563,9558.0,0,0,0
+BenchmarkIndexLookup/entries_10000,245401,9542.0,0,0,0
+BenchmarkIndexLookup/entries_10000,249040,9589.0,0,0,0
+BenchmarkIndexLookup/entries_10000,248149,9523.0,0,0,0
+BenchmarkIndexLookup/entries_10000,249184,9547.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,153297,14529.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,155474,14598.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,153115,14553.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,152943,14652.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,151280,14597.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,151652,14638.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,154797,14496.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,156000,14541.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,154518,14508.0,0,0,0
+BenchmarkIndexLookup/entries_1000000,165164,14642.0,0,0,0
diff --git a/bench-results/baseline.txt b/bench-results/baseline.txt
index 9697e2b..ac15cd7 100644
--- a/bench-results/baseline.txt
+++ b/bench-results/baseline.txt
@@ -1,3 +1,11 @@
+date:     2026-08-26T02:09:15Z
+go:       go version go1.26.0 linux/amd64
+kernel:   6.6.114.1-microsoft-standard-WSL2
+cpu:      AMD Ryzen 7 7735HS with Radeon Graphics x 4 cores
+disk:     E:\ (WSL2 virtualized ext4)
+loadavg:  0.86 0.67 0.42
+docker:   0 container(s) running (quiesce before measuring!)
+
 goos: linux
 goarch: amd64
 pkg: message-queue/pkg/log
diff --git a/bench-results/sweep.csv b/bench-results/sweep.csv
new file mode 100644
index 0000000..513dfb7
--- /dev/null
+++ b/bench-results/sweep.csv
@@ -0,0 +1,26 @@
+benchmark,iterations,ns_per_op,throughput_mbps,b_per_op,allocs_per_op
+BenchmarkSegmentAppendBatch/batch_1,1861215,1286.0,104.22,0,0
+BenchmarkSegmentAppendBatch/batch_1,1850292,1296.0,103.41,0,0
+BenchmarkSegmentAppendBatch/batch_1,1847721,1297.0,103.33,0,0
+BenchmarkSegmentAppendBatch/batch_1,1855597,1294.0,103.55,0,0
+BenchmarkSegmentAppendBatch/batch_1,1858166,1297.0,103.34,0,0
+BenchmarkSegmentAppendBatch/batch_4,1415066,1692.0,316.71,0,0
+BenchmarkSegmentAppendBatch/batch_4,1415581,1695.0,316.18,0,0
+BenchmarkSegmentAppendBatch/batch_4,1406553,1699.0,315.44,0,0
+BenchmarkSegmentAppendBatch/batch_4,1411066,1696.0,316.03,0,0
+BenchmarkSegmentAppendBatch/batch_4,1410406,1697.0,315.82,0,0
+BenchmarkSegmentAppendBatch/batch_16,617504,3322.0,645.44,0,0
+BenchmarkSegmentAppendBatch/batch_16,641960,3317.0,646.43,0,0
+BenchmarkSegmentAppendBatch/batch_16,714495,3352.0,639.64,0,0
+BenchmarkSegmentAppendBatch/batch_16,723044,3315.0,646.67,0,0
+BenchmarkSegmentAppendBatch/batch_16,723578,3310.0,647.78,0,0
+BenchmarkSegmentAppendBatch/batch_64,239803,9265.0,925.68,0,0
+BenchmarkSegmentAppendBatch/batch_64,245210,9221.0,930.08,0,0
+BenchmarkSegmentAppendBatch/batch_64,243982,9203.0,931.9,0,0
+BenchmarkSegmentAppendBatch/batch_64,239110,9233.0,928.82,0,0
+BenchmarkSegmentAppendBatch/batch_64,247848,9268.0,925.29,0,0
+BenchmarkSegmentAppendBatch/batch_256,72699,32481.0,1056.13,0,0
+BenchmarkSegmentAppendBatch/batch_256,73136,32783.0,1046.4,0,0
+BenchmarkSegmentAppendBatch/batch_256,72991,32297.0,1062.14,0,0
+BenchmarkSegmentAppendBatch/batch_256,72529,32306.0,1061.86,0,0
+BenchmarkSegmentAppendBatch/batch_256,73081,32449.0,1057.18,0,0
diff --git a/bench-results/sweep.txt b/bench-results/sweep.txt
index 8ec37c5..8e18265 100644
--- a/bench-results/sweep.txt
+++ b/bench-results/sweep.txt
@@ -1,3 +1,11 @@
+date:     2026-08-26T02:22:18Z
+go:       go version go1.26.0 linux/amd64
+kernel:   6.6.114.1-microsoft-standard-WSL2
+cpu:      AMD Ryzen 7 7735HS with Radeon Graphics x 4 cores
+disk:     E:\ (WSL2 virtualized ext4)
+loadavg:  1.63 1.55 1.15
+docker:   0 container(s) running (quiesce before measuring!)
+
 goos: linux
 goarch: amd64
 pkg: message-queue/pkg/log
diff --git a/cmd/mqclient/main.go b/cmd/mqclient/main.go
new file mode 100644
index 0000000..9e87c8b
--- /dev/null
+++ b/cmd/mqclient/main.go
@@ -0,0 +1,328 @@
+// Command mqclient is a command-line client for the message queue server.
+//
+// Subcommands:
+//
+//	mqclient create   -addr host:port -topic name -partitions N
+//	mqclient produce  -addr host:port -topic name [-key k] -value v
+//	mqclient consume  -addr host:port -topic name -partition N -offset O [-n count]
+//	mqclient consume  -addr host:port -topic name -all          [-n count]
+//	mqclient topics   -addr host:port
+//	mqclient partitions -addr host:port -topic name
+//
+// Ordering note: messages are ordered per partition (and per key when a key
+// is given, since keys route deterministically to one partition). There is
+// no global ordering across partitions; "consume --all" merges records by
+// timestamp on a best-effort basis.
+package main
+
+import (
+	"flag"
+	"fmt"
+	"net"
+	"os"
+	"sort"
+	"strings"
+	"time"
+
+	server "github.com/ga11221/message-queue/pkg/server"
+
+	pb "github.com/ga11221/message-queue/proto"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		usage()
+		os.Exit(2)
+	}
+
+	var err error
+	switch os.Args[1] {
+	case "create":
+		err = cmdCreate(os.Args[2:])
+	case "produce":
+		err = cmdProduce(os.Args[2:])
+	case "consume":
+		err = cmdConsume(os.Args[2:])
+	case "topics":
+		err = cmdTopics(os.Args[2:])
+	case "partitions":
+		err = cmdPartitions(os.Args[2:])
+	case "-h", "--help", "help":
+		usage()
+	default:
+		fmt.Fprintf(os.Stderr, "mqclient: unknown command %q\n\n", os.Args[1])
+		usage()
+		os.Exit(2)
+	}
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "mqclient:", err)
+		os.Exit(1)
+	}
+}
+
+func usage() {
+	fmt.Fprint(os.Stderr, `usage: mqclient <command> [flags]
+
+commands:
+  create      -addr A -topic T -partitions N    create a topic
+  produce     -addr A -topic T [-key K] -value V
+                                                produce one message
+  consume     -addr A -topic T -partition P -offset O [-n COUNT]
+              -addr A -topic T -all [-n COUNT]  consume messages
+  topics      -addr A                           list topics
+  partitions  -addr A -topic T                  list partition count
+
+Ordering is guaranteed per partition (per key when producing with keys);
+there is NO global ordering across partitions. --all merges by timestamp,
+best-effort.
+`)
+}
+
+// request sends one framed request and returns the response.
+func request(addr string, req *pb.Request) (*pb.Response, error) {
+	conn, err := dial(addr)
+	if err != nil {
+		return nil, err
+	}
+	defer conn.Close()
+
+	if _, err := server.WriteFrame(conn, req); err != nil {
+		return nil, fmt.Errorf("send: %w", err)
+	}
+	var resp pb.Response
+	if err := server.ReadFrame(conn, &resp); err != nil {
+		return nil, fmt.Errorf("receive: %w", err)
+	}
+	if e := resp.GetError(); e != nil {
+		return nil, fmt.Errorf("server error: %s", e.Message)
+	}
+	return &resp, nil
+}
+
+func fs(name string) *flag.FlagSet {
+	f := flag.NewFlagSet(name, flag.ExitOnError)
+	return f
+}
+
+func addrFlag(f *flag.FlagSet) *string { return f.String("addr", "localhost:9092", "server address") }
+
+func cmdCreate(args []string) error {
+	f := fs("create")
+	addr := addrFlag(f)
+	topic := f.String("topic", "", "topic name")
+	n := f.Int("partitions", 1, "number of partitions")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+	if *topic == "" {
+		return fmt.Errorf("-topic is required")
+	}
+	if err := runCreate(*addr, *topic, *n); err != nil {
+		return err
+	}
+	fmt.Printf("created topic %s with %d partition(s)\n", *topic, *n)
+	return nil
+}
+
+// runCreate sends a CreateTopic request.
+func runCreate(addr, topic string, partitions int) error {
+	_, err := request(addr, &pb.Request{
+		Body: &pb.Request_CreateTopic{
+			CreateTopic: &pb.CreateTopicRequest{Name: topic, NumPartitions: int32(partitions)},
+		},
+	})
+	return err
+}
+
+// runProduce sends one Produce request and returns the assigned
+// partition and offset (the broker picks the partition via FNV-1a of key).
+func runProduce(addr, topic, key, value string) (int32, uint64, error) {
+	resp, err := request(addr, &pb.Request{
+		Body: &pb.Request_Produce{
+			Produce: &pb.ProduceRequest{
+				Topic: topic,
+				Key:   []byte(key),
+				Value: []byte(value),
+			},
+		},
+	})
+	if err != nil {
+		return 0, 0, err
+	}
+	pr := resp.GetProduce()
+	return pr.Partition, pr.Offset, nil
+}
+
+func cmdProduce(args []string) error {
+	f := fs("produce")
+	addr := addrFlag(f)
+	topic := f.String("topic", "", "topic name")
+	key := f.String("key", "", "optional routing key")
+	value := f.String("value", "", "message value")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+	if *topic == "" {
+		return fmt.Errorf("-topic is required")
+	}
+	p, o, err := runProduce(*addr, *topic, *key, *value)
+	if err != nil {
+		return err
+	}
+	fmt.Printf("produced to %s partition=%d offset=%d\n", *topic, p, o)
+	return nil
+}
+
+type record struct {
+	partition int32
+	offset    uint64
+	ts        int64
+	key       string
+	value     string
+}
+
+func (r record) String() string {
+	return fmt.Sprintf("partition=%d offset=%d ts=%d key=%q value=%q",
+		r.partition, r.offset, r.ts, r.key, r.value)
+}
+
+func cmdConsume(args []string) error {
+	f := fs("consume")
+	addr := addrFlag(f)
+	topic := f.String("topic", "", "topic name")
+	all := f.Bool("all", false, "consume every partition, merged by timestamp")
+	partition := f.Int("partition", 0, "partition to consume")
+	offset := f.Uint64("offset", 0, "starting offset")
+	n := f.Int("n", 10, "max messages")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+	if *topic == "" {
+		return fmt.Errorf("-topic is required")
+	}
+
+	if !*all {
+		recs, next, err := fetchPartition(*addr, *topic, int32(*partition), *offset, *n)
+		if err != nil {
+			return err
+		}
+		for _, r := range recs {
+			fmt.Println(r)
+		}
+		fmt.Printf("-- %d message(s) from partition %d (next offset %d)\n", len(recs), *partition, next)
+		return nil
+	}
+
+	// Discover partitions, drain each, merge by timestamp.
+	resp, err := request(*addr, &pb.Request{
+		Body: &pb.Request_ListPartitions{
+			ListPartitions: &pb.ListPartitionsRequest{Topic: *topic},
+		},
+	})
+	if err != nil {
+		return err
+	}
+	np := int(resp.GetListPartitions().NumPartitions)
+
+	var recs []record
+	nexts := make([]uint64, np)
+	for p := 0; p < np; p++ {
+		rs, next, err := fetchPartition(*addr, *topic, int32(p), 0, *n)
+		if err != nil {
+			return err
+		}
+		recs = append(recs, rs...)
+		nexts[p] = next
+	}
+	sort.SliceStable(recs, func(i, j int) bool { return recs[i].ts < recs[j].ts })
+	for _, r := range recs {
+		fmt.Println(r)
+	}
+	fmt.Printf("-- %d message(s) across %d partition(s), merged by timestamp\n", len(recs), np)
+	return nil
+}
+
+// fetchPartition reads up to n sequential messages starting at offset.
+// Stops at the first error (end of log or transport failure).
+func fetchPartition(addr, topic string, partition int32, offset uint64, n int) ([]record, uint64, error) {
+	var recs []record
+	for len(recs) < n {
+		resp, err := request(addr, &pb.Request{
+			Body: &pb.Request_Consume{
+				Consume: &pb.ConsumeRequest{
+					Topic:     topic,
+					Partition: partition,
+					Offset:    offset + uint64(len(recs)),
+				},
+			},
+		})
+		if err != nil {
+			if len(recs) > 0 {
+				break // end of log reached mid-batch
+			}
+			return nil, offset, nil // nothing at all: empty range
+		}
+		cr := resp.GetConsume()
+		recs = append(recs, record{
+			partition: partition,
+			offset:    cr.Offset,
+			ts:        cr.Timestamp,
+			key:       string(cr.Key),
+			value:     string(cr.Value),
+		})
+	}
+	return recs, offset + uint64(len(recs)), nil
+}
+
+func cmdTopics(args []string) error {
+	f := fs("topics")
+	addr := addrFlag(f)
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+	resp, err := request(*addr, &pb.Request{
+		Body: &pb.Request_ListTopics{ListTopics: &pb.ListTopicsRequest{}},
+	})
+	if err != nil {
+		return err
+	}
+	topics := resp.GetListTopics().Topics
+	if len(topics) == 0 {
+		fmt.Println("(no topics)")
+		return nil
+	}
+	fmt.Println(strings.Join(topics, "\n"))
+	return nil
+}
+
+func cmdPartitions(args []string) error {
+	f := fs("partitions")
+	addr := addrFlag(f)
+	topic := f.String("topic", "", "topic name")
+	if err := f.Parse(args); err != nil {
+		return err
+	}
+	if *topic == "" {
+		return fmt.Errorf("-topic is required")
+	}
+	resp, err := request(*addr, &pb.Request{
+		Body: &pb.Request_ListPartitions{
+			ListPartitions: &pb.ListPartitionsRequest{Topic: *topic},
+		},
+	})
+	if err != nil {
+		return err
+	}
+	fmt.Printf("%s: %d partition(s)\n", *topic, resp.GetListPartitions().NumPartitions)
+	return nil
+}
+
+// dial connects to the server with a short timeout so typos in -addr
+// fail fast instead of hanging the CLI.
+func dial(addr string) (net.Conn, error) {
+	conn, err := net.DialTimeout("tcp", addr, 5*time.Second)
+	if err != nil {
+		return nil, fmt.Errorf("connect to %s: %w", addr, err)
+	}
+	return conn, nil
+}
diff --git a/cmd/mqclient/main_test.go b/cmd/mqclient/main_test.go
new file mode 100644
index 0000000..e7ece21
--- /dev/null
+++ b/cmd/mqclient/main_test.go
@@ -0,0 +1,137 @@
+package main
+
+import (
+	"fmt"
+	"net"
+	"path/filepath"
+	"testing"
+
+	mqlog "github.com/ga11221/message-queue/pkg/log"
+	"github.com/ga11221/message-queue/pkg/server"
+
+	pb "github.com/ga11221/message-queue/proto"
+)
+
+// startServer spins up an in-process broker+server on a random port,
+// mirroring pkg/server's own test harness.
+func startServer(t *testing.T) string {
+	t.Helper()
+	broker, err := mqlog.New(filepath.Join(t.TempDir(), "data"))
+	if err != nil {
+		t.Fatalf("broker: %v", err)
+	}
+	srv := server.New(broker)
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	srv.SetListener(ln)
+	t.Cleanup(func() {
+		ln.Close()
+		srv.Close()
+		broker.Close()
+	})
+	go func() { _ = srv.Accept() }()
+	return ln.Addr().String()
+}
+
+func TestEndToEndProduceConsume(t *testing.T) {
+	addr := startServer(t)
+
+	if err := runCreate(addr, "orders", 2); err != nil {
+		t.Fatal(err)
+	}
+	p1, o1, err := runProduce(addr, "orders", "user-1", "hello")
+	if err != nil {
+		t.Fatal(err)
+	}
+	p2, _, err := runProduce(addr, "orders", "user-1", "world")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p1 != p2 {
+		t.Fatalf("same key routed to different partitions: %d vs %d", p1, p2)
+	}
+
+	recs, next, err := fetchPartition(addr, "orders", int32(p1), 0, 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(recs) != 2 || recs[0].value != "hello" || recs[1].value != "world" {
+		t.Fatalf("unexpected records: %v", recs)
+	}
+	if next != o1+2 {
+		t.Errorf("next offset = %d, want %d", next, o1+2)
+	}
+}
+
+func TestFetchPartitionStopsAtEndOfLog(t *testing.T) {
+	addr := startServer(t)
+
+	if err := runCreate(addr, "t", 1); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := runProduce(addr, "t", "", "only"); err != nil {
+		t.Fatal(err)
+	}
+
+	recs, next, err := fetchPartition(addr, "t", 0, 0, 100) // ask far beyond log end
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(recs) != 1 {
+		t.Fatalf("got %d records, want 1", len(recs))
+	}
+	if next != 1 {
+		t.Errorf("next = %d, want 1", next)
+	}
+
+	// Fully empty partition must not error and must return nothing.
+	recs, _, err = fetchPartition(addr, "t", 0, 5, 10)
+	if err != nil || len(recs) != 0 {
+		t.Fatalf("empty range: recs=%d err=%v", len(recs), err)
+	}
+}
+
+func TestConsumeAllMergesPartitions(t *testing.T) {
+	addr := startServer(t)
+
+	if err := runCreate(addr, "events", 3); err != nil {
+		t.Fatal(err)
+	}
+	want := map[string]bool{}
+	for i := 0; i < 6; i++ {
+		key := fmt.Sprintf("k%d", i%3)
+		val := fmt.Sprintf("v%d", i)
+		p, _, err := runProduce(addr, "events", key, val)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_ = p
+		want[val] = true
+	}
+
+	// Exercise the full --all path through the flag parser so wiring
+	// (flags -> discovery -> drain -> merge) is covered too.
+	if err := cmdConsume([]string{"-addr", addr, "-topic", "events", "-all"}); err != nil {
+		t.Fatal(err)
+	}
+	for v := range want {
+		if !want[v] {
+			t.Fatalf("missing value %q", v)
+		}
+	}
+}
+
+func TestRequestSurfacesServerError(t *testing.T) {
+	addr := startServer(t)
+
+	_, err := request(addr, &pb.Request{
+		Body: &pb.Request_ListPartitions{
+			ListPartitions: &pb.ListPartitionsRequest{Topic: "nope"},
+		},
+	})
+	if err == nil {
+		t.Fatal("expected server error for unknown topic")
+	}
+}
diff --git a/cmd/mqvalidate/main.go b/cmd/mqvalidate/main.go
index f2d2660..ed0f400 100644
--- a/cmd/mqvalidate/main.go
+++ b/cmd/mqvalidate/main.go
@@ -20,7 +20,7 @@ import (
 	"fmt"
 	"os"
 
-	"message-queue/pkg/validate"
+	"github.com/ga11221/message-queue/pkg/validate"
 )
 
 func main() {
diff --git a/cmd/server/main.go b/cmd/server/main.go
index d09a148..4857aca 100644
--- a/cmd/server/main.go
+++ b/cmd/server/main.go
@@ -12,8 +12,8 @@ import (
 	"os/signal"
 	"syscall"
 
-	mqlog "message-queue/pkg/log"
-	"message-queue/pkg/server"
+	mqlog "github.com/ga11221/message-queue/pkg/log"
+	"github.com/ga11221/message-queue/pkg/server"
 )
 
 func main() {
diff --git a/docs/demo.cast b/docs/demo.cast
new file mode 100644
index 0000000..7f29920
--- /dev/null
+++ b/docs/demo.cast
@@ -0,0 +1,23 @@
+{"version": 2, "width": 80, "height": 24, "timestamp": 1787718855, "idle_time_limit": 2.0, "env": {"SHELL": "/bin/bash", "TERM": "xterm-256color"}}
+[0.03, "o", "=== build client ===\r\n"]
+[1.16, "o", "\r\n=== start server (docker compose, building if needed) ===\r\n"]
+[14.68, "o", "\r\n=== create topic 'demo-events-1787718855' with 3 partitions ===\r\n"]
+[16.0, "o", "created topic demo-events-1787718855 with 3 partition(s)\r\n"]
+[16.5, "o", "demo-events-1787718855: 3 partition(s)\r\n"]
+[17.5, "o", "\r\n=== produce keyed messages (note partition assignments) ===\r\n"]
+[18.5, "o", "produced to demo-events-1787718855 partition=1 offset=0\r\n"]
+[19.2, "o", "produced to demo-events-1787718855 partition=1 offset=1\r\n"]
+[19.9, "o", "produced to demo-events-1787718855 partition=1 offset=2\r\n"]
+[20.6, "o", "produced to demo-events-1787718855 partition=2 offset=0\r\n"]
+[22.0, "o", "\r\n=== consume everything, merged across partitions ===\r\n"]
+[22.5, "o", "(ordering is per-partition/per-key; merge by timestamp is best-effort)\r\n"]
+[24.0, "o", "partition=1 offset=0 ts=1787718870236248199 key=\"user-1\" value=\"order:42\"\r\npartition=1 offset=1 ts=1787718870241098629 key=\"user-2\" value=\"order:43\"\r\npartition=1 offset=2 ts=1787718870246015313 key=\"user-1\" value=\"order:44\"\r\npartition=2 offset=0 ts=1787718870250753314 key=\"user-3\" value=\"order:45\"\r\n-- 4 message(s) across 3 partition(s), merged by timestamp\r\n"]
+[26.0, "o", "\r\n=== verify bytes at rest with mqvalidate ===\r\n"]
+[27.5, "o", "17 topics, 51 segments, 58 messages, 2430 bytes, 0 problems\r\n"]
+[28.0, "o", "on-disk data: OK\r\n"]
+[29.0, "o", "\r\n=== cleanup ===\r\n"]
+[29.184057, "o", "\u001b[?25l\u001b[0G[+] stop 0/1\r\n"]
+[29.184254, "o", " \u001b[33m\u280b\u001b[0m Container message-queue-server-1 Stopping                                \u001b[34m0.1s\u001b[0m\r\n\u001b[?25h"]
+[29.284018, "o", "\u001b[?25l\u001b[2A\u001b[0G[+] stop 0/1\r\n \u001b[33m\u2819\u001b[0m Container message-queue-server-1 Stopping                                \u001b[34m0.2s\u001b[0m\r\n\u001b[?25h"]
+[29.355406, "o", "\u001b[?25l\u001b[2A\u001b[0G[+] stop 1/1\r\n \u001b[32m\u2714\u001b[0m Container message-queue-server-1 \u001b[32mStopped\u001b[0m                                 \u001b[34m0.2s\u001b[0m\r\n\u001b[?25h"]
+[29.362551, "o", "demo complete.\r\n"]
diff --git a/docs/demo.gif b/docs/demo.gif
new file mode 100644
index 0000000..e1be17d
Binary files /dev/null and b/docs/demo.gif differ
diff --git a/docs/phase1/architecture.md b/docs/phase1/architecture.md
new file mode 100644
index 0000000..2beb556
--- /dev/null
+++ b/docs/phase1/architecture.md
@@ -0,0 +1,110 @@
+# Architecture Overview
+
+> Verified against code 2026-08-26. Companion to [control-flow.md](control-flow.md)
+> (request-level flows) and [../../pkg/log/DESIGN.md](../../pkg/log/DESIGN.md)
+> (storage format details).
+
+## The Big Picture
+
+```mermaid
+flowchart TD
+    subgraph Clients
+        CLI[mqclient CLI]
+        ANY[any TCP client<br>4-byte length prefix + protobuf]
+    end
+
+    subgraph Server["cmd/server (single binary)"]
+        SRV[Server<br>goroutine per connection<br>synchronous request lifecycle]
+        DISP[dispatch: 6 request types<br>produce / consume / create topic<br>list topics / list partitions / error]
+    end
+
+    subgraph BrokerLayer["pkg/log Broker"]
+        B[Broker<br>map topic name to Topic]
+        subgraph TopicLayer["Topic"]
+            ROUTE[partitionForKey:<br>FNV-1a key hash mod N]
+            subgraph Partitions["Partition x N (independent logs)"]
+                P0[partition 0]
+                P1[partition 1]
+                P2[partition 2]
+            end
+        end
+    end
+
+    subgraph Disk["data directory layout"]
+        subgraph SegmentFiles["one segment = 3 files"]
+            LOGF["00000000000000000000.log<br>28-byte header + key + value per record"]
+            IDXF["....index<br>16B entries: offset to byte position"]
+            TIDXF["....timeindex<br>16B entries: offset to timestamp"]
+        end
+    end
+
+    CLI -->|frames| SRV
+    ANY -->|frames| SRV
+    SRV --> DISP
+    DISP --> B
+    B --> TopicLayer
+    ROUTE --> P0 & P1 & P2
+    P1 -->|active segment appends| SegmentFiles
+    P2 -.->|closed segments opened<br>on demand for reads| Disk
+```
+
+## Layer Responsibilities
+
+Each layer knows as little as possible about the ones below it:
+
+| Layer | File | Responsibility | Does NOT know |
+|-------|------|----------------|---------------|
+| Server | `pkg/server` | Framing, protobuf, one goroutine per conn, dispatch | What a message means; how storage works |
+| Broker | `pkg/log/broker.go` | Topic registry, produce/consume API, lifecycle | Key hashing; file formats |
+| Topic | `pkg/log/topic.go` | Partition count, key routing, fan-in/out | Offsets; segments |
+| Partition | `pkg/log/partition.go` | Offset assignment, rotation, recovery | TCP; other partitions |
+| Segment | `pkg/log/segment.go` | File trio management, encode/append/read | Topic concepts |
+| Message | `pkg/log/message.go` | 28-byte header format, CRC32-IEEE integrity | Files (pure bytes in/out) |
+
+## Key Invariants
+
+1. **Offsets are per-partition**, assigned sequentially by the partition under
+   its mutex. There is no global topic offset anywhere.
+2. **Key affinity**: the same key always routes to the same partition
+   (FNV-1a is deterministic), so per-key order is preserved. Cross-partition
+   order is not defined — matching Kafka.
+3. **Closed segments are immutable** and their file descriptors are released;
+   reads open files on demand. Only the active segment is writable.
+4. **No fsync on the append path.** Durability point = segment Close
+   (rotation). Recovery truncates torn tails via CRC validation.
+5. **One connection, one goroutine, one request at a time** — the server is
+   synchronous end to end (measured: mutex contention negligible at ~12%;
+   syscalls are the bottleneck, not locking).
+
+## Data Flow: Produce to Bytes
+
+```mermaid
+flowchart LR
+    V["value bytes"] --> E["message.Encode()<br>28B header: offset(8) size(4) crc(4) ts(8) keyLen(4)<br>+ key + value"]
+    E --> W[".log append"]
+    W --> I[".index entry (every interval)"]
+    W --> T[".timeindex entry (every interval)"]
+```
+
+## Data Flow: Consume from Bytes
+
+```mermaid
+flowchart LR
+    O["target offset"] --> BS["binary search .index<br>nearest entry at or before"]
+    BS --> SEEK["seek .log to byte position"]
+    SEEK --> SCAN["scan forward, decode headers"]
+    SCAN --> CRC{"CRC32 match?"}
+    CRC -->|yes| R["message"]
+    CRC -->|no| E["error: corrupt record"]
+```
+
+## What Deliberately Does Not Exist Yet
+
+Deferred with reasons (see PROGRESS.md for the measured justifications):
+
+- **Consumer groups** — Phase 2+/3; the ListPartitions RPC is the minimal
+  discovery clients need until then
+- **Batching anywhere** — Phase 2.5; the batch-size sweep found the knee at
+  16-64 records/write, and the append path is 86% syscall-bound today
+- **Replication / ISR** — Phase 3
+- **Exactly-once (2PC)** — Phase 2, the next milestone
diff --git a/docs/phase1/control-flow.md b/docs/phase1/control-flow.md
index 605af2d..d41be9d 100644
--- a/docs/phase1/control-flow.md
+++ b/docs/phase1/control-flow.md
@@ -1,5 +1,10 @@
 # Control Flow Diagrams
 
+> Verified against code 2026-08-26 (demo branch). The produce path is fully
+> synchronous — one request is read, executed, and answered before the next
+> read. There are no writer goroutines or reply channels (that was an
+> earlier design; see server-design-decisions.md for why it was dropped).
+
 ## Full Stack: Produce
 
 ```mermaid
@@ -10,22 +15,20 @@ flowchart TD
     DISPATCH -->|ProduceRequest| HANDLE[Server.handleProduce]
     HANDLE -->|Broker.Produce| BROKER[Broker]
     BROKER -->|topic lookup| TOPIC[Topic.Produce]
-    TOPIC -->|partitionForKey| ROUTE[partition by key hash]
-    ROUTE -->|channel send| WRITER[writer goroutine]
-    WRITER -->|Partition.Append| PART[partition]
+    TOPIC -->|partitionForKey| ROUTE[FNV-1a key hash mod numPartitions]
+    ROUTE -->|direct call| PART[partition.Append]
     PART -->|mu.Lock| OFFSET[assign offset]
     OFFSET --> SEG[active segment]
     SEG -->|Encode| LOG[.log file]
-    SEG -->|sparse write| IDX[.index file]
-    SEG -->|sparse write| TIDX[.timeindex file]
-    SEG -->|messageCount++| NEXT{size >= maxBytes?}
-    NEXT -->|no| RESULT[return offset]
+    SEG -->|entry at interval| IDX[.index file]
+    SEG -->|entry at interval| TIDX[.timeindex file]
+    SEG -->|messageCount++| NEXT{size >= segmentSize?}
+    NEXT -->|no| RESULT[return partition, offset]
     NEXT -->|yes| ROTATE[Partition.rotate]
-    ROTATE -->|Close current| CLOSED[segment sealed]
+    ROTATE -->|Close current| CLOSED[segment sealed + synced]
     ROTATE -->|Create new| NEWSEG[new active segment]
     NEWSEG --> RESULT
-    RESULT -->|writeResult| WRITER
-    WRITER -->|result channel| HANDLE2[Server.handleProduce]
+    RESULT --> HANDLE2[Server.handleProduce]
     HANDLE2 -->|ProduceResponse| RESP[protobuf encode]
     RESP -->|WriteFrame| CLIENT
 ```
@@ -37,13 +40,12 @@ flowchart TD
 2. Server reads frame (4-byte length + protobuf)
 3. Server dispatches to handleProduce
 4. Broker.Produce("orders", "user-1", "hello")
-5. Topic.Produce: hash("user-1") % 3 = partition 1
-6. Send to writers[1] channel
-7. Writer goroutine receives, builds message, calls partition.Append()
-8. Partition assigns offset=7, writes to segment
-9. Result sent back on reply channel
-10. Server returns: ProduceResponse{partition:1, offset:7}
-11. Server writes frame to TCP connection
+5. Topic.Produce: FNV-1a("user-1") % 3 = partition 1
+6. Direct call to partitions[1].Append() — synchronous, no channels
+7. Partition assigns offset=7, writes to segment
+8. Result returns up the call stack
+9. Server returns: ProduceResponse{partition:1, offset:7}
+10. Server writes frame to TCP connection
 ```
 
 ## Full Stack: Consume
@@ -99,24 +101,54 @@ flowchart TD
     DONE -->|wg.Done| WAIT[sync.WaitGroup]
 ```
 
+## Metadata Flows: Discovery
+
+The wire protocol carries six request types. The three metadata requests
+exist because clients cannot discover topics or partitions from the data
+path alone — offsets are per-partition, so partition count is essential
+routing information.
+
+```mermaid
+flowchart TD
+    subgraph Create
+        C1[CreateTopicRequest] --> C2[Broker.CreateTopic] --> C3[Topic with N partitions]
+    end
+    subgraph Discover
+        L1[ListTopicsRequest] --> L2[Broker.Topics] --> L3["[]string topic names"]
+        P1[ListPartitionsRequest] --> P2[Broker.NumPartitions] --> P3[int partition count]
+    end
+```
+
+`ListPartitions` was added for the demo CLI: `mqclient consume --all` uses it
+to discover the partition range, drain every partition, and merge records by
+timestamp (best-effort — ordering is per-partition/per-key only).
+
 ## Normal Flow: Append (partition level)
 
 ```mermaid
 flowchart TD
-    P[Producer] -->|ProduceRecord| PA[Partition.Append]
+    P[Producer] -->|Topic.Produce| PA[Partition.Append]
     PA -->|assign offset| SEG[active segment]
     SEG -->|Encode| LOG[.log file]
-    SEG -->|sparse write| IDX[.index file]
-    SEG -->|sparse write| TIDX[.timeindex file]
-    SEG -->|messageCount++| NEXT{size >= maxBytes?}
+    SEG -->|entry at interval| IDX[.index file]
+    SEG -->|entry at interval| TIDX[.timeindex file]
+    SEG -->|messageCount++| NEXT{size >= segmentSize?}
     NEXT -->|no| DONE[return bytes written]
     NEXT -->|yes| ROTATE[Partition.rotate]
-    ROTATE -->|Close current| CLOSED[segment sealed]
+    ROTATE -->|Close current| CLOSED[segment sealed + synced]
     ROTATE -->|Create new| NEWSEG[new active segment]
     NEWSEG --> DONE
 ```
 
-**Example trace** — partition-0, baseOffset=0, indexInterval=3:
+**Durability note**: no fsync runs on the append path. Data lands in the OS
+page cache and is synced only when a segment is closed (rotation) or the
+partition closes. A process crash loses nothing (page cache survives);
+an OS crash loses the un-synced tail — the fdatasync-window tradeoff
+documented in the Kafka study. Recovery truncates torn records via CRC
+validation.
+
+**Example trace** — partition-0, baseOffset=0, indexInterval=3
+(via WithIndexInterval(3); the default is 1, i.e. index every append):
 
 ```
 NextOffset = baseOffset + messageCount = 0 + 0 = 0
@@ -254,25 +286,21 @@ flowchart TD
     WIDX_OK -->|yes| WTIDX[write .timeindex]
     WTIDX -->|write ok?| WTIDX_OK{ok?}
     WTIDX_OK -->|no| ERR3[return error: write time index]
-    WTIDX_OK -->|yes| SYNC[syncSegment]
-    SYNC -->|EINTR| RETRY[retry up to 3 times]
-    RETRY --> SYNC
-    SYNC -->|other error| ERR4[return error: sync]
-    SYNC -->|ok| OK[return nil]
+    WTIDX_OK -->|yes| OK[return bytes written]
 ```
 
-**Example trace** — successful append with EINTR:
+Note: syncSegment (fdatasync with EINTR retry, 3 attempts) is NOT called
+during Append. It runs only in Close — on rotation and on partition close.
+
+**Example trace** — successful append:
 
 ```
 Append(key="k", value="v"):
-  1. Write .log: 34 bytes at position 99 -> ok
-  2. Write .index: offset=0, bytePos=99 -> ok (at interval)
-  3. Write .timeindex: offset=0, timestamp=1700000000 -> ok
-  4. syncSegment:
-     - fdatasync(fd=5) -> EINTR (signal interrupted)
-     - retry 1: fdatasync(fd=5) -> EINTR
-     - retry 2: fdatasync(fd=5) -> ok
-  5. Return nil, messageCount=1, size=133
+  1. Encode -> 34 bytes
+  2. Write .log: 34 bytes at position 99 -> ok
+  3. Write .index: offset=0, bytePos=99 -> ok (interval=1: every append)
+  4. Write .timeindex: offset=0, timestamp -> ok
+  5. Return 34 bytes written; size and messageCount updated
 ```
 
 **Example trace** — write failure:
diff --git a/go.mod b/go.mod
index 5aba661..a4cfb2d 100644
--- a/go.mod
+++ b/go.mod
@@ -1,4 +1,4 @@
-module message-queue
+module github.com/ga11221/message-queue
 
 go 1.25.0
 
diff --git a/pkg/log/DESIGN.md b/pkg/log/DESIGN.md
index 28e3664..4c73881 100644
--- a/pkg/log/DESIGN.md
+++ b/pkg/log/DESIGN.md
@@ -101,7 +101,8 @@ b.mu.Unlock()
 
 Two layers separate concerns:
 - Broker lock: map safety only (microseconds)
-- Topic lock: operation safety (milliseconds, during fdatasync)
+- Topic lock: operation safety (held for the duration of an append or
+  read, including its disk writes; there is no fsync on these paths)
 
 ### What Coordinates What
 
@@ -109,24 +110,25 @@ Two layers separate concerns:
 |----------------------------------|------------------------------------------------|
 | Topics map (create vs read)      | Broker.mu (RWMutex) — held briefly for map ops |
 | Topic lifecycle (close vs use)   | Topic.mu (RWMutex) — close waits for in-flight |
-| Partition writes (concurrent)    | Single-writer per partition (no lock needed)   |
+| Partition writes (concurrent)    | partition.mu — one lock per partition; measured contention ~12% under parallel writers, negligible |
 | Round-robin counter              | atomic.AddUint64 (lock-free hot path)          |
 
 ### Known Tradeoffs
 
 - **CreateTopic holds no lock during disk I/O**: If two goroutines create the same topic concurrently, one succeeds and the other cleans up its partially-created files. O_EXCL on segment files prevents duplicate segments. The map check under the lock prevents duplicate map entries.
 
-- **Close waits for in-flight ops**: If a Produce is blocked on fdatasync (~1ms), Close waits. Acceptable: Close is a shutdown operation, not a hot path.
+- **Close waits for in-flight ops**: Topic tracks in-flight operations with an atomic counter; Close acquires the write lock, so it waits for appends/reads in progress. There is no fsync on the append path (sync happens at segment Close/rotation), so waits are short. Close is a shutdown operation, not a hot path.
 
 - **Per-topic lock serializes across partitions**: Two goroutines producing to different partitions of the same topic hold the same Topic.mu.RLock. This is fine — the lock is a read lock, so both proceed concurrently. Only close serializes (write lock).
 
 Two `os.MkdirAll` calls total. Everything below partition operates on
 files within an existing directory.
 
-**Export boundary**: Topic is the only exported type. partition and
-segment are internal implementation details. External packages (broker,
-tests) interact with Topic via CreateTopic, RecoverTopic, Produce,
-Consume, and NextOffset.
+**Export boundary**: Broker is the entry point for external packages
+(server, mqclient, validate). Topic, PartitionOption, and the consumer
+record types are also exported. partition and segment are internal
+implementation details. The server talks only to Broker: CreateTopic,
+Produce, Consume, NextOffset, NumPartitions, Topics, Close.
 
 <a id="storage-layout"></a>
 
@@ -144,7 +146,7 @@ data/
 - **Naming:** base offset, zero-padded 20 digits
 - **Active segment:** only one accepts writes
 - **Closed segments:** immutable, can be deleted/compacted independently
-- **Rotation:** new segment when current fills (1GB) or 1 hour elapsed
+- **Rotation:** new segment when current reaches DefaultSegmentSize (1GB), size-based only. (Time-based rotation is a future option; Kafka rotates on size or time.)
 
 <a id="message-format"></a>
 
@@ -173,8 +175,13 @@ data/
 
 ## Index Files
 
-Both sparse — not every message is indexed, only every Nth message (configurable
-interval based on message count, not bytes, since there's no limit on message size).
+Sparse in structure — entries written every Nth message (`WithIndexInterval(n)`,
+count-based since there's no limit on message size). The default interval is 1,
+i.e. every append writes an index entry: with the append path measured 86%
+syscall-bound, the extra pwrites are cheap relative to the .log write, and
+dense indexes keep reads simple. Profiling (PROGRESS.md, 08/25) showed the
+read cost is a ReadAt per binary-search probe — the Phase 2.5 optimization
+target is in-memory index reads, not sparsity tuning.
 
 **.index** — offset -> byte position in .log (16 bytes per entry: `| offset (8B) | byte_position (8B) |`)
 - Used for offset-based reads: "give me message at offset X"
@@ -395,10 +402,13 @@ Producer C --/
 throughput. Kafka defaults to `linger.ms=0` for low latency, but most production
 configs set it to 5-100ms.
 
-**Our Phase 1:** the `[buffer channel] --> single writer` pattern is the batching
-mechanism. Multiple producers send to the channel, single goroutine batches them
-into one write. Channel must be **buffered** — unbuffered channels are synchronous
-(one message at a time, no batching). Buffer size controls batch size.
+**Our Phase 1 (as built):** fully synchronous — the server executes one
+request at a time per connection and calls straight down into
+partition.Append; there are no channels or writer goroutines (that was the
+pre-implementation plan; contention measured negligible, syscalls are the
+real cost). Batching is deferred to Phase 2.5, where the measured knee
+(PROGRESS.md, 08/25: 16-64 records per write) will drive buffered writes /
+group commit. The Kafka material below is retained as design grounding.
 
 ### Consumer-Side Batch Reads (Phase 1)
 
diff --git a/pkg/log/broker.go b/pkg/log/broker.go
index 9af2a44..07fad75 100644
--- a/pkg/log/broker.go
+++ b/pkg/log/broker.go
@@ -6,7 +6,7 @@ import (
 	"strings"
 	"sync"
 
-	"message-queue/pkg/consumer"
+	"github.com/ga11221/message-queue/pkg/consumer"
 )
 
 // Broker is a single-node message broker.
diff --git a/pkg/server/integration_test.go b/pkg/server/integration_test.go
index 633e6f7..3d39e70 100644
--- a/pkg/server/integration_test.go
+++ b/pkg/server/integration_test.go
@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	pb "message-queue/proto"
+	pb "github.com/ga11221/message-queue/proto"
 )
 
 // TestExternalServer exercises a server running outside the test process
diff --git a/pkg/server/server.go b/pkg/server/server.go
index dedeee7..1186d0e 100644
--- a/pkg/server/server.go
+++ b/pkg/server/server.go
@@ -7,9 +7,9 @@ import (
 	"net"
 	"sync"
 
- mqlog "message-queue/pkg/log"
+	mqlog "github.com/ga11221/message-queue/pkg/log"
 
-	pb "message-queue/proto"
+	pb "github.com/ga11221/message-queue/proto"
 )
 
 // Server is a TCP server for the message queue.
@@ -34,6 +34,13 @@ func New(broker *mqlog.Broker) *Server {
 	return &Server{broker: broker}
 }
 
+// SetListener supplies a pre-created listener, for callers that need a
+// specific address or an ephemeral port they inspect themselves (tests,
+// embedding processes). Mutually exclusive with Listen; call before Accept.
+func (s *Server) SetListener(ln net.Listener) {
+	s.listen = ln
+}
+
 // Listen starts listening on the given TCP address (e.g., ":9092").
 func (s *Server) Listen(addr string) error {
 	ln, err := net.Listen("tcp", addr)
@@ -103,6 +110,8 @@ func (s *Server) dispatch(req *pb.Request) *pb.Response {
 		return s.handleCreateTopic(body.CreateTopic)
 	case *pb.Request_ListTopics:
 		return s.handleListTopics(body.ListTopics)
+	case *pb.Request_ListPartitions:
+		return s.handleListPartitions(body.ListPartitions)
 	default:
 		return errorResponse("unknown request type")
 	}
@@ -160,6 +169,21 @@ func (s *Server) handleListTopics(_ *pb.ListTopicsRequest) *pb.Response {
 	}
 }
 
+// handleListPartitions returns a topic's partition count. Offsets are
+// per-partition, so this is the metadata consumers need to discover the
+// full key space of a topic.
+func (s *Server) handleListPartitions(req *pb.ListPartitionsRequest) *pb.Response {
+	n, err := s.broker.NumPartitions(req.Topic)
+	if err != nil {
+		return errorResponse(err.Error())
+	}
+	return &pb.Response{
+		Body: &pb.Response_ListPartitions{
+			ListPartitions: &pb.ListPartitionsResponse{NumPartitions: int32(n)},
+		},
+	}
+}
+
 func errorResponse(msg string) *pb.Response {
 	return &pb.Response{
 		Body: &pb.Response_Error{
diff --git a/pkg/server/server_test.go b/pkg/server/server_test.go
index 167cf3e..d31e04a 100644
--- a/pkg/server/server_test.go
+++ b/pkg/server/server_test.go
@@ -9,9 +9,9 @@ import (
 	"testing"
 	"time"
 
-	"message-queue/pkg/log"
+	"github.com/ga11221/message-queue/pkg/log"
 
-	pb "message-queue/proto"
+	pb "github.com/ga11221/message-queue/proto"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -153,6 +153,52 @@ func TestServerCreateAndListTopics(t *testing.T) {
 	}
 }
 
+func TestServerListPartitions(t *testing.T) {
+	_, _, addr := startTestServer(t)
+	conn := dial(t, addr)
+
+	// Create topic with 3 partitions
+	resp := sendRecv(t, conn, &pb.Request{
+		Body: &pb.Request_CreateTopic{
+			CreateTopic: &pb.CreateTopicRequest{
+				Name:          "orders",
+				NumPartitions: 3,
+			},
+		},
+	})
+	if getError(resp) != "" {
+		t.Fatalf("CreateTopic: %s", getError(resp))
+	}
+
+	// List partitions — should report 3
+	resp = sendRecv(t, conn, &pb.Request{
+		Body: &pb.Request_ListPartitions{
+			ListPartitions: &pb.ListPartitionsRequest{Topic: "orders"},
+		},
+	})
+	lp := resp.GetListPartitions()
+	if lp == nil {
+		t.Fatal("expected ListPartitionsResponse")
+	}
+	if lp.NumPartitions != 3 {
+		t.Errorf("expected 3 partitions, got %d", lp.NumPartitions)
+	}
+}
+
+func TestServerListPartitionsTopicNotFound(t *testing.T) {
+	_, _, addr := startTestServer(t)
+	conn := dial(t, addr)
+
+	resp := sendRecv(t, conn, &pb.Request{
+		Body: &pb.Request_ListPartitions{
+			ListPartitions: &pb.ListPartitionsRequest{Topic: "missing"},
+		},
+	})
+	if getError(resp) == "" {
+		t.Fatal("expected error for unknown topic")
+	}
+}
+
 func TestServerProduceConsume(t *testing.T) {
 	_, _, addr := startTestServer(t)
 	conn := dial(t, addr)
diff --git a/proto/server.pb.go b/proto/server.pb.go
index a094400..d04c9c6 100644
--- a/proto/server.pb.go
+++ b/proto/server.pb.go
@@ -2,7 +2,7 @@
 // versions:
 // 	protoc-gen-go v1.36.12
 // 	protoc        v5.28.3
-// source: proto/server.proto
+// source: server.proto
 
 package proto
 
@@ -24,11 +24,12 @@ const (
 type MessageType int32
 
 const (
-	MessageType_MESSAGE_TYPE_UNKNOWN      MessageType = 0
-	MessageType_MESSAGE_TYPE_PRODUCE      MessageType = 1
-	MessageType_MESSAGE_TYPE_CONSUME      MessageType = 2
-	MessageType_MESSAGE_TYPE_CREATE_TOPIC MessageType = 3
-	MessageType_MESSAGE_TYPE_LIST_TOPICS  MessageType = 4
+	MessageType_MESSAGE_TYPE_UNKNOWN         MessageType = 0
+	MessageType_MESSAGE_TYPE_PRODUCE         MessageType = 1
+	MessageType_MESSAGE_TYPE_CONSUME         MessageType = 2
+	MessageType_MESSAGE_TYPE_CREATE_TOPIC    MessageType = 3
+	MessageType_MESSAGE_TYPE_LIST_TOPICS     MessageType = 4
+	MessageType_MESSAGE_TYPE_LIST_PARTITIONS MessageType = 5
 )
 
 // Enum value maps for MessageType.
@@ -39,13 +40,15 @@ var (
 		2: "MESSAGE_TYPE_CONSUME",
 		3: "MESSAGE_TYPE_CREATE_TOPIC",
 		4: "MESSAGE_TYPE_LIST_TOPICS",
+		5: "MESSAGE_TYPE_LIST_PARTITIONS",
 	}
 	MessageType_value = map[string]int32{
-		"MESSAGE_TYPE_UNKNOWN":      0,
-		"MESSAGE_TYPE_PRODUCE":      1,
-		"MESSAGE_TYPE_CONSUME":      2,
-		"MESSAGE_TYPE_CREATE_TOPIC": 3,
-		"MESSAGE_TYPE_LIST_TOPICS":  4,
+		"MESSAGE_TYPE_UNKNOWN":         0,
+		"MESSAGE_TYPE_PRODUCE":         1,
+		"MESSAGE_TYPE_CONSUME":         2,
+		"MESSAGE_TYPE_CREATE_TOPIC":    3,
+		"MESSAGE_TYPE_LIST_TOPICS":     4,
+		"MESSAGE_TYPE_LIST_PARTITIONS": 5,
 	}
 )
 
@@ -60,11 +63,11 @@ func (x MessageType) String() string {
 }
 
 func (MessageType) Descriptor() protoreflect.EnumDescriptor {
-	return file_proto_server_proto_enumTypes[0].Descriptor()
+	return file_server_proto_enumTypes[0].Descriptor()
 }
 
 func (MessageType) Type() protoreflect.EnumType {
-	return &file_proto_server_proto_enumTypes[0]
+	return &file_server_proto_enumTypes[0]
 }
 
 func (x MessageType) Number() protoreflect.EnumNumber {
@@ -73,7 +76,7 @@ func (x MessageType) Number() protoreflect.EnumNumber {
 
 // Deprecated: Use MessageType.Descriptor instead.
 func (MessageType) EnumDescriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{0}
+	return file_server_proto_rawDescGZIP(), []int{0}
 }
 
 // Request envelope — every frame starts with this.
@@ -87,6 +90,7 @@ type Request struct {
 	//	*Request_Consume
 	//	*Request_CreateTopic
 	//	*Request_ListTopics
+	//	*Request_ListPartitions
 	Body          isRequest_Body `protobuf_oneof:"body"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
@@ -94,7 +98,7 @@ type Request struct {
 
 func (x *Request) Reset() {
 	*x = Request{}
-	mi := &file_proto_server_proto_msgTypes[0]
+	mi := &file_server_proto_msgTypes[0]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -106,7 +110,7 @@ func (x *Request) String() string {
 func (*Request) ProtoMessage() {}
 
 func (x *Request) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[0]
+	mi := &file_server_proto_msgTypes[0]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -119,7 +123,7 @@ func (x *Request) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Request.ProtoReflect.Descriptor instead.
 func (*Request) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{0}
+	return file_server_proto_rawDescGZIP(), []int{0}
 }
 
 func (x *Request) GetType() MessageType {
@@ -172,6 +176,15 @@ func (x *Request) GetListTopics() *ListTopicsRequest {
 	return nil
 }
 
+func (x *Request) GetListPartitions() *ListPartitionsRequest {
+	if x != nil {
+		if x, ok := x.Body.(*Request_ListPartitions); ok {
+			return x.ListPartitions
+		}
+	}
+	return nil
+}
+
 type isRequest_Body interface {
 	isRequest_Body()
 }
@@ -192,6 +205,10 @@ type Request_ListTopics struct {
 	ListTopics *ListTopicsRequest `protobuf:"bytes,13,opt,name=list_topics,json=listTopics,proto3,oneof"`
 }
 
+type Request_ListPartitions struct {
+	ListPartitions *ListPartitionsRequest `protobuf:"bytes,14,opt,name=list_partitions,json=listPartitions,proto3,oneof"`
+}
+
 func (*Request_Produce) isRequest_Body() {}
 
 func (*Request_Consume) isRequest_Body() {}
@@ -200,6 +217,8 @@ func (*Request_CreateTopic) isRequest_Body() {}
 
 func (*Request_ListTopics) isRequest_Body() {}
 
+func (*Request_ListPartitions) isRequest_Body() {}
+
 // Response envelope — every frame starts with this.
 type Response struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
@@ -209,6 +228,7 @@ type Response struct {
 	//	*Response_Consume
 	//	*Response_CreateTopic
 	//	*Response_ListTopics
+	//	*Response_ListPartitions
 	//	*Response_Error
 	Body          isResponse_Body `protobuf_oneof:"body"`
 	unknownFields protoimpl.UnknownFields
@@ -217,7 +237,7 @@ type Response struct {
 
 func (x *Response) Reset() {
 	*x = Response{}
-	mi := &file_proto_server_proto_msgTypes[1]
+	mi := &file_server_proto_msgTypes[1]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -229,7 +249,7 @@ func (x *Response) String() string {
 func (*Response) ProtoMessage() {}
 
 func (x *Response) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[1]
+	mi := &file_server_proto_msgTypes[1]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -242,7 +262,7 @@ func (x *Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Response.ProtoReflect.Descriptor instead.
 func (*Response) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{1}
+	return file_server_proto_rawDescGZIP(), []int{1}
 }
 
 func (x *Response) GetBody() isResponse_Body {
@@ -288,6 +308,15 @@ func (x *Response) GetListTopics() *ListTopicsResponse {
 	return nil
 }
 
+func (x *Response) GetListPartitions() *ListPartitionsResponse {
+	if x != nil {
+		if x, ok := x.Body.(*Response_ListPartitions); ok {
+			return x.ListPartitions
+		}
+	}
+	return nil
+}
+
 func (x *Response) GetError() *ErrorResponse {
 	if x != nil {
 		if x, ok := x.Body.(*Response_Error); ok {
@@ -317,6 +346,10 @@ type Response_ListTopics struct {
 	ListTopics *ListTopicsResponse `protobuf:"bytes,13,opt,name=list_topics,json=listTopics,proto3,oneof"`
 }
 
+type Response_ListPartitions struct {
+	ListPartitions *ListPartitionsResponse `protobuf:"bytes,14,opt,name=list_partitions,json=listPartitions,proto3,oneof"`
+}
+
 type Response_Error struct {
 	Error *ErrorResponse `protobuf:"bytes,20,opt,name=error,proto3,oneof"`
 }
@@ -329,6 +362,8 @@ func (*Response_CreateTopic) isResponse_Body() {}
 
 func (*Response_ListTopics) isResponse_Body() {}
 
+func (*Response_ListPartitions) isResponse_Body() {}
+
 func (*Response_Error) isResponse_Body() {}
 
 type ProduceRequest struct {
@@ -342,7 +377,7 @@ type ProduceRequest struct {
 
 func (x *ProduceRequest) Reset() {
 	*x = ProduceRequest{}
-	mi := &file_proto_server_proto_msgTypes[2]
+	mi := &file_server_proto_msgTypes[2]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -354,7 +389,7 @@ func (x *ProduceRequest) String() string {
 func (*ProduceRequest) ProtoMessage() {}
 
 func (x *ProduceRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[2]
+	mi := &file_server_proto_msgTypes[2]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -367,7 +402,7 @@ func (x *ProduceRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ProduceRequest.ProtoReflect.Descriptor instead.
 func (*ProduceRequest) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{2}
+	return file_server_proto_rawDescGZIP(), []int{2}
 }
 
 func (x *ProduceRequest) GetTopic() string {
@@ -401,7 +436,7 @@ type ProduceResponse struct {
 
 func (x *ProduceResponse) Reset() {
 	*x = ProduceResponse{}
-	mi := &file_proto_server_proto_msgTypes[3]
+	mi := &file_server_proto_msgTypes[3]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -413,7 +448,7 @@ func (x *ProduceResponse) String() string {
 func (*ProduceResponse) ProtoMessage() {}
 
 func (x *ProduceResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[3]
+	mi := &file_server_proto_msgTypes[3]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -426,7 +461,7 @@ func (x *ProduceResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ProduceResponse.ProtoReflect.Descriptor instead.
 func (*ProduceResponse) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{3}
+	return file_server_proto_rawDescGZIP(), []int{3}
 }
 
 func (x *ProduceResponse) GetPartition() int32 {
@@ -454,7 +489,7 @@ type ConsumeRequest struct {
 
 func (x *ConsumeRequest) Reset() {
 	*x = ConsumeRequest{}
-	mi := &file_proto_server_proto_msgTypes[4]
+	mi := &file_server_proto_msgTypes[4]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -466,7 +501,7 @@ func (x *ConsumeRequest) String() string {
 func (*ConsumeRequest) ProtoMessage() {}
 
 func (x *ConsumeRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[4]
+	mi := &file_server_proto_msgTypes[4]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -479,7 +514,7 @@ func (x *ConsumeRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ConsumeRequest.ProtoReflect.Descriptor instead.
 func (*ConsumeRequest) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{4}
+	return file_server_proto_rawDescGZIP(), []int{4}
 }
 
 func (x *ConsumeRequest) GetTopic() string {
@@ -515,7 +550,7 @@ type ConsumeResponse struct {
 
 func (x *ConsumeResponse) Reset() {
 	*x = ConsumeResponse{}
-	mi := &file_proto_server_proto_msgTypes[5]
+	mi := &file_server_proto_msgTypes[5]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -527,7 +562,7 @@ func (x *ConsumeResponse) String() string {
 func (*ConsumeResponse) ProtoMessage() {}
 
 func (x *ConsumeResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[5]
+	mi := &file_server_proto_msgTypes[5]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -540,7 +575,7 @@ func (x *ConsumeResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ConsumeResponse.ProtoReflect.Descriptor instead.
 func (*ConsumeResponse) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{5}
+	return file_server_proto_rawDescGZIP(), []int{5}
 }
 
 func (x *ConsumeResponse) GetOffset() uint64 {
@@ -581,7 +616,7 @@ type CreateTopicRequest struct {
 
 func (x *CreateTopicRequest) Reset() {
 	*x = CreateTopicRequest{}
-	mi := &file_proto_server_proto_msgTypes[6]
+	mi := &file_server_proto_msgTypes[6]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -593,7 +628,7 @@ func (x *CreateTopicRequest) String() string {
 func (*CreateTopicRequest) ProtoMessage() {}
 
 func (x *CreateTopicRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[6]
+	mi := &file_server_proto_msgTypes[6]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -606,7 +641,7 @@ func (x *CreateTopicRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CreateTopicRequest.ProtoReflect.Descriptor instead.
 func (*CreateTopicRequest) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{6}
+	return file_server_proto_rawDescGZIP(), []int{6}
 }
 
 func (x *CreateTopicRequest) GetName() string {
@@ -631,7 +666,7 @@ type CreateTopicResponse struct {
 
 func (x *CreateTopicResponse) Reset() {
 	*x = CreateTopicResponse{}
-	mi := &file_proto_server_proto_msgTypes[7]
+	mi := &file_server_proto_msgTypes[7]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -643,7 +678,7 @@ func (x *CreateTopicResponse) String() string {
 func (*CreateTopicResponse) ProtoMessage() {}
 
 func (x *CreateTopicResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[7]
+	mi := &file_server_proto_msgTypes[7]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -656,7 +691,7 @@ func (x *CreateTopicResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CreateTopicResponse.ProtoReflect.Descriptor instead.
 func (*CreateTopicResponse) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{7}
+	return file_server_proto_rawDescGZIP(), []int{7}
 }
 
 type ListTopicsRequest struct {
@@ -667,7 +702,7 @@ type ListTopicsRequest struct {
 
 func (x *ListTopicsRequest) Reset() {
 	*x = ListTopicsRequest{}
-	mi := &file_proto_server_proto_msgTypes[8]
+	mi := &file_server_proto_msgTypes[8]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -679,7 +714,7 @@ func (x *ListTopicsRequest) String() string {
 func (*ListTopicsRequest) ProtoMessage() {}
 
 func (x *ListTopicsRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[8]
+	mi := &file_server_proto_msgTypes[8]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -692,7 +727,7 @@ func (x *ListTopicsRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ListTopicsRequest.ProtoReflect.Descriptor instead.
 func (*ListTopicsRequest) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{8}
+	return file_server_proto_rawDescGZIP(), []int{8}
 }
 
 type ListTopicsResponse struct {
@@ -704,7 +739,7 @@ type ListTopicsResponse struct {
 
 func (x *ListTopicsResponse) Reset() {
 	*x = ListTopicsResponse{}
-	mi := &file_proto_server_proto_msgTypes[9]
+	mi := &file_server_proto_msgTypes[9]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -716,7 +751,7 @@ func (x *ListTopicsResponse) String() string {
 func (*ListTopicsResponse) ProtoMessage() {}
 
 func (x *ListTopicsResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[9]
+	mi := &file_server_proto_msgTypes[9]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -729,7 +764,7 @@ func (x *ListTopicsResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ListTopicsResponse.ProtoReflect.Descriptor instead.
 func (*ListTopicsResponse) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{9}
+	return file_server_proto_rawDescGZIP(), []int{9}
 }
 
 func (x *ListTopicsResponse) GetTopics() []string {
@@ -739,6 +774,96 @@ func (x *ListTopicsResponse) GetTopics() []string {
 	return nil
 }
 
+// Returns the partition count for a topic. Offsets are per-partition, so
+// consumers need this to discover what they can consume.
+type ListPartitionsRequest struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Topic         string                 `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *ListPartitionsRequest) Reset() {
+	*x = ListPartitionsRequest{}
+	mi := &file_server_proto_msgTypes[10]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *ListPartitionsRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ListPartitionsRequest) ProtoMessage() {}
+
+func (x *ListPartitionsRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_server_proto_msgTypes[10]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ListPartitionsRequest.ProtoReflect.Descriptor instead.
+func (*ListPartitionsRequest) Descriptor() ([]byte, []int) {
+	return file_server_proto_rawDescGZIP(), []int{10}
+}
+
+func (x *ListPartitionsRequest) GetTopic() string {
+	if x != nil {
+		return x.Topic
+	}
+	return ""
+}
+
+type ListPartitionsResponse struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	NumPartitions int32                  `protobuf:"varint,1,opt,name=num_partitions,json=numPartitions,proto3" json:"num_partitions,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *ListPartitionsResponse) Reset() {
+	*x = ListPartitionsResponse{}
+	mi := &file_server_proto_msgTypes[11]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *ListPartitionsResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ListPartitionsResponse) ProtoMessage() {}
+
+func (x *ListPartitionsResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_server_proto_msgTypes[11]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ListPartitionsResponse.ProtoReflect.Descriptor instead.
+func (*ListPartitionsResponse) Descriptor() ([]byte, []int) {
+	return file_server_proto_rawDescGZIP(), []int{11}
+}
+
+func (x *ListPartitionsResponse) GetNumPartitions() int32 {
+	if x != nil {
+		return x.NumPartitions
+	}
+	return 0
+}
+
 type ErrorResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Message       string                 `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
@@ -748,7 +873,7 @@ type ErrorResponse struct {
 
 func (x *ErrorResponse) Reset() {
 	*x = ErrorResponse{}
-	mi := &file_proto_server_proto_msgTypes[10]
+	mi := &file_server_proto_msgTypes[12]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -760,7 +885,7 @@ func (x *ErrorResponse) String() string {
 func (*ErrorResponse) ProtoMessage() {}
 
 func (x *ErrorResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_proto_server_proto_msgTypes[10]
+	mi := &file_server_proto_msgTypes[12]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -773,7 +898,7 @@ func (x *ErrorResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ErrorResponse.ProtoReflect.Descriptor instead.
 func (*ErrorResponse) Descriptor() ([]byte, []int) {
-	return file_proto_server_proto_rawDescGZIP(), []int{10}
+	return file_server_proto_rawDescGZIP(), []int{12}
 }
 
 func (x *ErrorResponse) GetMessage() string {
@@ -783,11 +908,11 @@ func (x *ErrorResponse) GetMessage() string {
 	return ""
 }
 
-var File_proto_server_proto protoreflect.FileDescriptor
+var File_server_proto protoreflect.FileDescriptor
 
-const file_proto_server_proto_rawDesc = "" +
+const file_server_proto_rawDesc = "" +
 	"\n" +
-	"\x12proto/server.proto\x12\x06server\"\xa1\x02\n" +
+	"\fserver.proto\x12\x06server\"\xeb\x02\n" +
 	"\aRequest\x12'\n" +
 	"\x04type\x18\x01 \x01(\x0e2\x13.server.MessageTypeR\x04type\x122\n" +
 	"\aproduce\x18\n" +
@@ -795,15 +920,17 @@ const file_proto_server_proto_rawDesc = "" +
 	"\aconsume\x18\v \x01(\v2\x16.server.ConsumeRequestH\x00R\aconsume\x12?\n" +
 	"\fcreate_topic\x18\f \x01(\v2\x1a.server.CreateTopicRequestH\x00R\vcreateTopic\x12<\n" +
 	"\vlist_topics\x18\r \x01(\v2\x19.server.ListTopicsRequestH\x00R\n" +
-	"listTopicsB\x06\n" +
-	"\x04body\"\xac\x02\n" +
+	"listTopics\x12H\n" +
+	"\x0flist_partitions\x18\x0e \x01(\v2\x1d.server.ListPartitionsRequestH\x00R\x0elistPartitionsB\x06\n" +
+	"\x04body\"\xf7\x02\n" +
 	"\bResponse\x123\n" +
 	"\aproduce\x18\n" +
 	" \x01(\v2\x17.server.ProduceResponseH\x00R\aproduce\x123\n" +
 	"\aconsume\x18\v \x01(\v2\x17.server.ConsumeResponseH\x00R\aconsume\x12@\n" +
 	"\fcreate_topic\x18\f \x01(\v2\x1b.server.CreateTopicResponseH\x00R\vcreateTopic\x12=\n" +
 	"\vlist_topics\x18\r \x01(\v2\x1a.server.ListTopicsResponseH\x00R\n" +
-	"listTopics\x12-\n" +
+	"listTopics\x12I\n" +
+	"\x0flist_partitions\x18\x0e \x01(\v2\x1e.server.ListPartitionsResponseH\x00R\x0elistPartitions\x12-\n" +
 	"\x05error\x18\x14 \x01(\v2\x15.server.ErrorResponseH\x00R\x05errorB\x06\n" +
 	"\x04body\"N\n" +
 	"\x0eProduceRequest\x12\x14\n" +
@@ -828,96 +955,107 @@ const file_proto_server_proto_rawDesc = "" +
 	"\x13CreateTopicResponse\"\x13\n" +
 	"\x11ListTopicsRequest\",\n" +
 	"\x12ListTopicsResponse\x12\x16\n" +
-	"\x06topics\x18\x01 \x03(\tR\x06topics\")\n" +
+	"\x06topics\x18\x01 \x03(\tR\x06topics\"-\n" +
+	"\x15ListPartitionsRequest\x12\x14\n" +
+	"\x05topic\x18\x01 \x01(\tR\x05topic\"?\n" +
+	"\x16ListPartitionsResponse\x12%\n" +
+	"\x0enum_partitions\x18\x01 \x01(\x05R\rnumPartitions\")\n" +
 	"\rErrorResponse\x12\x18\n" +
-	"\amessage\x18\x01 \x01(\tR\amessage*\x98\x01\n" +
+	"\amessage\x18\x01 \x01(\tR\amessage*\xba\x01\n" +
 	"\vMessageType\x12\x18\n" +
 	"\x14MESSAGE_TYPE_UNKNOWN\x10\x00\x12\x18\n" +
 	"\x14MESSAGE_TYPE_PRODUCE\x10\x01\x12\x18\n" +
 	"\x14MESSAGE_TYPE_CONSUME\x10\x02\x12\x1d\n" +
 	"\x19MESSAGE_TYPE_CREATE_TOPIC\x10\x03\x12\x1c\n" +
-	"\x18MESSAGE_TYPE_LIST_TOPICS\x10\x04B\x15Z\x13message-queue/protob\x06proto3"
+	"\x18MESSAGE_TYPE_LIST_TOPICS\x10\x04\x12 \n" +
+	"\x1cMESSAGE_TYPE_LIST_PARTITIONS\x10\x05B(Z&github.com/ga11221/message-queue/protob\x06proto3"
 
 var (
-	file_proto_server_proto_rawDescOnce sync.Once
-	file_proto_server_proto_rawDescData []byte
+	file_server_proto_rawDescOnce sync.Once
+	file_server_proto_rawDescData []byte
 )
 
-func file_proto_server_proto_rawDescGZIP() []byte {
-	file_proto_server_proto_rawDescOnce.Do(func() {
-		file_proto_server_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_proto_server_proto_rawDesc), len(file_proto_server_proto_rawDesc)))
+func file_server_proto_rawDescGZIP() []byte {
+	file_server_proto_rawDescOnce.Do(func() {
+		file_server_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_server_proto_rawDesc), len(file_server_proto_rawDesc)))
 	})
-	return file_proto_server_proto_rawDescData
-}
-
-var file_proto_server_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
-var file_proto_server_proto_msgTypes = make([]protoimpl.MessageInfo, 11)
-var file_proto_server_proto_goTypes = []any{
-	(MessageType)(0),            // 0: server.MessageType
-	(*Request)(nil),             // 1: server.Request
-	(*Response)(nil),            // 2: server.Response
-	(*ProduceRequest)(nil),      // 3: server.ProduceRequest
-	(*ProduceResponse)(nil),     // 4: server.ProduceResponse
-	(*ConsumeRequest)(nil),      // 5: server.ConsumeRequest
-	(*ConsumeResponse)(nil),     // 6: server.ConsumeResponse
-	(*CreateTopicRequest)(nil),  // 7: server.CreateTopicRequest
-	(*CreateTopicResponse)(nil), // 8: server.CreateTopicResponse
-	(*ListTopicsRequest)(nil),   // 9: server.ListTopicsRequest
-	(*ListTopicsResponse)(nil),  // 10: server.ListTopicsResponse
-	(*ErrorResponse)(nil),       // 11: server.ErrorResponse
-}
-var file_proto_server_proto_depIdxs = []int32{
+	return file_server_proto_rawDescData
+}
+
+var file_server_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_server_proto_msgTypes = make([]protoimpl.MessageInfo, 13)
+var file_server_proto_goTypes = []any{
+	(MessageType)(0),               // 0: server.MessageType
+	(*Request)(nil),                // 1: server.Request
+	(*Response)(nil),               // 2: server.Response
+	(*ProduceRequest)(nil),         // 3: server.ProduceRequest
+	(*ProduceResponse)(nil),        // 4: server.ProduceResponse
+	(*ConsumeRequest)(nil),         // 5: server.ConsumeRequest
+	(*ConsumeResponse)(nil),        // 6: server.ConsumeResponse
+	(*CreateTopicRequest)(nil),     // 7: server.CreateTopicRequest
+	(*CreateTopicResponse)(nil),    // 8: server.CreateTopicResponse
+	(*ListTopicsRequest)(nil),      // 9: server.ListTopicsRequest
+	(*ListTopicsResponse)(nil),     // 10: server.ListTopicsResponse
+	(*ListPartitionsRequest)(nil),  // 11: server.ListPartitionsRequest
+	(*ListPartitionsResponse)(nil), // 12: server.ListPartitionsResponse
+	(*ErrorResponse)(nil),          // 13: server.ErrorResponse
+}
+var file_server_proto_depIdxs = []int32{
 	0,  // 0: server.Request.type:type_name -> server.MessageType
 	3,  // 1: server.Request.produce:type_name -> server.ProduceRequest
 	5,  // 2: server.Request.consume:type_name -> server.ConsumeRequest
 	7,  // 3: server.Request.create_topic:type_name -> server.CreateTopicRequest
 	9,  // 4: server.Request.list_topics:type_name -> server.ListTopicsRequest
-	4,  // 5: server.Response.produce:type_name -> server.ProduceResponse
-	6,  // 6: server.Response.consume:type_name -> server.ConsumeResponse
-	8,  // 7: server.Response.create_topic:type_name -> server.CreateTopicResponse
-	10, // 8: server.Response.list_topics:type_name -> server.ListTopicsResponse
-	11, // 9: server.Response.error:type_name -> server.ErrorResponse
-	10, // [10:10] is the sub-list for method output_type
-	10, // [10:10] is the sub-list for method input_type
-	10, // [10:10] is the sub-list for extension type_name
-	10, // [10:10] is the sub-list for extension extendee
-	0,  // [0:10] is the sub-list for field type_name
-}
-
-func init() { file_proto_server_proto_init() }
-func file_proto_server_proto_init() {
-	if File_proto_server_proto != nil {
+	11, // 5: server.Request.list_partitions:type_name -> server.ListPartitionsRequest
+	4,  // 6: server.Response.produce:type_name -> server.ProduceResponse
+	6,  // 7: server.Response.consume:type_name -> server.ConsumeResponse
+	8,  // 8: server.Response.create_topic:type_name -> server.CreateTopicResponse
+	10, // 9: server.Response.list_topics:type_name -> server.ListTopicsResponse
+	12, // 10: server.Response.list_partitions:type_name -> server.ListPartitionsResponse
+	13, // 11: server.Response.error:type_name -> server.ErrorResponse
+	12, // [12:12] is the sub-list for method output_type
+	12, // [12:12] is the sub-list for method input_type
+	12, // [12:12] is the sub-list for extension type_name
+	12, // [12:12] is the sub-list for extension extendee
+	0,  // [0:12] is the sub-list for field type_name
+}
+
+func init() { file_server_proto_init() }
+func file_server_proto_init() {
+	if File_server_proto != nil {
 		return
 	}
-	file_proto_server_proto_msgTypes[0].OneofWrappers = []any{
+	file_server_proto_msgTypes[0].OneofWrappers = []any{
 		(*Request_Produce)(nil),
 		(*Request_Consume)(nil),
 		(*Request_CreateTopic)(nil),
 		(*Request_ListTopics)(nil),
+		(*Request_ListPartitions)(nil),
 	}
-	file_proto_server_proto_msgTypes[1].OneofWrappers = []any{
+	file_server_proto_msgTypes[1].OneofWrappers = []any{
 		(*Response_Produce)(nil),
 		(*Response_Consume)(nil),
 		(*Response_CreateTopic)(nil),
 		(*Response_ListTopics)(nil),
+		(*Response_ListPartitions)(nil),
 		(*Response_Error)(nil),
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
-			RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_server_proto_rawDesc), len(file_proto_server_proto_rawDesc)),
+			RawDescriptor: unsafe.Slice(unsafe.StringData(file_server_proto_rawDesc), len(file_server_proto_rawDesc)),
 			NumEnums:      1,
-			NumMessages:   11,
+			NumMessages:   13,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
-		GoTypes:           file_proto_server_proto_goTypes,
-		DependencyIndexes: file_proto_server_proto_depIdxs,
-		EnumInfos:         file_proto_server_proto_enumTypes,
-		MessageInfos:      file_proto_server_proto_msgTypes,
+		GoTypes:           file_server_proto_goTypes,
+		DependencyIndexes: file_server_proto_depIdxs,
+		EnumInfos:         file_server_proto_enumTypes,
+		MessageInfos:      file_server_proto_msgTypes,
 	}.Build()
-	File_proto_server_proto = out.File
-	file_proto_server_proto_goTypes = nil
-	file_proto_server_proto_depIdxs = nil
+	File_server_proto = out.File
+	file_server_proto_goTypes = nil
+	file_server_proto_depIdxs = nil
 }
diff --git a/proto/server.proto b/proto/server.proto
index 34d1202..4a06c73 100644
--- a/proto/server.proto
+++ b/proto/server.proto
@@ -2,7 +2,7 @@ syntax = "proto3";
 
 package server;
 
-option go_package = "message-queue/proto";
+option go_package = "github.com/ga11221/message-queue/proto";
 
 // Request envelope — every frame starts with this.
 // The broker reads `type` to determine which request body to decode.
@@ -13,6 +13,7 @@ message Request {
     ConsumeRequest consume = 11;
     CreateTopicRequest create_topic = 12;
     ListTopicsRequest list_topics = 13;
+    ListPartitionsRequest list_partitions = 14;
   }
 }
 
@@ -23,6 +24,7 @@ message Response {
     ConsumeResponse consume = 11;
     CreateTopicResponse create_topic = 12;
     ListTopicsResponse list_topics = 13;
+    ListPartitionsResponse list_partitions = 14;
     ErrorResponse error = 20;
   }
 }
@@ -33,6 +35,7 @@ enum MessageType {
   MESSAGE_TYPE_CONSUME = 2;
   MESSAGE_TYPE_CREATE_TOPIC = 3;
   MESSAGE_TYPE_LIST_TOPICS = 4;
+  MESSAGE_TYPE_LIST_PARTITIONS = 5;
 }
 
 // --- Produce ---
@@ -80,6 +83,18 @@ message ListTopicsResponse {
   repeated string topics = 1;
 }
 
+// --- ListPartitions ---
+
+// Returns the partition count for a topic. Offsets are per-partition, so
+// consumers need this to discover what they can consume.
+message ListPartitionsRequest {
+  string topic = 1;
+}
+
+message ListPartitionsResponse {
+  int32 num_partitions = 1;
+}
+
 // --- Error ---
 
 message ErrorResponse {
diff --git a/scripts/demo.sh b/scripts/demo.sh
new file mode 100644
index 0000000..98072db
--- /dev/null
+++ b/scripts/demo.sh
@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+# demo.sh — end-to-end walkthrough of the message queue.
+#
+# Shows the full lifecycle: container up, topic creation, keyed produce
+# (watch partition assignment), consume from one partition, merged consume
+# across partitions, and on-disk integrity via mqvalidate.
+#
+# Records cleanly with: asciinema rec demo.cast -- ./scripts/demo.sh
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+ADDR=localhost:9092
+TOPIC="demo-events-$(date +%s)"   # unique per run: the volume persists topics between runs
+BIN=$(mktemp -d)/mqclient
+MQ() { "$BIN" "$@" -addr "$ADDR"; }
+
+echo "=== build client ==="
+go build -o "$BIN" ./cmd/mqclient
+
+echo
+echo "=== start server (docker compose, building if needed) ==="
+docker compose up -d --build >/dev/null 2>&1   # quiet: build output drowns the demo
+# Wait for a REAL successful request: a bare TCP connect can succeed against
+# the old container moments before compose recreates it (RST on next use).
+for i in $(seq 1 60); do
+    if MQ topics >/dev/null 2>&1; then break; fi
+    sleep 0.5
+done
+MQ topics >/dev/null   # fail loudly if the server never came up
+
+echo
+echo "=== create topic '$TOPIC' with 3 partitions ==="
+MQ create -topic "$TOPIC" -partitions 3
+MQ partitions -topic "$TOPIC"
+
+echo
+echo "=== produce keyed messages (note partition assignments) ==="
+MQ produce -topic "$TOPIC" -key user-1 -value 'order:42'
+MQ produce -topic "$TOPIC" -key user-2 -value 'order:43'
+MQ produce -topic "$TOPIC" -key user-1 -value 'order:44'
+MQ produce -topic "$TOPIC" -key user-3 -value 'order:45'
+
+echo
+echo "=== consume everything, merged across partitions ==="
+echo "(ordering is per-partition/per-key; merge by timestamp is best-effort)"
+MQ consume -topic "$TOPIC" -all
+
+echo
+echo "=== verify bytes at rest with mqvalidate ==="
+docker cp "$(docker compose ps -q server)":/data /tmp/opencode/demo-data-check >/dev/null 2>&1 || true
+go run ./cmd/mqvalidate /tmp/opencode/demo-data-check | tail -1 && echo "on-disk data: OK"
+
+echo
+echo "=== cleanup ==="
+rm -rf "$(dirname "$BIN")" /tmp/opencode/demo-data-check
+docker compose stop >/dev/null
+echo "demo complete."