diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,77 @@
 # Changelog
 
+## 0.13.0.0 — 2026-11-12
+
+### Breaking Changes
+
+- `Adapter` gains a strict `telemetry :: AdapterTelemetry` field. Record
+  constructions must add `telemetry = defaultAdapterTelemetry`; positional
+  construction requires the extra argument. Adapters should expose the value in
+  their own configuration so users can override it.
+- Per-message processing spans now carry a link to the producer's span when
+  `Envelope.traceContext` decodes. Under the default `ParentAndLink` policy the span
+  is still the producer span's child, so existing traces keep their shape and gain
+  one link.
+- Batch processing spans are always roots and no longer join the first member's
+  trace. They link every member whose trace headers decode, in member order, each
+  link carrying `messaging.message.id`, up to 128 links. Valid contexts beyond
+  that are counted in `shibuya.batch.links.dropped`. Batch spans also carry
+  `messaging.batch.message_count`.
+- Processing and batch spans are renamed from `<destination> process` to
+  `process <destination>`, following the messaging conventions'
+  `{messaging.operation.name} {destination}`. Saved searches, alerts and sampling
+  rules that match the span or operation name must change: a Jaeger or Tempo
+  search for operation `orders process` becomes `process orders`. Lifecycle span
+  names are unchanged.
+
+### New Features
+
+- Processing and batch spans carry `messaging.operation.name = "process"`, which
+  the conventions mark Required, alongside `messaging.operation.type`.
+  `Shibuya.Telemetry.Semantic` exports `attrMessagingOperationName`.
+- New module `Shibuya.Telemetry.Correlation`, re-exported from `Shibuya.Adapter`,
+  `Shibuya.Telemetry` and `Shibuya`. It provides `TraceCorrelation`
+  (`ParentAndLink`, `LinkOnly`), `AdapterTelemetry` (constructor hidden, built from
+  `defaultAdapterTelemetry` by record update), and the link helpers `creationLink`,
+  `creationLinkWith` and `consumerSpanArgsLinkedTo`.
+- `LinkOnly` makes each processing span with a usable creation context the root of
+  a new trace, related to the producer only by the link. Use it for replayed,
+  fanned-out or long-delayed sources and with tail-based sampling. Without usable
+  headers both policies keep the ambient parent and add no link.
+- New module `Shibuya.Telemetry.Process` with `withProcessSpan` (in `Tracing`) and
+  `withProcessSpanIO` (for a plain `Tracer`). They open the same processing span
+  as the runner for code that drains an adapter outside it. With valid headers they
+  neither parent on nor link an ambient span; they are meant for independent
+  message processing.
+- `Shibuya.Telemetry.Correlation` also exports `maxBatchLinks`, `BatchLinks`,
+  `collectBatchLinks` and `collectBatchLinksBy`, and `Shibuya.Telemetry.Semantic`
+  exports `attrMessagingBatchMessageCount` and `attrShibuyaBatchLinksDropped`.
+- `Shibuya.Telemetry.Effect.withRootContext` runs an action with an empty
+  OpenTelemetry context attached.
+
+### Other Changes
+
+- With tracing disabled, per-message processing no longer decodes the W3C trace
+  headers. Before, each message with a `traceparent` was decoded even under
+  `runTracingNoop`.
+- Per-message processing allocates about 208 bytes less on every single-message
+  path. With tracing enabled and trace headers present it allocates about 210 bytes
+  less, even with the added link, because the context attachment and the span now
+  share one unlift.
+- Accept a measured allocation increase on `Async` runners whose handlers wait:
+  `concurrency-modes` `async-5` cells allocate 20–30% more than 0.12.0.0 (for
+  example 1.17–1.21 MB to 1.47–1.57 MB for 100 messages with a 1 ms handler delay),
+  `async-10` 4–5% and `async-20` 2–3% more, with unchanged time. `async-2`, every
+  `Ahead` cell and `hot-path/async8` are flat or lower. The per-message code these
+  cells run is the path that got cheaper everywhere else; the increase comes from
+  streamly's worker dispatch, which is sensitive to the shape of per-message work
+  (see `docs/plans/30-investigate-and-reduce-the-async-ahead-concurrency-allocation-regression.md`).
+- The release comparison against 0.12.0.0 measured 10,000-message processing
+  (`hot-path/serial-noop-10000`) at 9.247 to 9.098 ms, with allocation falling
+  from 19,918,565 to 17,837,870 bytes. Every timing comparison passed the 10% gate
+  in interleaved confirming runs at `--stdev 3`; measurements used package-default
+  RTS settings on a shared workstation.
+
 ## 0.12.0.0 — 2026-10-26
 
 ### Breaking Changes
diff --git a/shibuya-core.cabal b/shibuya-core.cabal
--- a/shibuya-core.cabal
+++ b/shibuya-core.cabal
@@ -1,6 +1,6 @@
 cabal-version: 3.12
 name: shibuya-core
-version: 0.12.0.0
+version: 0.13.0.0
 synopsis: Supervised queue processing framework for Haskell
 description:
   A supervised queue processing framework inspired by Broadway (Elixir).
@@ -47,7 +47,9 @@
     Shibuya.Policy
     Shibuya.Stream
     Shibuya.Telemetry
+    Shibuya.Telemetry.Correlation
     Shibuya.Telemetry.Effect
+    Shibuya.Telemetry.Process
     Shibuya.Telemetry.Propagation
     Shibuya.Telemetry.Semantic
 
@@ -132,6 +134,8 @@
     Shibuya.Runner.PauseSpec
     Shibuya.Runner.SupervisedSpec
     Shibuya.RunnerSpec
+    Shibuya.Telemetry.BatchCorrelationSpec
+    Shibuya.Telemetry.CorrelationSpec
     Shibuya.Telemetry.EffectSpec
     Shibuya.Telemetry.LifecycleSpec
     Shibuya.Telemetry.PropagationSpec
@@ -195,3 +199,27 @@
     shibuya-core,
     streamly-core,
     unliftio,
+
+-- Trace retention probes for the trace correlation work. Process-isolated for
+-- the same reason as the GC suites: weak-pointer and live-bytes assertions
+-- need nothing else in the process holding references (ADR 0001).
+test-suite shibuya-core-trace-retention-test
+  import: warnings
+  default-language: GHC2024
+  type: exitcode-stdio-1.0
+  hs-source-dirs: test-retention
+  main-is: Main.hs
+  ghc-options:
+    -threaded
+    -rtsopts
+    "-with-rtsopts=-N2 -T"
+
+  build-depends:
+    base ^>=4.21.0.0,
+    bytestring,
+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),
+    hs-opentelemetry-api,
+    hs-opentelemetry-exporter-in-memory ^>=1.0,
+    shibuya-core,
+    streamly-core,
+    text,
diff --git a/src/Shibuya.hs b/src/Shibuya.hs
--- a/src/Shibuya.hs
+++ b/src/Shibuya.hs
@@ -76,6 +76,9 @@
 
     -- * Adapter authoring
     Adapter (..),
+    AdapterTelemetry (traceCorrelation),
+    defaultAdapterTelemetry,
+    TraceCorrelation (..),
     AckHandle (..),
     Lease (..),
     Ingested (..),
@@ -114,7 +117,7 @@
   )
 where
 
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry)
 import Shibuya.App
   ( AppConfig (..),
     AppError (..),
diff --git a/src/Shibuya/Adapter.hs b/src/Shibuya/Adapter.hs
--- a/src/Shibuya/Adapter.hs
+++ b/src/Shibuya/Adapter.hs
@@ -3,12 +3,18 @@
 -- Adapter owns queue semantics; runner never touches offsets directly.
 module Shibuya.Adapter
   ( Adapter (..),
+
+    -- * Telemetry options
+    AdapterTelemetry (traceCorrelation),
+    defaultAdapterTelemetry,
+    TraceCorrelation (..),
   )
 where
 
 import Data.Text (Text)
 import Effectful (Eff)
 import Shibuya.Core.Ingested (Ingested)
+import Shibuya.Telemetry.Correlation (AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry)
 import Streamly.Data.Stream (Stream)
 
 -- | Queue adapter interface.
@@ -19,5 +25,10 @@
     -- | Stream of leased messages
     source :: Stream (Eff es) (Ingested es msg),
     -- | Stop polling, release resources
-    shutdown :: Eff es ()
+    shutdown :: Eff es (),
+    -- | Telemetry options for this adapter's processing spans, such as the
+    -- 'TraceCorrelation' policy. Start from 'defaultAdapterTelemetry' and
+    -- expose the value in the adapter's own configuration so users can
+    -- override it. Read once per processor.
+    telemetry :: !AdapterTelemetry
   }
diff --git a/src/Shibuya/Adapter/Mock.hs b/src/Shibuya/Adapter/Mock.hs
--- a/src/Shibuya/Adapter/Mock.hs
+++ b/src/Shibuya/Adapter/Mock.hs
@@ -16,7 +16,7 @@
 
 import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)
 import Effectful (Eff, IOE, liftIO, (:>))
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.Core.Ack (AckDecision)
 import Shibuya.Core.AckHandle (AckHandle (..))
 import Shibuya.Core.Ingested (Ingested, mkIngested)
@@ -30,7 +30,8 @@
   Adapter
     { adapterName = "mock:list",
       source = Stream.fromList msgs,
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
 
 -- | Tracking state for ack decisions.
diff --git a/src/Shibuya/Internal/Runner/BatchProcessor.hs b/src/Shibuya/Internal/Runner/BatchProcessor.hs
--- a/src/Shibuya/Internal/Runner/BatchProcessor.hs
+++ b/src/Shibuya/Internal/Runner/BatchProcessor.hs
@@ -87,22 +87,26 @@
 import Shibuya.Internal.Runner.KeyedScheduler (runKeyedScheduler)
 import Shibuya.Policy (Concurrency (..))
 import Shibuya.Prelude
+import Shibuya.Telemetry.Correlation (BatchLinks (..), collectBatchLinksBy, consumerSpanArgsLinkedTo)
 import Shibuya.Telemetry.Effect
   ( Tracing,
     addAttribute,
     addAttributes,
     addEvent,
+    isTracingEnabled,
     recordException,
     setStatus,
-    withExtractedContext,
+    withRootContext,
     withSpan',
   )
-import Shibuya.Telemetry.Propagation (extractTraceContext)
 import Shibuya.Telemetry.Semantic
-  ( attrMessagingDestinationName,
+  ( attrMessagingBatchMessageCount,
+    attrMessagingDestinationName,
     attrMessagingOperation,
+    attrMessagingOperationName,
     attrMessagingSystem,
     attrShibuyaBatchKey,
+    attrShibuyaBatchLinksDropped,
     attrShibuyaBatchSize,
     attrShibuyaBatchTrigger,
     attrShibuyaInflightCount,
@@ -184,138 +188,151 @@
   (BatchInfo, NonEmpty (Ingested es msg)) ->
   Eff es ()
 processCurrentBatchWithDetail detailed metricsHandle procId maxConc stopSignal exitPublisher handler (info, batch) = do
-  -- Use the first message's trace context as the batch span's parent. A batch
-  -- may span several traces; picking the first is a pragmatic single parent
-  -- (full fan-in links are a later refinement).
-  let firstMsg = NE.head batch
-      parentCtx = firstMsg.envelope.traceContext >>= extractTraceContext
-      ProcessorId pidText = procId
+  enabled <- isTracingEnabled
+  let ProcessorId pidText = procId
+      spanName = processSpanName pidText
+      -- A batch has many producers and a span has one parent, so the batch
+      -- span is always a root and links each member's creation context
+      -- instead (ADR 0014). Disabled tracing reads no member headers.
+      openBatchSpan body
+        | not enabled = withSpan' spanName consumerSpanArgs body
+        | otherwise =
+            let !batchLinks = collectBatchLinksBy (.envelope.messageId) (.envelope.traceContext) batch
+                !args = case batchLinks.links of
+                  [] -> consumerSpanArgs
+                  links -> consumerSpanArgsLinkedTo links
+             in withRootContext $
+                  withSpan' spanName args $ \traceSpan -> do
+                    when (batchLinks.dropped > 0) $
+                      addAttribute traceSpan attrShibuyaBatchLinksDropped batchLinks.dropped
+                    body traceSpan
 
-  withExtractedContext parentCtx $
-    withSpan' (processSpanName pidText) consumerSpanArgs $ \traceSpan -> do
-      -- Framework messaging.* attributes plus batch-scoped attributes.
-      let BatchKey keyText = info.batchKey
-          frameworkAttrs =
-            HashMap.fromList
-              [ (attrMessagingSystem, toAttribute ("shibuya" :: Text)),
-                (attrMessagingDestinationName, toAttribute pidText),
-                (attrMessagingOperation, toAttribute ("process" :: Text)),
-                (attrShibuyaBatchKey, toAttribute keyText),
-                (attrShibuyaBatchSize, toAttribute info.size),
-                (attrShibuyaBatchTrigger, toAttribute (triggerText info.trigger))
-              ]
-      addAttributes traceSpan frameworkAttrs
+  openBatchSpan $ \traceSpan -> do
+    -- Framework messaging.* attributes plus batch-scoped attributes.
+    let BatchKey keyText = info.batchKey
+        frameworkAttrs =
+          HashMap.fromList
+            [ (attrMessagingSystem, toAttribute ("shibuya" :: Text)),
+              (attrMessagingDestinationName, toAttribute pidText),
+              (attrMessagingOperation, toAttribute ("process" :: Text)),
+              (attrMessagingOperationName, toAttribute ("process" :: Text)),
+              (attrShibuyaBatchKey, toAttribute keyText),
+              (attrShibuyaBatchSize, toAttribute info.size),
+              (attrMessagingBatchMessageCount, toAttribute info.size),
+              (attrShibuyaBatchTrigger, toAttribute (triggerText info.trigger))
+            ]
+    addAttributes traceSpan frameworkAttrs
 
-      let process currentInflight = do
-            addAttribute traceSpan attrShibuyaInflightCount currentInflight
-            addAttribute traceSpan attrShibuyaInflightMax maxConc
+    let process currentInflight = do
+          addAttribute traceSpan attrShibuyaInflightCount currentInflight
+          addAttribute traceSpan attrShibuyaInflightMax maxConc
 
-            addEvent traceSpan (mkEvent eventBatchStarted [])
+          addEvent traceSpan (mkEvent eventBatchStarted [])
 
-            terminal <- liftIO $ readProcessorExit stopSignal
+          terminal <- liftIO $ readProcessorExit stopSignal
 
-            -- Run the handler under exception isolation. On any exception, record it
-            -- on the span and substitute the whole-batch retry default.
-            (handlerResult, skippedAfterHalt) <-
-              case terminal of
-                Just _ -> pure (Left (), True)
-                Nothing -> do
-                  result <-
-                    catchAny
-                      (Right <$> handler info (toMessage <$> batch))
-                      ( \ex -> do
-                          recordException traceSpan ex
-                          pure (Left ())
-                      )
-                  pure (result, False)
-            let (resolvedAck, handlerThrew) = case handlerResult of
-                  Right a -> (a, False)
-                  Left () -> (ackAll (AckRetry (RetryDelay 0)), True)
+          -- Run the handler under exception isolation. On any exception, record it
+          -- on the span and substitute the whole-batch retry default.
+          (handlerResult, skippedAfterHalt) <-
+            case terminal of
+              Just _ -> pure (Left (), True)
+              Nothing -> do
+                result <-
+                  catchAny
+                    (Right <$> handler info (toMessage <$> batch))
+                    ( \ex -> do
+                        recordException traceSpan ex
+                        pure (Left ())
+                    )
+                pure (result, False)
+          let (resolvedAck, handlerThrew) = case handlerResult of
+                Right a -> (a, False)
+                Left () -> (ackAll (AckRetry (RetryDelay 0)), True)
 
-            -- RELIABLE FINALIZATION: iterate OUR OWN retained list, never the
-            -- handler's output. For each retained message, choose its decision once
-            -- via findWithDefault, then call the idempotent adapter finalizer with
-            -- bounded retry. Do not let one adapter failure prevent attempts for the
-            -- rest of the batch.
-            results <-
-              mapM
-                ( \ingested -> do
-                    let explicitDecision = Map.lookup ingested.envelope.messageId resolvedAck.decisions
-                        d = fromMaybe resolvedAck.fallback explicitDecision
-                    finalResult <- finalizeWithRetry traceSpan ingested d
-                    let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor
-                    when detailed $ case (finalResult, d) of
-                      (Right (), AckOk) -> recordCursor
-                      (Right (), AckDeadLetter _) -> recordCursor
-                      _ -> pure ()
-                    pure (ingested.envelope.messageId, isJust explicitDecision, d, finalResult)
-                )
-                (NE.toList batch)
-            let decisions = [d | (_, _, d, _) <- results]
-                finalizeFailures = [(mid, ex) | (mid, _, _, Left ex) <- results]
+          -- RELIABLE FINALIZATION: iterate OUR OWN retained list, never the
+          -- handler's output. For each retained message, choose its decision once
+          -- via findWithDefault, then call the idempotent adapter finalizer with
+          -- bounded retry. Do not let one adapter failure prevent attempts for the
+          -- rest of the batch.
+          results <-
+            mapM
+              ( \ingested -> do
+                  let explicitDecision = Map.lookup ingested.envelope.messageId resolvedAck.decisions
+                      d = fromMaybe resolvedAck.fallback explicitDecision
+                  finalResult <- finalizeWithRetry traceSpan ingested d
+                  let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor
+                  when detailed $ case (finalResult, d) of
+                    (Right (), AckOk) -> recordCursor
+                    (Right (), AckDeadLetter _) -> recordCursor
+                    _ -> pure ()
+                  pure (ingested.envelope.messageId, isJust explicitDecision, d, finalResult)
+              )
+              (NE.toList batch)
+          let decisions = [d | (_, _, d, _) <- results]
+              finalizeFailures = [(mid, ex) | (mid, _, _, Left ex) <- results]
 
-            -- Compute halt and partial-failure signals from the resolved decisions.
-            let finalizationFailure =
-                  case finalizeFailures of
-                    [] -> Nothing
-                    failed@((firstMessageId, _) : _) ->
-                      Just $
-                        ProcessorFailed
-                          ( "batch finalization failed for message ids: "
-                              <> Text.intercalate ", " [tshow mid | (mid, _) <- failed]
-                          )
-                          (Just firstMessageId)
-                firstHalt = listToMaybe [r | AckHalt r <- decisions]
-                requestedExit = finalizationFailure <|> (ProcessorHalted <$> firstHalt)
-                overrideFailures =
-                  [ ()
-                  | (_, explicitlyNamed, d, _) <- results,
-                    explicitlyNamed,
-                    isFailing d
-                  ]
-                partialInc = not handlerThrew && not (null overrideFailures)
+          -- Compute halt and partial-failure signals from the resolved decisions.
+          let finalizationFailure =
+                case finalizeFailures of
+                  [] -> Nothing
+                  failed@((firstMessageId, _) : _) ->
+                    Just $
+                      ProcessorFailed
+                        ( "batch finalization failed for message ids: "
+                            <> Text.intercalate ", " [tshow mid | (mid, _) <- failed]
+                        )
+                        (Just firstMessageId)
+              firstHalt = listToMaybe [r | AckHalt r <- decisions]
+              requestedExit = finalizationFailure <|> (ProcessorHalted <$> firstHalt)
+              overrideFailures =
+                [ ()
+                | (_, explicitlyNamed, d, _) <- results,
+                  explicitlyNamed,
+                  isFailing d
+                ]
+              partialInc = not handlerThrew && not (null overrideFailures)
 
-            -- Span status: error on halt or exception, otherwise Ok.
-            addEvent traceSpan (mkEvent eventBatchCompleted [])
-            case requestedExit of
-              Just processorExit -> setStatus traceSpan (OTel.Error (processorExitText processorExit))
-              Nothing ->
-                if skippedAfterHalt
-                  then setStatus traceSpan (OTel.Error "batch skipped after halt")
-                  else
-                    if handlerThrew
-                      then setStatus traceSpan (OTel.Error "batch handler exception")
-                      else setStatus traceSpan OTel.Ok
+          -- Span status: error on halt or exception, otherwise Ok.
+          addEvent traceSpan (mkEvent eventBatchCompleted [])
+          case requestedExit of
+            Just processorExit -> setStatus traceSpan (OTel.Error (processorExitText processorExit))
+            Nothing ->
+              if skippedAfterHalt
+                then setStatus traceSpan (OTel.Error "batch skipped after halt")
+                else
+                  if handlerThrew
+                    then setStatus traceSpan (OTel.Error "batch handler exception")
+                    else setStatus traceSpan OTel.Ok
 
-            traverse_ (recordException traceSpan . snd) finalizeFailures
+          traverse_ (recordException traceSpan . snd) finalizeFailures
 
-            -- Record metrics: decrement in-flight, add per-message counter deltas,
-            -- advance batch counters, set Failed state on halt or exhausted
-            -- finalization retry.
-            liftIO $
-              recordBatchOutcomeMetrics
-                metricsHandle
-                (triggerMetric info.trigger)
-                info.size
-                handlerThrew
-                partialInc
-                (decisionMetric <$> decisions)
-                (processorExitText <$> requestedExit)
+          -- Record metrics: decrement in-flight, add per-message counter deltas,
+          -- advance batch counters, set Failed state on halt or exhausted
+          -- finalization retry.
+          liftIO $
+            recordBatchOutcomeMetrics
+              metricsHandle
+              (triggerMetric info.trigger)
+              info.size
+              handlerThrew
+              partialInc
+              (decisionMetric <$> decisions)
+              (processorExitText <$> requestedExit)
 
-            -- Halt: set the shared flag; do NOT throw (let the stream drain).
-            for_ requestedExit $ \processorExit ->
-              liftIO $ requestProcessorExit exitPublisher processorExit
-      if not detailed
-        then do
-          currentInflight <- liftIO $ beginProcessing metricsHandle maxConc
-          process currentInflight
-        else Exception.mask $ \restore -> do
-          currentInflight <- liftIO $ beginProcessing metricsHandle maxConc
-          slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle firstMsg.envelope.messageId
-          restore (process currentInflight)
-            `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)
-          liftIO (Metrics.releaseInFlightSlot metricsHandle slot)
-            `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)
+          -- Halt: set the shared flag; do NOT throw (let the stream drain).
+          for_ requestedExit $ \processorExit ->
+            liftIO $ requestProcessorExit exitPublisher processorExit
+    if not detailed
+      then do
+        currentInflight <- liftIO $ beginProcessing metricsHandle maxConc
+        process currentInflight
+      else Exception.mask $ \restore -> do
+        currentInflight <- liftIO $ beginProcessing metricsHandle maxConc
+        slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle (NE.head batch).envelope.messageId
+        restore (process currentInflight)
+          `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)
+        liftIO (Metrics.releaseInFlightSlot metricsHandle slot)
+          `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)
   where
     isFailing :: AckDecision -> Bool
     isFailing (AckDeadLetter _) = True
diff --git a/src/Shibuya/Internal/Runner/Supervised.hs b/src/Shibuya/Internal/Runner/Supervised.hs
--- a/src/Shibuya/Internal/Runner/Supervised.hs
+++ b/src/Shibuya/Internal/Runner/Supervised.hs
@@ -109,6 +109,7 @@
 import Shibuya.Internal.Runner.Pause (PauseHandle, newPauseHandle)
 import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))
 import Shibuya.Prelude
+import Shibuya.Telemetry.Correlation (AdapterTelemetry (traceCorrelation), TraceCorrelation)
 import Shibuya.Telemetry.Effect
   ( Tracing,
     addAttribute,
@@ -116,21 +117,19 @@
     addEvent,
     recordException,
     setStatus,
-    withExtractedContext,
-    withSpan',
   )
-import Shibuya.Telemetry.Propagation (extractTraceContext)
+import Shibuya.Telemetry.Process (withProcessSpan)
 import Shibuya.Telemetry.Semantic
   ( attrMessagingDestinationName,
     attrMessagingMessageId,
     attrMessagingOperation,
+    attrMessagingOperationName,
     attrMessagingSystem,
     attrShibuyaAckDecision,
     attrShibuyaDeadLetterReasonCode,
     attrShibuyaInflightCount,
     attrShibuyaInflightMax,
     attrShibuyaPartition,
-    consumerSpanArgs,
     eventAckDecision,
     eventHandlerCompleted,
     eventHandlerStarted,
@@ -308,6 +307,9 @@
   -- Create bounded inbox (this is where inboxSize is used for backpressure)
   inbox <- liftIO $ newBoundedInbox inboxSize
 
+  -- The adapter's trace correlation policy, read once per processor.
+  let !correlation = adapter.telemetry.traceCorrelation
+
   -- Signal when ingester completes (stream exhausted)
   streamDoneVar <- liftIO $ newTVarIO False
 
@@ -348,8 +350,8 @@
       -- the detail ownership action from the basic specialization. Selecting a
       -- partially applied worker retains a per-message closure.
       case Metrics.metricsDetailLevel metricsHandle of
-        Metrics.MetricsBasic -> runInIO $ processUntilDrained False metricsHandle procId ordering concurrency handler inbox streamDoneVar
-        Metrics.MetricsDetailed _ -> runInIO $ processUntilDrained True metricsHandle procId ordering concurrency handler inbox streamDoneVar
+        Metrics.MetricsBasic -> runInIO $ processUntilDrained False metricsHandle procId correlation ordering concurrency handler inbox streamDoneVar
+        Metrics.MetricsDetailed _ -> runInIO $ processUntilDrained True metricsHandle procId correlation ordering concurrency handler inbox streamDoneVar
       UIO.waitCatch ingesterAsync >>= \case
         Left ingesterErr -> do
           now <- getCurrentTime
@@ -613,13 +615,14 @@
   Bool ->
   MetricsHandle ->
   ProcessorId ->
+  TraceCorrelation ->
   OrderingPolicy ->
   Concurrency ->
   Handler es msg ->
   Inbox (Ingested es msg) ->
   TVar Bool ->
   Eff es ()
-processUntilDrained detailed metricsHandle procId ordering concurrency handler inbox streamDoneVar = do
+processUntilDrained detailed metricsHandle procId correlation ordering concurrency handler inbox streamDoneVar = do
   stopSignal <- liftIO newProcessorSignal
   let exitPublisher = newProcessorExitPublisherWithWake stopSignal streamDoneVar
 
@@ -633,13 +636,14 @@
         HashMap.fromList
           [ (attrMessagingSystem, toAttribute ("shibuya" :: Text)),
             (attrMessagingDestinationName, toAttribute pidText),
-            (attrMessagingOperation, toAttribute ("process" :: Text))
+            (attrMessagingOperation, toAttribute ("process" :: Text)),
+            (attrMessagingOperationName, toAttribute ("process" :: Text))
           ]
 
   withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
     let inboxStream = inboxToStream inbox streamDoneVar stopSignal
         runProcessAction ingested =
-          runInIO $ processOne detailed metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested
+          runInIO $ processOne detailed metricsHandle correlation spanName constantFrameworkAttrs maxConc exitPublisher handler ingested
 
         -- Restore normal interruptibility for handlers and finalizers while the
         -- concurrent Streamly scheduler retains its inherited mask.
@@ -694,6 +698,7 @@
   (IOE :> es, Tracing :> es) =>
   Bool ->
   MetricsHandle ->
+  TraceCorrelation ->
   Text ->
   HashMap.HashMap Text Attribute ->
   Int ->
@@ -701,146 +706,144 @@
   Handler es msg ->
   Ingested es msg ->
   Eff es ()
-processOne detailed metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested = do
-  -- Extract parent context from message headers for distributed tracing
-  let parentCtx = ingested.envelope.traceContext >>= extractTraceContext
+processOne detailed metricsHandle correlation spanName constantFrameworkAttrs maxConc exitPublisher handler ingested =
+  -- Link to (and, by policy, parent on) the producer's span from the
+  -- message's trace headers. With tracing disabled the headers are not read.
+  withProcessSpan correlation spanName ingested.envelope.traceContext $ \traceSpan -> do
+    -- Build the messaging.* attribute set: framework defaults first,
+    -- then the adapter's HashMap layered on top so adapter keys with
+    -- the same name override the framework's default. The explicit
+    -- 'HashMap.union' here is left-biased (left wins), which keeps
+    -- the precedence rule local and obvious instead of relying on
+    -- the order of repeated 'addAttribute' / 'addAttributes' calls
+    -- against the underlying mutable Span.
+    let messageId@(MessageId msgIdText) = ingested.envelope.messageId
+        frameworkAttrs =
+          HashMap.insert attrMessagingMessageId (toAttribute msgIdText) $
+            case ingested.envelope.partition of
+              Just p -> HashMap.insert attrShibuyaPartition (toAttribute p) constantFrameworkAttrs
+              Nothing -> constantFrameworkAttrs
+        mergedAttrs = HashMap.union ingested.envelope.attributes frameworkAttrs
+    addAttributes traceSpan mergedAttrs
 
-  withExtractedContext parentCtx $
-    withSpan' spanName consumerSpanArgs $ \traceSpan -> do
-      -- Build the messaging.* attribute set: framework defaults first,
-      -- then the adapter's HashMap layered on top so adapter keys with
-      -- the same name override the framework's default. The explicit
-      -- 'HashMap.union' here is left-biased (left wins), which keeps
-      -- the precedence rule local and obvious instead of relying on
-      -- the order of repeated 'addAttribute' / 'addAttributes' calls
-      -- against the underlying mutable Span.
-      let messageId@(MessageId msgIdText) = ingested.envelope.messageId
-          frameworkAttrs =
-            HashMap.insert attrMessagingMessageId (toAttribute msgIdText) $
-              case ingested.envelope.partition of
-                Just p -> HashMap.insert attrShibuyaPartition (toAttribute p) constantFrameworkAttrs
-                Nothing -> constantFrameworkAttrs
-          mergedAttrs = HashMap.union ingested.envelope.attributes frameworkAttrs
-      addAttributes traceSpan mergedAttrs
+    status <- ingested.deliveryStatus
+    case status of
+      DeliverySuperseded -> do
+        addEvent traceSpan $
+          mkEvent
+            eventAckDecision
+            [(attrShibuyaAckDecision, OTel.toAttribute ("superseded" :: Text))]
+        setStatus traceSpan OTel.Ok
+      DeliveryCurrent -> do
+        let process currentInflight = do
+              addAttribute traceSpan attrShibuyaInflightCount currentInflight
+              addAttribute traceSpan attrShibuyaInflightMax maxConc
 
-      status <- ingested.deliveryStatus
-      case status of
-        DeliverySuperseded -> do
-          addEvent traceSpan $
-            mkEvent
-              eventAckDecision
-              [(attrShibuyaAckDecision, OTel.toAttribute ("superseded" :: Text))]
-          setStatus traceSpan OTel.Ok
-        DeliveryCurrent -> do
-          let process currentInflight = do
-                addAttribute traceSpan attrShibuyaInflightCount currentInflight
-                addAttribute traceSpan attrShibuyaInflightMax maxConc
+              -- Record handler start event
+              addEvent traceSpan handlerStartedEvent
 
-                -- Record handler start event
-                addEvent traceSpan handlerStartedEvent
+              -- Call handler and finalizer separately. A handler exception is
+              -- substituted with immediate retry so an admitted delivery always receives a
+              -- finalization decision. Superseded deliveries are excluded.
+              --
+              -- This is a *separate* per-message 'catchAny' from the one inside
+              -- 'finalizeWithRetry' below. 0.7.1.0 used a single combined catch around
+              -- handler+finalize, which skipped finalization when the handler threw;
+              -- splitting them is what makes "always finalize admitted deliveries" hold. The extra
+              -- per-message exception frame costs ~126 bytes/message (measured, EP-31
+              -- S3: docs/plans/31-…-shared-per-message-…-allocation-regression.md) and is
+              -- accepted as the price of the always-finalize guarantee — do not merge the
+              -- two catches back together to reclaim it.
+              handlerResult <-
+                catchAny
+                  (Right <$> handler (toMessage ingested))
+                  ( \ex -> do
+                      recordException traceSpan ex
+                      pure (Left ex)
+                  )
+              let (decision, result) = case handlerResult of
+                    Right d -> (d, Right d)
+                    Left ex ->
+                      ( AckRetry (RetryDelay 0),
+                        Left $ HandlerException $ Text.pack $ show ex
+                      )
 
-                -- Call handler and finalizer separately. A handler exception is
-                -- substituted with immediate retry so an admitted delivery always receives a
-                -- finalization decision. Superseded deliveries are excluded.
-                --
-                -- This is a *separate* per-message 'catchAny' from the one inside
-                -- 'finalizeWithRetry' below. 0.7.1.0 used a single combined catch around
-                -- handler+finalize, which skipped finalization when the handler threw;
-                -- splitting them is what makes "always finalize admitted deliveries" hold. The extra
-                -- per-message exception frame costs ~126 bytes/message (measured, EP-31
-                -- S3: docs/plans/31-…-shared-per-message-…-allocation-regression.md) and is
-                -- accepted as the price of the always-finalize guarantee — do not merge the
-                -- two catches back together to reclaim it.
-                handlerResult <-
-                  catchAny
-                    (Right <$> handler (toMessage ingested))
-                    ( \ex -> do
-                        recordException traceSpan ex
-                        pure (Left ex)
-                    )
-                let (decision, result) = case handlerResult of
-                      Right d -> (d, Right d)
-                      Left ex ->
-                        ( AckRetry (RetryDelay 0),
-                          Left $ HandlerException $ Text.pack $ show ex
-                        )
+              when (isLeft handlerResult) $
+                addEvent traceSpan $
+                  mkEvent
+                    eventAckDecision
+                    [(attrShibuyaAckDecision, OTel.toAttribute ("ack_retry" :: Text))]
 
-                when (isLeft handlerResult) $
+              finalizeResult <- finalizeWithRetry traceSpan ingested decision
+
+              -- Record completion event and set status
+              case finalizeResult of
+                Left _ -> do
                   addEvent traceSpan $
                     mkEvent
                       eventAckDecision
-                      [(attrShibuyaAckDecision, OTel.toAttribute ("ack_retry" :: Text))]
-
-                finalizeResult <- finalizeWithRetry traceSpan ingested decision
+                      [(attrShibuyaAckDecision, OTel.toAttribute ("finalization_failed" :: Text))]
+                  setStatus traceSpan $ OTel.Error $ finalizationFailureText msgIdText
+                Right () -> case result of
+                  Right decision' -> do
+                    let decisionText = showAckDecision decision'
+                    addEvent traceSpan $
+                      mkEvent
+                        eventHandlerCompleted
+                        [(attrShibuyaAckDecision, OTel.toAttribute decisionText)]
+                    addAttribute traceSpan attrShibuyaAckDecision decisionText
 
-                -- Record completion event and set status
-                case finalizeResult of
-                  Left _ -> do
+                    -- Set span status based on decision
+                    case decision' of
+                      AckOk -> setStatus traceSpan OTel.Ok
+                      AckRetry _ -> setStatus traceSpan OTel.Ok
+                      AckDeadLetter reason -> do
+                        addAttribute
+                          traceSpan
+                          attrShibuyaDeadLetterReasonCode
+                          (deadLetterCodeText (deadLetterReasonCode reason))
+                        setStatus traceSpan $ OTel.Error $ renderDeadLetterReason reason
+                      AckHalt reason ->
+                        setStatus traceSpan $ OTel.Error $ showHaltReason reason
+                  Left err -> do
                     addEvent traceSpan $
                       mkEvent
                         eventAckDecision
-                        [(attrShibuyaAckDecision, OTel.toAttribute ("finalization_failed" :: Text))]
-                    setStatus traceSpan $ OTel.Error $ finalizationFailureText msgIdText
-                  Right () -> case result of
-                    Right decision' -> do
-                      let decisionText = showAckDecision decision'
-                      addEvent traceSpan $
-                        mkEvent
-                          eventHandlerCompleted
-                          [(attrShibuyaAckDecision, OTel.toAttribute decisionText)]
-                      addAttribute traceSpan attrShibuyaAckDecision decisionText
-
-                      -- Set span status based on decision
-                      case decision' of
-                        AckOk -> setStatus traceSpan OTel.Ok
-                        AckRetry _ -> setStatus traceSpan OTel.Ok
-                        AckDeadLetter reason -> do
-                          addAttribute
-                            traceSpan
-                            attrShibuyaDeadLetterReasonCode
-                            (deadLetterCodeText (deadLetterReasonCode reason))
-                          setStatus traceSpan $ OTel.Error $ renderDeadLetterReason reason
-                        AckHalt reason ->
-                          setStatus traceSpan $ OTel.Error $ showHaltReason reason
-                    Left err -> do
-                      addEvent traceSpan $
-                        mkEvent
-                          eventAckDecision
-                          [(attrShibuyaAckDecision, OTel.toAttribute ("error" :: Text))]
-                      setStatus traceSpan $ OTel.Error $ handlerErrorToText err
-
-                let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor
-                when detailed $ case (finalizeResult, result) of
-                  (Right (), Right AckOk) -> recordCursor
-                  (Right (), Right (AckDeadLetter _)) -> recordCursor
-                  _ -> pure ()
+                        [(attrShibuyaAckDecision, OTel.toAttribute ("error" :: Text))]
+                    setStatus traceSpan $ OTel.Error $ handlerErrorToText err
 
-                -- Decrement in-flight and update stats.
-                liftIO $
-                  case finalizeResult of
-                    Left _ -> finishFinalizationFailure metricsHandle (finalizationFailureText msgIdText)
-                    Right () -> finishProcessing metricsHandle (metricForResult result)
+              let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor
+              when detailed $ case (finalizeResult, result) of
+                (Right (), Right AckOk) -> recordCursor
+                (Right (), Right (AckDeadLetter _)) -> recordCursor
+                _ -> pure ()
 
-                -- Handle halt (set flag, don't throw - let stream drain)
+              -- Decrement in-flight and update stats.
+              liftIO $
                 case finalizeResult of
-                  Left _ ->
-                    liftIO $
-                      requestProcessorExit exitPublisher (ProcessorFailed (finalizationFailureText msgIdText) (Just messageId))
-                  Right () -> case result of
-                    Right (AckHalt reason) ->
-                      liftIO $ requestProcessorExit exitPublisher (ProcessorHalted reason)
-                    _ -> pure ()
-          if not detailed
-            then do
-              currentInflight <- liftIO $ beginProcessing metricsHandle maxConc
-              process currentInflight
-            else Exception.mask $ \restore -> do
-              currentInflight <- liftIO $ beginProcessing metricsHandle maxConc
-              slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle messageId
-              restore (process currentInflight)
-                `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)
-              liftIO (Metrics.releaseInFlightSlot metricsHandle slot)
-                `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)
+                  Left _ -> finishFinalizationFailure metricsHandle (finalizationFailureText msgIdText)
+                  Right () -> finishProcessing metricsHandle (metricForResult result)
+
+              -- Handle halt (set flag, don't throw - let stream drain)
+              case finalizeResult of
+                Left _ ->
+                  liftIO $
+                    requestProcessorExit exitPublisher (ProcessorFailed (finalizationFailureText msgIdText) (Just messageId))
+                Right () -> case result of
+                  Right (AckHalt reason) ->
+                    liftIO $ requestProcessorExit exitPublisher (ProcessorHalted reason)
+                  _ -> pure ()
+        if not detailed
+          then do
+            currentInflight <- liftIO $ beginProcessing metricsHandle maxConc
+            process currentInflight
+          else Exception.mask $ \restore -> do
+            currentInflight <- liftIO $ beginProcessing metricsHandle maxConc
+            slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle messageId
+            restore (process currentInflight)
+              `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)
+            liftIO (Metrics.releaseInFlightSlot metricsHandle slot)
+              `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)
   where
     isLeft :: Either a b -> Bool
     isLeft (Left _) = True
diff --git a/src/Shibuya/Telemetry.hs b/src/Shibuya/Telemetry.hs
--- a/src/Shibuya/Telemetry.hs
+++ b/src/Shibuya/Telemetry.hs
@@ -31,9 +31,17 @@
 
     -- * Semantic Conventions
     module Shibuya.Telemetry.Semantic,
+
+    -- * Trace Correlation
+    module Shibuya.Telemetry.Correlation,
+
+    -- * Processing Span
+    module Shibuya.Telemetry.Process,
   )
 where
 
+import Shibuya.Telemetry.Correlation
 import Shibuya.Telemetry.Effect
+import Shibuya.Telemetry.Process
 import Shibuya.Telemetry.Propagation
 import Shibuya.Telemetry.Semantic
diff --git a/src/Shibuya/Telemetry/Correlation.hs b/src/Shibuya/Telemetry/Correlation.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Telemetry/Correlation.hs
@@ -0,0 +1,171 @@
+-- | How a processing span relates to the span that created its message.
+--
+-- A message's /creation context/ is the span identity its producer encoded
+-- in the W3C trace headers the adapter extracted into
+-- 'Shibuya.Core.Types.Envelope'@.traceContext@. Every processing span links to
+-- the creation context when one decodes, as the OpenTelemetry messaging
+-- conventions recommend. Whether the span is also a /child/ of that context
+-- is the adapter's declared 'TraceCorrelation' policy.
+module Shibuya.Telemetry.Correlation
+  ( -- * Policy
+    TraceCorrelation (..),
+    AdapterTelemetry (traceCorrelation),
+    defaultAdapterTelemetry,
+
+    -- * Links
+    creationLink,
+    creationLinkWith,
+    consumerSpanArgsLinkedTo,
+
+    -- * Batch links
+    maxBatchLinks,
+    BatchLinks (..),
+    collectBatchLinks,
+    collectBatchLinksBy,
+  )
+where
+
+import Control.DeepSeq (NFData)
+import Data.HashMap.Strict qualified as HashMap
+import GHC.Generics (Generic)
+import OpenTelemetry.Attributes (AttributeMap, toAttribute)
+import OpenTelemetry.Trace.Core (NewLink (..), SpanArguments (..), SpanContext, SpanKind (..))
+import Shibuya.Core.Types (MessageId (..), TraceHeaders)
+import Shibuya.Telemetry.Propagation (extractTraceContext)
+import Shibuya.Telemetry.Semantic (attrMessagingMessageId)
+
+-- | How a processing span relates to the span that created the message.
+--
+-- Both policies link the processing span to the creation context when the
+-- message's trace headers decode. Without usable headers neither policy adds
+-- a link, and the span is opened in the ambient context exactly as before.
+data TraceCorrelation
+  = -- | Child of the creation context, and linked to it: one connected trace
+    -- from producer to consumer. Suits queues consumed soon after production.
+    ParentAndLink
+  | -- | A new root trace, related to the creation context only by a link.
+    -- Suits replayed event streams, long-delayed or retried jobs, fan-out to
+    -- many consumers, and tail-sampling deployments: the consumer trace is
+    -- sampled independently of the producer's.
+    LinkOnly
+  deriving stock (Eq, Show, Generic, Enum, Bounded)
+  deriving anyclass (NFData)
+
+-- | Telemetry options an adapter declares in its 'Shibuya.Adapter.Adapter'.
+--
+-- Build a value from 'defaultAdapterTelemetry' with record update:
+--
+-- @
+-- defaultAdapterTelemetry {traceCorrelation = LinkOnly}
+-- @
+--
+-- The constructor is not exported, so options can be added later as fields
+-- with defaults without breaking any adapter. The framework reads it once per
+-- processor, never per message.
+data AdapterTelemetry = AdapterTelemetry
+  { -- | Parent-versus-link policy for this adapter's processing spans.
+    traceCorrelation :: !TraceCorrelation
+  }
+  deriving stock (Eq, Show, Generic)
+  deriving anyclass (NFData)
+
+-- | @traceCorrelation = 'ParentAndLink'@.
+defaultAdapterTelemetry :: AdapterTelemetry
+defaultAdapterTelemetry = AdapterTelemetry {traceCorrelation = ParentAndLink}
+
+-- | Decode trace headers into a link to the message's creation context.
+--
+-- 'Nothing' when the headers are absent or malformed. The link carries no
+-- attributes (the shared empty map), holds an evaluated
+-- 'OpenTelemetry.Trace.Core.SpanContext', and keeps no reference to the
+-- header list.
+creationLink :: TraceHeaders -> Maybe NewLink
+creationLink = creationLinkWith HashMap.empty
+{-# INLINE creationLink #-}
+
+-- | As 'creationLink', with attributes describing the linked message (for
+-- example its @messaging.message.id@ on a batch span). The caller supplies an
+-- already evaluated map.
+creationLinkWith :: AttributeMap -> TraceHeaders -> Maybe NewLink
+creationLinkWith attrs headers = case extractTraceContext headers of
+  Nothing -> Nothing
+  Just ctx ->
+    let !link = NewLink {linkContext = ctx, linkAttributes = attrs}
+     in Just link
+{-# INLINE creationLinkWith #-}
+
+-- | Consumer-kind span arguments carrying the given links. The list must be
+-- fully evaluated by the caller, so the span references nothing else.
+--
+-- 'Shibuya.Telemetry.Semantic.consumerSpanArgs' remains the constant for
+-- spans without links.
+consumerSpanArgsLinkedTo :: [NewLink] -> SpanArguments
+consumerSpanArgsLinkedTo links =
+  SpanArguments
+    { kind = Consumer,
+      attributes = HashMap.empty,
+      links = links,
+      startTime = Nothing
+    }
+{-# INLINE consumerSpanArgsLinkedTo #-}
+
+-- | Maximum links Shibuya places on one batch span: the OpenTelemetry SDK's
+-- default per-span link limit. A lower configured SDK limit
+-- (@OTEL_SPAN_LINK_COUNT_LIMIT@) can still drop supplied links on its own.
+maxBatchLinks :: Int
+maxBatchLinks = 128
+
+-- | The creation-context links collected for one batch.
+data BatchLinks = BatchLinks
+  { -- | One link per member with a decodable creation context, in member
+    -- order, each carrying that member's @messaging.message.id@. Fully
+    -- evaluated; at most 'maxBatchLinks' long.
+    links :: ![NewLink],
+    -- | Members with a decodable creation context beyond 'maxBatchLinks'.
+    -- Absent and malformed contexts are never counted.
+    dropped :: !Int
+  }
+
+-- | 'collectBatchLinksBy' over @(message id, trace headers)@ pairs.
+collectBatchLinks :: [(MessageId, Maybe TraceHeaders)] -> BatchLinks
+collectBatchLinks = collectBatchLinksBy fst snd
+{-# INLINE collectBatchLinks #-}
+
+-- | Collect one link per member that has a decodable creation context, in
+-- order, each carrying @messaging.message.id@. At most 'maxBatchLinks' links
+-- are stored. Decoding continues past the bound so that 'dropped' counts the
+-- valid omitted contexts exactly, but no link or attribute map is built for
+-- them.
+--
+-- One strict pass with strict accumulators. A member whose raw headers equal
+-- the previous member's reuses that member's decoded context. Members that
+-- share a context still get separate links, because the links carry
+-- different message identifiers. The result references only evaluated span
+-- contexts and identifier text, never the members.
+collectBatchLinksBy :: (Foldable t) => (a -> MessageId) -> (a -> Maybe TraceHeaders) -> t a -> BatchLinks
+collectBatchLinksBy messageIdOf headersOf members =
+  case foldl' step (Acc 0 [] 0 Nothing Nothing) members of
+    Acc _ collected omitted _ _ -> BatchLinks {links = reverse collected, dropped = omitted}
+  where
+    step acc@(Acc count collected omitted previousHeaders previousContext) member =
+      case headersOf member of
+        Nothing -> acc
+        Just headers ->
+          let !context =
+                if Just headers == previousHeaders
+                  then previousContext
+                  else extractTraceContext headers
+           in case context of
+                Nothing -> Acc count collected omitted (Just headers) context
+                Just spanContext
+                  | count < maxBatchLinks ->
+                      let MessageId idText = messageIdOf member
+                          !attrs = HashMap.singleton attrMessagingMessageId $! toAttribute idText
+                          !link = NewLink {linkContext = spanContext, linkAttributes = attrs}
+                       in Acc (count + 1) (link : collected) omitted (Just headers) context
+                  | otherwise -> Acc count collected (omitted + 1) (Just headers) context
+{-# INLINE collectBatchLinksBy #-}
+
+-- Strict fold state: links so far, reversed, with their count; the overflow
+-- count; and the previous member's raw headers and decoded context.
+data Acc = Acc !Int ![NewLink] !Int !(Maybe TraceHeaders) !(Maybe SpanContext)
diff --git a/src/Shibuya/Telemetry/Effect.hs b/src/Shibuya/Telemetry/Effect.hs
--- a/src/Shibuya/Telemetry/Effect.hs
+++ b/src/Shibuya/Telemetry/Effect.hs
@@ -25,6 +25,7 @@
     getTracer,
     isTracingEnabled,
     withExtractedContext,
+    withRootContext,
 
     -- * Re-exports
     OTel.Span,
@@ -294,6 +295,24 @@
           (Ctx.attachContext newContext)
           Ctx.detachContext
           (\_token -> runInIO action)
+
+-- | Run an action with an empty OpenTelemetry context attached, so a span
+-- opened inside it is a root in a new trace. The previous context is restored
+-- afterwards, on normal exit, exception and asynchronous cancellation.
+-- A no-op when tracing is disabled.
+withRootContext ::
+  (Tracing :> es, IOE :> es) =>
+  Eff es a ->
+  Eff es a
+withRootContext action = do
+  TracingRep {tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then action
+    else withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
+      bracket
+        (Ctx.attachContext Ctx.empty)
+        Ctx.detachContext
+        (\_token -> runInIO action)
 
 --------------------------------------------------------------------------------
 -- Internal Helpers
diff --git a/src/Shibuya/Telemetry/Process.hs b/src/Shibuya/Telemetry/Process.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Telemetry/Process.hs
@@ -0,0 +1,120 @@
+-- | The Consumer-kind processing span for one message.
+--
+-- Shibuya's own runner opens every per-message processing span through
+-- 'withProcessSpan'. It is exported so code that drains an adapter outside
+-- the runner opens the same span, with the same link and parent rule, instead
+-- of a hand-written copy that can drift.
+--
+-- The rule, decided once per call:
+--
+-- 1. Tracing disabled: the body receives the shared non-recording sentinel
+--    span. The trace headers are not inspected.
+-- 2. Tracing enabled, no usable creation context (headers absent or
+--    malformed): the span is opened in the ambient context, with no link,
+--    exactly as Shibuya did before trace correlation policies existed.
+-- 3. Tracing enabled, the headers decode: the span links to the creation
+--    context. Under 'ParentAndLink' it is also a child of it; under
+--    'LinkOnly' it is a root in a new trace. The previous thread context is
+--    restored afterwards, also on exceptions and cancellation.
+--
+-- In case 3 neither policy preserves an enclosing ambient operation: it is
+-- neither the parent nor linked. These helpers are for /independent/ message
+-- processing. A caller that processes a message inside a meaningful
+-- enclosing operation, and wants that relationship recorded, must open its
+-- own span rather than use these helpers.
+--
+-- The functions set no attributes and record no outcome; both remain the
+-- caller's.
+module Shibuya.Telemetry.Process
+  ( withProcessSpan,
+    withProcessSpanIO,
+  )
+where
+
+import Control.Exception (bracket)
+import Data.Text (Text)
+import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), withEffToIO, (:>))
+import OpenTelemetry.Context qualified as Ctx
+import OpenTelemetry.Context.ThreadLocal qualified as Ctx
+import OpenTelemetry.Trace.Core (NewLink (..), Span, SpanArguments, Tracer)
+import OpenTelemetry.Trace.Core qualified as OTel
+import Shibuya.Core.Types (TraceHeaders)
+import Shibuya.Telemetry.Correlation (TraceCorrelation (..), consumerSpanArgsLinkedTo, creationLink)
+import Shibuya.Telemetry.Effect (Tracing, getTracer, isTracingEnabled, withSpan')
+import Shibuya.Telemetry.Semantic (consumerSpanArgs)
+
+-- | Open the Consumer-kind processing span for one message.
+--
+-- Linked to the creation context in the headers when one decodes; a child of
+-- it under 'ParentAndLink'; a root under 'LinkOnly'. Without usable headers
+-- the span is opened in the ambient context, as before. With tracing
+-- disabled the body receives the sentinel span and no header is decoded.
+-- With valid headers a meaningful ambient operation is neither parented nor
+-- linked: callers opt into independent message processing by using this
+-- helper (see the module documentation).
+withProcessSpan ::
+  (Tracing :> es, IOE :> es) =>
+  TraceCorrelation ->
+  -- | Full span name, e.g. @processSpanName destination@.
+  Text ->
+  -- | The message's trace headers ('Shibuya.Core.Types.Envelope'@.traceContext@).
+  Maybe TraceHeaders ->
+  (Span -> Eff es a) ->
+  Eff es a
+withProcessSpan policy name headers body = do
+  enabled <- isTracingEnabled
+  if not enabled
+    then withSpan' name consumerSpanArgs body
+    else case processSpanPlan policy headers of
+      Nothing -> withSpan' name consumerSpanArgs body
+      Just (context, args) -> do
+        tracer <- getTracer
+        -- One unlift covers both the context attachment and the span.
+        withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
+          inContext context $ OTel.inSpan' tracer name args (runInIO . body)
+{-# INLINE withProcessSpan #-}
+
+-- | The same span for callers that hold a 'Tracer' outside the 'Tracing'
+-- effect. It always behaves as tracing enabled; pass a tracer from a no-op
+-- provider to disable it.
+withProcessSpanIO ::
+  Tracer ->
+  TraceCorrelation ->
+  -- | Full span name, e.g. @processSpanName destination@.
+  Text ->
+  Maybe TraceHeaders ->
+  (Span -> IO a) ->
+  IO a
+withProcessSpanIO tracer policy name headers body =
+  case processSpanPlan policy headers of
+    Nothing -> OTel.inSpan' tracer name consumerSpanArgs body
+    Just (context, args) -> inContext context $ OTel.inSpan' tracer name args body
+{-# INLINE withProcessSpanIO #-}
+
+-- | The decision shared by both entry points. 'Nothing': open the span in
+-- the ambient context without links. @Just (context, args)@: attach
+-- @context@ (the creation context for 'ParentAndLink', the empty context for
+-- 'LinkOnly') and open the span with @args@, which carry the one link.
+--
+-- The headers are decoded once; the link, its one-element list and the
+-- arguments are evaluated before the span exists, so the span references
+-- neither the headers nor anything else of the message.
+processSpanPlan :: TraceCorrelation -> Maybe TraceHeaders -> Maybe (Ctx.Context, SpanArguments)
+processSpanPlan policy headers = case headers of
+  Nothing -> Nothing
+  Just hs -> case creationLink hs of
+    Nothing -> Nothing
+    Just link ->
+      let !args = consumerSpanArgsLinkedTo [link]
+          !context = case policy of
+            ParentAndLink -> Ctx.insertSpan (OTel.wrapSpanContext link.linkContext) Ctx.empty
+            LinkOnly -> Ctx.empty
+       in Just (context, args)
+{-# INLINE processSpanPlan #-}
+
+-- | Attach a context for the duration of an action, restoring the previous
+-- one on normal exit, exception and asynchronous cancellation.
+inContext :: Ctx.Context -> IO a -> IO a
+inContext context action =
+  bracket (Ctx.attachContext context) Ctx.detachContext (const action)
+{-# INLINE inContext #-}
diff --git a/src/Shibuya/Telemetry/Semantic.hs b/src/Shibuya/Telemetry/Semantic.hs
--- a/src/Shibuya/Telemetry/Semantic.hs
+++ b/src/Shibuya/Telemetry/Semantic.hs
@@ -15,6 +15,8 @@
     attrMessagingMessageId,
     attrMessagingDestinationName,
     attrMessagingOperation,
+    attrMessagingOperationName,
+    attrMessagingBatchMessageCount,
 
     -- * Attribute Keys (Shibuya-specific)
     attrShibuyaProcessorId,
@@ -26,6 +28,7 @@
     attrShibuyaBatchKey,
     attrShibuyaBatchSize,
     attrShibuyaBatchTrigger,
+    attrShibuyaBatchLinksDropped,
 
     -- * Event Names
     eventHandlerStarted,
@@ -57,19 +60,21 @@
 -- Span Names
 --------------------------------------------------------------------------------
 
--- | Span name for processing a single message.
+-- | Span name for processing a message or a batch: @\"process <destination>\"@,
+-- for example @\"process orders\"@.
 --
--- Follows the OpenTelemetry messaging-spans recommendation that consumer
--- spans be named @\"<destination> <operation>\"@ — for Shibuya the
--- destination is the processor id and the operation is @\"process\"@,
--- yielding e.g. @\"shibuya-consumer process\"@.
+-- The OpenTelemetry messaging conventions say the span name "SHOULD be
+-- @{messaging.operation.name} {destination}@". Shibuya's operation name is
+-- @process@ and the destination is the processor id. The destination stays
+-- the processor id even when an adapter overrides
+-- @messaging.destination.name@ per message, so the name remains a
+-- per-processor constant.
 --
--- This intentionally departs from
--- @hs-opentelemetry-instrumentation-hw-kafka-client@'s
--- @\"process <topic>\"@ order in favor of the spec's order; see
--- @docs/plans/2-align-opentelemetry-semantic-conventions.md@.
+-- Before 0.13.0.0 the name was @\"<destination> process\"@, following an
+-- earlier reading of the conventions
+-- (@docs/plans/2-align-opentelemetry-semantic-conventions.md@).
 processSpanName :: Text -> Text
-processSpanName destination = destination <> " process"
+processSpanName destination = "process " <> destination
 
 -- | Span name for the ingester lifecycle.
 ingestSpanName :: Text
@@ -104,6 +109,17 @@
 attrMessagingOperation :: Text
 attrMessagingOperation = unkey Sem.messaging_operation_type
 
+-- | @messaging.operation.name@: the system-specific operation name, Required
+-- by the messaging conventions. Shibuya's processing spans set @\"process\"@.
+-- ('attrMessagingOperation' is the operation /type/ key.)
+attrMessagingOperationName :: Text
+attrMessagingOperationName = unkey Sem.messaging_operation_name
+
+-- | @messaging.batch.message_count@: the number of messages in a batch, set
+-- on batch processing spans.
+attrMessagingBatchMessageCount :: Text
+attrMessagingBatchMessageCount = unkey Sem.messaging_batch_messageCount
+
 --------------------------------------------------------------------------------
 -- Attribute Keys (Shibuya-specific)
 --
@@ -164,6 +180,11 @@
 -- | Why the batch was emitted: size, timeout, or flush (@shibuya.batch.trigger@).
 attrShibuyaBatchTrigger :: Text
 attrShibuyaBatchTrigger = "shibuya.batch.trigger"
+
+-- | Members of a batch with a valid creation context that were not linked
+-- because the batch span reached its link bound. Set only when non-zero.
+attrShibuyaBatchLinksDropped :: Text
+attrShibuyaBatchLinksDropped = "shibuya.batch.links.dropped"
 
 --------------------------------------------------------------------------------
 -- Event Names
diff --git a/test-gc/Finished.hs b/test-gc/Finished.hs
--- a/test-gc/Finished.hs
+++ b/test-gc/Finished.hs
@@ -12,7 +12,7 @@
 import Control.Monad (forM, replicateM_, unless, void)
 import Data.List (isInfixOf)
 import Effectful (IOE, liftIO, runEff)
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.App
   ( AppConfig (..),
     ProcessorId (..),
@@ -113,7 +113,8 @@
   Adapter
     { adapterName = "gc-regression:finite",
       source = Stream.fromList (map message [1 .. count]),
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
 
 failedSource :: Adapter '[Tracing, IOE] String
@@ -121,7 +122,8 @@
   Adapter
     { adapterName = "gc-regression:failed",
       source = Stream.fromEffect (liftIO (throwIO (userError "the source failed on purpose"))),
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
 
 -- | One message, then a quiet queue: the source stays alive but produces nothing.
@@ -130,7 +132,8 @@
   Adapter
     { adapterName = "gc-regression:one-then-idle",
       source = Stream.unfoldrM step (0 :: Int),
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
   where
     step 0 = pure (Just (message 0, 1))
diff --git a/test-gc/Main.hs b/test-gc/Main.hs
--- a/test-gc/Main.hs
+++ b/test-gc/Main.hs
@@ -9,7 +9,7 @@
 import Control.Exception (SomeException, displayException, try)
 import Control.Monad (replicateM_)
 import Effectful (liftIO, runEff)
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.App
   ( AppConfig (..),
     ProcessorId (..),
@@ -56,7 +56,8 @@
             source = Stream.fromEffect $ liftIO $ do
               threadDelay 60_000_000
               fail "the idle adapter unexpectedly woke during the GC probe",
-            shutdown = pure ()
+            shutdown = pure (),
+            telemetry = defaultAdapterTelemetry
           }
   result <-
     runApp
diff --git a/test-retention/Main.hs b/test-retention/Main.hs
new file mode 100644
--- /dev/null
+++ b/test-retention/Main.hs
@@ -0,0 +1,305 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+-- | Trace retention probes.
+--
+-- This program owns its process (see ADR 0001): weak-pointer and live-bytes
+-- assertions depend on major collections and on nothing else in the process
+-- holding references, so they cannot share a process with the hspec suite.
+--
+-- Check A: a finished span, held by an exporter, does not keep the message
+-- that produced it alive.
+--
+-- Check B: live heap after processing ten times as many traced messages is
+-- not materially larger.
+--
+-- Each check prints one PASS/FAIL line; the program exits non-zero if any
+-- check fails.
+module Main (main) where
+
+import Control.Exception (evaluate)
+import Control.Monad (forM, unless, void)
+import Data.ByteString (ByteString)
+import Data.ByteString qualified as BS
+import Data.ByteString.Char8 qualified as BS8
+import Data.IORef (IORef, atomicModifyIORef', mkWeakIORef, newIORef, readIORef)
+import Data.List.NonEmpty (NonEmpty)
+import Data.Maybe (isJust)
+import Data.Text qualified as Text
+import Data.Word (Word8)
+import Effectful (Eff, IOE, liftIO, runEff, (:>))
+import Effectful.Exception qualified as Exception
+import GHC.Stats (GCDetails (..), RTSStats (..), getRTSStats)
+import Numeric (showHex)
+import OpenTelemetry.Attributes (emptyAttributes)
+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)
+import OpenTelemetry.Processor.Span (FlushResult (..), ShutdownResult (..), SpanProcessor (..))
+import OpenTelemetry.Trace.Core qualified as OTel
+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry)
+import Shibuya.App
+  ( QueueProcessor (..),
+    ackAllOk,
+    defaultAppConfig,
+    defaultBatchConfig,
+    runApp,
+    stopApp,
+    waitApp,
+  )
+import Shibuya.Batch (BatchAck, BatchConfig (..), BatchInfo)
+import Shibuya.Core.Ack (AckDecision (..))
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Ingested (Ingested, mkIngested)
+import Shibuya.Core.Metrics (ProcessorId (..))
+import Shibuya.Core.Types (Envelope (..), MessageId (..), TraceHeaders, mkEnvelope)
+import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))
+import Shibuya.Telemetry.Effect (Tracing, runTracing)
+import Streamly.Data.Stream qualified as Stream
+import System.Exit (exitFailure)
+import System.Mem (performMajorGC)
+import System.Mem.Weak (Weak, deRefWeak)
+import System.Timeout (timeout)
+
+-- | 'BatchDistinct' is a batch whose members each carry a different producer
+-- context, so the batch span holds one link per member up to the bound.
+data Shape = SingleSerial | SingleAsync8 | Batch | BatchDistinct
+  deriving stock (Eq, Show)
+
+-- | Batch cases are labelled by shape alone: batch spans take no policy.
+caseName :: Shape -> TraceCorrelation -> String
+caseName Batch _ = shapeName Batch
+caseName BatchDistinct _ = shapeName BatchDistinct
+caseName shape policy = shapeName shape <> "/" <> show policy
+
+shapeName :: Shape -> String
+shapeName = \case
+  SingleSerial -> "single-serial"
+  SingleAsync8 -> "single-async8"
+  Batch -> "batch"
+  BatchDistinct -> "batch/distinct-contexts"
+
+-- | A message payload. The 'token' is the identity the weak pointer watches:
+-- a weak pointer keyed on an 'IORef' follows its underlying mutable cell,
+-- which the optimiser cannot copy or unbox, so a dead weak pointer means the
+-- payload record really is unreachable. The 'blob' gives a retention bug a
+-- visible size.
+data Payload = Payload
+  { blob :: !ByteString,
+    token :: !(IORef ())
+  }
+
+main :: IO ()
+main = do
+  results <-
+    sequence
+      [ checkPayloadsCollected SingleSerial ParentAndLink,
+        checkPayloadsCollected SingleSerial LinkOnly,
+        checkPayloadsCollected Batch ParentAndLink,
+        checkPayloadsCollected BatchDistinct ParentAndLink,
+        checkLiveBytesFlat SingleSerial ParentAndLink,
+        checkLiveBytesFlat SingleSerial LinkOnly,
+        checkLiveBytesFlat SingleAsync8 ParentAndLink,
+        checkLiveBytesFlat Batch ParentAndLink,
+        checkLiveBytesFlat BatchDistinct ParentAndLink
+      ]
+  unless (and results) exitFailure
+
+--------------------------------------------------------------------------------
+-- Check A
+--------------------------------------------------------------------------------
+
+payloadCount :: Int
+payloadCount = 8
+
+checkPayloadsCollected :: Shape -> TraceCorrelation -> IO Bool
+checkPayloadsCollected shape policy = bounded label $ do
+  (processor, exported) <- inMemoryListExporter
+  provider <- OTel.createTracerProvider [processor] OTel.emptyTracerProviderOptions
+  let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "trace-retention-test" "" "" emptyAttributes) OTel.tracerOptions
+  batches <- newIORef (0 :: Int)
+  weaks <- runPayloadCase tracer shape policy batches
+  performMajorGC
+  performMajorGC
+  alive <- length . filter isJust <$> traverse deRefWeak weaks
+  -- Read the exporter only after collection, so it is demonstrably still
+  -- reachable (and still holds every finished span) while the payloads die.
+  spans <- readIORef exported
+  let consumerSpans = length [() | sp <- spans, OTel.spanKind sp == OTel.Consumer]
+  batchCount <- readIORef batches
+  let expectedConsumer = if isBatch shape then batchCount else payloadCount
+  void $ OTel.shutdownTracerProvider provider (Just 1_000_000)
+  report
+    label
+    (alive == 0 && consumerSpans == expectedConsumer && expectedConsumer > 0)
+    ( show consumerSpans
+        <> " Consumer spans (expected "
+        <> show expectedConsumer
+        <> "), "
+        <> show (length spans)
+        <> " exported in total, "
+        <> show (payloadCount - alive)
+        <> " of "
+        <> show payloadCount
+        <> " payloads collected"
+    )
+  where
+    label = caseName shape policy <> ": finished spans do not retain payloads"
+
+-- | Build the messages, process them to completion, and return only the weak
+-- pointers. Everything else this case allocated, including the application
+-- handle (whose adapter still references the message list), is unreachable
+-- once it returns.
+runPayloadCase :: OTel.Tracer -> Shape -> TraceCorrelation -> IORef Int -> IO [Weak (IORef ())]
+runPayloadCase tracer shape policy batches = do
+  built <- forM [1 .. payloadCount] $ \i -> do
+    token <- newIORef ()
+    weak <- mkWeakToken token
+    let payload = Payload {blob = BS.replicate (1024 * 1024) (fromIntegral i :: Word8), token}
+    pure (weak, message shape i payload)
+  let weaks = map fst built
+      adapter =
+        Adapter
+          { adapterName = "trace-retention:payload",
+            source = Stream.fromList (map snd built),
+            shutdown = pure (),
+            telemetry = defaultAdapterTelemetry
+          }
+  -- Force the weak pointers out of their tuples now; returning the lazy
+  -- @map fst built@ would keep the whole message list reachable.
+  mapM_ evaluate weaks
+  runEff $ runTracing tracer $ runToCompletion shape policy batches adapter
+  pure weaks
+{-# NOINLINE runPayloadCase #-}
+
+mkWeakToken :: IORef () -> IO (Weak (IORef ()))
+mkWeakToken ref = mkWeakIORef ref (pure ())
+
+--------------------------------------------------------------------------------
+-- Check B
+--------------------------------------------------------------------------------
+
+smallRun :: Int
+smallRun = 20_000
+
+checkLiveBytesFlat :: Shape -> TraceCorrelation -> IO Bool
+checkLiveBytesFlat shape policy = bounded label $ do
+  -- No exporter retains finished spans; a single discarding processor is
+  -- needed only because hs-opentelemetry makes spans non-recording when a
+  -- provider has no processors at all.
+  provider <- OTel.createTracerProvider [discardProcessor] OTel.emptyTracerProviderOptions
+  let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "trace-retention-test" "" "" emptyAttributes) OTel.tracerOptions
+  runSmallMessages tracer shape policy smallRun
+  before <- liveBytesAfterGC
+  runSmallMessages tracer shape policy (10 * smallRun)
+  after <- liveBytesAfterGC
+  void $ OTel.shutdownTracerProvider provider (Just 1_000_000)
+  let allowed = before + 1024 * 1024 + before `div` 10
+  report
+    label
+    (after <= allowed)
+    ( "N="
+        <> show smallRun
+        <> " -> "
+        <> show (10 * smallRun)
+        <> ", live bytes "
+        <> show before
+        <> " -> "
+        <> show after
+        <> " (limit "
+        <> show allowed
+        <> ")"
+    )
+  where
+    label = caseName shape policy <> ": live bytes flat"
+
+runSmallMessages :: OTel.Tracer -> Shape -> TraceCorrelation -> Int -> IO ()
+runSmallMessages tracer shape policy n = do
+  batches <- newIORef (0 :: Int)
+  let adapter =
+        Adapter
+          { adapterName = "trace-retention:flat",
+            -- A generator, not a list, so the source itself retains nothing.
+            source = fmap (\i -> message shape i (Text.pack (show i))) (Stream.enumerateFromTo 1 n),
+            shutdown = pure (),
+            telemetry = defaultAdapterTelemetry
+          }
+  runEff $ runTracing tracer $ runToCompletion shape policy batches adapter
+{-# NOINLINE runSmallMessages #-}
+
+liveBytesAfterGC :: IO Integer
+liveBytesAfterGC = do
+  performMajorGC
+  performMajorGC
+  stats <- getRTSStats
+  pure $ fromIntegral (gcdetails_live_bytes (gc stats))
+
+discardProcessor :: SpanProcessor
+discardProcessor =
+  SpanProcessor
+    { spanProcessorOnStart = \_ _ -> pure (),
+      spanProcessorOnEnd = \_ -> pure (),
+      spanProcessorShutdown = pure ShutdownSuccess,
+      spanProcessorForceFlush = pure FlushSuccess
+    }
+
+--------------------------------------------------------------------------------
+-- Shared
+--------------------------------------------------------------------------------
+
+-- | Run one processor over the adapter until its source is exhausted, then
+-- stop the application so no runner thread outlives the call.
+runToCompletion ::
+  (IOE :> es, Tracing :> es) =>
+  Shape ->
+  TraceCorrelation ->
+  IORef Int ->
+  Adapter es msg ->
+  Eff es ()
+runToCompletion shape policy batches adapter0 = do
+  -- Batch spans take no policy; single-message spans follow the adapter's.
+  let adapter = adapter0 {telemetry = defaultAdapterTelemetry {traceCorrelation = policy}}
+      processor = case shape of
+        SingleSerial -> QueueProcessor adapter (\_ -> pure AckOk) Unordered Serial
+        SingleAsync8 -> QueueProcessor adapter (\_ -> pure AckOk) Unordered (Async 8)
+        Batch -> batching
+        BatchDistinct -> batching
+      batching = BatchingProcessor adapter (countBatch batches) defaultBatchConfig {batchSize = 100} Unordered Serial
+  runApp defaultAppConfig [(ProcessorId "trace-retention", processor)] >>= \case
+    Left err -> liftIO $ fail ("runApp failed: " <> show err)
+    Right app -> Exception.finally (waitApp app) (stopApp app)
+
+countBatch :: (IOE :> es) => IORef Int -> BatchInfo -> NonEmpty a -> Eff es BatchAck
+countBatch batches _ _ = do
+  liftIO $ atomicModifyIORef' batches (\n -> (n + 1, ()))
+  pure ackAllOk
+
+isBatch :: Shape -> Bool
+isBatch = \case
+  Batch -> True
+  BatchDistinct -> True
+  _ -> False
+
+message :: (IOE :> es) => Shape -> Int -> msg -> Ingested es msg
+message shape i payload =
+  let headers = if shape == BatchDistinct then distinctTraceHeaders i else traceHeaders
+      envelope = (mkEnvelope (MessageId (Text.pack (show i))) payload) {traceContext = Just headers}
+   in mkIngested envelope (AckHandle (\_ -> pure ()))
+
+-- | A valid W3C context unique to message @i@.
+distinctTraceHeaders :: Int -> TraceHeaders
+distinctTraceHeaders i =
+  [("traceparent", BS8.pack ("00-" <> hex 32 (i + 0x1000) <> "-" <> hex 16 (i + 0x1000) <> "-01"))]
+  where
+    hex width n = let digits = showHex n "" in replicate (width - length digits) '0' <> digits
+
+traceHeaders :: TraceHeaders
+traceHeaders = [("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")]
+
+bounded :: String -> IO Bool -> IO Bool
+bounded label check =
+  timeout 60_000_000 check >>= \case
+    Just ok -> pure ok
+    Nothing -> report label False "timed out after 60 seconds"
+
+report :: String -> Bool -> String -> IO Bool
+report label ok detail = do
+  putStrLn ((if ok then "PASS: " else "FAIL: ") <> label <> " (" <> detail <> ")")
+  pure ok
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -18,6 +18,8 @@
 import Shibuya.Runner.PauseSpec qualified as Pause
 import Shibuya.Runner.SupervisedSpec qualified
 import Shibuya.RunnerSpec qualified
+import Shibuya.Telemetry.BatchCorrelationSpec qualified
+import Shibuya.Telemetry.CorrelationSpec qualified
 import Shibuya.Telemetry.EffectSpec qualified
 import Shibuya.Telemetry.LifecycleSpec qualified
 import Shibuya.Telemetry.PropagationSpec qualified
@@ -44,5 +46,7 @@
   Shibuya.Runner.SupervisedSpec.spec
   Shibuya.Telemetry.LifecycleSpec.spec
   Shibuya.Telemetry.EffectSpec.spec
+  Shibuya.Telemetry.CorrelationSpec.spec
+  Shibuya.Telemetry.BatchCorrelationSpec.spec
   Shibuya.Telemetry.PropagationSpec.spec
   Shibuya.Telemetry.SemanticSpec.spec
diff --git a/test/Shibuya/App/BatchSpec.hs b/test/Shibuya/App/BatchSpec.hs
--- a/test/Shibuya/App/BatchSpec.hs
+++ b/test/Shibuya/App/BatchSpec.hs
@@ -8,7 +8,7 @@
 import Data.Text qualified as Text
 import Data.Time (UTCTime (..), fromGregorian)
 import Effectful (Eff, IOE, liftIO, runEff, (:>))
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.Adapter.Mock (TrackingAck (..), newTrackingAck, trackingAckHandle)
 import Shibuya.App
   ( AppConfig (..),
@@ -179,7 +179,7 @@
 -- Finite adapter: ends as soon as its list is exhausted.
 listAdapter' :: [Ingested es String] -> Adapter es String
 listAdapter' messages =
-  Adapter {adapterName = "test:batch", source = Stream.fromList messages, shutdown = pure ()}
+  Adapter {adapterName = "test:batch", source = Stream.fromList messages, shutdown = pure (), telemetry = defaultAdapterTelemetry}
 
 -- Gated adapter: emits the messages, then blocks until 'shutdown' opens the gate,
 -- so a partial batch stays pending until stopApp triggers the flush. Implemented
@@ -190,7 +190,8 @@
   Adapter
     { adapterName = "test:gated",
       source = Stream.unfoldrM step messages,
-      shutdown = liftIO $ atomically $ writeTVar gate True
+      shutdown = liftIO $ atomically $ writeTVar gate True,
+      telemetry = defaultAdapterTelemetry
     }
   where
     step [] = do
diff --git a/test/Shibuya/App/LifecycleSpec.hs b/test/Shibuya/App/LifecycleSpec.hs
--- a/test/Shibuya/App/LifecycleSpec.hs
+++ b/test/Shibuya/App/LifecycleSpec.hs
@@ -15,7 +15,7 @@
 import Effectful.Exception qualified as Exception
 import OpenTelemetry.Attributes qualified as OTelAttributes
 import OpenTelemetry.Trace.Core qualified as OTel
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.App
   ( AppConfig (..),
     QueueProcessor (..),
@@ -581,7 +581,8 @@
   Adapter
     { adapterName = "test:list",
       source = Stream.fromList messages,
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
 
 failingAfterAdapter :: (IOE :> es) => Int -> Text.Text -> Adapter es String
@@ -589,7 +590,8 @@
   Adapter
     { adapterName = "test:failing",
       source = Stream.unfoldrM step (0 :: Int),
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
   where
     step n
@@ -603,7 +605,8 @@
   Adapter
     { adapterName = "test:infinite",
       source = Stream.unfoldrM step (1 :: Int),
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
   where
     step n = do
@@ -616,7 +619,8 @@
   Adapter
     { adapterName = "test:one-then-idle",
       source = Stream.unfoldrM step False,
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
   where
     step False = do
@@ -629,7 +633,8 @@
   Adapter
     { adapterName = "test:permanent-finalizer-failure",
       source = Stream.fromEffect (pure ingested),
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
   where
     envelope = mkEnvelope (MessageId "finalizer-failure") "message"
diff --git a/test/Shibuya/Batch/ReliabilitySpec.hs b/test/Shibuya/Batch/ReliabilitySpec.hs
--- a/test/Shibuya/Batch/ReliabilitySpec.hs
+++ b/test/Shibuya/Batch/ReliabilitySpec.hs
@@ -21,7 +21,7 @@
 import Data.Text (Text)
 import Data.Text qualified as Text
 import Effectful (Eff, IOE, liftIO, runEff, (:>))
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.Adapter.Mock
   ( TrackingAck,
     getTrackedDecisions,
@@ -155,7 +155,8 @@
   Adapter
     { adapterName = "test:gated-tracked",
       source = Stream.unfoldrM step (map (mkTrackedIngested tracking) envs),
-      shutdown = liftIO $ atomically $ writeTVar gate True
+      shutdown = liftIO $ atomically $ writeTVar gate True,
+      telemetry = defaultAdapterTelemetry
     }
   where
     step [] = do
@@ -556,7 +557,8 @@
               Adapter
                 { adapterName = "test:counting-source",
                   source = source,
-                  shutdown = pure ()
+                  shutdown = pure (),
+                  telemetry = defaultAdapterTelemetry
                 }
             cfg = (defaultBatchConfig @_ @Int) {batchSize = 1, batchTimeout = 30, batchKey = scenarioBatchKey}
             handler _ _ = do
diff --git a/test/Shibuya/Runner/BatchProcessorSpec.hs b/test/Shibuya/Runner/BatchProcessorSpec.hs
--- a/test/Shibuya/Runner/BatchProcessorSpec.hs
+++ b/test/Shibuya/Runner/BatchProcessorSpec.hs
@@ -13,7 +13,7 @@
 import Data.Time (UTCTime (..), fromGregorian)
 import Effectful (Eff, IOE, liftIO, runEff, (:>))
 import Shibuya (ProcessorFailure (..), ProcessorHalt (..))
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.Adapter.Mock
   ( TrackingAck (..),
     getTrackedDecisions,
@@ -453,7 +453,7 @@
 runStatusBatches detail handler batches = do
   master <- startMaster IgnoreAll
   let values = concatMap (NE.toList . snd) batches
-      adapter = Adapter "status-batch" (Stream.fromList values) (pure ())
+      adapter = Adapter "status-batch" (Stream.fromList values) (pure ()) defaultAdapterTelemetry
       config = defaultBatchConfig {batchSize = length values}
   proc <- runSupervisedBatchWith detail master 10 (ProcessorId "status-batch") Serial config adapter handler
   liftIO $ atomically $ readTVar proc.done >>= check
diff --git a/test/Shibuya/Runner/PauseSpec.hs b/test/Shibuya/Runner/PauseSpec.hs
--- a/test/Shibuya/Runner/PauseSpec.hs
+++ b/test/Shibuya/Runner/PauseSpec.hs
@@ -10,7 +10,7 @@
 import Effectful (liftIO, runEff)
 import Effectful.State.Static.Local qualified as State
 import Shibuya qualified as Public
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.Batch (BatchConfig (..), ackAllOk, defaultBatchConfig)
 import Shibuya.Core.Ack (AckDecision (..), HaltReason (..))
 import Shibuya.Core.AckHandle (AckHandle (..))
@@ -95,13 +95,13 @@
                     pure $ fmap (\() -> (mkIngested (mkEnvelope (MessageId "live") ()) (AckHandle (\_ -> liftIO $ atomically $ modifyTVar' finishes (+ 1))), n + 1)) next
                 )
                 (1 :: Int)
-            adapter = Adapter "live" source (liftIO $ atomically $ writeTVar stop True)
+            adapter = Adapter "live" source (liftIO $ atomically $ writeTVar stop True) defaultAdapterTelemetry
             work = liftIO ((atomically (writeTQueue starts ()) >> atomically (takeTMVar handlerGate))) >> pure AckOk
             processor =
               if batching
                 then Public.mkBatchProcessor adapter (\_ _ -> work >> pure ackAllOk) defaultBatchConfig {batchSize = 1}
                 else Public.mkProcessor adapter (\_ -> work)
-            siblingAdapter = Adapter "sibling" (Stream.fromList [mkIngested (mkEnvelope (MessageId "sibling") ()) (AckHandle (\_ -> pure ()))]) (pure ())
+            siblingAdapter = Adapter "sibling" (Stream.fromList [mkIngested (mkEnvelope (MessageId "sibling") ()) (AckHandle (\_ -> pure ()))]) (pure ()) defaultAdapterTelemetry
         result <- Public.runApp Public.defaultAppConfig [(ProcessorId "live", processor), (ProcessorId "sibling", Public.mkProcessor siblingAdapter (\_ -> liftIO (atomically $ putTMVar sibling ()) >> pure AckOk))]
         app <- either (liftIO . ioError . userError . show) pure result
         liftIO $ atomically (readTQueue polls) `shouldReturn` 1
@@ -168,7 +168,7 @@
       UIO.finally
         ( runEff $ runTracingNoop $ do
             let message = mkIngested (mkEnvelope (MessageId "terminal") ()) (AckHandle (\_ -> if failure then liftIO (ioError (userError "failure")) else pure ()))
-                adapter = Adapter "terminal" (Stream.fromList [message]) (pure ())
+                adapter = Adapter "terminal" (Stream.fromList [message]) (pure ()) defaultAdapterTelemetry
             proc <-
               runSupervised
                 master
diff --git a/test/Shibuya/Runner/SupervisedSpec.hs b/test/Shibuya/Runner/SupervisedSpec.hs
--- a/test/Shibuya/Runner/SupervisedSpec.hs
+++ b/test/Shibuya/Runner/SupervisedSpec.hs
@@ -16,7 +16,7 @@
 import Effectful (Eff, IOE, liftIO, runEff, (:>))
 import Effectful.State.Static.Local qualified as State
 import Shibuya (ProcessorHalt (..))
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.Adapter.Mock
   ( getTrackedDecisions,
     mkTrackedIngested,
@@ -287,7 +287,7 @@
                           pure $ Just (mkIngested (mkEnvelope (MessageId (Text.pack (show n))) value) (AckHandle (const $ pure ())), n + 1)
                   )
                   0
-              adapter = Adapter {adapterName = "test:state", source = source, shutdown = pure ()}
+              adapter = Adapter {adapterName = "test:state", source = source, shutdown = pure (), telemetry = defaultAdapterTelemetry}
               observe message = do
                 value <- State.get @Int
                 State.modify @Int (+ 10)
@@ -405,7 +405,7 @@
               mkIngested
                 (mkEnvelope (MessageId "failed-detail") ())
                 (AckHandle (\_ -> liftIO $ ioError (userError "failed finalizer")))
-            adapter = Adapter "failed-detail" (Stream.fromList [message]) (pure ())
+            adapter = Adapter "failed-detail" (Stream.fromList [message]) (pure ()) defaultAdapterTelemetry
         sp <-
           runSupervisedWith
             (Metrics.MetricsDetailed Metrics.defaultDetailOptions)
@@ -432,7 +432,7 @@
           master <- startMaster IgnoreAll
           let waitHere = liftIO (UIO.putMVar entered () >> UIO.takeMVar blocker)
               message = mkIngested (mkEnvelope (MessageId "cancel") ()) (AckHandle (\_ -> when inFinalizer waitHere))
-              adapter = Adapter "cancel-detail" (Stream.fromList [message]) (pure ())
+              adapter = Adapter "cancel-detail" (Stream.fromList [message]) (pure ()) defaultAdapterTelemetry
           sp <-
             runSupervisedWith
               (Metrics.MetricsDetailed Metrics.defaultDetailOptions)
@@ -1068,7 +1068,8 @@
                         Adapter
                           { adapterName = "test:failing-source",
                             source = failingSource,
-                            shutdown = pure ()
+                            shutdown = pure (),
+                            telemetry = defaultAdapterTelemetry
                           }
 
                   let handler _ = do
@@ -1126,7 +1127,8 @@
                         Adapter
                           { adapterName = "test:failing-source",
                             source = failingSource,
-                            shutdown = pure ()
+                            shutdown = pure (),
+                            telemetry = defaultAdapterTelemetry
                           }
 
                   let handler _ = do
@@ -1170,7 +1172,8 @@
                     Adapter
                       { adapterName = "test:failing-immediate",
                         source = failingSource,
-                        shutdown = pure ()
+                        shutdown = pure (),
+                        telemetry = defaultAdapterTelemetry
                       }
                   handler _ = pure AckOk
 
@@ -1263,7 +1266,8 @@
                         Adapter
                           { adapterName = "test:failing-A",
                             source = failingSourceA,
-                            shutdown = pure ()
+                            shutdown = pure (),
+                            telemetry = defaultAdapterTelemetry
                           }
 
                   -- Handler A counts messages
@@ -1522,7 +1526,8 @@
   Adapter
     { adapterName = "test:supervised",
       source = Stream.fromList messages,
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
 
 validDeadLetterCode :: Text.Text -> DeadLetterCode
@@ -1549,7 +1554,7 @@
         mkIngested
           ((mkEnvelope (MessageId "cursor") ()) {cursor = Just (CursorInt 5), partition = Just "p"})
           (AckHandle (\_ -> when failFinalizer (liftIO $ UIO.throwIO (userError "finalization failed"))))
-      adapter = Adapter "cursor" (Stream.fromList [message]) (pure ())
+      adapter = Adapter "cursor" (Stream.fromList [message]) (pure ()) defaultAdapterTelemetry
   proc <- runSupervisedWith (Metrics.MetricsDetailed Metrics.defaultDetailOptions) master 10 (ProcessorId "cursor") Unordered Serial adapter (\_ -> pure decision)
   liftIO $ atomically $ readTVar proc.done >>= check
   m <- getMetrics proc
diff --git a/test/Shibuya/RunnerSpec.hs b/test/Shibuya/RunnerSpec.hs
--- a/test/Shibuya/RunnerSpec.hs
+++ b/test/Shibuya/RunnerSpec.hs
@@ -6,7 +6,7 @@
 import Data.Text qualified as Text
 import Data.Time (UTCTime (..), fromGregorian)
 import Effectful (Eff, IOE, liftIO, runEff, (:>))
-import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry)
 import Shibuya.Adapter.Mock (TrackingAck (..), newTrackingAck, trackingAckHandle)
 import Shibuya.App (AppConfig (..), AppError (..), QueueProcessor (..), defaultAppConfig, mkBatchProcessor, mkProcessor, runApp, stopApp, waitApp)
 import Shibuya.Batch (ackAll, defaultBatchConfig)
@@ -296,7 +296,8 @@
   Adapter
     { adapterName = "test:mock",
       source = Stream.fromList messages,
-      shutdown = pure ()
+      shutdown = pure (),
+      telemetry = defaultAdapterTelemetry
     }
 
 -- | Handler that records processed messages
diff --git a/test/Shibuya/Telemetry/BatchCorrelationSpec.hs b/test/Shibuya/Telemetry/BatchCorrelationSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/Telemetry/BatchCorrelationSpec.hs
@@ -0,0 +1,220 @@
+-- | Fan-in links on batch processing spans.
+--
+-- A batch has many producers and a span has one parent, so the batch span is
+-- a root that links every member's creation context, bounded at
+-- 'maxBatchLinks', each link naming its member.
+module Shibuya.Telemetry.BatchCorrelationSpec (spec) where
+
+import Control.Exception (bracket)
+import Control.Monad (forM_, void)
+import Data.ByteString.Char8 qualified as BS8
+import Data.Foldable (toList)
+import Data.HashMap.Strict qualified as HashMap
+import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)
+import Data.Maybe (isNothing)
+import Data.Text (Text)
+import Data.Text qualified as Text
+import Effectful (Eff, IOE, liftIO, runEff, (:>))
+import Numeric (showHex)
+import OpenTelemetry.Attributes (Attribute, emptyAttributes, getAttributeMap, toAttribute)
+import OpenTelemetry.Context qualified as Ctx
+import OpenTelemetry.Context.ThreadLocal qualified as Ctx
+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)
+import OpenTelemetry.Trace.Core qualified as OTel
+import OpenTelemetry.Trace.Id (Base (..))
+import OpenTelemetry.Trace.Id qualified as OTel.Id
+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..))
+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)
+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry)
+import Shibuya.Adapter.Mock (listAdapter)
+import Shibuya.Batch (BatchConfig (..), ackAllOk, defaultBatchConfig)
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), mkIngested)
+import Shibuya.Core.Metrics (ProcessorId (..))
+import Shibuya.Core.Types (Envelope (..), MessageId (..), TraceHeaders, mkEnvelope)
+import Shibuya.Internal.Runner.Supervised (runWithMetricsBatch)
+import Shibuya.Policy (Concurrency (..))
+import Shibuya.Telemetry.Correlation (BatchLinks (..), collectBatchLinks, maxBatchLinks)
+import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop, withSpan')
+import Test.Hspec
+
+spec :: Spec
+spec = describe "Shibuya.Telemetry.Correlation (batches)" $ do
+  describe "collectBatchLinks" $ do
+    it "links members in order, each with its message id" $ do
+      let result = collectBatchLinks [(MessageId "a", Just (producer 1)), (MessageId "b", Nothing), (MessageId "c", Just (producer 3))]
+      map (hexSpan . (.linkContext)) result.links `shouldBe` [spanHex 1, spanHex 3]
+      map (HashMap.lookup "messaging.message.id" . (.linkAttributes)) result.links
+        `shouldBe` [Just (toAttribute ("a" :: Text)), Just (toAttribute ("c" :: Text))]
+      result.dropped `shouldBe` 0
+
+    it "keeps one link per member when members share a context" $ do
+      let result = collectBatchLinks [(MessageId "a", Just (producer 1)), (MessageId "b", Just (producer 1))]
+      map (hexSpan . (.linkContext)) result.links `shouldBe` [spanHex 1, spanHex 1]
+      map (HashMap.lookup "messaging.message.id" . (.linkAttributes)) result.links
+        `shouldBe` [Just (toAttribute ("a" :: Text)), Just (toAttribute ("b" :: Text))]
+
+    it "stores at most maxBatchLinks and counts only valid omitted contexts" $ do
+      let full = [(MessageId (tshow i), Just (producer i)) | i <- [1 .. maxBatchLinks]]
+          tailMembers =
+            [ (MessageId "absent", Nothing),
+              (MessageId "malformed", Just malformed),
+              (MessageId "valid-1", Just (producer 1000)),
+              (MessageId "valid-2", Just (producer 1001))
+            ]
+          result = collectBatchLinks (full <> tailMembers)
+      length result.links `shouldBe` maxBatchLinks
+      result.dropped `shouldBe` 2
+
+    it "keeps sparse valid contexts before the cap" $ do
+      let members = concat [[(MessageId (tshow i), Just (producer i)), (MessageId "gap", Nothing), (MessageId "bad", Just malformed)] | i <- [1 .. 10]]
+          result = collectBatchLinks members
+      length result.links `shouldBe` 10
+      result.dropped `shouldBe` 0
+
+  describe "batch spans through the runner" $ do
+    it "is a root with one link per producer, naming each member" $ do
+      [sp] <- batchSpans =<< runBatch [member "m1" (Just (producer 1)), member "m2" (Just (producer 2)), member "m3" (Just (producer 3))]
+      isNothing sp.parent `shouldBe` True
+      map fst sp.links `shouldBe` [spanHex 1, spanHex 2, spanHex 3]
+      map snd sp.links `shouldBe` map (Just . toAttribute) ["m1", "m2", "m3" :: Text]
+      sp.traceHex `notElem` map traceHex [1, 2, 3] `shouldBe` True
+      HashMap.lookup "messaging.batch.message_count" sp.attributes `shouldBe` Just (toAttribute (3 :: Int))
+      sp.name `shouldBe` "process batch-correlation"
+      HashMap.lookup "messaging.operation.name" sp.attributes `shouldBe` Just (toAttribute ("process" :: Text))
+      HashMap.lookup "messaging.operation.type" sp.attributes `shouldBe` Just (toAttribute ("process" :: Text))
+      HashMap.member "shibuya.batch.links.dropped" sp.attributes `shouldBe` False
+
+    it "links members that share a context separately" $ do
+      [sp] <- batchSpans =<< runBatch [member "m1" (Just (producer 1)), member "m2" (Just (producer 1)), member "m3" (Just (producer 3))]
+      map fst sp.links `shouldBe` [spanHex 1, spanHex 1, spanHex 3]
+      map snd sp.links `shouldBe` map (Just . toAttribute) ["m1", "m2", "m3" :: Text]
+
+    it "has no links and no parent when no member has headers" $ do
+      [sp] <- batchSpans =<< runBatch [member "m1" Nothing, member "m2" Nothing]
+      sp.links `shouldBe` []
+      isNothing sp.parent `shouldBe` True
+
+    it "links only the members that have headers" $ do
+      [sp] <- batchSpans =<< runBatch [member "m1" Nothing, member "m2" (Just (producer 2)), member "m3" (Just malformed)]
+      map snd sp.links `shouldBe` [Just (toAttribute ("m2" :: Text))]
+
+    it "bounds 200 distinct producers at 128 links and reports 72 dropped" $ do
+      [sp] <- batchSpans =<< runBatch [member (tshow i) (Just (producer i)) | i <- [1 .. 200]]
+      length sp.links `shouldBe` 128
+      HashMap.lookup "shibuya.batch.links.dropped" sp.attributes `shouldBe` Just (toAttribute (72 :: Int))
+      HashMap.lookup "messaging.batch.message_count" sp.attributes `shouldBe` Just (toAttribute (200 :: Int))
+
+    it "does not link a superseded member" $ do
+      let superseded = (member "m2" (Just (producer 2))) {deliveryStatus = pure DeliverySuperseded}
+      [sp] <- batchSpans =<< runBatch [member "m1" (Just (producer 1)), superseded, member "m3" (Just (producer 3))]
+      map snd sp.links `shouldBe` map (Just . toAttribute) ["m1", "m3" :: Text]
+      HashMap.lookup "messaging.batch.message_count" sp.attributes `shouldBe` Just (toAttribute (2 :: Int))
+
+    forM_ [minBound .. maxBound] $ \policy ->
+      it ("ignores the adapter's " <> show policy <> " policy") $ do
+        [sp] <- batchSpans =<< runBatchWith policy [member "m1" (Just (producer 1))]
+        isNothing sp.parent `shouldBe` True
+        map fst sp.links `shouldBe` [spanHex 1]
+
+    it "is still a root inside an ambient span, which is restored afterwards" $ do
+      restored <- newIORef False
+      spans <- withExporter $ \tracer -> runEff $
+        runTracing tracer $
+          withSpan' "ambient" OTel.defaultSpanArguments $ \ambient -> do
+            batchOnce ParentAndLink [member "m1" (Just (producer 1))]
+            liftIO $ do
+              ambientCtx <- OTel.getSpanContext ambient
+              current <- traverse OTel.getSpanContext . Ctx.lookupSpan =<< Ctx.getContext
+              writeIORef restored (current == Just ambientCtx)
+      [sp] <- batchSpans spans
+      isNothing sp.parent `shouldBe` True
+      readIORef restored `shouldReturn` True
+
+    it "runs the handler and exports nothing with tracing disabled" $ do
+      handled <- newIORef (0 :: Int)
+      spans <- withExporter $ \_ -> runEff $ runTracingNoop $ do
+        let adapter = listAdapter [member "m1" (Just (producer 1))]
+        void $ runWithMetricsBatch 16 (ProcessorId "batch-correlation") Serial batchConfig adapter $ \_ _ ->
+          liftIO (modifyIORef' handled (+ 1)) >> pure ackAllOk
+      readIORef handled `shouldReturn` 1
+      length spans `shouldBe` 0
+
+--------------------------------------------------------------------------------
+-- Fixtures
+--------------------------------------------------------------------------------
+
+batchConfig :: BatchConfig es msg
+batchConfig = defaultBatchConfig {batchSize = 1000, batchTimeout = 60}
+
+member :: (IOE :> es) => Text -> Maybe TraceHeaders -> Ingested es ()
+member msgId headers =
+  mkIngested ((mkEnvelope (MessageId msgId) ()) {traceContext = headers}) (AckHandle (\_ -> pure ()))
+
+-- | A distinct, valid producer context for each index.
+producer :: Int -> TraceHeaders
+producer i = [("traceparent", BS8.pack ("00-" <> Text.unpack (traceHex i) <> "-" <> Text.unpack (spanHex i) <> "-01"))]
+
+malformed :: TraceHeaders
+malformed = [("traceparent", "00-not-a-valid-traceparent-01")]
+
+traceHex :: Int -> Text
+traceHex i = padHex 32 (i + 0xabc)
+
+spanHex :: Int -> Text
+spanHex i = padHex 16 (i + 0x100)
+
+padHex :: Int -> Int -> Text
+padHex width n = let digits = showHex n "" in Text.pack (replicate (width - length digits) '0' <> digits)
+
+tshow :: (Show a) => a -> Text
+tshow = Text.pack . show
+
+batchOnce :: (IOE :> es, Tracing :> es) => TraceCorrelation -> [Ingested es ()] -> Eff es ()
+batchOnce policy members =
+  let adapter = (listAdapter members) {telemetry = defaultAdapterTelemetry {traceCorrelation = policy}}
+   in void $ runWithMetricsBatch 16 (ProcessorId "batch-correlation") Serial batchConfig adapter (\_ _ -> pure ackAllOk)
+
+runBatch :: [Ingested '[Tracing, IOE] ()] -> IO [OTel.ImmutableSpan]
+runBatch = runBatchWith ParentAndLink
+
+runBatchWith :: TraceCorrelation -> [Ingested '[Tracing, IOE] ()] -> IO [OTel.ImmutableSpan]
+runBatchWith policy members = withExporter $ \tracer -> runEff $ runTracing tracer $ batchOnce policy members
+
+withExporter :: (OTel.Tracer -> IO ()) -> IO [OTel.ImmutableSpan]
+withExporter action = do
+  (processor, spansRef) <- inMemoryListExporter
+  bracket
+    (OTel.createTracerProvider [processor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})
+    (\provider -> void $ OTel.shutdownTracerProvider provider (Just 5_000_000))
+    (\provider -> action (OTel.makeTracer provider (OTel.InstrumentationLibrary "batch-correlation-test" "" "" emptyAttributes) OTel.tracerOptions))
+  readIORef spansRef
+
+data BatchSpan = BatchSpan
+  { name :: Text,
+    traceHex :: Text,
+    parent :: Maybe OTel.SpanContext,
+    links :: [(Text, Maybe Attribute)],
+    attributes :: HashMap.HashMap Text Attribute
+  }
+
+batchSpans :: [OTel.ImmutableSpan] -> IO [BatchSpan]
+batchSpans spans = traverse describeSpan [sp | sp <- spans, OTel.spanKind sp == OTel.Consumer]
+  where
+    describeSpan sp = do
+      hot <- readIORef (OTel.spanHot sp)
+      parent <- traverse OTel.getSpanContext (OTel.spanParent sp)
+      pure
+        BatchSpan
+          { name = OTel.hotName hot,
+            traceHex = OTel.Id.traceIdBaseEncodedText Base16 (OTel.spanContext sp).traceId,
+            parent,
+            links =
+              [ (hexSpan (OTel.frozenLinkContext link), HashMap.lookup "messaging.message.id" (getAttributeMap (OTel.frozenLinkAttributes link)))
+              | link <- toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot))
+              ],
+            attributes = getAttributeMap (OTel.hotAttributes hot)
+          }
+
+hexSpan :: OTel.SpanContext -> Text
+hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 ctx.spanId
diff --git a/test/Shibuya/Telemetry/CorrelationSpec.hs b/test/Shibuya/Telemetry/CorrelationSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/Telemetry/CorrelationSpec.hs
@@ -0,0 +1,282 @@
+-- | Trace correlation policies for per-message processing spans.
+--
+-- Every case drives a real tracer whose provider exports into memory, and
+-- inspects the exported Consumer spans: their trace, their parent, and
+-- their links.
+module Shibuya.Telemetry.CorrelationSpec (spec) where
+
+import Control.Exception (bracket)
+import Control.Monad (forM_, void)
+import Data.ByteString (ByteString)
+import Data.Foldable (toList)
+import Data.HashMap.Strict qualified as HashMap
+import Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef)
+import Data.Maybe (isNothing)
+import Data.Text (Text)
+import Effectful (Eff, IOE, liftIO, runEff, (:>))
+import OpenTelemetry.Attributes (emptyAttributes, getAttributeMap, toAttribute)
+import OpenTelemetry.Context qualified as Ctx
+import OpenTelemetry.Context.ThreadLocal qualified as Ctx
+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)
+import OpenTelemetry.Trace.Core qualified as OTel
+import OpenTelemetry.Trace.Id (Base (..))
+import OpenTelemetry.Trace.Id qualified as OTel.Id
+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..))
+import OpenTelemetry.Trace.TraceState qualified as TraceState
+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)
+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry)
+import Shibuya.Adapter.Mock (listAdapter)
+import Shibuya.Core.Ack (AckDecision (..))
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), mkIngested)
+import Shibuya.Core.Metrics (ProcessorId (..))
+import Shibuya.Core.Types (Envelope (..), MessageId (..), TraceHeaders, mkEnvelope)
+import Shibuya.Internal.Runner.Supervised (runWithMetrics)
+import Shibuya.Telemetry.Correlation (consumerSpanArgsLinkedTo, creationLink, creationLinkWith)
+import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop, withSpan')
+import Shibuya.Telemetry.Process (withProcessSpan, withProcessSpanIO)
+import Shibuya.Telemetry.Propagation (currentTraceHeaders, extractTraceContext)
+import Shibuya.Telemetry.Semantic (processSpanName)
+import Test.Hspec
+
+spec :: Spec
+spec = describe "Shibuya.Telemetry.Correlation" $ do
+  describe "AdapterTelemetry" $ do
+    it "defaults to ParentAndLink and is changed by record update" $ do
+      defaultAdapterTelemetry.traceCorrelation `shouldBe` ParentAndLink
+      (defaultAdapterTelemetry {traceCorrelation = LinkOnly}).traceCorrelation `shouldBe` LinkOnly
+
+  describe "creationLink" $ do
+    it "decodes a valid traceparent into an attribute-free link" $
+      case creationLink producerHeaders of
+        Nothing -> expectationFailure "expected a link"
+        Just link -> do
+          hexTrace link.linkContext `shouldBe` producerTraceHex
+          hexSpan link.linkContext `shouldBe` producerSpanHex
+          HashMap.null link.linkAttributes `shouldBe` True
+
+    it "keeps tracestate on the linked context" $
+      case creationLink (producerHeaders <> [("tracestate", "congo=t61rcWkgMzE")]) of
+        Nothing -> expectationFailure "expected a link"
+        Just link -> TraceState.toList link.linkContext.traceState `shouldBe` [(TraceState.Key "congo", TraceState.Value "t61rcWkgMzE")]
+
+    it "returns Nothing without a traceparent" $
+      isNothing (creationLink [("tracestate", "congo=t61rcWkgMzE")]) `shouldBe` True
+
+    it "returns Nothing for a malformed traceparent" $
+      isNothing (creationLink malformedHeaders) `shouldBe` True
+
+    it "carries caller-supplied attributes with creationLinkWith" $ do
+      let attrs = HashMap.singleton "messaging.message.id" (toAttribute ("m-1" :: Text))
+      case creationLinkWith attrs producerHeaders of
+        Nothing -> expectationFailure "expected a link"
+        Just link -> link.linkAttributes `shouldBe` attrs
+
+    it "builds Consumer-kind arguments carrying the links" $ do
+      let links = maybe [] pure (creationLink producerHeaders)
+          args = consumerSpanArgsLinkedTo links
+      args.kind `shouldBe` OTel.Consumer
+      length args.links `shouldBe` 1
+
+  describe "processing spans through the runner" $ do
+    it "ParentAndLink: child of the producer, with one link to it" $ do
+      [sp] <- consumerSpans =<< runMessage ParentAndLink (Just producerHeaders) Nothing
+      hexTrace sp.context `shouldBe` producerTraceHex
+      fmap hexSpan sp.parent `shouldBe` Just producerSpanHex
+      map (\l -> (hexTrace l, hexSpan l)) sp.links `shouldBe` [(producerTraceHex, producerSpanHex)]
+
+    it "LinkOnly: a root in a new trace, with one link to the producer" $ do
+      [sp] <- consumerSpans =<< runMessage LinkOnly (Just producerHeaders) Nothing
+      hexTrace sp.context `shouldNotBe` producerTraceHex
+      isNothing sp.parent `shouldBe` True
+      map (\l -> (hexTrace l, hexSpan l)) sp.links `shouldBe` [(producerTraceHex, producerSpanHex)]
+
+    it "LinkOnly: still a root when an ambient span is active, which is restored afterwards" $ do
+      restored <- newIORef Nothing
+      spans <- withExporter $ \tracer -> runEff $
+        runTracing tracer $
+          withSpan' "ambient" OTel.defaultSpanArguments $ \ambient -> do
+            runOnce LinkOnly (Just producerHeaders) Nothing
+            liftIO $ do
+              ambientCtx <- OTel.getSpanContext ambient
+              current <- Ctx.lookupSpan <$> Ctx.getContext
+              currentCtx <- traverse OTel.getSpanContext current
+              writeIORef restored (Just (Just ambientCtx == currentCtx))
+      [sp] <- consumerSpans spans
+      isNothing sp.parent `shouldBe` True
+      length sp.links `shouldBe` 1
+      readIORef restored `shouldReturn` Just True
+
+    forM_ [minBound .. maxBound] $ \policy -> do
+      forM_ [("absent", Nothing), ("malformed", Just malformedHeaders)] $ \(label, headers) -> do
+        it (show policy <> ": " <> label <> " headers keep the ambient parent and add no link") $ do
+          (ambientSpanHex, spans) <- withAmbient $ \_tracer ->
+            runOnce policy headers Nothing
+          [sp] <- consumerSpans spans
+          sp.links `shouldBe` []
+          fmap hexSpan sp.parent `shouldBe` Just ambientSpanHex
+
+        it (show policy <> ": " <> label <> " headers without an ambient span give a root") $ do
+          [sp] <- consumerSpans =<< runMessage policy headers Nothing
+          sp.links `shouldBe` []
+          isNothing sp.parent `shouldBe` True
+
+    it "LinkOnly: currentTraceHeaders in the handler names the new trace and the Consumer span" $ do
+      seen <- newIORef Nothing
+      spans <- withExporter $ \tracer ->
+        runEff $
+          runTracing tracer $
+            runOnce LinkOnly (Just producerHeaders) (Just seen)
+      [sp] <- consumerSpans spans
+      Just (Just headers) <- readIORef seen
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "handler saw no decodable traceparent"
+        Just ctx -> do
+          hexTrace ctx `shouldBe` hexTrace sp.context
+          hexSpan ctx `shouldBe` hexSpan sp.context
+          hexTrace ctx `shouldNotBe` producerTraceHex
+
+    it "LinkOnly: a superseded delivery still opens a linked span that ends successfully" $ do
+      spans <- withExporter $ \tracer -> runEff $ runTracing tracer $ do
+        let message = (producedMessage (Just producerHeaders)) {deliveryStatus = pure DeliverySuperseded}
+        void $ runWithMetrics 4 (ProcessorId "correlation") (policyAdapter LinkOnly [message]) (\_ -> pure AckOk)
+      [sp] <- consumerSpans spans
+      isNothing sp.parent `shouldBe` True
+      length sp.links `shouldBe` 1
+      sp.status `shouldBe` OTel.Ok
+      sp.ackDecisions `shouldBe` [toAttribute ("superseded" :: Text)]
+
+    forM_ [minBound .. maxBound] $ \policy ->
+      it (show policy <> ": disabled tracing runs the handler and exports nothing") $ do
+        handled <- newIORef (0 :: Int)
+        spans <- withExporter $ \_tracer -> runEff $ runTracingNoop $ do
+          let adapter = policyAdapter policy [producedMessage (Just producerHeaders)]
+          void $ runWithMetrics 4 (ProcessorId "correlation") adapter (\_ -> liftIO (modifyIORef' handled (+ 1)) >> pure AckOk)
+        readIORef handled `shouldReturn` 1
+        length spans `shouldBe` 0
+
+  describe "withProcessSpan and withProcessSpanIO called directly" $ do
+    forM_ [minBound .. maxBound] $ \policy -> do
+      it (show policy <> ": both give the runner's span shape") $ do
+        [runner] <- consumerSpans =<< runMessage policy (Just producerHeaders) Nothing
+        [effect] <- consumerSpans =<< withExporter (\tracer -> runEff $ runTracing tracer $ withProcessSpan policy processName (Just producerHeaders) (\_ -> pure ()))
+        [io] <- consumerSpans =<< withExporter (\tracer -> withProcessSpanIO tracer policy processName (Just producerHeaders) (\_ -> pure ()))
+        forM_ [effect, io] $ \sp -> do
+          sp.name `shouldBe` runner.name
+          fmap hexSpan sp.parent `shouldBe` fmap hexSpan runner.parent
+          map hexSpan sp.links `shouldBe` map hexSpan runner.links
+          (hexTrace sp.context == producerTraceHex) `shouldBe` (hexTrace runner.context == producerTraceHex)
+
+      it (show policy <> ": the IO helper keeps the ambient parent without headers") $ do
+        (ambientSpanHex, spans) <- withAmbient $ \tracer ->
+          liftIO $ withProcessSpanIO tracer policy processName Nothing (\_ -> pure ())
+        [sp] <- consumerSpans spans
+        sp.links `shouldBe` []
+        fmap hexSpan sp.parent `shouldBe` Just ambientSpanHex
+
+      it (show policy <> ": withProcessSpan under runTracingNoop passes the sentinel span") $ do
+        recording <- runEff $ runTracingNoop $ withProcessSpan policy processName (Just producerHeaders) (liftIO . OTel.isRecording)
+        recording `shouldBe` False
+
+--------------------------------------------------------------------------------
+-- Fixtures
+--------------------------------------------------------------------------------
+
+producerTraceHex :: Text
+producerTraceHex = "0af7651916cd43dd8448eb211c80319c"
+
+producerSpanHex :: Text
+producerSpanHex = "b7ad6b7169203331"
+
+producerHeaders :: TraceHeaders
+producerHeaders = [("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")]
+
+malformedHeaders :: [(ByteString, ByteString)]
+malformedHeaders = [("traceparent", "00-not-a-valid-traceparent-01")]
+
+processName :: Text
+processName = processSpanName "correlation"
+
+producedMessage :: (IOE :> es) => Maybe TraceHeaders -> Ingested es ()
+producedMessage headers =
+  mkIngested ((mkEnvelope (MessageId "m-1") ()) {traceContext = headers}) (AckHandle (\_ -> pure ()))
+
+policyAdapter :: (IOE :> es) => TraceCorrelation -> [Ingested es msg] -> Adapter es msg
+policyAdapter policy messages =
+  (listAdapter messages) {telemetry = defaultAdapterTelemetry {traceCorrelation = policy}}
+
+-- | Process one message on the calling thread (the unsupervised runner path),
+-- so an ambient span on this thread is visible to the processing span.
+runOnce ::
+  (IOE :> es, Tracing :> es) =>
+  TraceCorrelation ->
+  Maybe TraceHeaders ->
+  Maybe (IORef (Maybe (Maybe TraceHeaders))) ->
+  Eff es ()
+runOnce policy headers seen =
+  void $
+    runWithMetrics 4 (ProcessorId "correlation") (policyAdapter policy [producedMessage headers]) $ \_ -> do
+      forM_ seen $ \ref -> currentTraceHeaders >>= liftIO . writeIORef ref . Just
+      pure AckOk
+
+runMessage :: TraceCorrelation -> Maybe TraceHeaders -> Maybe (IORef (Maybe (Maybe TraceHeaders))) -> IO [OTel.ImmutableSpan]
+runMessage policy headers seen =
+  withExporter $ \tracer -> runEff $ runTracing tracer $ runOnce policy headers seen
+
+-- | Run inside an ambient span and return that span's ID with the exports.
+withAmbient :: (OTel.Tracer -> Eff '[Tracing, IOE] ()) -> IO (Text, [OTel.ImmutableSpan])
+withAmbient action = do
+  ambientRef <- newIORef ""
+  spans <- withExporter $ \tracer -> runEff $
+    runTracing tracer $
+      withSpan' "ambient" OTel.defaultSpanArguments $ \ambient -> do
+        ctx <- liftIO $ OTel.getSpanContext ambient
+        liftIO $ writeIORef ambientRef (hexSpan ctx)
+        action tracer
+  ambientSpanHex <- readIORef ambientRef
+  pure (ambientSpanHex, spans)
+
+withExporter :: (OTel.Tracer -> IO ()) -> IO [OTel.ImmutableSpan]
+withExporter action = do
+  (processor, spansRef) <- inMemoryListExporter
+  bracket
+    (OTel.createTracerProvider [processor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})
+    (\provider -> void $ OTel.shutdownTracerProvider provider (Just 5_000_000))
+    (\provider -> action (OTel.makeTracer provider (OTel.InstrumentationLibrary "correlation-test" "" "" emptyAttributes) OTel.tracerOptions))
+  readIORef spansRef
+
+data ConsumerSpan = ConsumerSpan
+  { name :: Text,
+    context :: OTel.SpanContext,
+    parent :: Maybe OTel.SpanContext,
+    links :: [OTel.SpanContext],
+    status :: OTel.SpanStatus,
+    ackDecisions :: [OTel.Attribute]
+  }
+
+consumerSpans :: [OTel.ImmutableSpan] -> IO [ConsumerSpan]
+consumerSpans spans = traverse describeSpan [sp | sp <- spans, OTel.spanKind sp == OTel.Consumer]
+  where
+    describeSpan sp = do
+      hot <- readIORef (OTel.spanHot sp)
+      parent <- traverse OTel.getSpanContext (OTel.spanParent sp)
+      pure
+        ConsumerSpan
+          { name = OTel.hotName hot,
+            context = OTel.spanContext sp,
+            parent,
+            links = map OTel.frozenLinkContext (toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot))),
+            status = OTel.hotStatus hot,
+            ackDecisions =
+              [ value
+              | event <- toList (appendOnlyBoundedCollectionValues (OTel.hotEvents hot)),
+                OTel.eventName event == "shibuya.ack.decision",
+                Just value <- [HashMap.lookup "shibuya.ack.decision" (getAttributeMap (OTel.eventAttributes event))]
+              ]
+          }
+
+hexTrace :: OTel.SpanContext -> Text
+hexTrace ctx = OTel.Id.traceIdBaseEncodedText Base16 ctx.traceId
+
+hexSpan :: OTel.SpanContext -> Text
+hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 ctx.spanId
diff --git a/test/Shibuya/Telemetry/SemanticSpec.hs b/test/Shibuya/Telemetry/SemanticSpec.hs
--- a/test/Shibuya/Telemetry/SemanticSpec.hs
+++ b/test/Shibuya/Telemetry/SemanticSpec.hs
@@ -80,7 +80,7 @@
     case spans of
       [processingSpan] -> do
         hot <- readIORef (spanHot processingSpan)
-        hotName hot `shouldBe` "skip-proc process"
+        hotName hot `shouldBe` "process skip-proc"
         let events = toList (appendOnlyBoundedCollectionValues (hotEvents hot))
         map eventName events `shouldBe` ["shibuya.ack.decision"]
         hotStatus hot `shouldBe` Ok
@@ -109,8 +109,9 @@
     case spans of
       [s] -> do
         hot <- readIORef (spanHot s)
-        hotName hot `shouldBe` "test-proc process"
+        hotName hot `shouldBe` "process test-proc"
         let attrs = getAttributeMap (hotAttributes hot)
+        attrs `shouldHaveTextAttribute` ("messaging.operation.name", "process")
         attrs `shouldHaveTextAttribute` ("messaging.system", "shibuya")
         attrs `shouldHaveTextAttribute` ("messaging.message.id", "m-1")
         attrs `shouldHaveTextAttribute` ("messaging.destination.name", "test-proc")