diff --git a/docs/guide/handlers/batching.md b/docs/guide/handlers/batching.md
index ac74b8c6b..997765655 100644
--- a/docs/guide/handlers/batching.md
+++ b/docs/guide/handlers/batching.md
@@ -195,6 +195,57 @@ The durable inbox behaves just a little bit differently for message batching. Wo
 "handle" the individual messages, but does not mark them as handled in the message store until a batch message
 that refers to the original message is completely processed. 
 
+## De-duplicating a batch with `CoalesceBy`
+
+A very common batching workload is a "trigger storm": a bulk operation fires hundreds or thousands of
+"recalculate" messages that actually concern only a few dozen distinct entities. Batching already collapses
+those into one handler invocation, but the handler still sees every duplicate and recomputes the same entity
+many times. `CoalesceBy` de-duplicates the batch by a key so the handler sees **one message per distinct key,
+last message wins**:
+
+```csharp
+opts.BatchMessagesOf<RecalculateScores>(batching =>
+{
+    batching.BatchSize = 500;
+    batching.TriggerTime = 10.Seconds();
+
+    // The handler sees at most one RecalculateScores per AggregateId (the latest one)
+    batching.CoalesceBy((RecalculateScores x) => x.AggregateId);
+});
+```
+
+::: tip
+The key selector's lambda parameter must be **explicitly typed** to the batched element type (e.g.
+`(RecalculateScores x) => ...`) so both the message and key type arguments can be inferred.
+:::
+
+`CoalesceBy` is just sugar over the [`IMessageBatcher`](#custom-batching-strategies) seam — it installs a
+built-in `CoalescingMessageBatcher<T, TKey>` instead of the default batcher. Like the default it first groups
+by tenant id, then de-duplicates within each tenant group.
+
+Crucially, coalescing only changes **what the handler sees** — never what gets acknowledged. Every original
+member message still rides on the batch, so the transactional inbox/outbox tracking and dead-lettering behave
+exactly as they do for a normal (non-coalesced) batch. If you drop from 1,000 messages to 40 distinct keys,
+the handler runs once over 40 items, but all 1,000 member messages are settled with that batch.
+
+## Batch identity with `IBatchContext`
+
+A batched handler can inject `IBatchContext` to get read-only information about the batch it is processing —
+useful for correlating log entries, emitting batch-level metrics, or making per-batch decisions:
+
+```csharp
+public static void Handle(Item[] items, IBatchContext batch)
+{
+    // batch.BatchId is a stable id for this assembled batch (correlate your logs with it)
+    // batch.Members describes each original member message: MessageId, Attempts, SentAt
+    logger.LogInformation("Processing batch {BatchId} of {Count} members", batch.BatchId, batch.Members.Count);
+}
+```
+
+`IBatchContext` is purely informational; reading it never changes what is acknowledged or how the batch is
+settled. When combined with `CoalesceBy`, `Members` still lists **every** original member message (all the
+ones that settle with the batch), even though the `items` array the handler sees was de-duplicated.
+
 ## Custom Batching Strategies
 
 ::: info
diff --git a/src/Testing/CoreTests/Acceptance/batch_coalescing_and_context.cs b/src/Testing/CoreTests/Acceptance/batch_coalescing_and_context.cs
new file mode 100644
index 000000000..3e1ec394a
--- /dev/null
+++ b/src/Testing/CoreTests/Acceptance/batch_coalescing_and_context.cs
@@ -0,0 +1,212 @@
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Wolverine;
+using Wolverine.Runtime.Batching;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace CoreTests.Acceptance;
+
+public class CoalescingMessageBatcherTests
+{
+    private static Envelope envelopeFor(ScoreEvent message, string? tenantId = null)
+    {
+        return new Envelope(message) { TenantId = tenantId };
+    }
+
+    [Fact]
+    public void dedupes_by_key_last_wins_but_keeps_every_member_on_the_batch()
+    {
+        var batcher = new CoalescingMessageBatcher<ScoreEvent, string>(x => x.AggregateId);
+
+        var envelopes = new[]
+        {
+            envelopeFor(new ScoreEvent("A", 1)),
+            envelopeFor(new ScoreEvent("A", 2)),
+            envelopeFor(new ScoreEvent("A", 3)),
+            envelopeFor(new ScoreEvent("B", 1)),
+            envelopeFor(new ScoreEvent("B", 2))
+        };
+
+        var batch = batcher.Group(envelopes).ShouldHaveSingleItem();
+
+        // What the handler SEES: one message per key, last wins
+        var seen = batch.Message.ShouldBeOfType<ScoreEvent[]>();
+        seen.Length.ShouldBe(2);
+        seen.Single(x => x.AggregateId == "A").Version.ShouldBe(3);
+        seen.Single(x => x.AggregateId == "B").Version.ShouldBe(2);
+
+        // What gets SETTLED: every original member envelope still rides on the batch, so inbox/outbox
+        // tracking and dead-lettering are identical to a non-coalescing batch.
+        batch.Batch!.Length.ShouldBe(5);
+    }
+
+    [Fact]
+    public void groups_by_tenant_id_before_coalescing()
+    {
+        var batcher = new CoalescingMessageBatcher<ScoreEvent, string>(x => x.AggregateId);
+
+        var envelopes = new[]
+        {
+            envelopeFor(new ScoreEvent("A", 1), "tenant1"),
+            envelopeFor(new ScoreEvent("A", 2), "tenant2"),
+            envelopeFor(new ScoreEvent("A", 3), "tenant1")
+        };
+
+        var batches = batcher.Group(envelopes).ToArray();
+        batches.Length.ShouldBe(2);
+
+        var tenant1 = batches.Single(x => x.TenantId == "tenant1");
+        tenant1.Message.ShouldBeOfType<ScoreEvent[]>().ShouldHaveSingleItem().Version.ShouldBe(3);
+        tenant1.Batch!.Length.ShouldBe(2);
+
+        var tenant2 = batches.Single(x => x.TenantId == "tenant2");
+        tenant2.Message.ShouldBeOfType<ScoreEvent[]>().ShouldHaveSingleItem().Version.ShouldBe(2);
+        tenant2.Batch!.Length.ShouldBe(1);
+    }
+
+    [Fact]
+    public void coalesce_by_rejects_a_selector_for_the_wrong_element_type()
+    {
+        var options = new BatchingOptions(typeof(ScoreEvent));
+        Should.Throw<ArgumentOutOfRangeException>(() => options.CoalesceBy((Item x) => x.Name));
+    }
+
+    [Fact]
+    public void coalesce_by_installs_a_coalescing_batcher()
+    {
+        var options = new BatchingOptions(typeof(ScoreEvent));
+        options.CoalesceBy((ScoreEvent x) => x.AggregateId);
+        options.Batcher.ShouldBeOfType<CoalescingMessageBatcher<ScoreEvent, string>>();
+    }
+}
+
+public class coalescing_batch_processing_end_to_end : IAsyncLifetime
+{
+    private IHost theHost = null!;
+
+    public async Task InitializeAsync()
+    {
+        CoalescedScoreHandler.LastBatch = null;
+
+        #region sample_batch_coalesce_by
+        theHost = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.BatchMessagesOf<ScoreEvent>(batching =>
+                {
+                    batching.BatchSize = 500;
+                    batching.TriggerTime = 1.Seconds();
+
+                    // The handler only sees ONE ScoreEvent per AggregateId (the last one wins), instead
+                    // of recomputing the same aggregate many times. Every member message still settles
+                    // with the batch.
+                    batching.CoalesceBy((ScoreEvent x) => x.AggregateId);
+                }).Sequential();
+            }).StartAsync();
+        #endregion
+    }
+
+    public async Task DisposeAsync()
+    {
+        await theHost.StopAsync();
+        theHost.Dispose();
+    }
+
+    [Fact]
+    public async Task handler_only_sees_the_coalesced_messages()
+    {
+        Func<IMessageContext, Task> publish = async c =>
+        {
+            await c.PublishAsync(new ScoreEvent("A", 1));
+            await c.PublishAsync(new ScoreEvent("A", 2));
+            await c.PublishAsync(new ScoreEvent("A", 3));
+            await c.PublishAsync(new ScoreEvent("B", 1));
+            await c.PublishAsync(new ScoreEvent("B", 2));
+        };
+
+        await theHost.TrackActivity()
+            .WaitForMessageToBeReceivedAt<ScoreEvent[]>(theHost)
+            .ExecuteAndWaitAsync(publish);
+
+        var batch = CoalescedScoreHandler.LastBatch.ShouldNotBeNull();
+        batch.Length.ShouldBe(2);
+        batch.Single(x => x.AggregateId == "A").Version.ShouldBe(3);
+        batch.Single(x => x.AggregateId == "B").Version.ShouldBe(2);
+    }
+}
+
+public class batch_context_injection : IAsyncLifetime
+{
+    private IHost theHost = null!;
+
+    public async Task InitializeAsync()
+    {
+        BatchContextItemHandler.LastContext = null;
+
+        theHost = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.BatchMessagesOf<ContextItem>(batching =>
+                {
+                    batching.BatchSize = 500;
+                    batching.TriggerTime = 1.Seconds();
+                }).Sequential();
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await theHost.StopAsync();
+        theHost.Dispose();
+    }
+
+    [Fact]
+    public async Task batch_context_reports_the_batch_id_and_all_members()
+    {
+        Func<IMessageContext, Task> publish = async c =>
+        {
+            await c.PublishAsync(new ContextItem("one"));
+            await c.PublishAsync(new ContextItem("two"));
+            await c.PublishAsync(new ContextItem("three"));
+        };
+
+        await theHost.TrackActivity()
+            .WaitForMessageToBeReceivedAt<ContextItem[]>(theHost)
+            .ExecuteAndWaitAsync(publish);
+
+        var context = BatchContextItemHandler.LastContext.ShouldNotBeNull();
+        context.BatchId.ShouldNotBe(Guid.Empty);
+        context.Members.Count.ShouldBe(3);
+    }
+}
+
+#region sample_batch_coalesce_message
+public record ScoreEvent(string AggregateId, int Version);
+#endregion
+
+public static class CoalescedScoreHandler
+{
+    public static ScoreEvent[]? LastBatch;
+
+    public static void Handle(ScoreEvent[] events)
+    {
+        LastBatch = events;
+    }
+}
+
+public record ContextItem(string Name);
+
+public static class BatchContextItemHandler
+{
+    public static IBatchContext? LastContext;
+
+    #region sample_injecting_ibatchcontext
+    public static void Handle(ContextItem[] items, IBatchContext batch)
+    {
+        // batch.BatchId correlates all the log entries for this batch;
+        // batch.Members describes each original message that was grouped in.
+        LastContext = batch;
+    }
+    #endregion
+}
diff --git a/src/Wolverine/Runtime/Batching/BatchContextVariableSource.cs b/src/Wolverine/Runtime/Batching/BatchContextVariableSource.cs
new file mode 100644
index 000000000..c07383262
--- /dev/null
+++ b/src/Wolverine/Runtime/Batching/BatchContextVariableSource.cs
@@ -0,0 +1,50 @@
+using JasperFx.CodeGeneration;
+using JasperFx.CodeGeneration.Frames;
+using JasperFx.CodeGeneration.Model;
+using JasperFx.Core.Reflection;
+
+namespace Wolverine.Runtime.Batching;
+
+/// <summary>
+/// Makes <see cref="IBatchContext"/> injectable into any message handler, built from the active
+/// <see cref="Envelope"/> the same way <c>TenantId</c> and the "now" clock are supplied. In a batched
+/// handler the envelope is the assembled batch envelope, so the context reports the batch id and its
+/// members; in an ordinary handler the membership is simply empty.
+/// </summary>
+internal class BatchContextVariableSource : IVariableSource
+{
+    public bool Matches(Type type)
+    {
+        return type == typeof(IBatchContext);
+    }
+
+    public Variable Create(Type type)
+    {
+        return new BatchContextResolutionFrame().BatchContext;
+    }
+}
+
+internal class BatchContextResolutionFrame : SyncFrame
+{
+    private Variable _envelope = null!;
+
+    public BatchContextResolutionFrame()
+    {
+        BatchContext = new Variable(typeof(IBatchContext), "batchContext", this);
+    }
+
+    public Variable BatchContext { get; }
+
+    public override IEnumerable<Variable> FindVariables(IMethodVariables chain)
+    {
+        _envelope = chain.FindVariable(typeof(Envelope));
+        yield return _envelope;
+    }
+
+    public override void GenerateCode(GeneratedMethod method, ISourceWriter writer)
+    {
+        writer.WriteLine(
+            $"var {BatchContext.Usage} = {typeof(BatchContext).FullNameInCode()}.For({_envelope.Usage});");
+        Next?.GenerateCode(method, writer);
+    }
+}
diff --git a/src/Wolverine/Runtime/Batching/BatchingOptions.cs b/src/Wolverine/Runtime/Batching/BatchingOptions.cs
index 6e62f843f..2914894f4 100644
--- a/src/Wolverine/Runtime/Batching/BatchingOptions.cs
+++ b/src/Wolverine/Runtime/Batching/BatchingOptions.cs
@@ -35,6 +35,33 @@ public BatchingOptions(Type elementType)
 
     public IMessageBatcher Batcher { get; set; }
 
+    /// <summary>
+    /// De-duplicate the batched messages by a key so the handler only ever sees one message per
+    /// distinct key (the last message for that key wins). This is sugar over the <see cref="Batcher"/>
+    /// seam: it installs a <c>CoalescingMessageBatcher&lt;T,TKey&gt;</c>. It only changes what the
+    /// handler sees - every member envelope still settles with the batch, exactly like a normal batch.
+    /// The lambda parameter must be explicitly typed to the batched element type so both type
+    /// arguments can be inferred, e.g. <c>CoalesceBy((RecalculateScores x) =&gt; x.AggregateId)</c>.
+    /// </summary>
+    /// <param name="keySelector">Selects the de-duplication key from each message</param>
+    /// <typeparam name="T">The batched element type (must match <see cref="ElementType"/>)</typeparam>
+    /// <typeparam name="TKey">The de-duplication key type</typeparam>
+    public void CoalesceBy<T, TKey>(Func<T, TKey> keySelector)
+    {
+        if (keySelector == null)
+        {
+            throw new ArgumentNullException(nameof(keySelector));
+        }
+
+        if (typeof(T) != ElementType)
+        {
+            throw new ArgumentOutOfRangeException(nameof(keySelector),
+                $"The coalescing key selector must accept the batched element type '{ElementType.FullNameInCode()}', but accepted '{typeof(T).FullNameInCode()}'. Use an explicitly typed lambda parameter, e.g. CoalesceBy(({ElementType.Name} x) => x.SomeKey).");
+        }
+
+        Batcher = new CoalescingMessageBatcher<T, TKey>(keySelector);
+    }
+
     /// <summary>
     /// The maximum size of the message batch. Default is 100.
     /// </summary>
diff --git a/src/Wolverine/Runtime/Batching/CoalescingMessageBatcher.cs b/src/Wolverine/Runtime/Batching/CoalescingMessageBatcher.cs
new file mode 100644
index 000000000..21686bde2
--- /dev/null
+++ b/src/Wolverine/Runtime/Batching/CoalescingMessageBatcher.cs
@@ -0,0 +1,77 @@
+namespace Wolverine.Runtime.Batching;
+
+/// <summary>
+/// An <see cref="IMessageBatcher"/> that de-duplicates the batched messages by a user-supplied key so
+/// the handler only sees one message per distinct key (last message wins). Like
+/// <c>DefaultMessageBatcher&lt;T&gt;</c> it first groups by tenant id. Crucially it only changes
+/// <em>what the handler sees</em>: every member envelope still rides on the batch, so the transactional
+/// inbox/outbox settlement and dead-lettering behavior are identical to a non-coalescing batch.
+/// </summary>
+internal class CoalescingMessageBatcher<T, TKey> : IMessageBatcher
+{
+    private readonly Func<T, TKey> _keySelector;
+
+    public CoalescingMessageBatcher(Func<T, TKey> keySelector)
+    {
+        _keySelector = keySelector ?? throw new ArgumentNullException(nameof(keySelector));
+    }
+
+    public IEnumerable<Envelope> Group(IReadOnlyList<Envelope> envelopes)
+    {
+        // Group by tenant id exactly like DefaultMessageBatcher<T>. Empty string is the sentinel for a
+        // null TenantId because Dictionary<string, ...> does not allow null keys.
+        var groups = new Dictionary<string, List<Envelope>>();
+        foreach (var envelope in envelopes)
+        {
+            var tenant = envelope.TenantId ?? string.Empty;
+            if (!groups.TryGetValue(tenant, out var list))
+            {
+                list = new List<Envelope>();
+                groups[tenant] = list;
+            }
+
+            list.Add(envelope);
+        }
+
+        foreach (var group in groups)
+        {
+            // Coalesce by the user's key (last wins) for the array the handler SEES... Key on object so
+            // TKey stays unconstrained (a nullable key type is allowed); null keys are handled below.
+            var indexByKey = new Dictionary<object, int>();
+            var coalesced = new List<T>(group.Value.Count);
+            foreach (var envelope in group.Value)
+            {
+                if (envelope.Message is not T typed)
+                {
+                    continue;
+                }
+
+                var key = _keySelector(typed);
+
+                // A null key can't index a Dictionary; never coalesce null-keyed items - keep each.
+                if (key is not null && indexByKey.TryGetValue(key, out var index))
+                {
+                    coalesced[index] = typed;
+                }
+                else
+                {
+                    if (key is not null)
+                    {
+                        indexByKey[key] = coalesced.Count;
+                    }
+
+                    coalesced.Add(typed);
+                }
+            }
+
+            // ...but EVERY member envelope stays on the batch, so settlement (inbox/outbox tracking and
+            // dead-lettering) is identical to today. Only what the handler sees was reduced.
+            yield return new Envelope(coalesced.ToArray(), group.Value)
+            {
+                TenantId = group.Key.Length == 0 ? null : group.Key
+            };
+        }
+    }
+
+    public Type BatchMessageType => typeof(T[]);
+}
diff --git a/src/Wolverine/Runtime/Batching/IBatchContext.cs b/src/Wolverine/Runtime/Batching/IBatchContext.cs
new file mode 100644
index 000000000..fc8a9d658
--- /dev/null
+++ b/src/Wolverine/Runtime/Batching/IBatchContext.cs
@@ -0,0 +1,83 @@
+namespace Wolverine.Runtime.Batching;
+
+/// <summary>
+/// Read-only identity and membership information about the batch currently being processed by a
+/// batched message handler. Inject this into a <c>Handle(T[])</c> handler to correlate log entries,
+/// emit batch-level metrics, or make per-batch decisions. This is purely informational: it never
+/// changes what gets acknowledged or how the batch is settled.
+/// </summary>
+public interface IBatchContext
+{
+    /// <summary>
+    /// The unique id of the batch envelope Wolverine assembled for this batch.
+    /// </summary>
+    Guid BatchId { get; }
+
+    /// <summary>
+    /// Information about each original member message that was grouped into this batch. When
+    /// <see cref="Wolverine.WolverineOptions.BatchMessagesOf{T}"/> is configured with
+    /// <see cref="Wolverine.Runtime.Batching.BatchingOptions.CoalesceBy{T,TKey}"/>, every member
+    /// envelope is still listed here (settlement is unchanged) even though the handler only sees the
+    /// coalesced messages.
+    /// </summary>
+    IReadOnlyList<BatchMemberInfo> Members { get; }
+}
+
+/// <summary>
+/// Read-only metadata about a single member message inside a batch.
+/// </summary>
+public sealed class BatchMemberInfo
+{
+    public BatchMemberInfo(Guid messageId, int attempts, DateTimeOffset sentAt)
+    {
+        MessageId = messageId;
+        Attempts = attempts;
+        SentAt = sentAt;
+    }
+
+    /// <summary>The Wolverine message id of the original member envelope.</summary>
+    public Guid MessageId { get; }
+
+    /// <summary>The number of delivery attempts recorded for the member envelope.</summary>
+    public int Attempts { get; }
+
+    /// <summary>When the member envelope was originally sent.</summary>
+    public DateTimeOffset SentAt { get; }
+}
+
+// Public because the generated handler code (which has no access to Wolverine internals) calls the
+// For factory to read the internal Envelope.Batch; the constructor stays private.
+public sealed class BatchContext : IBatchContext
+{
+    /// <summary>
+    /// Build an <see cref="IBatchContext"/> from the active batch envelope. The batch envelope's own
+    /// id becomes <see cref="BatchId"/>; its <c>Batch</c> members become <see cref="Members"/>. If the
+    /// envelope is not actually a batch (no members), the context reports an empty membership.
+    /// </summary>
+    public static IBatchContext For(Envelope envelope)
+    {
+        var members = envelope.Batch;
+        if (members == null || members.Length == 0)
+        {
+            return new BatchContext(envelope.Id, Array.Empty<BatchMemberInfo>());
+        }
+
+        var infos = new BatchMemberInfo[members.Length];
+        for (var i = 0; i < members.Length; i++)
+        {
+            var member = members[i];
+            infos[i] = new BatchMemberInfo(member.Id, member.Attempts, member.SentAt);
+        }
+
+        return new BatchContext(envelope.Id, infos);
+    }
+
+    private BatchContext(Guid batchId, IReadOnlyList<BatchMemberInfo> members)
+    {
+        BatchId = batchId;
+        Members = members;
+    }
+
+    public Guid BatchId { get; }
+    public IReadOnlyList<BatchMemberInfo> Members { get; }
+}
diff --git a/src/Wolverine/WolverineOptions.cs b/src/Wolverine/WolverineOptions.cs
index 2fd66e2ab..3cce9a6ec 100644
--- a/src/Wolverine/WolverineOptions.cs
+++ b/src/Wolverine/WolverineOptions.cs
@@ -238,6 +238,7 @@ public WolverineOptions(string? assemblyName)
         CodeGeneration = new GenerationRules("Internal.Generated");
         CodeGeneration.Sources.Add(new NowTimeVariableSource());
         CodeGeneration.Sources.Add(new TenantIdSource());
+        CodeGeneration.Sources.Add(new Runtime.Batching.BatchContextVariableSource());
 
         // MassTransit shim variable sources
         CodeGeneration.Sources.Add(new Shims.MassTransit.ConsumeContextVariableSource());