diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,22 @@
 
 ## [Unreleased]
 
+## 0.24.0.0 — 2026-11-05
+
+### Breaking Changes
+
+- `AppHooks` gains `durableConsumers`; direct record constructors must supply it. Updating `emptyAppHooks` remains the preferred embedding pattern.
+
+- Exhaustive matches on `Keiro.Ops.Projection.Command` must handle `Consumers`,
+  and matches on `Keiro.Ops.Workflow.InspectionCommand` must handle `InspectCounts`.
+
+### New Features
+
+- Add embedded read-only `projection consumers --category` with coverage flags, matching bindings, bounded inventory-bound pagination and direct encoding.
+- Add read-only `wf inspect counts [--workflow-name NAME]` with the owning library JSON.
+- `projection status --json` items include sorted `feeding_subscriptions` from
+  registration metadata, empty for inline-only and unmanaged groups.
+
 ## 0.23.0.0 — 2026-10-30
 
 ### Breaking Changes
diff --git a/keiro-ops.cabal b/keiro-ops.cabal
--- a/keiro-ops.cabal
+++ b/keiro-ops.cabal
@@ -1,6 +1,6 @@
 cabal-version: 3.0
 name: keiro-ops
-version: 0.23.0.0
+version: 0.24.0.0
 synopsis: Operational command-line interface for Keiro deployments
 description:
   A standalone database-only console and embeddable command tree for inspecting
@@ -77,9 +77,9 @@
     effectful-core >=2.6 && <2.7 || >=2.7.1.1 && <2.8,
     hasql >=1.10 && <1.11,
     hasql-pool >=1.4 && <1.5,
-    keiro ^>=0.23.0.0,
-    keiro-migrations ^>=0.23.0.0,
-    keiro-pgmq ^>=0.23.0.0,
+    keiro ^>=0.24.0.0,
+    keiro-migrations ^>=0.24.0.0,
+    keiro-pgmq ^>=0.24.0.0,
     kiroku-store ^>=0.13.0.1,
     optparse-applicative >=0.18 && <0.20,
     text >=2.1 && <2.2,
@@ -110,7 +110,10 @@
     -rtsopts
     -with-rtsopts=-N
 
-  other-modules: ReactionInspectionSpec
+  other-modules:
+    CategoryConsumersSpec
+    ReactionInspectionSpec
+
   build-tool-depends: keiro-ops:keiro-ops
   build-depends:
     aeson >=2.2 && <2.3,
@@ -123,10 +126,10 @@
     hasql-transaction >=1.1 && <1.3,
     hspec >=2.11,
     keiki >=0.9.1 && <0.10,
-    keiro ^>=0.23.0.0,
+    keiro ^>=0.24.0.0,
     keiro-ops,
-    keiro-pgmq ^>=0.23.0.0,
-    keiro-test-support ^>=0.23.0.0,
+    keiro-pgmq ^>=0.24.0.0,
+    keiro-test-support ^>=0.24.0.0,
     kiroku-store ^>=0.13.0.1,
     optparse-applicative >=0.18 && <0.20,
     pgmq-migration >=0.7.0.0 && <0.8,
diff --git a/src/Keiro/Ops.hs b/src/Keiro/Ops.hs
--- a/src/Keiro/Ops.hs
+++ b/src/Keiro/Ops.hs
@@ -130,7 +130,7 @@
           (info (Pgmq <$> Pgmq.commandParser) (progDesc "Inspect and operate Keiro PGMQ queues"))
         <> command
           "projection"
-          (info (Projection <$> Projection.commandParser) (progDesc "Inspect and operate projection dedup state"))
+          (info (Projection <$> Projection.commandParserWithConsumers (isJust hooks.projectionCatalog || isJust hooks.durableConsumers)) (progDesc "Inspect and operate projection dedup state"))
         <> command
           "shard"
           (info (Shard <$> Shard.commandParser) (progDesc "Inspect and operate sharded-subscription ownership"))
@@ -240,6 +240,13 @@
     pure $ case encoded of
       Nothing -> (InspectionFailed "inspection_result_too_large" "encoded result exceeds the host limit; narrow the read", Nothing)
       Just body -> (Succeeded result, Just body)
+  Projection (Projection.Consumers options) -> case Projection.prepareConsumers hooks.projectionCatalog hooks.durableConsumers options of
+    Left (code, detail) -> pure (InspectionFailed code detail, Nothing)
+    Right (result, encoding) -> do
+      encoded <- encode encoding
+      pure $ case encoded of
+        Nothing -> (InspectionFailed "inspection_result_too_large" "encoded result exceeds the host limit; narrow the read", Nothing)
+        Just body -> (Succeeded result, Just body)
   _ -> runCommandWithHeaderCache cache hooks env command
 
 -- | Execute a parsed command after schema verification. Front ends must use
@@ -253,7 +260,7 @@
   Outbox outboxCommand -> Outbox.runCommand env outboxCommand
   Inbox inboxCommand -> Inbox.runCommand env inboxCommand
   Pgmq pgmqCommand -> Pgmq.runCommand env pgmqCommand
-  Projection projectionCommand -> Projection.runCommand env projectionCommand
+  Projection projectionCommand -> Projection.runCommandWithConsumers hooks.projectionCatalog hooks.durableConsumers env projectionCommand
   Shard shardCommand -> Shard.runCommand env shardCommand
   Snapshot snapshotCommand -> Snapshot.runCommand env snapshotCommand
   Stream streamCommand -> Stream.runCommand env streamCommand
diff --git a/src/Keiro/Ops/Embed.hs b/src/Keiro/Ops/Embed.hs
--- a/src/Keiro/Ops/Embed.hs
+++ b/src/Keiro/Ops/Embed.hs
@@ -13,6 +13,7 @@
 where
 
 import Keiro.Inspection.Aggregate (AggregateInspectors)
+import Keiro.Inspection.Consumers (DurableConsumerRegistry)
 import Keiro.Ops.ReplayAudit (OpsAuditConfig (..))
 import Keiro.Ops.Timer (TimerFire)
 import Keiro.Ops.Workflow (ResumeHook)
@@ -25,6 +26,7 @@
     replayAudit :: !(Maybe OpsAuditConfig),
     projectionCatalog :: !(Maybe ProjectionCatalogOperations),
     aggregates :: !(Maybe AggregateInspectors),
+    durableConsumers :: !(Maybe DurableConsumerRegistry),
     processManagers :: !(Maybe ProcessManagerInspectors)
   }
 
@@ -36,5 +38,6 @@
       replayAudit = Nothing,
       projectionCatalog = Nothing,
       aggregates = Nothing,
+      durableConsumers = Nothing,
       processManagers = Nothing
     }
diff --git a/src/Keiro/Ops/Projection.hs b/src/Keiro/Ops/Projection.hs
--- a/src/Keiro/Ops/Projection.hs
+++ b/src/Keiro/Ops/Projection.hs
@@ -5,7 +5,12 @@
 module Keiro.Ops.Projection
   ( Command (..),
     ProjectionStatusOptions (..),
+    ConsumersOptions (..),
+    prepareConsumers,
+    commandParserWithConsumers,
+    runCommandWithConsumers,
     projectionGroupStatusJson,
+    projectionGroupStatusRowJson,
     validateCommand,
     commandParser,
     isMutation,
@@ -14,19 +19,28 @@
 where
 
 import Data.Aeson (Value, object, (.=))
+import Data.Aeson qualified as Aeson
+import Data.Aeson.Encoding qualified as Encoding
+import Data.Aeson.KeyMap qualified as KeyMap
+import Data.Aeson.Types (Pair)
 import Data.Int (Int64)
+import Data.List (groupBy)
+import Data.Map.Lazy qualified as LazyMap
+import Data.Maybe (isJust)
 import Data.Text (Text)
 import Data.Text qualified as Text
 import Data.Time (UTCTime, defaultTimeLocale, formatTime, parseTimeM)
 import Data.Vector qualified as Vector
 import Effectful (Eff, IOE)
 import Effectful.Error.Static (Error)
-import Keiro.Inspection.Cursor (InspectionCursor (..))
+import Keiro.Inspection.Consumers qualified as Consumers
+import Keiro.Inspection.Cursor (InspectionCursor (..), decodeCursor, encodeCursorBounded)
 import Keiro.Ops.Env (OpsEnv (..), OutputMode (..))
 import Keiro.Ops.Parse (cursorReader, positiveIntReader)
 import Keiro.Ops.Render
 import Keiro.Projection (countAsyncProjectionDedupForBefore, pruneAsyncProjectionDedupForBefore)
-import Keiro.Projection.Catalog (projectionRevisionIdText, rebuildGroupIdText)
+import Keiro.Projection.Catalog (projectionIdText, projectionRevisionIdText, rebuildGroupIdText, subscriptionIdText)
+import Keiro.Projection.Catalog.Operations qualified as Catalog
 import Keiro.ReadModel (storeHeadPosition)
 import Keiro.ReadModel.Rebuild
 import Kiroku.Store.Effect (Store, runStoreIO)
@@ -37,7 +51,7 @@
     SubscriptionName (..),
     subscriptionCheckpointInventory,
   )
-import Kiroku.Store.Types (GlobalPosition (..))
+import Kiroku.Store.Types (CategoryName (..), GlobalPosition (..))
 import Options.Applicative hiding (action)
 import Prelude
 
@@ -46,18 +60,30 @@
 
 data Command
   = Status !ProjectionStatusOptions
+  | Consumers !ConsumersOptions
   | Position !Text
   | PruneDedup !Text !UTCTime
   deriving stock (Eq, Show)
 
 commandParser :: Parser Command
-commandParser =
+commandParser = commandParserWithConsumers False
+
+commandParserWithConsumers :: Bool -> Parser Command
+commandParserWithConsumers mounted =
   hsubparser
     ( command "status" (info (Status <$> (ProjectionStatusOptions <$> option positiveIntReader (long "limit" <> value 100 <> metavar "N") <*> optional (option cursorReader (long "after" <> metavar "CURSOR")))) (progDesc "Observe one bounded page of projection-group status"))
+        <> (if mounted then command "consumers" (info consumersParser (progDesc "Find declared consumers of one category in mounted configuration")) else mempty)
         <> command "position" (info positionParser (progDesc "Show one subscription's durable member checkpoints and floor"))
         <> command "prune-dedup" (info pruneParser (progDesc "Preview or prune one projection's old dedup rows"))
     )
   where
+    consumersParser =
+      Consumers
+        <$> ( ConsumersOptions
+                <$> textOption "category" "CATEGORY" "Exact stream category (case sensitive)"
+                <*> option positiveIntReader (long "limit" <> value 100 <> metavar "N" <> help "Page size, 1..500")
+                <*> optional (option cursorReader (long "after" <> metavar "CURSOR"))
+            )
     positionParser = Position <$> textOption "subscription" "NAME" "Durable Kiroku subscription name"
     pruneParser =
       PruneDedup
@@ -79,6 +105,7 @@
 
 validateCommand :: Command -> Either (Text, Text) ()
 validateCommand (Status options) = either (Left . statusError) (const (Right ())) (validateProjectionStatusRequest (statusInput options))
+validateCommand (Consumers options) = validateConsumers options
 validateCommand _ = Right ()
 
 statusError :: ProjectionStatusReadError -> (Text, Text)
@@ -92,21 +119,28 @@
 
 isMutation :: Command -> Bool
 isMutation = \case
+  Consumers {} -> False
   Status {} -> False
   Position {} -> False
   PruneDedup {} -> True
 
 runCommand :: OpsEnv -> Command -> IO OpsOutcome
-runCommand env = \case
+runCommand = runCommandWithConsumers Nothing Nothing
+
+runCommandWithConsumers :: Maybe Catalog.ProjectionCatalogOperations -> Maybe Consumers.DurableConsumerRegistry -> OpsEnv -> Command -> IO OpsOutcome
+runCommandWithConsumers catalog durable env = \case
+  Consumers options -> pure $ case prepareConsumers catalog durable options of
+    Left (code, detail) -> InspectionFailed code detail
+    Right (result, _) -> Succeeded result
   Status options -> runAction env action $ \case
     Left err -> uncurry InspectionFailed (statusError err)
     Right (page, head) ->
       Succeeded
         ( OpsResult
             ["group_id", "lifecycle_phase", "reads_allowed", "serving_position_basis"]
-            [[rebuildGroupIdText row.groupId, row.lifecyclePhase, Text.pack (show row.readsAllowed), basisText row.servingPositionBasis] | row <- page.items]
+            [[rebuildGroupIdText row.status.groupId, row.status.lifecyclePhase, Text.pack (show row.status.readsAllowed), basisText row.status.servingPositionBasis] | row <- page.items]
             ( object
-                ( ["observed_at" .= page.observedAt, "visible_store_head" .= positionInt head, "items" .= map projectionGroupStatusJson page.items]
+                ( ["observed_at" .= page.observedAt, "visible_store_head" .= positionInt head, "items" .= map projectionGroupStatusRowJson page.items]
                     <> maybe [] (\cursor -> ["next_cursor" .= cursor.cursorText]) page.next
                 )
             )
@@ -130,28 +164,36 @@
 
 -- | The frozen v1 owner relation, using ADR-28 position vocabulary.
 projectionGroupStatusJson :: ProjectionGroupStatusV1 -> Value
-projectionGroupStatusJson row =
-  object
-    [ "group_id" .= rebuildGroupIdText row.groupId,
-      "lifecycle_phase" .= row.lifecyclePhase,
-      "reads_allowed" .= row.readsAllowed,
-      "writes_allowed" .= row.writesAllowed,
-      "serving_revision_id" .= fmap projectionRevisionIdText row.servingRevisionId,
-      "serving_epoch" .= row.servingEpoch,
-      "serving_position_basis" .= basisText row.servingPositionBasis,
-      "serving_applied_position" .= fmap positionInt row.servingAppliedPosition,
-      "active_run_id" .= fmap rebuildRunIdText row.activeRunId,
-      "candidate_revision_id" .= fmap projectionRevisionIdText row.candidateRevisionId,
-      "candidate_rebuild_position" .= fmap positionInt row.candidateRebuildPosition,
-      "candidate_rebuild_head" .= fmap positionInt row.candidateRebuildHead,
-      "query_models" .= row.queryModels,
-      "rebuild_started_at" .= row.rebuildStartedAt,
-      "last_promoted_at" .= row.lastPromotedAt,
-      "failed_at" .= row.failedAt,
-      "failure_code" .= row.failureCode,
-      "failure_detail" .= row.failureDetail
-    ]
+projectionGroupStatusJson = object . projectionGroupStatusPairs
 
+-- | The bounded page item, including registered feeding subscriptions.
+projectionGroupStatusRowJson :: ProjectionGroupStatusRow -> Value
+projectionGroupStatusRowJson row =
+  -- Prepending avoids traversing the frozen row's pair list.
+  object (("feeding_subscriptions" .= row.feedingSubscriptions) : projectionGroupStatusPairs row.status)
+
+projectionGroupStatusPairs :: ProjectionGroupStatusV1 -> [Pair]
+projectionGroupStatusPairs row =
+  [ "group_id" .= rebuildGroupIdText row.groupId,
+    "lifecycle_phase" .= row.lifecyclePhase,
+    "reads_allowed" .= row.readsAllowed,
+    "writes_allowed" .= row.writesAllowed,
+    "serving_revision_id" .= fmap projectionRevisionIdText row.servingRevisionId,
+    "serving_epoch" .= row.servingEpoch,
+    "serving_position_basis" .= basisText row.servingPositionBasis,
+    "serving_applied_position" .= fmap positionInt row.servingAppliedPosition,
+    "active_run_id" .= fmap rebuildRunIdText row.activeRunId,
+    "candidate_revision_id" .= fmap projectionRevisionIdText row.candidateRevisionId,
+    "candidate_rebuild_position" .= fmap positionInt row.candidateRebuildPosition,
+    "candidate_rebuild_head" .= fmap positionInt row.candidateRebuildHead,
+    "query_models" .= row.queryModels,
+    "rebuild_started_at" .= row.rebuildStartedAt,
+    "last_promoted_at" .= row.lastPromotedAt,
+    "failed_at" .= row.failedAt,
+    "failure_code" .= row.failureCode,
+    "failure_detail" .= row.failureDetail
+  ]
+
 basisText :: ServingPositionBasis -> Text
 basisText = \case ServingPositionAppend -> "append"; ServingPositionCheckpoint -> "checkpoint"; ServingPositionUnmanaged -> "unmanaged"
 
@@ -256,3 +298,170 @@
 
 shellQuote :: Text -> Text
 shellQuote value = "'" <> Text.replace "'" "'\"'\"'" value <> "'"
+
+-- | One category per request; the host byte cap also bounds nested bindings.
+data ConsumersOptions = ConsumersOptions
+  { category :: !Text,
+    limit :: !Int,
+    after :: !(Maybe InspectionCursor)
+  }
+  deriving stock (Eq, Show)
+
+consumersSchema :: Text
+consumersSchema = "keiro/category-consumers/v1"
+
+type ConsumerKey = (Text, Text)
+
+type ConsumerInventory = (Maybe Text, Maybe Text)
+
+type ConsumerCursor = (Text, ConsumerInventory, ConsumerKey)
+
+validateConsumers :: ConsumersOptions -> Either (Text, Text) ()
+validateConsumers options
+  | Text.null options.category || Text.strip options.category /= options.category = Left ("invalid_category", "category must be nonempty and have no surrounding whitespace")
+  | options.limit < 1 || options.limit > 500 = Left ("invalid_page_size", "page size must be 1..500")
+  | otherwise = () <$ consumerCursor options
+
+consumerCursor :: ConsumersOptions -> Either (Text, Text) (Maybe ConsumerCursor)
+consumerCursor options = traverse decode options.after
+  where
+    decode cursor = case decodeCursor consumersSchema cursor of
+      Right payload@(category, _, (kind, name))
+        | category == options.category,
+          kind `elem` ["durable_subscriber", "process_manager", "projection_group"],
+          not (Text.null name) ->
+            Right payload
+      _ -> Left ("invalid_cursor", "cursor is malformed or belongs to another category/read")
+
+data ConsumerItem = ConsumerItem !Text !Text !(Maybe Text) ![ConsumerBinding]
+
+data ConsumerBinding = ConsumerBinding !(Maybe Text) !(Maybe Text) !(Maybe Text) !(Maybe Text) !Consumers.ConsumerSource
+
+consumerKey :: ConsumerItem -> ConsumerKey
+consumerKey (ConsumerItem kind name _ _) = (kind, name)
+
+-- | One filtered page supplies both presentations. On HTTP the canonical items
+-- array stays lazy, and every request encodes afresh inside the host's budget.
+prepareConsumers :: Maybe Catalog.ProjectionCatalogOperations -> Maybe Consumers.DurableConsumerRegistry -> ConsumersOptions -> Either (Text, Text) (OpsResult, Encoding.Encoding)
+prepareConsumers catalog durable options = do
+  validateConsumers options
+  if not (isJust catalog || isJust durable)
+    then Left ("unsupported_command", "consumer metadata hooks are not mounted")
+    else Right ()
+  cursor <- consumerCursor options
+  let inventory = (Catalog.projectionConsumerInventoryIdentity <$> catalog, Consumers.durableConsumerInventoryIdentity <$> durable)
+  case cursor of
+    Just (_, previous, _) | previous /= inventory -> Left ("invalid_cursor", "mounted consumer inventory changed; restart from the first page")
+    _ -> Right ()
+  let matching = maybe [] (durableItems options.category) durable <> maybe [] (projectionItems options.category) catalog
+      remaining = case cursor of
+        Nothing -> matching
+        Just (_, _, lastKey) -> dropWhile ((<= lastKey) . consumerKey) matching
+      window = take (options.limit + 1) remaining
+      page = take options.limit window
+  next <-
+    if length window <= options.limit
+      then Right Nothing
+      else case encodeCursorBounded consumersSchema (options.category, inventory, consumerKey (last page)) of
+        Nothing -> Left ("inspection_result_too_large", "consumer continuation exceeds cursor byte limit")
+        Just token -> Right (Just token)
+  let coverage = object ["projection_catalog" .= isJust catalog, "durable_subscribers" .= isJust durable]
+      fields =
+        [("schema", Aeson.String consumersSchema), ("category", Aeson.String options.category), ("scope", Aeson.String "mounted_configuration"), ("coverage", coverage), ("items", Aeson.toJSON page)]
+          <> maybe [] (\token -> [("next_cursor", Aeson.String token.cursorText)]) next
+      canonical = Aeson.Object (KeyMap.fromMap (LazyMap.fromList fields))
+      encoding =
+        Encoding.pairs
+          ( Encoding.pair "schema" (Encoding.text consumersSchema)
+              <> Encoding.pair "category" (Encoding.text options.category)
+              <> Encoding.pair "scope" (Encoding.text "mounted_configuration")
+              <> Encoding.pair "coverage" (Aeson.toEncoding coverage)
+              <> Encoding.pair "items" (Aeson.toEncoding page)
+              <> foldMap (Encoding.pair "next_cursor" . Encoding.text . (.cursorText)) next
+          )
+  pure (OpsResult ["kind", "name", "group_id", "subscription", "source"] (concatMap consumerRows page) canonical, encoding)
+
+projectionItems :: Text -> Catalog.ProjectionCatalogOperations -> [ConsumerItem]
+projectionItems category catalog =
+  [ ConsumerItem
+      "projection_group"
+      (rebuildGroupIdText group.groupId)
+      (Just (rebuildGroupIdText group.groupId))
+      [ ConsumerBinding
+          (Just (projectionIdText binding.projectionId))
+          (Just binding.handlerName)
+          (subscriptionIdText <$> binding.subscriptionId)
+          binding.subscriptionName
+          binding.source
+      | binding <- group.bindings
+      ]
+  | group <- Catalog.projectionConsumers catalog (CategoryName category)
+  ]
+
+durableItems :: Text -> Consumers.DurableConsumerRegistry -> [ConsumerItem]
+durableItems category registry =
+  [ ConsumerItem
+      (kindText first.kind)
+      first.name
+      Nothing
+      [ConsumerBinding Nothing Nothing Nothing (Just description.subscriptionName) description.source | description <- descriptions]
+  | descriptions@(first : _) <- groupBy sameConsumer (Consumers.durableConsumersForCategory registry (CategoryName category))
+  ]
+  where
+    sameConsumer left right = (left.kind, left.name) == (right.kind, right.name)
+    kindText Consumers.DurableSubscriber = "durable_subscriber"
+    kindText Consumers.ProcessManagerConsumer = "process_manager"
+
+instance Aeson.ToJSON ConsumerItem where
+  toJSON (ConsumerItem kind name group bindings) = object ["kind" .= kind, "name" .= name, "group_id" .= group, "bindings" .= bindings]
+  toEncoding (ConsumerItem kind name group bindings) =
+    Encoding.pairs
+      ( Encoding.pair "kind" (Encoding.text kind)
+          <> Encoding.pair "name" (Encoding.text name)
+          <> Encoding.pair "group_id" (Aeson.toEncoding group)
+          <> Encoding.pair "bindings" (Aeson.toEncoding bindings)
+      )
+
+instance Aeson.ToJSON ConsumerBinding where
+  toJSON (ConsumerBinding projection handler subscriptionId subscription source) =
+    object
+      [ "projection_id" .= projection,
+        "handler_name" .= handler,
+        "delivery" .= delivery subscription,
+        "subscription_id" .= subscriptionId,
+        "subscription_name" .= subscription,
+        "source" .= sourceJson source
+      ]
+  toEncoding (ConsumerBinding projection handler subscriptionId subscription source) =
+    Encoding.pairs
+      ( Encoding.pair "projection_id" (Aeson.toEncoding projection)
+          <> Encoding.pair "handler_name" (Aeson.toEncoding handler)
+          <> Encoding.pair "delivery" (Encoding.text (delivery subscription))
+          <> Encoding.pair "subscription_id" (Aeson.toEncoding subscriptionId)
+          <> Encoding.pair "subscription_name" (Aeson.toEncoding subscription)
+          <> Encoding.pair "source" (sourceEncoding source)
+      )
+
+delivery :: Maybe Text -> Text
+delivery Nothing = "inline"
+delivery (Just _) = "subscription"
+
+sourceJson :: Consumers.ConsumerSource -> Value
+sourceJson = \case
+  Consumers.AllCategories -> object ["kind" .= ("all" :: Text)]
+  Consumers.ExactCategory category -> object ["kind" .= ("category" :: Text), "category" .= category]
+  Consumers.CategoryNamespace namespace -> object ["kind" .= ("namespace" :: Text), "namespace" .= namespace]
+
+sourceEncoding :: Consumers.ConsumerSource -> Encoding.Encoding
+sourceEncoding source = Encoding.pairs $ case source of
+  Consumers.AllCategories -> Encoding.pair "kind" (Encoding.text "all")
+  Consumers.ExactCategory category -> Encoding.pair "kind" (Encoding.text "category") <> Encoding.pair "category" (Encoding.text category)
+  Consumers.CategoryNamespace namespace -> Encoding.pair "kind" (Encoding.text "namespace") <> Encoding.pair "namespace" (Encoding.text namespace)
+
+consumerRows :: ConsumerItem -> [[Text]]
+consumerRows (ConsumerItem kind name group bindings) =
+  [[kind, name, maybe "" id group, maybe "" id subscription, sourceText source] | ConsumerBinding _ _ _ subscription source <- bindings]
+  where
+    sourceText Consumers.AllCategories = "all categories"
+    sourceText (Consumers.ExactCategory category) = "category: " <> category
+    sourceText (Consumers.CategoryNamespace namespace) = "namespace: " <> namespace
diff --git a/src/Keiro/Ops/Workflow.hs b/src/Keiro/Ops/Workflow.hs
--- a/src/Keiro/Ops/Workflow.hs
+++ b/src/Keiro/Ops/Workflow.hs
@@ -122,6 +122,7 @@
 
 data InspectionCommand
   = InspectList !Inspection.WorkflowListQuery
+  | InspectCounts !Inspection.WorkflowCountsQuery
   | InspectHeader !WorkflowRef
   | InspectJournal !WorkflowRef !Inspection.DetailPageQuery
   | InspectSteps !WorkflowRef !Inspection.DetailPageQuery
@@ -184,6 +185,7 @@
 inspectionParser =
   hsubparser
     ( command "list" (info (InspectList <$> listQuery) (progDesc "Page workflow instances"))
+        <> command "counts" (info (InspectCounts . Inspection.WorkflowCountsQuery <$> optional (Text.pack <$> strOption (long "workflow-name" <> metavar "NAME"))) (progDesc "Count current instances by status class"))
         <> command "header" (info (InspectHeader <$> workflowRefParser) (progDesc "Read only the instance header"))
         <> command "journal" (info (InspectJournal <$> workflowRefParser <*> detailQuery) (progDesc "Page one journal generation"))
         <> command "steps" (info (InspectSteps <$> workflowRefParser <*> detailQuery) (progDesc "Page one generation's step index"))
@@ -313,6 +315,7 @@
 runInspection :: OpsEnv -> InspectionCommand -> IO OpsOutcome
 runInspection env = \case
   InspectList query -> page (Inspection.listWorkflowInstancesPage query)
+  InspectCounts query -> runAction env (Inspection.countWorkflowInstancesByStatus query) (Succeeded . jsonResult . Aeson.toJSON)
   InspectHeader ref -> runAction env (Instance.lookupInstance (refName ref) (refId ref)) (headerOutcome ref)
   InspectJournal ref query -> page (Inspection.readWorkflowJournalPage (refName ref) (refId ref) query)
   InspectSteps ref query -> page (Inspection.loadStepIndexPage (refName ref) (refId ref) query)
@@ -350,6 +353,7 @@
 validateCommand :: Command -> Either (Text, Text) ()
 validateCommand (Inspect command) = either (Left . inspectionError) Right $ case command of
   InspectList query -> Inspection.validateWorkflowListQuery query
+  InspectCounts _ -> Right ()
   InspectHeader _ -> Right ()
   InspectJournal ref query -> Inspection.validateJournalQuery (refName ref) (refId ref) query
   InspectSteps ref query -> Inspection.validateStepQuery (refName ref) (refId ref) query
diff --git a/test/CategoryConsumersSpec.hs b/test/CategoryConsumersSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/CategoryConsumersSpec.hs
@@ -0,0 +1,115 @@
+module CategoryConsumersSpec (spec) where
+
+import Control.Exception (finally)
+import Control.Monad (forM_, when)
+import Data.Aeson (Value (..), object, (.=))
+import Data.Aeson qualified as Aeson
+import Data.Aeson.Encoding qualified as Encoding
+import Data.Aeson.KeyMap qualified as KeyMap
+import Data.ByteString.Lazy qualified as Lazy
+import Data.IORef
+import Data.Text (Text)
+import Data.Text qualified as Text
+import Data.Vector qualified as Vector
+import Keiro.Inspection.Consumers qualified as C
+import Keiro.Inspection.Cursor
+import Keiro.Ops qualified as Ops
+import Keiro.Ops.Env
+import Keiro.Ops.Projection qualified as P
+import Keiro.Ops.Render
+import Keiro.Ops.WorkflowHeaderCache
+import Keiro.Test.Postgres
+import Kiroku.Store.Connection (ConnectionSettingsM (..))
+import Kiroku.Store.Effect (runStoreIO)
+import Kiroku.Store.Read (getStream)
+import Kiroku.Store.Types (StreamName (..))
+import Options.Applicative qualified as Opt
+import Test.Hspec
+
+spec :: Fixture -> Spec
+spec fixture = describe "category consumers commands" $ do
+  it "mounts for either metadata hook and distinguishes empty from missing coverage" $ do
+    let empty = registry []
+        hooks = Ops.emptyAppHooks {Ops.durableConsumers = Just empty}
+        args = ["projection", "consumers", "--category", "order"]
+        parses mounted = case Opt.execParserPure Opt.defaultPrefs (Opt.info (Ops.commandParser mounted) mempty) args of Opt.Success _ -> True; _ -> False
+    parses Ops.emptyAppHooks `shouldBe` False
+    parses hooks `shouldBe` True
+    Right (result, _) <- pure (P.prepareConsumers Nothing (Just empty) (options "order"))
+    field "items" result.jsonValue `shouldBe` Aeson.toJSON ([] :: [Value])
+    field "coverage" result.jsonValue `shouldBe` object ["projection_catalog" .= False, "durable_subscribers" .= True]
+    P.isMutation (P.Consumers (options "order")) `shouldBe` False
+  it "pages a deterministic sequence without omissions and binds tokens to category and inventory" $ do
+    let mounted = registry [description "z" "s-z", description "a" "s-a", description "m" "s-m", description "a" "s-a2"]
+        prepare request = P.prepareConsumers Nothing (Just mounted) request
+        firstRequest = P.ConsumersOptions "order" 1 Nothing
+    Right (first, _) <- pure (prepare firstRequest)
+    names first `shouldBe` ["a"]
+    let String token = field "next_cursor" first.jsonValue
+    Right (second, _) <- pure (prepare firstRequest {P.after = Just (InspectionCursor token)})
+    names second `shouldBe` ["m"]
+    let String next = field "next_cursor" second.jsonValue
+    Right (last, _) <- pure (prepare firstRequest {P.after = Just (InspectionCursor next)})
+    names last `shouldBe` ["z"]
+    field "next_cursor" last.jsonValue `shouldBe` Null
+    forM_ [first, second, last] $ \result -> Text.length (renderHuman result) `shouldSatisfy` (> 0)
+    let changed = registry [description "new" "s-new"]
+    failureCode (prepare firstRequest {P.category = "invoice", P.after = Just (InspectionCursor token)}) `shouldBe` Just "invalid_cursor"
+    failureCode (P.prepareConsumers Nothing (Just changed) firstRequest {P.after = Just (InspectionCursor token)}) `shouldBe` Just "invalid_cursor"
+    forM_ [InspectionCursor "bad", InspectionCursor (Text.replicate 4097 "a"), encodeCursor "other-kind" ("x" :: Text)] $ \cursor ->
+      P.validateCommand (P.Consumers firstRequest {P.after = Just cursor}) `shouldSatisfy` (\case Left ("invalid_cursor", _) -> True; _ -> False)
+    forM_ [0, 501] $ \limit -> failureCode (prepare firstRequest {P.limit = limit}) `shouldBe` Just "invalid_page_size"
+    forM_ ["", " order", "order "] $ \category -> failureCode (prepare firstRequest {P.category = category}) `shouldBe` Just "invalid_category"
+    let huge = registry [description (Text.replicate 5000 "x") "one", description (Text.replicate 5001 "x") "two"]
+    failureCode (P.prepareConsumers Nothing (Just huge) firstRequest) `shouldBe` Just "inspection_result_too_large"
+  it "renders and directly encodes afresh without touching the store or header cache" $ do
+    armed <- newIORef False
+    let observe _ = readIORef armed >>= \active -> when active (fail "category consumers touched the store")
+    withFreshResourceStoreWith fixture (\settings -> settings {observationHandler = Just observe}) $ \(unusable, _) ->
+      flip finally (writeIORef armed False) $ do
+        writeIORef armed True
+        let mounted = registry [description "escaped-\"\\\n界" "subscription", C.DurableConsumerDescription C.ProcessManagerConsumer "fulfillment" "orders" (C.ExactCategory "order")]
+            hooks = Ops.emptyAppHooks {Ops.durableConsumers = Just mounted}
+            env = OpsEnv unusable HumanTable False [] False
+            command = Ops.Projection (P.Consumers (options "order"))
+        runStoreIO unusable (getStream (StreamName "sentinel")) `shouldThrow` anyIOException
+        Succeeded result <- Ops.runCommand hooks env command
+        Text.length (renderHuman result) `shouldSatisfy` (> 0)
+        Text.length (jsonText result.jsonValue) `shouldSatisfy` (> 0)
+        cache <- newWorkflowHeaderCache 1048576 (const (fail "category consumers used the header cache"))
+        calls <- newIORef (0 :: Int)
+        let encode encoding = modifyIORef' calls (+ 1) >> pure (Just (Lazy.toStrict (Encoding.encodingToLazyByteString encoding)))
+        forM_ [1, 2 :: Int] $ \_ -> do
+          (Succeeded direct, Just bytes) <- Ops.runCommandWithReadEncoding encode cache hooks env command
+          direct `shouldBe` result
+          Aeson.eitherDecodeStrict bytes `shouldBe` Right result.jsonValue
+        readIORef calls `shouldReturn` 2
+        workflowHeaderCacheUsage cache `shouldReturn` (0, 0)
+        (refused, bytes) <- Ops.runCommandWithReadEncoding (const (pure Nothing)) cache hooks env command
+        refused `shouldSatisfy` (\case InspectionFailed "inspection_result_too_large" _ -> True; _ -> False)
+        bytes `shouldBe` Nothing
+        forM_ [0, 501] $ \limit -> do
+          invalid <- Ops.runCommand hooks env (Ops.Projection (P.Consumers (options "order") {P.limit = limit}))
+          invalid `shouldSatisfy` (\case InspectionFailed "invalid_page_size" _ -> True; _ -> False)
+
+options :: Text -> P.ConsumersOptions
+options category = P.ConsumersOptions category 100 Nothing
+
+description :: Text -> Text -> C.DurableConsumerDescription
+description name subscription = C.DurableConsumerDescription C.DurableSubscriber name subscription (C.ExactCategory "order")
+
+registry :: [C.DurableConsumerDescription] -> C.DurableConsumerRegistry
+registry = either (error . show) id . C.durableConsumerRegistry
+
+field :: Aeson.Key -> Value -> Value
+field key (Object fields) = maybe Null id (KeyMap.lookup key fields)
+field _ _ = Null
+
+names :: OpsResult -> [Text]
+names result = case field "items" result.jsonValue of
+  Array values -> [name | String name <- map (field "name") (Vector.toList values)]
+  _ -> []
+
+failureCode :: Either (Text, Text) a -> Maybe Text
+failureCode (Left (code, _)) = Just code
+failureCode _ = Nothing
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -2,7 +2,9 @@
 
 module Main (main) where
 
+import CategoryConsumersSpec qualified
 import Control.Exception (bracket)
+import Control.Monad (forM_)
 import Data.Aeson (object, (.=))
 import Data.Aeson qualified as Aeson
 import Data.Aeson.Key (Key)
@@ -60,6 +62,7 @@
 import Keiro.Workflow (StepName (..), WorkflowId (..), WorkflowJournalEvent (..), WorkflowName (..), appendJournalEntry)
 import Keiro.Workflow.Awakeable (AwakeableId (..))
 import Keiro.Workflow.Awakeable.Schema qualified as Awakeable
+import Keiro.Workflow.Inspection qualified as Inspection
 import Keiro.Workflow.Instance qualified as Instance
 import Keiro.Workflow.Resume (WorkflowDef (..), defaultWorkflowResumeOptions)
 import Keiro.Workflow.Sleep (sleepNamed)
@@ -93,6 +96,7 @@
       replayAudit = Just (OpsReplayAudit.OpsAuditConfig []),
       projectionCatalog = Just emptyCatalogOperations,
       aggregates = Just Map.empty,
+      durableConsumers = Nothing,
       processManagers = Just Map.empty
     }
 
@@ -115,6 +119,7 @@
 
 spec :: Fixture -> Spec
 spec fixture = do
+  CategoryConsumersSpec.spec fixture
   ReactionInspectionSpec.spec fixture
   describe "embedded command tree" do
     it "omits code-dependent commands from the standalone tree" do
@@ -204,6 +209,25 @@
       mapM_ (\args -> isParseFailure (parseOps embeddedHooks args) `shouldBe` True) [rebuild "-1", snapshot "-1", stream "-1", workflow "-1"]
       mapM_ (\args -> isParseSuccess (parseOps embeddedHooks args) `shouldBe` True) [rebuild "0", snapshot "0", stream "0", workflow "0"]
 
+  describe "workflow counts command" $ around (withFreshStore fixture) $ do
+    it "parses only the exact-name filter and remains read-only" $ \_ -> do
+      forM_ [["wf", "inspect", "counts"], ["wf", "inspect", "counts", "--workflow-name", "支払_%'\\ λ"]] $ \args ->
+        isParseSuccess (parseOps Ops.emptyAppHooks args) `shouldBe` True
+      forM_ ["--limit=1", "--after=cursor", "--status-class=running"] $ \option ->
+        isParseFailure (parseOps Ops.emptyAppHooks ["wf", "inspect", "counts", option]) `shouldBe` True
+      let command = OpsWorkflow.Inspect (OpsWorkflow.InspectCounts (Inspection.WorkflowCountsQuery Nothing))
+      OpsWorkflow.isMutation command `shouldBe` False
+      OpsWorkflow.validateCommand command `shouldBe` Right ()
+    it "renders the owning library JSON in the command result" $ \store -> do
+      let query = Inspection.WorkflowCountsQuery (Just "missing")
+      expected <- runStoreIO store (Inspection.countWorkflowInstancesByStatus query) >>= either (fail . show) pure
+      outcome <- OpsWorkflow.runCommand (opsEnv False store) (OpsWorkflow.Inspect (OpsWorkflow.InspectCounts query))
+      case outcome of
+        Succeeded result -> case result.jsonValue of
+          Aeson.Object fields -> KeyMap.lookup "counts" fields `shouldBe` Just (Aeson.toJSON expected.counts)
+          other -> expectationFailure (show other)
+        other -> expectationFailure (show other)
+
   describe "targeted stream repair command" $
     around (withFreshStore fixture) $ do
       it "requires a positive event admission limit" $ \_ -> do
@@ -1302,11 +1326,18 @@
       catalog <- expectValidatedCatalog (opsCatalog "status-codec")
       _ <- expectStore store (Rebuild.registerProjectionCatalog catalog)
       Right page <- expectStore store (Rebuild.listProjectionGroupStatusesPage (Rebuild.ProjectionStatusPageRequest 1 Nothing))
+      map (.feedingSubscriptions) page.items `shouldBe` [[]]
+      map (.status.servingPositionBasis) page.items `shouldBe` [Rebuild.ServingPositionAppend]
+      forM_ page.items $ \item -> case OpsProjection.projectionGroupStatusRowJson item of
+        Aeson.Object fields -> do
+          KeyMap.lookup "feeding_subscriptions" fields `shouldBe` Just (Aeson.toJSON ([] :: [Text]))
+          Aeson.Object (KeyMap.delete "feeding_subscriptions" fields) `shouldBe` OpsProjection.projectionGroupStatusJson item.status
+        _ -> expectationFailure "expected status item object"
       shown <- OpsProjection.runCommand (opsEnv False store) (OpsProjection.Status (OpsProjection.ProjectionStatusOptions 1 Nothing))
       case shown of
         Succeeded result -> case result.jsonValue of
           Aeson.Object fields -> do
-            KeyMap.lookup "items" fields `shouldBe` Just (Aeson.toJSON (map OpsProjection.projectionGroupStatusJson page.items))
+            KeyMap.lookup "items" fields `shouldBe` Just (Aeson.toJSON (map OpsProjection.projectionGroupStatusRowJson page.items))
             KeyMap.member "observed_at" fields `shouldBe` True
             KeyMap.lookup "visible_store_head" fields `shouldBe` Just (Aeson.toJSON (0 :: Int))
             KeyMap.lookup "next_cursor" fields `shouldBe` Nothing
@@ -1389,6 +1420,28 @@
       replacementOwners `shouldSatisfy` all (\(_, owner, _) -> owner == Just replacement)
 
   describe "snapshot handlers" $ around (withFreshStore fixture) do
+    it "preserves custom incompatible payloads and returns null for missing snapshots" $ \store -> do
+      appended <- seedKirokuEvent store "snapshot-raw" "018f5f43-8a70-7b9a-9a9b-59d391a76810" Nothing
+      forM_ ["missing-snapshot-stream", "snapshot-raw"] $ \name -> do
+        Succeeded result <- OpsSnapshot.runCommand (opsEnv False store) (OpsSnapshot.Show name)
+        result.jsonValue `shouldBe` Aeson.Null
+      let payload = object ["custom" .= [object ["secret" .= True]], "legacy" .= ("unrecognized shape" :: Text)]
+      expectStore store $ writeSnapshotRow (SnapshotWrite appended.streamId appended.streamVersion payload 999 "old-registers" "old-fold")
+      before <- expectStore store (lookupSnapshotRow appended.streamId)
+      streamBefore <- expectStore store (getStream (StreamName "snapshot-raw"))
+      Succeeded result <- OpsSnapshot.runCommand (opsEnv False store) (OpsSnapshot.Show "snapshot-raw")
+      case result.jsonValue of
+        Aeson.Object fields -> do
+          KeyMap.lookup "state" fields `shouldBe` Just payload
+          KeyMap.lookup "stream_version" fields `shouldBe` Just (Aeson.toJSON (1 :: Int))
+          KeyMap.lookup "state_codec_version" fields `shouldBe` Just (Aeson.toJSON (999 :: Int))
+          KeyMap.lookup "regfile_shape_hash" fields `shouldBe` Just (Aeson.String "old-registers")
+          KeyMap.lookup "state_shape_hash" fields `shouldBe` Just (Aeson.String "old-fold")
+          KeyMap.member "compatibility" fields `shouldBe` False
+        _ -> expectationFailure "expected raw snapshot object"
+      expectStore store (lookupSnapshotRow appended.streamId) `shouldReturn` before
+      expectStore store (getStream (StreamName "snapshot-raw")) `shouldReturn` streamBefore
+
     it "refuses uncovered truncation, passes matching coverage, and deletes advisories" $ \store -> do
       appended <- seedKirokuEvent store "snapshot-ops" "018f5f43-8a70-7b9a-9a9b-59d391a76811" Nothing
       let expected = OpsSnapshot.ExpectedDiscriminators 7 "regs-v7" "fold-v7"