diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,20 @@
 # Changelog
 
+## 0.13.0.1 — 2026-10-30
+
+### Bug Fixes
+
+- Preserve asynchronous cancellation through observability callbacks, publisher
+  recovery and initial notifier LISTEN setup, including concrete exception payloads.
+- Release the acquired notifier connection when startup LISTEN is cancelled;
+  retain synchronous callback suppression and publisher recovery behavior.
+
+### Other Changes
+
+- Export `withListenHookForTest` from `Kiroku.Store.Notification` for sequential,
+  process-local startup LISTEN injection. The previous hook is restored on exit;
+  production leaves it unset.
+
 ## 0.13.0.0 — 2026-10-28
 
 ### Breaking Changes
diff --git a/kiroku-store.cabal b/kiroku-store.cabal
--- a/kiroku-store.cabal
+++ b/kiroku-store.cabal
@@ -1,6 +1,6 @@
 cabal-version:   3.0
 name:            kiroku-store
-version:         0.13.0.0
+version:         0.13.0.1
 synopsis:        High-performance PostgreSQL event store
 description:
   Kiroku is a PostgreSQL-backed event store for Haskell applications. It
@@ -108,6 +108,7 @@
   hs-source-dirs:  test
   other-modules:
     Paths_kiroku_store
+    Test.AsyncCancellation
     Test.BrowseQueryPlans
     Test.BrowseReads
     Test.BrowseReadsMock
diff --git a/src/Kiroku/Store/Notification.hs b/src/Kiroku/Store/Notification.hs
--- a/src/Kiroku/Store/Notification.hs
+++ b/src/Kiroku/Store/Notification.hs
@@ -3,6 +3,7 @@
     NotifierStartError (..),
     startNotifier,
     stopNotifier,
+    withListenHookForTest,
     NamespaceRegistry,
     registerNamespace,
     notifyNamespace,
@@ -13,11 +14,12 @@
 import Control.Concurrent.Async (Async)
 import Control.Concurrent.Async qualified as Async
 import Control.Concurrent.STM (STM, TChan, TVar, atomically, modifyTVar', newBroadcastTChanIO, newTVar, newTVarIO, readTVar, readTVarIO, writeTChan, writeTVar)
-import Control.Exception (Exception, SomeAsyncException, SomeException, asyncExceptionFromException, bracketOnError, catch, throwIO, toException)
+import Control.Exception (Exception, SomeAsyncException, SomeException, asyncExceptionFromException, bracket, bracketOnError, catch, fromException, throwIO, toException)
 import Control.Monad (forM_, unless, void)
 import Control.Monad.IO.Class (MonadIO, liftIO)
 import Data.ByteString (ByteString)
 import Data.ByteString.Char8 qualified as BC
+import Data.IORef (IORef, newIORef, readIORef, writeIORef)
 import Data.Map.Strict (Map)
 import Data.Map.Strict qualified as Map
 import Data.Text (Text)
@@ -36,6 +38,7 @@
 import Hasql.Statement (Statement, unpreparable)
 import Kiroku.Store.Observability (KirokuEvent (..), emitOrDrop)
 import Kiroku.Store.Types (CategoryName (..), StreamName (..), categoryName)
+import System.IO.Unsafe (unsafePerformIO)
 
 -- | A Notifier manages a dedicated PostgreSQL connection for LISTEN/NOTIFY.
 -- It writes a @()@ tick to a broadcast 'TChan' on every notification,
@@ -128,6 +131,28 @@
   | n <= 0 = 1_000_000
   | otherwise = min maxReconnectDelayMicros (1_000_000 * 2 ^ (min (n - 1) 30))
 
+{-# NOINLINE listenHookRef #-}
+listenHookRef :: IORef (Maybe (IO () -> IO ()))
+listenHookRef = unsafePerformIO (newIORef Nothing)
+
+-- | Test-only process-local startup LISTEN hook. Use sequentially; the previous
+-- hook is restored even if the test throws. Production leaves it unset.
+withListenHookForTest :: (IO () -> IO ()) -> IO a -> IO a
+withListenHookForTest hook action =
+  bracket
+    ( do
+        previous <- readIORef listenHookRef
+        writeIORef listenHookRef (Just hook)
+        pure previous
+    )
+    (writeIORef listenHookRef)
+    (const action)
+
+startupListen :: IO () -> IO ()
+startupListen action = do
+  hook <- readIORef listenHookRef
+  maybe action ($ action) hook
+
 -- | Start a Notifier. Acquires a dedicated connection, sets
 -- @application_name@ to @kiroku-listener@ for operator visibility, issues
 -- LISTEN on the @\<schema\>.events@ channel, and spawns a thread that
@@ -148,7 +173,8 @@
 --
 -- On 'Async.AsyncCancelled' (from cancellation), exits cleanly.
 --
--- Initial acquire or LISTEN failure raises 'NotifierStartError'.
+-- Synchronous initial acquire or LISTEN failure raises 'NotifierStartError'.
+-- Asynchronous LISTEN exceptions propagate unchanged after releasing the connection.
 startNotifier ::
   (MonadIO m) =>
   -- | libpq connection string
@@ -164,10 +190,10 @@
   namespaces <- newTVarIO Map.empty
   let channel = toPgIdentifier (schema <> ".events")
   bracketOnError (acquireOrThrow connString) Connection.release $ \conn -> do
-    Notifications.listen conn channel
+    startupListen (Notifications.listen conn channel)
       `catch` \(e :: SomeException) ->
-        case asyncExceptionFromException e of
-          Just (ae :: SomeAsyncException) -> throwIO ae
+        case fromException e of
+          Just (_ :: SomeAsyncException) -> throwIO e
           Nothing -> throwIO (NotifierListenError e)
     connRef <- newTVarIO conn
     thread <- Async.async (listenerLoop chan catGenVar namespaces connRef channel connString mHandler)
diff --git a/src/Kiroku/Store/Observability.hs b/src/Kiroku/Store/Observability.hs
--- a/src/Kiroku/Store/Observability.hs
+++ b/src/Kiroku/Store/Observability.hs
@@ -51,7 +51,7 @@
   )
 where
 
-import Control.Exception (SomeAsyncException, SomeException, asyncExceptionFromException, catch, throwIO)
+import Control.Exception (SomeAsyncException, SomeException, catch, fromException, throwIO)
 import Data.Foldable (for_)
 import Data.Int (Int32, Int64)
 import Data.Time (NominalDiffTime)
@@ -246,8 +246,8 @@
 emitOrDrop mHandler evt =
   for_ mHandler $ \handler ->
     handler evt `catch` \(e :: SomeException) ->
-      case asyncExceptionFromException e of
-        Just (ae :: SomeAsyncException) -> throwIO ae
+      case fromException e of
+        Just (_ :: SomeAsyncException) -> throwIO e
         Nothing -> pure ()
 
 -- | Which database phase a 'KirokuEventSubscriptionDbError' fired in.
diff --git a/src/Kiroku/Store/Subscription/EventPublisher.hs b/src/Kiroku/Store/Subscription/EventPublisher.hs
--- a/src/Kiroku/Store/Subscription/EventPublisher.hs
+++ b/src/Kiroku/Store/Subscription/EventPublisher.hs
@@ -53,7 +53,7 @@
     writeTVar,
   )
 import Control.Concurrent.STM qualified as STM
-import Control.Exception (SomeAsyncException, SomeException, asyncExceptionFromException, catch, throwIO)
+import Control.Exception (SomeAsyncException, SomeException, catch, fromException, throwIO)
 import Control.Monad (when)
 import Control.Monad.IO.Class (MonadIO, liftIO)
 import Data.Foldable (for_)
@@ -232,8 +232,8 @@
       -- partial delivery, lastPublished is not advanced, so the next tick
       -- re-fetches under Kiroku's at-least-once delivery contract.
       fetchAndBroadcast `catch` \(e :: SomeException) ->
-        case asyncExceptionFromException e of
-          Just (ae :: SomeAsyncException) -> throwIO ae
+        case fromException e of
+          Just (_ :: SomeAsyncException) -> throwIO e
           Nothing -> emitOrDrop mHandler (KirokuEventPublisherLoopError e)
       loop
 
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -25,6 +25,7 @@
 import Kiroku.Store.Subscription.EventPublisher (publisherPosition)
 import Kiroku.Store.Subscription.Fsm qualified as Fsm
 import Kiroku.Test.Postgres (withMigratedTestDatabase)
+import Test.AsyncCancellation qualified as AsyncCancellation
 import Test.BrowseQueryPlans qualified as BrowseQueryPlans
 import Test.BrowseReads qualified as BrowseReads
 import Test.BrowseReadsMock qualified as BrowseReadsMock
@@ -138,6 +139,7 @@
     StreamNameLookup.noOpSpec
   CategoryIdleNoSpin.spec
   HandlerStall.spec
+  AsyncCancellation.spec
   PublisherCallbackResilience.spec
   PublisherDropCounter.spec
   PublisherIdleAdvance.spec
diff --git a/test/Test/AsyncCancellation.hs b/test/Test/AsyncCancellation.hs
new file mode 100644
--- /dev/null
+++ b/test/Test/AsyncCancellation.hs
@@ -0,0 +1,172 @@
+module Test.AsyncCancellation (spec, publisherSpec) where
+
+import Control.Concurrent (threadDelay)
+import Control.Concurrent.Async qualified as Async
+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar, tryPutMVar, tryReadMVar)
+import Control.Concurrent.STM (atomically)
+import Control.Exception (AsyncException (ThreadKilled, UserInterrupt), Exception (..), SomeException, asyncExceptionFromException, asyncExceptionToException, bracket, finally, throwIO, throwTo, try)
+import Control.Lens ((&), (.~), (^.))
+import Control.Monad (forM_, void)
+import Data.Aeson qualified as Aeson
+import Data.Generics.Labels ()
+import Data.IORef (newIORef, readIORef, writeIORef)
+import Data.Int (Int64)
+import Data.Text (Text)
+import Kiroku.Store
+import Kiroku.Store.Notification qualified as N
+import Kiroku.Store.Observability (emitOrDrop)
+import Kiroku.Store.Subscription.EventPublisher qualified as P
+import Kiroku.Test.Postgres (withMigratedTestDatabase)
+import System.Timeout (timeout)
+import Test.Helpers (listenerCount, makeEvent, withTestStoreSettings)
+import Test.Hspec
+
+-- The payload ensures rethrowing preserves a consumer-defined exception too.
+data PayloadCancellation = PayloadCancellation Int deriving stock (Eq, Show)
+
+instance Exception PayloadCancellation where
+  toException = asyncExceptionToException
+  fromException = asyncExceptionFromException
+
+data SyncFailure = SyncFailure deriving stock (Eq, Show)
+
+instance Exception SyncFailure
+
+cases :: [(String, SomeException, SomeException -> Bool)]
+cases =
+  [ ("AsyncCancelled", toException Async.AsyncCancelled, matches Async.AsyncCancelled),
+    ("ThreadKilled", toException ThreadKilled, matches ThreadKilled),
+    ("UserInterrupt", toException UserInterrupt, matches UserInterrupt),
+    ("custom payload", toException (PayloadCancellation 73), matches (PayloadCancellation 73))
+  ]
+  where
+    matches :: (Exception e, Eq e) => e -> SomeException -> Bool
+    matches expected e = fromException e == Just expected
+
+bounded :: IO a -> IO a
+bounded action = timeout 10_000_000 action >>= maybe (fail "asynchronous boundary timed out") pure
+
+expectException :: (SomeException -> Bool) -> Either SomeException a -> Expectation
+expectException matches = \case
+  Left e -> e `shouldSatisfy` matches
+  Right _ -> expectationFailure "expected asynchronous exception; action returned normally"
+
+emit :: IO () -> IO ()
+emit action = emitOrDrop (Just (const action)) KirokuEventNotifierReconnected
+
+-- Release the blocked action before the scoped async finalizer on every exit.
+-- This also makes the original, cancellation-swallowing implementation safe to test.
+crossThread :: (IO () -> IO ()) -> (Async.Async () -> IO ()) -> (SomeException -> Bool) -> IO ()
+crossThread boundary send matches = do
+  entered <- newEmptyMVar
+  release <- newEmptyMVar
+  Async.withAsync (boundary (putMVar entered () >> takeMVar release)) $ \target ->
+    ( do
+        bounded (takeMVar entered)
+        bounded (send target)
+        bounded (Async.waitCatch target) >>= expectException matches
+    )
+      `finally` void (tryPutMVar release ())
+
+awaitNoListeners :: Text -> Int64 -> IO ()
+awaitNoListeners conn baseline = bounded loop
+  where
+    loop = do
+      n <- listenerCount conn
+      if n == baseline then pure () else threadDelay 1_000 >> loop
+
+spec :: Spec
+spec = describe "asynchronous cancellation" $ do
+  it "allows absent and successful observation callbacks" $ do
+    emitOrDrop Nothing KirokuEventNotifierReconnected
+    called <- newIORef False
+    emit (writeIORef called True)
+    readIORef called `shouldReturn` True
+  it "suppresses synchronous observation failures" $ emit (throwIO SyncFailure)
+  forM_ cases $ \(label, exception, matches) -> do
+    it ("preserves observation " <> label) $
+      try (emit (throwIO exception)) >>= expectException matches
+    it ("preserves notifier startup " <> label <> " and releases its connection") $
+      withMigratedTestDatabase $ \conn -> do
+        baseline <- listenerCount conn
+        baseline `shouldBe` 0
+        N.withListenHookForTest
+          ( \_ -> do
+              listenerCount conn `shouldReturn` (baseline + 1)
+              throwIO exception
+          )
+          $ do
+            result <- try (N.startNotifier conn "kiroku" Nothing)
+            -- Even an incorrect successful startup must be cleaned up.
+            either (const (pure ())) N.stopNotifier result
+            expectException matches result
+        awaitNoListeners conn baseline
+  it "wraps synchronous notifier startup failure with its original cause and releases the connection" $
+    withMigratedTestDatabase $ \conn -> do
+      baseline <- listenerCount conn
+      N.withListenHookForTest (\_ -> throwIO SyncFailure) $ do
+        result <- try (N.startNotifier conn "kiroku" Nothing)
+        either (const (pure ())) N.stopNotifier result
+        case result of
+          Left e
+            | Just (N.NotifierListenError cause) <- fromException e ->
+                fromException cause `shouldBe` Just SyncFailure
+          _ -> expectationFailure "expected NotifierListenError with synchronous cause"
+      awaitNoListeners conn baseline
+  forM_ [False, True] $ \custom -> do
+    let send target = if custom then throwTo (Async.asyncThreadId target) (PayloadCancellation 73) else Async.cancel target
+        matches e = if custom then fromException e == Just (PayloadCancellation 73) else fromException e == Just Async.AsyncCancelled
+    it ("terminates a blocked observation callback, custom=" <> show custom) $
+      crossThread emit send matches
+    it ("terminates blocked notifier startup and releases its connection, custom=" <> show custom) $
+      withMigratedTestDatabase $ \conn -> do
+        baseline <- listenerCount conn
+        let boundary action =
+              N.withListenHookForTest (\_ -> (listenerCount conn `shouldReturn` (baseline + 1)) >> action) $
+                bracket (N.startNotifier conn "kiroku" Nothing) N.stopNotifier (const (pure ()))
+        crossThread boundary send matches
+        awaitNoListeners conn baseline
+
+publisherSpec :: Spec
+publisherSpec = describe "asynchronous publisher recovery" $ do
+  forM_ [False, True] $ \callback -> do
+    forM_ cases $ \(label, exception, matches) ->
+      it ("preserves " <> label <> ", callback=" <> show callback) $
+        publisherCase callback (throwIO exception) Nothing (pure ()) matches
+    forM_ [False, True] $ \custom ->
+      it ("terminates blocked action, callback=" <> show callback <> ", custom=" <> show custom) $ do
+        entered <- newEmptyMVar
+        release <- newEmptyMVar
+        let action = putMVar entered () >> takeMVar release
+            send target = do
+              bounded (takeMVar entered)
+              bounded (if custom then throwTo (Async.asyncThreadId target) (PayloadCancellation 73) else Async.cancel target)
+            matches e = if custom then fromException e == Just (PayloadCancellation 73) else fromException e == Just Async.AsyncCancelled
+        publisherCase callback action (Just send) (void (tryPutMVar release ())) matches
+
+publisherCase :: Bool -> IO () -> Maybe (Async.Async () -> IO ()) -> IO () -> (SomeException -> Bool) -> IO ()
+publisherCase callback action sender cleanup matches = do
+  loopError <- newEmptyMVar
+  let hook event =
+        if callback
+          then pure (Left (DecodeFailure (event ^. #eventId) "injected decode failure"))
+          else action >> pure (Right event)
+      observe event = case event of
+        KirokuEventPublisherDecodeFailed {} | callback -> action
+        KirokuEventPublisherLoopError {} -> void (tryPutMVar loopError ())
+        _ -> pure ()
+      tweak settings = settings & #storeSettings .~ defaultStoreSettings {decodeHook = Just hook} & #eventHandler .~ Just observe
+  withTestStoreSettings tweak $ \store -> do
+    let pub = store ^. #publisher
+        target = P.publisherThread pub
+    ( bracket (atomically (P.subscribePublisher pub 1 PauseAndResume)) (\(_, _, unsubscribe) -> unsubscribe) $ \_ -> do
+        Right _ <- runStoreIO store $ appendToStream (StreamName "async-boundary") NoStream [makeEvent "Cancel" (Aeson.object [])]
+        forM_ sender ($ target)
+        outcome <- bounded $ Async.race (takeMVar loopError) (Async.waitCatch target)
+        case outcome of
+          Left () -> expectationFailure "asynchronous exception became a publisher loop error"
+          Right result -> expectException matches result
+        tryReadMVar loopError `shouldReturn` Nothing
+        atomically (P.publisherPosition pub) `shouldReturn` GlobalPosition 0
+      )
+      `finally` cleanup
diff --git a/test/Test/PublisherCallbackResilience.hs b/test/Test/PublisherCallbackResilience.hs
--- a/test/Test/PublisherCallbackResilience.hs
+++ b/test/Test/PublisherCallbackResilience.hs
@@ -29,6 +29,7 @@
 import Kiroku.Store.SQL qualified as SQL
 import Kiroku.Store.Subscription.Effect qualified as SubEff
 import Kiroku.Store.Subscription.Fsm (SubscriptionState (..))
+import Test.AsyncCancellation qualified as AsyncCancellation
 import Test.Helpers (caughtUpEventHandler, makeEvent, validConsumerGroup, waitForPublisher, waitForSubscriptionLive, waitWithTimeout, withTestStoreSettings)
 import Test.Hspec
 
@@ -90,6 +91,7 @@
 
 spec :: Spec
 spec = describe "publisher callback resilience" $ do
+  AsyncCancellation.publisherSpec
   it "stops only the default subscriber while a sibling dead-letters and continues live" $ do
     defaultLive <- newEmptyMVar
     siblingLive <- newEmptyMVar
@@ -553,6 +555,8 @@
         Right (Right ()) -> pure ()
         Left timeout -> expectationFailure timeout
         Right (Left e) -> expectationFailure ("expected clean stop, got: " <> show e)
+      awaitCheckpoint store "publisher-decode-hook-resilience" 2
+      waitForPublisher store (GlobalPosition 2)
 
     seen <- reverse <$> readIORef delivered
     seen `shouldBe` [EventType "Boom", EventType "Ok"]