diff --git a/CHANGELOG.md b/CHANGELOG.md
index 60948d37..4920f5a2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Middleware / interceptor (#93):** a composable per-item hook — `IItemMiddleware<T>` returning
+  `MiddlewareResult<T>` (`Continue` to keep/replace an item, `Drop` to remove it) — attached to any
+  stream with the `WithMiddleware(...)` extensions (single or ordered chain). Lets cross-cutting
+  concerns (logging, validation, metrics, throttling, dedup) decorate an extractor/transformer output
+  or loader input without changing the component, and composes inside an `EtlPipeline` via
+  `Through(s => s.WithMiddleware(...))`. Dependency-free.
 - **Retry seam (#94):** `ExtractorBase`, `LoaderBase`, and `TransformerBase` gained a
   `protected virtual WrapWorkerExecution(...)` hook wrapped around every worker invocation. The
   default implementation is a no-op, so behaviour is unchanged; override it to run the worker through
diff --git a/src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs b/src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs
new file mode 100644
index 00000000..511f51df
--- /dev/null
+++ b/src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs
@@ -0,0 +1,25 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// A composable, reusable hook for cross-cutting per-item behaviour (logging, validation, metrics,
+/// throttling, deduplication) that can be attached to any stream without modifying the extractor,
+/// transformer, or loader that produced it. Attach one or more with
+/// <see cref="MiddlewareExtensions.WithMiddleware{T}(System.Collections.Generic.IAsyncEnumerable{T}, IItemMiddleware{T}, CancellationToken)"/>;
+/// they run in the order attached, each seeing the item the previous one passed on.
+/// </summary>
+/// <typeparam name="T">The item type flowing through the pipeline.</typeparam>
+public interface IItemMiddleware<T>
+{
+    /// <summary>
+    /// Invoked once per item. Return <see cref="MiddlewareResult.Continue{T}(T)"/> to keep the item
+    /// flowing (optionally replacing it), or <see cref="MiddlewareResult.Drop{T}"/> to remove it from
+    /// the stream.
+    /// </summary>
+    /// <param name="item">The item to process.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The outcome describing whether to keep or drop the item.</returns>
+    ValueTask<MiddlewareResult<T>> OnItemAsync(T item, CancellationToken token);
+}
diff --git a/src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs b/src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs
new file mode 100644
index 00000000..fb4565b1
--- /dev/null
+++ b/src/Wolfgang.Etl.Abstractions/MiddlewareExtensions.cs
@@ -0,0 +1,122 @@
+using System;
+using System.Collections.Generic;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// Extension methods that attach <see cref="IItemMiddleware{T}"/> to an
+/// <see cref="IAsyncEnumerable{T}"/> stream, so cross-cutting per-item behaviour composes onto any
+/// extractor / transformer output or loader input — and inside an <c>EtlPipeline</c> via
+/// <c>Through(stream =&gt; stream.WithMiddleware(...))</c> — without changing the component itself.
+/// </summary>
+public static class MiddlewareExtensions
+{
+    /// <summary>
+    /// Pipes every item of <paramref name="source"/> through <paramref name="middleware"/>. Items the
+    /// middleware drops (<see cref="MiddlewareResult.Drop{T}"/>) are removed from the stream; otherwise
+    /// the (possibly replaced) item is yielded.
+    /// </summary>
+    /// <typeparam name="T">The item type.</typeparam>
+    /// <param name="source">The stream to decorate.</param>
+    /// <param name="middleware">The middleware to run for each item.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The decorated stream.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="middleware"/> is <see langword="null"/>.</exception>
+    public static async IAsyncEnumerable<T> WithMiddleware<T>
+    (
+        this IAsyncEnumerable<T> source,
+        IItemMiddleware<T> middleware,
+        [EnumeratorCancellation] CancellationToken token = default
+    )
+    {
+        if (source is null)
+        {
+            throw new ArgumentNullException(nameof(source));
+        }
+
+        if (middleware is null)
+        {
+            throw new ArgumentNullException(nameof(middleware));
+        }
+
+        await foreach (var item in source.WithCancellation(token))
+        {
+            // Stryker disable once Boolean: equivalent — with no synchronization context in play, ConfigureAwait(false) and (true) are indistinguishable.
+            var result = await middleware.OnItemAsync(item, token).ConfigureAwait(false);
+            if (!result.Skip)
+            {
+                yield return result.Item;
+            }
+        }
+    }
+
+
+
+    /// <summary>
+    /// Pipes every item of <paramref name="source"/> through <paramref name="middlewares"/> in order:
+    /// each middleware sees the item the previous one passed on. If any middleware drops the item
+    /// (<see cref="MiddlewareResult.Drop{T}"/>), the remaining middleware is not run and the item is
+    /// removed from the stream.
+    /// </summary>
+    /// <typeparam name="T">The item type.</typeparam>
+    /// <param name="source">The stream to decorate.</param>
+    /// <param name="middlewares">The middleware chain, applied in enumeration order.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The decorated stream.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="middlewares"/> is <see langword="null"/>, or a member of <paramref name="middlewares"/> is <see langword="null"/>.</exception>
+    public static async IAsyncEnumerable<T> WithMiddleware<T>
+    (
+        this IAsyncEnumerable<T> source,
+        IEnumerable<IItemMiddleware<T>> middlewares,
+        [EnumeratorCancellation] CancellationToken token = default
+    )
+    {
+        if (source is null)
+        {
+            throw new ArgumentNullException(nameof(source));
+        }
+
+        if (middlewares is null)
+        {
+            throw new ArgumentNullException(nameof(middlewares));
+        }
+
+        // Snapshot the chain once so the same ordered set runs for every item.
+        var chain = new List<IItemMiddleware<T>>(middlewares);
+        foreach (var middleware in chain)
+        {
+            if (middleware is null)
+            {
+                // Stryker disable once String: the exception message is diagnostic-only, not a behavioural contract asserted by tests.
+                throw new ArgumentNullException(nameof(middlewares), "A middleware in the chain is null.");
+            }
+        }
+
+        await foreach (var item in source.WithCancellation(token))
+        {
+            var current = item;
+            var dropped = false;
+
+            foreach (var middleware in chain)
+            {
+                // Stryker disable once Boolean: equivalent — with no synchronization context in play, ConfigureAwait(false) and (true) are indistinguishable.
+                var result = await middleware.OnItemAsync(current, token).ConfigureAwait(false);
+                if (result.Skip)
+                {
+                    dropped = true;
+                    break;
+                }
+
+                current = result.Item;
+            }
+
+            if (!dropped)
+            {
+                yield return current;
+            }
+        }
+    }
+}
diff --git a/src/Wolfgang.Etl.Abstractions/MiddlewareResult.Factory.cs b/src/Wolfgang.Etl.Abstractions/MiddlewareResult.Factory.cs
new file mode 100644
index 00000000..b8f9a1c6
--- /dev/null
+++ b/src/Wolfgang.Etl.Abstractions/MiddlewareResult.Factory.cs
@@ -0,0 +1,25 @@
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// Factory methods for creating <see cref="MiddlewareResult{T}"/> values from an
+/// <see cref="IItemMiddleware{T}"/> implementation.
+/// </summary>
+public static class MiddlewareResult
+{
+    /// <summary>
+    /// Keeps the item in the stream, optionally replacing it with a transformed value.
+    /// </summary>
+    /// <typeparam name="T">The item type.</typeparam>
+    /// <param name="item">The item to pass on.</param>
+    /// <returns>A result that keeps <paramref name="item"/> flowing.</returns>
+    public static MiddlewareResult<T> Continue<T>(T item) => new(item, skip: false);
+
+
+
+    /// <summary>
+    /// Drops the current item from the stream.
+    /// </summary>
+    /// <typeparam name="T">The item type.</typeparam>
+    /// <returns>A result that discards the current item.</returns>
+    public static MiddlewareResult<T> Drop<T>() => new(default!, skip: true);
+}
diff --git a/src/Wolfgang.Etl.Abstractions/MiddlewareResult.cs b/src/Wolfgang.Etl.Abstractions/MiddlewareResult.cs
new file mode 100644
index 00000000..ec86e59b
--- /dev/null
+++ b/src/Wolfgang.Etl.Abstractions/MiddlewareResult.cs
@@ -0,0 +1,64 @@
+using System;
+using System.Collections.Generic;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// The outcome of running a single item through an <see cref="IItemMiddleware{T}"/>: the
+/// (possibly replaced) item to pass on, and whether the item should be dropped from the stream.
+/// Create one with <see cref="MiddlewareResult.Continue{T}(T)"/> to keep an item flowing or
+/// <see cref="MiddlewareResult.Drop{T}"/> to discard it.
+/// </summary>
+/// <typeparam name="T">The item type flowing through the pipeline.</typeparam>
+public readonly struct MiddlewareResult<T> : IEquatable<MiddlewareResult<T>>
+{
+    internal MiddlewareResult(T item, bool skip)
+    {
+        Item = item;
+        Skip = skip;
+    }
+
+
+
+    /// <summary>
+    /// The item to pass on to the next middleware (or to the stream). Meaningful only when
+    /// <see cref="Skip"/> is <see langword="false"/>.
+    /// </summary>
+    public T Item { get; }
+
+
+
+    /// <summary>
+    /// <see langword="true"/> to drop the item from the stream (later middleware is not run and the
+    /// item is not yielded); <see langword="false"/> to keep it.
+    /// </summary>
+    public bool Skip { get; }
+
+
+
+    /// <inheritdoc/>
+    public bool Equals(MiddlewareResult<T> other) =>
+        Skip == other.Skip && EqualityComparer<T>.Default.Equals(Item, other.Item);
+
+
+
+    /// <inheritdoc/>
+    public override bool Equals(object? obj) => obj is MiddlewareResult<T> other && Equals(other);
+
+
+
+    /// <inheritdoc/>
+    // Stryker disable once all: equivalent — any change to the hash formula still yields equal hash codes for equal values (the only GetHashCode contract), so no behavioural test can distinguish it.
+    public override int GetHashCode() =>
+        unchecked(((Skip ? 1 : 0) * 397) ^ (Item is null ? 0 : EqualityComparer<T>.Default.GetHashCode(Item)));
+
+
+
+    /// <summary>Indicates whether two results are equal.</summary>
+    public static bool operator ==(MiddlewareResult<T> left, MiddlewareResult<T> right) => left.Equals(right);
+
+
+
+    /// <summary>Indicates whether two results are not equal.</summary>
+    public static bool operator !=(MiddlewareResult<T> left, MiddlewareResult<T> right) => !left.Equals(right);
+}
diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
index fa63bcc8..e53b6f35 100644
--- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
+++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
@@ -3,3 +3,20 @@ virtual Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.WrapWorkerEx
 virtual Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.WrapWorkerExecution(System.Func<System.Threading.CancellationToken, System.Threading.Tasks.Task!>! workerFactory, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task!
 virtual Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.WrapWorkerExecution(System.Func<System.Threading.CancellationToken, System.Collections.Generic.IAsyncEnumerable<TDestination>!>! workerFactory, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable<TDestination>!
 Wolfgang.Etl.Abstractions.Report.Report(int currentItemCount, System.DateTimeOffset? startedAt, System.TimeSpan elapsed, int? totalItemCount = null) -> void
+Wolfgang.Etl.Abstractions.IItemMiddleware<T>
+Wolfgang.Etl.Abstractions.IItemMiddleware<T>.OnItemAsync(T item, System.Threading.CancellationToken token) -> System.Threading.Tasks.ValueTask<Wolfgang.Etl.Abstractions.MiddlewareResult<T>>
+Wolfgang.Etl.Abstractions.MiddlewareResult
+static Wolfgang.Etl.Abstractions.MiddlewareResult.Continue<T>(T item) -> Wolfgang.Etl.Abstractions.MiddlewareResult<T>
+static Wolfgang.Etl.Abstractions.MiddlewareResult.Drop<T>() -> Wolfgang.Etl.Abstractions.MiddlewareResult<T>
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>.MiddlewareResult() -> void
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>.Item.get -> T
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>.Skip.get -> bool
+Wolfgang.Etl.Abstractions.MiddlewareResult<T>.Equals(Wolfgang.Etl.Abstractions.MiddlewareResult<T> other) -> bool
+override Wolfgang.Etl.Abstractions.MiddlewareResult<T>.Equals(object? obj) -> bool
+override Wolfgang.Etl.Abstractions.MiddlewareResult<T>.GetHashCode() -> int
+static Wolfgang.Etl.Abstractions.MiddlewareResult<T>.operator ==(Wolfgang.Etl.Abstractions.MiddlewareResult<T> left, Wolfgang.Etl.Abstractions.MiddlewareResult<T> right) -> bool
+static Wolfgang.Etl.Abstractions.MiddlewareResult<T>.operator !=(Wolfgang.Etl.Abstractions.MiddlewareResult<T> left, Wolfgang.Etl.Abstractions.MiddlewareResult<T> right) -> bool
+Wolfgang.Etl.Abstractions.MiddlewareExtensions
+static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware<T>(this System.Collections.Generic.IAsyncEnumerable<T>! source, Wolfgang.Etl.Abstractions.IItemMiddleware<T>! middleware, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<T>!
+static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware<T>(this System.Collections.Generic.IAsyncEnumerable<T>! source, System.Collections.Generic.IEnumerable<Wolfgang.Etl.Abstractions.IItemMiddleware<T>!>! middlewares, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<T>!
diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/MiddlewareTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/MiddlewareTests.cs
new file mode 100644
index 00000000..92ae8342
--- /dev/null
+++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/MiddlewareTests.cs
@@ -0,0 +1,266 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Wolfgang.Etl.Abstractions;
+using Xunit;
+
+namespace Wolfgang.Etl.Abstractions.Tests.Unit;
+
+/// <summary>
+/// Covers the #93 middleware mechanism: <see cref="IItemMiddleware{T}"/>,
+/// <see cref="MiddlewareResult{T}"/>, and the <c>WithMiddleware</c> stream decorators — including
+/// composition inside an <see cref="EtlPipeline"/> <c>Through</c> stage.
+/// </summary>
+public class MiddlewareTests
+{
+    // ---------- MiddlewareResult value type ----------
+
+    [Fact]
+    public void MiddlewareResult_Continue_keeps_the_item()
+    {
+        var result = MiddlewareResult.Continue(42);
+
+        Assert.False(result.Skip);
+        Assert.Equal(42, result.Item);
+    }
+
+
+    [Fact]
+    public void MiddlewareResult_Drop_marks_the_item_skipped()
+    {
+        var result = MiddlewareResult.Drop<int>();
+
+        Assert.True(result.Skip);
+    }
+
+
+    [Fact]
+    public void MiddlewareResult_has_value_equality()
+    {
+        var a = MiddlewareResult.Continue(7);
+        var b = MiddlewareResult.Continue(7);
+        var different = MiddlewareResult.Continue(8);
+        var dropped = MiddlewareResult.Drop<int>();
+
+        Assert.Equal(a, b);
+        Assert.True(a == b);
+        Assert.False(a != b);
+        Assert.Equal(a.GetHashCode(), b.GetHashCode());
+        Assert.NotEqual(a, different);
+        Assert.True(a != different);
+        Assert.NotEqual(a, dropped);
+    }
+
+
+    // ---------- single middleware ----------
+
+    [Fact]
+    public async Task WithMiddleware_transforms_each_item()
+    {
+        var items = await Drain(AsyncSource(1, 2, 3).WithMiddleware(new TimesTenMiddleware()));
+
+        Assert.Equal(new[] { 10, 20, 30 }, items);
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_drops_items_the_middleware_skips()
+    {
+        var items = await Drain(AsyncSource(1, 2, 3, 4).WithMiddleware(new DropOddMiddleware()));
+
+        Assert.Equal(new[] { 2, 4 }, items);
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_flows_the_cancellation_token_to_the_middleware()
+    {
+        using var cts = new CancellationTokenSource();
+        var capturing = new TokenCapturingMiddleware();
+
+        await Drain(AsyncSource(1).WithMiddleware(capturing), cts.Token);
+
+        Assert.Equal(cts.Token, capturing.LastToken);
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_composes_inside_an_EtlPipeline_Through_stage()
+    {
+        var stream = EtlPipeline
+            .Create()
+            .From(AsyncSource(1, 2, 3, 4))
+            .Through(s => s.WithMiddleware(new DropOddMiddleware()))
+            .Through(s => s.WithMiddleware(new TimesTenMiddleware()))
+            .AsAsyncEnumerable();
+
+        var items = await Drain(stream);
+
+        Assert.Equal(new[] { 20, 40 }, items);
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_when_source_is_null_throws_ArgumentNullException()
+    {
+        await Assert.ThrowsAsync<ArgumentNullException>(
+            () => Drain(((IAsyncEnumerable<int>)null!).WithMiddleware(new TimesTenMiddleware())));
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_when_middleware_is_null_throws_ArgumentNullException()
+    {
+        await Assert.ThrowsAsync<ArgumentNullException>(
+            () => Drain(AsyncSource(1).WithMiddleware((IItemMiddleware<int>)null!)));
+    }
+
+
+    // ---------- middleware chain ----------
+
+    [Fact]
+    public async Task WithMiddleware_chain_runs_in_registration_order_each_seeing_the_previous_output()
+    {
+        var log = new List<string>();
+        var chain = new IItemMiddleware<int>[]
+        {
+            new RecordingMiddleware(log, "A", add: 10),
+            new RecordingMiddleware(log, "B", add: 100),
+        };
+
+        var items = await Drain(AsyncSource(1).WithMiddleware(chain));
+
+        Assert.Equal(new[] { 111 }, items);         // 1 -> +10 -> +100
+        Assert.Equal(new[] { "A:1", "B:11" }, log);  // B saw A's output
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_chain_stops_at_the_first_drop()
+    {
+        var log = new List<string>();
+        var chain = new IItemMiddleware<int>[]
+        {
+            new DropOddMiddleware(),
+            new RecordingMiddleware(log, "R", add: 0),
+        };
+
+        var items = await Drain(AsyncSource(1, 2, 3).WithMiddleware(chain));
+
+        Assert.Equal(new[] { 2 }, items);        // odds dropped before reaching R
+        Assert.Equal(new[] { "R:2" }, log);      // R only ran for the surviving even item
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_empty_chain_passes_items_through()
+    {
+        var items = await Drain(AsyncSource(1, 2, 3).WithMiddleware(Array.Empty<IItemMiddleware<int>>()));
+
+        Assert.Equal(new[] { 1, 2, 3 }, items);
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_chain_when_source_is_null_throws_ArgumentNullException()
+    {
+        await Assert.ThrowsAsync<ArgumentNullException>(
+            () => Drain(((IAsyncEnumerable<int>)null!).WithMiddleware(new IItemMiddleware<int>[] { new TimesTenMiddleware() })));
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_chain_when_middlewares_is_null_throws_ArgumentNullException()
+    {
+        var ex = await Assert.ThrowsAsync<ArgumentNullException>(
+            () => Drain(AsyncSource(1).WithMiddleware((IEnumerable<IItemMiddleware<int>>)null!)));
+
+        // The explicit guard names "middlewares"; without it the fallback List ctor would name "collection".
+        Assert.Equal("middlewares", ex.ParamName);
+    }
+
+
+    [Fact]
+    public async Task WithMiddleware_chain_when_a_member_is_null_throws_ArgumentNullException()
+    {
+        var chain = new IItemMiddleware<int>[] { new TimesTenMiddleware(), null! };
+
+        await Assert.ThrowsAsync<ArgumentNullException>(
+            () => Drain(AsyncSource(1).WithMiddleware(chain)));
+    }
+
+
+    // ---------- helpers ----------
+
+    private static async IAsyncEnumerable<int> AsyncSource(params int[] items)
+    {
+        foreach (var item in items)
+        {
+            await Task.Yield();
+            yield return item;
+        }
+    }
+
+
+    private static async Task<List<int>> Drain(IAsyncEnumerable<int> source, CancellationToken token = default)
+    {
+        var result = new List<int>();
+        await foreach (var item in source.WithCancellation(token).ConfigureAwait(false))
+        {
+            result.Add(item);
+        }
+
+        return result;
+    }
+
+
+    // ---------- doubles ----------
+
+    private sealed class TimesTenMiddleware : IItemMiddleware<int>
+    {
+        public ValueTask<MiddlewareResult<int>> OnItemAsync(int item, CancellationToken token) =>
+            new(MiddlewareResult.Continue(item * 10));
+    }
+
+
+    private sealed class DropOddMiddleware : IItemMiddleware<int>
+    {
+        public ValueTask<MiddlewareResult<int>> OnItemAsync(int item, CancellationToken token) =>
+            new(item % 2 == 0 ? MiddlewareResult.Continue(item) : MiddlewareResult.Drop<int>());
+    }
+
+
+    private sealed class TokenCapturingMiddleware : IItemMiddleware<int>
+    {
+        public CancellationToken LastToken { get; private set; }
+
+        public ValueTask<MiddlewareResult<int>> OnItemAsync(int item, CancellationToken token)
+        {
+            LastToken = token;
+            return new(MiddlewareResult.Continue(item));
+        }
+    }
+
+
+    private sealed class RecordingMiddleware : IItemMiddleware<int>
+    {
+        private readonly List<string> _log;
+        private readonly string _name;
+        private readonly int _add;
+
+        public RecordingMiddleware(List<string> log, string name, int add)
+        {
+            _log = log;
+            _name = name;
+            _add = add;
+        }
+
+        public ValueTask<MiddlewareResult<int>> OnItemAsync(int item, CancellationToken token)
+        {
+            _log.Add($"{_name}:{item}");
+            return new(MiddlewareResult.Continue(item + _add));
+        }
+    }
+}