@@ -1038,6 +1038,37 @@ cannot be forcibly cancelled by this API; hard deadlines require process isolati
 The existing `ordered_map` convenience function uses a fresh pool for one pass and
 releases it on exit. Neither API depends on a machine-learning framework or cloud.
 
+For uneven work sizes, also constrain the planned bytes held by submitted tasks,
+completed results and the currently yielded result:
+
+```python
+with OrderedPool(make_processor, workers=4, max_pending=8,
+                 max_pending_bytes=128 << 20,
+                 estimate_bytes=lambda request: request.peak_buffer_bytes) as pool:
+    with pool.map(requests) as results:
+        for result in results:
+            consume(result)
+            del result
+```
+
+The estimator runs once per input before submission. It must return a nonnegative
+integer and account for every buffer covered by the claimed allowance. A task larger
+than the allowance fails before execution; reduce its selection or batch size, or
+increase the allowance. Tasks wait in input order instead of bypassing a large task.
+Admission remains charged while a result is yielded and releases when iteration resumes.
+Estimation and input-iteration errors abort the pass immediately; worker failures still
+surface at their input position.
+
+One input awaiting admission may be retained as lookahead. The input iterable,
+worker caches, unaccounted scratch, allocator overhead and consumer-retained values
+are outside this planned-byte allowance. Use lightweight selection descriptors as
+inputs when bounding actual data-buffer lifetime. This is not a process RSS limit.
+
+`python -m arrayroute.benchmark_admission --output admission.json` compares count-only
+and byte admission in fresh processes with uneven synthetic buffers, verified output
+hashes, traced allocations and peak RSS. It measures queue memory, not scientific
+I/O or training throughput.
+
 ## Measure reader concurrency on your own data
 
 `arrayroute.probe.probe_reads` runs bounded pilots through any `ArrayReader` factory:
 
@@ -1,6 +1,6 @@
 [project]
 name = "arrayroute"
-version = "0.1.21"
+version = "0.1.22"
 description = "Measured, bounded selections over scientific chunked arrays"
 requires-python = ">=3.11"
 license = "Apache-2.0"
 
@@ -0,0 +1,103 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Measure ordered queue payloads with uneven buffers and a paced consumer."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import platform
+import random
+import resource
+import subprocess
+import sys
+import time
+import tracemalloc
+from pathlib import Path
+
+import numpy as np
+
+from .execution import ordered_map
+
+
+def run_trial(policy: str, unit_bytes: int) -> dict:
+    if policy not in {'count', 'bytes'} or unit_bytes < 1:
+        raise ValueError('Choose count/bytes admission and a positive buffer unit')
+    sizes = [unit_bytes * n for n in [1, 4, 2, 6] * 4]
+    expected = []
+    for i, size in enumerate(sizes):
+        digest = hashlib.sha256()
+        block = bytes([i]) * min(size, 65536)
+        whole, remainder = divmod(size, len(block))
+        for _ in range(whole):
+            digest.update(block)
+        digest.update(block[:remainder])
+        expected.append(digest.hexdigest())
+
+    def factory():
+        return lambda i: np.full(sizes[i], i, dtype=np.uint8)
+
+    digests = []
+    tracemalloc.start()
+    start = time.perf_counter()
+    with ordered_map(factory, range(len(sizes)), workers=4, max_pending=8,
+                     max_pending_bytes=unit_bytes * 8 if policy == 'bytes' else None,
+                     estimate_bytes=(lambda i: sizes[i]) if policy == 'bytes' else None) as outputs:
+        for value in outputs:
+            digests.append(hashlib.sha256(memoryview(value)).hexdigest())
+            time.sleep(.01)
+            del value
+    seconds = time.perf_counter() - start
+    _, peak = tracemalloc.get_traced_memory()
+    tracemalloc.stop()
+    if digests != expected:
+        raise ValueError('Ordered payload bytes differ from the independent reference')
+    rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+    return {
+        'policy': policy, 'unit_bytes': unit_bytes, 'payload_bytes': sum(sizes),
+        'planned_queue_bytes': unit_bytes * 8 if policy == 'bytes' else None,
+        'seconds': seconds, 'peak_traced_bytes': peak,
+        'peak_rss_bytes': rss if sys.platform == 'darwin' else rss * 1024,
+        'digest_sha256': hashlib.sha256(json.dumps(digests).encode()).hexdigest(),
+        'verified': True, 'workers': 4, 'max_pending': 8, 'consumer_delay_seconds': .01,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--output', type=Path)
+    parser.add_argument('--unit-mib', type=int, default=16)
+    parser.add_argument('--repeats', type=int, default=3)
+    parser.add_argument('--worker', choices=['count', 'bytes'])
+    args = parser.parse_args()
+    if args.unit_mib < 1 or args.repeats < 1:
+        parser.error('unit-mib and repeats must be positive')
+    if args.worker:
+        print(json.dumps(run_trial(args.worker, args.unit_mib * (1 << 20))))
+        return
+    if args.output is None or args.output.exists():
+        parser.error('Choose a new --output path')
+    rows = []
+    for repeat in range(args.repeats):
+        policies = ['count', 'bytes']
+        random.Random(20261023 + repeat).shuffle(policies)
+        for policy in policies:
+            completed = subprocess.run(
+                [sys.executable, '-m', 'arrayroute.benchmark_admission',
+                 '--worker', policy, '--unit-mib', str(args.unit_mib)],
+                capture_output=True, text=True, check=True, timeout=300,
+            )
+            rows.append({'repeat': repeat, **json.loads(completed.stdout)})
+    if len({row['digest_sha256'] for row in rows}) != 1:
+        raise ValueError('Policies or repeats changed payload bytes/order')
+    report = {'scope': 'Synthetic variable-buffer queue; not scientific I/O or training speed',
+              'platform': platform.platform(), 'python': platform.python_version(),
+              'numpy': np.__version__, 'trials': rows}
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    with args.output.open('x') as output:
+        json.dump(report, output, indent=2)
+        output.write('\n')
+    print(json.dumps(report, indent=2))
+
+
+if __name__ == '__main__':
+    main()
@@ -25,7 +25,11 @@ class OrderedPool(Generic[T, R]):
     Each active thread constructs one callable and keeps it until this pool closes.
     A pass owns at most ``max_pending`` tasks/results, including its yielded result.
     Caller-retained values, native caches and scratch are additional memory. This is
-    a task-count bound, not an RSS cap. Factories must not return shared mutable readers.
+    a task-count bound, not an RSS cap. Optional byte admission also bounds the sum
+    of caller-supplied task estimates until each result is consumed. Estimates must
+    cover the memory they claim; caches, scratch and caller-retained values remain
+    additional. One not-yet-admitted input may be retained for lookahead. Factories
+    must not return shared mutable readers.
 
     Map contexts are exclusive, including before consumption starts. Task failures
     surface in input order; no task is retried or skipped. Leaving a map context
@@ -40,15 +44,25 @@ def __init__(
         *,
         workers: int,
         max_pending: int | None = None,
+        max_pending_bytes: int | None = None,
+        estimate_bytes: Callable[[T], int] | None = None,
     ) -> None:
         if isinstance(workers, bool) or isinstance(max_pending, bool):
             raise ValueError("workers and max_pending must be positive integers")
         workers = operator.index(workers)
         capacity = workers if max_pending is None else operator.index(max_pending)
         if workers < 1 or capacity < 1:
             raise ValueError("workers and max_pending must be positive integers")
+        if (max_pending_bytes is None) != (estimate_bytes is None):
+            raise ValueError("Byte admission requires both max_pending_bytes and estimate_bytes")
+        if max_pending_bytes is not None:
+            if isinstance(max_pending_bytes, bool) or operator.index(max_pending_bytes) < 1:
+                raise ValueError("max_pending_bytes must be a positive integer")
+            max_pending_bytes = operator.index(max_pending_bytes)
         self._make_worker = make_worker
         self._capacity = capacity
+        self._byte_capacity = max_pending_bytes
+        self._estimate_bytes = estimate_bytes
         self._local = threading.local()
         self._state_lock = threading.Lock()
         self._active = False
@@ -74,7 +88,7 @@ def map(
             if self._active:
                 raise RuntimeError("Concurrent map contexts are not supported")
             self._active = True
-        pending: deque[Future[R]] = deque()
+        pending: deque[tuple[Future[R], int]] = deque()
 
         def check() -> None:
             if self._closed:
@@ -85,38 +99,61 @@ def check() -> None:
         def iterate() -> Generator[R, None, None]:
             source = iter(items)
             exhausted = False
+            deferred: tuple[T, int] | None = None
+            admitted_bytes = 0
             while True:
                 check()
                 while not exhausted and len(pending) < self._capacity:
                     check()
-                    try:
-                        item = next(source)
-                    except StopIteration:
-                        exhausted = True
+                    if deferred is None:
+                        try:
+                            item = next(source)
+                        except StopIteration:
+                            exhausted = True
+                            break
+                        amount = 0 if self._estimate_bytes is None else self._estimate_bytes(item)
+                        if isinstance(amount, bool) or operator.index(amount) < 0:
+                            raise ValueError("Task byte estimate must be a nonnegative integer")
+                        amount = operator.index(amount)
+                        if self._byte_capacity is not None and amount > self._byte_capacity:
+                            raise MemoryError(
+                                f"Task requires {amount} planned bytes; pending-byte allowance "
+                                f"is {self._byte_capacity}. "
+                                "Reduce the task size or raise the allowance."
+                            )
+                    else:
+                        item, amount = deferred
+                        deferred = None
+                    if (self._byte_capacity is not None
+                            and admitted_bytes + amount > self._byte_capacity):
+                        deferred = (item, amount)
+                        del item
                         break
-                    pending.append(self._executor.submit(self._execute, item, check))
+                    pending.append((self._executor.submit(self._execute, item, check), amount))
+                    admitted_bytes += amount
                     del item
                 if not pending:
                     return
                 # Keep the awaited future in the deque so cleanup also waits for it
                 # if the consumer is interrupted while future.result() is waiting.
-                future = pending[0]
+                future, amount = pending[0]
                 result = future.result()
                 pending.popleft()
                 del future
                 check()
                 yield result
                 del result
+                admitted_bytes -= amount
 
         output = iterate()
         try:
             yield output
         finally:
             try:
-                for future in pending:
+                for future, _ in pending:
                     future.cancel()
                 if pending:
-                    wait(pending)
+                    wait(future for future, _ in pending)
                 output.close()
                 pending.clear()
             finally:
@@ -147,13 +184,16 @@ def ordered_map(
     *,
     workers: int,
     max_pending: int | None = None,
+    max_pending_bytes: int | None = None,
+    estimate_bytes: Callable[[T], int] | None = None,
     cancelled: Callable[[], bool] | None = None,
 ) -> Iterator[Iterator[R]]:
     """Run one bounded pass and release its thread-owned workers on context exit.
 
     See ``OrderedPool`` for ordering, memory and cancellation guarantees. Use a pool
     explicitly when repeated passes should retain the same native readers and caches.
     """
-    with OrderedPool(make_worker, workers=workers, max_pending=max_pending) as pool:
+    with OrderedPool(make_worker, workers=workers, max_pending=max_pending,
+                     max_pending_bytes=max_pending_bytes, estimate_bytes=estimate_bytes) as pool:
         with pool.map(items, cancelled=cancelled) as output:
             yield output
@@ -0,0 +1,97 @@
+# SPDX-License-Identifier: Apache-2.0
+import threading
+
+import pytest
+
+from arrayroute.execution import OrderedPool, ProcessingCancelledError, ordered_map
+
+
+def test_bytes_limit_submission_without_reordering_or_reestimating():
+    started, estimated = [], []
+    second = threading.Event()
+    weights = [3, 3, 5, 1, 0]
+
+    def estimate(i):
+        estimated.append(i)
+        return weights[i]
+
+    def factory():
+        def process(i):
+            started.append(i)
+            if i == 0:
+                assert second.wait(5)
+            if i == 1:
+                second.set()
+            return i
+        return process
+
+    with ordered_map(factory, range(5), workers=3, max_pending=8,
+                     max_pending_bytes=6, estimate_bytes=estimate) as values:
+        assert next(values) == 0
+        assert set(started) == {0, 1}
+        assert estimated == [0, 1, 2]
+        assert next(values) == 1
+        assert set(started) == {0, 1}
+        assert list(values) == [2, 3, 4]
+    assert estimated == list(range(5))
+
+
+def test_one_full_budget_task_progresses_and_pool_can_be_reused():
+    with OrderedPool(lambda: lambda x: x, workers=8, max_pending=16,
+                     max_pending_bytes=10, estimate_bytes=lambda x: 10) as pool:
+        for _ in range(3):
+            with pool.map(range(100)) as values:
+                assert list(values) == list(range(100))
+
+
+def test_oversized_task_is_rejected_before_worker_creation():
+    with ordered_map(lambda: pytest.fail('No worker should be created'), [11], workers=2,
+                     max_pending_bytes=10, estimate_bytes=lambda x: x) as values:
+        with pytest.raises(MemoryError, match='11 planned bytes.*10'):
+            next(values)
+
+
+@pytest.mark.parametrize('budget,estimate', [(None, lambda x: 1), (1, None),
+                                            (0, lambda x: 1), (True, lambda x: 1)])
+def test_invalid_byte_admission(budget, estimate):
+    with pytest.raises(ValueError):
+        OrderedPool(lambda: lambda x: x, workers=1,
+                    max_pending_bytes=budget, estimate_bytes=estimate)
+
+
+@pytest.mark.parametrize('estimate', [-1, True, 1.5])
+def test_invalid_task_estimate_does_not_execute(estimate):
+    with ordered_map(lambda: pytest.fail('Invalid estimate must not execute'), [0], workers=1,
+                     max_pending_bytes=10, estimate_bytes=lambda x: estimate) as values:
+        with pytest.raises((ValueError, TypeError)):
+            next(values)
+
+
+def test_cancelled_deferred_task_is_not_submitted_and_next_pass_works():
+    cancelled = False
+    started = []
+    with OrderedPool(lambda: lambda x: started.append(x) or x, workers=2,
+                     max_pending_bytes=1, estimate_bytes=lambda x: 1) as pool:
+        with pool.map(range(5), cancelled=lambda: cancelled) as values:
+            assert next(values) == 0
+            cancelled = True
+            with pytest.raises(ProcessingCancelledError):
+                next(values)
+        assert started == [0]
+        with pool.map([9]) as values:
+            assert list(values) == [9]
+
+
+def test_failed_task_retains_order_and_releases_pass():
+    def process(i):
+        if i == 1:
+            raise OSError('decoder failed')
+        return i
+    with OrderedPool(lambda: process, workers=3, max_pending_bytes=5,
+                     estimate_bytes=lambda x: 3) as pool:
+        with pool.map(range(4)) as values:
+            assert next(values) == 0
+            with pytest.raises(OSError, match='decoder failed'):
+                next(values)
+        with pool.map([8]) as values:
+            assert list(values) == [8]