diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,29 @@
+# Changelog
+
+## 0.1.0.0 — 2026-02-24
+
+Initial release.
+
+### New Features
+
+- Multi-queue processing with `runApp` and `QueueProcessor` API
+- Backpressure via bounded inbox
+- AckHalt support to stop processing on halt decision
+- Serial, Ahead, and Async concurrent processing modes
+- Policy validation enforcement (StrictInOrder requires Serial)
+- Graceful shutdown with configurable drain timeout
+- NQE-based supervision via Master/Supervisor
+- OpenTelemetry tracing integration with W3C trace context propagation
+- Mock adapter for testing
+
+### Bug Fixes
+
+- Fix race conditions and incomplete cleanup in runner
+- Fix data races in concurrent test handlers using atomicModifyIORef'
+
+### Other Changes
+
+- Consolidate error handling with unified error types
+- Define own SupervisionStrategy type to decouple from NQE
+- Use registerDelay for cleaner timeout handling
+- Replace polling with STM blocking in waitApp
diff --git a/shibuya-core.cabal b/shibuya-core.cabal
new file mode 100644
--- /dev/null
+++ b/shibuya-core.cabal
@@ -0,0 +1,134 @@
+cabal-version: 3.14
+name: shibuya-core
+version: 0.1.0.0
+synopsis: Supervised queue processing framework for Haskell
+description:
+  A supervised queue processing framework inspired by Broadway (Elixir).
+  Provides unified queue abstraction, NQE-based supervision, backpressure
+  via bounded inbox, explicit ack semantics, and OpenTelemetry tracing.
+
+author: Nadeem Bitar
+maintainer: nadeem@gmail.com
+license: MIT
+build-type: Simple
+category: Concurrency
+extra-doc-files: CHANGELOG.md
+
+common warnings
+  ghc-options: -Wall
+
+library
+  import: warnings
+  exposed-modules:
+    Shibuya.Adapter
+    Shibuya.Adapter.Mock
+    Shibuya.App
+    Shibuya.Core
+    Shibuya.Core.Ack
+    Shibuya.Core.AckHandle
+    Shibuya.Core.Error
+    Shibuya.Core.Ingested
+    Shibuya.Core.Lease
+    Shibuya.Core.Types
+    Shibuya.Handler
+    Shibuya.Policy
+    Shibuya.Prelude
+    Shibuya.Runner.Master
+    Shibuya.Runner.Metrics
+    Shibuya.Runner.Supervised
+    Shibuya.Stream
+    Shibuya.Telemetry
+    Shibuya.Telemetry.Config
+    Shibuya.Telemetry.Effect
+    Shibuya.Telemetry.Propagation
+    Shibuya.Telemetry.Semantic
+
+  other-modules:
+    Shibuya.Runner.Halt
+    Shibuya.Runner.Ingester
+    Shibuya.Runner.Processor
+    Shibuya.Runner.Serial
+
+  default-extensions:
+    DeriveAnyClass
+    DerivingStrategies
+    DuplicateRecordFields
+    LambdaCase
+    NoFieldSelectors
+    OverloadedLabels
+    OverloadedRecordDot
+    OverloadedStrings
+    QuasiQuotes
+
+  build-depends:
+    aeson ^>=2.2,
+    base ^>=4.21.0.0,
+    bytestring ^>=0.12.2.0,
+    containers ^>=0.7,
+    effectful ^>=2.6.1.0,
+    effectful-core ^>=2.6.1.0,
+    generic-lens ^>=2.3.0.0,
+    hs-opentelemetry-api ^>=0.3,
+    hs-opentelemetry-propagator-w3c ^>=0.1,
+    lens ^>=5.3.5,
+    nqe ^>=0.6,
+    stm ^>=2.5,
+    streamly ^>=0.11,
+    streamly-core ^>=0.3,
+    text ^>=2.1.3,
+    time ^>=1.14,
+    unliftio ^>=0.2,
+    unordered-containers ^>=0.2,
+    uuid ^>=1.3,
+    vector ^>=0.13,
+
+  hs-source-dirs: src
+  default-language: GHC2024
+
+test-suite shibuya-core-test
+  import: warnings
+  default-language: GHC2024
+  type: exitcode-stdio-1.0
+  hs-source-dirs: test
+  main-is: Main.hs
+  default-extensions:
+    DeriveAnyClass
+    DerivingStrategies
+    DuplicateRecordFields
+    LambdaCase
+    NoFieldSelectors
+    OverloadedLabels
+    OverloadedRecordDot
+    OverloadedStrings
+    QuasiQuotes
+
+  ghc-options:
+    -threaded
+    -rtsopts
+    -with-rtsopts=-N
+
+  other-modules:
+    Shibuya.Core.AckSpec
+    Shibuya.Core.TypesSpec
+    Shibuya.PolicySpec
+    Shibuya.Runner.SupervisedSpec
+    Shibuya.RunnerSpec
+    Shibuya.Telemetry.EffectSpec
+    Shibuya.Telemetry.PropagationSpec
+
+  build-depends:
+    QuickCheck ^>=2.15,
+    base ^>=4.21.0.0,
+    bytestring,
+    effectful,
+    hs-opentelemetry-api,
+    hspec ^>=2.11,
+    memory,
+    nqe,
+    shibuya-core,
+    stm,
+    streamly,
+    streamly-core,
+    text,
+    time,
+    unliftio,
diff --git a/src/Shibuya/Adapter.hs b/src/Shibuya/Adapter.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Adapter.hs
@@ -0,0 +1,23 @@
+-- | Adapter API - library-author facing.
+-- Adapters bridge external systems to the framework.
+-- Adapter owns queue semantics; runner never touches offsets directly.
+module Shibuya.Adapter
+  ( Adapter (..),
+  )
+where
+
+import Data.Text (Text)
+import Effectful (Eff)
+import Shibuya.Core.Ingested (Ingested)
+import Streamly.Data.Stream (Stream)
+
+-- | Queue adapter interface.
+-- Provides a stream of ingested messages and shutdown capability.
+data Adapter es msg = Adapter
+  { -- | Name for logging/observability
+    adapterName :: !Text,
+    -- | Stream of leased messages
+    source :: Stream (Eff es) (Ingested es msg),
+    -- | Stop polling, release resources
+    shutdown :: Eff es ()
+  }
diff --git a/src/Shibuya/Adapter/Mock.hs b/src/Shibuya/Adapter/Mock.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Adapter/Mock.hs
@@ -0,0 +1,56 @@
+-- | Mock adapter for testing.
+-- Provides adapters that produce messages from in-memory sources.
+module Shibuya.Adapter.Mock
+  ( -- * Mock Adapters
+    listAdapter,
+
+    -- * Test Helpers
+    TrackingAck (..),
+    newTrackingAck,
+    trackingAckHandle,
+    getTrackedDecisions,
+  )
+where
+
+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)
+import Effectful (Eff, IOE, liftIO, (:>))
+import Shibuya.Adapter (Adapter (..))
+import Shibuya.Core.Ack (AckDecision)
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Ingested (Ingested)
+import Shibuya.Core.Types (MessageId (..))
+import Streamly.Data.Stream qualified as Stream
+
+-- | Create an adapter from a list of ingested messages.
+-- Useful for testing handlers with predetermined input.
+listAdapter :: (IOE :> es) => [Ingested es msg] -> Adapter es msg
+listAdapter msgs =
+  Adapter
+    { adapterName = "mock:list",
+      source = Stream.fromList msgs,
+      shutdown = pure ()
+    }
+
+-- | Tracking state for ack decisions.
+data TrackingAck = TrackingAck
+  { trackedDecisions :: IORef [(MessageId, AckDecision)]
+  }
+
+-- | Create an AckHandle that tracks all decisions made.
+-- Useful for testing that handlers make correct ack decisions.
+trackingAckHandle ::
+  (IOE :> es) =>
+  TrackingAck ->
+  MessageId ->
+  AckHandle es
+trackingAckHandle tracking msgId =
+  AckHandle $ \decision ->
+    liftIO $ modifyIORef' tracking.trackedDecisions ((msgId, decision) :)
+
+-- | Create a new TrackingAck.
+newTrackingAck :: (IOE :> es) => Eff es TrackingAck
+newTrackingAck = liftIO $ TrackingAck <$> newIORef []
+
+-- | Get all tracked decisions.
+getTrackedDecisions :: (IOE :> es) => TrackingAck -> Eff es [(MessageId, AckDecision)]
+getTrackedDecisions tracking = liftIO $ readIORef tracking.trackedDecisions
diff --git a/src/Shibuya/App.hs b/src/Shibuya/App.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/App.hs
@@ -0,0 +1,285 @@
+-- | Application entry point for running Shibuya queue processors.
+module Shibuya.App
+  ( -- * Running Processors
+    runApp,
+    QueueProcessor (..),
+    mkProcessor,
+    AppHandle (..),
+
+    -- * AppHandle Operations
+    getAppMetrics,
+    getAppMaster,
+    stopApp,
+    stopAppGracefully,
+    waitApp,
+
+    -- * Shutdown Configuration
+    ShutdownConfig (..),
+    defaultShutdownConfig,
+
+    -- * Supervision Strategy
+    SupervisionStrategy (..),
+
+    -- * Errors
+    AppError (..),
+
+    -- * Re-exports
+    ProcessorId (..),
+    ProcessorMetrics (..),
+  )
+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 Data.Foldable (traverse_)
+import Data.Map.Strict (Map)
+import Data.Map.Strict qualified as Map
+import Data.Text qualified as Text
+import Data.Time.Clock (NominalDiffTime)
+import Effectful (Eff, IOE, liftIO, (:>))
+import GHC.Generics (Generic)
+import Numeric.Natural (Natural)
+import Shibuya.Adapter (Adapter (..))
+import Shibuya.Core.Error (HandlerError (..), PolicyError (..), RuntimeError (..))
+import Shibuya.Handler (Handler)
+import Shibuya.Policy (Concurrency (..), Ordering (..), validatePolicy)
+import Shibuya.Runner.Master
+  ( Master,
+    getAllMetrics,
+    startMaster,
+    stopMaster,
+  )
+import Shibuya.Runner.Metrics
+  ( MetricsMap,
+    ProcessorId (..),
+    ProcessorMetrics (..),
+  )
+import Shibuya.Runner.Supervised
+  ( SupervisedProcessor (..),
+    runSupervised,
+  )
+import Shibuya.Telemetry.Effect (Tracing)
+import UnliftIO (SomeException, catch, displayException)
+import Prelude hiding (Ordering)
+
+--------------------------------------------------------------------------------
+-- Supervision Strategy
+--------------------------------------------------------------------------------
+
+-- | Supervision strategy for processor failures.
+--
+-- This is Shibuya's own type that maps to NQE's supervision strategies,
+-- decoupling users from the NQE library.
+data SupervisionStrategy
+  = -- | Ignore all child exits, keep running.
+    -- Failed processors are marked as Failed in metrics but don't affect others.
+    IgnoreFailures
+  | -- | Stop all processors if any fails.
+    -- A single processor failure triggers shutdown of all processors.
+    StopAllOnFailure
+  deriving stock (Eq, Show, Generic)
+
+-- | Convert Shibuya's strategy type to NQE's internal type.
+toNQEStrategy :: SupervisionStrategy -> NQE.Strategy
+toNQEStrategy = \case
+  IgnoreFailures -> NQE.IgnoreAll
+  StopAllOnFailure -> NQE.KillAll
+
+--------------------------------------------------------------------------------
+-- Shutdown Configuration
+--------------------------------------------------------------------------------
+
+-- | Configuration for graceful shutdown behavior.
+data ShutdownConfig = ShutdownConfig
+  { -- | Maximum time to wait for in-flight messages to drain.
+    -- After this timeout, remaining processors are forcefully stopped.
+    -- Default: 30 seconds.
+    drainTimeout :: !NominalDiffTime
+  }
+  deriving stock (Eq, Show, Generic)
+
+-- | Default shutdown configuration with 30 second drain timeout.
+defaultShutdownConfig :: ShutdownConfig
+defaultShutdownConfig = ShutdownConfig {drainTimeout = 30}
+
+--------------------------------------------------------------------------------
+-- Errors
+--------------------------------------------------------------------------------
+
+-- | Application errors.
+-- Uses structured error types from Shibuya.Core.Error.
+data AppError
+  = -- | Invalid policy configuration
+    AppPolicyError !PolicyError
+  | -- | Handler execution error
+    AppHandlerError !HandlerError
+  | -- | Runtime error
+    AppRuntimeError !RuntimeError
+  deriving stock (Eq, Show)
+
+-- | A queue processor pairs an adapter with its handler.
+-- The message type is existentially hidden, allowing heterogeneous queues.
+data QueueProcessor es where
+  QueueProcessor ::
+    { adapter :: Adapter es msg,
+      handler :: Handler es msg,
+      ordering :: Ordering,
+      concurrency :: Concurrency
+    } ->
+    QueueProcessor es
+
+-- | Convenience constructor with default policies (Unordered + Serial).
+-- Provides backward compatibility with existing code.
+mkProcessor :: Adapter es msg -> Handler es msg -> QueueProcessor es
+mkProcessor adapter handler = QueueProcessor adapter handler Unordered Serial
+
+-- | Handle for a running multi-queue application.
+-- Provides introspection and control over all processors.
+data AppHandle es = AppHandle
+  { -- | The master coordinator
+    master :: !Master,
+    -- | Map of processor IDs to their handles
+    processors :: !(Map ProcessorId (SupervisedProcessor, QueueProcessor es))
+  }
+
+-- | Run queue processors concurrently under NQE supervision.
+--
+-- Each processor runs independently. Returns immediately with a handle
+-- for introspection and control.
+--
+-- Example:
+--
+-- @
+-- result <- runApp IgnoreFailures 100
+--   [ ("orders", QueueProcessor ordersAdapter ordersHandler)
+--   , ("events", QueueProcessor eventsAdapter eventsHandler)
+--   ]
+-- @
+runApp ::
+  (IOE :> es, Tracing :> es) =>
+  -- | Supervision strategy
+  SupervisionStrategy ->
+  -- | Inbox size for backpressure
+  Int ->
+  -- | Named processors
+  [(ProcessorId, QueueProcessor es)] ->
+  Eff es (Either AppError (AppHandle es))
+runApp strategy inboxSize namedProcessors =
+  -- Validate all policies first
+  case validateAllPolicies namedProcessors of
+    Left err -> pure $ Left $ AppPolicyError err
+    Right () -> do
+      let nqeStrategy = toNQEStrategy strategy
+      catch
+        ( do
+            -- Start the master coordinator
+            master <- startMaster nqeStrategy
+
+            -- Spawn each processor under supervision
+            processors <- spawnProcessors master (fromIntegral inboxSize) namedProcessors
+
+            pure $
+              Right
+                AppHandle
+                  { master = master,
+                    processors = Map.fromList processors
+                  }
+        )
+        ( \(e :: SomeException) ->
+            pure $ Left $ AppRuntimeError $ SupervisorFailed $ Text.pack $ displayException e
+        )
+
+-- | Validate all processor policies before starting.
+validateAllPolicies :: [(ProcessorId, QueueProcessor es)] -> Either PolicyError ()
+validateAllPolicies = traverse_ validateOne
+  where
+    validateOne (_, QueueProcessor _ _ ord conc) = validatePolicy ord conc
+
+-- | Spawn all processors under supervision.
+spawnProcessors ::
+  (IOE :> es, Tracing :> es) =>
+  Master ->
+  Natural ->
+  [(ProcessorId, QueueProcessor es)] ->
+  Eff es [(ProcessorId, (SupervisedProcessor, QueueProcessor es))]
+spawnProcessors master inboxSize = traverse spawnOne
+  where
+    spawnOne (procId, qp@(QueueProcessor adapter handler _ordering concurrency)) = do
+      sp <- runSupervised master inboxSize procId concurrency adapter handler
+      pure (procId, (sp, qp))
+
+--------------------------------------------------------------------------------
+-- AppHandle Operations
+--------------------------------------------------------------------------------
+
+-- | Get metrics for all processors.
+getAppMetrics :: (IOE :> es) => AppHandle es -> Eff es MetricsMap
+getAppMetrics appHandle = getAllMetrics appHandle.master
+
+-- | Get the master handle for direct access.
+-- This is useful for integrating with the metrics server.
+getAppMaster :: AppHandle es -> Master
+getAppMaster appHandle = appHandle.master
+
+-- | Gracefully stop all processors with default configuration.
+-- Uses 'defaultShutdownConfig' (30 second drain timeout).
+-- For custom timeout, use 'stopAppGracefully'.
+stopApp :: (IOE :> es) => AppHandle es -> Eff es ()
+stopApp = void . stopAppGracefully defaultShutdownConfig
+
+-- | Gracefully stop all processors with configurable drain timeout.
+--
+-- Shutdown sequence:
+-- 1. Signal all adapters to stop producing (close source streams)
+-- 2. Wait for processors to drain in-flight messages (with timeout)
+-- 3. Force stop any remaining processors after timeout
+-- 4. Stop the master coordinator
+--
+-- 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)
+
+  -- 2. Wait for drain with timeout
+  let timeoutMicros = floor (config.drainTimeout * 1_000_000)
+  drained <- liftIO $ waitForDrainWithTimeout timeoutMicros (Map.elems appHandle.processors)
+
+  -- 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
+
+  -- 4. Stop master (cancels any remaining processors)
+  stopMaster appHandle.master
+
+  pure drained
+  where
+    shutdownAdapter (_, QueueProcessor adapter _ _ _) = adapter.shutdown
+
+-- | Wait for all processors to be done, with timeout.
+-- Returns True if all drained cleanly, False if timeout occurred.
+-- Note: Requires -threaded RTS for registerDelay to work properly.
+waitForDrainWithTimeout :: Int -> [(SupervisedProcessor, a)] -> IO Bool
+waitForDrainWithTimeout timeoutMicros processors = do
+  -- Create a timeout TVar that becomes True after the deadline
+  timeoutVar <- registerDelay timeoutMicros
+
+  -- Wait for either all done or timeout
+  atomically $
+    (allDone processors >> pure True)
+      `orElse` (readTVar timeoutVar >>= check >> pure False)
+  where
+    allDone :: [(SupervisedProcessor, a)] -> STM ()
+    allDone procs = forM_ procs $ \(sp, _) -> readTVar sp.done >>= check
+
+-- | Wait for all processors to complete.
+-- For infinite streams, this will block forever.
+-- Use 'stopApp' to gracefully terminate.
+--
+-- Uses STM to block efficiently until all processors are done,
+-- rather than polling.
+waitApp :: (IOE :> es) => AppHandle es -> Eff es ()
+waitApp appHandle = liftIO $ atomically $ do
+  -- Block until all done TVars are True
+  forM_ (Map.elems appHandle.processors) $ \(sp, _) ->
+    readTVar sp.done >>= check
diff --git a/src/Shibuya/Core.hs b/src/Shibuya/Core.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Core.hs
@@ -0,0 +1,94 @@
+-- | Shibuya Core - Public API
+--
+-- This module re-exports all public types and functions from the Shibuya framework.
+-- Import this module for application development.
+--
+-- Example:
+--
+-- @
+-- import Shibuya.Core
+--
+-- main = runEff $ do
+--   let processor = QueueProcessor myAdapter myHandler
+--   result <- runApp IgnoreFailures 100 [(ProcessorId "main", processor)]
+--   case result of
+--     Right handle -> waitApp handle
+--     Left err -> print err
+-- @
+module Shibuya.Core
+  ( -- * Message Types
+    MessageId (..),
+    Cursor (..),
+    Envelope (..),
+
+    -- * Ack Semantics
+    AckDecision (..),
+    RetryDelay (..),
+    DeadLetterReason (..),
+    HaltReason (..),
+
+    -- * Handler Types
+    AckHandle (..),
+    Lease (..),
+    Ingested (..),
+
+    -- * Handler
+    Handler,
+
+    -- * Adapter
+    Adapter (..),
+
+    -- * Policy
+    Ordering (..),
+    Concurrency (..),
+    validatePolicy,
+
+    -- * App
+    runApp,
+    AppError (..),
+    QueueProcessor (..),
+    mkProcessor,
+    AppHandle (..),
+    waitApp,
+    stopApp,
+    stopAppGracefully,
+    getAppMetrics,
+
+    -- * Shutdown Configuration
+    ShutdownConfig (..),
+    defaultShutdownConfig,
+
+    -- * Supervision Strategy
+    SupervisionStrategy (..),
+
+    -- * Errors
+    PolicyError (..),
+    HandlerError (..),
+    RuntimeError (..),
+
+    -- * Metrics
+    ProcessorId (..),
+    ProcessorState (..),
+    ProcessorMetrics (..),
+    StreamStats (..),
+    InFlightInfo (..),
+    MetricsMap,
+
+    -- * Halt
+    ProcessorHalt (..),
+  )
+where
+
+import Shibuya.Adapter (Adapter (..))
+import Shibuya.App (AppError (..), AppHandle (..), QueueProcessor (..), ShutdownConfig (..), SupervisionStrategy (..), defaultShutdownConfig, getAppMetrics, mkProcessor, runApp, stopApp, stopAppGracefully, waitApp)
+import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..), RetryDelay (..))
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Error (HandlerError (..), PolicyError (..), RuntimeError (..))
+import Shibuya.Core.Ingested (Ingested (..))
+import Shibuya.Core.Lease (Lease (..))
+import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))
+import Shibuya.Handler (Handler)
+import Shibuya.Policy (Concurrency (..), Ordering (..), validatePolicy)
+import Shibuya.Runner.Halt (ProcessorHalt (..))
+import Shibuya.Runner.Metrics (InFlightInfo (..), MetricsMap, ProcessorId (..), ProcessorMetrics (..), ProcessorState (..), StreamStats (..))
+import Prelude hiding (Ordering)
diff --git a/src/Shibuya/Core/Ack.hs b/src/Shibuya/Core/Ack.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Core/Ack.hs
@@ -0,0 +1,54 @@
+-- | Ack semantics for message processing.
+-- Handlers decide meaning, not mechanics.
+-- This is explicit to support halt-on-error for ordered streams.
+module Shibuya.Core.Ack
+  ( -- * Retry
+    RetryDelay (..),
+
+    -- * Dead Letter
+    DeadLetterReason (..),
+
+    -- * Halt
+    HaltReason (..),
+
+    -- * Handler Decision
+    AckDecision (..),
+  )
+where
+
+import Shibuya.Prelude
+
+-- | Delay before retry.
+newtype RetryDelay = RetryDelay {unRetryDelay :: NominalDiffTime}
+  deriving stock (Eq, Show)
+
+-- | Why a message is being dead-lettered.
+data DeadLetterReason
+  = -- | Message is permanently unprocessable
+    PoisonPill !Text
+  | -- | Message payload failed validation/parsing
+    InvalidPayload !Text
+  | -- | Retry limit exceeded
+    MaxRetriesExceeded
+  deriving stock (Eq, Show, Generic)
+
+-- | Why processing should halt.
+data HaltReason
+  = -- | Must stop to preserve ordering guarantees
+    HaltOrderedStream !Text
+  | -- | Unrecoverable error
+    HaltFatal !Text
+  deriving stock (Eq, Show, Generic)
+
+-- | Handler outcome (semantic, not mechanical).
+-- The handler returns this to express intent; the framework handles the mechanics.
+data AckDecision
+  = -- | Message processed successfully
+    AckOk
+  | -- | Retry after delay
+    AckRetry !RetryDelay
+  | -- | Move to dead letter queue
+    AckDeadLetter !DeadLetterReason
+  | -- | Stop processing
+    AckHalt !HaltReason
+  deriving stock (Eq, Show, Generic)
diff --git a/src/Shibuya/Core/AckHandle.hs b/src/Shibuya/Core/AckHandle.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Core/AckHandle.hs
@@ -0,0 +1,18 @@
+-- | AckHandle for mechanical acknowledgment.
+-- This is the Broadway.Acknowledger equivalent, but typed.
+-- Must be called exactly once; adapter enforces idempotency.
+module Shibuya.Core.AckHandle
+  ( AckHandle (..),
+  )
+where
+
+import Effectful (Eff)
+import Shibuya.Core.Ack (AckDecision)
+
+-- | Mechanical ack interface (adapter-provided).
+-- The handler's AckDecision is passed to finalize, which performs
+-- the actual commit/retry/dead-letter operation.
+newtype AckHandle es = AckHandle
+  { -- | Finalize the message with the given decision
+    finalize :: AckDecision -> Eff es ()
+  }
diff --git a/src/Shibuya/Core/Error.hs b/src/Shibuya/Core/Error.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Core/Error.hs
@@ -0,0 +1,54 @@
+-- | Unified error types for Shibuya.
+-- Provides structured errors with consistent representation across the library.
+module Shibuya.Core.Error
+  ( -- * Policy Errors
+    PolicyError (..),
+    policyErrorToText,
+
+    -- * Handler Errors
+    HandlerError (..),
+    handlerErrorToText,
+
+    -- * Runtime Errors
+    RuntimeError (..),
+    runtimeErrorToText,
+  )
+where
+
+import Shibuya.Prelude
+
+-- | Policy validation errors.
+data PolicyError
+  = -- | Invalid combination of ordering and concurrency
+    InvalidPolicyCombo !Text
+  deriving stock (Eq, Show, Generic)
+
+-- | Convert policy error to text for display.
+policyErrorToText :: PolicyError -> Text
+policyErrorToText (InvalidPolicyCombo msg) = msg
+
+-- | Handler execution errors.
+data HandlerError
+  = -- | Handler threw an exception
+    HandlerException !Text
+  | -- | Handler timed out
+    HandlerTimeout
+  deriving stock (Eq, Show, Generic)
+
+-- | Convert handler error to text for display.
+handlerErrorToText :: HandlerError -> Text
+handlerErrorToText (HandlerException msg) = msg
+handlerErrorToText HandlerTimeout = "Handler timed out"
+
+-- | Runtime errors during processing.
+data RuntimeError
+  = -- | Supervisor failed
+    SupervisorFailed !Text
+  | -- | Inbox overflow (backpressure failure)
+    InboxOverflow
+  deriving stock (Eq, Show, Generic)
+
+-- | Convert runtime error to text for display.
+runtimeErrorToText :: RuntimeError -> Text
+runtimeErrorToText (SupervisorFailed msg) = msg
+runtimeErrorToText InboxOverflow = "Inbox overflow"
diff --git a/src/Shibuya/Core/Ingested.hs b/src/Shibuya/Core/Ingested.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Core/Ingested.hs
@@ -0,0 +1,22 @@
+-- | Ingested message type - what handlers receive.
+-- Combines Broadway.Message + Acknowledger + optional lease.
+-- Exactly one thing flows through the system.
+module Shibuya.Core.Ingested
+  ( Ingested (..),
+  )
+where
+
+import Shibuya.Core.AckHandle (AckHandle)
+import Shibuya.Core.Lease (Lease)
+import Shibuya.Core.Types (Envelope)
+
+-- | What handlers receive for processing.
+-- Contains the message envelope, ack handle, and optional lease.
+data Ingested es msg = Ingested
+  { -- | Message metadata and payload
+    envelope :: !(Envelope msg),
+    -- | Handle for acknowledging the message
+    ack :: !(AckHandle es),
+    -- | Optional lease for visibility timeout extension
+    lease :: !(Maybe (Lease es))
+  }
diff --git a/src/Shibuya/Core/Lease.hs b/src/Shibuya/Core/Lease.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Core/Lease.hs
@@ -0,0 +1,20 @@
+-- | Lease types for temporary message ownership.
+-- Lease exists only to model temporary ownership (SQS visibility timeout, DB locks).
+-- Kafka adapters won't use this; SQS/Redis/DB queues will.
+module Shibuya.Core.Lease
+  ( Lease (..),
+  )
+where
+
+import Data.Text (Text)
+import Data.Time (NominalDiffTime)
+import Effectful (Eff)
+
+-- | Optional capability for sources with visibility/ownership.
+-- Allows handlers to extend the lease if processing takes longer than expected.
+data Lease es = Lease
+  { -- | Identifier for the lease
+    leaseId :: !Text,
+    -- | Extend the lease by the given duration
+    leaseExtend :: NominalDiffTime -> Eff es ()
+  }
diff --git a/src/Shibuya/Core/Types.hs b/src/Shibuya/Core/Types.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Core/Types.hs
@@ -0,0 +1,56 @@
+-- | Core types for the Shibuya framework.
+-- These types exist everywhere and should be extremely stable.
+-- No behavior, no effects, no policy - just identity + payload + metadata.
+module Shibuya.Core.Types
+  ( -- * Message Identity
+    MessageId (..),
+
+    -- * Cursor / Offset
+    Cursor (..),
+
+    -- * Message Envelope
+    Envelope (..),
+
+    -- * Trace Context
+    TraceHeaders,
+  )
+where
+
+import Data.ByteString (ByteString)
+import Data.String (IsString)
+import Shibuya.Prelude
+
+-- | Stable identity for idempotency & observability.
+-- Every message has a unique identifier.
+newtype MessageId = MessageId {unMessageId :: Text}
+  deriving stock (Eq, Ord, Show, Generic)
+  deriving newtype (IsString)
+
+-- | Optional cursor / offset / global position.
+-- Used to track position in ordered streams.
+data Cursor
+  = CursorInt !Int
+  | CursorText !Text
+  deriving stock (Eq, Ord, Show, Generic)
+
+-- | W3C Trace Context headers for distributed tracing.
+-- Contains traceparent and optionally tracestate headers.
+type TraceHeaders = [(ByteString, ByteString)]
+
+-- | Normalized message envelope (Broadway.Message equivalent).
+-- Contains message metadata plus the payload.
+data Envelope msg = Envelope
+  { -- | Unique message identifier
+    messageId :: !MessageId,
+    -- | Optional position/offset
+    cursor :: !(Maybe Cursor),
+    -- | Optional partition key (for Kafka-style queues)
+    partition :: !(Maybe Text),
+    -- | When the message was enqueued
+    enqueuedAt :: !(Maybe UTCTime),
+    -- | W3C trace context headers for distributed tracing
+    traceContext :: !(Maybe TraceHeaders),
+    -- | The actual message payload
+    payload :: !msg
+  }
+  deriving stock (Eq, Show, Functor, Generic)
diff --git a/src/Shibuya/Handler.hs b/src/Shibuya/Handler.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Handler.hs
@@ -0,0 +1,15 @@
+-- | Handler API - this is all application authors need to know.
+-- Handlers cannot ack directly, cannot influence concurrency,
+-- and express intent only via AckDecision.
+module Shibuya.Handler
+  ( Handler,
+  )
+where
+
+import Effectful (Eff)
+import Shibuya.Core.Ack (AckDecision)
+import Shibuya.Core.Ingested (Ingested)
+
+-- | Handler function type.
+-- Takes an ingested message and returns an ack decision.
+type Handler es msg = Ingested es msg -> Eff es AckDecision
diff --git a/src/Shibuya/Policy.hs b/src/Shibuya/Policy.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Policy.hs
@@ -0,0 +1,44 @@
+-- | Ordering and concurrency policies.
+-- Runner policy that maps ordering guarantees to concurrency constraints.
+module Shibuya.Policy
+  ( -- * Ordering
+    Ordering (..),
+
+    -- * Concurrency
+    Concurrency (..),
+
+    -- * Validation
+    validatePolicy,
+  )
+where
+
+import Shibuya.Core.Error (PolicyError (..))
+import Shibuya.Prelude
+import Prelude hiding (Ordering)
+
+-- | Message ordering guarantees.
+data Ordering
+  = -- | Event-sourced subscriptions - must be Serial
+    StrictInOrder
+  | -- | Kafka-style - parallel across partitions
+    PartitionedInOrder
+  | -- | No ordering guarantees
+    Unordered
+  deriving stock (Eq, Show, Generic)
+
+-- | Concurrency mode.
+data Concurrency
+  = -- | One message at a time
+    Serial
+  | -- | Prefetch N, process in order
+    Ahead !Int
+  | -- | Process N concurrently
+    Async !Int
+  deriving stock (Eq, Show, Generic)
+
+-- | Validate policy combinations.
+-- Invariant: StrictInOrder => Serial
+validatePolicy :: Ordering -> Concurrency -> Either PolicyError ()
+validatePolicy StrictInOrder (Ahead _) = Left $ InvalidPolicyCombo "StrictInOrder requires Serial concurrency"
+validatePolicy StrictInOrder (Async _) = Left $ InvalidPolicyCombo "StrictInOrder requires Serial concurrency"
+validatePolicy _ _ = Right ()
diff --git a/src/Shibuya/Prelude.hs b/src/Shibuya/Prelude.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Prelude.hs
@@ -0,0 +1,30 @@
+{-# LANGUAGE PackageImports #-}
+
+module Shibuya.Prelude
+  ( -- base
+    Generic,
+    MonadIO,
+    Natural,
+    -- text
+    Text,
+    -- time
+    UTCTime,
+    Day,
+    LocalTime,
+    NominalDiffTime,
+    getCurrentTime,
+    -- lens
+    module Control.Lens,
+    -- vector
+    Vector,
+  )
+where
+
+import "base" Control.Monad.IO.Class (MonadIO)
+import "base" GHC.Generics (Generic)
+import "base" Numeric.Natural (Natural)
+import "generic-lens" Data.Generics.Labels ()
+import "lens" Control.Lens
+import "text" Data.Text (Text)
+import "time" Data.Time (Day, LocalTime, NominalDiffTime, UTCTime, getCurrentTime)
+import "vector" Data.Vector (Vector)
diff --git a/src/Shibuya/Runner/Halt.hs b/src/Shibuya/Runner/Halt.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Runner/Halt.hs
@@ -0,0 +1,19 @@
+-- | Halt exception for processor termination.
+-- Thrown when a handler returns AckHalt to stop processing.
+module Shibuya.Runner.Halt
+  ( ProcessorHalt (..),
+  )
+where
+
+import Control.Exception (Exception)
+import Shibuya.Core.Ack (HaltReason)
+import Shibuya.Prelude
+
+-- | Exception thrown when processing should halt.
+-- The supervisor catches this to handle graceful shutdown.
+data ProcessorHalt = ProcessorHalt
+  { reason :: !HaltReason
+  }
+  deriving stock (Show, Generic)
+
+instance Exception ProcessorHalt
diff --git a/src/Shibuya/Runner/Ingester.hs b/src/Shibuya/Runner/Ingester.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Runner/Ingester.hs
@@ -0,0 +1,60 @@
+-- | Ingester - reads from adapter stream and sends to inbox.
+-- Provides backpressure via bounded inbox.
+module Shibuya.Runner.Ingester
+  ( runIngester,
+    runIngesterWithMetrics,
+  )
+where
+
+import Control.Concurrent.NQE.Process (Inbox, inboxToMailbox, send)
+import Control.Concurrent.STM (TVar, atomically, modifyTVar')
+import Effectful (Eff, IOE, liftIO, (:>))
+import Shibuya.Core.Ingested (Ingested)
+import Shibuya.Runner.Metrics (ProcessorMetrics (..), incReceived)
+import Streamly.Data.Fold qualified as Fold
+import Streamly.Data.Stream (Stream)
+import Streamly.Data.Stream qualified as Stream
+
+-- | Run the ingester: reads from stream, sends to inbox.
+-- Sends each element to the inbox with backpressure - if inbox is full, blocks.
+-- Completes when the stream completes.
+runIngester ::
+  (IOE :> es) =>
+  -- | Source stream from adapter
+  Stream (Eff es) (Ingested es msg) ->
+  -- | Target inbox (bounded for backpressure)
+  Inbox (Ingested es msg) ->
+  Eff es ()
+runIngester source inbox = do
+  let mailbox = inboxToMailbox inbox
+  -- Send each stream element to the mailbox, then drain
+  -- This provides backpressure: if mailbox is full (bounded), send blocks
+  Stream.fold Fold.drain $
+    Stream.mapM
+      (\msg -> send msg mailbox >> pure msg)
+      source
+
+-- | Run the ingester with metrics tracking.
+-- Increments 'received' count for each message sent to inbox.
+runIngesterWithMetrics ::
+  (IOE :> es) =>
+  -- | Metrics TVar (for updating received count)
+  TVar ProcessorMetrics ->
+  -- | Source stream from adapter
+  Stream (Eff es) (Ingested es msg) ->
+  -- | Target inbox (bounded for backpressure)
+  Inbox (Ingested es msg) ->
+  Eff es ()
+runIngesterWithMetrics metricsVar source inbox = do
+  let mailbox = inboxToMailbox inbox
+  Stream.fold Fold.drain $
+    Stream.mapM
+      ( \msg -> do
+          -- Increment received count
+          liftIO $ atomically $ modifyTVar' metricsVar $ \m ->
+            m {stats = incReceived m.stats}
+          -- Send to inbox (blocks if full - backpressure)
+          liftIO $ send msg mailbox
+          pure msg
+      )
+      source
diff --git a/src/Shibuya/Runner/Master.hs b/src/Shibuya/Runner/Master.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Runner/Master.hs
@@ -0,0 +1,195 @@
+-- | Master process - central coordinator for queue processors.
+-- Provides supervision, metrics collection, and control API.
+--
+-- Architecture:
+-- - Master is an NQE Process that handles control messages
+-- - Holds a Supervisor for managing child processors
+-- - Maintains TVar MetricsMap for O(1) metrics access
+-- - Processors register their metrics TVars with the Master
+module Shibuya.Runner.Master
+  ( -- * Master Handle
+    Master (..),
+    MasterState (..),
+
+    -- * Control Messages
+    MasterMessage (..),
+
+    -- * Starting the Master
+    startMaster,
+    stopMaster,
+
+    -- * Introspection
+    getAllMetrics,
+    getAllMetricsIO,
+    getProcessorMetrics,
+    getProcessorMetricsIO,
+
+    -- * Processor Management
+    registerProcessor,
+    unregisterProcessor,
+  )
+where
+
+import Control.Concurrent.NQE.Process
+  ( Inbox,
+    Listen,
+    Process (..),
+    newInbox,
+    query,
+    receive,
+  )
+import Control.Concurrent.NQE.Supervisor (Strategy (..), Supervisor)
+import Control.Concurrent.NQE.Supervisor qualified as Supervisor
+import Control.Concurrent.STM
+  ( TVar,
+    atomically,
+    modifyTVar',
+    newTVarIO,
+    readTVar,
+  )
+import Control.Monad (forever)
+import Data.Map.Strict (Map)
+import Data.Map.Strict qualified as Map
+import Effectful (Eff, IOE, liftIO, (:>))
+import Shibuya.Prelude
+import Shibuya.Runner.Metrics
+  ( MetricsMap,
+    ProcessorId,
+    ProcessorMetrics,
+  )
+import UnliftIO (Async, async, cancel, link)
+
+-- | Messages for the master process.
+data MasterMessage
+  = -- | Get metrics for all processors
+    GetAllMetrics !(Listen MetricsMap)
+  | -- | Get metrics for a specific processor
+    GetProcessorMetrics ProcessorId !(Listen (Maybe ProcessorMetrics))
+  | -- | Register a processor's metrics TVar
+    RegisterProcessor !ProcessorId !(TVar ProcessorMetrics) !(Listen ())
+  | -- | Unregister a processor
+    UnregisterProcessor !ProcessorId !(Listen ())
+  | -- | Shutdown all processors
+    Shutdown !(Listen ())
+
+-- | Master state held in TVars.
+data MasterState = MasterState
+  { -- | Map of processor IDs to their metrics TVars
+    metrics :: !(TVar (Map ProcessorId (TVar ProcessorMetrics))),
+    -- | The supervisor managing child processors
+    supervisor :: !Supervisor
+  }
+  deriving (Generic)
+
+-- | Master handle - provides access to the master process.
+data Master = Master
+  { -- | The async handle for the master
+    handle :: !(Async ()),
+    -- | Direct access to master state
+    state :: !MasterState,
+    -- | Inbox for sending messages
+    inbox :: !(Inbox MasterMessage)
+  }
+  deriving (Generic)
+
+-- | Start the master process.
+-- Returns a handle for interacting with the master.
+-- The caller is responsible for calling stopMaster when done.
+startMaster :: (IOE :> es) => Strategy -> Eff es Master
+startMaster strategy = liftIO $ do
+  -- Create supervisor
+  sup <- Supervisor.supervisor strategy
+
+  metricsMapVar <- newTVarIO Map.empty
+  let masterState = MasterState metricsMapVar sup
+
+  masterInbox <- newInbox
+
+  -- Start master loop
+  masterHandle <- async $ masterLoop masterState masterInbox
+  link masterHandle
+
+  pure
+    Master
+      { handle = masterHandle,
+        state = masterState,
+        inbox = masterInbox
+      }
+
+-- | Stop the master and all child processors.
+-- Cancels the supervisor first (which cancels all children via NQE's stopAll),
+-- then cancels the master message loop.
+stopMaster :: (IOE :> es) => Master -> Eff es ()
+stopMaster master = liftIO $ do
+  -- Cancel the supervisor first - this triggers NQE's stopAll which cancels all children
+  cancel (getProcessAsync master.state.supervisor)
+  -- Then cancel the master message loop
+  cancel (master ^. #handle)
+
+-- | The master process main loop.
+masterLoop :: MasterState -> Inbox MasterMessage -> IO ()
+masterLoop state inbox = forever $ do
+  msg <- receive inbox
+  handleMessage state msg
+
+-- | Handle a single master message.
+handleMessage :: MasterState -> MasterMessage -> IO ()
+handleMessage state msg = case msg of
+  GetAllMetrics respond -> do
+    -- Read all metrics from their TVars
+    metricsMap <- atomically $ do
+      tvarsMap <- readTVar state.metrics
+      traverse readTVar tvarsMap
+    atomically $ respond metricsMap
+  GetProcessorMetrics pid respond -> do
+    result <- atomically $ do
+      tvarsMap <- readTVar state.metrics
+      case Map.lookup pid tvarsMap of
+        Nothing -> pure Nothing
+        Just tvar -> Just <$> readTVar tvar
+    atomically $ respond result
+  RegisterProcessor pid metricsTVar respond -> do
+    atomically $ do
+      modifyTVar' state.metrics $ Map.insert pid metricsTVar
+    atomically $ respond ()
+  UnregisterProcessor pid respond -> do
+    atomically $ do
+      modifyTVar' state.metrics $ Map.delete pid
+    atomically $ respond ()
+  Shutdown respond -> do
+    -- Clear all processors
+    atomically $ modifyTVar' state.metrics (const Map.empty)
+    atomically $ respond ()
+
+-- | Get metrics for all processors.
+getAllMetrics :: (IOE :> es) => Master -> Eff es MetricsMap
+getAllMetrics master =
+  liftIO $
+    GetAllMetrics `query` master.inbox
+
+-- | Get metrics for all processors (IO version for web servers).
+getAllMetricsIO :: Master -> IO MetricsMap
+getAllMetricsIO master = GetAllMetrics `query` master.inbox
+
+-- | Get metrics for a specific processor.
+getProcessorMetrics :: (IOE :> es) => Master -> ProcessorId -> Eff es (Maybe ProcessorMetrics)
+getProcessorMetrics master pid =
+  liftIO $
+    GetProcessorMetrics pid `query` master.inbox
+
+-- | Get metrics for a specific processor (IO version for web servers).
+getProcessorMetricsIO :: Master -> ProcessorId -> IO (Maybe ProcessorMetrics)
+getProcessorMetricsIO master pid = GetProcessorMetrics pid `query` master.inbox
+
+-- | Register a processor with the master.
+-- The processor should call this with its metrics TVar.
+registerProcessor :: (IOE :> es) => Master -> ProcessorId -> TVar ProcessorMetrics -> Eff es ()
+registerProcessor master pid metricsTVar =
+  liftIO $
+    RegisterProcessor pid metricsTVar `query` master.inbox
+
+-- | Unregister a processor from the master.
+unregisterProcessor :: (IOE :> es) => Master -> ProcessorId -> Eff es ()
+unregisterProcessor master pid =
+  liftIO $
+    UnregisterProcessor pid `query` master.inbox
diff --git a/src/Shibuya/Runner/Metrics.hs b/src/Shibuya/Runner/Metrics.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Runner/Metrics.hs
@@ -0,0 +1,167 @@
+-- | Metrics and state tracking for processors.
+-- Provides introspection into what's happening in the system.
+module Shibuya.Runner.Metrics
+  ( -- * Processor State
+    ProcessorState (..),
+    ProcessorId (..),
+
+    -- * In-Flight Tracking
+    InFlightInfo (..),
+    emptyInFlightInfo,
+
+    -- * Stream Statistics
+    StreamStats (..),
+    emptyStreamStats,
+
+    -- * Combined Metrics
+    ProcessorMetrics (..),
+    emptyProcessorMetrics,
+
+    -- * Metrics Map
+    MetricsMap,
+
+    -- * Metrics Updates
+    incReceived,
+    incDropped,
+    incProcessed,
+    incFailed,
+  )
+where
+
+import Data.Aeson (FromJSON (..), FromJSONKey (..), ToJSON (..), ToJSONKey (..), object, withObject, (.:))
+import Data.Aeson qualified as Aeson
+import Data.Map.Strict (Map)
+import Data.Text qualified as Text
+import Shibuya.Prelude
+
+-- | Processor identifier.
+newtype ProcessorId = ProcessorId {unProcessorId :: Text}
+  deriving stock (Eq, Ord, Show, Generic)
+  deriving newtype (ToJSON, FromJSON, ToJSONKey, FromJSONKey)
+
+-- | Tracks concurrent in-flight messages.
+data InFlightInfo = InFlightInfo
+  { -- | Currently processing count
+    inFlight :: !Int,
+    -- | Configured max concurrency (1 for Serial)
+    maxConcurrency :: !Int
+  }
+  deriving stock (Eq, Show, Generic)
+
+instance ToJSON InFlightInfo where
+  toJSON info =
+    object
+      [ "inFlight" Aeson..= info.inFlight,
+        "maxConcurrency" Aeson..= info.maxConcurrency
+      ]
+
+instance FromJSON InFlightInfo where
+  parseJSON = withObject "InFlightInfo" $ \v ->
+    InFlightInfo
+      <$> v .: "inFlight"
+      <*> v .: "maxConcurrency"
+
+-- | Create empty in-flight info with given max concurrency.
+emptyInFlightInfo :: Int -> InFlightInfo
+emptyInFlightInfo = InFlightInfo 0
+
+-- | Processor runtime state.
+data ProcessorState
+  = -- | Waiting for messages
+    Idle
+  | -- | Currently processing (in-flight info, last activity time)
+    Processing !InFlightInfo !UTCTime
+  | -- | Failed with error (error message, timestamp)
+    Failed !Text !UTCTime
+  | -- | Processor has been stopped
+    Stopped
+  deriving stock (Eq, Show, Generic)
+
+instance ToJSON ProcessorState where
+  toJSON Idle = object ["status" Aeson..= ("idle" :: Text)]
+  toJSON (Processing info lastActivity) =
+    object
+      [ "status" Aeson..= ("processing" :: Text),
+        "inFlight" Aeson..= info.inFlight,
+        "maxConcurrency" Aeson..= info.maxConcurrency,
+        "lastActivity" Aeson..= lastActivity
+      ]
+  toJSON (Failed err timestamp) =
+    object
+      [ "status" Aeson..= ("failed" :: Text),
+        "error" Aeson..= err,
+        "timestamp" Aeson..= timestamp
+      ]
+  toJSON Stopped = object ["status" Aeson..= ("stopped" :: Text)]
+
+instance FromJSON ProcessorState where
+  parseJSON = withObject "ProcessorState" $ \v -> do
+    status <- v .: "status"
+    case status :: Text of
+      "idle" -> pure Idle
+      "processing" -> do
+        inFlightCount <- v .: "inFlight"
+        maxConc <- v .: "maxConcurrency"
+        lastActivity <- v .: "lastActivity"
+        pure $ Processing (InFlightInfo inFlightCount maxConc) lastActivity
+      "failed" -> Failed <$> v .: "error" <*> v .: "timestamp"
+      "stopped" -> pure Stopped
+      other -> fail $ "Unknown processor state: " <> Text.unpack other
+
+-- | Stream statistics.
+data StreamStats = StreamStats
+  { -- | Messages received from stream
+    received :: !Int,
+    -- | Messages dropped (backpressure)
+    dropped :: !Int,
+    -- | Messages successfully processed
+    processed :: !Int,
+    -- | Messages that failed processing
+    failed :: !Int
+  }
+  deriving stock (Eq, Show, Generic)
+  deriving anyclass (ToJSON, FromJSON)
+
+-- | Empty stream stats.
+emptyStreamStats :: StreamStats
+emptyStreamStats = StreamStats 0 0 0 0
+
+-- | Combined processor metrics.
+data ProcessorMetrics = ProcessorMetrics
+  { -- | Current state
+    state :: !ProcessorState,
+    -- | Statistics
+    stats :: !StreamStats,
+    -- | When the processor started
+    startedAt :: !UTCTime
+  }
+  deriving stock (Eq, Show, Generic)
+  deriving anyclass (ToJSON, FromJSON)
+
+-- | Empty processor metrics.
+emptyProcessorMetrics :: UTCTime -> ProcessorMetrics
+emptyProcessorMetrics now =
+  ProcessorMetrics
+    { state = Idle,
+      stats = emptyStreamStats,
+      startedAt = now
+    }
+
+-- | Map of processor IDs to their metrics.
+type MetricsMap = Map ProcessorId ProcessorMetrics
+
+-- | Increment received count.
+incReceived :: StreamStats -> StreamStats
+incReceived = #received %~ (+ 1)
+
+-- | Increment dropped count.
+incDropped :: StreamStats -> StreamStats
+incDropped = #dropped %~ (+ 1)
+
+-- | Increment processed count.
+incProcessed :: StreamStats -> StreamStats
+incProcessed = #processed %~ (+ 1)
+
+-- | Increment failed count.
+incFailed :: StreamStats -> StreamStats
+incFailed = #failed %~ (+ 1)
diff --git a/src/Shibuya/Runner/Processor.hs b/src/Shibuya/Runner/Processor.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Runner/Processor.hs
@@ -0,0 +1,78 @@
+-- | Processor - receives from inbox, calls handler, finalizes ack.
+-- Runs in a loop until terminated.
+module Shibuya.Runner.Processor
+  ( runProcessor,
+    runProcessorN,
+    drainInbox,
+  )
+where
+
+import Control.Concurrent.NQE.Process (Inbox, mailboxEmpty, receive)
+import Control.Monad (forever, unless)
+import Effectful (Eff, IOE, (:>))
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Ingested (Ingested (..))
+import Shibuya.Handler (Handler)
+
+-- | Run the processor forever: receive message, call handler, finalize ack.
+-- This runs until an exception is thrown or the thread is cancelled.
+runProcessor ::
+  (IOE :> es) =>
+  -- | Message handler
+  Handler es msg ->
+  -- | Source inbox
+  Inbox (Ingested es msg) ->
+  Eff es ()
+runProcessor handler inbox = forever $ processOne handler inbox
+
+-- | Run the processor for exactly N messages.
+-- Useful for testing.
+runProcessorN ::
+  (IOE :> es) =>
+  -- | Number of messages to process
+  Int ->
+  -- | Message handler
+  Handler es msg ->
+  -- | Source inbox
+  Inbox (Ingested es msg) ->
+  Eff es ()
+runProcessorN n handler inbox = go n
+  where
+    go 0 = pure ()
+    go remaining = do
+      processOne handler inbox
+      go (remaining - 1)
+
+-- | Process a single message: receive, handle, finalize.
+processOne ::
+  (IOE :> es) =>
+  Handler es msg ->
+  Inbox (Ingested es msg) ->
+  Eff es ()
+processOne handler inbox = do
+  -- Receive next message (blocks if inbox empty)
+  ingested <- receive inbox
+
+  -- Call the handler to get the ack decision
+  decision <- handler ingested
+
+  -- Finalize the ack (commit, retry, dead-letter, or halt)
+  -- Note: AckHalt is handled by the supervised runner (Supervised.hs throws ProcessorHalt)
+  ingested.ack.finalize decision
+
+-- | Drain all remaining messages from inbox until empty.
+-- Used after ingester completes to process buffered messages.
+drainInbox ::
+  (IOE :> es) =>
+  -- | Message handler
+  Handler es msg ->
+  -- | Source inbox
+  Inbox (Ingested es msg) ->
+  Eff es ()
+drainInbox handler inbox = go
+  where
+    go = do
+      empty <- mailboxEmpty inbox
+      unless empty $ do
+        processOne handler inbox
+        go
diff --git a/src/Shibuya/Runner/Serial.hs b/src/Shibuya/Runner/Serial.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Runner/Serial.hs
@@ -0,0 +1,44 @@
+-- | Serial Runner - processes messages one at a time.
+-- Simple sequential runner for ordered processing.
+module Shibuya.Runner.Serial
+  ( runSerial,
+  )
+where
+
+import Effectful (Eff, IOE, (:>))
+import Numeric.Natural (Natural)
+import Shibuya.Adapter (Adapter (..))
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Ingested (Ingested (..))
+import Shibuya.Handler (Handler)
+import Streamly.Data.Fold qualified as Fold
+import Streamly.Data.Stream qualified as Stream
+
+-- | Run the serial runner.
+-- Processes each message from the adapter stream sequentially:
+-- 1. Pull message from stream
+-- 2. Call handler
+-- 3. Finalize ack
+-- 4. Repeat until stream is exhausted
+--
+-- This is the simplest runner - no concurrency, no backpressure buffering.
+-- Suitable for ordered, low-throughput workloads.
+runSerial ::
+  (IOE :> es) =>
+  -- | Inbox size (unused in serial mode, for API compatibility)
+  Natural ->
+  -- | Queue adapter
+  Adapter es msg ->
+  -- | Message handler
+  Handler es msg ->
+  Eff es ()
+runSerial _inboxSize adapter handler = do
+  -- Process each message from the stream sequentially
+  Stream.fold Fold.drain $
+    Stream.mapM processOne adapter.source
+  where
+    processOne ingested = do
+      -- Call the handler to get the ack decision
+      decision <- handler ingested
+      -- Finalize the ack
+      ingested.ack.finalize decision
diff --git a/src/Shibuya/Runner/Supervised.hs b/src/Shibuya/Runner/Supervised.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Runner/Supervised.hs
@@ -0,0 +1,488 @@
+-- | Supervised runner - runs processors under NQE supervision with metrics.
+-- This is the production runner with introspection and control.
+--
+-- Architecture:
+-- - Each processor runs as a child under the Master's supervisor
+-- - Metrics are tracked in a TVar and registered with the Master
+-- - Backpressure is provided via bounded inbox between stream and processor
+module Shibuya.Runner.Supervised
+  ( -- * Running with Supervision
+    runSupervised,
+
+    -- * Standalone (without Master)
+    runWithMetrics,
+
+    -- * Processor Handle
+    SupervisedProcessor (..),
+
+    -- * Introspection
+    getMetrics,
+    getProcessorState,
+    isDone,
+  )
+where
+
+import Control.Concurrent.NQE.Process (Inbox, mailboxEmpty, mailboxEmptySTM, newBoundedInbox, receive, receiveSTM)
+import Control.Concurrent.NQE.Supervisor (addChild)
+import Control.Concurrent.STM
+  ( TVar,
+    atomically,
+    modifyTVar',
+    newTVarIO,
+    orElse,
+    readTVar,
+    readTVarIO,
+    retry,
+    writeTVar,
+  )
+import Control.Monad (unless)
+import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef)
+import Data.Text qualified as Text
+import Effectful (Eff, IOE, liftIO, withEffToIO, (:>))
+import Effectful.Dispatch.Static (unsafeEff_)
+import Effectful.Internal.Unlift (Limit (..), Persistence (..), UnliftStrategy (..))
+import OpenTelemetry.Trace.Core qualified as OTel
+import Shibuya.Adapter (Adapter (..))
+import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..))
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Error (HandlerError (..), handlerErrorToText)
+import Shibuya.Core.Ingested (Ingested (..))
+import Shibuya.Core.Types (Envelope (..), MessageId (..))
+import Shibuya.Handler (Handler)
+import Shibuya.Policy (Concurrency (..))
+import Shibuya.Prelude
+import Shibuya.Runner.Halt (ProcessorHalt (..))
+import Shibuya.Runner.Ingester (runIngesterWithMetrics)
+import Shibuya.Runner.Master (Master (..), MasterState (..), registerProcessor, unregisterProcessor)
+import Shibuya.Runner.Metrics
+  ( InFlightInfo (..),
+    ProcessorId (..),
+    ProcessorMetrics (..),
+    ProcessorState (..),
+    emptyProcessorMetrics,
+    incFailed,
+    incProcessed,
+  )
+import Shibuya.Telemetry.Effect
+  ( Tracing,
+    addAttribute,
+    addEvent,
+    recordException,
+    setStatus,
+    withExtractedContext,
+    withSpan',
+  )
+import Shibuya.Telemetry.Propagation (extractTraceContext)
+import Shibuya.Telemetry.Semantic
+  ( attrMessagingDestinationPartitionId,
+    attrMessagingMessageId,
+    attrMessagingSystem,
+    attrShibuyaAckDecision,
+    attrShibuyaInflightCount,
+    attrShibuyaInflightMax,
+    consumerSpanArgs,
+    eventAckDecision,
+    eventHandlerCompleted,
+    eventHandlerException,
+    eventHandlerStarted,
+    mkEvent,
+    processMessageSpanName,
+  )
+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, finally, throwIO)
+import UnliftIO qualified as UIO
+
+-- | Handle for a supervised processor.
+-- Provides introspection into the running processor.
+data SupervisedProcessor = SupervisedProcessor
+  { -- | Live metrics for this processor
+    metrics :: !(TVar ProcessorMetrics),
+    -- | The processor's ID
+    processorId :: !ProcessorId,
+    -- | Whether processing is complete
+    done :: !(TVar Bool),
+    -- | The async handle if running under supervision
+    child :: !(Maybe (Async ()))
+  }
+
+-- | Get current metrics for the processor.
+getMetrics :: (IOE :> es) => SupervisedProcessor -> Eff es ProcessorMetrics
+getMetrics sp = liftIO $ readTVarIO sp.metrics
+
+-- | Get current state of the processor.
+getProcessorState :: (IOE :> es) => SupervisedProcessor -> Eff es ProcessorState
+getProcessorState sp = liftIO $ (.state) <$> readTVarIO sp.metrics
+
+-- | Check if processing is done.
+isDone :: (IOE :> es) => SupervisedProcessor -> Eff es Bool
+isDone sp = liftIO $ readTVarIO sp.done
+
+-- | Run a processor under the Master's supervision with metrics tracking.
+--
+-- Architecture:
+-- 1. Creates a bounded inbox (using inboxSize for backpressure)
+-- 2. Spawns ingester async (reads from adapter, sends to inbox)
+-- 3. Runs processor loop (receives from inbox, calls handler)
+-- 4. Registers metrics with Master, unregisters on completion
+--
+-- Returns immediately with a handle for introspection.
+runSupervised ::
+  (IOE :> es, Tracing :> es) =>
+  Master ->
+  -- | Inbox size (for backpressure)
+  Natural ->
+  -- | Processor identifier
+  ProcessorId ->
+  -- | Concurrency mode
+  Concurrency ->
+  -- | Queue adapter
+  Adapter es msg ->
+  -- | Message handler
+  Handler es msg ->
+  Eff es SupervisedProcessor
+runSupervised master inboxSize procId concurrency adapter handler = do
+  now <- liftIO getCurrentTime
+
+  -- Initialize state
+  let initialMetrics = emptyProcessorMetrics now
+  metricsVar <- liftIO $ newTVarIO initialMetrics
+  doneVar <- liftIO $ newTVarIO False
+
+  -- Register with Master
+  registerProcessor master procId metricsVar
+
+  -- Add as supervised child using NQE's Supervisor
+  -- ConcUnlift Persistent allows the runInIO function to be used in the async child
+  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 metricsVar doneVar inboxSize concurrency adapter handler
+            `catch` \(ProcessorHalt _) -> pure () -- Convert halt to graceful exit
+        )
+          `finally` unregisterProcessor master procId
+
+  -- Link so exceptions propagate to the parent
+  unsafeEff_ $ UIO.link supervisedChild
+
+  pure
+    SupervisedProcessor
+      { metrics = metricsVar,
+        processorId = procId,
+        done = doneVar,
+        child = Just supervisedChild
+      }
+
+-- | Run a processor with metrics but without Master supervision.
+-- Useful for testing or simple single-processor setups.
+-- This blocks until the stream is exhausted and all messages are processed.
+runWithMetrics ::
+  (IOE :> es, Tracing :> es) =>
+  -- | Inbox size (for backpressure)
+  Natural ->
+  -- | Processor identifier
+  ProcessorId ->
+  -- | Queue adapter
+  Adapter es msg ->
+  -- | Message handler
+  Handler es msg ->
+  Eff es SupervisedProcessor
+runWithMetrics inboxSize procId adapter handler = do
+  now <- liftIO getCurrentTime
+
+  -- Initialize state
+  let initialMetrics = emptyProcessorMetrics now
+  metricsVar <- liftIO $ newTVarIO initialMetrics
+  doneVar <- liftIO $ newTVarIO False
+
+  -- Create bounded inbox
+  inbox <- liftIO $ newBoundedInbox inboxSize
+
+  -- Run ingester to completion (all messages sent to inbox)
+  runIngesterWithMetrics metricsVar adapter.source inbox
+
+  -- Drain remaining messages from inbox
+  drainInboxWithMetrics metricsVar handler inbox
+
+  -- Mark done
+  liftIO $ atomically $ writeTVar doneVar True
+
+  pure
+    SupervisedProcessor
+      { metrics = metricsVar,
+        processorId = procId,
+        done = doneVar,
+        child = Nothing
+      }
+
+-- | Run ingester and processor with a bounded inbox.
+-- Ingester reads from adapter stream, processor calls handler.
+-- When stream exhausts, processor drains remaining messages and exits.
+runIngesterAndProcessor ::
+  (IOE :> es, Tracing :> es) =>
+  TVar ProcessorMetrics ->
+  TVar Bool ->
+  Natural ->
+  Concurrency ->
+  Adapter es msg ->
+  Handler es msg ->
+  Eff es ()
+runIngesterAndProcessor metricsVar doneVar inboxSize concurrency adapter handler = do
+  -- Create bounded inbox (this is where inboxSize is used for backpressure)
+  inbox <- liftIO $ newBoundedInbox inboxSize
+
+  -- Signal when ingester completes (stream exhausted)
+  streamDoneVar <- liftIO $ newTVarIO False
+
+  -- Run ingester async, processor in main thread
+  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
+    let ingesterWithSignal =
+          runInIO (runIngesterWithMetrics metricsVar adapter.source inbox)
+            `finally` atomically (writeTVar streamDoneVar True)
+
+    UIO.withAsync ingesterWithSignal $ \_ ->
+      -- Processor: process messages, exit when stream done and inbox empty
+      runInIO $ processUntilDrained metricsVar concurrency handler inbox streamDoneVar
+
+  -- Mark done when processor exits
+  liftIO $ atomically $ writeTVar doneVar True
+
+-- | Convert inbox to a stream for use with streamly.
+-- Respects both the stream-done signal and halt flag.
+--
+-- Uses STM to atomically check done/empty and receive, avoiding a race
+-- condition where the processor could block on receive after the stream
+-- has completed but before the done flag was checked.
+inboxToStream ::
+  Inbox (Ingested es msg) ->
+  TVar Bool ->
+  IORef (Maybe HaltReason) ->
+  Stream.Stream IO (Ingested es msg)
+inboxToStream inbox streamDoneVar haltRef = 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
+        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
+                )
+          pure $ fmap (,()) result
+
+-- | Process messages from inbox until stream is done and inbox is empty.
+-- Supports Serial, Ahead, and Async concurrency modes.
+processUntilDrained ::
+  (IOE :> es, Tracing :> es) =>
+  TVar ProcessorMetrics ->
+  Concurrency ->
+  Handler es msg ->
+  Inbox (Ingested es msg) ->
+  TVar Bool ->
+  Eff es ()
+processUntilDrained metricsVar concurrency handler inbox streamDoneVar = do
+  haltRef <- liftIO $ newIORef Nothing
+
+  let maxConc = case concurrency of
+        Serial -> 1
+        Ahead n -> n
+        Async n -> n
+
+  withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do
+    let inboxStream = inboxToStream inbox streamDoneVar haltRef
+        processAction = runInIO . processOne metricsVar maxConc haltRef handler
+
+    case concurrency of
+      Serial ->
+        Stream.fold Fold.drain $
+          Stream.mapM processAction inboxStream
+      Ahead n ->
+        Stream.fold Fold.drain $
+          StreamP.parMapM (StreamP.maxBuffer n . StreamP.ordered True) processAction inboxStream
+      Async n ->
+        Stream.fold Fold.drain $
+          StreamP.parMapM (StreamP.maxBuffer n) processAction inboxStream
+
+    -- After draining, check if we halted
+    maybeHalt <- readIORef haltRef
+    case maybeHalt of
+      Just reason -> throwIO $ ProcessorHalt reason
+      Nothing -> pure ()
+
+-- | Drain inbox until empty, with metrics tracking.
+-- Used for testing with finite streams.
+drainInboxWithMetrics ::
+  (IOE :> es, Tracing :> es) =>
+  TVar ProcessorMetrics ->
+  Handler es msg ->
+  Inbox (Ingested es msg) ->
+  Eff es ()
+drainInboxWithMetrics metricsVar handler inbox = do
+  haltRef <- liftIO $ newIORef Nothing
+  go haltRef
+  where
+    go haltRef = do
+      empty <- liftIO $ mailboxEmpty inbox
+      unless empty $ do
+        ingested <- liftIO $ receive inbox
+        processOne metricsVar 1 haltRef handler ingested
+        -- Check if halted
+        halted <- liftIO $ readIORef haltRef
+        case halted of
+          Just reason -> throwIO $ ProcessorHalt reason
+          Nothing -> go haltRef
+
+-- | Process a single message with metrics tracking and tracing.
+-- Thread-safe for concurrent execution.
+processOne ::
+  (IOE :> es, Tracing :> es) =>
+  TVar ProcessorMetrics ->
+  Int ->
+  IORef (Maybe HaltReason) ->
+  Handler es msg ->
+  Ingested es msg ->
+  Eff es ()
+processOne metricsVar maxConc haltRef handler ingested = do
+  -- Extract parent context from message headers for distributed tracing
+  let parentCtx = ingested.envelope.traceContext >>= extractTraceContext
+
+  withExtractedContext parentCtx $
+    withSpan' processMessageSpanName consumerSpanArgs $ \traceSpan -> do
+      -- Add messaging attributes
+      let MessageId msgIdText = ingested.envelope.messageId
+      addAttribute traceSpan attrMessagingSystem ("shibuya" :: Text)
+      addAttribute traceSpan attrMessagingMessageId msgIdText
+
+      -- Add partition if present
+      case ingested.envelope.partition of
+        Just p -> addAttribute traceSpan attrMessagingDestinationPartitionId p
+        Nothing -> pure ()
+
+      -- Increment in-flight and add inflight attributes
+      now <- liftIO getCurrentTime
+      currentInflight <- liftIO $ atomically $ do
+        modifyTVar' metricsVar $ \m ->
+          let current = case m.state of
+                Processing info _ -> info.inFlight
+                _ -> 0
+           in m & #state .~ Processing (InFlightInfo (current + 1) maxConc) now
+        m <- readTVar metricsVar
+        pure $ case m.state of
+          Processing info _ -> info.inFlight
+          _ -> 1
+
+      addAttribute traceSpan attrShibuyaInflightCount currentInflight
+      addAttribute traceSpan attrShibuyaInflightMax maxConc
+
+      -- Record handler start event
+      addEvent traceSpan (mkEvent eventHandlerStarted [])
+
+      -- Call handler and finalize
+      result <-
+        catchAny
+          ( do
+              decision <- handler ingested
+              ingested.ack.finalize decision
+              pure (Right decision)
+          )
+          ( \ex -> do
+              recordException traceSpan ex
+              addEvent traceSpan (mkEvent eventHandlerException [])
+              pure $ Left $ HandlerException $ Text.pack $ show ex
+          )
+
+      -- Record completion event and set status
+      case result of
+        Right decision -> do
+          let decisionText = showAckDecision decision
+          addEvent traceSpan $
+            mkEvent
+              eventHandlerCompleted
+              [(attrShibuyaAckDecision, OTel.toAttribute decisionText)]
+          addAttribute traceSpan attrShibuyaAckDecision decisionText
+
+          -- Set span status based on decision
+          case decision of
+            AckOk -> setStatus traceSpan OTel.Ok
+            AckRetry _ -> setStatus traceSpan OTel.Ok
+            AckDeadLetter reason ->
+              setStatus traceSpan $ OTel.Error $ showDeadLetterReason reason
+            AckHalt reason ->
+              setStatus traceSpan $ OTel.Error $ showHaltReason reason
+        Left err -> do
+          addEvent traceSpan $
+            mkEvent
+              eventAckDecision
+              [(attrShibuyaAckDecision, OTel.toAttribute ("error" :: Text))]
+          setStatus traceSpan $ OTel.Error $ handlerErrorToText err
+
+      -- Decrement in-flight and update stats
+      now' <- liftIO getCurrentTime
+      liftIO $ atomically $ modifyTVar' metricsVar $ decrementAndUpdate result now'
+
+      -- Handle halt (set flag, don't throw - let stream drain)
+      case result of
+        Right (AckHalt reason) -> liftIO $ atomicWriteIORef haltRef (Just reason)
+        _ -> pure ()
+  where
+    showAckDecision :: AckDecision -> Text
+    showAckDecision AckOk = "ack_ok"
+    showAckDecision (AckRetry _) = "ack_retry"
+    showAckDecision (AckDeadLetter _) = "ack_dead_letter"
+    showAckDecision (AckHalt _) = "ack_halt"
+
+    showDeadLetterReason :: DeadLetterReason -> Text
+    showDeadLetterReason (PoisonPill t) = "poison_pill: " <> t
+    showDeadLetterReason (InvalidPayload t) = "invalid_payload: " <> t
+    showDeadLetterReason MaxRetriesExceeded = "max_retries_exceeded"
+
+    showHaltReason :: HaltReason -> Text
+    showHaltReason (HaltOrderedStream t) = "halt_ordered_stream: " <> t
+    showHaltReason (HaltFatal t) = "halt_fatal: " <> t
+
+-- | Decrement in-flight count and update stats based on result.
+decrementAndUpdate ::
+  Either HandlerError AckDecision ->
+  UTCTime ->
+  ProcessorMetrics ->
+  ProcessorMetrics
+decrementAndUpdate result now m =
+  let newState = case m.state of
+        Processing info _ ->
+          if info.inFlight <= 1
+            then Idle
+            else Processing (info {inFlight = info.inFlight - 1}) now
+        other -> other
+      newStats = case result of
+        Right AckOk -> incProcessed m.stats
+        Right (AckRetry _) -> incProcessed m.stats
+        Right (AckDeadLetter _) -> incFailed m.stats
+        Right (AckHalt _) -> m.stats -- Mark as failed with halt message
+        Left _ -> incFailed m.stats
+      finalState = case result of
+        Right (AckHalt reason) -> Failed (haltReasonText reason) now
+        Left err -> Failed (handlerErrorToText err) now
+        _ -> newState
+   in m {state = finalState, stats = newStats}
+  where
+    haltReasonText (HaltOrderedStream t) = t
+    haltReasonText (HaltFatal t) = t
diff --git a/src/Shibuya/Stream.hs b/src/Shibuya/Stream.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Stream.hs
@@ -0,0 +1,42 @@
+-- | Stream utilities for adapters.
+-- Common stream transformations for use with adapters.
+module Shibuya.Stream
+  ( -- * Stream Sources
+    pollingStream,
+
+    -- * Transformations
+    batchStream,
+    filterStream,
+  )
+where
+
+import Control.Concurrent (threadDelay)
+import Effectful (Eff, IOE, liftIO, (:>))
+import Streamly.Data.Fold qualified as Fold
+import Streamly.Data.Stream (Stream)
+import Streamly.Data.Stream qualified as Stream
+
+-- | Create a polling stream from a poll action.
+-- Polls at the given interval, yielding messages when available.
+pollingStream ::
+  (IOE :> es) =>
+  -- | Poll interval in microseconds
+  Int ->
+  -- | Poll action
+  Eff es (Maybe msg) ->
+  Stream (Eff es) msg
+pollingStream interval poll =
+  Stream.catMaybes $
+    Stream.repeatM $ do
+      mmsg <- poll
+      case mmsg of
+        Nothing -> liftIO (threadDelay interval) >> pure Nothing
+        Just msg -> pure (Just msg)
+
+-- | Batch messages into chunks.
+batchStream :: (Monad m) => Int -> Stream m msg -> Stream m [msg]
+batchStream n = Stream.foldMany (Fold.take n Fold.toList)
+
+-- | Filter messages.
+filterStream :: (Monad m) => (msg -> Bool) -> Stream m msg -> Stream m msg
+filterStream = Stream.filter
diff --git a/src/Shibuya/Telemetry.hs b/src/Shibuya/Telemetry.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Telemetry.hs
@@ -0,0 +1,42 @@
+-- | OpenTelemetry tracing support for Shibuya.
+--
+-- This module re-exports all telemetry-related types and functions.
+--
+-- == Quick Start
+--
+-- 1. Enable tracing in your config:
+--
+-- @
+-- config = defaultAppConfig
+--   { tracing = defaultTracingConfig { enabled = True }
+--   }
+-- @
+--
+-- 2. Set environment variables for the OTLP exporter:
+--
+-- @
+-- export OTEL_SERVICE_NAME="my-service"
+-- export OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4317"
+-- @
+--
+-- 3. Spans are automatically created for message processing.
+--    See "Shibuya.Telemetry.Effect" for custom instrumentation.
+module Shibuya.Telemetry
+  ( -- * Configuration
+    module Shibuya.Telemetry.Config,
+
+    -- * Effect
+    module Shibuya.Telemetry.Effect,
+
+    -- * Context Propagation
+    module Shibuya.Telemetry.Propagation,
+
+    -- * Semantic Conventions
+    module Shibuya.Telemetry.Semantic,
+  )
+where
+
+import Shibuya.Telemetry.Config
+import Shibuya.Telemetry.Effect
+import Shibuya.Telemetry.Propagation
+import Shibuya.Telemetry.Semantic
diff --git a/src/Shibuya/Telemetry/Config.hs b/src/Shibuya/Telemetry/Config.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Telemetry/Config.hs
@@ -0,0 +1,33 @@
+-- | Tracing configuration for Shibuya.
+-- Most configuration is handled via OTEL_* environment variables
+-- by the hs-opentelemetry SDK. This module provides the minimal
+-- application-level config.
+module Shibuya.Telemetry.Config
+  ( TracingConfig (..),
+    defaultTracingConfig,
+  )
+where
+
+import Data.Text (Text)
+import GHC.Generics (Generic)
+
+-- | Tracing configuration.
+-- The SDK reads most settings from OTEL_* environment variables.
+data TracingConfig = TracingConfig
+  { -- | Master switch for tracing. When False, runTracingNoop is used.
+    enabled :: !Bool,
+    -- | Service name (fallback if OTEL_SERVICE_NAME not set)
+    serviceName :: !Text,
+    -- | Service version for span attributes
+    serviceVersion :: !(Maybe Text)
+  }
+  deriving stock (Eq, Show, Generic)
+
+-- | Default tracing config with tracing disabled.
+defaultTracingConfig :: TracingConfig
+defaultTracingConfig =
+  TracingConfig
+    { enabled = False,
+      serviceName = "shibuya",
+      serviceVersion = Nothing
+    }
diff --git a/src/Shibuya/Telemetry/Effect.hs b/src/Shibuya/Telemetry/Effect.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Telemetry/Effect.hs
@@ -0,0 +1,312 @@
+{-# LANGUAGE TypeFamilies #-}
+
+-- | Tracing effect for OpenTelemetry integration.
+-- Provides a simple, low-overhead tracing API that wraps hs-opentelemetry.
+module Shibuya.Telemetry.Effect
+  ( -- * Effect
+    Tracing,
+
+    -- * Runners
+    runTracing,
+    runTracingNoop,
+
+    -- * Span Operations
+    withSpan,
+    withSpan',
+
+    -- * Span Modification
+    addAttribute,
+    addAttributes,
+    addEvent,
+    recordException,
+    setStatus,
+
+    -- * Context
+    getTracer,
+    isTracingEnabled,
+    withExtractedContext,
+
+    -- * Re-exports
+    OTel.Span,
+    OTel.SpanArguments (..),
+    OTel.SpanKind (..),
+    OTel.SpanStatus (..),
+    OTel.NewEvent (..),
+    OTel.Tracer,
+    OTel.defaultSpanArguments,
+    OTel.toAttribute,
+  )
+where
+
+import Control.Exception (Exception, bracket_)
+import Data.ByteString qualified as BS
+import Data.HashMap.Strict (HashMap)
+import Data.HashMap.Strict qualified as HashMap
+import Data.Text (Text)
+import Effectful (Dispatch (..), DispatchOf, Eff, Effect, IOE, liftIO, withEffToIO, (:>))
+import Effectful.Dispatch.Static
+  ( SideEffects (..),
+    StaticRep,
+    evalStaticRep,
+    getStaticRep,
+  )
+import Effectful.Internal.Unlift (Limit (..), Persistence (..), UnliftStrategy (..))
+import OpenTelemetry.Attributes (Attribute, ToAttribute)
+import OpenTelemetry.Attributes qualified as Attr
+import OpenTelemetry.Context qualified as Ctx
+import OpenTelemetry.Context.ThreadLocal qualified as Ctx
+import OpenTelemetry.Trace.Core qualified as OTel
+import OpenTelemetry.Trace.Id qualified as OTel.Id
+import OpenTelemetry.Trace.TraceState qualified as OTel.TraceState
+
+--------------------------------------------------------------------------------
+-- Effect Definition
+--------------------------------------------------------------------------------
+
+-- | Tracing effect for OpenTelemetry spans.
+data Tracing :: Effect
+
+type instance DispatchOf Tracing = 'Static 'WithSideEffects
+
+-- | Internal representation carrying the Tracer and enabled flag.
+data instance StaticRep Tracing = TracingRep
+  { tracer :: !OTel.Tracer,
+    tracingEnabled :: !Bool
+  }
+
+--------------------------------------------------------------------------------
+-- Runners
+--------------------------------------------------------------------------------
+
+-- | Run with actual tracing enabled.
+-- The Tracer should be obtained from a TracerProvider.
+--
+-- Example:
+--
+-- @
+-- provider <- OTel.initializeTracerProvider
+-- let tracer = OTel.makeTracer provider "shibuya" OTel.tracerOptions
+-- runTracing tracer myAction
+-- @
+runTracing ::
+  (IOE :> es) =>
+  OTel.Tracer ->
+  Eff (Tracing : es) a ->
+  Eff es a
+runTracing t = evalStaticRep (TracingRep t True)
+
+-- | Run with tracing disabled (zero overhead).
+-- All tracing operations become no-ops.
+runTracingNoop ::
+  (IOE :> es) =>
+  Eff (Tracing : es) a ->
+  Eff es a
+runTracingNoop action = do
+  -- Create a minimal no-op tracer
+  noopProvider <- liftIO $ OTel.createTracerProvider [] OTel.emptyTracerProviderOptions
+  let noopTracer = OTel.makeTracer noopProvider instrumentationLib OTel.tracerOptions
+  evalStaticRep (TracingRep noopTracer False) action
+  where
+    instrumentationLib =
+      OTel.InstrumentationLibrary
+        { OTel.libraryName = "shibuya-noop",
+          OTel.libraryVersion = "",
+          OTel.librarySchemaUrl = "",
+          OTel.libraryAttributes = Attr.emptyAttributes
+        }
+
+--------------------------------------------------------------------------------
+-- Span Operations
+--------------------------------------------------------------------------------
+
+-- | Create a span wrapping an action.
+-- The span is automatically ended when the action completes.
+--
+-- Example:
+--
+-- @
+-- withSpan "process.message" consumerSpanArgs $ do
+--   -- your code here
+-- @
+withSpan ::
+  (Tracing :> es, IOE :> es) =>
+  Text ->
+  OTel.SpanArguments ->
+  Eff es a ->
+  Eff es a
+withSpan name args action = do
+  TracingRep {tracer, tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then action
+    else withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
+      OTel.inSpan tracer name args (runInIO action)
+
+-- | Create a span with access to the Span handle.
+-- Use this when you need to add attributes or events to the span.
+--
+-- Example:
+--
+-- @
+-- withSpan' "process.message" consumerSpanArgs $ \span -> do
+--   addAttribute span "message.id" msgId
+--   -- your code here
+-- @
+withSpan' ::
+  (Tracing :> es, IOE :> es) =>
+  Text ->
+  OTel.SpanArguments ->
+  (OTel.Span -> Eff es a) ->
+  Eff es a
+withSpan' name args f = do
+  TracingRep {tracer, tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then do
+      -- Create a dropped span for the callback
+      dummySpan <- liftIO mkDummySpan
+      f dummySpan
+    else withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
+      OTel.inSpan' tracer name args $ \traceSpan ->
+        runInIO (f traceSpan)
+
+--------------------------------------------------------------------------------
+-- Span Modification
+--------------------------------------------------------------------------------
+
+-- | Add an attribute to a span.
+addAttribute ::
+  (Tracing :> es, IOE :> es, ToAttribute a) =>
+  OTel.Span ->
+  Text ->
+  a ->
+  Eff es ()
+addAttribute traceSpan key val = do
+  TracingRep {tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then pure ()
+    else liftIO $ OTel.addAttribute traceSpan key val
+
+-- | Add multiple attributes to a span.
+addAttributes ::
+  (Tracing :> es, IOE :> es) =>
+  OTel.Span ->
+  HashMap Text Attribute ->
+  Eff es ()
+addAttributes traceSpan attrs = do
+  TracingRep {tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then pure ()
+    else liftIO $ OTel.addAttributes traceSpan attrs
+
+-- | Add an event to a span.
+addEvent ::
+  (Tracing :> es, IOE :> es) =>
+  OTel.Span ->
+  OTel.NewEvent ->
+  Eff es ()
+addEvent traceSpan event = do
+  TracingRep {tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then pure ()
+    else liftIO $ OTel.addEvent traceSpan event
+
+-- | Record an exception on a span.
+-- This adds an "exception" event with stack trace information.
+recordException ::
+  (Tracing :> es, IOE :> es, Exception e) =>
+  OTel.Span ->
+  e ->
+  Eff es ()
+recordException traceSpan ex = do
+  TracingRep {tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then pure ()
+    else liftIO $ OTel.recordException traceSpan HashMap.empty Nothing ex
+
+-- | Set the status of a span.
+setStatus ::
+  (Tracing :> es, IOE :> es) =>
+  OTel.Span ->
+  OTel.SpanStatus ->
+  Eff es ()
+setStatus traceSpan status = do
+  TracingRep {tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then pure ()
+    else liftIO $ OTel.setStatus traceSpan status
+
+--------------------------------------------------------------------------------
+-- Context Operations
+--------------------------------------------------------------------------------
+
+-- | Get the current tracer.
+getTracer ::
+  (Tracing :> es) =>
+  Eff es OTel.Tracer
+getTracer = do
+  TracingRep {tracer} <- getStaticRep
+  pure tracer
+
+-- | Check if tracing is enabled.
+isTracingEnabled ::
+  (Tracing :> es) =>
+  Eff es Bool
+isTracingEnabled = do
+  TracingRep {tracingEnabled} <- getStaticRep
+  pure tracingEnabled
+
+-- | Run an action with an extracted parent context.
+-- Use this to link spans to parent traces from message headers.
+--
+-- Example:
+--
+-- @
+-- let parentCtx = extractTraceContext messageHeaders
+-- withExtractedContext parentCtx $ do
+--   withSpan "child.span" consumerSpanArgs $ do
+--     -- this span will be a child of parentCtx
+-- @
+withExtractedContext ::
+  (Tracing :> es, IOE :> es) =>
+  Maybe OTel.SpanContext ->
+  Eff es a ->
+  Eff es a
+withExtractedContext Nothing action = action
+withExtractedContext (Just parentCtx) action = do
+  TracingRep {tracingEnabled} <- getStaticRep
+  if not tracingEnabled
+    then action
+    else do
+      -- Insert the parent span into thread-local context using bracket pattern
+      let parentSpan = OTel.wrapSpanContext parentCtx
+          newContext = Ctx.insertSpan parentSpan Ctx.empty
+      withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->
+        bracket_
+          (Ctx.attachContext newContext)
+          Ctx.detachContext
+          (runInIO action)
+
+--------------------------------------------------------------------------------
+-- Internal Helpers
+--------------------------------------------------------------------------------
+
+-- | Create a dummy/dropped span for use when tracing is disabled.
+-- All operations on this span are no-ops.
+mkDummySpan :: IO OTel.Span
+mkDummySpan = do
+  -- Create a FrozenSpan with invalid (all-zeros) IDs
+  -- This creates a "Dropped" span that no-ops all operations
+  let dummyTraceId = case OTel.Id.bytesToTraceId (BS.replicate 16 0) of
+        Right tid -> tid
+        Left _ -> error "mkDummySpan: failed to create dummy TraceId"
+      dummySpanId = case OTel.Id.bytesToSpanId (BS.replicate 8 0) of
+        Right sid -> sid
+        Left _ -> error "mkDummySpan: failed to create dummy SpanId"
+      dummySpanContext =
+        OTel.SpanContext
+          { OTel.traceFlags = OTel.defaultTraceFlags,
+            OTel.isRemote = False,
+            OTel.traceId = dummyTraceId,
+            OTel.spanId = dummySpanId,
+            OTel.traceState = OTel.TraceState.empty
+          }
+  pure $ OTel.wrapSpanContext dummySpanContext
diff --git a/src/Shibuya/Telemetry/Propagation.hs b/src/Shibuya/Telemetry/Propagation.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Telemetry/Propagation.hs
@@ -0,0 +1,51 @@
+-- | W3C Trace Context propagation for Shibuya.
+-- Extracts and injects trace context from/to message headers.
+module Shibuya.Telemetry.Propagation
+  ( -- * Extraction
+    extractTraceContext,
+
+    -- * Injection
+    injectTraceContext,
+
+    -- * Re-export
+    TraceHeaders,
+  )
+where
+
+import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C
+import OpenTelemetry.Trace.Core (Span, SpanContext)
+import Shibuya.Core.Types (TraceHeaders)
+
+-- | Extract SpanContext from W3C trace headers.
+-- Returns Nothing if headers are missing or malformed.
+--
+-- Example:
+--
+-- @
+-- let headers = [("traceparent", "00-abc123...-def456...-01")]
+-- case extractTraceContext headers of
+--   Nothing -> -- no valid trace context
+--   Just ctx -> -- use ctx as parent
+-- @
+extractTraceContext :: TraceHeaders -> Maybe SpanContext
+extractTraceContext headers =
+  let traceparent = lookup "traceparent" headers
+      tracestate = lookup "tracestate" headers
+   in W3C.decodeSpanContext traceparent tracestate
+
+-- | Inject current span's context into headers for propagation.
+-- Use this when producing messages that should carry trace context.
+--
+-- Example:
+--
+-- @
+-- headers <- injectTraceContext currentSpan
+-- -- headers contains [("traceparent", "..."), ("tracestate", "...")]
+-- @
+injectTraceContext :: Span -> IO TraceHeaders
+injectTraceContext otelSpan = do
+  (traceparent, tracestate) <- W3C.encodeSpanContext otelSpan
+  pure
+    [ ("traceparent", traceparent),
+      ("tracestate", tracestate)
+    ]
diff --git a/src/Shibuya/Telemetry/Semantic.hs b/src/Shibuya/Telemetry/Semantic.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Telemetry/Semantic.hs
@@ -0,0 +1,162 @@
+-- | Semantic conventions for Shibuya OpenTelemetry instrumentation.
+-- Provides span names, attribute keys, and helpers following OTel conventions.
+module Shibuya.Telemetry.Semantic
+  ( -- * Span Names
+    processMessageSpanName,
+    ingestSpanName,
+
+    -- * Attribute Keys (OTel Messaging Conventions)
+    attrMessagingSystem,
+    attrMessagingMessageId,
+    attrMessagingDestinationName,
+    attrMessagingDestinationPartitionId,
+
+    -- * Attribute Keys (Shibuya-specific)
+    attrShibuyaProcessorId,
+    attrShibuyaInflightCount,
+    attrShibuyaInflightMax,
+    attrShibuyaAckDecision,
+
+    -- * Event Names
+    eventHandlerStarted,
+    eventHandlerCompleted,
+    eventHandlerException,
+    eventAckDecision,
+
+    -- * SpanArguments Helpers
+    consumerSpanArgs,
+    internalSpanArgs,
+
+    -- * NewEvent Helper
+    mkEvent,
+  )
+where
+
+import Data.HashMap.Strict qualified as HashMap
+import Data.Text (Text)
+import OpenTelemetry.Attributes (Attribute)
+import OpenTelemetry.Trace.Core
+  ( NewEvent (..),
+    SpanArguments (..),
+    SpanKind (..),
+  )
+
+--------------------------------------------------------------------------------
+-- Span Names
+--------------------------------------------------------------------------------
+
+-- | Span name for processing a single message.
+processMessageSpanName :: Text
+processMessageSpanName = "shibuya.process.message"
+
+-- | Span name for the ingester lifecycle.
+ingestSpanName :: Text
+ingestSpanName = "shibuya.ingest"
+
+--------------------------------------------------------------------------------
+-- Attribute Keys (OTel Messaging Semantic Conventions)
+-- See: https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/
+--------------------------------------------------------------------------------
+
+-- | The messaging system identifier.
+attrMessagingSystem :: Text
+attrMessagingSystem = "messaging.system"
+
+-- | The message identifier.
+attrMessagingMessageId :: Text
+attrMessagingMessageId = "messaging.message.id"
+
+-- | The destination (queue/topic) name.
+attrMessagingDestinationName :: Text
+attrMessagingDestinationName = "messaging.destination.name"
+
+-- | The partition identifier.
+attrMessagingDestinationPartitionId :: Text
+attrMessagingDestinationPartitionId = "messaging.destination.partition.id"
+
+--------------------------------------------------------------------------------
+-- Attribute Keys (Shibuya-specific)
+--------------------------------------------------------------------------------
+
+-- | The processor identifier.
+attrShibuyaProcessorId :: Text
+attrShibuyaProcessorId = "shibuya.processor.id"
+
+-- | Current number of in-flight messages.
+attrShibuyaInflightCount :: Text
+attrShibuyaInflightCount = "shibuya.inflight.count"
+
+-- | Maximum concurrency limit.
+attrShibuyaInflightMax :: Text
+attrShibuyaInflightMax = "shibuya.inflight.max"
+
+-- | The ack decision made by the handler.
+attrShibuyaAckDecision :: Text
+attrShibuyaAckDecision = "shibuya.ack.decision"
+
+--------------------------------------------------------------------------------
+-- Event Names
+--------------------------------------------------------------------------------
+
+-- | Event recorded when handler execution starts.
+eventHandlerStarted :: Text
+eventHandlerStarted = "handler.started"
+
+-- | Event recorded when handler execution completes successfully.
+eventHandlerCompleted :: Text
+eventHandlerCompleted = "handler.completed"
+
+-- | Event recorded when handler throws an exception.
+eventHandlerException :: Text
+eventHandlerException = "handler.exception"
+
+-- | Event recorded when ack decision is made.
+eventAckDecision :: Text
+eventAckDecision = "ack.decision"
+
+--------------------------------------------------------------------------------
+-- SpanArguments Helpers
+--------------------------------------------------------------------------------
+
+-- | SpanArguments for message consumption (Consumer kind).
+consumerSpanArgs :: SpanArguments
+consumerSpanArgs =
+  SpanArguments
+    { kind = Consumer,
+      attributes = HashMap.empty,
+      links = [],
+      startTime = Nothing
+    }
+
+-- | SpanArguments for internal operations (Internal kind).
+internalSpanArgs :: SpanArguments
+internalSpanArgs =
+  SpanArguments
+    { kind = Internal,
+      attributes = HashMap.empty,
+      links = [],
+      startTime = Nothing
+    }
+
+--------------------------------------------------------------------------------
+-- NewEvent Helper
+--------------------------------------------------------------------------------
+
+-- | Create a NewEvent with the given name and attributes.
+-- Use 'toAttribute' to convert values to Attribute.
+--
+-- Example:
+--
+-- @
+-- mkEvent "handler.completed"
+--   [ ("duration_ms", toAttribute (42 :: Int))
+--   , ("status", toAttribute ("ok" :: Text))
+--   ]
+-- @
+mkEvent :: Text -> [(Text, Attribute)] -> NewEvent
+mkEvent name attrs =
+  NewEvent
+    { newEventName = name,
+      newEventAttributes = HashMap.fromList attrs,
+      newEventTimestamp = Nothing
+    }
diff --git a/test/Main.hs b/test/Main.hs
new file mode 100644
--- /dev/null
+++ b/test/Main.hs
@@ -0,0 +1,22 @@
+{-# LANGUAGE ImportQualifiedPost #-}
+
+module Main (main) where
+
+import Shibuya.Core.AckSpec qualified
+import Shibuya.Core.TypesSpec qualified
+import Shibuya.PolicySpec qualified
+import Shibuya.Runner.SupervisedSpec qualified
+import Shibuya.RunnerSpec qualified
+import Shibuya.Telemetry.EffectSpec qualified
+import Shibuya.Telemetry.PropagationSpec qualified
+import Test.Hspec
+
+main :: IO ()
+main = hspec $ do
+  describe "Shibuya.Core.Types" Shibuya.Core.TypesSpec.spec
+  describe "Shibuya.Core.Ack" Shibuya.Core.AckSpec.spec
+  describe "Shibuya.Policy" Shibuya.PolicySpec.spec
+  describe "Shibuya.Runner" Shibuya.RunnerSpec.spec
+  Shibuya.Runner.SupervisedSpec.spec
+  Shibuya.Telemetry.EffectSpec.spec
+  Shibuya.Telemetry.PropagationSpec.spec
diff --git a/test/Shibuya/Core/AckSpec.hs b/test/Shibuya/Core/AckSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/Core/AckSpec.hs
@@ -0,0 +1,90 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module Shibuya.Core.AckSpec (spec) where
+
+import Data.Time (secondsToNominalDiffTime)
+import Shibuya.Core.Ack
+import Test.Hspec
+
+spec :: Spec
+spec = do
+  describe "RetryDelay" $ do
+    it "wraps NominalDiffTime" $ do
+      let delay = RetryDelay (secondsToNominalDiffTime 30)
+      delay.unRetryDelay `shouldBe` secondsToNominalDiffTime 30
+
+    it "supports Eq" $ do
+      let d1 = RetryDelay (secondsToNominalDiffTime 10)
+          d2 = RetryDelay (secondsToNominalDiffTime 10)
+          d3 = RetryDelay (secondsToNominalDiffTime 20)
+      d1 `shouldBe` d2
+      d1 `shouldNotBe` d3
+
+  describe "DeadLetterReason" $ do
+    it "distinguishes all constructors" $ do
+      let r1 = PoisonPill "bad message"
+          r2 = InvalidPayload "parse error"
+          r3 = MaxRetriesExceeded
+      r1 `shouldNotBe` r2
+      r2 `shouldNotBe` r3
+      r1 `shouldNotBe` r3
+
+    it "PoisonPill carries message" $ do
+      let r = PoisonPill "corrupt data"
+      case r of
+        PoisonPill msg -> msg `shouldBe` "corrupt data"
+        _ -> expectationFailure "wrong constructor"
+
+    it "InvalidPayload carries message" $ do
+      let r = InvalidPayload "JSON decode failed"
+      case r of
+        InvalidPayload msg -> msg `shouldBe` "JSON decode failed"
+        _ -> expectationFailure "wrong constructor"
+
+  describe "HaltReason" $ do
+    it "HaltOrderedStream carries message" $ do
+      let r = HaltOrderedStream "ordering violation"
+      case r of
+        HaltOrderedStream msg -> msg `shouldBe` "ordering violation"
+        _ -> expectationFailure "wrong constructor"
+
+    it "HaltFatal carries message" $ do
+      let r = HaltFatal "database connection lost"
+      case r of
+        HaltFatal msg -> msg `shouldBe` "database connection lost"
+        _ -> expectationFailure "wrong constructor"
+
+  describe "AckDecision" $ do
+    it "distinguishes all constructors" $ do
+      let d1 = AckOk
+          d2 = AckRetry (RetryDelay 10)
+          d3 = AckDeadLetter MaxRetriesExceeded
+          d4 = AckHalt (HaltFatal "error")
+      d1 `shouldNotBe` d2
+      d2 `shouldNotBe` d3
+      d3 `shouldNotBe` d4
+      d1 `shouldNotBe` d4
+
+    it "AckOk equals AckOk" $ do
+      AckOk `shouldBe` AckOk
+
+    it "AckRetry carries delay" $ do
+      let delay = RetryDelay (secondsToNominalDiffTime 60)
+          decision = AckRetry delay
+      case decision of
+        AckRetry d -> d `shouldBe` delay
+        _ -> expectationFailure "wrong constructor"
+
+    it "AckDeadLetter carries reason" $ do
+      let reason = PoisonPill "unprocessable"
+          decision = AckDeadLetter reason
+      case decision of
+        AckDeadLetter r -> r `shouldBe` reason
+        _ -> expectationFailure "wrong constructor"
+
+    it "AckHalt carries reason" $ do
+      let reason = HaltOrderedStream "must stop"
+          decision = AckHalt reason
+      case decision of
+        AckHalt r -> r `shouldBe` reason
+        _ -> expectationFailure "wrong constructor"
diff --git a/test/Shibuya/Core/TypesSpec.hs b/test/Shibuya/Core/TypesSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/Core/TypesSpec.hs
@@ -0,0 +1,72 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module Shibuya.Core.TypesSpec (spec) where
+
+import Data.Time (UTCTime (..), fromGregorian)
+import Shibuya.Core.Types
+import Test.Hspec
+
+spec :: Spec
+spec = do
+  describe "MessageId" $ do
+    it "supports Eq" $ do
+      MessageId "abc" `shouldBe` MessageId "abc"
+      MessageId "abc" `shouldNotBe` MessageId "xyz"
+
+    it "supports Ord for Map keys" $ do
+      MessageId "a" `compare` MessageId "b" `shouldBe` LT
+      MessageId "b" `compare` MessageId "a" `shouldBe` GT
+      MessageId "a" `compare` MessageId "a" `shouldBe` EQ
+
+    it "supports IsString for OverloadedStrings" $ do
+      let msgId :: MessageId = "test-message"
+      msgId.unMessageId `shouldBe` "test-message"
+
+  describe "Cursor" $ do
+    it "CursorInt compares numerically" $ do
+      CursorInt 1 `compare` CursorInt 2 `shouldBe` LT
+      CursorInt 10 `compare` CursorInt 2 `shouldBe` GT
+
+    it "CursorText compares lexicographically" $ do
+      CursorText "a" `compare` CursorText "b" `shouldBe` LT
+      CursorText "z" `compare` CursorText "a" `shouldBe` GT
+
+    it "CursorInt comes before CursorText in Ord" $ do
+      -- This is the derived Ord behavior
+      CursorInt 999 `compare` CursorText "a" `shouldBe` LT
+
+  describe "Envelope" $ do
+    it "Functor: fmap id == id" $ do
+      let env = testEnvelope (42 :: Int)
+      fmap id env `shouldBe` env
+
+    it "Functor: fmap (f . g) == fmap f . fmap g" $ do
+      let env = testEnvelope (1 :: Int)
+          f = (+ 1)
+          g = (* 2)
+      fmap (f . g) env `shouldBe` (fmap f . fmap g) env
+
+    it "preserves all fields through fmap" $ do
+      let env = testEnvelope ("hello" :: String)
+          mapped = fmap length env
+      mapped.messageId `shouldBe` env.messageId
+      mapped.cursor `shouldBe` env.cursor
+      mapped.partition `shouldBe` env.partition
+      mapped.enqueuedAt `shouldBe` env.enqueuedAt
+      mapped.traceContext `shouldBe` env.traceContext
+      mapped.payload `shouldBe` 5
+
+-- Test helper
+testEnvelope :: msg -> Envelope msg
+testEnvelope msg =
+  Envelope
+    { messageId = MessageId "test-id",
+      cursor = Just (CursorInt 42),
+      partition = Just "partition-0",
+      enqueuedAt = Just testTime,
+      traceContext = Nothing,
+      payload = msg
+    }
+
+testTime :: UTCTime
+testTime = UTCTime (fromGregorian 2024 1 1) 0
diff --git a/test/Shibuya/PolicySpec.hs b/test/Shibuya/PolicySpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/PolicySpec.hs
@@ -0,0 +1,69 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module Shibuya.PolicySpec (spec) where
+
+import Shibuya.Policy
+import Test.Hspec
+import Prelude hiding (Ordering)
+
+spec :: Spec
+spec = do
+  describe "Ordering" $ do
+    it "all constructors are distinguishable" $ do
+      StrictInOrder `shouldNotBe` PartitionedInOrder
+      PartitionedInOrder `shouldNotBe` Unordered
+      StrictInOrder `shouldNotBe` Unordered
+
+  describe "Concurrency" $ do
+    it "all constructors are distinguishable" $ do
+      Serial `shouldNotBe` Ahead 1
+      Ahead 1 `shouldNotBe` Async 1
+      Serial `shouldNotBe` Async 1
+
+    it "Ahead preserves Int value" $ do
+      let c = Ahead 42
+      case c of
+        Ahead n -> n `shouldBe` 42
+        _ -> expectationFailure "wrong constructor"
+
+    it "Async preserves Int value" $ do
+      let c = Async 100
+      case c of
+        Async n -> n `shouldBe` 100
+        _ -> expectationFailure "wrong constructor"
+
+  describe "validatePolicy" $ do
+    describe "StrictInOrder" $ do
+      it "allows Serial" $ do
+        validatePolicy StrictInOrder Serial `shouldBe` Right ()
+
+      it "rejects Ahead" $ do
+        validatePolicy StrictInOrder (Ahead 5) `shouldSatisfy` isLeft
+
+      it "rejects Async" $ do
+        validatePolicy StrictInOrder (Async 5) `shouldSatisfy` isLeft
+
+    describe "PartitionedInOrder" $ do
+      it "allows Serial" $ do
+        validatePolicy PartitionedInOrder Serial `shouldBe` Right ()
+
+      it "allows Ahead" $ do
+        validatePolicy PartitionedInOrder (Ahead 10) `shouldBe` Right ()
+
+      it "allows Async" $ do
+        validatePolicy PartitionedInOrder (Async 10) `shouldBe` Right ()
+
+    describe "Unordered" $ do
+      it "allows Serial" $ do
+        validatePolicy Unordered Serial `shouldBe` Right ()
+
+      it "allows Ahead" $ do
+        validatePolicy Unordered (Ahead 10) `shouldBe` Right ()
+
+      it "allows Async" $ do
+        validatePolicy Unordered (Async 10) `shouldBe` Right ()
+
+-- Helper
+isLeft :: Either a b -> Bool
+isLeft (Left _) = True
+isLeft (Right _) = False
diff --git a/test/Shibuya/Runner/SupervisedSpec.hs b/test/Shibuya/Runner/SupervisedSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/Runner/SupervisedSpec.hs
@@ -0,0 +1,905 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module Shibuya.Runner.SupervisedSpec (spec) where
+
+import Control.Concurrent.NQE.Supervisor (Strategy (..))
+import Control.Concurrent.STM (readTVarIO)
+import Control.Monad (forM, replicateM)
+import Data.IORef (IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef)
+import Data.Text qualified as Text
+import Data.Time (UTCTime (..), fromGregorian)
+import Effectful (Eff, IOE, liftIO, runEff, (:>))
+import Shibuya.Adapter (Adapter (..))
+import Shibuya.App
+  ( ShutdownConfig (..),
+    SupervisionStrategy (..),
+    mkProcessor,
+    runApp,
+    stopAppGracefully,
+  )
+import Shibuya.Core (ProcessorHalt (..))
+import Shibuya.Core.Ack (AckDecision (..), HaltReason (..))
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Ingested (Ingested (..))
+import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))
+import Shibuya.Handler (Handler)
+import Shibuya.Policy (Concurrency (..))
+import Shibuya.Runner.Master
+  ( getAllMetrics,
+    startMaster,
+    stopMaster,
+  )
+import Shibuya.Runner.Metrics
+  ( InFlightInfo (..),
+    ProcessorId (..),
+    ProcessorMetrics (..),
+    ProcessorState (..),
+    StreamStats (..),
+  )
+import Shibuya.Runner.Supervised
+  ( SupervisedProcessor (..),
+    getMetrics,
+    isDone,
+    runSupervised,
+    runWithMetrics,
+  )
+import Shibuya.Telemetry.Effect (runTracingNoop)
+import Streamly.Data.Stream qualified as Stream
+import Test.Hspec
+import UnliftIO (SomeException, try)
+import UnliftIO.Concurrent (threadDelay)
+
+spec :: Spec
+spec = do
+  describe "Shibuya.Runner.Master" $ do
+    it "starts and stops cleanly" $ do
+      result <- runEff $ runTracingNoop $ do
+        master <- startMaster IgnoreAll
+        stopMaster master
+        pure ()
+      result `shouldBe` ()
+
+    it "returns empty metrics initially" $ do
+      metrics <- runEff $ runTracingNoop $ do
+        master <- startMaster IgnoreAll
+        m <- getAllMetrics master
+        stopMaster master
+        pure m
+      metrics `shouldSatisfy` null
+
+  describe "Shibuya.Runner.Supervised" $ do
+    describe "runWithMetrics" $ do
+      it "processes messages and tracks metrics" $ do
+        (finalMetrics, processedMsgs) <- runEff $ runTracingNoop $ do
+          -- Track processed messages
+          processedRef <- liftIO $ newIORef ([] :: [String])
+
+          -- Create test messages
+          messages <- createTestMessages 3
+
+          -- Create adapter and handler
+          let adapter = testAdapter messages
+              handler = testHandler processedRef
+              procId = ProcessorId "test-processor"
+
+          -- Run with metrics
+          sp <- runWithMetrics 10 procId adapter handler
+
+          -- Get results
+          metrics <- getMetrics sp
+          msgs <- liftIO $ readIORef processedRef
+          pure (metrics, msgs)
+
+        -- Verify metrics
+        finalMetrics.stats.received `shouldBe` 3
+        finalMetrics.stats.processed `shouldBe` 3
+        finalMetrics.stats.failed `shouldBe` 0
+        finalMetrics.state `shouldBe` Idle
+
+        -- Verify all messages were processed
+        length processedMsgs `shouldBe` 3
+
+      it "tracks failed messages" $ do
+        finalMetrics <- runEff $ runTracingNoop $ do
+          messages <- createTestMessages 3
+
+          let adapter = testAdapter messages
+              handler = failingHandler
+              procId = ProcessorId "failing-processor"
+
+          sp <- runWithMetrics 10 procId adapter handler
+          getMetrics sp
+
+        -- All should be failed (handler throws)
+        finalMetrics.stats.failed `shouldBe` 3
+
+      it "marks processor as done when stream exhausted" $ do
+        done <- runEff $ runTracingNoop $ do
+          messages <- createTestMessages 2
+
+          let adapter = testAdapter messages
+              handler = alwaysAckOk
+              procId = ProcessorId "quick-processor"
+
+          sp <- runWithMetrics 10 procId adapter handler
+          isDone sp
+
+        done `shouldBe` True
+
+    describe "AckHalt behavior" $ do
+      it "stops processing when handler returns AckHalt" $ do
+        processedCount <- do
+          processedRef <- newIORef (0 :: Int)
+
+          let handler _ = do
+                count <- liftIO $ readIORef processedRef
+                liftIO $ modifyIORef' processedRef (+ 1)
+                if count >= 2
+                  then pure $ AckHalt (HaltFatal "stopping after 3")
+                  else pure AckOk
+
+          -- Try to run and catch ProcessorHalt
+          result <- try $ runEff $ runTracingNoop $ do
+            messages <- createTestMessages 10
+            let adapter = testAdapter messages
+                procId = ProcessorId "halting-processor"
+            runWithMetrics 10 procId adapter handler
+
+          -- Verify ProcessorHalt was thrown
+          case result of
+            Left (ProcessorHalt reason) ->
+              reason `shouldBe` HaltFatal "stopping after 3"
+            Right _ ->
+              expectationFailure "Expected ProcessorHalt exception"
+
+          readIORef processedRef
+
+        -- Should have processed exactly 3 messages before halt
+        processedCount `shouldBe` 3
+
+      it "updates metrics to halted state on AckHalt" $ do
+        finalMetrics <- runEff $ runTracingNoop $ do
+          messages <- createTestMessages 5
+
+          let handler _ = pure $ AckHalt (HaltFatal "immediate halt")
+              adapter = testAdapter messages
+
+          master <- startMaster IgnoreAll
+
+          -- runSupervised catches ProcessorHalt, so we can check metrics
+          sp <- runSupervised master 10 (ProcessorId "halt-metrics") Serial adapter handler
+
+          -- Wait for processor to halt
+          liftIO $ threadDelay 100000 -- 100ms
+          m <- liftIO $ readTVarIO sp.metrics
+          stopMaster master
+          pure m
+
+        case finalMetrics.state of
+          Failed msg _ -> msg `shouldSatisfy` (Text.isInfixOf "immediate halt")
+          other -> expectationFailure $ "Expected Failed state, got: " ++ show other
+
+      it "halt in one supervised processor doesn't affect others" $ do
+        (countA, countB) <- runEff $ runTracingNoop $ do
+          countARef <- liftIO $ newIORef (0 :: Int)
+          countBRef <- liftIO $ newIORef (0 :: Int)
+
+          messagesA <- createTestMessages 10
+          messagesB <- createTestMessages 10
+
+          -- Handler A halts after 2 messages
+          let handlerA _ = do
+                count <- liftIO $ readIORef countARef
+                liftIO $ modifyIORef' countARef (+ 1)
+                if count >= 1
+                  then pure $ AckHalt (HaltFatal "A stopping")
+                  else pure AckOk
+
+          -- Handler B processes all messages
+          let handlerB _ = do
+                liftIO $ modifyIORef' countBRef (+ 1)
+                pure AckOk
+
+          let adapterA = testAdapter messagesA
+              adapterB = testAdapter messagesB
+
+          master <- startMaster IgnoreAll
+
+          _spA <- runSupervised master 10 (ProcessorId "A") Serial adapterA handlerA
+          _spB <- runSupervised master 10 (ProcessorId "B") Serial adapterB handlerB
+
+          -- Wait for both to complete
+          liftIO $ threadDelay 500000 -- 500ms
+          cA <- liftIO $ readIORef countARef
+          cB <- liftIO $ readIORef countBRef
+
+          stopMaster master
+          pure (cA, cB)
+
+        -- A should have stopped after 2 messages
+        countA `shouldBe` 2
+        -- B should have processed all 10
+        countB `shouldBe` 10
+
+    describe "Concurrency modes" $ do
+      describe "Ahead mode" $ do
+        it "processes all messages successfully" $ do
+          countRef <- newIORef (0 :: Int)
+
+          finalCount <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 5
+
+            let handler _ = do
+                  liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))
+                  liftIO $ threadDelay 10000 -- 10ms
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _ <- runSupervised master 10 (ProcessorId "ahead") (Ahead 3) adapter handler
+
+            -- Wait for completion
+            liftIO $ threadDelay 300000 -- 300ms
+            stopMaster master
+            liftIO $ readIORef countRef
+
+          finalCount `shouldBe` 5
+
+        it "respects maxBuffer limit" $ do
+          -- This tests that we don't start too many concurrent tasks
+          maxInFlightRef <- newIORef (0 :: Int)
+          currentInFlightRef <- newIORef (0 :: Int)
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 10
+
+            let handler _ = do
+                  -- Atomically increment current in-flight and get new value
+                  cur <-
+                    liftIO $
+                      atomicModifyIORef' currentInFlightRef (\n -> let n' = n + 1 in (n', n'))
+                  -- Update max
+                  liftIO $ atomicModifyIORef' maxInFlightRef (\m -> (max m cur, ()))
+                  -- Simulate some work
+                  liftIO $ threadDelay 50000 -- 50ms
+                  -- Decrement in-flight
+                  liftIO $ atomicModifyIORef' currentInFlightRef (\c -> (c - 1, ()))
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            -- Use Ahead with max 3 concurrent
+            _sp <- runSupervised master 10 (ProcessorId "ahead-limit") (Ahead 3) adapter handler
+
+            liftIO $ threadDelay 1000000 -- 1s to let it complete
+            stopMaster master
+
+          maxInFlight <- readIORef maxInFlightRef
+          -- Max in-flight should not exceed buffer size + 1 (buffer + currently processing)
+          maxInFlight `shouldSatisfy` (<= 4)
+
+      describe "Async mode" $ do
+        it "processes all messages concurrently" $ do
+          countRef <- newIORef (0 :: Int)
+
+          finalCount <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 5
+
+            let handler _ = do
+                  liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))
+                  liftIO $ threadDelay 10000 -- 10ms
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _ <- runSupervised master 10 (ProcessorId "async") (Async 5) adapter handler
+
+            -- Wait for completion
+            liftIO $ threadDelay 300000 -- 300ms
+            stopMaster master
+            liftIO $ readIORef countRef
+
+          finalCount `shouldBe` 5
+
+        it "respects concurrency limit" $ do
+          maxInFlightRef <- newIORef (0 :: Int)
+          currentInFlightRef <- newIORef (0 :: Int)
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 10
+
+            let handler _ = do
+                  cur <-
+                    liftIO $
+                      atomicModifyIORef' currentInFlightRef (\n -> let n' = n + 1 in (n', n'))
+                  liftIO $ atomicModifyIORef' maxInFlightRef (\m -> (max m cur, ()))
+                  liftIO $ threadDelay 50000 -- 50ms
+                  liftIO $ atomicModifyIORef' currentInFlightRef (\c -> (c - 1, ()))
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _sp <- runSupervised master 10 (ProcessorId "async-limit") (Async 3) adapter handler
+
+            liftIO $ threadDelay 1000000 -- 1s
+            stopMaster master
+
+          maxInFlight <- readIORef maxInFlightRef
+          maxInFlight `shouldSatisfy` (<= 4)
+
+      describe "Halt with concurrency" $ do
+        it "waits for in-flight to complete before halting" $ do
+          processedRef <- newIORef ([] :: [Int])
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 10
+
+            let handler ingested = do
+                  let msgNum = case ingested.envelope.cursor of
+                        Just (CursorInt n) -> n
+                        _ -> 0
+                  -- Simulate work
+                  liftIO $ threadDelay 50000 -- 50ms
+                  liftIO $ atomicModifyIORef' processedRef (\xs -> (msgNum : xs, ()))
+                  -- Halt on message 3
+                  if msgNum == 3
+                    then pure $ AckHalt (HaltFatal "halt on 3")
+                    else pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _sp <- runSupervised master 10 (ProcessorId "halt-concurrent") (Async 3) adapter handler
+
+            liftIO $ threadDelay 500000 -- 500ms
+            stopMaster master
+
+          processed <- readIORef processedRef
+          -- With Async 3, messages 1, 2, 3 start together
+          -- When 3 halts, 1 and 2 should complete
+          length processed `shouldSatisfy` (>= 3)
+
+        it "stops reading new messages after halt decision" $ do
+          processedRef <- newIORef (0 :: Int)
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 20
+
+            let handler _ = do
+                  count <-
+                    liftIO $
+                      atomicModifyIORef' processedRef (\n -> let n' = n + 1 in (n', n'))
+                  liftIO $ threadDelay 20000 -- 20ms
+                  -- Halt after processing 5 messages
+                  if count >= 5
+                    then pure $ AckHalt (HaltFatal "halt at 5")
+                    else pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _sp <- runSupervised master 10 (ProcessorId "halt-stop-read") (Async 3) adapter handler
+
+            liftIO $ threadDelay 500000 -- 500ms
+            stopMaster master
+
+          processed <- readIORef processedRef
+          -- Should process around 5-8 messages (5 before halt + a few in-flight)
+          -- but definitely not all 20
+          processed `shouldSatisfy` (< 15)
+
+      describe "Error handling" $ do
+        it "handler exception doesn't affect other in-flight handlers" $ do
+          resultsRef <- newIORef ([] :: [Either String Int])
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 5
+
+            let handler ingested = do
+                  let msgNum = case ingested.envelope.cursor of
+                        Just (CursorInt n) -> n
+                        _ -> 0
+                  -- Message 2 throws an exception
+                  if msgNum == 2
+                    then do
+                      liftIO $ atomicModifyIORef' resultsRef (\xs -> (Left "error on 2" : xs, ()))
+                      error "Intentional failure on message 2"
+                    else do
+                      liftIO $ threadDelay 30000 -- 30ms
+                      liftIO $ atomicModifyIORef' resultsRef (\xs -> (Right msgNum : xs, ()))
+                      pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _ <- runSupervised master 10 (ProcessorId "error-handling") (Async 3) adapter handler
+
+            liftIO $ threadDelay 500000 -- 500ms
+            stopMaster master
+
+          results <- readIORef resultsRef
+          -- Message 2 should have errored
+          let errors = [e | Left e <- results]
+          length errors `shouldBe` 1
+          -- Other messages (1, 3, 4, 5) should have succeeded
+          let successes = [n | Right n <- results]
+          length successes `shouldSatisfy` (>= 3)
+
+      describe "Metrics tracking" $ do
+        it "tracks in-flight count during concurrent processing" $ do
+          maxInFlightObserved <- newIORef (0 :: Int)
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 5
+
+            let handler _ = do
+                  liftIO $ threadDelay 100000 -- 100ms
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            sp <- runSupervised master 10 (ProcessorId "metrics-test") (Async 3) adapter handler
+
+            -- Check metrics while processing
+            liftIO $ do
+              threadDelay 50000 -- 50ms - should be in the middle of processing
+              metrics <- readTVarIO sp.metrics
+              case metrics.state of
+                Processing info _ -> modifyIORef' maxInFlightObserved (max info.inFlight)
+                _ -> pure ()
+
+            liftIO $ threadDelay 600000 -- 600ms to complete
+            stopMaster master
+
+          maxObserved <- readIORef maxInFlightObserved
+          -- Should have observed at least 2 concurrent handlers
+          maxObserved `shouldSatisfy` (>= 2)
+
+        it "reports correct maxConcurrency in metrics" $ do
+          observedMaxConc <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 3
+
+            let handler _ = do
+                  liftIO $ threadDelay 50000 -- 50ms
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            sp <- runSupervised master 10 (ProcessorId "max-conc") (Ahead 7) adapter handler
+
+            -- Check metrics while processing
+            result <- liftIO $ do
+              threadDelay 25000 -- 25ms
+              metrics <- readTVarIO sp.metrics
+              case metrics.state of
+                Processing info _ -> pure $ Just info.maxConcurrency
+                _ -> pure Nothing
+
+            liftIO $ threadDelay 300000
+            stopMaster master
+            pure result
+
+          observedMaxConc `shouldBe` Just 7
+
+      describe "Ahead mode ordering" $ do
+        it "completes handlers and emits results in input order" $ do
+          -- Track the order in which handlers START and COMPLETE
+          startOrderRef <- newIORef ([] :: [Int])
+          completeOrderRef <- newIORef ([] :: [Int])
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 5
+
+            let handler ingested = do
+                  let msgNum = case ingested.envelope.cursor of
+                        Just (CursorInt n) -> n
+                        _ -> 0
+                  -- Record start
+                  liftIO $ atomicModifyIORef' startOrderRef (\xs -> (msgNum : xs, ()))
+                  -- Variable delay: message 1 is slowest, 5 is fastest
+                  liftIO $ threadDelay ((6 - msgNum) * 20000)
+                  -- Record completion
+                  liftIO $ atomicModifyIORef' completeOrderRef (\xs -> (msgNum : xs, ()))
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _ <- runSupervised master 10 (ProcessorId "ahead-order") (Ahead 5) adapter handler
+
+            liftIO $ threadDelay 500000 -- 500ms
+            stopMaster master
+
+          -- With Ahead mode, all 5 should complete
+          completeOrder <- readIORef completeOrderRef
+          length completeOrder `shouldBe` 5
+
+    -- The key property: all messages were processed
+    -- (Ordering of side effects may vary, but stream output is ordered)
+
+    describe "Robustness" $ do
+      describe "Adapter source exceptions" $ do
+        it "handles adapter source throwing mid-stream" $ do
+          processedRef <- newIORef (0 :: Int)
+
+          _ <- runEff $ runTracingNoop $ do
+            -- Create an adapter where source throws after 3 messages
+            _goodMessages <- createTestMessages 3
+            -- Use fromEffect to throw after yielding good messages
+            let failingSource = Stream.unfoldrM step (0 :: Int)
+                  where
+                    step n
+                      | n < 3 = do
+                          msg <- liftIO $ createSingleMessage (n + 1)
+                          pure $ Just (msg, n + 1)
+                      | n == 3 = error "Network failure mid-stream"
+                      | otherwise = pure Nothing
+
+            let adapter =
+                  Adapter
+                    { adapterName = "test:failing-source",
+                      source = failingSource,
+                      shutdown = pure ()
+                    }
+
+            let handler _ = do
+                  liftIO $ modifyIORef' processedRef (+ 1)
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            _ <- runSupervised master 10 (ProcessorId "failing-adapter") Serial adapter handler
+
+            -- Wait for processing
+            liftIO $ threadDelay 200000 -- 200ms
+            stopMaster master
+
+          -- Should have processed the 3 good messages before the error
+          processed <- readIORef processedRef
+          processed `shouldBe` 3
+
+      describe "Rapid start/stop cycles" $ do
+        it "handles rapid start/stop without resource leaks" $ do
+          -- Run 50 rapid cycles
+          results <- replicateM 50 $ runEff $ runTracingNoop $ do
+            master <- startMaster IgnoreAll
+            messages <- createTestMessages 5
+
+            let adapter = testAdapter messages
+                handler _ = pure AckOk
+
+            sp <- runSupervised master 10 (ProcessorId "rapid") Serial adapter handler
+
+            -- Minimal wait
+            liftIO $ threadDelay 10000 -- 10ms
+            stopMaster master
+
+            -- Check it stopped cleanly
+            isDone sp
+
+          -- All should complete (no hangs, no crashes)
+          length results `shouldBe` 50
+
+        it "handles concurrent start/stop cycles" $ do
+          -- Run 20 concurrent cycles
+          countRef <- newIORef (0 :: Int)
+
+          _ <- runEff $ runTracingNoop $ do
+            master <- startMaster IgnoreAll
+
+            -- Start 10 processors rapidly
+            _sps <- forM [(1 :: Int) .. 10] $ \i -> do
+              messages <- createTestMessages 3
+              let adapter = testAdapter messages
+                  handler _ = do
+                    liftIO $ atomicModifyIORef' countRef (\n -> (n + 1, ()))
+                    liftIO $ threadDelay 5000 -- 5ms
+                    pure AckOk
+              runSupervised master 10 (ProcessorId $ "concurrent-" <> Text.pack (show i)) Serial adapter handler
+
+            -- Wait briefly then stop
+            liftIO $ threadDelay 100000 -- 100ms
+            stopMaster master
+
+          -- Should have processed messages from multiple processors
+          count <- readIORef countRef
+          count `shouldSatisfy` (> 0)
+
+      describe "KillAll supervision strategy" $ do
+        it "stops all processors when adapter source fails with KillAll strategy" $ do
+          countARef <- newIORef (0 :: Int)
+          countBRef <- newIORef (0 :: Int)
+
+          _ :: Either SomeException () <- try $ runEff $ runTracingNoop $ do
+            messagesB <- createTestMessages 20
+
+            -- Adapter A throws after 3 messages (ingester crash, not handler)
+            let failingSourceA = Stream.unfoldrM step (0 :: Int)
+                  where
+                    step n
+                      | n < 3 = do
+                          msg <- liftIO $ createSingleMessage (n + 1)
+                          pure $ Just (msg, n + 1)
+                      | otherwise = error "Adapter A source failed!"
+
+            let adapterA =
+                  Adapter
+                    { adapterName = "test:failing-A",
+                      source = failingSourceA,
+                      shutdown = pure ()
+                    }
+
+            -- Handler A counts messages
+            let handlerA _ = do
+                  liftIO $ modifyIORef' countARef (+ 1)
+                  liftIO $ threadDelay 10000 -- 10ms
+                  pure AckOk
+
+            -- Handler B processes slowly
+            let handlerB _ = do
+                  liftIO $ modifyIORef' countBRef (+ 1)
+                  liftIO $ threadDelay 30000 -- 30ms
+                  pure AckOk
+
+            let adapterB = testAdapter messagesB
+
+            -- Use KillAll strategy
+            master <- startMaster KillAll
+
+            _spA <- runSupervised master 10 (ProcessorId "killall-A") Serial adapterA handlerA
+            _spB <- runSupervised master 10 (ProcessorId "killall-B") Serial adapterB handlerB
+
+            -- Wait for A to fail and trigger KillAll
+            liftIO $ threadDelay 500000 -- 500ms
+            stopMaster master
+
+          -- A should have processed messages before adapter failed
+          countA <- readIORef countARef
+          countA `shouldSatisfy` (<= 3)
+
+          -- B should have been killed (not processed all 20)
+          countB <- readIORef countBRef
+          countB `shouldSatisfy` (< 20)
+
+      describe "Multiple concurrent halts" $ do
+        it "handles multiple handlers returning AckHalt simultaneously" $ do
+          haltCountRef <- newIORef (0 :: Int)
+          processedRef <- newIORef (0 :: Int)
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 10
+
+            let handler ingested = do
+                  let msgNum = case ingested.envelope.cursor of
+                        Just (CursorInt n) -> n
+                        _ -> 0
+                  liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))
+                  liftIO $ threadDelay 30000 -- 30ms
+                  -- Messages 2, 3, 4 all return halt
+                  if msgNum `elem` [2, 3, 4]
+                    then do
+                      liftIO $ atomicModifyIORef' haltCountRef (\n -> (n + 1, ()))
+                      pure $ AckHalt (HaltFatal $ "halt from " <> Text.pack (show msgNum))
+                    else pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _ <- runSupervised master 10 (ProcessorId "multi-halt") (Async 5) adapter handler
+
+            liftIO $ threadDelay 500000 -- 500ms
+            stopMaster master
+
+          -- Multiple halts should have been recorded
+          haltCount <- readIORef haltCountRef
+          haltCount `shouldSatisfy` (>= 1) -- At least one halt processed
+
+          -- Should not have processed all messages
+          processed <- readIORef processedRef
+          processed `shouldSatisfy` (< 10)
+
+      describe "Stability under load" $ do
+        it "processes many messages without issues" $ do
+          processedRef <- newIORef (0 :: Int)
+
+          finalCount <- runEff $ runTracingNoop $ do
+            -- 500 messages
+            messages <- createTestMessages 500
+
+            let handler _ = do
+                  liftIO $ atomicModifyIORef' processedRef (\n -> (n + 1, ()))
+                  pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _sp <- runSupervised master 100 (ProcessorId "load-test") (Async 10) adapter handler
+
+            -- Wait for completion
+            liftIO $ threadDelay 2000000 -- 2s
+            stopMaster master
+            liftIO $ readIORef processedRef
+
+          finalCount `shouldBe` 500
+
+        it "handles mixed success and failure under load" $ do
+          successRef <- newIORef (0 :: Int)
+          failRef <- newIORef (0 :: Int)
+
+          _ <- runEff $ runTracingNoop $ do
+            messages <- createTestMessages 200
+
+            let handler ingested = do
+                  let msgNum = case ingested.envelope.cursor of
+                        Just (CursorInt n) -> n
+                        _ -> 0
+                  -- Every 7th message fails
+                  if msgNum `mod` 7 == 0
+                    then do
+                      liftIO $ atomicModifyIORef' failRef (\n -> (n + 1, ()))
+                      error "Intentional failure"
+                    else do
+                      liftIO $ atomicModifyIORef' successRef (\n -> (n + 1, ()))
+                      pure AckOk
+
+            master <- startMaster IgnoreAll
+            let adapter = testAdapter messages
+            _ <- runSupervised master 50 (ProcessorId "mixed-load") (Async 5) adapter handler
+
+            liftIO $ threadDelay 1000000 -- 1s
+            stopMaster master
+
+          success <- readIORef successRef
+          failures <- readIORef failRef
+
+          -- Should have processed many messages
+          success `shouldSatisfy` (> 150)
+          -- Should have recorded failures
+          failures `shouldSatisfy` (> 20)
+
+  describe "Graceful Shutdown" $ do
+    describe "stopAppGracefully" $ do
+      it "drains in-flight messages before stopping" $ do
+        processedRef <- newIORef (0 :: Int)
+
+        drained <- runEff $ runTracingNoop $ do
+          messages <- createTestMessages 10
+
+          let handler _ = do
+                liftIO $ do
+                  threadDelay 50000 -- 50ms per message
+                  modifyIORef' processedRef (+ 1)
+                pure AckOk
+
+          let adapter = testAdapter messages
+              processor = mkProcessor adapter handler
+
+          result <- runApp IgnoreFailures 10 [(ProcessorId "drain-test", processor)]
+          case result of
+            Left _ -> pure False
+            Right appHandle -> do
+              -- Give some time for processing to start
+              liftIO $ threadDelay 100000 -- 100ms
+
+              -- Graceful shutdown with generous timeout
+              let config = ShutdownConfig {drainTimeout = 5} -- 5 seconds
+              stopAppGracefully config appHandle
+
+        drained `shouldBe` True
+
+        -- All messages should have been processed
+        processed <- readIORef processedRef
+        processed `shouldBe` 10
+
+      it "respects timeout when processors are slow" $ do
+        processedRef <- newIORef (0 :: Int)
+
+        drained <- runEff $ runTracingNoop $ do
+          messages <- createTestMessages 20
+
+          let handler _ = do
+                liftIO $ do
+                  threadDelay 500000 -- 500ms per message - very slow
+                  modifyIORef' processedRef (+ 1)
+                pure AckOk
+
+          let adapter = testAdapter messages
+              processor = mkProcessor adapter handler
+
+          result <- runApp IgnoreFailures 5 [(ProcessorId "timeout-test", processor)]
+          case result of
+            Left _ -> pure True -- Treat error as "drained" for simplicity
+            Right appHandle -> do
+              -- Start immediately
+              liftIO $ threadDelay 100000 -- 100ms
+
+              -- Very short timeout (0.3 seconds)
+              let config = ShutdownConfig {drainTimeout = 0.3}
+              stopAppGracefully config appHandle
+
+        -- Should timeout (not all drained)
+        drained `shouldBe` False
+
+        -- Should have processed only some messages
+        processed <- readIORef processedRef
+        processed `shouldSatisfy` (< 20)
+
+      it "returns True when all processors finish quickly" $ do
+        drained <- runEff $ runTracingNoop $ do
+          messages <- createTestMessages 5
+
+          let handler _ = pure AckOk -- Instant processing
+          let adapter = testAdapter messages
+              processor = mkProcessor adapter handler
+
+          result <- runApp IgnoreFailures 10 [(ProcessorId "quick-test", processor)]
+          case result of
+            Left _ -> pure False
+            Right appHandle -> do
+              -- Give time to complete
+              liftIO $ threadDelay 200000 -- 200ms
+              let config = ShutdownConfig {drainTimeout = 1}
+              stopAppGracefully config appHandle
+
+        drained `shouldBe` True
+
+-- Test helpers
+
+testTime :: UTCTime
+testTime = UTCTime (fromGregorian 2026 1 1) 0
+
+createTestMessages :: (IOE :> es) => Int -> Eff es [Ingested es String]
+createTestMessages n = mapM createMessage [1 .. n]
+  where
+    createMessage i = do
+      let msgId = MessageId $ "msg-" <> Text.pack (show i)
+          env =
+            Envelope
+              { messageId = msgId,
+                cursor = Just (CursorInt i),
+                partition = Nothing,
+                enqueuedAt = Just testTime,
+                traceContext = Nothing,
+                payload = "message-" <> show i
+              }
+          ackHandle = AckHandle $ \_ -> pure ()
+      pure $
+        Ingested
+          { envelope = env,
+            ack = ackHandle,
+            lease = Nothing
+          }
+
+-- | Create a single message (for use in streaming contexts)
+-- Polymorphic over effect stack for use with tracing.
+createSingleMessage :: (IOE :> es) => Int -> IO (Ingested es String)
+createSingleMessage i = do
+  let msgId = MessageId $ "msg-" <> Text.pack (show i)
+      env =
+        Envelope
+          { messageId = msgId,
+            cursor = Just (CursorInt i),
+            partition = Nothing,
+            enqueuedAt = Just testTime,
+            traceContext = Nothing,
+            payload = "message-" <> show i
+          }
+      ackHandle = AckHandle $ \_ -> pure ()
+  pure $
+    Ingested
+      { envelope = env,
+        ack = ackHandle,
+        lease = Nothing
+      }
+
+testAdapter :: [Ingested es String] -> Adapter es String
+testAdapter messages =
+  Adapter
+    { adapterName = "test:supervised",
+      source = Stream.fromList messages,
+      shutdown = pure ()
+    }
+
+testHandler :: (IOE :> es) => IORef [String] -> Handler es String
+testHandler ref ingested = do
+  liftIO $ modifyIORef' ref (ingested.envelope.payload :)
+  pure AckOk
+
+failingHandler :: Handler es msg
+failingHandler _ = error "Intentional failure"
+
+alwaysAckOk :: Handler es msg
+alwaysAckOk _ = pure AckOk
diff --git a/test/Shibuya/RunnerSpec.hs b/test/Shibuya/RunnerSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/RunnerSpec.hs
@@ -0,0 +1,258 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module Shibuya.RunnerSpec (spec) where
+
+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)
+import Data.Text qualified as Text
+import Data.Time (UTCTime (..), fromGregorian)
+import Effectful (Eff, IOE, liftIO, runEff, (:>))
+import Shibuya.Adapter (Adapter (..))
+import Shibuya.Adapter.Mock (TrackingAck (..), newTrackingAck, trackingAckHandle)
+import Shibuya.App (AppError (..), QueueProcessor (..), SupervisionStrategy (..), mkProcessor, runApp, waitApp)
+import Shibuya.Core.Ack (AckDecision (..))
+import Shibuya.Core.AckHandle (AckHandle (..))
+import Shibuya.Core.Error (PolicyError (..))
+import Shibuya.Core.Ingested (Ingested (..))
+import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))
+import Shibuya.Handler (Handler)
+import Shibuya.Policy (Concurrency (..), Ordering (..))
+import Shibuya.Runner.Metrics (ProcessorId (..))
+import Shibuya.Telemetry.Effect (runTracingNoop)
+import Streamly.Data.Stream qualified as Stream
+import Test.Hspec
+import Prelude hiding (Ordering)
+
+spec :: Spec
+spec = do
+  describe "runApp" $ do
+    it "processes messages from mock adapter" $ do
+      result <- runEff $ runTracingNoop $ do
+        -- Track processed messages
+        processedRef <- liftIO $ newIORef ([] :: [String])
+
+        -- Create test messages
+        messages <- createTestMessages 3
+
+        -- Create adapter and handler
+        let adapter = testAdapter messages
+            handler = testHandler processedRef
+            processor = mkProcessor adapter handler
+
+        -- Run the app
+        res <-
+          runApp
+            IgnoreFailures
+            100
+            [ (ProcessorId "test", processor)
+            ]
+
+        case res of
+          Left err -> pure $ Left err
+          Right appHandle -> do
+            waitApp appHandle
+            pure $ Right ()
+
+      -- Verify result
+      result `shouldBe` Right ()
+
+    it "calls finalize for each message" $ do
+      (decisions, result) <- runEff $ runTracingNoop $ do
+        -- Track ack decisions
+        tracking <- newTrackingAck
+
+        -- Create test messages with tracking acks
+        messages <- createTrackedMessages tracking 3
+
+        -- Create adapter and handler
+        let adapter = testAdapter messages
+            handler = alwaysAckOk
+            processor = mkProcessor adapter handler
+
+        -- Run the app
+        res <-
+          runApp
+            IgnoreFailures
+            100
+            [ (ProcessorId "test", processor)
+            ]
+
+        case res of
+          Left err -> do
+            decs <- liftIO $ readIORef tracking.trackedDecisions
+            pure (decs, Left err)
+          Right appHandle -> do
+            waitApp appHandle
+            decs <- liftIO $ readIORef tracking.trackedDecisions
+            pure (decs, Right ())
+
+      -- Verify all messages were acked
+      result `shouldBe` Right ()
+      length decisions `shouldBe` 3
+      -- All should be AckOk (decisions are in reverse order)
+      all ((== AckOk) . snd) decisions `shouldBe` True
+
+    it "returns AppHandle for multiple processors" $ do
+      result <- runEff $ runTracingNoop $ do
+        messages1 <- createTestMessages 2
+        messages2 <- createTestMessages 2
+
+        let adapter1 = testAdapter messages1
+            adapter2 = testAdapter messages2
+            handler = alwaysAckOk
+            proc1 = mkProcessor adapter1 handler
+            proc2 = mkProcessor adapter2 handler
+
+        res <-
+          runApp
+            IgnoreFailures
+            100
+            [ (ProcessorId "proc1", proc1),
+              (ProcessorId "proc2", proc2)
+            ]
+
+        case res of
+          Left err -> pure $ Left err
+          Right appHandle -> do
+            waitApp appHandle
+            pure $ Right ()
+
+      result `shouldBe` Right ()
+
+  describe "Policy validation" $ do
+    it "rejects StrictInOrder with Async" $ do
+      result <- runEff $ runTracingNoop $ do
+        messages <- createTestMessages 3
+        let adapter = testAdapter messages
+            handler = alwaysAckOk
+            -- Invalid combination: StrictInOrder requires Serial
+            processor = QueueProcessor adapter handler StrictInOrder (Async 5)
+
+        runApp IgnoreFailures 100 [(ProcessorId "invalid", processor)]
+
+      case result of
+        Left (AppPolicyError (InvalidPolicyCombo _)) -> pure ()
+        Left err -> expectationFailure $ "Expected AppPolicyError, got: " ++ show err
+        Right _ -> expectationFailure "Expected policy validation to fail"
+
+    it "rejects StrictInOrder with Ahead" $ do
+      result <- runEff $ runTracingNoop $ do
+        messages <- createTestMessages 3
+        let adapter = testAdapter messages
+            handler = alwaysAckOk
+            processor = QueueProcessor adapter handler StrictInOrder (Ahead 5)
+
+        runApp IgnoreFailures 100 [(ProcessorId "invalid", processor)]
+
+      case result of
+        Left (AppPolicyError (InvalidPolicyCombo _)) -> pure ()
+        Left err -> expectationFailure $ "Expected AppPolicyError, got: " ++ show err
+        Right _ -> expectationFailure "Expected policy validation to fail"
+
+    it "accepts valid policy combinations" $ do
+      result <- runEff $ runTracingNoop $ do
+        messages <- createTestMessages 2
+        let adapter = testAdapter messages
+            handler = alwaysAckOk
+            -- Valid combinations
+            proc1 = QueueProcessor adapter handler Unordered (Async 3)
+            proc2 = QueueProcessor adapter handler PartitionedInOrder (Ahead 3)
+
+        res <-
+          runApp
+            IgnoreFailures
+            100
+            [ (ProcessorId "async", proc1),
+              (ProcessorId "ahead", proc2)
+            ]
+        case res of
+          Left err -> pure $ Left err
+          Right appHandle -> do
+            waitApp appHandle
+            pure $ Right ()
+
+      result `shouldBe` Right ()
+
+  describe "mkProcessor" $ do
+    it "creates processor with Unordered ordering" $ do
+      messages <- runEff $ createTestMessages 1
+      let adapter = testAdapter messages
+          handler = alwaysAckOk
+          QueueProcessor _ _ ordering _ = mkProcessor adapter handler
+      ordering `shouldBe` Unordered
+
+    it "creates processor with Serial concurrency" $ do
+      messages <- runEff $ createTestMessages 1
+      let adapter = testAdapter messages
+          handler = alwaysAckOk
+          QueueProcessor _ _ _ concurrency = mkProcessor adapter handler
+      concurrency `shouldBe` Serial
+
+-- Test helpers
+
+testTime :: UTCTime
+testTime = UTCTime (fromGregorian 2024 1 1) 0
+
+-- | Create N test messages with simple string payloads
+createTestMessages :: (IOE :> es) => Int -> Eff es [Ingested es String]
+createTestMessages n = mapM createMessage [1 .. n]
+  where
+    createMessage i = do
+      let msgId = MessageId $ "msg-" <> (if i < 10 then "0" else "") <> Text.pack (show i)
+          env =
+            Envelope
+              { messageId = msgId,
+                cursor = Just (CursorInt i),
+                partition = Nothing,
+                enqueuedAt = Just testTime,
+                traceContext = Nothing,
+                payload = "message-" <> show i
+              }
+          ackHandle = AckHandle $ \_ -> pure () -- No-op ack
+      pure $
+        Ingested
+          { envelope = env,
+            ack = ackHandle,
+            lease = Nothing
+          }
+
+-- | Create N test messages with tracking acks
+createTrackedMessages :: (IOE :> es) => TrackingAck -> Int -> Eff es [Ingested es String]
+createTrackedMessages tracking n = mapM createMessage [1 .. n]
+  where
+    createMessage i = do
+      let msgId = MessageId $ "msg-" <> (if i < 10 then "0" else "") <> Text.pack (show i)
+          env =
+            Envelope
+              { messageId = msgId,
+                cursor = Just (CursorInt i),
+                partition = Nothing,
+                enqueuedAt = Just testTime,
+                traceContext = Nothing,
+                payload = "message-" <> show i
+              }
+          ackHandle = trackingAckHandle tracking msgId
+      pure $
+        Ingested
+          { envelope = env,
+            ack = ackHandle,
+            lease = Nothing
+          }
+
+-- | Create a test adapter from a list of messages
+testAdapter :: [Ingested es String] -> Adapter es String
+testAdapter messages =
+  Adapter
+    { adapterName = "test:mock",
+      source = Stream.fromList messages,
+      shutdown = pure ()
+    }
+
+-- | Handler that records processed messages
+testHandler :: (IOE :> es) => IORef [String] -> Handler es String
+testHandler ref ingested = do
+  liftIO $ modifyIORef' ref (ingested.envelope.payload :)
+  pure AckOk
+
+-- | Handler that always returns AckOk
+alwaysAckOk :: Handler es msg
+alwaysAckOk _ = pure AckOk
diff --git a/test/Shibuya/Telemetry/EffectSpec.hs b/test/Shibuya/Telemetry/EffectSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/Telemetry/EffectSpec.hs
@@ -0,0 +1,108 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+-- | Tests for the Tracing effect.
+module Shibuya.Telemetry.EffectSpec (spec) where
+
+import Data.IORef (newIORef, readIORef, writeIORef)
+import Data.Text (Text)
+import Effectful (liftIO, runEff)
+import OpenTelemetry.Trace.Core qualified as OTel
+import Shibuya.Telemetry.Effect
+  ( addAttribute,
+    addEvent,
+    getTracer,
+    isTracingEnabled,
+    runTracingNoop,
+    setStatus,
+    withSpan,
+    withSpan',
+  )
+import Shibuya.Telemetry.Semantic (consumerSpanArgs, mkEvent)
+import Test.Hspec
+
+spec :: Spec
+spec = describe "Shibuya.Telemetry.Effect" $ do
+  describe "runTracingNoop" $ do
+    it "executes actions and returns result" $ do
+      result <- runEff $ runTracingNoop $ do
+        pure (42 :: Int)
+      result `shouldBe` 42
+
+    it "executes nested withSpan without error" $ do
+      result <- runEff $ runTracingNoop $ do
+        withSpan "outer" consumerSpanArgs $ do
+          withSpan "inner" consumerSpanArgs $ do
+            pure ("success" :: String)
+      result `shouldBe` "success"
+
+    it "executes withSpan' and provides span handle" $ do
+      result <- runEff $ runTracingNoop $ do
+        withSpan' "test-span" consumerSpanArgs $ \traceSpan -> do
+          -- These should be no-ops but not error
+          addAttribute traceSpan "key" ("value" :: Text)
+          addEvent traceSpan (mkEvent "test.event" [])
+          setStatus traceSpan OTel.Ok
+          pure (123 :: Int)
+      result `shouldBe` 123
+
+    it "reports tracing as disabled" $ do
+      enabled <- runEff $ runTracingNoop $ isTracingEnabled
+      enabled `shouldBe` False
+
+    it "provides a tracer (noop)" $ do
+      result <- runEff $ runTracingNoop $ do
+        _tracer <- getTracer
+        pure ("got tracer" :: String)
+      result `shouldBe` "got tracer"
+
+    it "handles exceptions in actions correctly" $ do
+      ref <- newIORef (0 :: Int)
+      result <- runEff $ runTracingNoop $ do
+        withSpan "outer" consumerSpanArgs $ do
+          liftIO $ writeIORef ref 1
+          withSpan "inner" consumerSpanArgs $ do
+            liftIO $ writeIORef ref 2
+            pure ()
+        liftIO $ readIORef ref
+      result `shouldBe` 2
+
+    it "works with effectful actions" $ do
+      ref <- newIORef ([] :: [String])
+      let modifyIORef' r f = do
+            val <- readIORef r
+            writeIORef r (f val)
+      result <- runEff $ runTracingNoop $ do
+        withSpan "span1" consumerSpanArgs $ do
+          liftIO $ modifyIORef' ref ("span1" :)
+        withSpan "span2" consumerSpanArgs $ do
+          liftIO $ modifyIORef' ref ("span2" :)
+        liftIO $ readIORef ref
+      -- Order is reversed due to cons
+      result `shouldBe` ["span2", "span1"]
+
+  describe "Span operations (noop mode)" $ do
+    it "addAttribute is a no-op" $ do
+      result <- runEff $ runTracingNoop $ do
+        withSpan' "test" consumerSpanArgs $ \traceSpan -> do
+          addAttribute traceSpan "string" ("value" :: Text)
+          addAttribute traceSpan "int" (42 :: Int)
+          addAttribute traceSpan "bool" True
+          pure ("done" :: String)
+      result `shouldBe` ("done" :: String)
+
+    it "addEvent is a no-op" $ do
+      result <- runEff $ runTracingNoop $ do
+        withSpan' "test" consumerSpanArgs $ \traceSpan -> do
+          addEvent traceSpan (mkEvent "event1" [])
+          addEvent traceSpan (mkEvent "event2" [("key", OTel.toAttribute ("val" :: Text))])
+          pure ("done" :: String)
+      result `shouldBe` ("done" :: String)
+
+    it "setStatus is a no-op" $ do
+      result <- runEff $ runTracingNoop $ do
+        withSpan' "test" consumerSpanArgs $ \traceSpan -> do
+          setStatus traceSpan OTel.Ok
+          setStatus traceSpan (OTel.Error "test error")
+          setStatus traceSpan OTel.Unset
+          pure ("done" :: String)
+      result `shouldBe` ("done" :: String)
diff --git a/test/Shibuya/Telemetry/PropagationSpec.hs b/test/Shibuya/Telemetry/PropagationSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/Telemetry/PropagationSpec.hs
@@ -0,0 +1,121 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+-- | Tests for W3C Trace Context propagation.
+module Shibuya.Telemetry.PropagationSpec (spec) where
+
+import Data.ByteArray.Encoding (Base (..))
+import Data.ByteString (ByteString)
+import OpenTelemetry.Trace.Core qualified as OTel
+import OpenTelemetry.Trace.Id qualified as OTel.Id
+import Shibuya.Telemetry.Propagation (extractTraceContext)
+import Test.Hspec
+
+spec :: Spec
+spec = describe "Shibuya.Telemetry.Propagation" $ do
+  describe "extractTraceContext" $ do
+    it "extracts valid W3C traceparent header" $ do
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")
+            ]
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "Expected to extract trace context"
+        Just ctx -> do
+          -- Verify trace ID
+          OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "0af7651916cd43dd8448eb211c80319c"
+          -- Verify span ID
+          OTel.Id.spanIdBaseEncodedText Base16 (OTel.spanId ctx) `shouldBe` "b7ad6b7169203331"
+
+    it "extracts traceparent with tracestate" $ do
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"),
+              ("tracestate", "congo=t61rcWkgMzE,rojo=00f067aa0ba902b7")
+            ]
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "Expected to extract trace context"
+        Just ctx -> do
+          OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "0af7651916cd43dd8448eb211c80319c"
+          OTel.Id.spanIdBaseEncodedText Base16 (OTel.spanId ctx) `shouldBe` "b7ad6b7169203331"
+
+    it "returns Nothing for missing traceparent" $ do
+      let headers :: [(ByteString, ByteString)]
+          headers = []
+      extractTraceContext headers `shouldBe` Nothing
+
+    it "returns Nothing for invalid traceparent format" $ do
+      let headers :: [(ByteString, ByteString)]
+          headers = [("traceparent", "invalid-format")]
+      extractTraceContext headers `shouldBe` Nothing
+
+    it "parses traceparent with non-standard version (permissive)" $ do
+      -- Note: hs-opentelemetry propagator is permissive and accepts non-00 versions
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("traceparent", "ff-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")
+            ]
+      -- Library accepts this - validation is left to the application
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "Library should parse non-standard version"
+        Just _ctx -> pure ()
+
+    it "parses traceparent with all-zero trace ID (permissive)" $ do
+      -- Note: hs-opentelemetry propagator doesn't reject all-zero trace IDs
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("traceparent", "00-00000000000000000000000000000000-b7ad6b7169203331-01")
+            ]
+      -- Library accepts this - validation is left to the application
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "Library should parse all-zero trace ID"
+        Just ctx -> OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "00000000000000000000000000000000"
+
+    it "parses traceparent with all-zero span ID (permissive)" $ do
+      -- Note: hs-opentelemetry propagator doesn't reject all-zero span IDs
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-0000000000000000-01")
+            ]
+      -- Library accepts this - validation is left to the application
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "Library should parse all-zero span ID"
+        Just ctx -> OTel.Id.spanIdBaseEncodedText Base16 (OTel.spanId ctx) `shouldBe` "0000000000000000"
+
+    it "handles unsampled trace flag (00)" $ do
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-00")
+            ]
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "Expected to extract trace context"
+        Just _ctx -> pure () -- Just verify it parses
+    it "handles sampled trace flag (01)" $ do
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")
+            ]
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "Expected to extract trace context"
+        Just _ctx -> pure () -- Just verify it parses
+    it "ignores case-sensitivity of header names" $ do
+      -- W3C spec says header names are case-insensitive
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("Traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")
+            ]
+      -- Note: This depends on how lookup handles case.
+      -- The current implementation uses exact match, so this should return Nothing
+      -- This is a known limitation - header normalization should happen at the adapter level
+      extractTraceContext headers `shouldBe` Nothing
+
+    it "extracts only traceparent when tracestate is malformed" $ do
+      let headers :: [(ByteString, ByteString)]
+          headers =
+            [ ("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"),
+              ("tracestate", "malformed tracestate value!!!")
+            ]
+      -- Should still extract the traceparent even if tracestate is invalid
+      case extractTraceContext headers of
+        Nothing -> expectationFailure "Expected to extract trace context despite malformed tracestate"
+        Just ctx -> do
+          OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "0af7651916cd43dd8448eb211c80319c"