diff --git a/README.md b/README.md
index 5102450..60722e1 100644
--- a/README.md
+++ b/README.md
@@ -291,12 +291,39 @@ and after shutdown it is a counted no-op rather than a panic.
 ```go
 s := b.Stats()
 
-// s.Queued        -> queue depth; the value to alert on
-// s.Pending       -> accepted work not yet finished, including in-flight batches
-// s.Rejected      -> refused enqueues
-// s.DroppedErrors -> diagnostics lost because Errors() was not drained
+// Where work currently is. These three are disjoint, so together they show
+// whether a backlog is waiting on the queue, on batching, or on the processor:
+//   s.Queued    -> published, not yet taken by the aggregator (alert on this)
+//   s.BatchHeld -> held by the aggregator: filling, or waiting for a free worker
+//   s.InFlight  -> inside a processor call
+
+// Totals.
+//   s.Pending        -> conservative drain obligation: accepted-or-reserved work,
+//                       including in-flight batches. It counts publishers that have
+//                       reserved but not yet published, so it equals
+//                       accepted-but-unfinished work only once PublishersInGate == 0
+//   s.Accepted       -> successful enqueues
+//   s.Rejected       -> refused enqueues
+//   s.Completed / s.Failed / s.Panicked -> mutually exclusive terminal outcomes
+//   s.BatchesFlushed -> batches emitted. After a terminal drain,
+//                       (Completed+Failed+Panicked)/BatchesFlushed is mean batch size
+//   s.DroppedErrors  -> diagnostics lost because Errors() was not drained
 ```
 
+`BatchesFlushed` is the coalescing signal when tuning `WithBatchInterval`: a mean
+batch size well below `BatchSize` means windows are closing on the timer rather than
+filling, so the interval is costing latency without buying batching. Include every
+terminal outcome in that mean — a failed or panicked batch was still flushed, so
+dividing by `Completed` alone undercounts whenever the processor errors.
+
+A rising `BatchHeld` with `InFlight` at its ceiling means batches are ready but every
+worker is busy — that is the signal to raise `WithConcurrency`, not to shrink the
+window.
+
+The snapshot is eventually consistent, not transactional: each field is read
+independently, so use it to observe where work sits, not as an accounting identity
+while load is in flight.
+
 ### Handling Errors
 
 Whenever the processor function returns an error, the batcher will send the error in the `Errors()` channel:
diff --git a/docs/improvements/plan-perf.md b/docs/improvements/plan-perf.md
index eb96bc8..000ed2d 100644
--- a/docs/improvements/plan-perf.md
+++ b/docs/improvements/plan-perf.md
@@ -1080,8 +1080,10 @@ rather than intuition.
   sealing instead of panicking; `Close()` no longer abandoning the drain at its
   deadline; **`Len()` now counting accepted-but-unfinished work including
   in-flight batches, so it can be non-zero with an empty queue**; `Join`'s
-  clarified quiescence-snapshot contract; the removal of the `rill` and `chann`
-  dependencies; **the module's minimum Go version moving from 1.22.4 to 1.25.0**
+  clarified quiescence-snapshot contract; **`Config()` returning a value snapshot
+  instead of a live mutable pointer**, which removes the ability to reconfigure a
+  running batcher and the data race that came with it; the removal of the `rill` and
+  `chann` dependencies; **the module's minimum Go version moving from 1.22.4 to 1.25.0**
   (required by `golang.org/x/sys` v0.46.0, pulled in when dependencies were
   modernised in 2.1); and any `ProvideBatcherInFX` signature change.
 - **`Config()` must stop exposing live mutable state.** It currently returns the
diff --git a/docs/improvements/thresholds.md b/docs/improvements/thresholds.md
index 61b4332..551d207 100644
--- a/docs/improvements/thresholds.md
+++ b/docs/improvements/thresholds.md
@@ -41,9 +41,12 @@ the blocking signal rather than timings.
 | Recovery wrapper allocations, non-panic | exactly 0                                    | `TestRecoveredPanicAddsNoSteadyStateAllocations`              |
 | Scenario recorder allocations per item  | ≤ 1 total, and must not grow with run length | `TestHarnessRecorderDoesNotAllocatePerItem` (`test/scenario`) |
 | Goroutines per running batcher, `n=1`   | exactly 2                                    | `TestGoroutineBudgetPerRunningBatcher`                        |
+| `Stats()` allocations                   | exactly 0                                    | `TestStatsIsAllocationFree`                                   |
 
-`Stats()` is a fixed set of atomic loads returning a value type, so it has no
-allocation gate of its own; the `Add` gate covers the hot path that matters.
+`Stats()` returns a value type built from atomic loads plus one queue length check
+under the queue's existing mutex. It now carries its own allocation gate, because
+metrics scraping runs continuously in production and generating garbage per scrape
+would be a real cost.
 
 The recorder threshold is not "exactly 0" because `AllocsPerItem` measures the
 whole pipeline, including Batcher's own per-batch allocations, not just the
@@ -104,10 +107,44 @@ zero leaked.
 
 ## Conditional gates (Milestone 4.2 only)
 
-| Gate                                       | Threshold                    |
-| ------------------------------------------ | ---------------------------- |
-| Sparse-window allocated bytes/flush        | > 2 KB/flush to justify work |
-| Allocation regression ceiling, any scenario | ≤ +2% allocated bytes        |
+| Gate                                        | Threshold                    | Measured        |
+| ------------------------------------------- | ---------------------------- | --------------- |
+| Sparse-window allocated bytes/flush         | > 2 KB/flush to justify work | 55,944 B/flush  |
+| Allocation regression ceiling, any scenario | ≤ +2% allocated bytes        | +1.12% (worst)  |
+
+Milestone 4.2 was **triggered and implemented**. Sparse-window waste measured far
+above the justification threshold: at a 1ms window with `BatchSize=1000` the
+aggregator reserved 55,944 B/flush to hold a single item, and 559,944 B/flush at
+`BatchSize=10000`.
+
+Results per workload, adaptive versus the previous full-capacity strategy:
+
+| Workload                | Change  |
+| ----------------------- | ------- |
+| steady sparse           | -97.9%  |
+| small batches           | -93.3%  |
+| full batches (control)  | -0.00%  |
+| alternating sparse/full | +1.12%  |
+| burst after idle        | -72.2%  |
+| bimodal                 | -3.45%  |
+
+The alternating case is the one that killed the rejected EWMA estimator, which
+allocated *more* than doing nothing there. Recent-max keeps it inside the +2%
+budget.
+
+These figures are **measured evidence, not an enforced gate.** `BenchmarkCapacity*`
+reports allocated bytes for each workload but does not compare them against the +2%
+limit, and `capacity_test.go` asserts estimator behaviour (bounds, adaptation,
+decay) rather than allocation deltas. Re-check the numbers with:
+
+```sh
+go test -run='^$' -bench='BenchmarkCapacity' -benchmem -count=1 ./pkg/batcher
+```
+
+Turning this into a gate needs a stored allocation baseline for the reference
+runner, which does not exist yet for the same reason the throughput gate is still
+advisory. Until then, a regression here is caught by reading the benchmark output,
+not by CI.
 
 The second gate must hold for alternating sparse/full, burst-after-idle, and
 bimodal workloads, not just the sparse case the optimisation targets. An
diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go
index 4c04cc7..b32e11e 100644
--- a/pkg/batcher/batcher.go
+++ b/pkg/batcher/batcher.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"runtime/debug"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -41,6 +42,17 @@ type Config[T any] struct {
 type Batcher[T any] struct {
 	config *Config[T]
 
+	// configFrozen becomes true before any processing goroutine starts. Options
+	// are construction-time configuration: applying one to an existing batcher is
+	// a no-op, so a caller cannot race the aggregator or processor by mutating
+	// BatchSize, BatchInterval or ProcessorFunc after New returns.
+	configFrozen atomic.Bool
+
+	// runtime is the immutable snapshot the processing goroutines use. It is taken
+	// in New, before any goroutine exists, so no worker ever reads the Config
+	// struct a caller may still hold a reference to.
+	runtime runtimeConfig[T]
+
 	gate     *admissionGate
 	counters counters
 
@@ -69,6 +81,19 @@ type Batcher[T any] struct {
 	shutdownDone chan struct{}
 }
 
+// runtimeConfig is the frozen view of Config that processing goroutines read.
+//
+// Config is a plain struct and callers can hold a reference to one, so reading it
+// from a worker is a data race by construction. Copying the fields the pipeline
+// needs, once, before any goroutine starts, removes that whole class rather than
+// relying on every read site to remember to use a local.
+type runtimeConfig[T any] struct {
+	batchSize     int
+	batchInterval time.Duration
+	workers       int
+	processor     Processor[T]
+}
+
 // New creates a new Batcher with the given options.
 func New[T any](options ...Option[T]) *Batcher[T] {
 	b := &Batcher[T]{
@@ -93,6 +118,30 @@ func New[T any](options ...Option[T]) *Batcher[T] {
 
 	b.validateConfig()
 
+	// No option may change configuration after this point, including a batcher
+	// constructed with WithSkipAutoStart. SkipAutoStart controls lifecycle, not
+	// whether the configuration is still mutable.
+	b.configFrozen.Store(true)
+
+	b.runtime = runtimeConfig[T]{
+		batchSize:     b.config.BatchSize,
+		batchInterval: b.config.BatchInterval,
+		workers:       b.config.Concurrency,
+		processor:     b.config.ProcessorFunc,
+	}
+
+	if b.runtime.batchSize < 1 {
+		b.runtime.batchSize = DefaultBatchSize
+	}
+
+	if b.runtime.workers < 1 {
+		b.runtime.workers = 1
+	}
+
+	if b.runtime.processor == nil {
+		b.runtime.processor = NoOpProcessor[T]
+	}
+
 	b.input = newQueue[T](b.config.MaxQueueSize)
 	b.errorsChan = make(chan error, b.config.ErrorBufferSize)
 
@@ -138,9 +187,17 @@ func (b *Batcher[T]) Start() {
 	})
 }
 
-// Config returns the batcher's configuration.
-func (b *Batcher[T]) Config() *Config[T] {
-	return b.config
+// Config returns a copy of the batcher's configuration.
+//
+// It is a snapshot, not a handle. Returning the live pointer let a caller mutate a
+// running batcher's batch size, interval or processor from another goroutine, which
+// is a data race against the aggregation loop and could change batching semantics
+// mid-batch. Options are meant to be applied at construction; mutating them
+// afterwards was never coherent, so this closes the hole rather than documenting it.
+//
+// Callers that need to change configuration should construct a new Batcher.
+func (b *Batcher[T]) Config() Config[T] {
+	return *b.config
 }
 
 // Add enqueues an item. It is the compatibility fast path and returns no error.
@@ -223,11 +280,13 @@ func (b *Batcher[T]) Stats() Stats {
 		IntakePending:    b.counters.intakePending.Load(),
 		PublishersInGate: b.gate.inGate(),
 		Queued:           int64(b.input.length()),
+		BatchHeld:        b.counters.batchHeld.Load(),
 		InFlight:         b.counters.inFlight.Load(),
 		Accepted:         b.counters.accepted.Load(),
 		Completed:        b.counters.completed.Load(),
 		Failed:           b.counters.failed.Load(),
 		Panicked:         b.counters.panicked.Load(),
+		BatchesFlushed:   b.counters.batchesFlushed.Load(),
 		Rejected:         b.counters.rejected.Load(),
 		DroppedErrors:    b.counters.droppedErrors.Load(),
 	}
@@ -294,12 +353,18 @@ func (b *Batcher[T]) Errors() <-chan error {
 func (b *Batcher[T]) run() {
 	defer close(b.stopped)
 
-	batches := make(chan []T)
+	// Use the immutable runtime snapshot taken by New before any goroutine starts.
+	// In particular, ProcessorFunc is passed to workers rather than dereferenced
+	// from Config per batch. This keeps every processing read off the mutable Config
+	// struct, so a caller with a stale reference cannot race the pipeline.
+	var (
+		batchSize     = b.runtime.batchSize
+		batchInterval = b.runtime.batchInterval
+		workers       = b.runtime.workers
+		processor     = b.runtime.processor
+	)
 
-	workers := b.config.Concurrency
-	if workers < 1 {
-		workers = 1
-	}
+	batches := make(chan []T)
 
 	var processing sync.WaitGroup
 
@@ -310,7 +375,7 @@ func (b *Batcher[T]) run() {
 			defer processing.Done()
 
 			for items := range batches {
-				b.process(items)
+				b.process(processor, items)
 			}
 		}()
 	}
@@ -324,9 +389,10 @@ func (b *Batcher[T]) run() {
 	}()
 
 	var (
-		batch  []T
-		timer  *time.Timer
-		timerC <-chan time.Time
+		batch      []T
+		timer      *time.Timer
+		timerC     <-chan time.Time
+		capacities = newCapacityEstimator(batchSize)
 	)
 
 	stopTimer := func() {
@@ -356,17 +422,28 @@ func (b *Batcher[T]) run() {
 
 		stopTimer()
 
+		// The send blocks until a worker is free. Until it returns, the batch is
+		// still aggregator-held, which is the state BatchHeld exists to expose:
+		// releasing the counter before the send would hide a batch waiting on
+		// saturated workers.
 		batches <- items
+
+		b.counters.dispatched(len(items))
+		capacities.observe(len(items))
 	}
 
 	take := func(item T) {
 		if len(batch) == 0 {
-			batch = make([]T, 0, b.config.BatchSize)
+			// The capacity estimator is owned by this aggregation goroutine. It
+			// reduces timer-flush allocation pressure without pooling or reusing a
+			// slice after the processor receives it, so callers may still retain the
+			// batch slice exactly as before.
+			batch = make([]T, 0, capacities.capacity())
 
 			if timer == nil {
-				timer = time.NewTimer(b.config.BatchInterval)
+				timer = time.NewTimer(batchInterval)
 			} else {
-				timer.Reset(b.config.BatchInterval)
+				timer.Reset(batchInterval)
 			}
 
 			timerC = timer.C
@@ -374,22 +451,50 @@ func (b *Batcher[T]) run() {
 
 		batch = append(batch, item)
 
-		if len(batch) >= b.config.BatchSize {
+		if len(batch) >= batchSize {
 			flush()
 		}
 	}
 
 	// drainReady empties whatever is currently queued. The aggregator drains
 	// greedily rather than one item per wakeup, so a burst costs one signal.
+	//
+	// It transfers a run of items under a single queue lock rather than locking per
+	// item. The aggregator is the only consumer and keeps no invariant between items,
+	// so a block transfer is indistinguishable from N pops to it, while per-item
+	// locking was what serialised producers: profiling attributed 86% of mutex delay
+	// and 61% of CPU to the push path contending with this loop.
+	//
+	// Each transfer takes only what the batch being built can still accept. That
+	// bounds how long the queue lock is held, and it keeps the capacity contract
+	// exact: transferred items move straight into batch, so no additional
+	// batch-sized buffer of accepted work exists outside the queue. Draining a full
+	// batchSize regardless of the batch's remaining space added exactly that third
+	// buffer and pushed accepted work past N + 2*BatchSize + gate.
+	//
+	// drained is reused across wakeups to keep steady-state draining allocation-free.
+	// It is deliberately separate from batch: batch is handed to a worker and may be
+	// retained by the processor, whereas drained never leaves this goroutine.
+	drained := make([]T, 0, batchSize)
+
 	drainReady := func() {
 		for {
-			item, ok := b.input.pop()
-			if !ok {
+			room := batchSize - len(batch)
+			if room <= 0 {
+				// take flushes at batchSize, so this only happens if a flush could not
+				// complete. Fall back to one item and let take drive the flush.
+				room = 1
+			}
+
+			drained = b.input.popBatch(drained[:0], room)
+			if len(drained) == 0 {
 				return
 			}
 
-			b.counters.received(1)
-			take(item)
+			for _, item := range drained {
+				b.counters.received(1)
+				take(item)
+			}
 		}
 	}
 
@@ -429,7 +534,7 @@ func (b *Batcher[T]) run() {
 // The ordering is fixed and load-bearing: whatever happens, exactly one terminal
 // category is counted and the drain obligation is released exactly once. Leaving it
 // unreleased would inflate Pending permanently and hang shutdown forever.
-func (b *Batcher[T]) process(items []T) {
+func (b *Batcher[T]) process(processor Processor[T], items []T) {
 	b.counters.inFlight.Add(int64(len(items)))
 
 	outcome := outcomeCompleted
@@ -454,7 +559,7 @@ func (b *Batcher[T]) process(items []T) {
 		})
 	}()
 
-	if err := b.config.ProcessorFunc(items); err != nil {
+	if err := processor(items); err != nil {
 		outcome = outcomeFailed
 
 		b.publishError(err)
diff --git a/pkg/batcher/capacity.go b/pkg/batcher/capacity.go
new file mode 100644
index 0000000..5a68aee
--- /dev/null
+++ b/pkg/batcher/capacity.go
@@ -0,0 +1,165 @@
+package batcher
+
+// capacityEstimator chooses the initial capacity for the next batch slice.
+//
+// The aggregator allocates one slice per batch. Reserving BatchSize every time is
+// correct but wasteful whenever batches close on the timer rather than on size: a
+// 1ms window with BatchSize=1000 was measured reserving ~56KB per flush to hold a
+// single item, and ~560KB at BatchSize=10000.
+//
+// The estimator tracks the maximum batch size seen in a recent window and rounds it
+// up to a power of two, so capacity follows observed demand instead of the
+// configured ceiling.
+//
+// Why recent-max rather than a mean:
+//
+// An EWMA of the mean was evaluated during planning and rejected. On traffic
+// alternating between tiny and full batches it allocated *more* than doing nothing,
+// because the mean sits between the two modes: every large batch then grows from a
+// too-small start, and each growth reallocates and copies. Tracking the maximum
+// means a workload containing large batches keeps capacity sized for them, so
+// alternating traffic behaves like the full-capacity strategy rather than worse.
+//
+// Concurrency: this is owned exclusively by the aggregator goroutine and updated
+// once per flush. It deliberately has no locks and no atomics; nothing else may
+// touch it.
+type capacityEstimator struct {
+	// batchSize is the configured ceiling. Capacity never exceeds it, so a
+	// processor that retains its slice never holds more than today's worst case.
+	batchSize int
+
+	// ceiling is the current power-of-two capacity, derived from recentMax. It
+	// starts at batchSize for the first batch so a full-batch workload never pays
+	// append growth before the estimator has any evidence. The first observed
+	// sparse batch then adapts it downward.
+	ceiling int
+
+	// initialized records whether the first observed batch has set a data-driven
+	// ceiling. It distinguishes the initial full-capacity safety value from a
+	// ceiling reached because recent demand genuinely saturated BatchSize.
+	initialized bool
+
+	// recentMax is the largest batch observed in the current evaluation window.
+	recentMax int
+
+	// seen counts observations in the current window.
+	seen int
+}
+
+const (
+	// capacityFloor keeps very sparse traffic from reallocating on a batch that
+	// happens to be slightly larger than the last. 16 items is small enough that
+	// the floor itself is not meaningful waste.
+	capacityFloor = 16
+
+	// capacityWindow is how many flushes are observed before the ceiling is
+	// re-derived. Long enough that a single small batch cannot collapse capacity
+	// for a busy batcher, short enough that a batcher which quiets down releases
+	// its oversized reservation promptly.
+	capacityWindow = 32
+
+	// capacityHeadroom scales the observed maximum before rounding, so a workload
+	// sitting just above a power of two does not reallocate on every batch.
+	capacityHeadroom = 5
+	capacityDivisor  = 4 // headroom/divisor == 1.25x
+)
+
+func newCapacityEstimator(batchSize int) *capacityEstimator {
+	if batchSize < 1 {
+		batchSize = 1
+	}
+
+	return &capacityEstimator{
+		batchSize: batchSize,
+		// Start pessimistic: reserve the configured ceiling until there is evidence
+		// that batches are smaller. Starting small instead made a size-triggered
+		// workload pay repeated append growth on its very first batch, measured as a
+		// 2.2% allocation regression against the full-capacity strategy.
+		ceiling: batchSize,
+	}
+}
+
+// capacity returns the initial capacity to allocate for the next batch.
+func (e *capacityEstimator) capacity() int {
+	return e.ceiling
+}
+
+// observe records the size of a completed batch.
+//
+// The ceiling rises immediately when a batch exceeds it, because under-allocating a
+// large batch costs a reallocation and copy right away. It falls only at window
+// boundaries, so capacity decays gradually rather than oscillating.
+func (e *capacityEstimator) observe(size int) {
+	if size > e.recentMax {
+		e.recentMax = size
+	}
+
+	if !e.initialized {
+		// First evidence: replace the pessimistic starting ceiling outright, so a
+		// sparse workload stops reserving BatchSize after a single batch rather than
+		// waiting for a full evaluation window.
+		e.initialized = true
+		e.ceiling = clampCapacity(roundUpPow2(scaleHeadroom(size)), e.batchSize)
+	} else if size > e.ceiling {
+		e.ceiling = clampCapacity(roundUpPow2(scaleHeadroom(size)), e.batchSize)
+	}
+
+	e.seen++
+
+	if e.seen < capacityWindow {
+		return
+	}
+
+	e.ceiling = clampCapacity(roundUpPow2(scaleHeadroom(e.recentMax)), e.batchSize)
+	e.recentMax = 0
+	e.seen = 0
+}
+
+// scaleHeadroom applies the headroom factor with integer arithmetic.
+func scaleHeadroom(n int) int {
+	if n <= 0 {
+		return 0
+	}
+
+	return n * capacityHeadroom / capacityDivisor
+}
+
+// roundUpPow2 returns the smallest power of two at or above n.
+func roundUpPow2(n int) int {
+	if n <= capacityFloor {
+		return capacityFloor
+	}
+
+	p := capacityFloor
+	for p < n {
+		next := p << 1
+		// Stop before overflow; the caller clamps to batchSize anyway.
+		if next <= p {
+			return p
+		}
+
+		p = next
+	}
+
+	return p
+}
+
+// clampCapacity keeps capacity within [capacityFloor, batchSize].
+//
+// The upper clamp is a contract, not an optimisation: it bounds what a processor
+// that retains its batch slice can hold.
+func clampCapacity(n, batchSize int) int {
+	if n > batchSize {
+		return batchSize
+	}
+
+	if n < capacityFloor {
+		if batchSize < capacityFloor {
+			return batchSize
+		}
+
+		return capacityFloor
+	}
+
+	return n
+}
diff --git a/pkg/batcher/capacity_bench_test.go b/pkg/batcher/capacity_bench_test.go
new file mode 100644
index 0000000..85c141c
--- /dev/null
+++ b/pkg/batcher/capacity_bench_test.go
@@ -0,0 +1,112 @@
+package batcher
+
+import "testing"
+
+// Capacity strategy comparison for Milestone 4.2.
+//
+// The aggregator reserves make([]T, 0, BatchSize) for every batch. When a batch
+// closes on the timer holding far fewer items, the difference is pure waste. These
+// benchmarks compare the current strategy against the recent-max estimator across
+// the workloads the plan requires, including the ones that must NOT regress.
+//
+// EWMA-of-mean was rejected during planning because it regressed alternating
+// traffic; recentMaxCapacity is benchmarked here against that same pattern to
+// confirm the replacement does not repeat the mistake.
+
+type benchItem struct {
+	_ [64]byte
+}
+
+const benchBatchSize = 1_000
+
+var benchSink []benchItem
+
+// fullCapacity is today's strategy: always reserve BatchSize.
+func fullCapacity(sizes []int) {
+	for _, n := range sizes {
+		batch := make([]benchItem, 0, benchBatchSize)
+
+		for range n {
+			batch = append(batch, benchItem{})
+		}
+
+		benchSink = batch
+	}
+}
+
+// adaptive uses the recent-max estimator under test.
+func adaptive(sizes []int) {
+	est := newCapacityEstimator(benchBatchSize)
+
+	for _, n := range sizes {
+		batch := make([]benchItem, 0, est.capacity())
+
+		for range n {
+			batch = append(batch, benchItem{})
+		}
+
+		est.observe(len(batch))
+
+		benchSink = batch
+	}
+}
+
+func repeat(pattern []int, times int) []int {
+	out := make([]int, 0, len(pattern)*times)
+	for range times {
+		out = append(out, pattern...)
+	}
+
+	return out
+}
+
+func benchBoth(b *testing.B, sizes []int) {
+	b.Run("full", func(b *testing.B) {
+		b.ReportAllocs()
+
+		for range b.N {
+			fullCapacity(sizes)
+		}
+	})
+
+	b.Run("adaptive", func(b *testing.B) {
+		b.ReportAllocs()
+
+		for range b.N {
+			adaptive(sizes)
+		}
+	})
+}
+
+// BenchmarkCapacitySteadySparse is the target workload: every batch holds one item.
+func BenchmarkCapacitySteadySparse(b *testing.B) {
+	benchBoth(b, repeat([]int{1}, 200))
+}
+
+// BenchmarkCapacitySmallBatches models a moderate rate closing on the timer.
+func BenchmarkCapacitySmallBatches(b *testing.B) {
+	benchBoth(b, repeat([]int{50}, 200))
+}
+
+// BenchmarkCapacityFullBatches is the control: reserved capacity is fully used, so
+// adaptive must not do worse.
+func BenchmarkCapacityFullBatches(b *testing.B) {
+	benchBoth(b, repeat([]int{benchBatchSize}, 100))
+}
+
+// BenchmarkCapacityAlternating is the pattern that killed EWMA-of-mean.
+func BenchmarkCapacityAlternating(b *testing.B) {
+	benchBoth(b, repeat([]int{1, benchBatchSize}, 100))
+}
+
+// BenchmarkCapacityBurstAfterIdle models a long sparse period then a burst.
+func BenchmarkCapacityBurstAfterIdle(b *testing.B) {
+	sizes := append(repeat([]int{1}, 150), repeat([]int{benchBatchSize}, 50)...)
+	benchBoth(b, sizes)
+}
+
+// BenchmarkCapacityBimodal models mostly small batches with regular large ones.
+func BenchmarkCapacityBimodal(b *testing.B) {
+	pattern := []int{20, 20, 20, 20, 20, 20, 20, 20, 20, 800}
+	benchBoth(b, repeat(pattern, 20))
+}
diff --git a/pkg/batcher/capacity_test.go b/pkg/batcher/capacity_test.go
new file mode 100644
index 0000000..a8c11a4
--- /dev/null
+++ b/pkg/batcher/capacity_test.go
@@ -0,0 +1,187 @@
+package batcher
+
+import "testing"
+
+// Capacity estimator unit tests.
+//
+// These pin the properties the aggregator relies on, independently of timing: the
+// bound that protects a retained slice, adaptation in both directions, and the
+// behaviour on the two workloads that made the rejected EWMA estimator worse than
+// doing nothing.
+
+// TestCapacityNeverExceedsBatchSize pins the retention bound. A processor may keep
+// its batch slice, so capacity must never exceed the configured ceiling; otherwise
+// a retained slice could hold more than today's worst case.
+func TestCapacityNeverExceedsBatchSize(t *testing.T) {
+	t.Parallel()
+
+	for _, batchSize := range []int{1, 7, 16, 100, 1_000} {
+		est := newCapacityEstimator(batchSize)
+
+		if got := est.capacity(); got > batchSize {
+			t.Fatalf("batchSize=%d: initial capacity %d exceeds ceiling", batchSize, got)
+		}
+
+		// Feed sizes at and beyond the ceiling, including absurd ones.
+		for _, size := range []int{0, 1, batchSize, batchSize * 4, batchSize * 1000} {
+			est.observe(size)
+
+			if got := est.capacity(); got > batchSize {
+				t.Fatalf("batchSize=%d: capacity %d exceeds ceiling after observing %d",
+					batchSize, got, size)
+			}
+		}
+	}
+}
+
+// TestCapacityStartsAtBatchSize pins the pessimistic start.
+//
+// Starting small measured as a 2.2% allocation regression on size-triggered
+// workloads, because the first full batch grew repeatedly before any evidence
+// existed. The first batch therefore reserves the ceiling.
+func TestCapacityStartsAtBatchSize(t *testing.T) {
+	t.Parallel()
+
+	est := newCapacityEstimator(1_000)
+
+	if got := est.capacity(); got != 1_000 {
+		t.Fatalf("initial capacity = %d, want 1000 (no evidence yet)", got)
+	}
+}
+
+// TestCapacityAdaptsDownAfterFirstSparseBatch pins that a sparse workload stops
+// reserving the ceiling immediately, rather than waiting a full evaluation window.
+func TestCapacityAdaptsDownAfterFirstSparseBatch(t *testing.T) {
+	t.Parallel()
+
+	est := newCapacityEstimator(1_000)
+
+	est.observe(1)
+
+	got := est.capacity()
+
+	if got >= 1_000 {
+		t.Fatalf("capacity = %d after a 1-item batch, want well below 1000", got)
+	}
+
+	if got < capacityFloor {
+		t.Fatalf("capacity = %d, want at least the floor %d", got, capacityFloor)
+	}
+}
+
+// TestCapacityRisesImmediatelyOnLargerBatch pins that growth is not deferred to a
+// window boundary. Under-allocating a large batch costs a reallocation right away,
+// so the ceiling must react on the spot.
+func TestCapacityRisesImmediatelyOnLargerBatch(t *testing.T) {
+	t.Parallel()
+
+	est := newCapacityEstimator(1_000)
+
+	est.observe(1)
+
+	small := est.capacity()
+
+	est.observe(500)
+
+	grown := est.capacity()
+
+	if grown <= small {
+		t.Fatalf("capacity did not rise: %d -> %d after observing 500", small, grown)
+	}
+
+	if grown < 500 {
+		t.Fatalf("capacity = %d, want at least the observed 500", grown)
+	}
+}
+
+// TestCapacityFullWorkloadStaysAtBatchSize pins the no-regression control.
+// Repeated full batches keep the recent maximum at the configured ceiling, so the
+// estimator must continue allocating BatchSize and never pay append growth.
+func TestCapacityFullWorkloadStaysAtBatchSize(t *testing.T) {
+	t.Parallel()
+
+	est := newCapacityEstimator(1_000)
+
+	for range capacityWindow * 2 {
+		est.observe(1_000)
+
+		if got := est.capacity(); got != 1_000 {
+			t.Fatalf("capacity = %d after a full batch, want 1000", got)
+		}
+	}
+}
+
+// TestCapacityDecaysAfterWindow pins that a batcher which quiets down releases its
+// oversized reservation, so a formerly busy batcher does not hold capacity forever.
+func TestCapacityDecaysAfterWindow(t *testing.T) {
+	t.Parallel()
+
+	est := newCapacityEstimator(1_000)
+
+	// Reach the ceiling under a busy workload.
+	est.observe(1_000)
+
+	busy := est.capacity()
+
+	if busy != 1_000 {
+		t.Fatalf("capacity = %d, want 1000 after a full batch", busy)
+	}
+
+	// Two full windows of sparse traffic.
+	for range capacityWindow * 2 {
+		est.observe(1)
+	}
+
+	quiet := est.capacity()
+
+	if quiet >= busy {
+		t.Fatalf("capacity did not decay: %d -> %d after sustained sparse traffic",
+			busy, quiet)
+	}
+}
+
+// TestCapacityHandlesAlternatingTraffic is the regression guard for the estimator
+// that was rejected during planning.
+//
+// An EWMA of the mean allocated MORE than doing nothing on this pattern, because the
+// mean sits between the two modes and every large batch grew from a too-small start.
+// Tracking the maximum must keep capacity sized for the large mode.
+func TestCapacityHandlesAlternatingTraffic(t *testing.T) {
+	t.Parallel()
+
+	est := newCapacityEstimator(1_000)
+
+	for range 50 {
+		est.observe(1)
+		est.observe(1_000)
+	}
+
+	if got := est.capacity(); got != 1_000 {
+		t.Fatalf("capacity = %d on alternating traffic, want 1000: a mean-based "+
+			"estimator would sit between the modes and reallocate every large batch",
+			got)
+	}
+}
+
+// TestRoundUpPow2 pins the rounding helper, including its floor and idempotence.
+func TestRoundUpPow2(t *testing.T) {
+	t.Parallel()
+
+	cases := map[int]int{
+		-5: capacityFloor,
+		0:  capacityFloor,
+		1:  capacityFloor,
+		16: capacityFloor,
+		17: 32,
+		32: 32,
+		33: 64,
+		63: 64,
+		65: 128,
+	}
+
+	for input, want := range cases {
+		if got := roundUpPow2(input); got != want {
+			t.Errorf("roundUpPow2(%d) = %d, want %d", input, got, want)
+		}
+	}
+}
diff --git a/pkg/batcher/options.go b/pkg/batcher/options.go
index fef5324..df975ff 100644
--- a/pkg/batcher/options.go
+++ b/pkg/batcher/options.go
@@ -4,11 +4,25 @@ import (
 	"time"
 )
 
+// Option configures a Batcher during New.
+//
+// Options are construction-time only. New applies them, validates the resulting
+// configuration, and freezes it before any processing goroutine can start. Calling
+// an Option on a Batcher after New returns is a deliberate no-op: runtime
+// reconfiguration was never coherent (the pipeline snapshots its configuration at
+// start), and allowing it would reintroduce races against workers.
+//
+// To change configuration, construct a new Batcher. In particular, WithSkipAutoStart
+// delays lifecycle start; it does not leave the configuration mutable until Start.
 type Option[T any] func(*Batcher[T])
 
 // WithProcessor sets the processor function to be called for each batch.
 func WithProcessor[T any](fn Processor[T]) Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		b.config.ProcessorFunc = fn
 	}
 }
@@ -16,6 +30,10 @@ func WithProcessor[T any](fn Processor[T]) Option[T] {
 // WithBatchSize sets the batch size.
 func WithBatchSize[T any](batchSize int) Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		if batchSize <= 0 {
 			batchSize = 1000
 		}
@@ -27,6 +45,10 @@ func WithBatchSize[T any](batchSize int) Option[T] {
 // WithBatchInterval sets the batch interval.
 func WithBatchInterval[T any](batchInterval time.Duration) Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		if batchInterval <= 0 {
 			batchInterval = 1 * time.Second
 		}
@@ -35,9 +57,17 @@ func WithBatchInterval[T any](batchInterval time.Duration) Option[T] {
 	}
 }
 
-// WithSkipAutoStart skips the automatic start of the batcher.
+// WithSkipAutoStart skips automatic lifecycle start.
+//
+// It does not defer configuration freeze: options are still applied only during New.
+// Call Start when ready to process, or Shutdown to drain queued work without an
+// explicit Start.
 func WithSkipAutoStart[T any]() Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		b.config.SkipAutoStart = true
 	}
 }
@@ -59,6 +89,10 @@ func WithSkipAutoStart[T any]() Option[T] {
 // the peak by up to two batches.
 func WithMaxQueueSize[T any](maxQueueSize int) Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		if maxQueueSize < 0 {
 			maxQueueSize = 0
 		}
@@ -71,6 +105,10 @@ func WithMaxQueueSize[T any](maxQueueSize int) Option[T] {
 // reporting ErrTimeout. The drain is never abandoned when the grace expires.
 func WithCloseGrace[T any](grace time.Duration) Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		if grace <= 0 {
 			grace = DefaultCloseGrace
 		}
@@ -92,6 +130,10 @@ func WithCloseGrace[T any](grace time.Duration) Option[T] {
 // loss.
 func WithErrorBufferSize[T any](size int) Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		if size <= 0 {
 			size = DefaultErrorBufferSize
 		}
@@ -122,6 +164,10 @@ func WithErrorBufferSize[T any](size int) Option[T] {
 // remaining processor time *plus* a full BatchInterval.
 func WithConcurrency[T any](concurrency int) Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		if concurrency < 1 {
 			concurrency = 1
 		}
@@ -145,6 +191,10 @@ func WithConcurrency[T any](concurrency int) Option[T] {
 //   - every accepted item is processed exactly once by the pool itself.
 func WithoutOrderedProcessing[T any]() Option[T] {
 	return func(b *Batcher[T]) {
+		if b.configFrozen.Load() {
+			return
+		}
+
 		b.config.UnorderedProcessingAcknowledged = true
 	}
 }
diff --git a/pkg/batcher/options_freeze_test.go b/pkg/batcher/options_freeze_test.go
new file mode 100644
index 0000000..6654f30
--- /dev/null
+++ b/pkg/batcher/options_freeze_test.go
@@ -0,0 +1,90 @@
+package batcher_test
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/NSXBet/batcher/internal/test"
+	"github.com/NSXBet/batcher/pkg/batcher"
+	"github.com/stretchr/testify/require"
+)
+
+// TestOptionsAreFrozenAfterNew pins the construction-time option contract.
+//
+// Option[T] is a callable function, so callers can retain one and invoke it after
+// New returns. Before this test, that was a data race: WithProcessor wrote the live
+// config while process() read ProcessorFunc, and the other options could similarly
+// change batching semantics mid-run. Configuration was never coherent at runtime,
+// so every post-New option application is deliberately a no-op.
+//
+// Run under -race: this fails if a future option bypasses configFrozen and writes a
+// live field again.
+func TestOptionsAreFrozenAfterNew(t *testing.T) {
+	t.Parallel()
+
+	var (
+		originalCalls atomic.Int64
+		mutatedCalls  atomic.Int64
+	)
+
+	original := batcher.Processor[test.BatchItem](func([]test.BatchItem) error {
+		originalCalls.Add(1)
+
+		return nil
+	})
+
+	mutated := batcher.Processor[test.BatchItem](func([]test.BatchItem) error {
+		mutatedCalls.Add(1)
+
+		return nil
+	})
+
+	b := batcher.New(
+		batcher.WithBatchSize[test.BatchItem](1),
+		batcher.WithBatchInterval[test.BatchItem](time.Millisecond),
+		batcher.WithProcessor(original),
+	)
+
+	defer func() { require.NoError(t, b.Close()) }()
+
+	// Apply every mutable option concurrently with active processing. They must all
+	// become no-ops after New; the original config remains the one run() snapshots.
+	var wg sync.WaitGroup
+
+	for range 4 {
+		wg.Add(1)
+
+		go func() {
+			defer wg.Done()
+
+			for range 200 {
+				batcher.WithProcessor(mutated)(b)
+				batcher.WithBatchSize[test.BatchItem](99)(b)
+				batcher.WithBatchInterval[test.BatchItem](time.Hour)(b)
+				batcher.WithConcurrency[test.BatchItem](8)(b)
+				batcher.WithMaxQueueSize[test.BatchItem](1)(b)
+			}
+		}()
+	}
+
+	for i := range 400 {
+		b.Add(test.BatchItem{Key: string(rune('a' + i%26))})
+	}
+
+	wg.Wait()
+
+	require.NoError(t, b.Join(10*time.Second))
+
+	config := b.Config()
+
+	require.Equal(t, 1, config.BatchSize)
+	require.Equal(t, time.Millisecond, config.BatchInterval)
+	require.Equal(t, 1, config.Concurrency)
+	require.Equal(t, 0, config.MaxQueueSize)
+	require.Equal(t, int64(400), originalCalls.Load(),
+		"processing must keep using the construction-time processor")
+	require.Zero(t, mutatedCalls.Load(),
+		"a post-start WithProcessor must not replace the original")
+}
diff --git a/pkg/batcher/options_test.go b/pkg/batcher/options_test.go
index 844b35f..dd11baf 100644
--- a/pkg/batcher/options_test.go
+++ b/pkg/batcher/options_test.go
@@ -10,67 +10,52 @@ import (
 )
 
 func TestWithProcessor(t *testing.T) {
-	// ARRANGE
-	b := batcher.New[test.BatchItem]()
+	called := false
 
-	var (
-		pr        batcher.Processor[test.BatchItem]
-		processor batcher.Processor[test.BatchItem]
-	)
+	processor := batcher.Processor[test.BatchItem](func(_ []test.BatchItem) error {
+		called = true
 
-	processor = func(_ []test.BatchItem) error {
 		return nil
-	}
-
-	// ACT
-	batcher.WithProcessor(processor)(b)
+	})
 
-	// ASSERT
-	pr = b.Config().ProcessorFunc
-	require.IsType(t, processor, pr)
+	// Options are construction-time configuration. Applying one after New starts
+	// the batcher races its aggregation goroutine and was never a coherent runtime
+	// reconfiguration API.
+	b := batcher.New(batcher.WithProcessor(processor))
+	defer func() { _ = b.Close() }()
+
+	// Invoke the configured processor rather than checking its type. Every
+	// Processor[T] has the same type, including the default no-op, so a type
+	// assertion would pass even if WithProcessor had done nothing.
+	require.NoError(t, b.Config().ProcessorFunc(nil))
+	require.True(t, called, "the configured processor must replace the default")
 }
 
 func TestWithBatchSize(t *testing.T) {
-	// ARRANGE
-	b := batcher.New[test.BatchItem]()
-
-	// ACT
-	batcher.WithBatchSize[test.BatchItem](1000)(b)
+	b := batcher.New(batcher.WithBatchSize[test.BatchItem](1000))
+	defer func() { _ = b.Close() }()
 
-	// ASSERT
 	require.Equal(t, 1000, b.Config().BatchSize)
 
 	t.Run("WithBatchSize - zero size", func(t *testing.T) {
-		// ARRANGE
-		b := batcher.New[test.BatchItem]()
+		b := batcher.New(batcher.WithBatchSize[test.BatchItem](0))
+		defer func() { _ = b.Close() }()
 
-		// ACT
-		batcher.WithBatchSize[test.BatchItem](0)(b)
-
-		// ASSERT
 		require.Equal(t, batcher.DefaultBatchSize, b.Config().BatchSize)
 	})
 }
 
 func TestWithBatchInterval(t *testing.T) {
-	// ARRANGE
-	b := batcher.New[test.BatchItem]()
-	duration := 1 * time.Second
-
-	// ACT
-	batcher.WithBatchInterval[test.BatchItem](duration)(b)
+	duration := time.Second
+	b := batcher.New(batcher.WithBatchInterval[test.BatchItem](duration))
+	defer func() { _ = b.Close() }()
 
-	// ASSERT
 	require.Equal(t, duration, b.Config().BatchInterval)
 
 	t.Run("WithBatchInterval - zero duration", func(t *testing.T) {
-		// ARRANGE
-		b := batcher.New[test.BatchItem]()
-
-		// ACT
-		batcher.WithBatchInterval[test.BatchItem](0)(b)
+		b := batcher.New(batcher.WithBatchInterval[test.BatchItem](0))
+		defer func() { _ = b.Close() }()
 
-		// ASSERT
 		require.Equal(t, batcher.DefaultBatchInterval, b.Config().BatchInterval)
 	})
 }
diff --git a/pkg/batcher/queue.go b/pkg/batcher/queue.go
index 209ff9c..46a7e5e 100644
--- a/pkg/batcher/queue.go
+++ b/pkg/batcher/queue.go
@@ -109,7 +109,108 @@ func (q *queue[T]) tryPush(item T) bool {
 	return true
 }
 
+// popBatch moves up to max queued items onto dst and returns the extended slice.
+//
+// This exists because the aggregator is a single consumer that drains greedily: it
+// pops until the queue is empty on every wakeup. Doing that one item at a time takes
+// and releases the mutex once per item, and under concurrent producers those
+// acquisitions are what serialise the queue. Profiling an 8-producer, 200k-item MPSC
+// workload attributed 86% of all mutex delay to the push path waiting behind exactly
+// this traffic, and 61% of CPU samples to push.
+//
+// Transferring a run under one acquisition is safe precisely because the consumer is
+// single and holds no invariant between items: it appends them to the batch it is
+// building, so one block transfer and N single pops are indistinguishable to it.
+//
+// max bounds the transfer so a large backlog cannot be moved in one acquisition,
+// which would hold the lock for an unbounded time and starve producers. Passing
+// dst[:0] reuses the caller's buffer, so steady-state draining does not allocate.
+//
+// Measured against per-item pop on the same workload (darwin/arm64, Apple M4 Pro,
+// Go 1.26.5, 200k items): 5.4x faster with one producer, 1.1x with 256, and
+// unbounded-mode allocation down from ~8MiB to ~0.2MiB per run. For reference, a
+// lock-free MPSC queue (xsync/v4 UMPSCQueue) measured within noise of this on the
+// same workload, so the mutex was never the cost -- per-item locking was.
+func (q *queue[T]) popBatch(dst []T, max int) []T {
+	if max <= 0 {
+		return dst
+	}
+
+	q.mu.Lock()
+
+	queued := len(q.items) - q.head
+	if queued == 0 {
+		// Fully drained: reset the cursor and keep the backing array, so the next
+		// batch of pushes reuses capacity instead of allocating.
+		if q.head > 0 {
+			q.items = q.items[:0]
+			q.head = 0
+		}
+
+		q.mu.Unlock()
+
+		return dst
+	}
+
+	n := min(queued, max)
+
+	dst = append(dst, q.items[q.head:q.head+n]...)
+
+	// Clear the transferred slots so popped items are not kept alive by the backing
+	// array.
+	clear(q.items[q.head : q.head+n])
+
+	q.head += n
+
+	remaining := len(q.items) - q.head
+
+	q.reclaimLocked(remaining)
+
+	q.mu.Unlock()
+
+	if remaining > 0 {
+		// Keep the latch armed: the aggregator drains greedily, but re-arming means
+		// a missed wakeup cannot strand items.
+		signal(q.notEmpty)
+	}
+
+	signal(q.notFull)
+
+	return dst
+}
+
+// reclaimLocked keeps the backing array proportional to queue depth rather than to
+// total throughput. The caller must hold q.mu.
+//
+// Without this, push always appends and the full-drain reset only fires when the
+// queue is observed completely empty. A steady producer that keeps even one item
+// resident never triggers it, so the array grows without bound. Measured before this
+// compaction: a queue holding a single item reached 219,136 slots after 200,000
+// pushes.
+//
+// Compacting when head >= remaining keeps the copy cost amortised O(1) per item,
+// because each compaction halves the live region's offset and moves at most as many
+// items as have been consumed since the last one.
+func (q *queue[T]) reclaimLocked(remaining int) {
+	if remaining == 0 {
+		q.items = q.items[:0]
+		q.head = 0
+
+		return
+	}
+
+	if q.head >= remaining {
+		copy(q.items, q.items[q.head:])
+		clear(q.items[remaining:])
+
+		q.items = q.items[:remaining]
+		q.head = 0
+	}
+}
+
 // pop removes the oldest item, reporting false when the queue is empty.
+//
+// popBatch is the aggregator's path; this remains for single-item callers and tests.
 func (q *queue[T]) pop() (T, bool) {
 	var zero T
 
@@ -135,25 +236,7 @@ func (q *queue[T]) pop() (T, bool) {
 
 	remaining := len(q.items) - q.head
 
-	// Reclaim the consumed prefix once it dominates the slice.
-	//
-	// Without this, the backing array grows with total throughput rather than with
-	// queue depth: push always appends, and the full-drain reset above only fires
-	// when the queue is observed completely empty. A steady producer that keeps even
-	// one item resident never triggers it, so the array grows without bound.
-	// Measured before this compaction: a queue holding a single item reached 219,136
-	// slots after 200,000 pushes.
-	//
-	// Compacting when head >= remaining keeps the copy cost amortised O(1) per item,
-	// because each compaction halves the live region's offset and moves at most as
-	// many items as have been consumed since the last one.
-	if q.head >= remaining {
-		copy(q.items, q.items[q.head:])
-		clear(q.items[remaining:])
-
-		q.items = q.items[:remaining]
-		q.head = 0
-	}
+	q.reclaimLocked(remaining)
 
 	q.mu.Unlock()
 
diff --git a/pkg/batcher/queue_bench_test.go b/pkg/batcher/queue_bench_test.go
new file mode 100644
index 0000000..37204e8
--- /dev/null
+++ b/pkg/batcher/queue_bench_test.go
@@ -0,0 +1,66 @@
+package batcher
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+)
+
+// BenchmarkQueueBatchDrainMPSC measures Batcher's queue in its real access shape:
+// multiple publishers and exactly one greedy-draining aggregator.
+//
+// It deliberately exercises the queue directly instead of BenchmarkBatcherEnqueue:
+// the latter measures the public producer API (admission, lifecycle counters and the
+// processor) and is the right compatibility benchmark, but it cannot attribute a
+// CPU or mutex profile to the intake queue itself.
+//
+// Capture profiles with:
+//
+//	go test -run '^$' -bench '^BenchmarkQueueBatchDrainMPSC$' -benchtime=5s \
+//	  -cpuprofile cpu.out -memprofile mem.out -mutexprofile mutex.out ./pkg/batcher
+//
+// Inspect with `go tool pprof -http=:0 cpu.out` (and the other profiles). In
+// particular, compare mutex delay in queue.push against earlier profiles before
+// reintroducing any per-item locking or replacing the queue with an atomic structure.
+func BenchmarkQueueBatchDrainMPSC(b *testing.B) {
+	q := newQueue[int](0)
+	sealCh := make(chan struct{})
+	ctx := context.Background()
+
+	var received atomic.Int64
+
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		buf := make([]int, 0, DefaultBatchSize)
+
+		for received.Load() < int64(b.N) {
+			<-q.ready()
+
+			for {
+				buf = q.popBatch(buf[:0], DefaultBatchSize)
+				if len(buf) == 0 {
+					break
+				}
+
+				received.Add(int64(len(buf)))
+			}
+		}
+	}()
+
+	b.ReportAllocs()
+	b.ResetTimer()
+
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			if err := q.push(ctx, 1, sealCh); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+
+	b.StopTimer()
+	<-done
+}
diff --git a/pkg/batcher/queue_test.go b/pkg/batcher/queue_test.go
index 8b1817d..ef0c5c5 100644
--- a/pkg/batcher/queue_test.go
+++ b/pkg/batcher/queue_test.go
@@ -3,6 +3,7 @@ package batcher
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -74,3 +75,176 @@ func TestQueueCompactionPreservesOrder(t *testing.T) {
 	_, ok := q.pop()
 	require.False(t, ok)
 }
+
+// TestPopBatchPreservesFIFOAcrossCompaction is the equivalence check that matters:
+// batch draining must be indistinguishable from repeated pop, including across the
+// prefix compaction that moves the live region.
+func TestPopBatchPreservesFIFOAcrossCompaction(t *testing.T) {
+	t.Parallel()
+
+	q := newQueue[int](0)
+	sealCh := make(chan struct{})
+
+	const total = 5_000
+
+	var (
+		got  []int
+		next int
+		buf  []int
+	)
+
+	// Interleave pushes and partial batch drains so head advances, the suffix is
+	// compacted, and drains land mid-array rather than always at offset zero.
+	for next < total {
+		for range 7 {
+			if next >= total {
+				break
+			}
+
+			require.NoError(t, q.push(context.Background(), next, sealCh))
+
+			next++
+		}
+
+		buf = q.popBatch(buf[:0], 3)
+		got = append(got, buf...)
+	}
+
+	for {
+		buf = q.popBatch(buf[:0], 64)
+		if len(buf) == 0 {
+			break
+		}
+
+		got = append(got, buf...)
+	}
+
+	require.Len(t, got, total)
+
+	for i, v := range got {
+		require.Equal(t, i, v, "batch draining must preserve FIFO order")
+	}
+}
+
+// TestPopBatchRespectsMaxAndReportsEmpty pins the transfer cap, which is what bounds
+// how long the queue lock is held. Without it a large backlog would move in one
+// acquisition and starve producers for an unbounded time.
+func TestPopBatchRespectsMaxAndReportsEmpty(t *testing.T) {
+	t.Parallel()
+
+	q := newQueue[int](0)
+	sealCh := make(chan struct{})
+
+	for i := range 10 {
+		require.NoError(t, q.push(context.Background(), i, sealCh))
+	}
+
+	got := q.popBatch(nil, 4)
+	require.Equal(t, []int{0, 1, 2, 3}, got, "at most max items may be transferred")
+
+	got = q.popBatch(got[:0], 100)
+	require.Equal(t, []int{4, 5, 6, 7, 8, 9}, got,
+		"a max above the queued count must transfer only what is queued")
+
+	require.Empty(t, q.popBatch(got[:0], 8), "an empty queue must transfer nothing")
+
+	// A non-positive max is a caller bug; returning dst unchanged keeps it from
+	// silently draining the queue or panicking.
+	require.NoError(t, q.push(context.Background(), 42, sealCh))
+	require.Empty(t, q.popBatch(nil, 0))
+	require.Equal(t, 1, q.length(), "max<=0 must not consume items")
+}
+
+// TestPopBatchReusesCallerBuffer pins the allocation property the aggregator relies
+// on: steady-state draining into a reused buffer must not allocate.
+func TestPopBatchReusesCallerBuffer(t *testing.T) {
+	// AllocsPerRun temporarily changes process-wide GC settings, so Go rejects it
+	// from a parallel test.
+	q := newQueue[int](0)
+	sealCh := make(chan struct{})
+
+	buf := make([]int, 0, 64)
+
+	// Warm up so slice growth and the queue's own backing array settle.
+	for range 64 {
+		require.NoError(t, q.push(context.Background(), 1, sealCh))
+	}
+
+	buf = q.popBatch(buf[:0], 64)
+	require.Len(t, buf, 64)
+
+	allocs := testing.AllocsPerRun(200, func() {
+		for range 32 {
+			_ = q.push(context.Background(), 1, sealCh)
+		}
+
+		buf = q.popBatch(buf[:0], 32)
+	})
+
+	require.Zero(t, allocs,
+		"draining into a reused buffer must not allocate; got %.1f allocs/op", allocs)
+}
+
+// TestPopBatchDrainedQueueReclaimsStorage pins that batch draining keeps the same
+// memory bound as pop: backing storage proportional to depth, not to throughput.
+func TestPopBatchDrainedQueueReclaimsStorage(t *testing.T) {
+	t.Parallel()
+
+	q := newQueue[int](0)
+	sealCh := make(chan struct{})
+
+	var buf []int
+
+	require.NoError(t, q.push(context.Background(), 0, sealCh))
+
+	const cycles = 50_000
+
+	for i := range cycles {
+		require.NoError(t, q.push(context.Background(), i, sealCh))
+
+		buf = q.popBatch(buf[:0], 1)
+		require.Len(t, buf, 1, "cycle %d: queue unexpectedly empty", i)
+	}
+
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	require.Equal(t, 1, len(q.items)-q.head, "one item must remain live")
+	require.LessOrEqual(t, cap(q.items), 8,
+		"backing capacity must stay proportional to depth; got cap=%d after %d cycles",
+		cap(q.items), cycles)
+}
+
+// TestPopBatchSignalsBoundedProducers pins the actual release mechanism: a batch
+// transfer from a bounded queue arms notFull. push selects on this latch while full,
+// so it is the signal that wakes parked publishers. Testing the channel directly is
+// stronger than racing a goroutine against the drain: a publisher can be scheduled
+// to retry after space exists even if the signal is accidentally removed.
+func TestPopBatchSignalsBoundedProducers(t *testing.T) {
+	t.Parallel()
+
+	q := newQueue[int](4)
+	sealCh := make(chan struct{})
+
+	for i := range 4 {
+		require.NoError(t, q.push(context.Background(), i, sealCh))
+	}
+
+	require.False(t, q.tryPush(99), "queue must be full")
+
+	// Ensure the signal observed below comes from this popBatch, not an earlier pop.
+	select {
+	case <-q.notFull:
+	default:
+	}
+
+	buf := q.popBatch(nil, 4)
+	require.Len(t, buf, 4)
+
+	select {
+	case <-q.notFull:
+		// The one-place latch is armed. A publisher parked in push can now retry.
+	case <-time.After(time.Second):
+		t.Fatal("popBatch did not arm notFull; bounded publishers would remain parked")
+	}
+}
diff --git a/pkg/batcher/retention_test.go b/pkg/batcher/retention_test.go
new file mode 100644
index 0000000..e213da9
--- /dev/null
+++ b/pkg/batcher/retention_test.go
@@ -0,0 +1,150 @@
+package batcher_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/NSXBet/batcher/pkg/batcher"
+	"github.com/stretchr/testify/require"
+)
+
+// TestProcessorMayRetainBatchSlices is the contract test for adaptive batch
+// capacity.
+//
+// The capacity estimator changes how large each batch slice starts, but it must not
+// change slice *ownership*: the processor receives a slice it may keep, and no later
+// batch may write through it. This is the property that ruled out pooling during
+// planning, so it is asserted directly rather than assumed.
+//
+// The test retains every batch, mutates the retained copies after the fact, and then
+// verifies that nothing the batcher did afterwards corrupted them.
+func TestProcessorMayRetainBatchSlices(t *testing.T) {
+	t.Parallel()
+
+	var (
+		mu       sync.Mutex
+		retained [][]int
+	)
+
+	b := batcher.New(
+		// Small batches with a short interval: many flushes, so the estimator adapts
+		// repeatedly while batches are being retained.
+		batcher.WithBatchSize[int](8),
+		batcher.WithBatchInterval[int](2*time.Millisecond),
+		batcher.WithProcessor(func(items []int) error {
+			// Deliberately keep the slice the batcher handed us, without copying.
+			mu.Lock()
+			retained = append(retained, items)
+			mu.Unlock()
+
+			return nil
+		}),
+	)
+
+	const total = 500
+
+	for i := range total {
+		b.Add(i)
+
+		// Vary arrival pacing so batches close on both size and timer, driving the
+		// estimator up and down during the run.
+		if i%37 == 0 {
+			time.Sleep(3 * time.Millisecond)
+		}
+	}
+
+	require.NoError(t, b.Shutdown(context.Background()))
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	// Every accepted item must appear exactly once across the retained slices, and
+	// each slice must still hold what it held when the processor saw it.
+	seen := make(map[int]int, total)
+	count := 0
+
+	for i, batch := range retained {
+		require.NotEmpty(t, batch, "batch %d: no empty batch may be emitted", i)
+		require.LessOrEqual(t, len(batch), 8,
+			"batch %d: len must never exceed BatchSize", i)
+
+		for _, item := range batch {
+			seen[item]++
+			count++
+		}
+	}
+
+	require.Equal(t, total, count,
+		"retained slices must still contain every accepted item exactly once; "+
+			"a reused backing array would show duplicates or missing values")
+
+	for i := range total {
+		require.Equal(t, 1, seen[i], "item %d appeared %d times", i, seen[i])
+	}
+}
+
+// TestRetainedBatchSlicesAreNotAliased pins the stronger property: two retained
+// batches must not share backing storage. If they did, mutating one would corrupt
+// the other, which is exactly the failure mode a pooled or reused slice introduces.
+func TestRetainedBatchSlicesAreNotAliased(t *testing.T) {
+	t.Parallel()
+
+	var (
+		mu       sync.Mutex
+		retained [][]int
+	)
+
+	b := batcher.New(
+		batcher.WithBatchSize[int](4),
+		batcher.WithBatchInterval[int](2*time.Millisecond),
+		batcher.WithProcessor(func(items []int) error {
+			mu.Lock()
+			retained = append(retained, items)
+			mu.Unlock()
+
+			return nil
+		}),
+	)
+
+	for i := range 100 {
+		b.Add(i)
+	}
+
+	require.NoError(t, b.Shutdown(context.Background()))
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	require.Greater(t, len(retained), 1, "need several batches to compare")
+
+	// Snapshot the retained contents, then overwrite every retained slice. If any
+	// two shared storage, the overwrite of a later batch would change an earlier one.
+	snapshots := make([][]int, len(retained))
+	for i, batch := range retained {
+		snapshots[i] = append([]int(nil), batch...)
+	}
+
+	for i, batch := range retained {
+		for j := range batch {
+			batch[j] = -(i + 1)
+		}
+	}
+
+	for i, batch := range retained {
+		for j := range batch {
+			require.Equal(t, -(i + 1), batch[j],
+				"batch %d slot %d was changed by a write to another batch, so the "+
+					"batches share backing storage", i, j)
+		}
+	}
+
+	// And the snapshots prove the original contents were distinct per batch.
+	total := 0
+	for _, snap := range snapshots {
+		total += len(snap)
+	}
+
+	require.Equal(t, 100, total, "every accepted item must have been delivered once")
+}
diff --git a/pkg/batcher/runtime_snapshot_test.go b/pkg/batcher/runtime_snapshot_test.go
new file mode 100644
index 0000000..83fb5c9
--- /dev/null
+++ b/pkg/batcher/runtime_snapshot_test.go
@@ -0,0 +1,84 @@
+package batcher
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestProcessingReadsOnlyTheRuntimeSnapshot pins that processing goroutines never
+// read the mutable Config struct.
+//
+// Config is a plain struct and a caller can hold a reference to one, so any read of
+// it from a worker is a data race by construction. run() previously snapshotted
+// batch size, interval and worker count but still dereferenced
+// b.config.ProcessorFunc on every batch, which made the rest of that snapshot
+// pointless: one live field is enough to race.
+//
+// This test writes the Config field directly, bypassing the frozen-option guard
+// entirely, because the guard is a separate defence. Under -race it fails if any
+// processing path reads Config again.
+func TestProcessingReadsOnlyTheRuntimeSnapshot(t *testing.T) {
+	t.Parallel()
+
+	var processed int
+
+	b := New(
+		WithBatchSize[int](1),
+		WithBatchInterval[int](time.Millisecond),
+		WithProcessor(func(items []int) error {
+			processed += len(items)
+
+			return nil
+		}),
+	)
+
+	t.Cleanup(func() { _ = b.Close() })
+
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		for range 300 {
+			b.Add(1)
+		}
+	}()
+
+	// A hostile or stale caller mutating configuration while work is in flight.
+	for range 300 {
+		b.config.ProcessorFunc = func([]int) error { return nil }
+		b.config.BatchSize = 999
+		b.config.BatchInterval = time.Hour
+	}
+
+	<-done
+
+	require.NoError(t, b.Join(10*time.Second))
+
+	// The pipeline must still be running the configuration it started with.
+	require.Equal(t, 1, b.runtime.batchSize)
+	require.Equal(t, time.Millisecond, b.runtime.batchInterval)
+	require.Equal(t, 1, b.runtime.workers)
+	require.NotNil(t, b.runtime.processor)
+	require.Positive(t, processed,
+		"the original processor must have run; a swapped-in one would not count here")
+}
+
+// TestRuntimeSnapshotNormalisesInvalidConfig pins the defaults applied when the
+// snapshot is taken, so a zero or negative value cannot produce a pipeline with no
+// capacity or no processor.
+func TestRuntimeSnapshotNormalisesInvalidConfig(t *testing.T) {
+	t.Parallel()
+
+	b := New[int](WithSkipAutoStart[int]())
+
+	t.Cleanup(func() { _ = b.Close() })
+
+	// Defaults, not zero values.
+	require.Positive(t, b.runtime.batchSize)
+	require.GreaterOrEqual(t, b.runtime.workers, 1)
+	require.NotNil(t, b.runtime.processor,
+		"a nil processor must normalise to the no-op rather than panicking a worker")
+}
diff --git a/pkg/batcher/stats.go b/pkg/batcher/stats.go
index a346ab9..52eb788 100644
--- a/pkg/batcher/stats.go
+++ b/pkg/batcher/stats.go
@@ -41,6 +41,12 @@ type Stats struct {
 	// the aggregator. This is the queue depth to alert on.
 	Queued int64
 
+	// BatchHeld is items owned by the aggregator: they have left the intake queue
+	// but have not yet entered a worker. This includes items accumulating in the
+	// current partial batch and a flushed batch blocked on the unbuffered worker
+	// dispatch channel. It is distinct from Queued and InFlight.
+	BatchHeld int64
+
 	// InFlight is items currently inside a processor call. It is updated by workers
 	// on batch boundaries, not on the enqueue path, so observability does not add
 	// another contended producer atomic.
@@ -62,6 +68,14 @@ type Stats struct {
 	// expired.
 	Rejected uint64
 
+	// BatchesFlushed counts batches emitted by the aggregator, in batches rather
+	// than items. After a terminal drain, (Completed + Failed + Panicked) /
+	// BatchesFlushed gives the mean batch size. All terminal outcomes are included:
+	// a processor failure or recovered panic does not make its batch disappear from
+	// the coalescing figure. A mean far below BatchSize means the window is closing
+	// on the timer rather than on size.
+	BatchesFlushed uint64
+
 	// DroppedErrors counts diagnostics discarded because the Errors() buffer was
 	// full. A non-zero value means diagnostics are being lost, not that batches
 	// failed: it is a signal that Errors() is not being drained fast enough.
@@ -76,15 +90,17 @@ type Stats struct {
 // contended atomics per Add measured up to 4x the cost of one, so nothing is added
 // to the hot path without a reason.
 type counters struct {
-	pending       atomic.Int64
-	intakePending atomic.Int64
-	accepted      atomic.Uint64
-	inFlight      atomic.Int64
-	completed     atomic.Uint64
-	failed        atomic.Uint64
-	panicked      atomic.Uint64
-	rejected      atomic.Uint64
-	droppedErrors atomic.Uint64
+	pending        atomic.Int64
+	intakePending  atomic.Int64
+	accepted       atomic.Uint64
+	batchHeld      atomic.Int64
+	inFlight       atomic.Int64
+	completed      atomic.Uint64
+	failed         atomic.Uint64
+	panicked       atomic.Uint64
+	batchesFlushed atomic.Uint64
+	rejected       atomic.Uint64
+	droppedErrors  atomic.Uint64
 }
 
 // reserve claims a drain obligation before publication.
@@ -112,9 +128,27 @@ func (c *counters) accept() {
 	c.accepted.Add(1)
 }
 
-// received records the aggregator taking an item out of the queue.
+// received records the aggregator taking an item out of the queue and into the
+// batch it is currently accumulating.
+//
+// This is a transfer, not a decrement: the item stops being queued and starts
+// being aggregator-held, so both counters move together. Decrementing intake
+// alone would make the item invisible in every field between leaving the queue
+// and entering a worker.
 func (c *counters) received(n int) {
 	c.intakePending.Add(int64(-n))
+	c.batchHeld.Add(int64(n))
+}
+
+// dispatched records a batch leaving the aggregator for a worker.
+//
+// batchHeld is released here rather than when the processor returns, because
+// inFlight takes ownership at that point; counting both would double-count the
+// batch. It runs once per batch on the aggregation path, so it adds no per-item
+// producer cost.
+func (c *counters) dispatched(n int) {
+	c.batchHeld.Add(int64(-n))
+	c.batchesFlushed.Add(1)
 }
 
 // terminal records exactly one outcome for a finished batch and releases its
diff --git a/pkg/batcher/stats_test.go b/pkg/batcher/stats_test.go
new file mode 100644
index 0000000..e3c198c
--- /dev/null
+++ b/pkg/batcher/stats_test.go
@@ -0,0 +1,232 @@
+package batcher_test
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/NSXBet/batcher/pkg/batcher"
+	"github.com/stretchr/testify/require"
+)
+
+// Stats snapshot contract tests.
+//
+// Stats is intentionally an eventually consistent diagnostic interface: each
+// field is read independently, so callers cannot use one live snapshot as a
+// transactional accounting assertion. These tests pin the meaningful contract:
+// ownership transitions are represented, terminal accounting converges after the
+// drain, and observing the snapshot is allocation-free and cannot perturb Add.
+
+// TestStatsShowsAggregatorHeldPartialBatch distinguishes the three ownership
+// states. A partial batch has left the intake queue but cannot yet be in flight,
+// because no worker receives it until the timer/size/shutdown flush. That item must
+// be visible as BatchHeld rather than disappearing between Queued and InFlight.
+func TestStatsShowsAggregatorHeldPartialBatch(t *testing.T) {
+	t.Parallel()
+
+	b := batcher.New(
+		batcher.WithBatchSize[int](100),
+		batcher.WithBatchInterval[int](time.Hour), // only shutdown can flush it
+		batcher.WithProcessor(batcher.NoOpProcessor[int]),
+	)
+
+	b.Add(1)
+
+	require.Eventually(t, func() bool {
+		stats := b.Stats()
+
+		return stats.BatchHeld == 1 && stats.Queued == 0 && stats.InFlight == 0
+	}, 5*time.Second, time.Millisecond,
+		"an item received by the aggregator but not dispatched must be BatchHeld")
+
+	before := b.Stats()
+
+	require.Equal(t, int64(1), before.Pending)
+	require.Equal(t, int64(1), before.BatchHeld)
+	require.Equal(t, uint64(0), before.BatchesFlushed)
+
+	require.NoError(t, b.Close())
+
+	after := b.Stats()
+
+	require.Zero(t, after.Pending)
+	require.Zero(t, after.BatchHeld)
+	require.Zero(t, after.InFlight)
+	require.Equal(t, uint64(1), after.BatchesFlushed,
+		"the shutdown partial batch is still a flushed batch")
+	require.Equal(t, uint64(1), after.Completed)
+}
+
+// TestStatsCountsBatchesFlushedAcrossEveryFlushPath pins that BatchesFlushed is a
+// batch counter, independent of whether a flush happened by size, timer, or final
+// shutdown. This is the coalescing signal operators need when choosing a window.
+func TestStatsCountsBatchesFlushedAcrossEveryFlushPath(t *testing.T) {
+	t.Parallel()
+
+	const (
+		batchSize = 3
+		interval  = 30 * time.Millisecond
+	)
+
+	b := batcher.New(
+		batcher.WithBatchSize[int](batchSize),
+		batcher.WithBatchInterval[int](interval),
+		batcher.WithProcessor(batcher.NoOpProcessor[int]),
+	)
+
+	// Size flush.
+	for i := range batchSize {
+		b.Add(i)
+	}
+
+	require.Eventually(t, func() bool {
+		return b.Stats().BatchesFlushed >= 1
+	}, 5*time.Second, time.Millisecond)
+
+	// Timer flush.
+	b.Add(100)
+	b.Add(101)
+
+	require.Eventually(t, func() bool {
+		return b.Stats().BatchesFlushed >= 2
+	}, 5*time.Second, time.Millisecond)
+
+	// Shutdown partial flush.
+	b.Add(200)
+
+	require.NoError(t, b.Close())
+
+	stats := b.Stats()
+
+	require.Equal(t, uint64(3), stats.BatchesFlushed)
+	require.Equal(t, uint64(batchSize+2+1), stats.Completed)
+	require.Zero(t, stats.Pending)
+	require.Zero(t, stats.BatchHeld)
+	require.Zero(t, stats.InFlight)
+}
+
+// TestStatsTerminalOutcomesAreMutuallyExclusive pins accounting at the only point
+// where a multi-field equality is meaningful: after terminal drain, when no
+// publisher or worker can still be moving an item between fields.
+func TestStatsTerminalOutcomesAreMutuallyExclusive(t *testing.T) {
+	t.Parallel()
+
+	var calls atomic.Int64
+	failure := errors.New("processor failed")
+
+	b := batcher.New(
+		batcher.WithBatchSize[int](1),
+		batcher.WithBatchInterval[int](time.Millisecond),
+		batcher.WithProcessor(func([]int) error {
+			switch calls.Add(1) {
+			case 1:
+				return nil
+			case 2:
+				return failure
+			default:
+				panic("poison batch")
+			}
+		}),
+	)
+
+	// Do not let diagnostics block the scenario: outcomes, not diagnostics,
+	// are under test here.
+	go func() {
+		for range b.Errors() {
+		}
+	}()
+
+	for i := range 3 {
+		b.Add(i)
+	}
+
+	require.NoError(t, b.Shutdown(context.Background()))
+
+	stats := b.Stats()
+
+	require.Equal(t, uint64(3), stats.Accepted)
+	require.Equal(t, uint64(1), stats.Completed)
+	require.Equal(t, uint64(1), stats.Failed)
+	require.Equal(t, uint64(1), stats.Panicked)
+	require.Equal(t, stats.Accepted,
+		stats.Completed+stats.Failed+stats.Panicked,
+		"terminal categories are mutually exclusive and exhaustive after drain")
+	require.Zero(t, stats.Pending)
+	require.Zero(t, stats.IntakePending)
+	require.Zero(t, stats.BatchHeld)
+	require.Zero(t, stats.InFlight)
+	require.Zero(t, stats.PublishersInGate)
+}
+
+// TestStatsIsAllocationFree pins the read-side performance contract. Queued takes
+// the queue mutex for a length check, but neither that lock nor the atomic loads
+// may allocate. Metrics scraping must never generate garbage.
+func TestStatsIsAllocationFree(t *testing.T) {
+	// Deliberately not parallel: AllocsPerRun requires exclusive GOMAXPROCS.
+	b := batcher.New(
+		batcher.WithBatchSize[int](100),
+		batcher.WithBatchInterval[int](time.Hour),
+		batcher.WithProcessor(batcher.NoOpProcessor[int]),
+	)
+
+	defer func() { require.NoError(t, b.Close()) }()
+
+	for range 100 {
+		b.Add(1)
+	}
+
+	allocs := testing.AllocsPerRun(2_000, func() {
+		_ = b.Stats()
+	})
+
+	require.Zero(t, allocs, "Stats must not allocate")
+}
+
+// TestStatsIsEventuallyConsistent, rather than pretending live counters are a
+// transaction, makes the boundary explicit. While the processor is blocked, the
+// snapshot has a stable meaningful ownership state. Once it is released and
+// terminal drain completes, the conservation equality holds. Callers must not
+// infer that arbitrary intermediate snapshots satisfy the latter equality.
+func TestStatsIsEventuallyConsistent(t *testing.T) {
+	t.Parallel()
+
+	entered := make(chan struct{})
+	release := make(chan struct{})
+
+	b := batcher.New(
+		batcher.WithBatchSize[int](1),
+		batcher.WithBatchInterval[int](time.Millisecond),
+		batcher.WithProcessor(func([]int) error {
+			close(entered)
+			<-release
+
+			return nil
+		}),
+	)
+
+	b.Add(1)
+
+	select {
+	case <-entered:
+	case <-time.After(5 * time.Second):
+		t.Fatal("processor did not start")
+	}
+
+	live := b.Stats()
+
+	require.Equal(t, int64(1), live.InFlight,
+		"a stable live snapshot exposes the item in its current ownership state")
+	require.Equal(t, int64(1), live.Pending)
+
+	close(release)
+	require.NoError(t, b.Shutdown(context.Background()))
+
+	settled := b.Stats()
+
+	require.Zero(t, settled.Pending)
+	require.Equal(t, settled.Accepted,
+		settled.Completed+settled.Failed+settled.Panicked,
+		"only a terminally drained snapshot is a valid accounting assertion")
+}
diff --git a/test/scenario/allocation_evidence_test.go b/test/scenario/allocation_evidence_test.go
new file mode 100644
index 0000000..f5d33be
--- /dev/null
+++ b/test/scenario/allocation_evidence_test.go
@@ -0,0 +1,103 @@
+package scenario_test
+
+import (
+	"testing"
+	"time"
+	"unsafe"
+
+	"github.com/NSXBet/batcher/test/scenario"
+	"github.com/stretchr/testify/require"
+)
+
+// TestSparseWindowAllocationEvidence measures allocated bytes per flush for the
+// workload Milestone 4.2 targets: a small window with a large BatchSize, where
+// every batch closes on the timer holding far fewer items than it reserved
+// capacity for.
+//
+// This exists to decide whether 4.2 should be implemented at all. The plan gates it
+// on sparse-window allocation pressure above 2 KB/flush.
+//
+// The "bytes/flush (est)" column is the PRE-ADAPTIVE BASELINE, deliberately kept
+// after 4.2 landed: it is (BatchSize - actual) * sizeof(T), the waste a fixed
+// make([]T, 0, BatchSize) reservation would incur. The aggregator now reserves
+// capacities.capacity() instead, so this column reports the waste adaptive capacity
+// removed rather than waste that remains. It is the figure the milestone is measured
+// against, which is why it is not recomputed from the current reservation.
+//
+// The measurements are report-only: no allocation threshold is asserted, because the
+// numbers are an input to a decision rather than a gate. The scenario's own validity
+// is still enforced -- a run that timed out is not evidence of anything, so
+// result.TimedOut fails the test.
+func TestSparseWindowAllocationEvidence(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip("measurement only; runs in the full suite")
+	}
+
+	cases := []struct {
+		name      string
+		batchSize int
+		window    time.Duration
+		rate      int
+	}{
+		// The worst realistic case: 1ms window, 1000-item BatchSize, sparse arrivals.
+		{"sparse/1ms/batch=1000", 1_000, time.Millisecond, 200},
+		{"sparse/5ms/batch=1000", 1_000, 5 * time.Millisecond, 200},
+		{"sparse/1ms/batch=10000", 10_000, time.Millisecond, 200},
+		// Moderate rate: batches still close on the timer well below BatchSize.
+		{"steady/5ms/batch=1000", 1_000, 5 * time.Millisecond, 10_000},
+		// Control: batches fill, so reserved capacity is actually used.
+		{"full/100ms/batch=1000", 1_000, 100 * time.Millisecond, 50_000},
+	}
+
+	t.Logf("%-26s %-11s %-9s %-12s %s",
+		"scenario", "mean_batch", "batches", "allocs/item", "bytes/flush (pre-4.2 est)")
+
+	for _, c := range cases {
+		result := scenario.Run(scenario.Config{
+			Name:           c.name,
+			BatchSize:      c.batchSize,
+			BatchInterval:  c.window,
+			Arrival:        scenario.Steady(c.rate, time.Second),
+			Processor:      scenario.NoOpProcessor(),
+			LatenessBudget: time.Second,
+		})
+
+		require.False(t, result.TimedOut,
+			"%s: scenario timed out; allocation evidence is invalid", c.name)
+
+		// scenario.Item is the harness payload. BatchSize, not the adaptive estimate,
+		// is deliberately used here: this reproduces what a fixed reservation would
+		// have cost, which is the baseline 4.2 is compared against.
+		itemSize := float64(unsafe.Sizeof(scenario.Item{}))
+
+		wastePerFlush := float64(c.batchSize-int(result.MeanBatchSize)) * itemSize
+		if wastePerFlush < 0 {
+			wastePerFlush = 0
+		}
+
+		t.Logf("%-26s %-11.0f %-9d %-12.3f %.0f B",
+			c.name, result.MeanBatchSize, result.Batches,
+			result.AllocsPerItem, wastePerFlush)
+	}
+}
+
+// TestSparseWindowAllocationEvidenceRejectsTimedOutRun pins that evidence is not
+// logged when the processor did not finish. A timed-out run has incomplete batches
+// and allocation counts, so treating it as capacity evidence would be misleading.
+func TestSparseWindowAllocationEvidenceRejectsTimedOutRun(t *testing.T) {
+	t.Parallel()
+
+	result := scenario.Run(scenario.Config{
+		Name:               "timed-out-evidence",
+		BatchSize:          1,
+		BatchInterval:      time.Millisecond,
+		Arrival:            scenario.Steady(1_000, 100*time.Millisecond),
+		Processor:          scenario.FixedProcessor(100 * time.Millisecond),
+		CompletionDeadline: time.Millisecond,
+		LatenessBudget:     time.Second,
+	})
+
+	require.True(t, result.TimedOut, "the deliberately blocked run must time out")
+}