diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,52 @@
 # Changelog
 
+## 0.10.0.0 — 2026-09-21
+
+### Breaking Changes
+
+- `ShutdownConfig` gains `totalShutdownTimeout`; existing record construction must choose an
+  overall shutdown deadline. The default keeps the 30-second drain timeout and adds a
+  60-second bound covering adapter shutdown and graceful drain before forced supervisor
+  stop begins.
+- `ConfigError` gains `DuplicateProcessorId`; `PolicyError` gains `InvalidConcurrency` and
+  `ConcurrencyCapacityOverflow`. Exhaustive matches must handle the new constructors.
+- Permanent framework-owned finalization failure now throws `ProcessorFailure` with the
+  failed message identity instead of being caught as graceful `ProcessorHalt`.
+- `ProcessorState.Processing` gains a sampled last-progress timestamp in addition to its
+  burst-start timestamp. Exhaustive constructor matches must accept the third field; its
+  JSON object gains the additive `lastProgress` field.
+
+### Bug Fixes
+
+- Reject duplicate processor IDs and invalid or overflow-prone concurrency before acquiring
+  resources.
+- Wake idle Serial, Ahead, Async, partitioned, and batch intake when a handler halts or
+  finalization fails.
+- Stop keyed input immediately on worker failure and close the worker start-gate ownership
+  gap.
+- Make startup and shutdown exception safe: all adapter shutdowns are attempted, the master
+  is always stopped, a blocking adapter is bounded by the total deadline, and repeated or
+  concurrent stop calls invoke adapters once.
+- Retain a bounded internal terminal lifecycle snapshot after live metrics unregister so
+  ignored failures remain observable with processor and message identity.
+- Observe batch ticker failure from the consuming stream instead of allowing an unmonitored
+  ticker death to leave the batcher blocked.
+- Reset active state on the final completion, restamp every observed burst, floor duplicate
+  completions at zero in-flight, and track progress during metrics sampling without adding a
+  clock read to the handler hot path.
+
+### Other Changes
+
+- Document the accepted batching resource boundary: `inboxSize` does not cap distinct
+  in-progress batch keys, so callers must bound externally controlled key cardinality. The
+  release evidence observed a conservative upper envelope of 703 bytes per additional key
+  over 1,000 to 50,000 keys; this finite measurement is not an implementation-enforced
+  production limit.
+- Depend directly on `effectful-core`, which provides every Effectful module used by the core,
+  tests, examples, and benchmarks. The accepted range keeps the 2.6 family and 2.7.1.1 or
+  later while excluding 2.7.0.0 through 2.7.1.0, whose upstream changelog records a
+  per-operation performance regression for dynamically dispatched effects.
+
 ## 0.9.0.3 — 2026-09-20
 
 ### Bug Fixes
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.9.0.3
+version: 0.10.0.0
 synopsis: Supervised queue processing framework for Haskell
 description:
   A supervised queue processing framework inspired by Broadway (Elixir).
@@ -67,12 +67,13 @@
 
   build-depends:
     aeson ^>=2.2,
+    async ^>=2.2,
     atomic-primops ^>=0.8.8,
     base ^>=4.21.0.0,
     bytestring ^>=0.12.2.0,
     containers ^>=0.7,
     deepseq ^>=1.5,
-    effectful >=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-propagator-w3c ^>=1.0,
     hs-opentelemetry-semantic-conventions ^>=1.40,
@@ -137,7 +138,7 @@
     base ^>=4.21.0.0,
     bytestring,
     containers,
-    effectful,
+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),
     hs-opentelemetry-api,
     hs-opentelemetry-exporter-in-memory ^>=1.0,
     hspec ^>=2.11,
@@ -166,7 +167,7 @@
 
   build-depends:
     base ^>=4.21.0.0,
-    effectful,
+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),
     shibuya-core,
     streamly-core,
     unliftio,
@@ -184,7 +185,7 @@
 
   build-depends:
     base ^>=4.21.0.0,
-    effectful,
+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),
     nqe,
     shibuya-core,
     streamly-core,
diff --git a/src/Shibuya.hs b/src/Shibuya.hs
--- a/src/Shibuya.hs
+++ b/src/Shibuya.hs
@@ -42,6 +42,7 @@
     renderDeadLetterReason,
     HaltReason (..),
     ProcessorHalt (..),
+    ProcessorFailure (..),
 
     -- * Batch processing
     BatchHandler,
@@ -148,6 +149,6 @@
 import Shibuya.Core.Retry
 import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), Headers, MessageId (..), TraceHeaders, mkEnvelope)
 import Shibuya.Handler (Handler)
-import Shibuya.Internal.Runner.Halt (ProcessorHalt (..))
+import Shibuya.Internal.Runner.Halt (ProcessorFailure (..), ProcessorHalt (..))
 import Shibuya.Policy (Concurrency (..), OrderingPolicy (..), validatePolicy)
 import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop)
diff --git a/src/Shibuya/App.hs b/src/Shibuya/App.hs
--- a/src/Shibuya/App.hs
+++ b/src/Shibuya/App.hs
@@ -41,14 +41,28 @@
 where
 
 import Control.Concurrent.NQE.Supervisor qualified as NQE
-import Control.Concurrent.STM (STM, atomically, check, orElse, readTVar, registerDelay)
-import Control.Monad (forM_, void)
+import Control.Concurrent.STM
+  ( STM,
+    atomically,
+    check,
+    newEmptyTMVarIO,
+    newTVarIO,
+    orElse,
+    putTMVar,
+    readTMVar,
+    readTVar,
+    registerDelay,
+    writeTVar,
+  )
+import Control.Monad (forM, forM_, void)
 import Data.Bifunctor (first)
 import Data.Foldable (traverse_)
 import Data.Map.Strict qualified as Map
+import Data.Set qualified as Set
 import Data.Text qualified as Text
 import Data.Time.Clock (NominalDiffTime)
-import Effectful (Eff, IOE, liftIO, (:>))
+import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, withEffToIO, (:>))
+import Effectful.Exception qualified as Exception
 import GHC.Generics (Generic)
 import Numeric.Natural (Natural)
 import Shibuya.Adapter (Adapter (..))
@@ -59,13 +73,23 @@
     ProcessorId (..),
     ProcessorMetrics (..),
   )
-import Shibuya.Internal.App (AppHandle (..), QueueProcessor (..), mkBatchProcessor, mkProcessor)
+import Shibuya.Internal.App
+  ( AppHandle (..),
+    OwnershipFailure (..),
+    QueueProcessor (..),
+    acquireOwned,
+    mkBatchProcessor,
+    mkProcessor,
+  )
 import Shibuya.Internal.Runner.Master
   ( Master,
     getAllMetrics,
     getAllMetricsIO,
     getProcessorMetrics,
     getProcessorMetricsIO,
+    markMasterDraining,
+    markMasterRunning,
+    markProcessorDraining,
     startMaster,
     stopMaster,
   )
@@ -76,7 +100,8 @@
   )
 import Shibuya.Policy (Concurrency (..), OrderingPolicy (..), validatePolicy)
 import Shibuya.Telemetry.Effect (Tracing)
-import UnliftIO (SomeException, catch, displayException, try)
+import UnliftIO (SomeException, displayException)
+import UnliftIO qualified as UIO
 
 --------------------------------------------------------------------------------
 -- Supervision Strategy
@@ -112,13 +137,18 @@
   { -- | Maximum time to wait for in-flight messages to drain.
     -- After this timeout, remaining processors are forcefully stopped.
     -- Default: 30 seconds.
-    drainTimeout :: !NominalDiffTime
+    drainTimeout :: !NominalDiffTime,
+    -- | Maximum time for adapter shutdown plus graceful draining before
+    -- forced supervisor stop begins.
+    -- Default: 60 seconds.
+    totalShutdownTimeout :: !NominalDiffTime
   }
   deriving stock (Eq, Show, Generic)
 
--- | Default shutdown configuration with 30 second drain timeout.
+-- | Default shutdown configuration with a 30 second drain timeout and a
+-- 60 second bound on the graceful shutdown phase.
 defaultShutdownConfig :: ShutdownConfig
-defaultShutdownConfig = ShutdownConfig {drainTimeout = 30}
+defaultShutdownConfig = ShutdownConfig {drainTimeout = 30, totalShutdownTimeout = 60}
 
 --------------------------------------------------------------------------------
 -- Errors
@@ -174,36 +204,68 @@
   Eff es (Either AppError (AppHandle es))
 runApp config namedProcessors =
   -- Validate all policies (and batch configs) first
-  case validateAppConfig config *> validateAllPolicies namedProcessors of
+  case validateAppConfig config *> validateUniqueProcessorIds namedProcessors *> validateAllPolicies namedProcessors of
     Left err -> pure $ Left err
     Right () -> do
       let nqeStrategy = toNQEStrategy config.strategy
-      catch
-        ( do
-            master <- startMaster nqeStrategy
-            spawnResult <- try $ spawnProcessors master (fromIntegral config.inboxSize) namedProcessors
-            case spawnResult of
-              Left (e :: SomeException) -> do
-                stopMaster master
-                pure $ Left $ AppRuntimeError $ SupervisorFailed $ Text.pack $ displayException e
-              Right processors ->
-                pure $
-                  Right
-                    AppHandle
-                      { master = master,
-                        processors = Map.fromList processors
-                      }
-        )
-        ( \(e :: SomeException) ->
-            pure $ Left $ AppRuntimeError $ SupervisorFailed $ Text.pack $ displayException e
-        )
+      startupResult <-
+        acquireOwned
+          (startMaster nqeStrategy)
+          stopMaster
+          ( \master -> do
+              processors <- spawnProcessors master (fromIntegral config.inboxSize) namedProcessors
+              markMasterRunning master
+              shutdownStarted <- liftIO $ newTVarIO False
+              shutdownResult <- liftIO newEmptyTMVarIO
+              pure
+                AppHandle
+                  { master = master,
+                    processors = Map.fromList processors,
+                    shutdownStarted,
+                    shutdownResult
+                  }
+          )
+      case startupResult of
+        Right appHandle -> pure $ Right appHandle
+        Left (OwnerAcquisitionFailed failure) -> startupFailure failure
+        Left (OwnedActionFailed failure cleanupFailure) ->
+          startupFailureWithCleanup failure cleanupFailure
+  where
+    startupFailure failure
+      | Exception.isAsyncException failure = Exception.throwIO failure
+      | otherwise =
+          pure $ Left $ AppRuntimeError $ SupervisorFailed $ Text.pack $ displayException failure
 
+    startupFailureWithCleanup failure cleanupFailure
+      | Exception.isAsyncException failure = Exception.throwIO failure
+      | otherwise =
+          let cleanupSuffix = case cleanupFailure of
+                Nothing -> ""
+                Just cleanupException ->
+                  "; supervisor cleanup also failed: " <> Text.pack (displayException cleanupException)
+           in pure $
+                Left $
+                  AppRuntimeError $
+                    SupervisorFailed $
+                      Text.pack (displayException failure) <> cleanupSuffix
+
 -- | Validate app configuration before starting any processor.
 validateAppConfig :: AppConfig -> Either AppError ()
 validateAppConfig config
   | config.inboxSize < 1 = Left $ AppConfigInvalid $ InvalidInboxSize config.inboxSize
   | otherwise = Right ()
 
+-- | Reject duplicate processor identifiers before the master or any adapter is
+-- acquired. Keeping this separate from handle construction prevents a live
+-- processor from being silently discarded by 'Map.fromList'.
+validateUniqueProcessorIds :: [(ProcessorId, QueueProcessor es)] -> Either AppError ()
+validateUniqueProcessorIds = go Set.empty
+  where
+    go _ [] = Right ()
+    go seen ((pid, _) : rest)
+      | pid `Set.member` seen = Left $ AppConfigInvalid $ DuplicateProcessorId pid
+      | otherwise = go (Set.insert pid seen) rest
+
 -- | Validate all processor policies (and batch configs) before starting.
 validateAllPolicies :: [(ProcessorId, QueueProcessor es)] -> Either AppError ()
 validateAllPolicies = traverse_ validateOne
@@ -267,12 +329,12 @@
 getAppMaster appHandle = appHandle.master
 
 -- | Gracefully stop all processors with default configuration.
--- Uses 'defaultShutdownConfig' (30 second drain timeout).
+-- Uses 'defaultShutdownConfig' (30 second drain timeout, 60 second graceful-phase bound).
 -- For custom timeout, use 'stopAppGracefully'.
 stopApp :: (IOE :> es) => AppHandle es -> Eff es ()
 stopApp = void . stopAppGracefully defaultShutdownConfig
 
--- | Gracefully stop all processors with configurable drain timeout.
+-- | Gracefully stop all processors with configurable drain and total timeout.
 --
 -- Shutdown sequence:
 -- 1. Signal all adapters to stop producing (close source streams)
@@ -282,25 +344,65 @@
 --
 -- Returns whether all processors drained cleanly (True) or were forced (False).
 stopAppGracefully :: (IOE :> es) => ShutdownConfig -> AppHandle es -> Eff es Bool
-stopAppGracefully config appHandle = do
-  -- 1. Signal adapters to stop producing
-  mapM_ shutdownAdapter (Map.elems appHandle.processors)
+stopAppGracefully config appHandle =
+  Exception.mask $ \restore -> do
+    isLeader <-
+      liftIO $
+        atomically $ do
+          started <- readTVar appHandle.shutdownStarted
+          if started
+            then pure False
+            else writeTVar appHandle.shutdownStarted True >> pure True
+    if isLeader
+      then do
+        result <- Exception.try @SomeException (restore performShutdown)
+        liftIO $ atomically $ putTMVar appHandle.shutdownResult result
+        either Exception.throwIO pure result
+      else do
+        result <- restore $ liftIO $ atomically $ readTMVar appHandle.shutdownResult
+        either Exception.throwIO pure result
+  where
+    performShutdown = Exception.mask $ \restore -> do
+      let totalTimeoutMicros = nominalToMicros config.totalShutdownTimeout
+      outcome <-
+        Exception.try $
+          restore $
+            withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
+              UIO.timeout totalTimeoutMicros (runInIO shutdownAndDrain)
 
-  -- 2. Wait for drain with timeout
-  let timeoutMicros = floor (config.drainTimeout * 1_000_000)
-  drained <- liftIO $ waitForDrainWithTimeout timeoutMicros (Map.elems appHandle.processors)
+      -- Master cleanup is unconditional: adapter failures, external cancellation,
+      -- drain cancellation, and the total deadline all converge here.
+      stopOutcome <- Exception.try @SomeException (stopMaster appHandle.master)
+      case outcome of
+        Left (primaryFailure :: SomeException) -> Exception.throwIO primaryFailure
+        Right Nothing -> finishStopOutcome stopOutcome False
+        Right (Just drained) -> finishStopOutcome stopOutcome drained
 
-  -- 3. Log warning if forced shutdown (caller can check return value)
-  -- Note: We don't log here to avoid IO dependencies, caller can log if needed
+    shutdownAndDrain = do
+      markMasterDraining appHandle.master
+      forM_ (Map.keys appHandle.processors) $
+        markProcessorDraining appHandle.master
 
-  -- 4. Stop master (cancels any remaining processors)
-  stopMaster appHandle.master
+      -- Catch only synchronous adapter failures. External cancellation and the
+      -- total timeout must abort this phase so the master is force-stopped.
+      shutdownResults <-
+        forM (Map.elems appHandle.processors) $ \processor ->
+          Exception.trySync (shutdownAdapter processor)
 
-  pure drained
-  where
+      case [failure | Left failure <- shutdownResults] of
+        firstFailure : _ -> Exception.throwIO firstFailure
+        [] -> do
+          let drainTimeoutMicros = nominalToMicros config.drainTimeout
+          liftIO $ waitForDrainWithTimeout drainTimeoutMicros (Map.elems appHandle.processors)
+
     shutdownAdapter (_, qp) = case qp of
       QueueProcessor {adapter} -> adapter.shutdown
       BatchingProcessor {adapter} -> adapter.shutdown
+
+    finishStopOutcome (Left stopFailure) _ = Exception.throwIO stopFailure
+    finishStopOutcome (Right ()) drained = pure drained
+
+    nominalToMicros timeout = max 0 (floor (timeout * 1_000_000))
 
 -- | Wait for all processors to be done, with timeout.
 -- Returns True if all drained cleanly, False if timeout occurred.
diff --git a/src/Shibuya/Core/Error.hs b/src/Shibuya/Core/Error.hs
--- a/src/Shibuya/Core/Error.hs
+++ b/src/Shibuya/Core/Error.hs
@@ -20,17 +20,26 @@
 where
 
 import Data.Text qualified as Text
+import Shibuya.Core.Metrics (ProcessorId (..))
 import Shibuya.Prelude
 
 -- | Policy validation errors.
 data PolicyError
   = -- | Invalid combination of ordering and concurrency
     InvalidPolicyCombo !Text
+  | -- | Ahead and Async concurrency must be positive.
+    InvalidConcurrency !Int
+  | -- | A derived concurrency buffer would overflow 'Int'.
+    ConcurrencyCapacityOverflow !Int
   deriving stock (Eq, Show, Generic)
 
 -- | Convert policy error to text for display.
 policyErrorToText :: PolicyError -> Text
 policyErrorToText (InvalidPolicyCombo msg) = msg
+policyErrorToText (InvalidConcurrency n) =
+  "concurrency must be >= 1, got " <> Text.pack (show n)
+policyErrorToText (ConcurrencyCapacityOverflow n) =
+  "concurrency is too large to derive a bounded buffer safely, got " <> Text.pack (show n)
 
 -- | Handler execution errors.
 data HandlerError
@@ -56,9 +65,13 @@
 data ConfigError
   = -- | inboxSize must be >= 1; 0 stalls ingestion, negatives are nonsense.
     InvalidInboxSize !Int
+  | -- | Processor identifiers must be unique within one application.
+    DuplicateProcessorId !ProcessorId
   deriving stock (Eq, Show, Generic)
 
 -- | Convert configuration error to text for display.
 configErrorToText :: ConfigError -> Text
 configErrorToText (InvalidInboxSize n) =
   "inboxSize must be >= 1, got " <> Text.pack (show n)
+configErrorToText (DuplicateProcessorId (ProcessorId pid)) =
+  "processor IDs must be unique, duplicate: " <> pid
diff --git a/src/Shibuya/Core/Metrics.hs b/src/Shibuya/Core/Metrics.hs
--- a/src/Shibuya/Core/Metrics.hs
+++ b/src/Shibuya/Core/Metrics.hs
@@ -29,6 +29,7 @@
     HotCounters (..),
     MetricsHandle (..),
     newMetricsHandle,
+    newMetricsHandleWithClock,
     sampleMetrics,
     incrementReceived,
     beginProcessing,
@@ -49,14 +50,18 @@
 where
 
 import Control.Concurrent.STM (TVar, atomically, modifyTVar', newTVarIO, readTVarIO)
-import Control.Monad (unless, void, when)
-import Data.Aeson (FromJSON (..), FromJSONKey (..), ToJSON (..), ToJSONKey (..), object, withObject, (.:))
+import Control.Monad (void, when)
+import Data.Aeson (FromJSON (..), FromJSONKey (..), ToJSON (..), ToJSONKey (..), object, withObject, (.:), (.:?))
 import Data.Aeson qualified as Aeson
 import Data.Atomics.Counter (AtomicCounter, incrCounter, readCounter)
 import Data.Atomics.Counter qualified as Counter
-import Data.IORef (IORef, newIORef, readIORef, writeIORef)
+import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef, writeIORef)
 import Data.Map.Strict (Map)
+import Data.Maybe (fromMaybe)
 import Data.Text qualified as Text
+import Data.Time.Clock (addUTCTime)
+import Data.Word (Word64)
+import GHC.Clock (getMonotonicTimeNSec)
 import Shibuya.Prelude
 
 -- | Processor identifier.
@@ -94,8 +99,8 @@
 data ProcessorState
   = -- | Waiting for messages
     Idle
-  | -- | Currently processing (in-flight info, last activity time)
-    Processing !InFlightInfo !UTCTime
+  | -- | Currently processing (in-flight info, sampled burst start, sampled last progress)
+    Processing !InFlightInfo !UTCTime !UTCTime
   | -- | Failed with error (error message, timestamp)
     Failed !Text !UTCTime
   | -- | Processor has been stopped
@@ -104,12 +109,13 @@
 
 instance ToJSON ProcessorState where
   toJSON Idle = object ["status" Aeson..= ("idle" :: Text)]
-  toJSON (Processing info lastActivity) =
+  toJSON (Processing info lastActivity lastProgress) =
     object
       [ "status" Aeson..= ("processing" :: Text),
         "inFlight" Aeson..= info.inFlight,
         "maxConcurrency" Aeson..= info.maxConcurrency,
-        "lastActivity" Aeson..= lastActivity
+        "lastActivity" Aeson..= lastActivity,
+        "lastProgress" Aeson..= lastProgress
       ]
   toJSON (Failed err timestamp) =
     object
@@ -128,7 +134,8 @@
         inFlightCount <- v .: "inFlight"
         maxConc <- v .: "maxConcurrency"
         lastActivity <- v .: "lastActivity"
-        pure $ Processing (InFlightInfo inFlightCount maxConc) lastActivity
+        lastProgress <- v .:? "lastProgress"
+        pure $ Processing (InFlightInfo inFlightCount maxConc) lastActivity (fromMaybe lastActivity lastProgress)
       "failed" -> Failed <$> v .: "error" <*> v .: "timestamp"
       "stopped" -> pure Stopped
       other -> fail $ "Unknown processor state: " <> Text.unpack other
@@ -219,18 +226,32 @@
   { hot :: !HotCounters,
     maxConcurrencyRef :: !(IORef Int),
     burstStartedRef :: !(IORef UTCTime),
+    lastProgressRef :: !(IORef Word64),
+    progressOriginNs :: !Word64,
+    progressOriginTime :: !UTCTime,
+    progressClock :: !(IO Word64),
+    lastObservedProgress :: !(IORef (Int, Int, Int)),
     stateActiveRef :: !(IORef Bool),
     cold :: !(TVar ProcessorMetrics)
   }
 
 newMetricsHandle :: UTCTime -> IO MetricsHandle
-newMetricsHandle now = do
+newMetricsHandle = newMetricsHandleWithClock getMonotonicTimeNSec
+
+-- | Create a metrics handle with an injectable monotonic clock. The custom
+-- clock is intended for deterministic tests; production callers should use
+-- 'newMetricsHandle'.
+newMetricsHandleWithClock :: IO Word64 -> UTCTime -> IO MetricsHandle
+newMetricsHandleWithClock clock now = do
+  progressOriginNs <- clock
   received <- Counter.newCounter 0
   processed <- Counter.newCounter 0
   failed <- Counter.newCounter 0
   inFlight <- Counter.newCounter 0
   maxConcurrencyRef <- newIORef 1
   burstStartedRef <- newIORef now
+  lastProgressRef <- newIORef progressOriginNs
+  lastObservedProgress <- newIORef (0, 0, 0)
   stateActiveRef <- newIORef False
   cold <- newTVarIO (emptyProcessorMetrics now)
   pure
@@ -244,6 +265,11 @@
             },
         maxConcurrencyRef = maxConcurrencyRef,
         burstStartedRef = burstStartedRef,
+        lastProgressRef = lastProgressRef,
+        progressOriginNs = progressOriginNs,
+        progressOriginTime = now,
+        progressClock = clock,
+        lastObservedProgress = lastObservedProgress,
         stateActiveRef = stateActiveRef,
         cold = cold
       }
@@ -256,7 +282,10 @@
   failed <- readCounter handle.hot.failed
   inFlight <- readCounter handle.hot.inFlight
   maxConcurrency <- readIORef handle.maxConcurrencyRef
+  observeProgress handle processed failed inFlight
   burstStartedAt <- readIORef handle.burstStartedRef
+  lastProgressNs <- readIORef handle.lastProgressRef
+  let lastProgressAt = monotonicToUTC handle lastProgressNs
   let sampledStats =
         StreamStats
           { received = received,
@@ -266,7 +295,7 @@
       sampledState = case coldSnapshot.state of
         Failed err timestamp -> Failed err timestamp
         Stopped -> Stopped
-        _ | inFlight > 0 -> Processing (InFlightInfo inFlight maxConcurrency) burstStartedAt
+        _ | inFlight > 0 -> Processing (InFlightInfo inFlight maxConcurrency) burstStartedAt lastProgressAt
         _ -> Idle
   pure coldSnapshot {state = sampledState, stats = sampledStats}
 
@@ -278,15 +307,8 @@
 beginProcessing handle maxConcurrency = do
   currentInflight <- incrCounter 1 handle.hot.inFlight
   when (currentInflight == 1) $ do
-    stateActive <- readIORef handle.stateActiveRef
-    unless stateActive $ do
-      now <- getCurrentTime
-      writeIORef handle.maxConcurrencyRef maxConcurrency
-      writeIORef handle.burstStartedRef now
-      writeIORef handle.stateActiveRef True
-      atomically $
-        modifyTVar' handle.cold $ \m ->
-          m {state = Processing (InFlightInfo currentInflight maxConcurrency) now}
+    writeIORef handle.maxConcurrencyRef maxConcurrency
+    writeIORef handle.stateActiveRef True
   pure currentInflight
 
 finishProcessing :: MetricsHandle -> Either Text AckDecisionMetric -> IO ()
@@ -297,7 +319,8 @@
     Right CountNeither -> pure ()
     Right (CountHalt _) -> pure ()
     Left _ -> void $ incrCounter 1 handle.hot.failed
-  void $ incrCounter (-1) handle.hot.inFlight
+  remaining <- decrementCounterFloorZero handle.hot.inFlight
+  when (remaining == 0) $ writeIORef handle.stateActiveRef False
   case result of
     Left failureText -> setFailed failureText
     Right (CountHalt reasonText) -> setFailed reasonText
@@ -349,7 +372,7 @@
   when (failedDelta /= 0) $
     void $
       incrCounter failedDelta handle.hot.failed
-  void $ incrCounter (-1) handle.hot.inFlight
+  remaining <- decrementCounterFloorZero handle.hot.inFlight
   now <- getCurrentTime
   atomically $
     modifyTVar' handle.cold $ \m ->
@@ -368,7 +391,42 @@
        in m {state = newState, batch = newBatch}
   case firstHalt of
     Just _ -> writeIORef handle.stateActiveRef False
-    Nothing -> pure ()
+    Nothing -> when (remaining == 0) $ writeIORef handle.stateActiveRef False
+
+observeProgress :: MetricsHandle -> Int -> Int -> Int -> IO ()
+observeProgress handle processed failed inFlight = do
+  let current = (processed, failed, inFlight)
+  previous@(_, _, previousInFlight) <- readIORef handle.lastObservedProgress
+  when (current /= previous) $ do
+    progressNs <- handle.progressClock
+    atomicWriteIORef handle.lastProgressRef progressNs
+    atomicWriteIORef handle.lastObservedProgress current
+    -- Burst boundaries are sampled observations: an observed idle-to-active
+    -- transition restamps lastActivity, while a continuously active processor
+    -- only advances lastProgress. An idle gap wholly between samples cannot
+    -- affect stuck detection because changed counters still advance progress.
+    when (inFlight > 0 && previousInFlight == 0) $
+      atomicWriteIORef handle.burstStartedRef (monotonicToUTC handle progressNs)
+
+monotonicToUTC :: MetricsHandle -> Word64 -> UTCTime
+monotonicToUTC handle progressNs =
+  let elapsedNs
+        | progressNs >= handle.progressOriginNs = progressNs - handle.progressOriginNs
+        | otherwise = 0
+      elapsedSeconds = fromIntegral elapsedNs / 1_000_000_000
+   in addUTCTime elapsedSeconds handle.progressOriginTime
+
+decrementCounterFloorZero :: AtomicCounter -> IO Int
+decrementCounterFloorZero counter = do
+  remaining <- incrCounter (-1) counter
+  if remaining >= 0
+    then pure remaining
+    else do
+      -- Underflow means a caller completed work it did not begin. Repair the
+      -- decrement so externally sampled in-flight never remains below zero;
+      -- the valid hot path stays one fetch-and-add operation.
+      Counter.incrCounter_ 1 counter
+      pure 0
 
 data BatchTriggerMetric
   = CountTriggerSize
diff --git a/src/Shibuya/Internal/App.hs b/src/Shibuya/Internal/App.hs
--- a/src/Shibuya/Internal/App.hs
+++ b/src/Shibuya/Internal/App.hs
@@ -6,10 +6,15 @@
     mkProcessor,
     mkBatchProcessor,
     AppHandle (..),
+    OwnershipFailure (..),
+    acquireOwned,
   )
 where
 
+import Control.Concurrent.STM (TMVar, TVar)
 import Data.Map.Strict (Map)
+import Effectful (Eff, IOE, (:>))
+import Effectful.Exception qualified as Exception
 import Shibuya.Adapter (Adapter (..))
 import Shibuya.Batch (BatchConfig, BatchHandler)
 import Shibuya.Core.Metrics (ProcessorId (..))
@@ -17,7 +22,43 @@
 import Shibuya.Internal.Runner.Master (Master)
 import Shibuya.Internal.Runner.Supervised (SupervisedProcessor)
 import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))
+import UnliftIO (SomeException)
 
+-- | Failure while acquiring an owner or while transferring resources into an
+-- acquired owner's custody. Kept in this internal module so tests can inject a
+-- deterministic cancellation barrier into the exact primitive used by
+-- 'Shibuya.App.runApp'.
+data OwnershipFailure
+  = OwnerAcquisitionFailed !SomeException
+  | OwnedActionFailed !SomeException !(Maybe SomeException)
+  deriving stock (Show)
+
+-- | Acquire an owner under masking, restore interruptibility while acquiring
+-- the resources it will own, and clean the owner up before reporting any
+-- synchronous or asynchronous failure from that action.
+acquireOwned ::
+  (IOE :> es) =>
+  Eff es owner ->
+  (owner -> Eff es ()) ->
+  (owner -> Eff es value) ->
+  Eff es (Either OwnershipFailure value)
+acquireOwned acquireOwner releaseOwner acquireResources =
+  Exception.mask $ \restore -> do
+    ownerResult <- Exception.try @SomeException acquireOwner
+    case ownerResult of
+      Left failure -> pure $ Left $ OwnerAcquisitionFailed failure
+      Right owner -> do
+        resourceResult <- Exception.try @SomeException $ restore $ acquireResources owner
+        case resourceResult of
+          Right value -> pure $ Right value
+          Left failure -> do
+            cleanupResult <- Exception.try @SomeException $ releaseOwner owner
+            pure $
+              Left $
+                OwnedActionFailed
+                  failure
+                  (either Just (const Nothing) cleanupResult)
+
 -- | A queue processor pairs an adapter with a handler. The message type is
 -- existentially hidden, allowing heterogeneous queues in one @runApp@ call.
 --
@@ -58,5 +99,9 @@
   { -- | The master coordinator
     master :: !Master,
     -- | Map of processor IDs to their handles
-    processors :: !(Map ProcessorId (SupervisedProcessor, QueueProcessor es))
+    processors :: !(Map ProcessorId (SupervisedProcessor, QueueProcessor es)),
+    -- | Coordinates repeated and concurrent graceful-stop calls. The first
+    -- caller performs shutdown; every caller observes the same terminal result.
+    shutdownStarted :: !(TVar Bool),
+    shutdownResult :: !(TMVar (Either SomeException Bool))
   }
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
@@ -35,13 +35,13 @@
 import Control.Applicative ((<|>))
 import Data.Foldable (for_, traverse_)
 import Data.HashMap.Strict qualified as HashMap
-import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef)
 import Data.List.NonEmpty (NonEmpty)
 import Data.List.NonEmpty qualified as NE
 import Data.Map.Strict qualified as Map
 import Data.Maybe (fromMaybe, isJust, listToMaybe)
 import Data.Text qualified as Text
 import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, withEffToIO, (:>))
+import GHC.IO (unsafeUnmask)
 import OpenTelemetry.Attributes (toAttribute)
 import OpenTelemetry.Trace.Core qualified as OTel
 import Shibuya.Batch
@@ -71,7 +71,16 @@
   )
 import Shibuya.Core.Types (Envelope (..))
 import Shibuya.Internal.Runner.Finalize (finalizeWithRetry)
-import Shibuya.Internal.Runner.Halt (ProcessorHalt (..))
+import Shibuya.Internal.Runner.Halt
+  ( ProcessorExit (..),
+    ProcessorExitPublisher,
+    ProcessorSignal,
+    newProcessorExitPublisher,
+    newProcessorSignal,
+    readProcessorExit,
+    requestProcessorExit,
+    throwProcessorExit,
+  )
 import Shibuya.Internal.Runner.KeyedScheduler (runKeyedScheduler)
 import Shibuya.Policy (Concurrency (..))
 import Shibuya.Prelude
@@ -104,7 +113,7 @@
 import Streamly.Data.Fold qualified as Fold
 import Streamly.Data.Stream (Stream)
 import Streamly.Data.Stream qualified as Stream
-import UnliftIO (catchAny, throwIO)
+import UnliftIO (catchAny)
 
 -- | Execute one emitted batch and finalize every retained message resiliently.
 --
@@ -116,11 +125,12 @@
   MetricsHandle ->
   ProcessorId ->
   Int ->
-  IORef (Maybe HaltReason) ->
+  ProcessorSignal ->
+  ProcessorExitPublisher ->
   BatchHandler es msg ->
   (BatchInfo, NonEmpty (Ingested es msg)) ->
   Eff es ()
-processOneBatch metricsHandle procId maxConc haltRef handler (info, batch) = do
+processOneBatch 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).
@@ -150,12 +160,12 @@
 
       addEvent traceSpan (mkEvent eventBatchStarted [])
 
-      alreadyHalted <- liftIO $ readIORef haltRef
+      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 alreadyHalted of
+        case terminal of
           Just _ -> pure (Left (), True)
           Nothing -> do
             result <-
@@ -188,15 +198,18 @@
           finalizeFailures = [(mid, ex) | (mid, _, _, Left ex) <- results]
 
       -- Compute halt and partial-failure signals from the resolved decisions.
-      let finalizationHalt =
+      let finalizationFailure =
             case finalizeFailures of
               [] -> Nothing
-              failed ->
+              failed@((firstMessageId, _) : _) ->
                 Just $
-                  HaltFatal $
-                    "batch finalization failed for message ids: "
-                      <> Text.intercalate ", " [tshow mid | (mid, _) <- failed]
-          firstHalt = finalizationHalt <|> listToMaybe [r | AckHalt r <- decisions]
+                  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,
@@ -207,8 +220,8 @@
 
       -- Span status: error on halt or exception, otherwise Ok.
       addEvent traceSpan (mkEvent eventBatchCompleted [])
-      case firstHalt of
-        Just reason -> setStatus traceSpan (OTel.Error (haltReasonText reason))
+      case requestedExit of
+        Just processorExit -> setStatus traceSpan (OTel.Error (processorExitText processorExit))
         Nothing ->
           if skippedAfterHalt
             then setStatus traceSpan (OTel.Error "batch skipped after halt")
@@ -230,11 +243,11 @@
           handlerThrew
           partialInc
           (decisionMetric <$> decisions)
-          (haltReasonText <$> firstHalt)
+          (processorExitText <$> requestedExit)
 
       -- Halt: set the shared flag; do NOT throw (let the stream drain).
-      for_ firstHalt $ \reason ->
-        liftIO $ atomicWriteIORef haltRef (Just reason)
+      for_ requestedExit $ \processorExit ->
+        liftIO $ requestProcessorExit exitPublisher processorExit
   where
     isFailing :: AckDecision -> Bool
     isFailing (AckDeadLetter _) = True
@@ -261,6 +274,10 @@
 haltReasonText (HaltOrderedStream t) = t
 haltReasonText (HaltFatal t) = t
 
+processorExitText :: ProcessorExit -> Text
+processorExitText (ProcessorHalted reason) = haltReasonText reason
+processorExitText (ProcessorFailed failure _) = failure
+
 tshow :: (Show a) => a -> Text
 tshow = Text.pack . show
 
@@ -276,21 +293,28 @@
   Concurrency ->
   BatchHandler es msg ->
   Stream IO (BatchInfo, NonEmpty (Ingested es msg)) ->
-  IORef (Maybe HaltReason) ->
+  ProcessorSignal ->
+  ProcessorExitPublisher ->
   Eff es ()
-processBatchesUntilDrained metricsHandle procId concurrency handler batchesStream haltRef = do
+processBatchesUntilDrained metricsHandle procId concurrency handler batchesStream stopSignal exitPublisher = do
   let maxConc = case concurrency of
         Serial -> 1
         Ahead n -> n
         Async n -> n
 
   withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
-    let batchAction = runInIO . processOneBatch metricsHandle procId maxConc haltRef handler
+    -- The supervisor keeps framework coordination masked; only the owned batch
+    -- action is unmasked so user code and finalizers remain cancellable.
+    let runBatchAction batch =
+          runInIO $
+            processOneBatch metricsHandle procId maxConc stopSignal exitPublisher handler batch
+        batchAction = unsafeUnmask . runBatchAction
         pendingLimit = max 2 (2 * max 1 maxConc)
     case concurrency of
       Serial ->
-        Stream.fold Fold.drain $
-          Stream.mapM batchAction batchesStream
+        unsafeUnmask $
+          Stream.fold Fold.drain $
+            Stream.mapM runBatchAction batchesStream
       Ahead n ->
         runKeyedScheduler (max 1 n) pendingLimit (Just . fstBatchKey) batchAction batchesStream
       Async n ->
@@ -313,12 +337,13 @@
 runBatchesWithMetrics procId concurrency handler batches = do
   now <- liftIO getCurrentTime
   metricsHandle <- liftIO $ newMetricsHandle now
-  haltRef <- liftIO $ newIORef Nothing
+  stopSignal <- liftIO newProcessorSignal
+  let exitPublisher = newProcessorExitPublisher stopSignal
 
   let batchesStream = Stream.fromList batches
-  processBatchesUntilDrained metricsHandle procId concurrency handler batchesStream haltRef
+  processBatchesUntilDrained metricsHandle procId concurrency handler batchesStream stopSignal exitPublisher
 
-  maybeHalt <- liftIO $ readIORef haltRef
-  case maybeHalt of
-    Just reason -> throwIO (ProcessorHalt reason)
+  maybeExit <- liftIO $ readProcessorExit stopSignal
+  case maybeExit of
+    Just processorExit -> liftIO $ throwProcessorExit processorExit
     Nothing -> liftIO $ sampleMetrics metricsHandle
diff --git a/src/Shibuya/Internal/Runner/Batcher.hs b/src/Shibuya/Internal/Runner/Batcher.hs
--- a/src/Shibuya/Internal/Runner/Batcher.hs
+++ b/src/Shibuya/Internal/Runner/Batcher.hs
@@ -29,6 +29,7 @@
 
     -- * IO engine
     runBatcher,
+    runBatcherWithTickHook,
   )
 where
 
@@ -40,6 +41,7 @@
     readTVar,
     readTVarIO,
     retry,
+    throwSTM,
     writeTVar,
   )
 import Control.Monad (when)
@@ -60,7 +62,7 @@
 import Streamly.Data.Stream (Stream)
 import Streamly.Data.Stream qualified as Stream
 import UnliftIO (finally, throwIO)
-import UnliftIO.Async (Async, async, cancel, waitCatch)
+import UnliftIO.Async (Async, async, cancel, pollSTM, waitCatch)
 
 -- | In-progress state for one batch key.
 data Accum es msg = Accum
@@ -191,7 +193,18 @@
   BatchConfig es msg ->
   Stream IO (Ingested es msg) ->
   Stream IO (ReadyBatch es msg)
-runBatcher outputCapacity cfg input =
+runBatcher = runBatcherWithTickHook (pure ())
+
+-- | Test seam for deterministic ticker-failure injection. The hook runs once
+-- after each tick delay and before the clock/state step. Production uses a
+-- no-op hook through 'runBatcher'.
+runBatcherWithTickHook ::
+  IO () ->
+  Natural ->
+  BatchConfig es msg ->
+  Stream IO (Ingested es msg) ->
+  Stream IO (ReadyBatch es msg)
+runBatcherWithTickHook tickHook outputCapacity cfg input =
   Stream.bracketIO acquire release consume
   where
     tickMicros = nominalToMicros (fromMaybe cfg.batchTimeout cfg.tickInterval)
@@ -219,6 +232,7 @@
 
           tickerLoop = do
             threadDelay tickMicros
+            tickHook
             done <- readTVarIO doneVar
             if done
               then pure ()
@@ -235,31 +249,36 @@
       cancel tickerA
       cancel consumerA
 
-    consume (stateVar, doneVar, consumerA, _tickerA) = drainQueue consumerA stateVar doneVar
+    consume (stateVar, doneVar, consumerA, tickerA) = drainQueue consumerA tickerA stateVar doneVar
 
 -- | Stream finished batches out of the bounded queue, ending when the consumer
 -- has flushed everything (doneVar), the queue has drained, and the consumer
 -- async is known to have completed successfully.
 drainQueue ::
   Async () ->
+  Async () ->
   TVar (BatcherState es msg, Seq (ReadyBatch es msg)) ->
   TVar Bool ->
   Stream IO (ReadyBatch es msg)
-drainQueue consumerA stateVar doneVar = Stream.unfoldrM step ()
+drainQueue consumerA tickerA stateVar doneVar = Stream.unfoldrM step ()
   where
     step _ = do
       drainStep <-
         atomically $ do
-          (state, pending) <- readTVar stateVar
-          case Seq.viewl pending of
-            rb Seq.:< rest -> do
-              writeTVar stateVar (state, rest)
-              pure (DrainReady rb)
-            Seq.EmptyL -> do
-              done <- readTVar doneVar
-              if done
-                then pure DrainDone
-                else retry
+          tickerStatus <- pollSTM tickerA
+          case tickerStatus of
+            Just (Left tickerFailure) -> throwSTM tickerFailure
+            _ -> do
+              (state, pending) <- readTVar stateVar
+              case Seq.viewl pending of
+                rb Seq.:< rest -> do
+                  writeTVar stateVar (state, rest)
+                  pure (DrainReady rb)
+                Seq.EmptyL -> do
+                  done <- readTVar doneVar
+                  if done
+                    then pure DrainDone
+                    else retry
       case drainStep of
         DrainReady rb -> pure (Just (rb, ()))
         DrainDone ->
diff --git a/src/Shibuya/Internal/Runner/Halt.hs b/src/Shibuya/Internal/Runner/Halt.hs
--- a/src/Shibuya/Internal/Runner/Halt.hs
+++ b/src/Shibuya/Internal/Runner/Halt.hs
@@ -6,12 +6,26 @@
 -- Thrown when a handler returns AckHalt to stop processing.
 module Shibuya.Internal.Runner.Halt
   ( ProcessorHalt (..),
+    ProcessorFailure (..),
+    ProcessorExit (..),
+    ProcessorSignal,
+    ProcessorExitPublisher,
+    newProcessorSignal,
+    newProcessorExitPublisher,
+    newProcessorExitPublisherWithWake,
+    readProcessorExit,
+    requestProcessorExit,
+    throwProcessorExit,
   )
 where
 
-import Control.Exception (Exception)
+import Control.Concurrent.STM (TVar, atomically, writeTVar)
+import Control.Exception (Exception, mask_)
+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)
 import Shibuya.Core.Ack (HaltReason)
+import Shibuya.Core.Types (MessageId)
 import Shibuya.Prelude
+import UnliftIO (throwIO)
 
 -- | Exception thrown when processing should halt.
 -- The supervisor catches this to handle graceful shutdown.
@@ -21,3 +35,75 @@
   deriving stock (Show, Generic)
 
 instance Exception ProcessorHalt
+
+-- | An infrastructure failure that must remain distinguishable from a handler's
+-- deliberate 'AckHalt'. The optional message identity is retained when the
+-- failure occurred while finalizing a delivery.
+data ProcessorFailure = ProcessorFailure !Text !(Maybe MessageId)
+  deriving stock (Show, Generic)
+
+instance Exception ProcessorFailure
+
+-- | The first terminal request observed by a processor. Infrastructure failure
+-- takes precedence over a graceful halt if concurrent work reports both.
+data ProcessorExit
+  = ProcessorHalted !HaltReason
+  | ProcessorFailed !Text !(Maybe MessageId)
+  deriving stock (Eq, Show, Generic)
+
+-- | A cheap hot-path stop observation. The corresponding publisher is kept
+-- separate so message actions retain one opaque cold-path pointer rather than
+-- capturing and field-splitting the signal and its STM wake cell.
+newtype ProcessorSignal = ProcessorSignal
+  { terminalExit :: IORef (Maybe ProcessorExit)
+  }
+
+-- | Opaque cold-path terminal publisher. The function closure may retain both
+-- the signal and an intake wake cell, but hot message closures retain only this
+-- single pointer. The box prevents GHC from field-splitting those two captured
+-- cells back into every nested per-message closure.
+data ProcessorExitPublisher = ProcessorExitPublisher (ProcessorExit -> IO ())
+
+newProcessorSignal :: IO ProcessorSignal
+newProcessorSignal = ProcessorSignal <$> newIORef Nothing
+
+newProcessorExitPublisher :: ProcessorSignal -> ProcessorExitPublisher
+newProcessorExitPublisher signal =
+  ProcessorExitPublisher $ \requested ->
+    mask_ $ publishProcessorExit signal requested
+{-# OPAQUE newProcessorExitPublisher #-}
+
+newProcessorExitPublisherWithWake :: ProcessorSignal -> TVar Bool -> ProcessorExitPublisher
+newProcessorExitPublisherWithWake signal intakeWake =
+  ProcessorExitPublisher $ \requested ->
+    -- Publish the terminal outcome before the STM wakeup. Masking prevents
+    -- cancellation from leaving only the outcome set; this path runs once per
+    -- terminal request, not once per message.
+    mask_ $ do
+      publishProcessorExit signal requested
+      atomically $ writeTVar intakeWake True
+{-# OPAQUE newProcessorExitPublisherWithWake #-}
+
+readProcessorExit :: ProcessorSignal -> IO (Maybe ProcessorExit)
+readProcessorExit = readIORef . (.terminalExit)
+{-# INLINE readProcessorExit #-}
+
+requestProcessorExit :: ProcessorExitPublisher -> ProcessorExit -> IO ()
+requestProcessorExit (ProcessorExitPublisher publish) = publish
+{-# OPAQUE requestProcessorExit #-}
+
+publishProcessorExit :: ProcessorSignal -> ProcessorExit -> IO ()
+publishProcessorExit signal requested =
+  atomicModifyIORef' signal.terminalExit $ \current ->
+    ( case (current, requested) of
+        (Just ProcessorFailed {}, _) -> current
+        (_, ProcessorFailed {}) -> Just requested
+        (Nothing, _) -> Just requested
+        (Just ProcessorHalted {}, ProcessorHalted {}) -> current,
+      ()
+    )
+
+throwProcessorExit :: ProcessorExit -> IO a
+throwProcessorExit (ProcessorHalted reason) = throwIO (ProcessorHalt reason)
+throwProcessorExit (ProcessorFailed failure messageId) =
+  throwIO (ProcessorFailure failure messageId)
diff --git a/src/Shibuya/Internal/Runner/KeyedScheduler.hs b/src/Shibuya/Internal/Runner/KeyedScheduler.hs
--- a/src/Shibuya/Internal/Runner/KeyedScheduler.hs
+++ b/src/Shibuya/Internal/Runner/KeyedScheduler.hs
@@ -30,7 +30,7 @@
 import Data.Unique (Unique, newUnique)
 import Streamly.Data.Fold qualified as Fold
 import Streamly.Data.Stream qualified as Stream
-import UnliftIO (Async, SomeException, async, cancel, catchAny, finally, throwIO, withAsync)
+import UnliftIO (Async, SomeException, async, cancel, catchAny, finally, mask_, throwIO, withAsync)
 
 -- | Run every item in a stream through the worker action, at most
 -- @maxConcurrency@ at a time. Items with the same @Just key@ run strictly in
@@ -69,6 +69,11 @@
           SchedulerDone Nothing -> pure ()
           SchedulerDone (Just ex) -> throwIO ex
           StartItem item -> do
+            -- Keep worker creation, registration, and gate release in one masked
+            -- ownership transfer. The worker cannot pass the gate before its
+            -- handle is present in the cancellation registry. The loop has one
+            -- surrounding mask, so this path does not allocate a new mask frame
+            -- for every item.
             workerId <- newUnique
             startGate <- newEmptyMVar
             worker <- async $ do
@@ -79,7 +84,7 @@
             loop
 
   withAsync reader $ \_reader ->
-    loop `finally` cancelWorkers
+    mask_ (loop `finally` cancelWorkers)
 
 data KeyedSchedulerState key item = KeyedSchedulerState
   { inputDone :: !Bool,
@@ -155,21 +160,24 @@
   STM (SchedulerStep item)
 nextSchedulerStep maxConcurrency itemKey scheduler = do
   s <- readTVar scheduler
-  case (s.running < maxConcurrency, popStartable itemKey s.activeKeys s.pending) of
-    (True, Just (item, rest)) -> do
-      writeTVar
-        scheduler
-        s
-          { activeKeys = maybe s.activeKeys (`Set.insert` s.activeKeys) (itemKey item),
-            running = s.running + 1,
-            pending = rest
-          }
-      pure (StartItem item)
-    _
-      | s.inputDone && Seq.null s.pending && s.running == 0 ->
-          pure (SchedulerDone s.firstFailure)
-      | otherwise ->
-          retry
+  case s.firstFailure of
+    Just failure -> pure (SchedulerDone (Just failure))
+    Nothing ->
+      case (s.running < maxConcurrency, popStartable itemKey s.activeKeys s.pending) of
+        (True, Just (item, rest)) -> do
+          writeTVar
+            scheduler
+            s
+              { activeKeys = maybe s.activeKeys (`Set.insert` s.activeKeys) (itemKey item),
+                running = s.running + 1,
+                pending = rest
+              }
+          pure (StartItem item)
+        _
+          | s.inputDone && Seq.null s.pending && s.running == 0 ->
+              pure (SchedulerDone Nothing)
+          | otherwise ->
+              retry
 
 finishItem ::
   (Ord key) =>
diff --git a/src/Shibuya/Internal/Runner/Master.hs b/src/Shibuya/Internal/Runner/Master.hs
--- a/src/Shibuya/Internal/Runner/Master.hs
+++ b/src/Shibuya/Internal/Runner/Master.hs
@@ -23,13 +23,28 @@
     getAllMetricsIO,
     getProcessorMetrics,
     getProcessorMetricsIO,
+    MasterPhase (..),
+    getMasterPhase,
+    getMasterPhaseIO,
+    ProcessorLifecycle (..),
+    LifecycleSnapshot,
+    getLifecycleSnapshot,
+    getLifecycleSnapshotIO,
 
     -- * Processor Management
     registerProcessor,
     unregisterProcessor,
+    markMasterRunning,
+    markMasterDraining,
+    markProcessorDraining,
+    markProcessorStopped,
+    markProcessorStoppedIO,
+    markProcessorFailed,
+    markProcessorFailedIO,
   )
 where
 
+import Control.Concurrent.Async (asyncWithUnmask)
 import Control.Concurrent.NQE.Process (Process (..), newMailbox)
 import Control.Concurrent.NQE.Supervisor (Strategy (..), Supervisor)
 import Control.Concurrent.NQE.Supervisor qualified as Supervisor
@@ -40,6 +55,8 @@
     newTVarIO,
     readTVar,
   )
+import Control.Exception qualified as Exception
+import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, newIORef, readIORef)
 import Data.Map.Strict (Map)
 import Data.Map.Strict qualified as Map
 import Effectful (Eff, IOE, liftIO, (:>))
@@ -50,13 +67,20 @@
     ProcessorMetrics,
     sampleMetrics,
   )
+import Shibuya.Core.Types (MessageId)
 import Shibuya.Prelude
-import UnliftIO (async, cancel)
+import UnliftIO (cancel)
 
--- | Master state held in TVars.
+-- | Master ownership and observation state.
 data MasterState = MasterState
-  { -- | Map of processor IDs to their metrics handles
-    metrics :: !(TVar (Map ProcessorId MetricsHandle)),
+  { -- | Live metrics and retained lifecycle state share one STM ownership
+    -- cell. Registration can therefore publish both atomically without adding
+    -- another per-master TVar to the startup path.
+    registry :: !(TVar MasterRegistry),
+    -- | The master phase is sampled independently by health endpoints. Keeping
+    -- it in an atomic reference avoids paying for a standalone STM transaction
+    -- on every stop while processor registry updates remain transactional.
+    phaseRef :: !(IORef MasterPhase),
     -- | The supervisor managing child processors
     supervisor :: !Supervisor,
     -- | Whether child failures should be linked into the spawning thread.
@@ -66,6 +90,31 @@
   }
   deriving (Generic)
 
+data MasterRegistry = MasterRegistry
+  { liveMetrics :: !(Map ProcessorId MetricsHandle),
+    lifecycles :: !LifecycleSnapshot
+  }
+  deriving (Generic)
+
+-- | Lifecycle phase of the master itself. Processor terminal state is retained
+-- separately in 'LifecycleSnapshot'.
+data MasterPhase
+  = MasterStarting
+  | MasterRunning
+  | MasterDraining
+  | MasterStopped
+  deriving stock (Eq, Show, Generic)
+
+-- | Internal lifecycle state retained for the configured processor set.
+data ProcessorLifecycle
+  = LifecycleRunning
+  | LifecycleDraining
+  | LifecycleStopped
+  | LifecycleFailed !Text !(Maybe MessageId)
+  deriving stock (Eq, Show, Generic)
+
+type LifecycleSnapshot = Map ProcessorId ProcessorLifecycle
+
 -- | Master handle - owns the shared supervisor and metrics registry.
 newtype Master = Master
   { -- | Direct access to master state
@@ -86,23 +135,34 @@
 -- is simply collected. Processor failures still reach the caller, exactly once,
 -- through the per-processor links installed when 'propagateFailures' is set.
 startMaster :: (IOE :> es) => Strategy -> Eff es Master
-startMaster strategy = liftIO $ do
+startMaster strategy = liftIO $ Exception.mask_ $ do
   (inbox, mailbox) <- newMailbox
-  supAsync <- async (Supervisor.supervisorProcess strategy inbox)
+  -- The parent stays masked through the ownership transfer, but the long-lived
+  -- supervisor must run unmasked. Inheriting the parent's masking state makes
+  -- every supervisor cycle retain exception machinery and materially regresses
+  -- repeated startup/shutdown.
+  supAsync <- asyncWithUnmask $ \unmask ->
+    unmask (Supervisor.supervisorProcess strategy inbox)
+  -- Everything after 'async' is a non-blocking ownership transfer under the
+  -- mask. Cancellation is delivered only after the completed 'Master' returns
+  -- to 'acquireOwned', which then owns cleanup; there is no interruptible gap
+  -- that needs an extra exception frame here.
   let sup = Process supAsync mailbox
-
-  metricsMapVar <- newTVarIO Map.empty
+  registryVar <- newTVarIO $ MasterRegistry Map.empty Map.empty
+  phaseRef <- newIORef MasterStarting
   let propagate = case strategy of
         KillAll -> True
         IgnoreGraceful -> True
         IgnoreAll -> False
         Notify _ -> False
-  pure Master {state = MasterState metricsMapVar sup propagate}
+  pure Master {state = MasterState registryVar phaseRef sup propagate}
 
 -- | Stop the master and all child processors.
 -- Cancels the supervisor, which cancels all children via NQE's stopAll.
 stopMaster :: (IOE :> es) => Master -> Eff es ()
-stopMaster master = liftIO $ cancel (getProcessAsync master.state.supervisor)
+stopMaster master = liftIO $ do
+  atomicWriteIORef master.state.phaseRef MasterStopped
+  cancel (getProcessAsync master.state.supervisor)
 
 -- | Get metrics for all processors.
 getAllMetrics :: (IOE :> es) => Master -> Eff es MetricsMap
@@ -111,8 +171,8 @@
 -- | Get metrics for all processors (IO version for web servers).
 getAllMetricsIO :: Master -> IO MetricsMap
 getAllMetricsIO master = do
-  handlesMap <- atomically $ readTVar master.state.metrics
-  traverse sampleMetrics handlesMap
+  registry <- atomically $ readTVar master.state.registry
+  traverse sampleMetrics registry.liveMetrics
 
 -- | Get metrics for a specific processor.
 getProcessorMetrics :: (IOE :> es) => Master -> ProcessorId -> Eff es (Maybe ProcessorMetrics)
@@ -121,16 +181,92 @@
 -- | Get metrics for a specific processor (IO version for web servers).
 getProcessorMetricsIO :: Master -> ProcessorId -> IO (Maybe ProcessorMetrics)
 getProcessorMetricsIO master pid = do
-  handlesMap <- atomically $ readTVar master.state.metrics
-  traverse sampleMetrics (Map.lookup pid handlesMap)
+  registry <- atomically $ readTVar master.state.registry
+  traverse sampleMetrics (Map.lookup pid registry.liveMetrics)
 
+-- | Read the master lifecycle phase.
+getMasterPhase :: (IOE :> es) => Master -> Eff es MasterPhase
+getMasterPhase = liftIO . getMasterPhaseIO
+
+-- | IO variant for health integrations.
+getMasterPhaseIO :: Master -> IO MasterPhase
+getMasterPhaseIO = readIORef . (.state.phaseRef)
+
+-- | Read the retained processor lifecycle snapshot.
+getLifecycleSnapshot :: (IOE :> es) => Master -> Eff es LifecycleSnapshot
+getLifecycleSnapshot = liftIO . getLifecycleSnapshotIO
+
+-- | IO variant for metrics and health integrations.
+getLifecycleSnapshotIO :: Master -> IO LifecycleSnapshot
+getLifecycleSnapshotIO master = (.lifecycles) <$> atomically (readTVar master.state.registry)
+
 -- | Register a processor with the master.
 -- The processor should call this with its metrics handle.
 registerProcessor :: (IOE :> es) => Master -> ProcessorId -> MetricsHandle -> Eff es ()
 registerProcessor master pid metricsHandle =
-  liftIO $ atomically $ modifyTVar' master.state.metrics $ Map.insert pid metricsHandle
+  liftIO $
+    atomically $
+      modifyTVar' master.state.registry $ \registry ->
+        registry
+          { liveMetrics = Map.insert pid metricsHandle registry.liveMetrics,
+            lifecycles = Map.insert pid LifecycleRunning registry.lifecycles
+          }
 
 -- | Unregister a processor from the master.
 unregisterProcessor :: (IOE :> es) => Master -> ProcessorId -> Eff es ()
 unregisterProcessor master pid =
-  liftIO $ atomically $ modifyTVar' master.state.metrics $ Map.delete pid
+  liftIO $
+    atomically $
+      modifyTVar' master.state.registry $ \registry ->
+        registry {liveMetrics = Map.delete pid registry.liveMetrics}
+
+markMasterRunning :: (IOE :> es) => Master -> Eff es ()
+markMasterRunning master = liftIO $ advanceMasterPhase master MasterRunning
+
+markMasterDraining :: (IOE :> es) => Master -> Eff es ()
+markMasterDraining master = liftIO $ advanceMasterPhase master MasterDraining
+
+advanceMasterPhase :: Master -> MasterPhase -> IO ()
+advanceMasterPhase master next =
+  atomicModifyIORef' master.state.phaseRef $ \current ->
+    (case current of MasterStopped -> MasterStopped; _ -> next, ())
+
+markProcessorDraining :: (IOE :> es) => Master -> ProcessorId -> Eff es ()
+markProcessorDraining master pid =
+  liftIO $
+    atomically $
+      modifyTVar' master.state.registry $ \registry ->
+        registry
+          { lifecycles =
+              Map.adjust
+                (\case LifecycleRunning -> LifecycleDraining; terminal -> terminal)
+                pid
+                registry.lifecycles
+          }
+
+markProcessorStopped :: (IOE :> es) => Master -> ProcessorId -> Eff es ()
+markProcessorStopped master = liftIO . markProcessorStoppedIO master
+
+markProcessorStoppedIO :: Master -> ProcessorId -> IO ()
+markProcessorStoppedIO master pid =
+  atomically $
+    modifyTVar' master.state.registry $ \registry ->
+      registry
+        { lifecycles =
+            Map.adjust
+              (\case LifecycleFailed failure messageId -> LifecycleFailed failure messageId; _ -> LifecycleStopped)
+              pid
+              registry.lifecycles
+        }
+
+markProcessorFailed :: (IOE :> es) => Master -> ProcessorId -> Text -> Maybe MessageId -> Eff es ()
+markProcessorFailed master pid failure messageId =
+  liftIO $ markProcessorFailedIO master pid failure messageId
+
+markProcessorFailedIO :: Master -> ProcessorId -> Text -> Maybe MessageId -> IO ()
+markProcessorFailedIO master pid failure messageId =
+  atomically $
+    modifyTVar' master.state.registry $ \registry ->
+      registry
+        { lifecycles = Map.insert pid (LifecycleFailed failure messageId) registry.lifecycles
+        }
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
@@ -41,12 +41,15 @@
     retry,
     writeTVar,
   )
+import Control.Exception qualified as IOException
 import Control.Monad (when)
+import Data.Foldable (traverse_)
 import Data.HashMap.Strict qualified as HashMap
-import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef)
 import Data.Text qualified as Text
 import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, withEffToIO, (:>))
 import Effectful.Dispatch.Static (unsafeEff_)
+import Effectful.Exception qualified as Exception
+import GHC.IO (unsafeUnmask)
 import OpenTelemetry.Attributes (Attribute, toAttribute)
 import OpenTelemetry.Trace.Core qualified as OTel
 import Shibuya.Adapter (Adapter (..))
@@ -79,10 +82,28 @@
 import Shibuya.Internal.Runner.BatchProcessor (processBatchesUntilDrained)
 import Shibuya.Internal.Runner.Batcher (runBatcher)
 import Shibuya.Internal.Runner.Finalize (finalizeWithRetry)
-import Shibuya.Internal.Runner.Halt (ProcessorHalt (..))
+import Shibuya.Internal.Runner.Halt
+  ( ProcessorExit (..),
+    ProcessorExitPublisher,
+    ProcessorFailure (..),
+    ProcessorHalt (..),
+    ProcessorSignal,
+    newProcessorExitPublisherWithWake,
+    newProcessorSignal,
+    readProcessorExit,
+    requestProcessorExit,
+    throwProcessorExit,
+  )
 import Shibuya.Internal.Runner.Ingester (runIngesterWithMetrics)
 import Shibuya.Internal.Runner.KeyedScheduler (runKeyedScheduler)
-import Shibuya.Internal.Runner.Master (Master (..), MasterState (..), registerProcessor, unregisterProcessor)
+import Shibuya.Internal.Runner.Master
+  ( Master (..),
+    MasterState (..),
+    markProcessorFailedIO,
+    markProcessorStoppedIO,
+    registerProcessor,
+    unregisterProcessor,
+  )
 import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))
 import Shibuya.Prelude
 import Shibuya.Telemetry.Effect
@@ -116,7 +137,7 @@
 import Streamly.Data.Fold qualified as Fold
 import Streamly.Data.Stream qualified as Stream
 import Streamly.Data.Stream.Prelude qualified as StreamP
-import UnliftIO (Async, catch, catchAny, displayException, finally, throwIO)
+import UnliftIO (Async, SomeException, catchAny, displayException, finally)
 import UnliftIO qualified as UIO
 
 -- | Handle for a supervised processor.
@@ -169,7 +190,7 @@
   -- | Message handler
   Handler es msg ->
   Eff es SupervisedProcessor
-runSupervised master inboxSize procId ordering concurrency adapter handler = do
+runSupervised master inboxSize procId ordering concurrency adapter handler = Exception.mask_ $ do
   now <- liftIO getCurrentTime
 
   -- Initialize state
@@ -181,17 +202,19 @@
 
   -- Add as supervised child using NQE's Supervisor
   -- ConcUnlift Persistent allows the runInIO function to be used in the async child
+  -- NQE 0.6.6 masks its child-registration transfer, so this action starts in
+  -- MaskedInterruptible. Keep framework coordination in that state and unmask
+  -- only the owned adapter and message actions below: a whole-runner restore
+  -- makes Streamly add exception bookkeeping to every unordered item.
   supervisedChild <- withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
-    addChild master.state.supervisor $
-      runInIO
-        ( -- Catch ProcessorHalt to prevent propagation via link
-          -- (Halt is intentional, not a failure - other processors should continue)
-          ( runIngesterAndProcessor metricsHandle procId inboxSize ordering concurrency adapter handler
-              `catch` \(ProcessorHalt _) -> pure () -- Convert halt to graceful exit
-          )
-            `finally` unregisterProcessor master procId
-        )
-        `finally` atomically (writeTVar doneVar True)
+    let processorAction =
+          runInIO $
+            runIngesterAndProcessor metricsHandle procId inboxSize ordering concurrency adapter handler
+        unregisterAction = runInIO $ unregisterProcessor master procId
+     in addChild master.state.supervisor $
+          superviseProcessorLifecycleIO master procId processorAction
+            `finally` unregisterAction
+            `finally` atomically (writeTVar doneVar True)
 
   -- Link so exceptions propagate to the parent for strategies that request it.
   when master.state.propagateFailures $
@@ -264,8 +287,10 @@
   withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
     -- Ingester: run until stream exhausts, then signal done
     -- Use finally to ensure streamDoneVar is always set, even if ingester fails
+    -- 'unsafeUnmask' is scoped inside the surrounding withAsync/finally owner;
+    -- cancellation therefore reaches adapter code without bypassing cleanup.
     let ingesterWithSignal =
-          runInIO (runIngesterWithMetrics metricsHandle adapter.source inbox)
+          unsafeUnmask (runInIO (runIngesterWithMetrics metricsHandle adapter.source inbox))
             `finally` atomically (writeTVar streamDoneVar True)
 
     UIO.withAsync ingesterWithSignal $ \ingesterAsync -> do
@@ -302,7 +327,7 @@
   -- | Batch handler
   BatchHandler es msg ->
   Eff es SupervisedProcessor
-runSupervisedBatch master inboxSize procId concurrency batchConfig adapter batchHandler = do
+runSupervisedBatch master inboxSize procId concurrency batchConfig adapter batchHandler = Exception.mask_ $ do
   now <- liftIO getCurrentTime
 
   metricsHandle <- liftIO $ newMetricsHandle now
@@ -311,9 +336,9 @@
   registerProcessor master procId metricsHandle
 
   supervisedChild <- withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
-    addChild master.state.supervisor $
-      runInIO
-        ( ( runIngesterAndProcessorBatch
+    let processorAction =
+          runInIO $
+            runIngesterAndProcessorBatch
               metricsHandle
               procId
               inboxSize
@@ -321,11 +346,11 @@
               batchConfig
               adapter
               batchHandler
-              `catch` \(ProcessorHalt _) -> pure ()
-          )
-            `finally` unregisterProcessor master procId
-        )
-        `finally` atomically (writeTVar doneVar True)
+        unregisterAction = runInIO $ unregisterProcessor master procId
+     in addChild master.state.supervisor $
+          superviseProcessorLifecycleIO master procId processorAction
+            `finally` unregisterAction
+            `finally` atomically (writeTVar doneVar True)
 
   when master.state.propagateFailures $
     unsafeEff_ $
@@ -339,6 +364,29 @@
         child = Just supervisedChild
       }
 
+-- | Convert deliberate halt into successful termination while retaining every
+-- infrastructure failure in the master's bounded terminal snapshot. Async
+-- cancellation is a stop, not a processor failure, and is rethrown after the
+-- snapshot transition so supervisor cleanup keeps its normal semantics.
+superviseProcessorLifecycleIO :: Master -> ProcessorId -> IO () -> IO ()
+superviseProcessorLifecycleIO master procId action = do
+  outcome <- IOException.try @SomeException action
+  case outcome of
+    Right () -> markProcessorStoppedIO master procId
+    Left unexpected -> do
+      case IOException.fromException unexpected of
+        Just (ProcessorHalt _) -> markProcessorStoppedIO master procId
+        Nothing ->
+          case IOException.fromException unexpected of
+            Just (ProcessorFailure message messageId) ->
+              markProcessorFailedIO master procId message messageId
+            Nothing
+              | Exception.isAsyncException unexpected -> markProcessorStoppedIO master procId
+              | otherwise -> markProcessorFailedIO master procId (Text.pack (displayException unexpected)) Nothing
+      case IOException.fromException unexpected of
+        Just (ProcessorHalt _) -> pure ()
+        Nothing -> IOException.throwIO unexpected
+
 -- | Run a batching processor with metrics but without Master supervision.
 -- Blocks until the adapter stream is exhausted and every accumulated batch has
 -- been processed (including the end-of-input flush). Useful for tests.
@@ -406,15 +454,16 @@
 runIngesterAndProcessorBatch metricsHandle procId inboxSize concurrency batchConfig adapter batchHandler = do
   inbox <- liftIO $ newBoundedInbox inboxSize
   streamDoneVar <- liftIO $ newTVarIO False
-  haltRef <- liftIO $ newIORef Nothing
+  stopSignal <- liftIO newProcessorSignal
+  let exitPublisher = newProcessorExitPublisherWithWake stopSignal streamDoneVar
 
   withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
     let ingesterWithSignal =
-          runInIO (runIngesterWithMetrics metricsHandle adapter.source inbox)
+          unsafeUnmask (runInIO (runIngesterWithMetrics metricsHandle adapter.source inbox))
             `finally` atomically (writeTVar streamDoneVar True)
 
     UIO.withAsync ingesterWithSignal $ \ingesterAsync -> do
-      let inboxStream = inboxToStream inbox streamDoneVar haltRef
+      let inboxStream = inboxToStream inbox streamDoneVar stopSignal
           readyBatchStream = runBatcher inboxSize batchConfig inboxStream
           batchProcessor =
             runInIO $ do
@@ -424,9 +473,10 @@
                 concurrency
                 batchHandler
                 readyBatchStream
-                haltRef
-              maybeHalt <- liftIO (readIORef haltRef)
-              maybe (pure ()) (throwIO . ProcessorHalt) maybeHalt
+                stopSignal
+                exitPublisher
+              maybeExit <- liftIO (readProcessorExit stopSignal)
+              maybe (pure ()) (liftIO . throwProcessorExit) maybeExit
       batchProcessor `catchAny` \processorErr -> do
         now <- getCurrentTime
         atomically $
@@ -451,32 +501,26 @@
 inboxToStream ::
   Inbox (Ingested es msg) ->
   TVar Bool ->
-  IORef (Maybe HaltReason) ->
+  ProcessorSignal ->
   Stream.Stream IO (Ingested es msg)
-inboxToStream inbox streamDoneVar haltRef = Stream.unfoldrM step ()
+inboxToStream inbox streamDoneVar stopSignal = Stream.unfoldrM step ()
   where
     step _ = do
-      -- Check halt flag first (outside STM since it's an IORef)
-      halted <- readIORef haltRef
-      case halted of
-        Just _ -> pure Nothing -- Stop reading
+      terminal <- readProcessorExit stopSignal
+      case terminal of
+        Just _ -> pure Nothing
         Nothing -> do
-          -- Atomically either receive a message or detect completion.
-          -- This avoids TOCTOU race where we check done/empty separately
-          -- and then block on receive after the stream has completed.
           result <-
             atomically $
-              -- Try to receive a message
               (Just <$> receiveSTM inbox)
-                `orElse`
-                -- Or check if we're done (stream exhausted and inbox empty)
-                ( do
-                    done <- readTVar streamDoneVar
-                    empty <- mailboxEmptySTM inbox
-                    if done && empty
-                      then pure Nothing
-                      else retry -- Inbox empty but stream not done, wait for message
-                )
+                `orElse` do
+                  -- Source completion and terminal publication share this
+                  -- wake cell. A populated inbox therefore keeps the original
+                  -- receive-only hot branch, while either terminal event wakes
+                  -- an empty intake wait.
+                  done <- readTVar streamDoneVar
+                  empty <- mailboxEmptySTM inbox
+                  if done && empty then pure Nothing else retry
           pure $ fmap (,()) result
 
 -- | Process messages from inbox until stream is done and inbox is empty.
@@ -492,7 +536,8 @@
   TVar Bool ->
   Eff es ()
 processUntilDrained metricsHandle procId ordering concurrency handler inbox streamDoneVar = do
-  haltRef <- liftIO $ newIORef Nothing
+  stopSignal <- liftIO newProcessorSignal
+  let exitPublisher = newProcessorExitPublisherWithWake stopSignal streamDoneVar
 
   let maxConc = case concurrency of
         Serial -> 1
@@ -508,8 +553,13 @@
           ]
 
   withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
-    let inboxStream = inboxToStream inbox streamDoneVar haltRef
-        processAction = runInIO . processOne metricsHandle spanName constantFrameworkAttrs maxConc haltRef handler
+    let inboxStream = inboxToStream inbox streamDoneVar stopSignal
+        runProcessAction ingested =
+          runInIO $
+            processOne metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested
+        -- Restore normal interruptibility for handlers and finalizers while the
+        -- concurrent Streamly scheduler retains its inherited mask.
+        processAction = unsafeUnmask . runProcessAction
         partitioned n =
           runKeyedScheduler
             (max 1 n)
@@ -520,8 +570,11 @@
 
     case (ordering, concurrency) of
       (_, Serial) ->
-        Stream.fold Fold.drain $
-          Stream.mapM processAction inboxStream
+        -- Unmask the serial region once. Unmasking each message is measurable
+        -- overhead, while this path has no concurrent scheduler to protect.
+        unsafeUnmask $
+          Stream.fold Fold.drain $
+            Stream.mapM runProcessAction inboxStream
       (PartitionedInOrder, Ahead n) ->
         partitioned n
       (PartitionedInOrder, Async n) ->
@@ -543,10 +596,8 @@
           StreamP.parMapM (StreamP.maxThreads n . StreamP.maxBuffer (2 * n)) processAction inboxStream
 
     -- After draining, check if we halted
-    maybeHalt <- readIORef haltRef
-    case maybeHalt of
-      Just reason -> throwIO $ ProcessorHalt reason
-      Nothing -> pure ()
+    maybeExit <- readProcessorExit stopSignal
+    traverse_ throwProcessorExit maybeExit
 
 handlerStartedEvent :: OTel.NewEvent
 handlerStartedEvent = mkEvent eventHandlerStarted []
@@ -560,11 +611,11 @@
   Text ->
   HashMap.HashMap Text Attribute ->
   Int ->
-  IORef (Maybe HaltReason) ->
+  ProcessorExitPublisher ->
   Handler es msg ->
   Ingested es msg ->
   Eff es ()
-processOne metricsHandle spanName constantFrameworkAttrs maxConc haltRef handler ingested = do
+processOne metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested = do
   -- Extract parent context from message headers for distributed tracing
   let parentCtx = ingested.envelope.traceContext >>= extractTraceContext
 
@@ -577,7 +628,7 @@
       -- the precedence rule local and obvious instead of relying on
       -- the order of repeated 'addAttribute' / 'addAttributes' calls
       -- against the underlying mutable Span.
-      let MessageId msgIdText = ingested.envelope.messageId
+      let messageId@(MessageId msgIdText) = ingested.envelope.messageId
           frameworkAttrs =
             HashMap.insert attrMessagingMessageId (toAttribute msgIdText) $
               case ingested.envelope.partition of
@@ -674,9 +725,11 @@
       -- Handle halt (set flag, don't throw - let stream drain)
       case finalizeResult of
         Left _ ->
-          liftIO $ atomicWriteIORef haltRef (Just (HaltFatal (finalizationFailureText msgIdText)))
+          liftIO $
+            requestProcessorExit exitPublisher (ProcessorFailed (finalizationFailureText msgIdText) (Just messageId))
         Right () -> case result of
-          Right (AckHalt reason) -> liftIO $ atomicWriteIORef haltRef (Just reason)
+          Right (AckHalt reason) ->
+            liftIO $ requestProcessorExit exitPublisher (ProcessorHalted reason)
           _ -> pure ()
   where
     isLeft :: Either a b -> Bool
diff --git a/src/Shibuya/Policy.hs b/src/Shibuya/Policy.hs
--- a/src/Shibuya/Policy.hs
+++ b/src/Shibuya/Policy.hs
@@ -44,6 +44,21 @@
 -- | Validate policy combinations.
 -- Invariant: StrictInOrder => Serial
 validatePolicy :: OrderingPolicy -> Concurrency -> Either PolicyError ()
-validatePolicy StrictInOrder (Ahead _) = Left $ InvalidPolicyCombo "StrictInOrder requires Serial concurrency"
-validatePolicy StrictInOrder (Async _) = Left $ InvalidPolicyCombo "StrictInOrder requires Serial concurrency"
-validatePolicy _ _ = Right ()
+validatePolicy ordering concurrency = do
+  validateConcurrency concurrency
+  validateCombination ordering concurrency
+  where
+    validateConcurrency Serial = Right ()
+    validateConcurrency (Ahead n) = validateBound n
+    validateConcurrency (Async n) = validateBound n
+
+    validateBound n
+      | n < 1 = Left $ InvalidConcurrency n
+      | n > maxBound `div` 2 = Left $ ConcurrencyCapacityOverflow n
+      | otherwise = Right ()
+
+    validateCombination StrictInOrder (Ahead _) =
+      Left $ InvalidPolicyCombo "StrictInOrder requires Serial concurrency"
+    validateCombination StrictInOrder (Async _) =
+      Left $ InvalidPolicyCombo "StrictInOrder requires Serial concurrency"
+    validateCombination _ _ = Right ()
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
@@ -2,16 +2,19 @@
 
 module Shibuya.App.LifecycleSpec (spec) where
 
-import Control.Concurrent (threadDelay)
+import Control.Concurrent (newEmptyMVar, putMVar, threadDelay)
 import Control.Concurrent.NQE.Supervisor (Strategy (..))
-import Control.Concurrent.STM (readTVarIO)
+import Control.Concurrent.STM (atomically, check, readTVar, readTVarIO, retry)
 import Control.Exception (SomeException, mask, try)
-import Control.Monad (join, void)
+import Control.Monad (forM_, join, void)
 import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)
 import Data.Map.Strict qualified as Map
 import Data.Text qualified as Text
 import Data.Time (UTCTime (..), diffUTCTime, fromGregorian, getCurrentTime)
-import Effectful (Eff, IOE, liftIO, runEff, (:>))
+import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, runEff, withEffToIO, (:>))
+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.App
   ( AppConfig (..),
@@ -31,17 +34,49 @@
 import Shibuya.Core.Ingested (Ingested, mkIngested)
 import Shibuya.Core.Metrics (ProcessorId (..), ProcessorMetrics (..), ProcessorState (..), sampleMetrics)
 import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), mkEnvelope)
-import Shibuya.Internal.App (AppHandle (..))
-import Shibuya.Internal.Runner.Master (startMaster, stopMaster)
+import Shibuya.Internal.App
+  ( AppHandle (..),
+    OwnershipFailure (..),
+    acquireOwned,
+  )
+import Shibuya.Internal.Runner.Master
+  ( ProcessorLifecycle (..),
+    getLifecycleSnapshot,
+    startMaster,
+    stopMaster,
+  )
 import Shibuya.Internal.Runner.Supervised (SupervisedProcessor (..), runSupervised)
 import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))
-import Shibuya.Telemetry.Effect (Tracing, runTracingNoop)
+import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop)
 import Streamly.Data.Stream qualified as Stream
 import Test.Hspec
 import UnliftIO qualified as UIO
 
 spec :: Spec
 spec = describe "Shibuya.App lifecycle" $ do
+  it "cleans an acquired startup owner when cancellation lands at the transfer barrier" $ do
+    transferReached <- newEmptyMVar
+    cleanupCalled <- newEmptyMVar
+    worker <-
+      UIO.async $
+        runEff $
+          acquireOwned
+            (pure ())
+            (const $ liftIO $ putMVar cleanupCalled ())
+            (\() -> liftIO (putMVar transferReached ()) >> liftIO (atomically retry))
+
+    UIO.takeMVar transferReached
+    UIO.cancel worker
+    result <- UIO.wait worker
+    cleaned <- UIO.timeout 1_000_000 (UIO.takeMVar cleanupCalled)
+
+    case result of
+      Left (OwnedActionFailed failure Nothing) ->
+        Exception.isAsyncException failure `shouldBe` True
+      Left other -> expectationFailure $ "unexpected ownership failure: " <> show other
+      Right () -> expectationFailure "startup transfer unexpectedly completed"
+    cleaned `shouldBe` Just ()
+
   it "waitApp returns after a handler halts" $ do
     result <-
       UIO.timeout 5_000_000 $
@@ -52,7 +87,7 @@
                 processor = mkProcessor (testAdapter messages) handler
             app <- runAppOrFail IgnoreFailures 10 [(ProcessorId "halt", processor)]
             waitApp app
-            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
             pure ()
 
     result `shouldBe` Just ()
@@ -68,7 +103,7 @@
             app <- runAppOrFail IgnoreFailures 10 [(ProcessorId "halt-stop", processor)]
             liftIO $ threadDelay 200_000
             startedAt <- liftIO getCurrentTime
-            drained <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+            drained <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
             finishedAt <- liftIO getCurrentTime
             pure (drained, diffUTCTime finishedAt startedAt)
 
@@ -99,11 +134,196 @@
                 processor = mkBatchProcessor (testAdapter messages) handler config
             app <- runAppOrFail IgnoreFailures 10 [(ProcessorId "batch-halt", processor)]
             waitApp app
-            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
             pure ()
 
     result `shouldBe` Just ()
 
+  it "halt wakes idle intake in every queue strategy and batch mode" $ do
+    let queue concurrency ordering =
+          QueueProcessor oneThenIdleAdapter (const (pure (AckHalt (HaltFatal "idle halt")))) ordering concurrency
+        batch =
+          mkBatchProcessor
+            oneThenIdleAdapter
+            (\_ _ -> pure (ackAll (AckHalt (HaltFatal "idle batch halt"))))
+            defaultBatchConfig {batchSize = 10, batchTimeout = 0.05}
+        processors =
+          [ (ProcessorId "idle-serial", queue Serial Unordered),
+            (ProcessorId "idle-ahead", queue (Ahead 2) Unordered),
+            (ProcessorId "idle-async", queue (Async 2) Unordered),
+            (ProcessorId "idle-keyed", queue (Async 2) PartitionedInOrder),
+            (ProcessorId "idle-batch", batch)
+          ]
+
+    result <-
+      UIO.timeout 5_000_000 $
+        runEff $
+          runTracingNoop $ do
+            app <- runAppOrFail IgnoreFailures 10 processors
+            waitApp app
+            getLifecycleSnapshot app.master
+
+    case result of
+      Nothing -> expectationFailure "idle halt did not wake every processor"
+      Just snapshot ->
+        Map.elems snapshot `shouldSatisfy` all (== LifecycleStopped)
+
+  it "IgnoreFailures retains finalizer failure and message identity after waitApp" $ do
+    result <-
+      UIO.timeout 5_000_000 $
+        runEff $
+          runTracingNoop $ do
+            let pid = ProcessorId "retained-finalizer-failure"
+                processor = mkProcessor permanentFinalizerFailureAdapter alwaysAckOk
+            app <- runAppOrFail IgnoreFailures 10 [(pid, processor)]
+            waitApp app
+            snapshot <- getLifecycleSnapshot app.master
+            stopMaster app.master
+            pure (Map.lookup pid snapshot)
+
+    case result of
+      Just (Just (LifecycleFailed message (Just messageId))) -> do
+        message `shouldSatisfy` Text.isInfixOf "finalization failed"
+        messageId `shouldBe` MessageId "finalizer-failure"
+      other -> expectationFailure $ "expected retained finalizer failure, got: " <> show other
+
+  it "retains finalizer failure with tracing enabled" $ do
+    provider <- OTel.createTracerProvider [] OTel.emptyTracerProviderOptions
+    let instrumentation =
+          OTel.InstrumentationLibrary
+            { OTel.libraryName = "shibuya-lifecycle-test",
+              OTel.libraryVersion = "",
+              OTel.librarySchemaUrl = "",
+              OTel.libraryAttributes = OTelAttributes.emptyAttributes
+            }
+        tracer = OTel.makeTracer provider instrumentation OTel.tracerOptions
+        pid = ProcessorId "traced-finalizer-failure"
+
+    result <-
+      UIO.timeout 5_000_000 $
+        runEff $
+          runTracing tracer $ do
+            app <- runAppOrFail IgnoreFailures 10 [(pid, mkProcessor permanentFinalizerFailureAdapter alwaysAckOk)]
+            waitApp app
+            snapshot <- getLifecycleSnapshot app.master
+            stopMaster app.master
+            pure (Map.lookup pid snapshot)
+
+    case result of
+      Just (Just (LifecycleFailed _ (Just (MessageId "finalizer-failure")))) -> pure ()
+      other -> expectationFailure $ "expected traced finalizer failure, got: " <> show other
+
+  it "StopAllOnFailure delivers exhausted finalization exactly once" $ do
+    let failing = mkProcessor permanentFinalizerFailureAdapter alwaysAckOk
+    result <-
+      UIO.timeout 15_000_000 $
+        UIO.withAsync (countLinkedDeliveries [(ProcessorId "finalizer-single-delivery", failing)]) UIO.wait
+    result `shouldBe` Just 1
+
+  it "attempts every adapter shutdown and stops all children when one throws" $ do
+    secondShutdownRef <- newIORef False
+    (shutdownResult, secondCalled, waited) <-
+      runEff $
+        runTracingNoop $ do
+          let throwing =
+                infiniteAdapter
+                  { adapterName = "test:throwing-shutdown",
+                    shutdown = liftIO $ ioError (userError "first shutdown failed")
+                  }
+              observing =
+                infiniteAdapter
+                  { adapterName = "test:observing-shutdown",
+                    shutdown = liftIO $ writeIORef secondShutdownRef True
+                  }
+          app <-
+            runAppOrFail
+              IgnoreFailures
+              10
+              [ (ProcessorId "a-throwing-shutdown", mkProcessor throwing alwaysAckOk),
+                (ProcessorId "z-observing-shutdown", mkProcessor observing alwaysAckOk)
+              ]
+          result <-
+            Exception.try @SomeException $
+              stopAppGracefully
+                (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2})
+                app
+          didWait <- liftIO $ UIO.timeout 1_000_000 (waitForDoneIO app)
+          observed <- liftIO $ readIORef secondShutdownRef
+          pure (result, observed, didWait)
+
+    shutdownResult `shouldSatisfy` isLeft
+    secondCalled `shouldBe` True
+    waited `shouldBe` Just ()
+
+  it "bounds a never-returning adapter shutdown with the total deadline" $ do
+    result <-
+      UIO.timeout 2_000_000 $
+        runEff $
+          runTracingNoop $ do
+            let blocking =
+                  infiniteAdapter
+                    { adapterName = "test:blocking-shutdown",
+                      shutdown = liftIO $ atomically retry
+                    }
+            app <- runAppOrFail IgnoreFailures 10 [(ProcessorId "blocking-shutdown", mkProcessor blocking alwaysAckOk)]
+            drained <-
+              stopAppGracefully
+                (ShutdownConfig {drainTimeout = 10, totalShutdownTimeout = 0.1})
+                app
+            waitApp app
+            pure drained
+
+    result `shouldBe` Just False
+
+  it "runs adapter shutdown once for concurrent and repeated stop calls" $ do
+    shutdownCountRef <- newIORef (0 :: Int)
+    ((first, second, repeated), shutdownCount) <-
+      runEff $
+        runTracingNoop $ do
+          let adapter =
+                infiniteAdapter
+                  { adapterName = "test:coordinated-shutdown",
+                    shutdown = liftIO $ modifyIORef' shutdownCountRef (+ 1)
+                  }
+              config = ShutdownConfig {drainTimeout = 0.05, totalShutdownTimeout = 1}
+          app <- runAppOrFail IgnoreFailures 10 [(ProcessorId "coordinated-shutdown", mkProcessor adapter alwaysAckOk)]
+          (firstResult, secondResult) <-
+            withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
+              UIO.concurrently
+                (runInIO $ stopAppGracefully config app)
+                (runInIO $ stopAppGracefully config app)
+          repeatedResult <- stopAppGracefully config app
+          count <- liftIO $ readIORef shutdownCountRef
+          pure ((firstResult, secondResult, repeatedResult), count)
+
+    first `shouldBe` False
+    second `shouldBe` first
+    repeated `shouldBe` first
+    shutdownCount `shouldBe` 1
+
+  it "stops the master when the shutdown caller is cancelled during drain" $ do
+    result <-
+      UIO.timeout 5_000_000 $
+        runEff $
+          runTracingNoop $ do
+            handlerStarted <- liftIO newEmptyMVar
+            let handler _ = do
+                  liftIO $ putMVar handlerStarted ()
+                  liftIO $ atomically retry
+                processor = mkProcessor infiniteAdapter handler
+                config = ShutdownConfig {drainTimeout = 10, totalShutdownTimeout = 20}
+            app <- runAppOrFail IgnoreFailures 10 [(ProcessorId "cancel-drain", processor)]
+            liftIO $ UIO.takeMVar handlerStarted
+            withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
+              stopper <- UIO.async (runInIO $ stopAppGracefully config app)
+              threadDelay 20_000
+              UIO.cancel stopper
+              stopResult <- UIO.waitCatch stopper
+              completed <- UIO.timeout 1_000_000 (waitForDoneIO app)
+              pure (isLeft stopResult, completed)
+
+    result `shouldBe` Just (True, Just ())
+
   it "graceful completion under StopAllOnFailure does not kill siblings" $ do
     countBRef <- newIORef (0 :: Int)
 
@@ -126,7 +346,7 @@
                 10
                 [(ProcessorId "complete-A", procA), (ProcessorId "complete-B", procB)]
             waitApp app
-            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
             liftIO $ readIORef countBRef
 
     result `shouldBe` Just 30
@@ -159,7 +379,7 @@
                 10
                 [(ProcessorId "halt-A", procA), (ProcessorId "halt-B", procB)]
             waitApp app
-            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
             liftIO $ readIORef countBRef
 
     result `shouldBe` Just 30
@@ -185,7 +405,7 @@
                   10
                   [(ProcessorId "fail-A", procA), (ProcessorId "fail-B", procB)]
               liftIO $ threadDelay 500_000
-              _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+              _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
               pure ()
         )
         UIO.waitCatch
@@ -254,7 +474,7 @@
                 [(ProcessorId "ignore-fail-A", procA), (ProcessorId "ignore-fail-B", procB)]
             waitApp app
             metricsA <- processorMetrics app (ProcessorId "ignore-fail-A")
-            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+            _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
             countB <- liftIO $ readIORef countBRef
             pure (metricsA, countB)
 
@@ -305,7 +525,7 @@
                 writeIORef stopRef $
                   runEff $
                     runTracingNoop $
-                      void (stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app)
+                      void (stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app)
     let arrivesWithin :: Int -> IO Bool
         arrivesWithin micros = either (const True) (const False) <$> try @SomeException (restore (threadDelay micros))
         countExtra :: Int -> IO Int
@@ -329,6 +549,14 @@
         Nothing -> liftIO $ expectationFailure ("missing processor: " <> show pid) >> error "unreachable"
         Just (SupervisedProcessor {metrics = metricsHandle}, _) -> liftIO $ sampleMetrics metricsHandle
 
+waitForDoneIO :: AppHandle es -> IO ()
+waitForDoneIO app =
+  case app of
+    AppHandle {processors = processorsMap} ->
+      atomically $
+        forM_ (Map.elems processorsMap) $
+          \(sp, _) -> readTVar sp.done >>= check
+
 alwaysAckOk :: (Applicative f) => a -> f AckDecision
 alwaysAckOk _ = pure AckOk
 
@@ -382,3 +610,34 @@
       liftIO $ threadDelay 5_000
       msg <- createTestMessage n
       pure (Just (msg, n + 1))
+
+oneThenIdleAdapter :: (IOE :> es) => Adapter es String
+oneThenIdleAdapter =
+  Adapter
+    { adapterName = "test:one-then-idle",
+      source = Stream.unfoldrM step False,
+      shutdown = pure ()
+    }
+  where
+    step False = do
+      msg <- createTestMessage 1
+      pure (Just (msg, True))
+    step True = liftIO $ atomically retry
+
+permanentFinalizerFailureAdapter :: (IOE :> es) => Adapter es String
+permanentFinalizerFailureAdapter =
+  Adapter
+    { adapterName = "test:permanent-finalizer-failure",
+      source = Stream.fromEffect (pure ingested),
+      shutdown = pure ()
+    }
+  where
+    envelope = mkEnvelope (MessageId "finalizer-failure") "message"
+    ingested =
+      mkIngested
+        envelope
+        (AckHandle $ \_ -> liftIO $ ioError (userError "permanent finalizer failure"))
+
+isLeft :: Either a b -> Bool
+isLeft (Left _) = True
+isLeft (Right _) = False
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
@@ -201,7 +201,7 @@
       adapter = trackedListAdapter tracking (scenarioEnvelopes s)
       proc = mkBatchProcessor adapter (intendedHandler (scenarioIntended s)) (scenarioConfig s)
   app <- runAppOrFail 100 [(pid, proc)]
-  _drained <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+  _drained <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
   m <- metricsFor app pid
   tracked <- getTrackedDecisions tracking
   pure (tracked, m)
@@ -214,27 +214,26 @@
 spec = describe "Shibuya.Batch reliability" $ do
   describe "successful-finalization property" $ do
     it "finalizes every normal-path message once with the intended decision" $
-      withMaxSuccess 50 $
-        forAll genScenario $ \s -> monadicIO $ do
-          (tracked, metrics) <- run (runScenario s)
-          let expected = scenarioIntended s
-          case finalizedExactlyOnce tracked expected of
-            Left err -> do
-              monitor (counterexample ("successful-finalization violated: " <> err))
-              assert False
-            Right () -> pure ()
-          monitor
-            ( counterexample
-                ( "accounting: processed="
-                    <> show metrics.stats.processed
-                    <> " failed="
-                    <> show metrics.stats.failed
-                    <> " n="
-                    <> show s.msgCount
-                )
-            )
-          assert (metrics.stats.processed + metrics.stats.failed == s.msgCount)
-          assert (metrics.batch.batchedMessages == s.msgCount)
+      forAll genScenario $ \s -> monadicIO $ do
+        (tracked, metrics) <- run (runScenario s)
+        let expected = scenarioIntended s
+        case finalizedExactlyOnce tracked expected of
+          Left err -> do
+            monitor (counterexample ("successful-finalization violated: " <> err))
+            assert False
+          Right () -> pure ()
+        monitor
+          ( counterexample
+              ( "accounting: processed="
+                  <> show metrics.stats.processed
+                  <> " failed="
+                  <> show metrics.stats.failed
+                  <> " n="
+                  <> show s.msgCount
+              )
+          )
+        assert (metrics.stats.processed + metrics.stats.failed == s.msgCount)
+        assert (metrics.batch.batchedMessages == s.msgCount)
 
   -- Non-vacuity of the checker itself: feed it perturbed tracked lists and prove
   -- it fires. Without these, a checker that always returns Right () would pass
@@ -274,7 +273,7 @@
         app <- runAppOrFail 100 [(pid, proc)]
         liftIO $ threadDelay 400000 -- 400 ms > 100 ms timeout: ticker flushes
         t0 <- getTrackedDecisions tracking
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
         o <- liftIO $ readIORef observedRef
         pure (t0, o)
       finalizedExactlyOnce tracked (Map.fromList [(MessageId ("msg-" <> tshowT i), AckOk) | i <- [1 .. 3 :: Int]])
@@ -295,7 +294,7 @@
             adapter = trackedListAdapter tracking (fixedEnvelopes 5)
             proc = mkBatchProcessor adapter handler cfg
         app <- runAppOrFail 100 [(pid, proc)]
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
         t <- getTrackedDecisions tracking
         m <- metricsFor app pid
         pure (t, m)
@@ -321,7 +320,7 @@
             adapter = trackedListAdapter tracking (fixedEnvelopes 4)
             proc = mkBatchProcessor adapter handler cfg
         app <- runAppOrFail 100 [(pid, proc)]
-        d <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+        d <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
         t <- getTrackedDecisions tracking
         pure (t, d)
       let expected = Map.fromList [(MessageId ("msg-" <> tshowT i), AckRetry (RetryDelay 0)) | i <- [1 .. 4 :: Int]]
@@ -351,7 +350,7 @@
         doneState <- case Map.lookup pid app.processors of
           Just (sp, _) -> liftIO $ readTVarIO sp.done
           Nothing -> liftIO $ ioError (userError "missing batch-key-failure processor")
-        d <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+        d <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
         t <- getTrackedDecisions tracking
         pure (t, m.state, doneState, d)
       case mState of
@@ -378,7 +377,7 @@
             handler _ _ = pure ackAllOk
             proc = mkBatchProcessor (listAdapter [ing]) handler cfg
         app <- runAppOrFail 100 [(pid, proc)]
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
         t <- liftIO $ readIORef trackRef
         a <- liftIO $ readIORef attemptRef
         pure (t, a)
@@ -398,7 +397,7 @@
             proc = mkBatchProcessor (listAdapter [ing1, ing2]) handler cfg
         app <- runAppOrFail 100 [(pid, proc)]
         liftIO $ threadDelay 700000 -- allow the [10,50,250]ms retry schedule to exhaust
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
         t <- getTrackedDecisions tracking
         m <- metricsFor app pid
         pure (t, m.state)
@@ -423,7 +422,7 @@
         app <- runAppOrFail 100 [(pidA, procA), (pidB, procB)]
         liftIO $ threadDelay 300000
         mA' <- metricsFor app pidA
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}) app
         tA <- getTrackedDecisions trackingA
         tB <- getTrackedDecisions trackingB
         pure (tA, tB, mA')
@@ -457,7 +456,7 @@
             proc = mkBatchProcessor adapter handler cfg
         app <- runAppOrFail 100 [(pid, proc)]
         liftIO $ threadDelay 100000 -- accumulate (no size/timeout flush)
-        d <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+        d <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
         t <- getTrackedDecisions tracking
         o <- liftIO $ readIORef observedRef
         pure (t, d, o)
@@ -477,7 +476,7 @@
             adapter = trackedListAdapter tracking (keyedEnvelopes [BatchKey "ka", BatchKey "kb"] 6)
             proc = mkBatchProcessor adapter handler cfg
         app <- runAppOrFail 100 [(pid, proc)]
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
         t <- getTrackedDecisions tracking
         o <- liftIO $ readIORef observedRef
         pure (t, o)
@@ -516,7 +515,7 @@
                   concurrency = Async 2
                 }
         app <- runAppOrFail 100 [(pid, proc)]
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6}) app
         v <- liftIO $ readTVarIO violatedVar
         t <- getTrackedDecisions tracking
         pure (v, t)
@@ -536,7 +535,7 @@
             adapter = trackedListAdapter tracking (fixedEnvelopes 20)
             proc = mkBatchProcessor adapter handler cfg
         app <- runAppOrFail 2 [(pid, proc)] -- inbox size 2
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 10}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 10, totalShutdownTimeout = 11}) app
         getTrackedDecisions tracking
       finalizedExactlyOnce tracked (Map.fromList [(MessageId ("msg-" <> tshowT i), AckOk) | i <- [1 .. 20 :: Int]])
         `shouldBe` Right ()
@@ -575,7 +574,7 @@
         liftIO $ threadDelay 300000
         pulled <- liftIO $ readIORef pulledRef
         liftIO $ atomically $ writeTVar gate True
-        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 10}) app
+        _ <- stopAppGracefully (ShutdownConfig {drainTimeout = 10, totalShutdownTimeout = 11}) app
         t <- getTrackedDecisions tracking
         liftIO $ pulled `shouldSatisfy` (<= allowedPulls)
         pure (pulled, t)
@@ -602,7 +601,7 @@
                 }
         app <- runAppOrFail 2 [(pid, proc)]
         liftIO $ threadDelay 50000
-        d <- stopAppGracefully (ShutdownConfig {drainTimeout = 0.05}) app
+        d <- stopAppGracefully (ShutdownConfig {drainTimeout = 0.05, totalShutdownTimeout = 1}) app
         t1 <- getTrackedDecisions tracking
         liftIO $ threadDelay 300000
         t2 <- getTrackedDecisions tracking
diff --git a/test/Shibuya/PolicySpec.hs b/test/Shibuya/PolicySpec.hs
--- a/test/Shibuya/PolicySpec.hs
+++ b/test/Shibuya/PolicySpec.hs
@@ -2,6 +2,7 @@
 
 module Shibuya.PolicySpec (spec) where
 
+import Shibuya.Core.Error (PolicyError (..))
 import Shibuya.Policy
 import Test.Hspec
 
@@ -61,6 +62,29 @@
 
       it "allows Async" $ do
         validatePolicy Unordered (Async 10) `shouldBe` Right ()
+
+    describe "resource bounds" $ do
+      mapM_
+        ( \concurrency ->
+            it ("rejects nonpositive " <> show concurrency) $
+              validatePolicy Unordered concurrency `shouldBe` Left (InvalidConcurrency 0)
+        )
+        [Ahead 0, Async 0]
+
+      mapM_
+        ( \concurrency ->
+            it ("rejects negative " <> show concurrency) $
+              validatePolicy PartitionedInOrder concurrency `shouldBe` Left (InvalidConcurrency (-1))
+        )
+        [Ahead (-1), Async (-1)]
+
+      let overflowing = maxBound `div` 2 + 1
+      mapM_
+        ( \concurrency ->
+            it ("rejects derived-capacity overflow for " <> show concurrency) $
+              validatePolicy Unordered concurrency `shouldBe` Left (ConcurrencyCapacityOverflow overflowing)
+        )
+        [Ahead overflowing, Async overflowing]
 
     describe "validatePolicy matrix" $ do
       let ok = True
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
@@ -17,7 +17,7 @@
 import Data.Text qualified as Text
 import Data.Time (UTCTime (..), fromGregorian)
 import Effectful (Eff, IOE, liftIO, runEff, (:>))
-import Shibuya (ProcessorHalt (..))
+import Shibuya (ProcessorFailure (..), ProcessorHalt (..))
 import Shibuya.Adapter.Mock
   ( TrackingAck (..),
     getTrackedDecisions,
@@ -230,10 +230,11 @@
         pure ()
 
       case result of
-        Left (ProcessorHalt (HaltFatal msg)) ->
+        Left (ProcessorFailure msg (Just messageId)) -> do
           msg `shouldSatisfy` Text.isInfixOf "perm-1"
-        Left other -> expectationFailure ("unexpected halt reason: " <> show other)
-        Right () -> expectationFailure "expected ProcessorHalt from exhausted finalization"
+          messageId `shouldBe` MessageId "perm-1"
+        Left other -> expectationFailure ("unexpected processor failure: " <> show other)
+        Right () -> expectationFailure "expected ProcessorFailure from exhausted finalization"
 
       -- The other message was still attempted and finalized despite msg-1 failing.
       tracked <- runEff $ runTracingNoop $ getTrackedDecisions tracking
diff --git a/test/Shibuya/Runner/BatcherSpec.hs b/test/Shibuya/Runner/BatcherSpec.hs
--- a/test/Shibuya/Runner/BatcherSpec.hs
+++ b/test/Shibuya/Runner/BatcherSpec.hs
@@ -1,6 +1,7 @@
 module Shibuya.Runner.BatcherSpec (spec) where
 
 import Control.Concurrent (threadDelay)
+import Control.Concurrent.STM (atomically, retry)
 import Data.List (sort)
 import Data.List.NonEmpty qualified as NE
 import Data.Maybe (fromMaybe)
@@ -21,6 +22,7 @@
   ( ReadyBatch,
     emptyBatcherState,
     runBatcher,
+    runBatcherWithTickHook,
     stepArrival,
     stepFlush,
     stepTick,
@@ -29,6 +31,7 @@
 import Streamly.Data.Stream qualified as Stream
 import Test.Hspec
 import Test.QuickCheck
+import UnliftIO qualified as UIO
 
 -- The engine is parameterized by an effect stack and payload; the pure core
 -- treats them as phantom. We pick es = 'E' (an empty effect stack, kind pinned
@@ -178,6 +181,25 @@
       out <- Stream.fold Fold.toList (runBatcher 8 cfg slow)
       any (\(info, _) -> info.trigger == TriggerTimeout) out `shouldBe` True
       sort (batchIds out) `shouldBe` [MessageId "m-0", MessageId "m-1"]
+
+    it "propagates ticker failure while input remains idle" $ do
+      let cfg =
+            (partitionKeyConfig 100)
+              { batchTimeout = 1,
+                tickInterval = Just 0.01
+              }
+          idle = Stream.repeatM (atomically retry :: IO (Ingested E String))
+          failTick = ioError (userError "ticker failed")
+
+      result <-
+        UIO.timeout 1_000_000 $
+          UIO.tryAny $
+            Stream.fold Fold.drain (runBatcherWithTickHook failTick 8 cfg idle)
+
+      case result of
+        Just (Left failure) ->
+          Text.pack (UIO.displayException failure) `shouldSatisfy` Text.isInfixOf "ticker failed"
+        other -> expectationFailure $ "expected ticker failure, got: " <> show other
 
 -- QuickCheck generators and properties -------------------------------------
 
diff --git a/test/Shibuya/Runner/PartitionOrderingSpec.hs b/test/Shibuya/Runner/PartitionOrderingSpec.hs
--- a/test/Shibuya/Runner/PartitionOrderingSpec.hs
+++ b/test/Shibuya/Runner/PartitionOrderingSpec.hs
@@ -18,12 +18,15 @@
 import Shibuya.Core.Metrics (ProcessorId (..))
 import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))
 import Shibuya.Core.Types qualified as Core
+import Shibuya.Internal.Runner.KeyedScheduler (runKeyedScheduler)
 import Shibuya.Internal.Runner.Master (startMaster, stopMaster)
 import Shibuya.Internal.Runner.Supervised (SupervisedProcessor (..), runSupervised)
 import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))
 import Shibuya.Telemetry.Effect (runTracingNoop)
+import Streamly.Data.Stream qualified as Stream
 import Test.Hspec
 import Test.QuickCheck
+import UnliftIO qualified as UIO
 import UnliftIO.Concurrent (threadDelay)
 
 data Payload = Payload
@@ -55,28 +58,27 @@
 spec :: Spec
 spec = describe "Shibuya.Runner.PartitionOrdering" $ do
   it "finalizes each partition in arrival order and exactly once" $
-    property $
-      withMaxSuccess 30 $ \(PartitionCase messages concurrency) ->
-        ioProperty $ do
-          finalized <- runPartitioned messages concurrency
-          let ids = makeIds (length messages)
-              partitionById = Map.fromList (zip ids (map fst messages))
-              finalizedIds = map fst finalized
-              partitions = nub [p | Just p <- map fst messages]
-              finalizedFor p =
-                [ msgId
-                | msgId <- finalizedIds,
-                  Map.lookup msgId partitionById == Just (Just p)
-                ]
-              arrivedFor p =
-                [ msgId
-                | (msgId, Just p') <- zip ids (map fst messages),
-                  p == p'
-                ]
-          pure $
-            counterexample ("finalized=" <> show finalizedIds) $
-              sort finalizedIds === sort ids
-                .&&. conjoin [finalizedFor p === arrivedFor p | p <- partitions]
+    property $ \(PartitionCase messages concurrency) ->
+      ioProperty $ do
+        finalized <- runPartitioned messages concurrency
+        let ids = makeIds (length messages)
+            partitionById = Map.fromList (zip ids (map fst messages))
+            finalizedIds = map fst finalized
+            partitions = nub [p | Just p <- map fst messages]
+            finalizedFor p =
+              [ msgId
+              | msgId <- finalizedIds,
+                Map.lookup msgId partitionById == Just (Just p)
+              ]
+            arrivedFor p =
+              [ msgId
+              | (msgId, Just p') <- zip ids (map fst messages),
+                p == p'
+              ]
+        pure $
+          counterexample ("finalized=" <> show finalizedIds) $
+            sort finalizedIds === sort ids
+              .&&. conjoin [finalizedFor p === arrivedFor p | p <- partitions]
 
   it "respects the global concurrency bound" $ do
     maxInFlightRef <- newIORef (0 :: Int)
@@ -118,6 +120,26 @@
             (`elemIndexOrMax` orderedIds)
             ["msg-1", "msg-3", "msg-5"]
     all (< lastSlowIndex) fastIndexes `shouldBe` True
+
+  it "propagates a keyed worker failure without waiting for infinite input" $ do
+    successorsRef <- newIORef (0 :: Int)
+    let infiniteInput = Stream.unfoldrM (\n -> pure (Just (n, n + 1))) (0 :: Int)
+        worker n
+          | n == 0 = ioError (userError "keyed worker failed")
+          | otherwise = atomicModifyIORef' successorsRef (\count -> (count + 1, ()))
+
+    result <-
+      UIO.timeout 1_000_000 $
+        UIO.tryAny $
+          runKeyedScheduler 2 4 (Just . (`mod` 2)) worker infiniteInput
+
+    case result of
+      Just (Left failure) ->
+        Text.pack (UIO.displayException failure) `shouldSatisfy` Text.isInfixOf "keyed worker failed"
+      other -> expectationFailure $ "expected prompt keyed worker failure, got: " <> show other
+
+    successors <- readIORef successorsRef
+    successors `shouldSatisfy` (< 20)
 
 runPartitioned ::
   [(Maybe Text, Int)] ->
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
@@ -3,7 +3,8 @@
 module Shibuya.Runner.SupervisedSpec (spec) where
 
 import Control.Concurrent.NQE.Supervisor (Strategy (..))
-import Control.Concurrent.STM (atomically, check, readTVar, readTVarIO)
+import Control.Concurrent.STM (atomically, check, newEmptyTMVarIO, readTVar, readTVarIO, takeTMVar, tryPutTMVar)
+import Control.Exception (MaskingState (..), getMaskingState)
 import Control.Monad (forM, forM_, replicateM)
 import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, modifyIORef', newIORef, readIORef)
 import Data.Map.Strict qualified as Map
@@ -98,6 +99,25 @@
       processorMetrics `shouldBe` Just Nothing
 
   describe "Shibuya.Internal.Runner.Supervised" $ do
+    it "restores interruptibility inside a supervised child" $ do
+      observed <- newEmptyTMVarIO
+      result <-
+        UIO.timeout 1_000_000 $
+          runEff $
+            runTracingNoop $ do
+              messages <- createTestMessages 1
+              let handler _ = do
+                    state <- liftIO getMaskingState
+                    liftIO $ atomically $ tryPutTMVar observed state >> pure ()
+                    pure AckOk
+              master <- startMaster IgnoreAll
+              _ <- runSupervised master 1 (ProcessorId "masking-state") Unordered Serial (testAdapter messages) handler
+              state <- liftIO $ atomically $ takeTMVar observed
+              stopMaster master
+              pure state
+
+      result `shouldBe` Just Unmasked
+
     describe "runWithMetrics" $ do
       it "processes messages and tracks metrics" $ do
         (finalMetrics, processedMsgs) <- runEff $ runTracingNoop $ do
@@ -603,7 +623,7 @@
               threadDelay 50000 -- 50ms - should be in the middle of processing
               metrics <- sampleMetrics sp.metrics
               case metrics.state of
-                Processing info _ -> modifyIORef' maxInFlightObserved (max info.inFlight)
+                Processing info _ _ -> modifyIORef' maxInFlightObserved (max info.inFlight)
                 _ -> pure ()
 
             liftIO $ threadDelay 600000 -- 600ms to complete
@@ -630,7 +650,7 @@
               threadDelay 25000 -- 25ms
               metrics <- sampleMetrics sp.metrics
               case metrics.state of
-                Processing info _ -> pure $ Just info.maxConcurrency
+                Processing info _ _ -> pure $ Just info.maxConcurrency
                 _ -> pure Nothing
 
             liftIO $ threadDelay 300000
@@ -1054,7 +1074,7 @@
               liftIO $ threadDelay 100000 -- 100ms
 
               -- Graceful shutdown with generous timeout
-              let config = ShutdownConfig {drainTimeout = 5} -- 5 seconds
+              let config = ShutdownConfig {drainTimeout = 5, totalShutdownTimeout = 6} -- 5 seconds
               stopAppGracefully config appHandle
 
         drained `shouldBe` True
@@ -1086,7 +1106,7 @@
               liftIO $ threadDelay 100000 -- 100ms
 
               -- Very short timeout (0.3 seconds)
-              let config = ShutdownConfig {drainTimeout = 0.3}
+              let config = ShutdownConfig {drainTimeout = 0.3, totalShutdownTimeout = 1.3}
               stopAppGracefully config appHandle
 
         -- Should timeout (not all drained)
@@ -1110,7 +1130,7 @@
             Right appHandle -> do
               -- Give time to complete
               liftIO $ threadDelay 200000 -- 200ms
-              let config = ShutdownConfig {drainTimeout = 1}
+              let config = ShutdownConfig {drainTimeout = 1, totalShutdownTimeout = 2}
               stopAppGracefully config appHandle
 
         drained `shouldBe` True
diff --git a/test/Shibuya/RunnerSpec.hs b/test/Shibuya/RunnerSpec.hs
--- a/test/Shibuya/RunnerSpec.hs
+++ b/test/Shibuya/RunnerSpec.hs
@@ -8,7 +8,8 @@
 import Effectful (Eff, IOE, liftIO, runEff, (:>))
 import Shibuya.Adapter (Adapter (..))
 import Shibuya.Adapter.Mock (TrackingAck (..), newTrackingAck, trackingAckHandle)
-import Shibuya.App (AppConfig (..), AppError (..), QueueProcessor (..), defaultAppConfig, mkProcessor, runApp, stopApp, waitApp)
+import Shibuya.App (AppConfig (..), AppError (..), QueueProcessor (..), defaultAppConfig, mkBatchProcessor, mkProcessor, runApp, stopApp, waitApp)
+import Shibuya.Batch (ackAll, defaultBatchConfig)
 import Shibuya.Core.Ack (AckDecision (..))
 import Shibuya.Core.AckHandle (AckHandle (..))
 import Shibuya.Core.Error (ConfigError (..), PolicyError (..))
@@ -17,7 +18,7 @@
 import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), mkEnvelope)
 import Shibuya.Handler (Handler)
 import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))
-import Shibuya.Telemetry.Effect (runTracingNoop)
+import Shibuya.Telemetry.Effect (Tracing, runTracingNoop)
 import Streamly.Data.Stream qualified as Stream
 import Test.Hspec
 
@@ -46,6 +47,27 @@
         Left err -> expectationFailure $ "Expected AppConfigInvalid, got: " ++ show err
         Right _ -> expectationFailure "Expected config validation to fail"
 
+    describe "processor identity validation" $ do
+      it "rejects duplicate ordinary processor IDs before adapter acquisition" $ do
+        assertDuplicateRejected $ \adapter ->
+          [ (ProcessorId "duplicate", mkProcessor adapter alwaysAckOk),
+            (ProcessorId "duplicate", mkProcessor adapter alwaysAckOk)
+          ]
+
+      it "rejects duplicate batch processor IDs before adapter acquisition" $ do
+        assertDuplicateRejected $ \adapter ->
+          let batchHandler _ _ = pure (ackAll AckOk)
+           in [ (ProcessorId "duplicate", mkBatchProcessor adapter batchHandler defaultBatchConfig),
+                (ProcessorId "duplicate", mkBatchProcessor adapter batchHandler defaultBatchConfig)
+              ]
+
+      it "rejects duplicate mixed processor IDs before adapter acquisition" $ do
+        assertDuplicateRejected $ \adapter ->
+          let batchHandler _ _ = pure (ackAll AckOk)
+           in [ (ProcessorId "duplicate", mkProcessor adapter alwaysAckOk),
+                (ProcessorId "duplicate", mkBatchProcessor adapter batchHandler defaultBatchConfig)
+              ]
+
     it "processes messages from mock adapter" $ do
       result <- runEff $ runTracingNoop $ do
         -- Track processed messages
@@ -143,6 +165,25 @@
       result `shouldBe` Right ()
 
   describe "Policy validation" $ do
+    it "rejects nonpositive and overflowing concurrency before adapter acquisition" $ do
+      let overflow = maxBound `div` 2 + 1
+          batchHandler _ _ = pure (ackAll AckOk)
+      assertPolicyRejected
+        (InvalidConcurrency 0)
+        (\adapter -> QueueProcessor adapter alwaysAckOk Unordered (Ahead 0))
+      assertPolicyRejected
+        (InvalidConcurrency (-1))
+        (\adapter -> QueueProcessor adapter alwaysAckOk PartitionedInOrder (Async (-1)))
+      assertPolicyRejected
+        (InvalidConcurrency 0)
+        (\adapter -> (mkBatchProcessor adapter batchHandler defaultBatchConfig) {concurrency = Async 0})
+      assertPolicyRejected
+        (ConcurrencyCapacityOverflow overflow)
+        (\adapter -> QueueProcessor adapter alwaysAckOk Unordered (Async overflow))
+      assertPolicyRejected
+        (ConcurrencyCapacityOverflow overflow)
+        (\adapter -> (mkBatchProcessor adapter batchHandler defaultBatchConfig) {concurrency = Ahead overflow})
+
     it "rejects StrictInOrder with Async" $ do
       result <- runEff $ runTracingNoop $ do
         messages <- createTestMessages 3
@@ -267,3 +308,48 @@
 -- | Handler that always returns AckOk
 alwaysAckOk :: Handler es msg
 alwaysAckOk _ = pure AckOk
+
+assertDuplicateRejected ::
+  (Adapter '[Tracing, IOE] String -> [(ProcessorId, QueueProcessor '[Tracing, IOE])]) ->
+  Expectation
+assertDuplicateRejected mkProcessors = do
+  acquiredRef <- newIORef (0 :: Int)
+  result <- runEff $ runTracingNoop $ do
+    messages <- createTestMessages 1
+    let adapter =
+          (testAdapter messages)
+            { source =
+                Stream.mapM
+                  (\msg -> liftIO (modifyIORef' acquiredRef (+ 1)) >> pure msg)
+                  (Stream.fromList messages)
+            }
+    runApp defaultAppConfig (mkProcessors adapter)
+
+  case result of
+    Left (AppConfigInvalid (DuplicateProcessorId (ProcessorId "duplicate"))) -> pure ()
+    Left err -> expectationFailure $ "Expected duplicate-ID config error, got: " ++ show err
+    Right _ -> expectationFailure "Expected duplicate processor IDs to be rejected"
+  readIORef acquiredRef `shouldReturn` 0
+
+assertPolicyRejected ::
+  PolicyError ->
+  (Adapter '[Tracing, IOE] String -> QueueProcessor '[Tracing, IOE]) ->
+  Expectation
+assertPolicyRejected expectedError mkProcessorUnderTest = do
+  acquiredRef <- newIORef (0 :: Int)
+  result <- runEff $ runTracingNoop $ do
+    messages <- createTestMessages 1
+    let adapter =
+          (testAdapter messages)
+            { source =
+                Stream.mapM
+                  (\msg -> liftIO (modifyIORef' acquiredRef (+ 1)) >> pure msg)
+                  (Stream.fromList messages)
+            }
+    runApp defaultAppConfig [(ProcessorId "invalid-policy", mkProcessorUnderTest adapter)]
+
+  case result of
+    Left (AppPolicyError actualError) -> actualError `shouldBe` expectedError
+    Left err -> expectationFailure $ "Expected policy error, got: " ++ show err
+    Right _ -> expectationFailure "Expected concurrency policy to be rejected"
+  readIORef acquiredRef `shouldReturn` 0