diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,42 @@
 # Changelog
 
+## 0.9.1.0 — 2026-09-21
+
+### Bug Fixes
+
+- Preserve the earliest unresolved delivery when multiple buffered Kafka
+  callbacks request retry. Delivery tokens distinguish an original callback
+  from the replay that may resolve its recovery boundary, so later buffered or
+  duplicate callbacks cannot advance the stored offset past unresolved work.
+- Throw exhausted acknowledgement operations synchronously from the finalizer
+  boundary. Core can now retain them as processor failures even after ingestion
+  has stopped instead of depending on a future source poll of the fatal slot.
+  The public `KafkaAcknowledgementException` carries the underlying
+  `KafkaError` for callers that need to classify it.
+- Fence callbacks retained across a revoke/assign cycle when callers install
+  `kafkaRebalanceHandler`. Assignment generations prevent an old owner from
+  storing, seeking, or pausing a partition after reassignment.
+- Serialize each delivery finalizer with exception-safe ownership. Successful
+  duplicates are no-ops, while failed or cancelled attempts remain retryable.
+
+### Other Changes
+
+- Require `shibuya-core ^>=0.10.0.0` across the library, tests, benchmark,
+  and example so the adapter is certified against the lifecycle-remediation
+  release candidate. Keep all three repository packages on version `0.9.1.0`.
+- Add deterministic reference-model, cancellation, timeout, terminal-failure,
+  repeated-shutdown, buffered-retry, restart, and actual-reassignment coverage.
+- Make the live restart fixture stream delivery identities to external files,
+  emit a reconciled per-delivery ledger, and fail on duplicates, missing
+  deliveries, unexpected deliveries, or malformed payloads without retaining
+  the identity set in the measured process heap.
+- Keep the documented absence of a DLQ producer: `AckDeadLetter` still warns
+  and stores the offset deliberately.
+- Exclude `effectful-core` 2.7.0.0 through 2.7.1.0 from the library, tests,
+  benchmark, and examples because upstream records a per-operation performance
+  regression for dynamically dispatched effects. The 2.6 family and 2.7.1.1 or
+  later remain accepted.
+
 ## 0.9.0.1 — 2026-09-15
 
 ### Other Changes
diff --git a/shibuya-kafka-adapter.cabal b/shibuya-kafka-adapter.cabal
--- a/shibuya-kafka-adapter.cabal
+++ b/shibuya-kafka-adapter.cabal
@@ -1,6 +1,6 @@
 cabal-version: 3.12
 name: shibuya-kafka-adapter
-version: 0.9.0.1
+version: 0.9.1.0
 synopsis: Kafka adapter for the Shibuya queue processing framework
 description:
   A Shibuya adapter that integrates with Apache Kafka via kafka-effectful
@@ -61,13 +61,13 @@
     base ^>=4.21.0.0,
     bytestring ^>=0.12,
     containers ^>=0.7,
-    effectful-core >=2.6.1 && <2.8,
+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),
     hs-opentelemetry-api ^>=1.0,
     hs-opentelemetry-semantic-conventions ^>=1.40,
     hw-kafka-client >=5.3 && <6,
     hw-kafka-streamly ^>=0.2,
     kafka-effectful ^>=0.3.1.0,
-    shibuya-core ^>=0.9.0.1,
+    shibuya-core ^>=0.10.0.0,
     stm ^>=2.5,
     streamly ^>=0.11,
     streamly-core ^>=0.3,
@@ -112,14 +112,14 @@
     base ^>=4.21.0.0,
     bytestring,
     containers,
-    effectful-core,
+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),
     hs-opentelemetry-api ^>=1.0,
     hs-opentelemetry-semantic-conventions ^>=1.40,
     hw-kafka-client,
     kafka-effectful,
     process,
     random,
-    shibuya-core ^>=0.9.0.1,
+    shibuya-core ^>=0.10.0.0,
     shibuya-kafka-adapter,
     stm,
     streamly,
diff --git a/src/Shibuya/Adapter/Kafka.hs b/src/Shibuya/Adapter/Kafka.hs
--- a/src/Shibuya/Adapter/Kafka.hs
+++ b/src/Shibuya/Adapter/Kafka.hs
@@ -32,7 +32,8 @@
 -- 3. On @AckOk@, the offset is stored locally; auto-commit or consumer close
 --    later flushes stored offsets to the broker.
 -- 4. On @AckRetry@, the offset is not stored. The adapter seeks the partition
---    back to the failed message so Kafka can redeliver it.
+--    back to the earliest unresolved delivery so Kafka can redeliver it. Later
+--    buffered callbacks cannot replace or commit past that recovery boundary.
 -- 5. On @AckDeadLetter@, the offset is stored after a loud stderr warning.
 -- 6. On @AckHalt@, the partition is paused and offset is not stored.
 --
@@ -65,9 +66,10 @@
 -- are filtered out of the poll stream. Any error that survives that filter is
 -- fatal by construction (for example, an SSL handshake failure, an authentication
 -- failure, or an invalid broker configuration) and terminates the stream by
--- throwing through the 'Effectful.Error.Static.Error' @KafkaError@ effect. The
--- caller observes the failure by receiving a @Left err@ from the
--- @runError \@KafkaError@ scope around 'Shibuya.App.runApp'.
+-- throwing through the 'Effectful.Error.Static.Error' @KafkaError@ effect.
+-- Exhausted acknowledgement operations instead throw a synchronous typed
+-- exception at the finalizer boundary; Shibuya records that as a processor
+-- failure even if ingestion has already ended.
 --
 -- == AckHalt Partition Pause Semantics
 --
@@ -84,15 +86,15 @@
 -- 'kafkaRebalanceHandler' is optional. Install it with
 -- @Kafka.Consumer.setCallback (Kafka.Consumer.rebalanceCallback (kafkaRebalanceHandler state))@
 -- before creating the consumer when you want stderr visibility into assignment
--- changes and eager cleanup of retry barriers for revoked partitions. Without it,
--- the seek barrier still self-heals when messages are finalized at or below the
--- barrier offset. Cooperative rebalance fencing of in-flight work is outside this
--- adapter's scope.
+-- changes and assignment-generation fencing of callbacks retained by a revoked
+-- owner. Without it, retry recovery remains safe within one assignment, but stale
+-- callbacks are not fenced across ownership changes.
 module Shibuya.Adapter.Kafka
   ( -- * Adapter
     kafkaAdapter,
     kafkaAdapterWith,
     KafkaAdapterState,
+    KafkaAcknowledgementException (..),
     newKafkaAdapterState,
     kafkaRebalanceHandler,
 
@@ -116,8 +118,6 @@
 import Control.Concurrent.STM (atomically, writeTVar)
 import Control.Monad.IO.Class (liftIO)
 import Data.ByteString (ByteString)
-import Data.IORef (atomicModifyIORef')
-import Data.Map.Strict qualified as Map
 import Data.Set qualified as Set
 import Data.Text qualified as Text
 import Effectful (Eff, IOE, (:>))
@@ -126,10 +126,10 @@
 import Kafka.Consumer.Types (ConsumerGroupId (..), OffsetCommit (..), RebalanceEvent (..))
 import Kafka.Consumer.Types qualified as KC
 import Kafka.Effectful.Consumer.Effect (KafkaConsumer, commitAllOffsets, subscription)
-import Kafka.Types (BatchSize (..), BrokerAddress (..), KafkaError (..), PartitionId, Timeout (..), TopicName (..))
+import Kafka.Types (BatchSize (..), BrokerAddress (..), KafkaError (..), Timeout (..), TopicName (..))
 import Shibuya.Adapter (Adapter (..))
 import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..), defaultConfig)
-import Shibuya.Adapter.Kafka.Internal (KafkaAdapterState (..), dropStaleRecords, ingestedStream, kafkaSource, mkIngested, newKafkaAdapterState, withConsumerLock)
+import Shibuya.Adapter.Kafka.Internal (KafkaAcknowledgementException (..), KafkaAdapterState (..), dropStaleRecords, ingestedStream, kafkaSource, markPartitionsAssigned, markPartitionsRevoked, mkIngested, newKafkaAdapterState, withConsumerLock)
 import System.IO (hPutStrLn, stderr)
 
 -- | Create a Kafka adapter with the given configuration.
@@ -208,8 +208,9 @@
 --
 -- Install with 'Kafka.Consumer.setCallback' and
 -- 'Kafka.Consumer.rebalanceCallback' before creating the consumer. The callback
--- logs every rebalance event to stderr and clears pending retry barriers for
--- revoked partitions. It does not fence in-flight work.
+-- logs every rebalance event to stderr, clears local retry state on revocation,
+-- and advances an assignment generation so late callbacks from a revoked owner
+-- cannot store, seek, or pause a later assignment.
 kafkaRebalanceHandler ::
   KafkaAdapterState ->
   KC.KafkaConsumer ->
@@ -218,12 +219,7 @@
 kafkaRebalanceHandler state _consumer event = do
   hPutStrLn stderr $ "[shibuya-kafka-adapter] rebalance: " <> show event
   case event of
-    RebalanceRevoke revoked ->
-      clearRevokedBarriers revoked
-    _ ->
-      pure ()
-  where
-    clearRevokedBarriers :: [(TopicName, PartitionId)] -> IO ()
-    clearRevokedBarriers revoked =
-      atomicModifyIORef' state.seekBarrier $ \barriers ->
-        (foldr Map.delete barriers revoked, ())
+    RebalanceBeforeAssign assigned -> markPartitionsAssigned state assigned
+    RebalanceAssign assigned -> markPartitionsAssigned state assigned
+    RebalanceBeforeRevoke revoked -> markPartitionsRevoked state revoked
+    RebalanceRevoke revoked -> markPartitionsRevoked state revoked
diff --git a/src/Shibuya/Adapter/Kafka/Internal.hs b/src/Shibuya/Adapter/Kafka/Internal.hs
--- a/src/Shibuya/Adapter/Kafka/Internal.hs
+++ b/src/Shibuya/Adapter/Kafka/Internal.hs
@@ -3,7 +3,10 @@
 module Shibuya.Adapter.Kafka.Internal
   ( -- * Adapter State
     KafkaAdapterState (..),
+    KafkaAcknowledgementException (..),
     newKafkaAdapterState,
+    markPartitionsAssigned,
+    markPartitionsRevoked,
     withConsumerLock,
 
     -- * Stream Construction
@@ -22,13 +25,15 @@
 import Control.Concurrent (threadDelay)
 import Control.Concurrent.MVar (MVar, newMVar, putMVar, takeMVar)
 import Control.Concurrent.STM (TVar, newTVarIO, readTVarIO)
+import Control.Monad (unless, when)
 import Data.ByteString (ByteString)
 import Data.Function ((&))
-import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)
+import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, newIORef, readIORef)
 import Data.Map.Strict (Map)
 import Data.Map.Strict qualified as Map
 import Data.Text qualified as Text
 import Data.Time.Clock (NominalDiffTime)
+import Data.Word (Word64)
 import Effectful (Eff, IOE, (:>))
 import Effectful qualified
 import Effectful.Error.Static (Error, catchError, throwError)
@@ -55,10 +60,60 @@
 
 type PartitionKey = (TopicName, PartitionId)
 
+newtype DeliveryToken = DeliveryToken Word64
+  deriving stock (Eq, Ord, Show)
+
+newtype AssignmentGeneration = AssignmentGeneration Word64
+  deriving stock (Eq, Show)
+
+data RetryBarrier = RetryBarrier
+  { offset :: !Offset,
+    retryToken :: !DeliveryToken
+  }
+
+data PartitionAckState = PartitionAckState
+  { barrier :: !(Maybe RetryBarrier),
+    validFromToken :: !DeliveryToken,
+    assignmentGeneration :: !AssignmentGeneration,
+    assigned :: !Bool
+  }
+
+data AckState = AckState
+  { nextDeliveryToken :: !Word64,
+    partitions :: !(Map PartitionKey PartitionAckState)
+  }
+
+data DeliveryAttempt = DeliveryAttempt
+  { partition :: !PartitionKey,
+    recordOffset :: !Offset,
+    token :: !DeliveryToken,
+    generation :: !AssignmentGeneration
+  }
+
+-- | A Kafka operation failed after the adapter's bounded retry budget. This is
+-- a synchronous exception so Shibuya's finalizer boundary can retry it and,
+-- when exhausted, retain a processor failure with the delivery identity.
+newtype KafkaAcknowledgementException = KafkaAcknowledgementException KafkaError
+  deriving stock (Show)
+
+instance Exception.Exception KafkaAcknowledgementException
+
+initialPartitionAckState :: PartitionAckState
+initialPartitionAckState =
+  PartitionAckState
+    { barrier = Nothing,
+      validFromToken = DeliveryToken 0,
+      assignmentGeneration = AssignmentGeneration 0,
+      assigned = True
+    }
+
+initialAckState :: AckState
+initialAckState = AckState {nextDeliveryToken = 1, partitions = Map.empty}
+
 -- | Mutable state shared by the source stream and ack handles.
 data KafkaAdapterState = KafkaAdapterState
   { shutdownVar :: !(TVar Bool),
-    seekBarrier :: !(IORef (Map PartitionKey Offset)),
+    ackState :: !(IORef AckState),
     fatalError :: !(IORef (Maybe KafkaError)),
     -- | Serializes every librdkafka consumer operation. Under
     --     'Shibuya.App.runApp' the consumer handle is shared between the ingester
@@ -78,10 +133,43 @@
 newKafkaAdapterState =
   KafkaAdapterState
     <$> newTVarIO False
-    <*> newIORef Map.empty
+    <*> newIORef initialAckState
     <*> newIORef Nothing
     <*> newMVar ()
 
+-- | Mark partitions assigned to this adapter generation. A revoke/assign cycle
+-- advances the generation, so ack handles retained from the previous owner can
+-- no longer store, seek, or pause the newly assigned partition.
+markPartitionsAssigned :: KafkaAdapterState -> [PartitionKey] -> IO ()
+markPartitionsAssigned state = setPartitionsAssigned state True
+
+-- | Mark partitions revoked and discard only the local retry ledger. Kafka's
+-- committed offset remains the durable recovery boundary; callbacks from the
+-- old generation are fenced by the incremented assignment generation.
+markPartitionsRevoked :: KafkaAdapterState -> [PartitionKey] -> IO ()
+markPartitionsRevoked state = setPartitionsAssigned state False
+
+setPartitionsAssigned :: KafkaAdapterState -> Bool -> [PartitionKey] -> IO ()
+setPartitionsAssigned state isAssigned keys =
+  atomicModifyIORef' state.ackState $ \ackState' ->
+    let partitions' = foldr (Map.alter (Just . transition . maybe initialPartitionAckState id)) ackState'.partitions keys
+     in (ackState' {partitions = partitions'}, ())
+  where
+    transition partitionState
+      | partitionState.assigned == isAssigned =
+          if isAssigned
+            then partitionState
+            else partitionState {barrier = Nothing}
+      | otherwise =
+          partitionState
+            { barrier = Nothing,
+              validFromToken = DeliveryToken 0,
+              assignmentGeneration = nextGeneration partitionState.assignmentGeneration,
+              assigned = isAssigned
+            }
+
+    nextGeneration (AssignmentGeneration generation') = AssignmentGeneration (generation' + 1)
+
 -- | Run a librdkafka consumer operation while holding the shared consumer
 -- lock, guaranteeing no other consumer call runs concurrently on the same
 -- handle. See 'consumerLock' for why this is mandatory. The lock is always
@@ -157,10 +245,12 @@
   Stream.filterM $ \case
     Left _ -> pure True
     Right cr -> do
-      barriers <- Effectful.liftIO $ readIORef state.seekBarrier
-      pure $ case Map.lookup (partitionKey cr) barriers of
-        Nothing -> True
-        Just barrierOff -> cr.crOffset <= barrierOff
+      ackState' <- Effectful.liftIO $ readIORef state.ackState
+      let partitionState = Map.findWithDefault initialPartitionAckState (partitionKey cr) ackState'.partitions
+      pure $
+        partitionState.assigned && case partitionState.barrier of
+          Nothing -> True
+          Just retryBarrier -> cr.crOffset <= retryBarrier.offset
 
 -- | Create an 'AckHandle' for a single 'ConsumerRecord'.
 --
@@ -175,35 +265,79 @@
   KafkaAdapterState ->
   KafkaAdapterConfig ->
   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->
-  AckHandle es
-mkAckHandle state config cr = AckHandle $ \case
-  AckOk ->
-    ackAttempt state (storeGuarded state cr)
+  Eff es (AckHandle es)
+mkAckHandle state config cr = do
+  attempt <- Effectful.liftIO $ registerDelivery state cr
+  finalizerLock <- Effectful.liftIO $ newMVar ()
+  completed <- Effectful.liftIO $ newIORef False
+  pure $ AckHandle $ \decision ->
+    Exception.bracket_
+      (Effectful.liftIO (takeMVar finalizerLock))
+      (Effectful.liftIO (putMVar finalizerLock ()))
+      $ do
+        isCompleted <- Effectful.liftIO $ readIORef completed
+        unless isCompleted $ do
+          finalizeAttempt state config attempt cr decision
+          Effectful.liftIO $ atomicWriteIORef completed True
+
+registerDelivery :: KafkaAdapterState -> ConsumerRecord k v -> IO DeliveryAttempt
+registerDelivery state cr =
+  atomicModifyIORef' state.ackState $ \ackState' ->
+    let partition = partitionKey cr
+        partitionState = Map.findWithDefault initialPartitionAckState partition ackState'.partitions
+        token = DeliveryToken ackState'.nextDeliveryToken
+        attempt =
+          DeliveryAttempt
+            { partition,
+              recordOffset = cr.crOffset,
+              token,
+              generation = partitionState.assignmentGeneration
+            }
+        ackState'' =
+          ackState'
+            { nextDeliveryToken = ackState'.nextDeliveryToken + 1,
+              partitions = Map.insert partition partitionState ackState'.partitions
+            }
+     in (ackState'', attempt)
+
+finalizeAttempt ::
+  (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>
+  KafkaAdapterState ->
+  KafkaAdapterConfig ->
+  DeliveryAttempt ->
+  ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->
+  AckDecision ->
+  Eff es ()
+finalizeAttempt state config attempt cr = \case
+  AckOk -> storeGuarded state attempt cr (pure ())
   AckRetry (RetryDelay delay) -> do
     Effectful.liftIO $ delayRetry delay
-    Effectful.liftIO $
-      atomicModifyIORef' state.seekBarrier $ \barriers ->
-        (Map.insert (partitionKey cr) cr.crOffset barriers, ())
-    ackAttempt state $
-      withConsumerLock state $
-        seekPartitions
-          [ TopicPartition
-              { tpTopicName = cr.crTopic,
-                tpPartition = cr.crPartition,
-                tpOffset = PartitionOffset (unOffset cr.crOffset)
-              }
-          ]
-          (boundedLockTimeout config.pollTimeout)
-  AckDeadLetter reason -> do
-    Effectful.liftIO $
-      hPutStrLn stderr $
-        "[shibuya-kafka-adapter] WARNING: dead-lettered message DROPPED (no DLQ producer): "
-          <> show (cr.crTopic, cr.crPartition, cr.crOffset)
-          <> " reason="
-          <> Text.unpack (renderDeadLetterReason reason)
-    ackAttempt state (storeGuarded state cr)
-  AckHalt _ ->
-    ackAttempt state (withConsumerLock state (pausePartitions [(cr.crTopic, cr.crPartition)]))
+    mbTarget <- Effectful.liftIO $ recordRetry state attempt
+    case mbTarget of
+      Nothing -> pure ()
+      Just target ->
+        ackAttempt state $
+          withConsumerLock state $
+            seekPartitions
+              [ TopicPartition
+                  { tpTopicName = cr.crTopic,
+                    tpPartition = cr.crPartition,
+                    tpOffset = PartitionOffset (unOffset target)
+                  }
+              ]
+              (boundedLockTimeout config.pollTimeout)
+  AckDeadLetter reason ->
+    storeGuarded state attempt cr $
+      Effectful.liftIO $
+        hPutStrLn stderr $
+          "[shibuya-kafka-adapter] WARNING: dead-lettered message DROPPED (no DLQ producer): "
+            <> show (cr.crTopic, cr.crPartition, cr.crOffset)
+            <> " reason="
+            <> Text.unpack (renderDeadLetterReason reason)
+  AckHalt _ -> do
+    isAccepted <- Effectful.liftIO $ deliveryIsAccepted state attempt
+    when isAccepted $
+      ackAttempt state (withConsumerLock state (pausePartitions [(cr.crTopic, cr.crPartition)]))
 
 ackAttempt ::
   (Error KafkaError :> es, IOE :> es) =>
@@ -218,7 +352,9 @@
     go attempt =
       action `catchError` \_ err ->
         if isFatal err || attempt >= maxAttempts
-          then Effectful.liftIO $ recordFatalError state err
+          then do
+            Effectful.liftIO $ recordFatalError state err
+            Exception.throwIO (KafkaAcknowledgementException err)
           else do
             Effectful.liftIO $ threadDelay retryDelayMicros
             go (attempt + 1)
@@ -230,21 +366,69 @@
     Nothing -> (Just err, ())
 
 storeGuarded ::
-  (KafkaConsumer :> es, IOE :> es) =>
+  (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>
   KafkaAdapterState ->
+  DeliveryAttempt ->
   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->
+  Eff es () ->
   Eff es ()
-storeGuarded state cr = do
-  shouldStore <-
-    Effectful.liftIO $
-      atomicModifyIORef' state.seekBarrier $ \barriers ->
-        case Map.lookup (partitionKey cr) barriers of
-          Nothing -> (barriers, True)
-          Just barrierOff
-            | cr.crOffset <= barrierOff -> (Map.delete (partitionKey cr) barriers, True)
-            | otherwise -> (barriers, False)
-  if shouldStore then withConsumerLock state (storeOffsetMessage cr) else pure ()
+storeGuarded state attempt cr beforeStore = do
+  shouldStore <- Effectful.liftIO $ claimStore state attempt
+  when shouldStore $ do
+    beforeStore
+    ackAttempt state (withConsumerLock state (storeOffsetMessage cr))
 
+recordRetry :: KafkaAdapterState -> DeliveryAttempt -> IO (Maybe Offset)
+recordRetry state attempt =
+  atomicModifyIORef' state.ackState $ \ackState' ->
+    let partitionState = Map.findWithDefault initialPartitionAckState attempt.partition ackState'.partitions
+     in if not (attemptIsAccepted partitionState attempt)
+          then (ackState', Nothing)
+          else
+            let retryBarrier = case partitionState.barrier of
+                  Nothing -> RetryBarrier attempt.recordOffset attempt.token
+                  Just existing
+                    | attempt.recordOffset < existing.offset -> RetryBarrier attempt.recordOffset attempt.token
+                    | attempt.recordOffset == existing.offset && attempt.token > existing.retryToken ->
+                        existing {retryToken = attempt.token}
+                    | otherwise -> existing
+                partitionState' = partitionState {barrier = Just retryBarrier}
+                ackState'' = ackState' {partitions = Map.insert attempt.partition partitionState' ackState'.partitions}
+             in (ackState'', Just retryBarrier.offset)
+
+claimStore :: KafkaAdapterState -> DeliveryAttempt -> IO Bool
+claimStore state attempt =
+  atomicModifyIORef' state.ackState $ \ackState' ->
+    let partitionState = Map.findWithDefault initialPartitionAckState attempt.partition ackState'.partitions
+     in if not (attemptIsAccepted partitionState attempt)
+          then (ackState', False)
+          else case partitionState.barrier of
+            Nothing -> (ackState', True)
+            Just retryBarrier
+              | attempt.recordOffset > retryBarrier.offset -> (ackState', False)
+              | attempt.recordOffset < retryBarrier.offset -> (ackState', True)
+              | attempt.token <= retryBarrier.retryToken -> (ackState', False)
+              | otherwise ->
+                  let partitionState' =
+                        partitionState
+                          { barrier = Nothing,
+                            validFromToken = max partitionState.validFromToken attempt.token
+                          }
+                      ackState'' = ackState' {partitions = Map.insert attempt.partition partitionState' ackState'.partitions}
+                   in (ackState'', True)
+
+deliveryIsAccepted :: KafkaAdapterState -> DeliveryAttempt -> IO Bool
+deliveryIsAccepted state attempt = do
+  ackState' <- readIORef state.ackState
+  let partitionState = Map.findWithDefault initialPartitionAckState attempt.partition ackState'.partitions
+  pure $ attemptIsAccepted partitionState attempt
+
+attemptIsAccepted :: PartitionAckState -> DeliveryAttempt -> Bool
+attemptIsAccepted partitionState attempt =
+  partitionState.assigned
+    && partitionState.assignmentGeneration == attempt.generation
+    && attempt.token >= partitionState.validFromToken
+
 partitionKey :: ConsumerRecord k v -> PartitionKey
 partitionKey cr = (cr.crTopic, cr.crPartition)
 
@@ -261,11 +445,13 @@
   KafkaAdapterState ->
   KafkaAdapterConfig ->
   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->
-  Ingested es (Maybe ByteString)
-mkIngested state config cr =
-  Core.mkIngested
-    (consumerRecordToEnvelope cr)
-    (mkAckHandle state config cr)
+  Eff es (Ingested es (Maybe ByteString))
+mkIngested state config cr = do
+  ackHandle <- mkAckHandle state config cr
+  pure $
+    Core.mkIngested
+      (consumerRecordToEnvelope cr)
+      ackHandle
 
 -- | Transform a poll stream of @Either KafkaError ConsumerRecord@ into a
 -- stream of 'Ingested'.
@@ -280,10 +466,10 @@
 -- injects a synthetic @Left@ without standing up a real consumer.
 ingestedStream ::
   (Error KafkaError :> es) =>
-  (ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> Ingested es (Maybe ByteString)) ->
+  (ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> Eff es (Ingested es (Maybe ByteString))) ->
   Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))) ->
   Stream (Eff es) (Ingested es (Maybe ByteString))
 ingestedStream mkI =
   Stream.mapMaybeM $ \case
-    Right cr -> pure (Just (mkI cr))
+    Right cr -> Just <$> mkI cr
     Left err -> throwError err
diff --git a/test/Kafka/TestEnv.hs b/test/Kafka/TestEnv.hs
--- a/test/Kafka/TestEnv.hs
+++ b/test/Kafka/TestEnv.hs
@@ -7,6 +7,7 @@
     -- * Producing
     produceMessages,
     produceKeyedMessages,
+    producePartitionMessages,
 
     -- * Consuming via Adapter
     consumeN,
@@ -157,6 +158,25 @@
       flushProducer
   case result of
     Left err -> error $ "Failed to produce: " <> show err
+    Right () -> pure ()
+
+-- | Produce payloads to exact partitions for deterministic rebalance tests.
+producePartitionMessages :: TestEnv -> [(Int, ByteString)] -> IO ()
+producePartitionMessages env records = do
+  result <- runEff . runError @KafkaError $ do
+    runKafkaProducer (mkProducerProps env) $ do
+      forM_ records $ \(partition, payload) ->
+        produceMessage
+          ProducerRecord
+            { prTopic = env.testTopic,
+              prPartition = SpecifiedPartition partition,
+              prKey = Nothing,
+              prValue = Just payload,
+              prHeaders = mempty
+            }
+      flushProducer
+  case result of
+    Left err -> error $ "Failed to produce to partitions: " <> show err
     Right () -> pure ()
 
 -- | Consume N messages from the test topic via the adapter, applying the given ack decision.
diff --git a/test/Shibuya/Adapter/Kafka/AckHandleTest.hs b/test/Shibuya/Adapter/Kafka/AckHandleTest.hs
--- a/test/Shibuya/Adapter/Kafka/AckHandleTest.hs
+++ b/test/Shibuya/Adapter/Kafka/AckHandleTest.hs
@@ -1,29 +1,43 @@
 module Shibuya.Adapter.Kafka.AckHandleTest (tests) where
 
+import Control.Concurrent.Async qualified as Async
+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)
+import Control.Exception (try)
 import Data.ByteString (ByteString)
 import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, newIORef, readIORef)
 import Data.Int (Int64)
-import Effectful (Eff, IOE, liftIO, runEff, (:>))
+import Data.Map.Strict qualified as Map
+import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, runEff, withEffToIO, (:>))
 import Effectful.Dispatch.Dynamic (interpret)
 import Effectful.Error.Static (Error, runErrorNoCallStack, throwError)
 import Kafka.Consumer (RdKafkaRespErrT (..))
-import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), PartitionOffset (..), Timestamp (..), TopicPartition (..))
+import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), PartitionOffset (..), RebalanceEvent (..), Timestamp (..), TopicPartition (..))
 import Kafka.Effectful.Consumer.Effect (KafkaConsumer (..))
 import Kafka.Types (BatchSize (..), KafkaError (..), PartitionId (..), Timeout (..), TopicName (..))
+import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter.Kafka (kafkaRebalanceHandler)
 import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..))
-import Shibuya.Adapter.Kafka.Internal (KafkaAdapterState (..), ingestedStream, kafkaSource, mkAckHandle, newKafkaAdapterState)
+import Shibuya.Adapter.Kafka.Internal (KafkaAcknowledgementException (..), KafkaAdapterState (..), ingestedStream, kafkaSource, mkAckHandle, mkIngested, newKafkaAdapterState)
+import Shibuya.App (ProcessorId (..), defaultAppConfig, getAppMaster, mkProcessor, runApp, waitApp)
 import Shibuya.Core.Ack (AckDecision (..), HaltReason (..), RetryDelay (..))
 import Shibuya.Core.AckHandle (AckHandle (..))
 import Shibuya.Core.Ingested (Ingested)
+import Shibuya.Internal.Runner.Master (ProcessorLifecycle (..), getLifecycleSnapshot)
+import Shibuya.Telemetry.Effect (runTracingNoop)
 import Streamly.Data.Fold qualified as Fold
 import Streamly.Data.Stream qualified as Stream
+import System.Random (mkStdGen, randomR)
+import System.Timeout (timeout)
 import Test.Tasty (TestTree, testGroup)
 import Test.Tasty.HUnit (assertEqual, assertFailure, testCase)
 
 data MockState = MockState
   { storeAttempts :: !Int,
+    storedOffsets :: ![Offset],
     pauseAttempts :: !Int,
     seekCalls :: ![TopicPartition],
+    seekTimeouts :: ![Timeout],
+    storeBlock :: !(Maybe (MVar (), MVar ())),
     storeFailuresRemaining :: !Int,
     pauseFailuresRemaining :: !Int,
     seekFailuresRemaining :: !Int,
@@ -37,11 +51,20 @@
   testGroup
     "AckHandle"
     [ testCase "transient store failures retry and then succeed" testTransientStoreRetry,
-      testCase "persistent transient store failure records fatal slot without throwing" testPersistentStoreFailure,
-      testCase "fatal store failure records fatal slot after one attempt" testFatalStoreFailure,
-      testCase "AckHalt pause failure does not throw and records fatal slot" testAckHaltPauseFailure,
+      testCase "persistent transient store failure records fatal slot and throws" testPersistentStoreFailure,
+      testCase "fatal store failure records fatal slot and throws after one attempt" testFatalStoreFailure,
+      testCase "AckHalt pause failure records fatal slot and throws" testAckHaltPauseFailure,
       testCase "AckRetry seeks exact failed offset and does not store" testAckRetrySeeks,
+      testCase "AckRetry caps the consumer-lock seek timeout" testAckRetryBoundsSeek,
       testCase "seek barrier prevents stale successor store" testBarrierSkipsSuccessorStore,
+      testCase "earliest retry survives later retry and acknowledgement" testEarliestRetrySurvives,
+      testCase "one delivery cannot resolve its own retry" testRetryRequiresRedelivery,
+      testCase "repeated retry on one delivery is idempotent" testRepeatedRetry,
+      testCase "fixed-seed sequences match the earliest-unresolved reference model" testReferenceModelSeeds,
+      testCase "exhausted acknowledgement throws immediately" testPersistentStoreFailureThrows,
+      testCase "revocation fences an old delivery callback" testRevocationFencesCallback,
+      testCase "cancellation releases finalizer and consumer ownership" testCancellationReleasesOwnership,
+      testCase "terminal acknowledgement failure reaches the core lifecycle" testTerminalFailureReachesCore,
       testCase "source observes fatal slot before polling" testSourceObservesFatalSlot
     ]
 
@@ -61,8 +84,7 @@
   let err = KafkaResponseError RdKafkaRespErrTransport
   mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}
   state <- newKafkaAdapterState
-  result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk
-  assertRight result
+  assertAckFailure err $ runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk
   final <- readIORef mock
   fatal <- readIORef state.fatalError
   assertEqual "store attempts" 3 final.storeAttempts
@@ -73,8 +95,7 @@
   let err = KafkaBadConfiguration
   mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}
   state <- newKafkaAdapterState
-  result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk
-  assertRight result
+  assertAckFailure err $ runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk
   final <- readIORef mock
   fatal <- readIORef state.fatalError
   assertEqual "store attempts" 1 final.storeAttempts
@@ -85,8 +106,7 @@
   let err = KafkaResponseError RdKafkaRespErrTransport
   mock <- newIORef defaultMockState {pauseFailuresRemaining = 99, pauseError = err}
   state <- newKafkaAdapterState
-  result <- runFinalizer mock $ finalizeRecord state (recordAt 42) (AckHalt (HaltFatal "stop"))
-  assertRight result
+  assertAckFailure err $ runFinalizer mock $ finalizeRecord state (recordAt 42) (AckHalt (HaltFatal "stop"))
   final <- readIORef mock
   fatal <- readIORef state.fatalError
   assertEqual "pause attempts" 3 final.pauseAttempts
@@ -105,6 +125,18 @@
     [TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset 42)]
     final.seekCalls
 
+testAckRetryBoundsSeek :: IO ()
+testAckRetryBoundsSeek = do
+  mock <- newIORef defaultMockState
+  state <- newKafkaAdapterState
+  let slowConfig = testConfig {pollTimeout = Timeout 5000}
+  result <- runFinalizer mock $ do
+    AckHandle finalize <- mkAckHandle state slowConfig (recordAt 42)
+    finalize (AckRetry (RetryDelay 0))
+  assertRight result
+  final <- readIORef mock
+  assertEqual "seek timeout" [Timeout 100] final.seekTimeouts
+
 testBarrierSkipsSuccessorStore :: IO ()
 testBarrierSkipsSuccessorStore = do
   mock <- newIORef defaultMockState
@@ -117,6 +149,164 @@
   final <- readIORef mock
   assertEqual "only retried message stored" 1 final.storeAttempts
 
+testEarliestRetrySurvives :: IO ()
+testEarliestRetrySurvives = do
+  mock <- newIORef defaultMockState
+  state <- newKafkaAdapterState
+  result <- runFinalizer mock $ do
+    finalizeRecord state (recordAt 42) (AckRetry (RetryDelay 0))
+    finalizeRecord state (recordAt 43) (AckRetry (RetryDelay 0))
+    finalizeRecord state (recordAt 43) AckOk
+  assertRight result
+  final <- readIORef mock
+  assertEqual
+    "both retries seek the earliest unresolved offset"
+    [ TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset 42),
+      TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset 42)
+    ]
+    final.seekCalls
+  assertEqual "later acknowledgement remains fenced" 0 final.storeAttempts
+
+testRetryRequiresRedelivery :: IO ()
+testRetryRequiresRedelivery = do
+  mock <- newIORef defaultMockState
+  state <- newKafkaAdapterState
+  result <- runFinalizer mock $ do
+    AckHandle finalize <- mkAckHandle state testConfig (recordAt 42)
+    finalize (AckRetry (RetryDelay 0))
+    finalize AckOk
+  assertRight result
+  final <- readIORef mock
+  assertEqual "same delivery cannot store after requesting retry" 0 final.storeAttempts
+
+testRepeatedRetry :: IO ()
+testRepeatedRetry = do
+  mock <- newIORef defaultMockState
+  state <- newKafkaAdapterState
+  result <- runFinalizer mock $ do
+    AckHandle finalize <- mkAckHandle state testConfig (recordAt 42)
+    finalize (AckRetry (RetryDelay 0))
+    finalize (AckRetry (RetryDelay 0))
+  assertRight result
+  final <- readIORef mock
+  assertEqual "one successful retry performs one seek" 1 (length final.seekCalls)
+
+testReferenceModelSeeds :: IO ()
+testReferenceModelSeeds =
+  -- EP-44's integrated release gate requires at least 1,000 recorded model
+  -- cases. Keep the range deterministic so a failure names its replayable
+  -- seed and the ordinary package suite exercises the full gate.
+  mapM_ runSeed [400040 .. 401039]
+  where
+    runSeed seed = do
+      let generator = mkStdGen seed
+          (baseDelta, generator') = randomR (0, 5 :: Int64) generator
+          (gap, generator'') = randomR (1, 5 :: Int64) generator'
+          (laterFirst, _) = randomR (False, True) generator''
+          base = 40 + baseDelta
+          later = base + gap
+          (firstOffset, secondOffset) = if laterFirst then (later, base) else (base, later)
+          retryOrder = [firstOffset, secondOffset]
+          expectedSeeks = map toTopicPartition (runningMinimum retryOrder)
+      mock <- newIORef defaultMockState
+      state <- newKafkaAdapterState
+      result <- runFinalizer mock $ do
+        first <- mkAckHandle state testConfig (recordAt firstOffset)
+        second <- mkAckHandle state testConfig (recordAt secondOffset)
+        finalizeHandle first (AckRetry (RetryDelay 0))
+        finalizeHandle second (AckRetry (RetryDelay 0))
+        prematureLater <- mkAckHandle state testConfig (recordAt later)
+        finalizeHandle prematureLater AckOk
+        replayBase <- mkAckHandle state testConfig (recordAt base)
+        finalizeHandle replayBase AckOk
+        replayLater <- mkAckHandle state testConfig (recordAt later)
+        finalizeHandle replayLater AckOk
+      assertRight result
+      final <- readIORef mock
+      assertEqual ("seed " <> show seed <> " seek boundary") expectedSeeks final.seekCalls
+      assertEqual ("seed " <> show seed <> " stored offsets") [Offset base, Offset later] final.storedOffsets
+
+    runningMinimum = \case
+      [] -> []
+      first : rest -> scanl min first rest
+
+    toTopicPartition offset =
+      TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset offset)
+
+    finalizeHandle (AckHandle finalize) = finalize
+
+testPersistentStoreFailureThrows :: IO ()
+testPersistentStoreFailureThrows = do
+  let err = KafkaResponseError RdKafkaRespErrTransport
+  mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}
+  state <- newKafkaAdapterState
+  assertAckFailure err $ runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk
+
+testRevocationFencesCallback :: IO ()
+testRevocationFencesCallback = do
+  mock <- newIORef defaultMockState
+  state <- newKafkaAdapterState
+  result <- runFinalizer mock $ do
+    AckHandle finalize <- mkAckHandle state testConfig (recordAt 42)
+    liftIO $
+      kafkaRebalanceHandler
+        state
+        (error "consumer handle is not inspected")
+        (RebalanceRevoke [(TopicName "orders", PartitionId 0)])
+    finalize AckOk
+  assertRight result
+  final <- readIORef mock
+  assertEqual "revoked callback cannot store" 0 final.storeAttempts
+
+testCancellationReleasesOwnership :: IO ()
+testCancellationReleasesOwnership = do
+  started <- newEmptyMVar
+  release <- newEmptyMVar
+  mock <- newIORef defaultMockState {storeBlock = Just (started, release)}
+  state <- newKafkaAdapterState
+  result <- runFinalizer mock $ do
+    AckHandle finalize <- mkAckHandle state testConfig (recordAt 42)
+    withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
+      worker <- Async.async (runInIO (finalize AckOk))
+      takeMVar started
+      Async.cancel worker
+      atomicModifyIORef' mock (\mockState -> (mockState {storeBlock = Nothing}, ()))
+      mbCompleted <- timeout 1000000 (runInIO (finalize AckOk))
+      case mbCompleted of
+        Nothing -> assertFailure "finalizer lock or consumer lock remained held after cancellation"
+        Just () -> pure ()
+  assertRight result
+  final <- readIORef mock
+  assertEqual "cancelled attempt plus successful retry" 2 final.storeAttempts
+
+testTerminalFailureReachesCore :: IO ()
+testTerminalFailureReachesCore = do
+  let err = KafkaBadConfiguration
+      processorId = ProcessorId "kafka-terminal-ack-failure"
+  mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}
+  state <- newKafkaAdapterState
+  result <-
+    timeout 5000000 $
+      runEff . runErrorNoCallStack @KafkaError . runMockConsumer mock . runTracingNoop $ do
+        ingested <- mkIngested state testConfig (recordAt 42)
+        let adapter =
+              Adapter
+                { adapterName = "kafka:test-terminal-ack-failure",
+                  source = Stream.fromList [ingested],
+                  shutdown = pure ()
+                }
+            processor = mkProcessor adapter (\_ -> pure AckOk)
+        appResult <- runApp defaultAppConfig [(processorId, processor)]
+        case appResult of
+          Left appError -> error $ "runApp failed: " <> show appError
+          Right appHandle -> do
+            waitApp appHandle
+            lifecycle <- getLifecycleSnapshot (getAppMaster appHandle)
+            pure (Map.lookup processorId lifecycle)
+  case result of
+    Just (Right (Just (LifecycleFailed _ _))) -> pure ()
+    other -> assertFailure $ "expected retained terminal acknowledgement failure, got: " <> show other
+
 testSourceObservesFatalSlot :: IO ()
 testSourceObservesFatalSlot = do
   let err = KafkaBadConfiguration
@@ -144,9 +334,9 @@
   Eff es a
 runMockConsumer mock =
   interpret $ \_env -> \case
-    StoreOffsetMessage _ -> attemptStore mock
+    StoreOffsetMessage cr -> attemptStore mock cr.crOffset
     PausePartitions _ -> attemptPause mock
-    SeekPartitions tps _ -> recordSeek mock tps
+    SeekPartitions tps seekTimeout -> recordSeek mock tps seekTimeout
     PollMessage _ -> error "AckHandleTest: PollMessage not exercised"
     PollMessageBatch _ _ -> error "AckHandleTest: PollMessageBatch not exercised"
     PollMessageEither _ -> error "AckHandleTest: PollMessageEither not exercised"
@@ -164,17 +354,29 @@
 
 unreachableBuilder ::
   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->
-  Ingested es (Maybe ByteString)
+  Eff es (Ingested es (Maybe ByteString))
 unreachableBuilder _ = error "AckHandleTest: source should not yield records"
 
-attemptStore :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> Eff es ()
-attemptStore mock = do
-  mbErr <-
+attemptStore ::
+  (IOE :> es, Error KafkaError :> es) =>
+  IORef MockState ->
+  Offset ->
+  Eff es ()
+attemptStore mock offset = do
+  (mbBlock, mbErr) <-
     liftIO $
       atomicModifyIORef' mock $ \s ->
         let remaining = s.storeFailuresRemaining
-            s' = s {storeAttempts = s.storeAttempts + 1, storeFailuresRemaining = max 0 (remaining - 1)}
-         in (s', if remaining > 0 then Just s.storeError else Nothing)
+            s' =
+              s
+                { storeAttempts = s.storeAttempts + 1,
+                  storedOffsets = s.storedOffsets <> [offset],
+                  storeFailuresRemaining = max 0 (remaining - 1)
+                }
+         in (s', (s.storeBlock, if remaining > 0 then Just s.storeError else Nothing))
+  liftIO $ case mbBlock of
+    Nothing -> pure ()
+    Just (started, release) -> putMVar started () >> takeMVar release
   maybe (pure ()) throwError mbErr
 
 attemptPause :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> Eff es ()
@@ -187,13 +389,18 @@
          in (s', if remaining > 0 then Just s.pauseError else Nothing)
   maybe (pure ()) throwError mbErr
 
-recordSeek :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> [TopicPartition] -> Eff es ()
-recordSeek mock tps = do
+recordSeek :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> [TopicPartition] -> Timeout -> Eff es ()
+recordSeek mock tps seekTimeout = do
   mbErr <-
     liftIO $
       atomicModifyIORef' mock $ \s ->
         let remaining = s.seekFailuresRemaining
-            s' = s {seekCalls = s.seekCalls <> tps, seekFailuresRemaining = max 0 (remaining - 1)}
+            s' =
+              s
+                { seekCalls = s.seekCalls <> tps,
+                  seekTimeouts = s.seekTimeouts <> [seekTimeout],
+                  seekFailuresRemaining = max 0 (remaining - 1)
+                }
          in (s', if remaining > 0 then Just s.seekError else Nothing)
   maybe (pure ()) throwError mbErr
 
@@ -204,8 +411,9 @@
   AckDecision ->
   Eff es ()
 finalizeRecord state cr decision =
-  let AckHandle finalize = mkAckHandle state testConfig cr
-   in finalize decision
+  do
+    AckHandle finalize <- mkAckHandle state testConfig cr
+    finalize decision
 
 recordAt :: Int64 -> ConsumerRecord (Maybe ByteString) (Maybe ByteString)
 recordAt offset =
@@ -231,8 +439,11 @@
 defaultMockState =
   MockState
     { storeAttempts = 0,
+      storedOffsets = [],
       pauseAttempts = 0,
       seekCalls = [],
+      seekTimeouts = [],
+      storeBlock = Nothing,
       storeFailuresRemaining = 0,
       pauseFailuresRemaining = 0,
       seekFailuresRemaining = 0,
@@ -245,3 +456,10 @@
 assertRight = \case
   Left err -> assertFailure $ "expected Right, got Left: " <> show err
   Right _ -> pure ()
+
+assertAckFailure :: KafkaError -> IO (Either KafkaError ()) -> IO ()
+assertAckFailure expected action = do
+  result <- try @KafkaAcknowledgementException action
+  case result of
+    Left (KafkaAcknowledgementException actual) -> assertEqual "acknowledgement error" expected actual
+    Right value -> assertFailure $ "expected KafkaAcknowledgementException, got: " <> show value
diff --git a/test/Shibuya/Adapter/Kafka/AdapterTest.hs b/test/Shibuya/Adapter/Kafka/AdapterTest.hs
--- a/test/Shibuya/Adapter/Kafka/AdapterTest.hs
+++ b/test/Shibuya/Adapter/Kafka/AdapterTest.hs
@@ -4,7 +4,7 @@
 module Shibuya.Adapter.Kafka.AdapterTest (tests) where
 
 import Data.ByteString (ByteString)
-import Effectful (runEff)
+import Effectful (Eff, runEff)
 import Effectful.Error.Static (runError)
 import Kafka.Consumer.Types (ConsumerRecord)
 import Kafka.Types (KafkaError (..))
@@ -28,7 +28,7 @@
 -- never fires.
 unreachableBuilder ::
   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->
-  Ingested es (Maybe ByteString)
+  Eff es (Ingested es (Maybe ByteString))
 unreachableBuilder _ = error "AdapterTest: Right branch should not be reached"
 
 testFatalPropagation :: IO ()
diff --git a/test/Shibuya/Adapter/Kafka/IntegrationTest.hs b/test/Shibuya/Adapter/Kafka/IntegrationTest.hs
--- a/test/Shibuya/Adapter/Kafka/IntegrationTest.hs
+++ b/test/Shibuya/Adapter/Kafka/IntegrationTest.hs
@@ -1,23 +1,28 @@
 module Shibuya.Adapter.Kafka.IntegrationTest (tests) where
 
+import Control.Concurrent.Async qualified as Async
+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
+import Control.Concurrent.STM (TChan, atomically, newTChanIO, readTChan, writeTChan)
 import Control.Exception (throwIO)
 import Control.Monad (forM)
 import Control.Monad.IO.Class (liftIO)
 import Data.ByteString (ByteString)
 import Data.ByteString.Char8 qualified as BS8
-import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)
+import Data.IORef (atomicModifyIORef', modifyIORef', newIORef, readIORef, writeIORef)
 import Data.List (nub, sort)
 import Data.Maybe (mapMaybe)
 import Data.Text qualified as Text
-import Effectful (runEff)
+import Effectful (Eff, Limit (..), Persistence (..), UnliftStrategy (..), runEff, withEffToIO)
 import Effectful.Error.Static (runError)
-import Kafka.Consumer.Types (OffsetCommit (..), OffsetReset (..))
+import Kafka.Consumer.Types (ConsumerRecord (..), OffsetCommit (..), OffsetReset (..), RebalanceEvent (..))
 import Kafka.Effectful.Consumer
   ( brokersList,
     groupId,
     noAutoOffsetStore,
     offsetReset,
+    rebalanceCallback,
     runKafkaConsumer,
+    setCallback,
     topics,
   )
 import Kafka.Effectful.Consumer.Effect (commitAllOffsets, pollMessageBatch)
@@ -28,16 +33,18 @@
     createTopicWithPartitions,
     produceKeyedMessages,
     produceMessages,
+    producePartitionMessages,
     withTestEnv,
   )
 import Kafka.Types
   ( BatchSize (..),
     KafkaError,
+    PartitionId (..),
     Timeout (..),
     TopicName (..),
   )
 import Shibuya.Adapter (Adapter (..))
-import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter)
+import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter, kafkaAdapterWith, kafkaRebalanceHandler, newKafkaAdapterState)
 import Shibuya.App (ProcessorId (..), defaultAppConfig, mkProcessor, runApp, waitApp)
 import Shibuya.Core.Ack (AckDecision (..), RetryDelay (..))
 import Shibuya.Core.AckHandle (AckHandle (..))
@@ -60,8 +67,12 @@
       testCase "Batch polling" testBatchPolling,
       testCase "Graceful shutdown" testGracefulShutdown,
       testCase "Idle graceful shutdown completes promptly" testIdleGracefulShutdown,
+      testCase "Repeated shutdown is idempotent" testRepeatedShutdown,
       testCase "AckRetry redelivers within the same session" testAckRetryRedelivery,
+      testCase "later buffered retry cannot replace the earliest recovery boundary" testBufferedRetryBoundary,
       testCase "AckRetry is not committed past when session exits" testAckRetryAbandonedSession,
+      testCase "late callback after revocation cannot advance the broker offset" testRevokedCallbackDoesNotCommit,
+      testCase "actual reassignment fences a late callback from the old owner" testActualReassignmentFence,
       testCase "Handler exception redelivers instead of skipping" testHandlerExceptionRedelivery
     ]
 
@@ -205,6 +216,22 @@
     Just (Left (_cs, err)) -> assertFailure $ "idle shutdown failed: " <> show err
     Just (Right ()) -> pure ()
 
+testRepeatedShutdown :: IO ()
+testRepeatedShutdown = withTestEnv $ \env -> do
+  createTopic env
+
+  timedResult <- timeout 3000000 $ runEff . runError @KafkaError $ do
+    let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore
+        sub = topics [env.testTopic] <> offsetReset Earliest
+    runKafkaConsumer props sub $ do
+      Adapter {shutdown} <- kafkaAdapter (testConfig env)
+      shutdown
+      shutdown
+  case timedResult of
+    Nothing -> assertFailure "repeated shutdown did not terminate promptly"
+    Just (Left (_cs, err)) -> assertFailure $ "repeated shutdown failed: " <> show err
+    Just (Right ()) -> pure ()
+
 testAckRetryRedelivery :: IO ()
 testAckRetryRedelivery = withTestEnv $ \env -> do
   createTopic env
@@ -251,6 +278,55 @@
     Left (_cs, err) -> assertFailure $ "post-commit verification failed: " <> show err
     Right () -> pure ()
 
+testBufferedRetryBoundary :: IO ()
+testBufferedRetryBoundary = withTestEnv $ \env -> do
+  createTopic env
+  let payloads = ["boundary-42", "boundary-43"]
+  produceMessages env payloads
+
+  result <- runEff . runError @KafkaError $ do
+    let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore
+        sub = topics [env.testTopic] <> offsetReset Earliest
+    runKafkaConsumer props sub $ do
+      Adapter {source} <- kafkaAdapter (testConfig env)
+      firstBuffered <- liftIO $ newIORef Nothing
+      deliveryIndex <- liftIO $ newIORef (0 :: Int)
+      replayedPayloads <- liftIO $ newIORef ([] :: [ByteString])
+      Stream.fold Fold.drain
+        $ Stream.mapM
+          ( \ingested -> do
+              index <- liftIO $ atomicModifyIORef' deliveryIndex (\n -> (n + 1, n))
+              case index of
+                0 -> liftIO $ writeIORef firstBuffered (Just ingested)
+                1 -> do
+                  first <-
+                    liftIO (readIORef firstBuffered) >>= \case
+                      Just value -> pure value
+                      Nothing -> error "missing first buffered delivery"
+                  finalizeIngested first (AckRetry (RetryDelay 0))
+                  finalizeIngested ingested (AckRetry (RetryDelay 0))
+                _ -> do
+                  case ingested of
+                    Ingested {envelope = Envelope {payload = Just payload}} ->
+                      liftIO $ modifyIORef' replayedPayloads (<> [payload])
+                    _ -> pure ()
+                  finalizeIngested ingested AckOk
+          )
+        $ Stream.take 4 source
+      replayed <- liftIO $ readIORef replayedPayloads
+      liftIO $
+        assertEqual
+          "retry seeks the earliest delivery before its successor"
+          payloads
+          replayed
+      commitAllOffsets OffsetCommit
+  case result of
+    Left (_cs, err) -> assertFailure $ "buffered retry boundary failed: " <> show err
+    Right () -> pure ()
+
+  noRedelivery <- pollPayloads env 3
+  assertEqual "both replayed deliveries committed" [] noRedelivery
+
 testAckRetryAbandonedSession :: IO ()
 testAckRetryAbandonedSession = withTestEnv $ \env -> do
   createTopic env
@@ -295,6 +371,100 @@
   delivered <- readIORef redelivered
   assertBool ("expected ab-2 redelivery, saw " <> show delivered) ("ab-2" `elem` delivered)
 
+testRevokedCallbackDoesNotCommit :: IO ()
+testRevokedCallbackDoesNotCommit = withTestEnv $ \env -> do
+  createTopic env
+  produceMessages env ["revoked-late-callback"]
+
+  firstSession <- runEff . runError @KafkaError $ do
+    let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore
+        sub = topics [env.testTopic] <> offsetReset Earliest
+    runKafkaConsumer props sub $ do
+      state <- liftIO newKafkaAdapterState
+      Adapter {source} <- kafkaAdapterWith state (testConfig env)
+      delivered <- Stream.fold Fold.toList $ Stream.take 1 source
+      case delivered of
+        [ingested] -> do
+          liftIO $
+            kafkaRebalanceHandler
+              state
+              (error "consumer handle is not inspected")
+              (RebalanceRevoke [(env.testTopic, PartitionId 0)])
+          finalizeIngested ingested AckOk
+        other -> liftIO $ assertFailure $ "expected one delivery before revoke, got " <> show (length other)
+  case firstSession of
+    Left (_cs, err) -> assertFailure $ "revoked session failed: " <> show err
+    Right () -> pure ()
+
+  replayed <- consumeN env 1 AckOk
+  assertEqual
+    "revoked delivery remains recoverable"
+    [Just "revoked-late-callback"]
+    [envelope.payload | envelope <- replayed]
+
+testActualReassignmentFence :: IO ()
+testActualReassignmentFence = withTestEnv $ \env -> do
+  createTopicWithPartitions env 2
+  producePartitionMessages env [(0, "rebalance-partition-0"), (1, "rebalance-partition-1")]
+  state <- newKafkaAdapterState
+  events <- newTChanIO
+  handlesReady <- newEmptyMVar
+  releaseFirstConsumer <- newEmptyMVar
+
+  let callback consumer event = do
+        kafkaRebalanceHandler state consumer event
+        atomically $ writeTChan events event
+      firstProps =
+        brokersList [env.testBroker]
+          <> groupId env.testGroupId
+          <> noAutoOffsetStore
+          <> setCallback (rebalanceCallback callback)
+      sub = topics [env.testTopic] <> offsetReset Earliest
+      firstConsumer =
+        runEff . runError @KafkaError $
+          runKafkaConsumer firstProps sub $ do
+            Adapter {source} <- kafkaAdapterWith state (testConfig env)
+            delivered <- Stream.fold Fold.toList $ Stream.take 2 source
+            withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
+              putMVar handlesReady (map (callbackHandle runInIO) delivered)
+              takeMVar releaseFirstConsumer
+      secondConsumer =
+        runEff . runError @KafkaError $
+          runKafkaConsumer
+            (brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore)
+            sub
+            ( do
+                _ <- forM [1 .. 20 :: Int] $ \_ -> pollMessageBatch (Timeout 250) (BatchSize 100)
+                pure ()
+            )
+
+  Async.withAsync firstConsumer $ \firstAsync -> do
+    mbHandles <- timeout 10000000 (takeMVar handlesReady)
+    handles <- case mbHandles of
+      Nothing -> assertFailure "first consumer did not receive both partitions" >> pure []
+      Just value -> pure value
+    revoked <- Async.withAsync secondConsumer $ \secondAsync -> do
+      mbRevoked <- timeout 10000000 (awaitRevokedPartition env.testTopic events)
+      revoked <- case mbRevoked of
+        Nothing -> assertFailure "second consumer did not trigger a revocation" >> pure (PartitionId (-1))
+        Just value -> pure value
+      case lookup revoked handles of
+        Nothing -> assertFailure $ "no retained handle for revoked partition " <> show revoked
+        Just lateFinalize -> lateFinalize AckOk
+      putMVar releaseFirstConsumer ()
+      firstResult <- Async.wait firstAsync
+      case firstResult of
+        Left (_cs, err) -> assertFailure $ "first reassignment consumer failed: " <> show err
+        Right () -> pure ()
+      Async.cancel secondAsync
+      pure revoked
+
+    replayed <- pollPayloads env 5
+    let expected = if revoked == PartitionId 0 then "rebalance-partition-0" else "rebalance-partition-1"
+    assertBool
+      ("expected revoked partition payload to remain recoverable, saw " <> show replayed)
+      (expected `elem` replayed)
+
 testHandlerExceptionRedelivery :: IO ()
 testHandlerExceptionRedelivery = withTestEnv $ \env -> do
   createTopic env
@@ -341,3 +511,40 @@
 
 countPayload :: ByteString -> [ByteString] -> Int
 countPayload target = length . filter (== target)
+
+finalizeIngested :: Ingested es payload -> AckDecision -> Eff es ()
+finalizeIngested Ingested {ack = AckHandle finalize} = finalize
+
+pollPayloads :: TestEnv -> Int -> IO [ByteString]
+pollPayloads env pollCount = do
+  result <- runEff . runError @KafkaError $ do
+    let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore
+        sub = topics [env.testTopic] <> offsetReset Earliest
+    runKafkaConsumer props sub $ do
+      batches <- forM [1 .. pollCount] $ \_ ->
+        pollMessageBatch (Timeout 500) (BatchSize 100)
+      pure [payload | Right record <- concat batches, Just payload <- [record.crValue]]
+  case result of
+    Left (_cs, err) -> assertFailure ("poll verification failed: " <> show err) >> pure []
+    Right payloads -> pure payloads
+
+callbackHandle ::
+  (forall a. Eff es a -> IO a) ->
+  Ingested es payload ->
+  (PartitionId, AckDecision -> IO ())
+callbackHandle runInIO Ingested {envelope = Envelope {partition}, ack = AckHandle finalize} =
+  case partition of
+    Just partitionText -> (PartitionId (read (Text.unpack partitionText)), runInIO . finalize)
+    Nothing -> error "Kafka delivery did not carry a partition"
+
+awaitRevokedPartition :: TopicName -> TChan RebalanceEvent -> IO PartitionId
+awaitRevokedPartition topic events = atomically loop
+  where
+    loop = do
+      event <- readTChan events
+      case event of
+        RebalanceRevoke revoked ->
+          case [partition | (topic', partition) <- revoked, topic' == topic] of
+            partition : _ -> pure partition
+            [] -> loop
+        _ -> loop