diff --git a/README.md b/README.md
index 3ce274d..5102450 100644
--- a/README.md
+++ b/README.md
@@ -220,6 +220,44 @@ if err := b.Shutdown(ctx); err != nil {
 `Shutdown` is resumable: a later call waits on the same drain rather than starting
 a new one, so an expired deadline never costs you accepted work.
 
+### Concurrent processing and ordering
+
+By default, Batcher processes **one batch at a time**. This guarantees that a
+processor is never invoked concurrently and that a single producer's batches are
+processed in publication order. Use this default when the processor holds
+unsynchronised state or when cross-batch ordering matters.
+
+A slow processor can make a small batch window ineffective at this setting: a 5ms
+window behind a 50ms processor is effectively bounded by the processor.
+
+The per-item worst case is additive, not a maximum. The interval timer starts when a
+batch takes its first item, so while the aggregator is blocked handing the previous
+batch to the busy worker no timer is running. An item arriving in that window waits
+the remaining processor time *plus* a full interval.
+
+When the processor is goroutine-safe and cross-batch ordering does not matter,
+explicitly opt into worker concurrency:
+
+```go
+b := batcher.New(
+    batcher.WithProcessor(processor),
+    batcher.WithConcurrency[Item](4),
+    batcher.WithoutOrderedProcessing[Item](), // required acknowledgement
+)
+```
+
+`WithoutOrderedProcessing` is deliberately required. `WithConcurrency(n > 1)`
+without it panics at construction, because concurrent processing gives up two
+properties callers may rely on:
+
+- batches may start, interleave, and complete in any order;
+- `processor` may be invoked concurrently, so it must be goroutine-safe.
+
+Items retain publication order **within** each batch at every concurrency. The
+worker dispatch is unbuffered, so `WithMaxQueueSize` continues to bound queued
+work; concurrency trades larger batches for more downstream calls, not an
+unbounded hidden dispatch queue.
+
 ### Back-pressure and rejection
 
 By default the queue is unbounded, which absorbs bursts well but turns a sustained
diff --git a/docs/improvements/plan-perf.md b/docs/improvements/plan-perf.md
index d4243c8..eb96bc8 100644
--- a/docs/improvements/plan-perf.md
+++ b/docs/improvements/plan-perf.md
@@ -512,7 +512,7 @@ Scenario matrix: windows `500µs, 1, 2, 5, 10, 20, 50, 100ms` × arrivals
   | Sparse-window allocated bytes/flush, 4.2 trigger | > 2 KB/flush to justify work       |
   | 4.2 allocation regression ceiling, any scenario  | ≤ +2% allocated bytes              |
   | Goroutines per idle batcher after 2.2            | exactly 0                          |
-  | Goroutines per running n=1 batcher after 2.2     | exactly 1 (aggregator)             |
+  | Goroutines per running n=1 batcher after 2.2     | exactly 2 (aggregator + processor) |
   | Goroutines after `closed`                        | equal to pre-construction baseline |
 
   Latency percentiles are deliberately absent: they are reported, never gated.
@@ -708,8 +708,10 @@ close-to-terminate model is replaced here, atomically, in one PR.
   baseline. Removing the `chann` relay is expected to make `Add` *faster* than
   today's 335 ns/op (prototype: 136-168 ns/op); a regression here indicates the
   owned queue is mis-implemented.
-- **Goroutine assertions:** 0 per idle batcher, 1 per running `n=1` batcher, and
-  exact return to the pre-construction baseline after `closed`. No queue owns a
+- **Goroutine assertions:** 0 per unstarted batcher, 2 per running `n=1` batcher
+  (aggregator + serial processor), `1 + n` per running `n>1` batcher (aggregator
+  + workers), and exact return to the pre-construction baseline after `closed`.
+  Measured: n=1=2, n=2=3, n=4=5, n=8=9, all with zero leaked. No queue owns a
   goroutine.
 
 **Dependencies**: Milestones 2.1 and 1.3.
@@ -839,9 +841,10 @@ Prototype evidence, 5ms window and 50ms processor at 10,000 items/s:
 | decoupled, n=4                   | 124        | **9.6ms** |
 
 Naive decoupling at `n=1` is **2.5x worse**, because it adds a stage without
-adding capacity. Inline aggregation already self-tunes: items pool while the
-processor runs, so the next batch is larger. Therefore the `n=1` path is left
-inline and unchanged; only `n>1` introduces a worker pool.
+adding capacity. Phase 2 already has one unbuffered aggregation→processing handoff
+for behaviour preservation; the relevant constraint is that `n=1` adds **no
+additional worker-pool dispatch** and invokes the processor serially. Only `n>1`
+introduces worker-pool dispatch and real concurrent processing.
 
 ### Milestone 3.1 — Concurrency configuration and ordering contract
 
@@ -850,11 +853,10 @@ inline and unchanged; only `n>1` introduces a worker pool.
 - Add `WithConcurrency(n)`; set `DefaultConcurrency = 1`.
 - Add `WithoutOrderedProcessing()` as an acknowledgement-only gate.
 - Panic at construction when `n > 1` without the acknowledgement.
-- Keep the `n = 1` aggregation path **without a worker-handoff stage**. It is
-  not byte-for-byte identical after Phase 2 — it necessarily uses the new gate,
-  accounting, drain, recovery, and stats machinery — but it preserves the
-  important performance shape: aggregation invokes the processor inline and no
-  additional dispatch queue exists.
+- Keep the `n = 1` path without an **additional worker-pool dispatch**. Phase 2
+  already has one unbuffered aggregation→processing handoff; it is retained because
+  removing it changes observable latency. At `n=1` the processor remains serial
+  and no worker pool or extra dispatch queue is introduced.
 - Document FIFO-by-publication-order and processor mutual-exclusion guarantees
   for `n = 1`.
 
@@ -880,7 +882,7 @@ inline and unchanged; only `n>1` introduces a worker pool.
   *within* each batch.
 - **Bounded dispatch:** the aggregator-to-worker handoff is unbuffered, so
   backpressure remains at admission and total accepted work stays bounded by
-  `N + BatchSize + (n × BatchSize)`. An unbounded dispatch queue would silently
+  `N + (1 + n) × BatchSize`. An unbounded dispatch queue would silently
   void the `MaxQueueSize` contract.
 - Add `Stats().InFlight`, tracked per worker, so the snapshot separates queued,
   batch-held, and in-flight work, and so the drain waits for active workers.
@@ -893,11 +895,26 @@ inline and unchanged; only `n>1` introduces a worker pool.
 - **Semantic-difference matrix:** scenarios compare `n=1` and `n>1` timer
   cadence, batch-size distribution, admission blocking, pending depth, and
   flush timing under the same slow processor. Documentation labels the observed
-  differences contractual: n=1 aggregates inline; n>1 aggregates independently
-  until the unbuffered worker dispatch is unavailable; cross-batch ordering is
-  not guaranteed at n>1.
+  differences contractual: n=1 uses the existing serial aggregation→processor
+  handoff; n>1 aggregates independently until the **unbuffered** worker dispatch
+  is unavailable; cross-batch ordering is not guaranteed at n>1.
+
+  The implementation was measured at 10k items/s with a 50ms processor and a 5ms
+  window on **darwin/arm64, Apple M4 Pro, GOMAXPROCS=12, Go 1.26.5**: n=1 p50 70ms
+  / mean batch 476; n=2 p50 25ms / 244; n=4 p50 15ms / 123; n=8 p50 4ms / 62.
+
+  Treat the ratio between rows as the finding and the absolute values as this
+  host's. A re-run on the same machine under different load measured n=1 p50 120ms
+  and n=8 p50 54ms — the same direction, roughly 2x rather than 17x. Anything
+  quoting a single multiplier out of this table will be wrong somewhere else, which
+  is why the tests assert only the relative ordering (`n>1` p50 below `n=1` p50)
+  and never a millisecond threshold.
+
+  With a 100ms window and a 20ms processor (timer-bound rather than
+  processor-bound), n=1 and n=4 both measured ~70ms p50, which is the control
+  showing concurrency does not alter the batching rule.
 - Total accepted-but-not-terminal items never exceed
-  `N + BatchSize + (n × BatchSize) + PublishersInGate` under a blocked processor;
+  `N + (1 + n) × BatchSize + PublishersInGate` under a blocked processor;
   the test records `PublishersInGate` separately and demonstrates that its maximum
   equals the number of intentionally parked publisher goroutines.
 - **Worker-mode drain test:** shutdown while a full batch is in flight on a busy
@@ -910,13 +927,18 @@ inline and unchanged; only `n>1` introduces a worker pool.
 - Goroutine count returns to the pre-construction baseline after `closed`, for
   `n=1` and `n>1`, using this leak-check protocol: sample after `Shutdown`
   returns, retry up to 50 times at 20ms intervals to allow scheduler settle,
-  then require exact equality with the pre-construction count. No unexplained
-  allowance is permitted; any runtime-owned goroutine must be named in the test.
+  then require the count to be at most the pre-construction baseline. It is `<=`
+  rather than `==` because an unrelated parallel test can retire a goroutine during
+  the settle window, which would make an equality assertion flaky without
+  indicating a leak. A count *above* the baseline is the leak signal, and any
+  runtime-owned goroutine must still be named in the test.
 - **Per-batcher goroutine budget is asserted by a test table** covering `new`,
   `running` at `n=1`, `running` at `n>1`, `draining`, and `closed`. Because 2.2
   replaces both `chann` queues with goroutine-free owned queues, the achievable
-  budget is: `new` = 0; `running` at `n=1` = 1 (aggregator); `running` at `n>1`
-  = 1 + n; `closed` = 0. Any deviation must be explained in the test rather than
+  budget is: `new` = 0; `running` at `n=1` = 2 (aggregator + serial processor);
+  `running` at `n>1` = 1 + n; `closed` = 0. (An earlier prediction of 1 for `n=1`
+  was invalidated by Phase 2, which deliberately retains the separate serial
+  processor goroutine to preserve latency semantics.) Any deviation must be explained in the test rather than
   absorbed into an allowance. The drain coordinator runs on the caller's
   `Shutdown` goroutine and must not add a persistent goroutine; if an
   implementation needs one, the table and this budget must be updated with it
diff --git a/docs/improvements/thresholds.md b/docs/improvements/thresholds.md
index 5cb8983..61b4332 100644
--- a/docs/improvements/thresholds.md
+++ b/docs/improvements/thresholds.md
@@ -40,7 +40,7 @@ the blocking signal rather than timings.
 | `Add` allocations, unbounded path       | exactly 0                                    | `TestAddAllocatesNothingPerCall`                              |
 | 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          | exactly 2                                    | `TestGoroutineBudgetPerRunningBatcher`                        |
+| Goroutines per running batcher, `n=1`   | exactly 2                                    | `TestGoroutineBudgetPerRunningBatcher`                        |
 
 `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.
@@ -68,23 +68,16 @@ Compare with `benchstat` over `-count=10`. A single run is not evidence. This is
 
 ## Goroutine gates (blocking)
 
-These are the gates in force today. They must agree with the allocation table
-above, which enforces the same running-batcher count.
+These are the gates in force today. The n=1 row agrees with the allocation table
+above; `n>1` is now a real, acknowledged configuration and has its own worker-pool
+gate. Every count is enforced by a test, not merely documented.
 
-| Gate                               | Threshold                                 |
-| ---------------------------------- | ----------------------------------------- |
-| Goroutines per unstarted batcher   | exactly 0                                 |
-| Goroutines per running batcher     | exactly 2 (aggregator + serial processor) |
-| Goroutines after terminal `closed` | equal to pre-construction baseline        |
-
-Enforced by `TestGoroutineBudgetPerRunningBatcher`.
-
-Phase 3 introduces explicit worker concurrency and will replace the single
-running-batcher row with `1 + n` (aggregator plus workers). That row is
-deliberately absent here rather than stated as a gate, because a threshold for a
-configuration this code cannot express is not enforceable — and two rows claiming
-different counts for the same batcher is worse than one row that is merely
-incomplete.
+| Gate                               | Threshold                                 | Enforced by |
+| ---------------------------------- | ----------------------------------------- | ----------- |
+| Goroutines per unstarted batcher   | exactly 0                                 | `TestWorkerGoroutineBudget` |
+| Goroutines per running `n=1`       | exactly 2 (aggregator + serial processor) | `TestGoroutineBudgetPerRunningBatcher`, `TestWorkerGoroutineBudget` |
+| Goroutines per running `n>1`       | exactly 1 + n (aggregator + workers)      | `TestWorkerGoroutineBudget` |
+| Goroutines after terminal `closed` | at most the pre-construction baseline     | `TestGoroutineBudgetPerRunningBatcher`, `TestWorkerGoroutineBudget` |
 
 Current `main` owns 6 goroutines per batcher. Phase 2.1 removed `rill`
 (**measured 6 → 5**) and Phase 2.2 removed both `chann` relays and the input
@@ -105,7 +98,9 @@ draining into aggregation regressed sequential `Add` by 39-50% because the
 
 Both earlier estimates assumed aggregation and processing could share one
 goroutine. They cannot without changing observable latency, which is why the
-enforced count is 2 rather than 1.
+enforced count is 2 at `n=1`. Phase 3 owns the worker model on top of that base:
+`n=1` is 2 goroutines and `n>1` is `1 + n`, measured as n=2→3, n=4→5, n=8→9 with
+zero leaked.
 
 ## Conditional gates (Milestone 4.2 only)
 
diff --git a/pkg/batcher/batcher.go b/pkg/batcher/batcher.go
index 25fd33c..4c04cc7 100644
--- a/pkg/batcher/batcher.go
+++ b/pkg/batcher/batcher.go
@@ -3,6 +3,7 @@ package batcher
 import (
 	"context"
 	"errors"
+	"fmt"
 	"runtime/debug"
 	"sync"
 	"time"
@@ -31,6 +32,10 @@ type Config[T any] struct {
 	CloseGrace      time.Duration
 	ErrorBufferSize int
 	ProcessorFunc   Processor[T]
+
+	// UnorderedProcessingAcknowledged records that the caller accepted the
+	// ordering trade required for Concurrency > 1. See WithoutOrderedProcessing.
+	UnorderedProcessingAcknowledged bool
 }
 
 type Batcher[T any] struct {
@@ -86,6 +91,8 @@ func New[T any](options ...Option[T]) *Batcher[T] {
 		option(b)
 	}
 
+	b.validateConfig()
+
 	b.input = newQueue[T](b.config.MaxQueueSize)
 	b.errorsChan = make(chan error, b.config.ErrorBufferSize)
 
@@ -96,6 +103,25 @@ func New[T any](options ...Option[T]) *Batcher[T] {
 	return b
 }
 
+// validateConfig rejects configurations whose guarantees contradict each other.
+//
+// This panics rather than returning an error because New has never been able to
+// fail, and adding an error return would break every existing call site for a
+// mistake that is always a programming error: the combination is fixed at
+// construction, so it is caught by the first test run rather than in production.
+func (b *Batcher[T]) validateConfig() {
+	if b.config.Concurrency > 1 && !b.config.UnorderedProcessingAcknowledged {
+		panic(fmt.Sprintf(
+			"batcher: WithConcurrency(%d) requires WithoutOrderedProcessing(). "+
+				"Concurrent processing gives up cross-batch ordering and lets the "+
+				"processor be invoked concurrently, so the processor must be "+
+				"goroutine-safe. Add WithoutOrderedProcessing() to acknowledge this, "+
+				"or keep WithConcurrency(1).",
+			b.config.Concurrency,
+		))
+	}
+}
+
 // Start begins processing. It is idempotent: repeated or concurrent calls start
 // exactly one processing lifecycle.
 //
@@ -197,6 +223,7 @@ func (b *Batcher[T]) Stats() Stats {
 		IntakePending:    b.counters.intakePending.Load(),
 		PublishersInGate: b.gate.inGate(),
 		Queued:           int64(b.input.length()),
+		InFlight:         b.counters.inFlight.Load(),
 		Accepted:         b.counters.accepted.Load(),
 		Completed:        b.counters.completed.Load(),
 		Failed:           b.counters.failed.Load(),
@@ -245,30 +272,55 @@ func (b *Batcher[T]) Errors() <-chan error {
 //   - Empty batches are never emitted.
 //   - Shutdown flushes the pending partial batch rather than discarding it.
 //
-// Aggregation runs here and processing runs in its own goroutine, connected by an
-// unbuffered channel. That separation is load-bearing: it lets the next batch
-// accumulate while the current one is being processed. Merging the two measurably
-// changes latency — with a 50ms processor at 10k items/s a merged loop made a 5ms
-// window faster than a 100ms one, inverting the documented baseline — and that is
-// a Phase 3 decision, not this milestone's.
+// Aggregation runs here and processing runs in Concurrency separate goroutines,
+// connected by an UNBUFFERED channel. Two properties follow, and both are
+// deliberate:
+//
+//   - The separation lets the next batch accumulate while the current one is being
+//     processed. Merging them measurably changes latency: with a 50ms processor at
+//     10k items/s a merged loop made a 5ms window faster than a 100ms one,
+//     inverting the documented baseline.
+//   - The channel is unbuffered, so back-pressure stays at admission. A buffered
+//     dispatch queue would silently void the MaxQueueSize contract by holding
+//     batches nobody counted, so accepted-but-unfinished work stays bounded by
+//     MaxQueueSize + (1 + Concurrency) × BatchSize + publishers in the gate. The
+//     leading batch is the one the aggregator holds after it leaves the queue; the
+//     Concurrency term is the batches inside processors.
+//
+// At Concurrency 1 there is exactly one worker, so the processor is never invoked
+// concurrently and batches are processed in publication order. Above 1 the caller
+// has acknowledged, via WithoutOrderedProcessing, that cross-batch ordering and
+// processor mutual exclusion are given up.
 func (b *Batcher[T]) run() {
 	defer close(b.stopped)
 
 	batches := make(chan []T)
 
-	processing := make(chan struct{})
+	workers := b.config.Concurrency
+	if workers < 1 {
+		workers = 1
+	}
 
-	go func() {
-		defer close(processing)
+	var processing sync.WaitGroup
 
-		for items := range batches {
-			b.process(items)
-		}
-	}()
+	processing.Add(workers)
+
+	for range workers {
+		go func() {
+			defer processing.Done()
+
+			for items := range batches {
+				b.process(items)
+			}
+		}()
+	}
 
+	// Closing the dispatch channel stops the workers, and waiting for them is what
+	// makes the drain complete: shutdown must not report success while a worker is
+	// still inside the processor.
 	defer func() {
 		close(batches)
-		<-processing
+		processing.Wait()
 	}()
 
 	var (
@@ -378,12 +430,16 @@ func (b *Batcher[T]) run() {
 // 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) {
+	b.counters.inFlight.Add(int64(len(items)))
+
 	outcome := outcomeCompleted
 
 	defer func() {
 		b.counters.terminal(outcome, len(items))
 	}()
 
+	defer b.counters.inFlight.Add(int64(-len(items)))
+
 	defer func() {
 		recovered := recover()
 		if recovered == nil {
diff --git a/pkg/batcher/batcher_test.go b/pkg/batcher/batcher_test.go
index 9a07bac..60c3e35 100644
--- a/pkg/batcher/batcher_test.go
+++ b/pkg/batcher/batcher_test.go
@@ -16,6 +16,8 @@ func TestCanCreateBatcherWithDefaultConfig(t *testing.T) {
 	// ACT
 	b := batcher.New[test.BatchItem]()
 
+	t.Cleanup(func() { _ = b.Close() })
+
 	var (
 		expected batcher.Processor[test.BatchItem]
 		actual   batcher.Processor[test.BatchItem]
@@ -44,6 +46,8 @@ func TestCanCreateBatcher(t *testing.T) {
 		batcher.WithBatchInterval[test.BatchItem](1*time.Second),
 	)
 
+	t.Cleanup(func() { _ = b.Close() })
+
 	// ASSERT
 	require.NotNil(t, b)
 	require.Equal(t, 1000, b.Config().BatchSize)
@@ -54,6 +58,8 @@ func TestCanAddItemsToBatch(t *testing.T) {
 	// ARRANGE
 	b := batcher.New[test.BatchItem]()
 
+	t.Cleanup(func() { _ = b.Close() })
+
 	// ACT
 	b.Add(test.BatchItem{})
 
@@ -65,6 +71,8 @@ func TestCanProcessItemsWithNoOp(t *testing.T) {
 	// ARRANGE
 	b := batcher.New[test.BatchItem]()
 
+	t.Cleanup(func() { _ = b.Close() })
+
 	// ACT
 	for i := 0; i < 1000; i++ {
 		b.Add(test.BatchItem{Key: fmt.Sprintf("key_%d", i)})
@@ -89,6 +97,8 @@ func TestCanProcessItemsWithCustomProcessor(t *testing.T) {
 		batcher.WithBatchInterval[test.BatchItem](1*time.Millisecond),
 	)
 
+	t.Cleanup(func() { _ = b.Close() })
+
 	// ACT
 	for i := 0; i < 1000; i++ {
 		b.Add(test.BatchItem{Key: fmt.Sprintf("key_%d", i)})
@@ -113,6 +123,8 @@ func TestCanProcessItemsWithStructProcessor(t *testing.T) {
 		batcher.WithBatchInterval[test.BatchItem](1*time.Millisecond),
 	)
 
+	t.Cleanup(func() { _ = b.Close() })
+
 	// ACT
 	for i := 0; i < 1000; i++ {
 		b.Add(test.BatchItem{Key: fmt.Sprintf("key_%d", i)})
@@ -138,6 +150,8 @@ func TestCanAddManyMoreItemsThanBatchSize(t *testing.T) {
 		batcher.WithBatchInterval[test.BatchItem](1*time.Millisecond),
 	)
 
+	t.Cleanup(func() { _ = b.Close() })
+
 	// ACT
 	for i := 0; i < 100000; i++ {
 		b.Add(test.BatchItem{Key: fmt.Sprintf("key_%d", i)})
@@ -248,6 +262,8 @@ func TestProcessesEntireBatchesIfTimerHasNotExpired(t *testing.T) {
 		}),
 	)
 
+	t.Cleanup(func() { _ = b.Close() })
+
 	// ACT
 	for range 10 {
 		b.Add(test.BatchItem{})
diff --git a/pkg/batcher/concurrency_test.go b/pkg/batcher/concurrency_test.go
new file mode 100644
index 0000000..00d3913
--- /dev/null
+++ b/pkg/batcher/concurrency_test.go
@@ -0,0 +1,314 @@
+package batcher_test
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/NSXBet/batcher/pkg/batcher"
+	"github.com/stretchr/testify/require"
+)
+
+// Concurrency configuration and the n=1 ordering contract.
+//
+// These tests pin what a caller is entitled to assume at the default
+// concurrency, and pin that raising concurrency cannot happen by accident. The
+// guarantees are only worth documenting if something enforces them, because they
+// currently hold by construction and a later refactor could remove them silently.
+
+// TestDefaultConcurrencyIsOne pins the default.
+//
+// It was 3 before this milestone, but read nowhere, so observed parallelism was
+// always 1. Changing the constant to 1 keeps existing callers on the behaviour
+// they already had; shipping 3 as a live value would have parallelised every
+// current user's processor without asking.
+func TestDefaultConcurrencyIsOne(t *testing.T) {
+	t.Parallel()
+
+	require.Equal(t, 1, batcher.DefaultConcurrency)
+
+	b := batcher.New(batcher.WithProcessor(batcher.NoOpProcessor[int]))
+
+	t.Cleanup(func() { _ = b.Close() })
+
+	require.Equal(t, 1, b.Config().Concurrency,
+		"a batcher built with no options must be serial")
+}
+
+// TestConcurrencyAboveOneRequiresAcknowledgement pins the gate.
+//
+// The acknowledgement carries no behaviour of its own; it exists so that giving
+// up ordering and processor mutual exclusion is stated in the code that does it.
+// Failing at construction means the mistake surfaces on the first test run rather
+// than as a rare production data race.
+func TestConcurrencyAboveOneRequiresAcknowledgement(t *testing.T) {
+	t.Parallel()
+
+	require.PanicsWithValue(t,
+		"batcher: WithConcurrency(3) requires WithoutOrderedProcessing(). "+
+			"Concurrent processing gives up cross-batch ordering and lets the "+
+			"processor be invoked concurrently, so the processor must be "+
+			"goroutine-safe. Add WithoutOrderedProcessing() to acknowledge this, "+
+			"or keep WithConcurrency(1).",
+		func() {
+			batcher.New(
+				batcher.WithProcessor(batcher.NoOpProcessor[int]),
+				batcher.WithConcurrency[int](3),
+			)
+		},
+		"WithConcurrency(n>1) without the acknowledgement must panic at construction")
+}
+
+// TestConcurrencyOneNeedsNoAcknowledgement pins that the gate does not get in the
+// way of the safe configuration.
+func TestConcurrencyOneNeedsNoAcknowledgement(t *testing.T) {
+	t.Parallel()
+
+	require.NotPanics(t, func() {
+		b := batcher.New(
+			batcher.WithProcessor(batcher.NoOpProcessor[int]),
+			batcher.WithConcurrency[int](1),
+		)
+
+		_ = b.Close()
+	}, "WithConcurrency(1) surrenders nothing, so it must not require the gate")
+}
+
+// TestAcknowledgedConcurrencyIsAccepted pins that the acknowledged combination
+// constructs. Milestone 3.2 makes n>1 actually parallel; here it only has to be
+// a legal configuration.
+func TestAcknowledgedConcurrencyIsAccepted(t *testing.T) {
+	t.Parallel()
+
+	require.NotPanics(t, func() {
+		b := batcher.New(
+			batcher.WithProcessor(batcher.NoOpProcessor[int]),
+			batcher.WithConcurrency[int](4),
+			batcher.WithoutOrderedProcessing[int](),
+		)
+
+		t.Cleanup(func() { _ = b.Close() })
+
+		require.Equal(t, 4, b.Config().Concurrency)
+	})
+}
+
+// TestConcurrencyIsClampedToAtLeastOne pins that a nonsensical value degrades to
+// the safe default rather than producing a batcher with no processing capacity.
+func TestConcurrencyIsClampedToAtLeastOne(t *testing.T) {
+	t.Parallel()
+
+	for _, n := range []int{0, -1, -100} {
+		b := batcher.New(
+			batcher.WithProcessor(batcher.NoOpProcessor[int]),
+			batcher.WithConcurrency[int](n),
+		)
+
+		require.Equal(t, 1, b.Config().Concurrency,
+			"WithConcurrency(%d) must clamp to 1, not disable processing", n)
+
+		require.NoError(t, b.Close())
+	}
+}
+
+// TestSerialProcessingPreservesPublicationOrder pins FIFO at n=1 across all three
+// flush triggers in one run: size-triggered batches, a timer-triggered batch, and
+// the final batch flushed by shutdown.
+//
+// Ordering is defined by publication order from a single producer. That is the
+// authoritative point because concurrent producers can reserve before they
+// publish, so reservation order and queue order are not the same thing.
+func TestSerialProcessingPreservesPublicationOrder(t *testing.T) {
+	t.Parallel()
+
+	const (
+		batchSize = 4
+		// Two full batches, then a gap for a timer flush, then a partial batch that
+		// only shutdown can flush.
+		firstBurst  = batchSize * 2
+		timerBatch  = 3
+		finalBatch  = 2
+		totalItems  = firstBurst + timerBatch + finalBatch
+		interval    = 40 * time.Millisecond
+		flushMargin = 4 * interval
+	)
+
+	var (
+		mu       sync.Mutex
+		observed []string
+	)
+
+	b := batcher.New(
+		batcher.WithBatchSize[string](batchSize),
+		batcher.WithBatchInterval[string](interval),
+		batcher.WithProcessor(func(items []string) error {
+			mu.Lock()
+			observed = append(observed, items...)
+			mu.Unlock()
+
+			return nil
+		}),
+	)
+
+	expected := make([]string, 0, totalItems)
+
+	publish := func(count int) {
+		for range count {
+			key := fmt.Sprintf("item-%03d", len(expected))
+			expected = append(expected, key)
+			b.Add(key)
+		}
+	}
+
+	publish(firstBurst)
+
+	// Let the size-triggered batches drain, then publish a batch that can only
+	// leave on the timer.
+	require.NoError(t, b.Join(5*time.Second))
+	publish(timerBatch)
+
+	time.Sleep(flushMargin)
+	require.NoError(t, b.Join(5*time.Second))
+
+	// The remainder is flushed by shutdown.
+	publish(finalBatch)
+	require.NoError(t, b.Close())
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	require.Equal(t, expected, observed,
+		"at n=1 a single producer's items must be processed in publication order "+
+			"across size, timer, and shutdown flushes")
+}
+
+// TestSerialProcessingNeverInvokesProcessorConcurrently pins processor mutual
+// exclusion at n=1, using a detector inside the processor rather than inferring
+// it from timing.
+//
+// This is the guarantee that lets a processor keep unsynchronised state, so it
+// must be enforced rather than assumed.
+func TestSerialProcessingNeverInvokesProcessorConcurrently(t *testing.T) {
+	t.Parallel()
+
+	var (
+		active     atomic.Int64
+		maxActive  atomic.Int64
+		invocation atomic.Int64
+	)
+
+	b := batcher.New(
+		batcher.WithBatchSize[int](1), // one batch per item: maximum opportunity to overlap
+		batcher.WithBatchInterval[int](time.Millisecond),
+		batcher.WithProcessor(func([]int) error {
+			current := active.Add(1)
+
+			for {
+				observed := maxActive.Load()
+				if current <= observed || maxActive.CompareAndSwap(observed, current) {
+					break
+				}
+			}
+
+			// Hold the processor open so any concurrent invocation would overlap.
+			time.Sleep(time.Millisecond)
+
+			invocation.Add(1)
+			active.Add(-1)
+
+			return nil
+		}),
+	)
+
+	const items = 60
+
+	for i := range items {
+		b.Add(i)
+	}
+
+	require.NoError(t, b.Close())
+
+	require.Equal(t, int64(items), invocation.Load(),
+		"every item must be processed")
+	require.Equal(t, int64(1), maxActive.Load(),
+		"at n=1 the processor must never be invoked concurrently")
+}
+
+// TestConcurrentProducersPreservePerProducerOrder pins the ordering guarantee that
+// actually holds with several producers, and deliberately does not claim more.
+//
+// Go does not order concurrent channel sends, so there is no total order across
+// producers to preserve. Asserting one would encode a guarantee the library cannot
+// make and would fail intermittently. What must hold is that each producer's own
+// subsequence stays in order.
+func TestConcurrentProducersPreservePerProducerOrder(t *testing.T) {
+	t.Parallel()
+
+	const (
+		producers   = 4
+		perProducer = 50
+	)
+
+	var (
+		mu       sync.Mutex
+		observed []string
+	)
+
+	b := batcher.New(
+		batcher.WithBatchSize[string](8),
+		batcher.WithBatchInterval[string](2*time.Millisecond),
+		batcher.WithProcessor(func(items []string) error {
+			mu.Lock()
+			observed = append(observed, items...)
+			mu.Unlock()
+
+			return nil
+		}),
+	)
+
+	var wg sync.WaitGroup
+
+	for p := range producers {
+		wg.Add(1)
+
+		go func(producer int) {
+			defer wg.Done()
+
+			for i := range perProducer {
+				b.Add(fmt.Sprintf("p%d-%03d", producer, i))
+			}
+		}(p)
+	}
+
+	wg.Wait()
+
+	require.NoError(t, b.Close())
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	require.Len(t, observed, producers*perProducer, "no item may be lost")
+
+	// Extract each producer's subsequence in observed order and require it to be
+	// ascending. Cross-producer interleaving is unconstrained by design.
+	perProducerSeen := make(map[string][]string, producers)
+
+	for _, key := range observed {
+		producer := key[:2]
+		perProducerSeen[producer] = append(perProducerSeen[producer], key)
+	}
+
+	require.Len(t, perProducerSeen, producers)
+
+	for producer, seen := range perProducerSeen {
+		expected := make([]string, perProducer)
+		for i := range expected {
+			expected[i] = fmt.Sprintf("%s-%03d", producer, i)
+		}
+
+		require.Equal(t, expected, seen,
+			"producer %s: its own items must stay in publication order", producer)
+	}
+}
diff --git a/pkg/batcher/constants.go b/pkg/batcher/constants.go
index 80110ea..f898b99 100644
--- a/pkg/batcher/constants.go
+++ b/pkg/batcher/constants.go
@@ -9,8 +9,17 @@ const (
 	// DefaultBatchInterval is the default batch interval.
 	DefaultBatchInterval = 1 * time.Second
 
-	// DefaultConcurrency is the default concurrency.
-	DefaultConcurrency = 3
+	// DefaultConcurrency is how many batches are processed at once by default.
+	//
+	// It is 1, not 3. The previous value of 3 was advertised but read nowhere, so
+	// observed parallelism was always 1; changing the default to match reality
+	// keeps existing users on the behaviour they already had. Raising it silently
+	// would have parallelised every current user's processor without asking, which
+	// is unsafe for a processor holding unsynchronised state.
+	//
+	// Values above 1 require WithoutOrderedProcessing, because they give up
+	// cross-batch ordering and processor mutual exclusion.
+	DefaultConcurrency = 1
 
 	// DefaultCloseGrace is how long Close waits for the drain to finish before
 	// reporting that it is incomplete. The drain continues regardless.
diff --git a/pkg/batcher/options.go b/pkg/batcher/options.go
index 4771f9e..fef5324 100644
--- a/pkg/batcher/options.go
+++ b/pkg/batcher/options.go
@@ -99,3 +99,52 @@ func WithErrorBufferSize[T any](size int) Option[T] {
 		b.config.ErrorBufferSize = size
 	}
 }
+
+// WithConcurrency sets how many batches may be processed at once.
+//
+// The default is 1, which guarantees the processor is never invoked concurrently
+// and that batches are processed in publication order.
+//
+// Values above 1 require WithoutOrderedProcessing as well. Construction panics
+// otherwise: raising concurrency silently discards two guarantees callers may be
+// relying on, so the trade has to be acknowledged in the code that makes it, not
+// discovered in production. See WithoutOrderedProcessing for what is given up.
+//
+// Concurrency above 1 is what stops a slow processor from bounding the effective
+// batch interval. At n = 1 the steady-state flush interval is
+// max(BatchInterval, processor duration), because one batch must finish before the
+// next can start.
+//
+// Per-item worst case is worse than that maximum, and additive rather than a
+// maximum. While the aggregator is blocked handing a batch to the busy worker, the
+// interval timer is not running: it is armed only when the next batch takes its
+// first item. An item that arrives during the blocked window therefore waits the
+// remaining processor time *plus* a full BatchInterval.
+func WithConcurrency[T any](concurrency int) Option[T] {
+	return func(b *Batcher[T]) {
+		if concurrency < 1 {
+			concurrency = 1
+		}
+
+		b.config.Concurrency = concurrency
+	}
+}
+
+// WithoutOrderedProcessing acknowledges that concurrent processing gives up
+// ordering guarantees. It changes no behaviour on its own.
+//
+// It exists purely so that enabling concurrency is explicit about its cost. With
+// WithConcurrency(n) for n > 1:
+//
+//   - batches may start, interleave, and complete in any order;
+//   - the processor may be invoked concurrently, so it must be goroutine-safe.
+//
+// What is retained at any concurrency:
+//
+//   - items keep publication order *within* each batch;
+//   - every accepted item is processed exactly once by the pool itself.
+func WithoutOrderedProcessing[T any]() Option[T] {
+	return func(b *Batcher[T]) {
+		b.config.UnorderedProcessingAcknowledged = true
+	}
+}
diff --git a/pkg/batcher/stats.go b/pkg/batcher/stats.go
index d455e7b..a346ab9 100644
--- a/pkg/batcher/stats.go
+++ b/pkg/batcher/stats.go
@@ -41,6 +41,11 @@ type Stats struct {
 	// the aggregator. This is the queue depth to alert on.
 	Queued 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.
+	InFlight int64
+
 	// Accepted counts successful publications. A rejected or cancelled enqueue
 	// never increments it.
 	Accepted uint64
@@ -74,6 +79,7 @@ type counters struct {
 	pending       atomic.Int64
 	intakePending atomic.Int64
 	accepted      atomic.Uint64
+	inFlight      atomic.Int64
 	completed     atomic.Uint64
 	failed        atomic.Uint64
 	panicked      atomic.Uint64
diff --git a/pkg/batcher/worker_test.go b/pkg/batcher/worker_test.go
new file mode 100644
index 0000000..4c64f7c
--- /dev/null
+++ b/pkg/batcher/worker_test.go
@@ -0,0 +1,522 @@
+package batcher_test
+
+import (
+	"context"
+	"errors"
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/NSXBet/batcher/pkg/batcher"
+	"github.com/stretchr/testify/require"
+)
+
+// Worker pool behaviour at acknowledged concurrency.
+//
+// These tests cover the properties that make concurrency safe rather than merely
+// faster: that it actually removes head-of-line blocking, that it does not void
+// the queue bound, that shutdown waits for active workers, and that workers do not
+// leak.
+
+func unorderedBatcher[T any](workers int, opts ...batcher.Option[T]) []batcher.Option[T] {
+	return append(opts,
+		batcher.WithConcurrency[T](workers),
+		batcher.WithoutOrderedProcessing[T](),
+	)
+}
+
+// TestConcurrentWorkersProcessBatchesInParallel is the point of the whole phase.
+//
+// At n=1 a slow processor bounds the effective batch interval, because one batch
+// must finish before the next can start. This asserts that n>1 actually overlaps
+// processor calls rather than merely being configured to.
+func TestConcurrentWorkersProcessBatchesInParallel(t *testing.T) {
+	t.Parallel()
+
+	const (
+		workers   = 4
+		batchSize = 1
+		items     = 8
+	)
+
+	var (
+		active    atomic.Int64
+		maxActive atomic.Int64
+		done      sync.WaitGroup
+	)
+
+	done.Add(items)
+
+	// Hold every worker inside the processor until all of them have arrived, so
+	// overlap is proven rather than inferred from timing.
+	release := make(chan struct{})
+	arrived := make(chan struct{}, items)
+
+	b := batcher.New(unorderedBatcher[int](workers,
+		batcher.WithBatchSize[int](batchSize),
+		batcher.WithBatchInterval[int](time.Millisecond),
+		batcher.WithProcessor(func([]int) error {
+			defer done.Done()
+
+			current := active.Add(1)
+
+			for {
+				observed := maxActive.Load()
+				if current <= observed || maxActive.CompareAndSwap(observed, current) {
+					break
+				}
+			}
+
+			arrived <- struct{}{}
+			<-release
+
+			active.Add(-1)
+
+			return nil
+		}),
+	)...)
+
+	for i := range items {
+		b.Add(i)
+	}
+
+	// Wait for exactly `workers` concurrent invocations. If the pool were serial,
+	// only one would ever arrive and this would time out.
+	deadline := time.After(10 * time.Second)
+
+	for range workers {
+		select {
+		case <-arrived:
+		case <-deadline:
+			t.Fatalf("only %d concurrent processor invocations; expected %d",
+				active.Load(), workers)
+		}
+	}
+
+	require.Equal(t, int64(workers), maxActive.Load(),
+		"n=%d must invoke the processor %d times concurrently", workers, workers)
+
+	close(release)
+	done.Wait()
+
+	require.NoError(t, b.Close())
+
+	stats := b.Stats()
+	require.Equal(t, uint64(items), stats.Completed)
+	require.Zero(t, stats.Pending)
+	require.Zero(t, stats.InFlight, "no work may remain in flight after shutdown")
+}
+
+// TestConcurrencyPreservesOrderWithinEachBatch pins the guarantee that survives
+// concurrency. Cross-batch order is given up; within a batch, items keep
+// publication order, which is what lets a processor rely on its slice.
+func TestConcurrencyPreservesOrderWithinEachBatch(t *testing.T) {
+	t.Parallel()
+
+	const (
+		workers   = 4
+		batchSize = 10
+		batches   = 20
+	)
+
+	var (
+		mu        sync.Mutex
+		disorders int
+	)
+
+	b := batcher.New(unorderedBatcher[int](workers,
+		batcher.WithBatchSize[int](batchSize),
+		batcher.WithBatchInterval[int](5*time.Millisecond),
+		batcher.WithProcessor(func(items []int) error {
+			for i := 1; i < len(items); i++ {
+				if items[i] < items[i-1] {
+					mu.Lock()
+					disorders++
+					mu.Unlock()
+				}
+			}
+
+			return nil
+		}),
+	)...)
+
+	for i := range batchSize * batches {
+		b.Add(i)
+	}
+
+	require.NoError(t, b.Close())
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	require.Zero(t, disorders,
+		"items must retain publication order within each batch even at n>1")
+}
+
+// TestUnbufferedDispatchKeepsAcceptedWorkBounded pins the capacity contract.
+//
+// A buffered dispatch channel would hold batches nobody accounted for, silently
+// voiding MaxQueueSize. The bound is
+// MaxQueueSize + (1 + n) × BatchSize + PublishersInGate -- the leading batch is the
+// aggregator's held batch, the n term is batches inside processors -- and the test
+// records PublishersInGate separately so the terms stay distinguishable.
+func TestUnbufferedDispatchKeepsAcceptedWorkBounded(t *testing.T) {
+	t.Parallel()
+
+	const (
+		workers      = 3
+		batchSize    = 4
+		maxQueueSize = 8
+		parked       = 5
+	)
+
+	// Released via t.Cleanup, guarded by sync.Once, so a failing assertion cannot
+	// strand the parked workers and publishers. t.FailNow exits the test goroutine,
+	// so anything left to the end of the function body would never run and a clean
+	// assertion failure would become a package-level timeout.
+	release := make(chan struct{})
+
+	var releaseOnce sync.Once
+
+	releaseAll := func() { releaseOnce.Do(func() { close(release) }) }
+
+	t.Cleanup(releaseAll)
+
+	b := batcher.New(unorderedBatcher[int](workers,
+		batcher.WithBatchSize[int](batchSize),
+		batcher.WithBatchInterval[int](time.Millisecond),
+		batcher.WithMaxQueueSize[int](maxQueueSize),
+		batcher.WithProcessor(func([]int) error {
+			<-release
+
+			return nil
+		}),
+	)...)
+
+	// Park publishers deliberately: each one blocks on a full bounded queue and is
+	// therefore a caller goroutine Batcher cannot bound, only report.
+	var wg sync.WaitGroup
+
+	for i := range parked {
+		wg.Add(1)
+
+		go func(v int) {
+			defer wg.Done()
+
+			_ = b.Enqueue(context.Background(), v)
+		}(i)
+	}
+
+	// Saturate until admission starts refusing, so the queue is genuinely full.
+	for attempt := range 1_000 {
+		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+		err := b.Enqueue(ctx, attempt)
+
+		cancel()
+
+		if err != nil {
+			break
+		}
+	}
+
+	t.Cleanup(func() { _ = b.Close() })
+
+	limit := int64(maxQueueSize + batchSize + workers*batchSize + parked)
+
+	maxObservedGate := int64(0)
+
+	for range 50 {
+		stats := b.Stats()
+
+		if stats.PublishersInGate > maxObservedGate {
+			maxObservedGate = stats.PublishersInGate
+		}
+
+		require.LessOrEqual(t, stats.Pending, limit,
+			"accepted-but-unfinished work must stay within "+
+				"MaxQueueSize + (1+n)*BatchSize + PublishersInGate = %d "+
+				"(pending=%d queued=%d inflight=%d gate=%d)",
+			limit, stats.Pending, stats.Queued, stats.InFlight, stats.PublishersInGate)
+
+		time.Sleep(2 * time.Millisecond)
+	}
+
+	require.LessOrEqual(t, maxObservedGate, int64(parked),
+		"publishers in the gate must not exceed the goroutines intentionally parked")
+
+	releaseAll()
+	wg.Wait()
+}
+
+// TestShutdownWaitsForActiveWorkers pins that a batch inside a worker is drained
+// rather than abandoned, which is the worker-mode version of the Phase 2 guarantee.
+//
+// This is also the case that distinguishes IntakePending from Pending: with a batch
+// dispatched to a busy worker and nothing left in the queue, an aggregator that
+// waited on Pending would block on a receive that can never arrive.
+func TestShutdownWaitsForActiveWorkers(t *testing.T) {
+	t.Parallel()
+
+	const (
+		workers   = 4
+		batchSize = 2
+		items     = 8
+	)
+
+	var processed atomic.Int64
+
+	started := make(chan struct{}, items)
+	release := make(chan struct{})
+
+	var releaseOnce sync.Once
+
+	releaseAll := func() { releaseOnce.Do(func() { close(release) }) }
+
+	// Registered before the assertions below: two of them call t.Fatalf, which would
+	// otherwise leave four workers and the Shutdown goroutine parked forever.
+	t.Cleanup(releaseAll)
+
+	b := batcher.New(unorderedBatcher[int](workers,
+		batcher.WithBatchSize[int](batchSize),
+		batcher.WithBatchInterval[int](time.Millisecond),
+		batcher.WithProcessor(func(batch []int) error {
+			started <- struct{}{}
+			<-release
+			processed.Add(int64(len(batch)))
+
+			return nil
+		}),
+	)...)
+
+	for i := range items {
+		b.Add(i)
+	}
+
+	// Wait until work is genuinely in flight before shutting down.
+	select {
+	case <-started:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the processor was never invoked")
+	}
+
+	require.Eventually(t, func() bool {
+		return b.Stats().InFlight >= batchSize
+	}, 5*time.Second, time.Millisecond,
+		"Stats().InFlight must expose work currently inside a worker")
+
+	shutdownDone := make(chan error, 1)
+
+	go func() { shutdownDone <- b.Shutdown(context.Background()) }()
+
+	// Shutdown must not complete while workers are held.
+	select {
+	case err := <-shutdownDone:
+		t.Fatalf("shutdown completed while workers were still active: %v", err)
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	releaseAll()
+
+	select {
+	case err := <-shutdownDone:
+		require.NoError(t, err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("shutdown never completed after workers were released")
+	}
+
+	stats := b.Stats()
+
+	require.Equal(t, int64(items), processed.Load(),
+		"every accepted item must be processed, not discarded")
+	require.Equal(t, uint64(items), stats.Accepted)
+	require.Equal(t, stats.Accepted, stats.Completed+stats.Failed+stats.Panicked,
+		"accounting invariant must hold at quiescence")
+	require.Zero(t, stats.Pending)
+	require.Zero(t, stats.IntakePending)
+	require.Zero(t, stats.InFlight)
+	require.True(t, b.IsClosed())
+}
+
+// TestPanicInWorkerDoesNotReduceConcurrency pins that recovery is scoped to the
+// batch, not the worker. A panic that killed its worker would silently degrade
+// throughput for the life of the process.
+func TestPanicInWorkerDoesNotReduceConcurrency(t *testing.T) {
+	t.Parallel()
+
+	const workers = 3
+
+	var calls atomic.Int64
+
+	b := batcher.New(unorderedBatcher[int](workers,
+		batcher.WithBatchSize[int](1),
+		batcher.WithBatchInterval[int](time.Millisecond),
+		batcher.WithProcessor(func([]int) error {
+			// Panic on the first several batches: enough to kill every worker if
+			// recovery were scoped to the loop instead of the batch.
+			if calls.Add(1) <= int64(workers) {
+				panic("poison batch")
+			}
+
+			return nil
+		}),
+	)...)
+
+	go func() {
+		for range b.Errors() {
+		}
+	}()
+
+	const items = 40
+
+	for i := range items {
+		b.Add(i)
+	}
+
+	require.NoError(t, b.Close())
+
+	stats := b.Stats()
+
+	require.Equal(t, uint64(workers), stats.Panicked)
+	require.Equal(t, uint64(items)-uint64(workers), stats.Completed,
+		"batches after the panics must still be processed by surviving workers")
+	require.Zero(t, stats.Pending, "a panic must not strand a drain obligation")
+}
+
+// TestWorkerGoroutineBudget pins the per-batcher goroutine cost across states.
+//
+// Callers create one batcher per tenant or key, so this is a real cost, and a leak
+// would accumulate one set per batcher for the life of the process.
+func TestWorkerGoroutineBudget(t *testing.T) {
+	// Not parallel: goroutine counting requires no other test starting batchers.
+	for _, workers := range []int{1, 2, 4} {
+		baseline := settledGoroutines()
+
+		b := batcher.New(unorderedBatcher[int](workers,
+			batcher.WithBatchSize[int](100),
+			batcher.WithBatchInterval[int](time.Hour), // stay idle
+			batcher.WithProcessor(batcher.NoOpProcessor[int]),
+		)...)
+
+		// aggregator + one goroutine per worker.
+		want := 1 + workers
+
+		running := 0
+
+		for range 100 {
+			running = runtime.NumGoroutine()
+
+			if running >= baseline+want {
+				break
+			}
+
+			time.Sleep(20 * time.Millisecond)
+		}
+
+		require.InDelta(t, float64(want), float64(running-baseline), 0.5,
+			"n=%d must own exactly %d goroutines (aggregator + %d workers); "+
+				"measured %d over baseline %d",
+			workers, want, workers, running-baseline, baseline)
+
+		require.NoError(t, b.Close())
+
+		settled := settledGoroutines()
+
+		require.LessOrEqual(t, settled, baseline,
+			"n=%d must return to the pre-construction goroutine count after Close "+
+				"(baseline %d, settled %d)", workers, baseline, settled)
+	}
+}
+
+// TestConcurrentWorkersUnderRace exercises worker completion against close, error
+// publication, and concurrent enqueue simultaneously. It exists to be run under
+// -race, where an unsynchronised worker interaction shows up as a reported race
+// rather than as a rare wrong answer.
+func TestConcurrentWorkersUnderRace(t *testing.T) {
+	t.Parallel()
+
+	const (
+		trials    = 40
+		workers   = 4
+		producers = 4
+		perActor  = 25
+	)
+
+	failure := errors.New("processor failed")
+
+	for trial := range trials {
+		b := batcher.New(unorderedBatcher[int](workers,
+			batcher.WithBatchSize[int](8),
+			batcher.WithBatchInterval[int](time.Millisecond),
+			batcher.WithProcessor(func(items []int) error {
+				// Alternate success and failure so error publication races worker
+				// completion and channel close.
+				if len(items)%2 == 0 {
+					return failure
+				}
+
+				return nil
+			}),
+		)...)
+
+		drained := make(chan struct{})
+
+		go func() {
+			defer close(drained)
+
+			for range b.Errors() {
+			}
+		}()
+
+		var wg sync.WaitGroup
+
+		for range producers {
+			wg.Add(1)
+
+			go func() {
+				defer wg.Done()
+
+				for i := range perActor {
+					b.Add(i)
+				}
+			}()
+		}
+
+		closed := make(chan error, 1)
+
+		go func() { closed <- b.Close() }()
+
+		wg.Wait()
+
+		require.NoError(t, <-closed, "trial %d", trial)
+		<-drained
+
+		stats := b.Stats()
+
+		require.Zero(t, stats.Pending, "trial %d", trial)
+		require.Zero(t, stats.InFlight, "trial %d", trial)
+		require.Equal(t, stats.Accepted, stats.Completed+stats.Failed+stats.Panicked,
+			"trial %d: accounting invariant must hold", trial)
+	}
+}
+
+// settledGoroutines waits for the goroutine count to stop changing, so a scheduler
+// that has not yet reaped finished goroutines is not mistaken for a leak.
+func settledGoroutines() int {
+	previous := -1
+
+	for range 100 {
+		runtime.GC()
+		time.Sleep(20 * time.Millisecond)
+
+		current := runtime.NumGoroutine()
+		if current == previous {
+			return current
+		}
+
+		previous = current
+	}
+
+	return previous
+}
diff --git a/test/scenario/concurrency_test.go b/test/scenario/concurrency_test.go
new file mode 100644
index 0000000..6d1cdd3
--- /dev/null
+++ b/test/scenario/concurrency_test.go
@@ -0,0 +1,260 @@
+package scenario_test
+
+import (
+	"os"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/NSXBet/batcher/pkg/batcher"
+	"github.com/NSXBet/batcher/test/scenario"
+	"github.com/stretchr/testify/require"
+)
+
+// concurrencyOptions builds the acknowledged worker-pool configuration. n=1 needs
+// no options, which keeps the comparison honest: the serial case is the library's
+// default rather than a specially configured mode.
+func concurrencyOptions(workers int) []batcher.Option[scenario.Item] {
+	if workers <= 1 {
+		return nil
+	}
+
+	return []batcher.Option[scenario.Item]{
+		batcher.WithConcurrency[scenario.Item](workers),
+		batcher.WithoutOrderedProcessing[scenario.Item](),
+	}
+}
+
+// TestConcurrencyRemovesSlowProcessorCoupling is the measurement the whole plan
+// exists to justify.
+//
+// At n=1 a batch must finish before the next can start, so the effective flush
+// interval is max(BatchInterval, processor duration) and a small window buys
+// nothing. This asserts that acknowledged concurrency actually breaks that
+// coupling under an open-loop load, rather than merely being configurable.
+//
+// The assertion is relative — n>1 must beat n=1 on the same scenario — because
+// absolute latency depends on the host, and a fixed millisecond threshold would
+// encode the machine it was written on.
+func TestConcurrencyRemovesSlowProcessorCoupling(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip("timing-sensitive; runs in the full suite only")
+	}
+
+	const (
+		window      = 5 * time.Millisecond
+		serviceTime = 50 * time.Millisecond
+		rate        = 10_000
+		duration    = 1500 * time.Millisecond
+	)
+
+	run := func(workers int) scenario.Result {
+		return scenario.Run(scenario.Config{
+			Name:           "coupling",
+			BatchSize:      100_000, // never size-triggered: isolate the timer
+			BatchInterval:  window,
+			Arrival:        scenario.Steady(rate, duration),
+			Processor:      scenario.FixedProcessor(serviceTime),
+			BatcherOptions: concurrencyOptions(workers),
+			Producers:      runtime.NumCPU(),
+			LatenessBudget: time.Second,
+		})
+	}
+
+	serial := run(1)
+	concurrent := run(8)
+
+	t.Logf("n=1  p50=%-10s p99=%-10s mean_batch=%-7.0f downstream/s=%.0f",
+		serial.EndToEnd.P50, serial.EndToEnd.P99,
+		serial.MeanBatchSize, serial.DownstreamPerSec)
+	t.Logf("n=8  p50=%-10s p99=%-10s mean_batch=%-7.0f downstream/s=%.0f",
+		concurrent.EndToEnd.P50, concurrent.EndToEnd.P99,
+		concurrent.MeanBatchSize, concurrent.DownstreamPerSec)
+
+	require.False(t, serial.TimedOut, "serial run must complete")
+	require.False(t, concurrent.TimedOut, "concurrent run must complete")
+
+	require.Less(t, concurrent.EndToEnd.P50, serial.EndToEnd.P50,
+		"with a %s processor, concurrency must reduce p50 latency: "+
+			"at n=1 the effective interval is bounded by the processor, not the window",
+		serviceTime)
+
+	// Concurrency adds capacity rather than changing the batching rule, so more
+	// batches leave per second and each is correspondingly smaller.
+	require.Greater(t, concurrent.DownstreamPerSec, serial.DownstreamPerSec,
+		"concurrency must raise the flush rate")
+	require.Less(t, concurrent.MeanBatchSize, serial.MeanBatchSize,
+		"a higher flush rate means less pooling per batch")
+}
+
+// TestSemanticDifferenceMatrix records how n=1 and n>1 differ under an identical
+// slow processor, which Milestone 3.2 requires as documentation rather than as a
+// pass/fail gate.
+//
+// Only the load-bearing contract is asserted: latency must improve monotonically
+// enough that n=8 beats n=1. The rest is reported, because batch-size distribution
+// and admission blocking depend on host scheduling.
+func TestSemanticDifferenceMatrix(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip("timing-sensitive; runs in the full suite only")
+	}
+
+	const (
+		window      = 5 * time.Millisecond
+		serviceTime = 20 * time.Millisecond
+		rate        = 10_000
+		duration    = time.Second
+	)
+
+	type row struct {
+		workers int
+		result  scenario.Result
+	}
+
+	rows := make([]row, 0, 4)
+
+	for _, workers := range []int{1, 2, 4, 8} {
+		rows = append(rows, row{
+			workers: workers,
+			result: scenario.Run(scenario.Config{
+				Name:           "matrix",
+				BatchSize:      100_000,
+				BatchInterval:  window,
+				Arrival:        scenario.Steady(rate, duration),
+				Processor:      scenario.FixedProcessor(serviceTime),
+				BatcherOptions: concurrencyOptions(workers),
+				Producers:      runtime.NumCPU(),
+				LatenessBudget: time.Second,
+			}),
+		})
+	}
+
+	t.Logf("window=%s processor=%s arrival=%d/s", window, serviceTime, rate)
+	t.Logf("%-4s %-11s %-11s %-11s %-12s %-12s %s",
+		"n", "p50", "p99", "max", "mean_batch", "calls/s", "admission_p99")
+
+	for _, r := range rows {
+		t.Logf("%-4d %-11s %-11s %-11s %-12.0f %-12.0f %s",
+			r.workers,
+			r.result.EndToEnd.P50.Round(time.Microsecond),
+			r.result.EndToEnd.P99.Round(time.Microsecond),
+			r.result.EndToEnd.Max.Round(time.Microsecond),
+			r.result.MeanBatchSize,
+			r.result.DownstreamPerSec,
+			r.result.AdmissionBlocking.P99.Round(time.Microsecond),
+		)
+
+		require.False(t, r.result.TimedOut, "n=%d run must complete", r.workers)
+		require.Zero(t, r.result.RejectedCount,
+			"n=%d: unbounded admission must accept everything", r.workers)
+	}
+
+	serial, highest := rows[0].result, rows[len(rows)-1].result
+
+	require.Less(t, highest.EndToEnd.P50, serial.EndToEnd.P50,
+		"n=%d must reduce p50 versus serial processing", rows[len(rows)-1].workers)
+}
+
+// TestConcurrencyDoesNotHelpWhenNotProcessorBound is the control for the claim
+// above.
+//
+// If concurrency appeared to improve a window that was never processor-bound, the
+// measurement would be suspect: the improvement must come from removing the
+// processor bottleneck, not from perturbing the timer. With a 100ms window and a
+// 20ms processor there is no bottleneck to remove, so latency should be
+// essentially unchanged.
+func TestConcurrencyDoesNotHelpWhenNotProcessorBound(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip("timing-sensitive; runs in the full suite only")
+	}
+
+	const (
+		window      = 100 * time.Millisecond
+		serviceTime = 20 * time.Millisecond
+		rate        = 10_000
+		duration    = time.Second
+	)
+
+	run := func(workers int) scenario.Result {
+		return scenario.Run(scenario.Config{
+			Name:           "not-bound",
+			BatchSize:      100_000,
+			BatchInterval:  window,
+			Arrival:        scenario.Steady(rate, duration),
+			Processor:      scenario.FixedProcessor(serviceTime),
+			BatcherOptions: concurrencyOptions(workers),
+			Producers:      runtime.NumCPU(),
+			LatenessBudget: time.Second,
+		})
+	}
+
+	serial := run(1)
+	concurrent := run(4)
+
+	t.Logf("n=1  p50=%s mean_batch=%.0f", serial.EndToEnd.P50, serial.MeanBatchSize)
+	t.Logf("n=4  p50=%s mean_batch=%.0f", concurrent.EndToEnd.P50, concurrent.MeanBatchSize)
+
+	require.False(t, serial.TimedOut, "serial run must complete")
+	require.False(t, concurrent.TimedOut, "concurrent run must complete")
+
+	// The window dominates, so both should sit near it. Allow a generous margin:
+	// the point is that concurrency does not change the regime, not that timing is
+	// exact on a shared runner.
+	require.InDelta(t, serial.EndToEnd.P50.Seconds(), concurrent.EndToEnd.P50.Seconds(),
+		(window / 2).Seconds(),
+		"when the window already exceeds the processor, concurrency must not change "+
+			"the latency regime; an improvement here would mean the batching rule "+
+			"changed rather than the bottleneck being removed")
+}
+
+// TestConcurrencyMatrixReport prints the full concurrency sweep as an artifact for
+// the Phase 5 default-window decision. Opt-in, because it is measurement rather
+// than verification.
+func TestConcurrencyMatrixReport(t *testing.T) {
+	if os.Getenv("SCENARIO_MATRIX") == "" {
+		t.Skip("set SCENARIO_MATRIX=1 to run the concurrency report")
+	}
+
+	var results []scenario.Result
+
+	for _, workers := range []int{1, 2, 4, 8} {
+		for _, window := range []time.Duration{
+			time.Millisecond,
+			5 * time.Millisecond,
+			10 * time.Millisecond,
+			100 * time.Millisecond,
+		} {
+			result := scenario.Run(scenario.Config{
+				Name:           "n=" + itoa(workers),
+				BatchSize:      1_000,
+				BatchInterval:  window,
+				Arrival:        scenario.Steady(10_000, 2*time.Second),
+				Processor:      scenario.FixedProcessor(20 * time.Millisecond),
+				BatcherOptions: concurrencyOptions(workers),
+				Producers:      runtime.NumCPU(),
+				Warmup:         200 * time.Millisecond,
+				Seed:           1,
+				LatenessBudget: 2 * time.Millisecond,
+			})
+
+			// A run that gave up waiting has not measured steady-state behaviour, so
+			// reporting it as evidence would be misleading.
+			if result.TimedOut {
+				t.Errorf("n=%d window=%s timed out; excluded from the report",
+					workers, window)
+
+				continue
+			}
+
+			results = append(results, result)
+		}
+	}
+
+	require.NoError(t, scenario.WriteReport(os.Stdout, results))
+}
diff --git a/test/scenario/matrix_test.go b/test/scenario/matrix_test.go
index 76d71d9..88f954e 100644
--- a/test/scenario/matrix_test.go
+++ b/test/scenario/matrix_test.go
@@ -2,9 +2,11 @@ package scenario_test
 
 import (
 	"os"
+	"runtime"
 	"testing"
 	"time"
 
+	"github.com/NSXBet/batcher/pkg/batcher"
 	"github.com/NSXBet/batcher/test/scenario"
 )
 
@@ -79,6 +81,43 @@ func TestScenarioMatrix(t *testing.T) {
 		}
 	}
 
+	// Phase 3 comparison: same slow downstream, serial versus acknowledged worker
+	// pool. This is the evidence used to decide whether a 5-10ms window is worth
+	// recommending for a service with processor-bound batches.
+	for _, workers := range []int{1, 2, 4, 8} {
+		options := []batcher.Option[scenario.Item](nil)
+		if workers > 1 {
+			options = []batcher.Option[scenario.Item]{
+				batcher.WithConcurrency[scenario.Item](workers),
+				batcher.WithoutOrderedProcessing[scenario.Item](),
+			}
+		}
+
+		for _, window := range []time.Duration{time.Millisecond, 5 * time.Millisecond, 10 * time.Millisecond, 100 * time.Millisecond} {
+			result := scenario.Run(scenario.Config{
+				Name:           "concurrency/n=" + itoa(workers),
+				BatchSize:      1_000,
+				BatchInterval:  window,
+				Arrival:        scenario.Steady(10_000, 2*time.Second),
+				Processor:      scenario.FixedProcessor(20 * time.Millisecond),
+				BatcherOptions: options,
+				Producers:      runtime.NumCPU(),
+				Warmup:         200 * time.Millisecond,
+				Seed:           1,
+				LatenessBudget: 2 * time.Millisecond,
+			})
+
+			if result.TimedOut {
+				t.Errorf("concurrency/n=%d window=%s timed out; excluded from the report",
+					workers, window)
+
+				continue
+			}
+
+			results = append(results, result)
+		}
+	}
+
 	if err := scenario.WriteReport(os.Stdout, results); err != nil {
 		t.Fatalf("writing report: %v", err)
 	}
diff --git a/test/scenario/run.go b/test/scenario/run.go
index f46460e..c5e1e58 100644
--- a/test/scenario/run.go
+++ b/test/scenario/run.go
@@ -45,6 +45,13 @@ type Config struct {
 	Arrival   Arrival
 	Processor Processor
 
+	// BatcherOptions are applied after the harness's batch size, interval and
+	// processor options. They make the same open-loop scenario usable across
+	// implementation modes, such as the acknowledged worker pool in Phase 3.
+	// The harness owns Item, so callers cannot accidentally configure a batcher
+	// for a different payload type.
+	BatcherOptions []batcher.Option[Item]
+
 	// Producers is how many goroutines offer the schedule concurrently.
 	//
 	// The schedule is partitioned round-robin, so the *set* of offer times is
@@ -237,17 +244,33 @@ func Run(cfg Config) Result {
 		completedItems  atomic.Int64
 		completedSignal = make(chan struct{}, 1)
 
-		procRNG = rand.New(rand.NewSource(cfg.Seed))
-		start   time.Time
+		// procRNG is guarded because BatcherOptions may enable worker concurrency,
+		// and math/rand.Rand is not goroutine-safe. JitteredProcessor and
+		// SlowOutlierProcessor read it from inside the processor, so several workers
+		// can call it at once; the matrix sweep does exactly that. Without the mutex
+		// the race detector fires in math/rand.(*rngSource).Uint64 and the service
+		// times become undefined, which would silently invalidate the measurement.
+		procRNGMu sync.Mutex
+		procRNG   = rand.New(rand.NewSource(cfg.Seed))
+		start     time.Time
 	)
 
-	b := batcher.New(
+	// serviceTime keeps the lock around the RNG read only, not around the sleep, so
+	// concurrent workers still overlap their simulated downstream latency.
+	serviceTime := func(batchSize int) time.Duration {
+		procRNGMu.Lock()
+		defer procRNGMu.Unlock()
+
+		return cfg.Processor.ServiceTime(procRNG, batchSize)
+	}
+
+	options := []batcher.Option[Item]{
 		batcher.WithBatchSize[Item](cfg.BatchSize),
 		batcher.WithBatchInterval[Item](cfg.BatchInterval),
 		batcher.WithProcessor(func(items []Item) error {
 			procStart := int64(time.Since(start))
 
-			if d := cfg.Processor.ServiceTime(procRNG, len(items)); d > 0 {
+			if d := serviceTime(len(items)); d > 0 {
 				time.Sleep(d)
 			}
 
@@ -273,7 +296,11 @@ func Run(cfg Config) Result {
 
 			return cfg.Processor.Err
 		}),
-	)
+	}
+
+	options = append(options, cfg.BatcherOptions...)
+
+	b := batcher.New(options...)
 
 	// Drain diagnostics so an erroring processor cannot grow memory unboundedly
 	// during the run.
diff --git a/test/scenario/scenario_test.go b/test/scenario/scenario_test.go
index 8e427b9..eb44b4a 100644
--- a/test/scenario/scenario_test.go
+++ b/test/scenario/scenario_test.go
@@ -55,19 +55,30 @@ func TestHarnessReportsLatenessAndInvalidatesBadRuns(t *testing.T) {
 		})
 	}
 
-	// Probe what this host actually achieves, then assert the guard accepts a
-	// budget above it and rejects one below it.
+	// Probe what this host achieves, for the log only. Deriving the budget for a
+	// *later* run from a single probe made this test flaky: under parallel CPU
+	// contention the next run's overshoot can exceed the probe's p99 by more than any
+	// small margin, so the assertion failed for scheduling reasons rather than
+	// because the guard was wrong.
 	probe := measure(time.Hour)
 	require.Positive(t, probe.Lateness.Count, "lateness must be recorded")
 	t.Logf("host p99 sleep overshoot: %s", probe.Lateness.P99)
 
-	generous := measure(probe.Lateness.P99 + 50*time.Millisecond)
-	require.True(t, generous.LatenessValid,
-		"a run within its lateness budget must be valid (p99 %s)", generous.Lateness.P99)
+	// Both directions are what actually needs asserting, and each is checked against
+	// a budget that cannot be marginal:
+	//
+	//   - an hour is unreachable by any scheduling delay, so the run must be valid;
+	//   - a nanosecond is unmeetable by any real scheduler, so it must be invalid.
+	//
+	// That tests the guard rather than the host's timer precision.
+	require.True(t, probe.LatenessValid,
+		"a run with an unreachable budget must be valid (p99 %s)", probe.Lateness.P99)
 
-	strict := measure(time.Nanosecond) // no real scheduler can meet this
+	strict := measure(time.Nanosecond)
 	require.False(t, strict.LatenessValid,
 		"a run whose lateness exceeds its budget must be marked invalid")
+	require.Positive(t, strict.Lateness.P99,
+		"the invalid run must still report the lateness it measured")
 }
 
 // TestHarnessRecorderDoesNotAllocatePerItem is the harness self-check required