diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,20 @@
 
 ## [Unreleased]
 
+## 0.24.0.0 — 2026-11-05
+
+### Breaking Changes
+
+- `ProjectionStatusPage.items` now contains `ProjectionGroupStatusRow`; access the
+  unchanged frozen v1 record through `status`.
+
+### New Features
+
+- Add pure `projectionConsumers` and validated `Keiro.Inspection.Consumers` durable-input metadata with exact, all-category and namespace scopes.
+- Add `countWorkflowInstancesByStatus` with exact seven-class current-header counts, optional exact-name filtering and canonical 64-bit JSON, without schema or writer changes.
+- Bounded projection status items carry registered `feedingSubscriptions` beside the
+  unchanged v1 row; complete-list and lookup APIs retain their contracts.
+
 ## 0.23.0.0 — 2026-10-30
 
 ### Breaking Changes
diff --git a/keiro.cabal b/keiro.cabal
--- a/keiro.cabal
+++ b/keiro.cabal
@@ -1,6 +1,6 @@
 cabal-version: 3.0
 name: keiro
-version: 0.23.0.0
+version: 0.24.0.0
 synopsis: Event sourcing framework and workflow engine
 description:
   A library that composes kiroku, keiki, and shibuya into an
@@ -71,6 +71,7 @@
     Keiro.Inbox.Schema
     Keiro.Inbox.Types
     Keiro.Inspection.Aggregate
+    Keiro.Inspection.Consumers
     Keiro.Inspection.Cursor
     Keiro.Outbox
     Keiro.Outbox.Identity
@@ -176,7 +177,7 @@
     hs-opentelemetry-semantic-conventions >=1.40 && <2,
     keiki >=0.9 && <0.10,
     keiki-codec-json >=0.9 && <0.10,
-    keiro-core ^>=0.23.0.0,
+    keiro-core ^>=0.24.0.0,
     kiroku-store ^>=0.13.0.1,
     lens >=5.2 && <5.4,
     mmzk-typeid >=0.7 && <0.8,
@@ -203,6 +204,7 @@
     CatalogOperationsSpec
     CatalogSpec
     Compatibility.ReadModel011
+    ConsumerInspectionSpec
     ExternalReadSpec
     GroupRebuildSpec
     GuardedCatalogExample
@@ -247,8 +249,8 @@
     keiki,
     keiki-codec-json,
     keiro,
-    keiro-migrations ^>=0.23.0.0,
-    keiro-test-support ^>=0.23.0.0,
+    keiro-migrations ^>=0.24.0.0,
+    keiro-test-support ^>=0.24.0.0,
     kiroku-store ^>=0.13.0.1,
     process >=1.6 && <1.7,
     shibuya-core ^>=0.12.0.0,
@@ -284,8 +286,8 @@
     keiki >=0.9 && <0.10,
     keiki-codec-json >=0.9 && <0.10,
     keiro,
-    keiro-core ^>=0.23.0.0,
-    keiro-test-support ^>=0.23.0.0,
+    keiro-core ^>=0.24.0.0,
+    keiro-test-support ^>=0.24.0.0,
     kiroku-store ^>=0.13.0.1,
     shibuya-core ^>=0.12.0.0,
     shibuya-kiroku-adapter ^>=0.7.0.1,
@@ -334,8 +336,8 @@
     hs-opentelemetry-sdk >=1.0 && <1.1,
     keiki >=0.9 && <0.10,
     keiro,
-    keiro-core ^>=0.23.0.0,
-    keiro-test-support ^>=0.23.0.0,
+    keiro-core ^>=0.24.0.0,
+    keiro-test-support ^>=0.24.0.0,
     kiroku-store ^>=0.13.0.1,
     process >=1.6 && <1.7,
     shibuya-core ^>=0.12.0.0,
diff --git a/src/Keiro/Inspection/Consumers.hs b/src/Keiro/Inspection/Consumers.hs
new file mode 100644
--- /dev/null
+++ b/src/Keiro/Inspection/Consumers.hs
@@ -0,0 +1,99 @@
+{-# LANGUAGE OverloadedRecordDot #-}
+
+-- | Data-only declarations of durable inputs. Hosts should construct these from
+-- the same configuration used by their workers; opaque adapters cannot be
+-- inferred from a process manager's journal category.
+module Keiro.Inspection.Consumers
+  ( ConsumerSource (..),
+    consumerSourceMatches,
+    DurableConsumerKind (..),
+    DurableConsumerDescription (..),
+    DurableConsumerRegistry,
+    DurableConsumerError (..),
+    durableConsumerRegistry,
+    durableConsumerInventoryIdentity,
+    durableConsumersForCategory,
+  )
+where
+
+import Data.List (foldl')
+import Data.Map.Strict qualified as Map
+import Data.Set qualified as Set
+import Data.Text (Text)
+import Data.Text qualified as Text
+import Keiro.Projection.Catalog.Preimage
+import Kiroku.Store.Types (CategoryName (..))
+
+data ConsumerSource = AllCategories | ExactCategory !Text | CategoryNamespace !Text
+  deriving stock (Eq, Ord, Show)
+
+consumerSourceMatches :: CategoryName -> ConsumerSource -> Bool
+consumerSourceMatches (CategoryName category) = \case
+  AllCategories -> True
+  ExactCategory declared -> category == declared
+  CategoryNamespace namespace -> category == namespace || Text.isPrefixOf (namespace <> ":") category
+
+-- | Constructor order is the public discovery order.
+data DurableConsumerKind = DurableSubscriber | ProcessManagerConsumer
+  deriving stock (Eq, Ord, Show)
+
+data DurableConsumerDescription = DurableConsumerDescription
+  { kind :: !DurableConsumerKind,
+    name :: !Text,
+    subscriptionName :: !Text,
+    source :: !ConsumerSource
+  }
+  deriving stock (Eq, Ord, Show)
+
+data DurableConsumerError
+  = InvalidConsumerDeclaration !Text
+  | ConflictingSubscriptionScope !Text
+  deriving stock (Eq, Show)
+
+-- Descriptions are deduplicated and ordered once, and the digest is forced at
+-- construction. Neither requests nor response bytes are retained here.
+data DurableConsumerRegistry = DurableConsumerRegistry !Text ![DurableConsumerDescription]
+  deriving stock (Eq, Show)
+
+durableConsumerRegistry :: [DurableConsumerDescription] -> Either DurableConsumerError DurableConsumerRegistry
+durableConsumerRegistry descriptions = do
+  mapM_ validate descriptions
+  _ <- foldl' addScope (Right Map.empty) descriptions
+  let ordered = Set.toAscList (Set.fromList descriptions)
+      digest = hashPreimage "keiro/durable-consumers/v1" (PList (map descriptionPreimage ordered))
+  pure (DurableConsumerRegistry digest ordered)
+  where
+    validText text = not (Text.null text) && Text.strip text == text
+    require label text
+      | validText text = Right ()
+      | otherwise = Left (InvalidConsumerDeclaration label)
+    validate description = do
+      require "consumer name" description.name
+      require "subscription name" description.subscriptionName
+      case description.source of
+        AllCategories -> Right ()
+        ExactCategory category -> require "category" category
+        CategoryNamespace namespace -> require "namespace" namespace
+    addScope result description = do
+      scopes <- result
+      case Map.lookup description.subscriptionName scopes of
+        Just scope | scope /= description.source -> Left (ConflictingSubscriptionScope description.subscriptionName)
+        _ -> Right (Map.insert description.subscriptionName description.source scopes)
+    descriptionPreimage description =
+      PRecord
+        "consumer"
+        [ PText (case description.kind of DurableSubscriber -> "durable_subscriber"; ProcessManagerConsumer -> "process_manager"),
+          PText description.name,
+          PText description.subscriptionName,
+          case description.source of
+            AllCategories -> PRecord "all" []
+            ExactCategory category -> PRecord "category" [PText category]
+            CategoryNamespace namespace -> PRecord "namespace" [PText namespace]
+        ]
+
+durableConsumerInventoryIdentity :: DurableConsumerRegistry -> Text
+durableConsumerInventoryIdentity (DurableConsumerRegistry digest _) = digest
+
+durableConsumersForCategory :: DurableConsumerRegistry -> CategoryName -> [DurableConsumerDescription]
+durableConsumersForCategory (DurableConsumerRegistry _ descriptions) category =
+  filter (consumerSourceMatches category . (.source)) descriptions
diff --git a/src/Keiro/Projection/Catalog/Operations.hs b/src/Keiro/Projection/Catalog/Operations.hs
--- a/src/Keiro/Projection/Catalog/Operations.hs
+++ b/src/Keiro/Projection/Catalog/Operations.hs
@@ -1,9 +1,15 @@
+{-# LANGUAGE OverloadedRecordDot #-}
+
 -- | Operator-neutral inventory and rebuild actions derived from one validated
 -- projection catalog. This module deliberately contains no command parser,
 -- renderer, confirmation policy, or database connection details.
 module Keiro.Projection.Catalog.Operations
   ( ProjectionCatalogOperations,
     projectionCatalogOperations,
+    ProjectionConsumerGroup (..),
+    ProjectionConsumerBinding (..),
+    projectionConsumers,
+    projectionConsumerInventoryIdentity,
     CatalogInventoryReport (..),
     RebuildPreview (..),
     RegisteredRebuildPreview (..),
@@ -56,6 +62,7 @@
 import Data.Time (DiffTime)
 import Data.UUID qualified as UUID
 import Effectful (Eff, IOE, (:>))
+import Keiro.Inspection.Consumers (ConsumerSource (..), consumerSourceMatches)
 import Keiro.Prelude
 import Keiro.Projection.Catalog
 import Keiro.ReadModel (HeadScope (..))
@@ -122,6 +129,56 @@
 
 projectionCatalogOperations :: ValidatedProjectionCatalog -> ProjectionCatalogOperations
 projectionCatalogOperations = ProjectionCatalogOperations
+
+-- | Matching declarations, grouped by the shared rebuild lifecycle. These
+-- describe source selection, never decoder/handler relevance or worker liveness.
+data ProjectionConsumerGroup = ProjectionConsumerGroup
+  { groupId :: !RebuildGroupId,
+    bindings :: ![ProjectionConsumerBinding]
+  }
+  deriving stock (Eq, Show, Generic)
+
+data ProjectionConsumerBinding = ProjectionConsumerBinding
+  { projectionId :: !ProjectionId,
+    handlerName :: !Text,
+    subscriptionId :: !(Maybe SubscriptionId),
+    subscriptionName :: !(Maybe Text),
+    source :: !ConsumerSource
+  }
+  deriving stock (Eq, Ord, Show, Generic)
+
+projectionConsumerInventoryIdentity :: ProjectionCatalogOperations -> Text
+projectionConsumerInventoryIdentity (ProjectionCatalogOperations catalog) =
+  catalogFingerprintText (Keiro.Projection.Catalog.catalogFingerprint catalog)
+
+projectionConsumers :: ProjectionCatalogOperations -> CategoryName -> [ProjectionConsumerGroup]
+projectionConsumers (ProjectionCatalogOperations catalog) category =
+  [ ProjectionConsumerGroup groupId (Set.toAscList bindings)
+  | (groupId, bindings) <- Map.toAscList grouped
+  ]
+  where
+    inventory = catalogInventory catalog
+    sources = Map.fromList [(source.sourceId, scope source.sourceScope) | source <- inventory.inventorySources]
+    subscriptions = Map.fromList [(subscription.subscriptionId, subscription) | subscription <- inventory.inventorySubscriptions]
+    scope AllStreams = AllCategories
+    scope (CategorySource (CategoryName value)) = ExactCategory value
+    grouped =
+      Map.fromListWith
+        Set.union
+        [ (projection.rebuildGroupId, Set.singleton binding)
+        | projection <- inventory.inventoryProjections,
+          handler <- projection.handlers,
+          Just binding <- [bindingFor projection handler],
+          consumerSourceMatches category binding.source
+        ]
+    bindingFor projection = \case
+      InventoryInlineHandler name -> do
+        source <- Map.lookup projection.sourceId sources
+        pure (ProjectionConsumerBinding projection.projectionId name Nothing Nothing source)
+      InventoryAsyncHandler name subscriptionId _ -> do
+        subscription <- Map.lookup subscriptionId subscriptions
+        source <- Map.lookup subscription.sourceId sources
+        pure (ProjectionConsumerBinding projection.projectionId name (Just subscriptionId) (Just subscription.subscriptionName) source)
 
 -- | Stable, JSON-friendly envelope for the complete heterogeneous inventory.
 data CatalogInventoryReport = CatalogInventoryReport
diff --git a/src/Keiro/ReadModel/Rebuild.hs b/src/Keiro/ReadModel/Rebuild.hs
--- a/src/Keiro/ReadModel/Rebuild.hs
+++ b/src/Keiro/ReadModel/Rebuild.hs
@@ -78,6 +78,7 @@
     ProjectionStatusPageRequest (..),
     ProjectionStatusReadError (..),
     ProjectionStatusPage (..),
+    ProjectionGroupStatusRow (..),
     projectionGroupsCursorKind,
     validateProjectionStatusRequest,
     listProjectionGroupStatusesPage,
diff --git a/src/Keiro/ReadModel/Rebuild/Status.hs b/src/Keiro/ReadModel/Rebuild/Status.hs
--- a/src/Keiro/ReadModel/Rebuild/Status.hs
+++ b/src/Keiro/ReadModel/Rebuild/Status.hs
@@ -10,6 +10,7 @@
     ProjectionStatusPageRequest (..),
     ProjectionStatusReadError (..),
     ProjectionStatusPage (..),
+    ProjectionGroupStatusRow (..),
     projectionGroupsCursorKind,
     validateProjectionStatusRequest,
     listProjectionGroupStatusesPage,
@@ -84,8 +85,16 @@
   | ProjectionStatusResultTooLarge
   deriving stock (Eq, Show, Generic)
 
+-- | A frozen v1 row plus the registered cursor authority. Names are stored
+-- sorted for checkpoint groups; append and unmanaged groups have an empty list.
+data ProjectionGroupStatusRow = ProjectionGroupStatusRow
+  { status :: !ProjectionGroupStatusV1,
+    feedingSubscriptions :: ![Text]
+  }
+  deriving stock (Eq, Show, Generic)
+
 data ProjectionStatusPage = ProjectionStatusPage
-  {observedAt :: !UTCTime, items :: ![ProjectionGroupStatusV1], next :: !(Maybe InspectionCursor)}
+  {observedAt :: !UTCTime, items :: ![ProjectionGroupStatusRow], next :: !(Maybe InspectionCursor)}
   deriving stock (Eq, Show, Generic)
 
 projectionGroupsCursorKind :: Text
@@ -108,6 +117,8 @@
 -- lookups through the frozen owner relation. The OFFSET 0 boundary preserves
 -- parameterized lookups instead of flattening the view into fleet-wide joins.
 -- A group's own model/checkpoint cardinality remains part of its read cost.
+-- The authority constraint proves non-checkpoint arrays are empty, so only
+-- checkpoint rows need the additional subscription-name probe.
 listProjectionGroupStatusesPage :: (Store :> es) => ProjectionStatusPageRequest -> Eff es (Either ProjectionStatusReadError ProjectionStatusPage)
 listProjectionGroupStatusesPage request = case validateProjectionStatusRequest request of
   Left err -> pure (Left err)
@@ -122,23 +133,31 @@
             maybe
               (Left ProjectionStatusResultTooLarge)
               (Right . Just)
-              (encodeCursorBounded projectionGroupsCursorKind ("projection_group_status_v1" :: Text, rebuildGroupIdText (row ^. #groupId)))
+              (encodeCursorBounded projectionGroupsCursorKind ("projection_group_status_v1" :: Text, rebuildGroupIdText (row ^. #status . #groupId)))
           _ -> Right Nothing
     pure (ProjectionStatusPage observed selected <$> continuation)
   where
     size = request ^. #pageSize
 
-pageProjectionGroupStatusesStmt :: Bool -> Statement (Maybe Text, Int64) [ProjectionGroupStatusV1]
+pageProjectionGroupStatusesStmt :: Bool -> Statement (Maybe Text, Int64) [ProjectionGroupStatusRow]
 pageProjectionGroupStatusesStmt after =
   preparable
     ( "WITH inspection_groups AS MATERIALIZED (SELECT group_id FROM keiro.keiro_projection_rebuild_groups"
         <> (if after then " WHERE group_id > $1" else " WHERE $1::text IS NULL")
-        <> " ORDER BY group_id LIMIT $2) SELECT status.* FROM inspection_groups page CROSS JOIN LATERAL ("
+        <> " ORDER BY group_id LIMIT $2) SELECT status.*, CASE WHEN status.serving_position_basis = 'checkpoint'"
+        <> " THEN coalesce((SELECT subscription_names FROM keiro.keiro_projection_group_cursors feeding_authority WHERE group_id = page.group_id OFFSET 0), ARRAY[]::text[])"
+        <> " ELSE ARRAY[]::text[] END AS feeding_subscriptions FROM inspection_groups page CROSS JOIN LATERAL ("
         <> statusSelect
         <> " WHERE group_id = page.group_id OFFSET 0) status ORDER BY status.group_id"
     )
     (contrazip2 (E.param (E.nullable E.text)) (E.param (E.nonNullable E.int8)))
-    (D.rowList projectionGroupStatusDecoder)
+    (D.rowList projectionGroupStatusRowDecoder)
+
+projectionGroupStatusRowDecoder :: D.Row ProjectionGroupStatusRow
+projectionGroupStatusRowDecoder =
+  ProjectionGroupStatusRow
+    <$> projectionGroupStatusDecoder
+    <*> D.column (D.nonNullable (D.listArray (D.nonNullable D.text)))
 
 -- | List the complete public status relation in stable group-id order.
 listProjectionGroupStatuses ::
diff --git a/src/Keiro/Workflow/Inspection.hs b/src/Keiro/Workflow/Inspection.hs
--- a/src/Keiro/Workflow/Inspection.hs
+++ b/src/Keiro/Workflow/Inspection.hs
@@ -14,6 +14,10 @@
     WorkflowListOrdering (..),
     WorkflowListQuery (..),
     defaultWorkflowListQuery,
+    WorkflowCountsQuery (..),
+    WorkflowStatusCounts (..),
+    WorkflowCounts (..),
+    countWorkflowInstancesByStatus,
     WorkflowListPage (..),
     listWorkflowInstancesPage,
     DetailPageQuery (..),
@@ -70,6 +74,18 @@
 
 maxInspectionPageSize :: Int
 maxInspectionPageSize = 500
+
+-- | Exact counts of current instance headers, including retained terminal rows.
+-- This scans the selected population in one database snapshot. Direct callers
+-- must configure their own statement deadline; the read changes no settings.
+countWorkflowInstancesByStatus :: (Store :> es) => WorkflowCountsQuery -> Eff es WorkflowCounts
+countWorkflowInstancesByStatus = Instance.inspectionInstanceCounts
+
+instance ToJSON WorkflowStatusCounts where
+  toJSON c = object ["running" .= c.running, "retrying" .= c.retrying, "sleeping" .= c.sleeping, "awaiting" .= c.awaiting, "completed" .= c.completed, "cancelled" .= c.cancelled, "failed" .= c.failed]
+
+instance ToJSON WorkflowCounts where
+  toJSON result = object ["observed_at" .= result.observedAt, "counts" .= result.counts]
 
 data InspectionError = InvalidPageSize !Int | InvalidCursor !CursorError | CursorGenerationMismatch !Int !Int | CursorResultTooLarge
   deriving stock (Eq, Show, Generic)
diff --git a/src/Keiro/Workflow/Inspection/Query.hs b/src/Keiro/Workflow/Inspection/Query.hs
--- a/src/Keiro/Workflow/Inspection/Query.hs
+++ b/src/Keiro/Workflow/Inspection/Query.hs
@@ -2,6 +2,7 @@
 -- schema consume them without depending on the rendering/page facade.
 module Keiro.Workflow.Inspection.Query where
 
+import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty)
 import Data.Text (Text)
 import Data.Time (UTCTime)
@@ -60,6 +61,25 @@
 
 defaultWorkflowListQuery :: WorkflowListQuery
 defaultWorkflowListQuery = WorkflowListQuery Nothing Nothing Nothing Nothing Nothing Nothing ByKey 100 Nothing
+
+data WorkflowCountsQuery = WorkflowCountsQuery
+  {workflowName :: !(Maybe Text)}
+  deriving stock (Eq, Show, Generic)
+
+data WorkflowStatusCounts = WorkflowStatusCounts
+  { running :: !Int64,
+    retrying :: !Int64,
+    sleeping :: !Int64,
+    awaiting :: !Int64,
+    completed :: !Int64,
+    cancelled :: !Int64,
+    failed :: !Int64
+  }
+  deriving stock (Eq, Show, Generic)
+
+data WorkflowCounts = WorkflowCounts
+  {observedAt :: !UTCTime, counts :: !WorkflowStatusCounts}
+  deriving stock (Eq, Show, Generic)
 
 -- | Validated exclusive database keys, decoded by the inspection facade.
 -- The default key tuple is (workflow_id, workflow_name).
diff --git a/src/Keiro/Workflow/Instance.hs b/src/Keiro/Workflow/Instance.hs
--- a/src/Keiro/Workflow/Instance.hs
+++ b/src/Keiro/Workflow/Instance.hs
@@ -26,6 +26,7 @@
     lookupInstanceConditionalSession,
     listWorkflowInstances,
     inspectionInstanceRows,
+    inspectionInstanceCounts,
     cancelWorkflow,
     claimInstance,
     renewInstanceLeaseTx,
@@ -896,14 +897,6 @@
     columns = "workflow_id, workflow_name, generation, status, attempts, last_error, next_attempt_at, wake_after, leased_by, lease_expires_at, created_at, updated_at, completed_at"
     -- Classification is inspection-only. Reuse existing status/key access;
     -- no generated column or inspection index adds work to ordinary writes.
-    classPredicate = \case
-      ClassRunning -> "status = 'running' AND attempts = 0"
-      ClassRetrying -> "status = 'running' AND attempts <> 0"
-      ClassSleeping -> "status = 'suspended' AND wake_after IS NOT NULL"
-      ClassAwaiting -> "status = 'suspended' AND wake_after IS NULL"
-      ClassCompleted -> "status = 'completed'"
-      ClassCancelled -> "status = 'cancelled'"
-      ClassFailed -> "status = 'failed'"
     sql =
       "/* keiro-inspection-page-v2 */ WITH inspection_page AS MATERIALIZED (SELECT "
         <> columns
@@ -917,3 +910,32 @@
         <> order
     params = filterParams <> (const (fromIntegral count :: Int32) >$< E.param (E.nonNullable E.int4))
     decoder = D.rowList ((,) <$> D.column (D.nonNullable D.timestamptz) <*> getCompose (instanceFields (Compose . D.column . D.nullable) (Compose . fmap Just . D.column . D.nullable)))
+
+-- Shared only by inspection reads; lifecycle writes do no counting work.
+classPredicate :: WorkflowStatusClass -> Text
+classPredicate = \case
+  ClassRunning -> "status = 'running' AND attempts = 0"
+  ClassRetrying -> "status = 'running' AND attempts <> 0"
+  ClassSleeping -> "status = 'suspended' AND wake_after IS NOT NULL"
+  ClassAwaiting -> "status = 'suspended' AND wake_after IS NULL"
+  ClassCompleted -> "status = 'completed'"
+  ClassCancelled -> "status = 'cancelled'"
+  ClassFailed -> "status = 'failed'"
+
+-- | Exact current-header counts in a single snapshot. The caller
+-- owns statement/deadline policy, as with other direct inspection reads.
+inspectionInstanceCounts :: (Store :> es) => WorkflowCountsQuery -> Eff es WorkflowCounts
+inspectionInstanceCounts query = runTransaction (Tx.statement () statement)
+  where
+    statement = preparable sql params (D.singleRow decoder)
+    sql =
+      "/* keiro-inspection-counts-v1 */ SELECT now(), "
+        <> Text.intercalate ", " ["count(*) FILTER (WHERE " <> classPredicate c <> ")" | c <- [minBound .. maxBound]]
+        <> " FROM keiro.keiro_workflows"
+        <> maybe "" (const " WHERE workflow_name = $1") query.workflowName
+    params = maybe E.noParams (\name -> const name >$< E.param (E.nonNullable E.text)) query.workflowName
+    total = D.column (D.nonNullable D.int8)
+    decoder =
+      WorkflowCounts
+        <$> D.column (D.nonNullable D.timestamptz)
+        <*> (WorkflowStatusCounts <$> total <*> total <*> total <*> total <*> total <*> total <*> total)
diff --git a/test/CatalogOperationsSpec.hs b/test/CatalogOperationsSpec.hs
--- a/test/CatalogOperationsSpec.hs
+++ b/test/CatalogOperationsSpec.hs
@@ -15,6 +15,7 @@
 import Data.Vector qualified as Vector
 import Effectful (Eff, IOE)
 import Effectful.Error.Static (Error)
+import Keiro.Inspection.Consumers (ConsumerSource (..))
 import Keiro.Prelude
 import Keiro.Projection.Catalog
 import Keiro.Projection.Catalog.Operations qualified as Operations
@@ -23,12 +24,58 @@
 import Kiroku.Store qualified as Store
 import Kiroku.Store.Effect (Store)
 import Kiroku.Store.Error (StoreError)
-import Kiroku.Store.Types (GlobalPosition (..))
+import Kiroku.Store.Types (CategoryName (..), GlobalPosition (..))
 import Test.Hspec
 import "hasql-transaction" Hasql.Transaction qualified as Tx
 
 spec :: Fixture -> Spec
 spec fixture = do
+  describe "category consumers projection metadata" $ do
+    it "returns all-stream and exact groups with only matching bindings in stable order" $ do
+      let allCatalog = Catalog.additiveCatalog & #sources . ix 0 . #sourceScope .~ AllStreams
+      validated <- expectValid allCatalog
+      let operations = Operations.projectionCatalogOperations validated
+          unmatched = Operations.projectionConsumers operations (CategoryName "unknown")
+          matched = Operations.projectionConsumers operations (CategoryName "catalog-additive")
+      map (^. #groupId) unmatched `shouldBe` [Catalog.mainGroupId]
+      map (^. #source) (concatMap (^. #bindings) unmatched) `shouldBe` [AllCategories, AllCategories]
+      map (^. #groupId) matched `shouldBe` [Catalog.additiveGroupId, Catalog.mainGroupId]
+      let first = head (unmatched ^.. traverse . #bindings . traverse)
+      first ^. #subscriptionName `shouldBe` Just "catalog-async-subscription"
+      let reordered =
+            allCatalog
+              & #sources
+              %~ reverse
+              & #subscriptions
+              %~ reverse
+              & #projectionSets
+              %~ reverse
+      again <- expectValid reordered
+      Operations.projectionConsumers (Operations.projectionCatalogOperations again) (CategoryName "catalog-additive") `shouldBe` matched
+    it "keeps only the matching binding when sources share a lifecycle group" $ do
+      let input = Catalog.twoSourceCatalog (CategorySource (CategoryName "order")) (CategorySource (CategoryName "invoice")) (identityOrError mkSourceId "invoice-source")
+      validated <- expectValid input
+      let lookup category = Operations.projectionConsumers (Operations.projectionCatalogOperations validated) (CategoryName category)
+      map (^. #groupId) (lookup "order") `shouldBe` [Catalog.mainGroupId]
+      map (^. #handlerName) (concatMap (^. #bindings) (lookup "order")) `shouldBe` ["catalog-inline"]
+      map (^. #handlerName) (concatMap (^. #bindings) (lookup "invoice")) `shouldBe` ["catalog-async"]
+    it "ignores unused sources and subscriptions and returns successful empty exact matches" $ do
+      let source = head (Catalog.validCatalog ^. #sources)
+          subscription = head (Catalog.validCatalog ^. #subscriptions)
+          unusedId = identityOrError mkSourceId "unused-source"
+          exact =
+            Catalog.validCatalog
+              & #sources
+              .~ [source & #sourceScope .~ CategorySource (CategoryName "order"), source & #sourceId .~ unusedId & #sourceScope .~ AllStreams]
+              & #subscriptions
+              .~ [subscription, subscription & #subscriptionId .~ identityOrError mkSubscriptionId "unused-subscription" & #subscriptionName .~ "unused" & #subscriptionSource .~ unusedId]
+      validated <- expectValid exact
+      let operations = Operations.projectionCatalogOperations validated
+      Operations.projectionConsumers operations (CategoryName "Order") `shouldBe` []
+      Operations.projectionConsumers operations (CategoryName "unknown") `shouldBe` []
+      let bindings = concatMap (^. #bindings) (Operations.projectionConsumers operations (CategoryName "order"))
+      map (^. #subscriptionName) bindings `shouldBe` [Just "catalog-async-subscription", Nothing]
+      map (^. #source) bindings `shouldBe` [ExactCategory "order", ExactCategory "order"]
   describe "projection catalog operations reports" $ do
     it "derives complete versioned inventory and destructive preview from one catalog" $ do
       validated <- expectValid (operationsCatalog passingVerification)
diff --git a/test/CatalogSpec.hs b/test/CatalogSpec.hs
--- a/test/CatalogSpec.hs
+++ b/test/CatalogSpec.hs
@@ -2,6 +2,7 @@
   ( spec,
     CatalogEvent (..),
     validCatalog,
+    twoSourceCatalog,
     bridgeCatalog,
     bridgeRevisionV1,
     bridgeRevisionV2,
diff --git a/test/ConsumerInspectionSpec.hs b/test/ConsumerInspectionSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/ConsumerInspectionSpec.hs
@@ -0,0 +1,39 @@
+module ConsumerInspectionSpec (spec) where
+
+import Data.Either (isLeft)
+import Keiro.Inspection.Consumers
+import Kiroku.Store.Types (CategoryName (..))
+import Test.Hspec
+
+spec :: Spec
+spec = describe "category consumers durable metadata" $ do
+  it "uses declared inputs rather than the manager's fulfillment journal" $ do
+    let manager = DurableConsumerDescription ProcessManagerConsumer "fulfillment" "order-input" (ExactCategory "order")
+    Right registry <- pure (durableConsumerRegistry [manager])
+    durableConsumersForCategory registry (CategoryName "order") `shouldBe` [manager]
+    durableConsumersForCategory registry (CategoryName "fulfillment") `shouldBe` []
+    durableConsumersForCategory registry (CategoryName "Order") `shouldBe` []
+  it "matches literal namespaces at colon boundaries and all categories" $ do
+    let namespace = DurableConsumerDescription DurableSubscriber "namespace" "ns" (CategoryNamespace "pm")
+        all = DurableConsumerDescription DurableSubscriber "all" "all" AllCategories
+    Right registry <- pure (durableConsumerRegistry [namespace, all])
+    mapM_ (\category -> durableConsumersForCategory registry (CategoryName category) `shouldBe` [all, namespace]) ["pm", "pm:orders", "pm:orders:child"]
+    mapM_ (\category -> durableConsumersForCategory registry (CategoryName category) `shouldBe` [all]) ["pmm", "PM", "orders"]
+  it "coalesces duplicates, permits shared subscriptions, and orders independent of declarations" $ do
+    let first = DurableConsumerDescription DurableSubscriber "a" "shared" (ExactCategory "order")
+        second = DurableConsumerDescription ProcessManagerConsumer "b" "shared" (ExactCategory "order")
+        third = DurableConsumerDescription DurableSubscriber "a" "second" AllCategories
+    Right registry <- pure (durableConsumerRegistry [second, first, third, first])
+    durableConsumerRegistry [third, second, first] `shouldBe` Right registry
+    durableConsumersForCategory registry (CategoryName "order") `shouldBe` [third, first, second]
+    durableConsumerRegistry [first, second {source = ExactCategory "invoice"}]
+      `shouldBe` Left (ConflictingSubscriptionScope "shared")
+    Right changed <- pure (durableConsumerRegistry [first {name = "changed"}])
+    durableConsumerInventoryIdentity changed `shouldNotBe` durableConsumerInventoryIdentity registry
+  it "rejects blank identities and invalid scopes and permits explicitly empty registries" $ do
+    let valid = DurableConsumerDescription DurableSubscriber "consumer" "subscription" AllCategories
+    mapM_
+      (\description -> durableConsumerRegistry [description] `shouldSatisfy` isLeft)
+      [valid {name = ""}, valid {name = " c"}, valid {subscriptionName = " "}, valid {source = ExactCategory ""}, valid {source = ExactCategory "order "}, valid {source = CategoryNamespace " pm"}]
+    Right empty <- pure (durableConsumerRegistry [])
+    durableConsumersForCategory empty (CategoryName "order") `shouldBe` []
diff --git a/test/GroupRebuildSpec.hs b/test/GroupRebuildSpec.hs
--- a/test/GroupRebuildSpec.hs
+++ b/test/GroupRebuildSpec.hs
@@ -83,6 +83,12 @@
     validated <- expectValid Catalog.catalogWithMissingSubscription
     _ <- expectStore store (registerProjectionCatalog validated) >>= shouldBeRight
 
+    Right page <- expectStore store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 Nothing))
+    map (^. #feedingSubscriptions) (page ^. #items)
+      `shouldBe` [["catalog-async-subscription", "catalog-missing-subscription"]]
+    map (^. #status . #servingPositionBasis) (page ^. #items)
+      `shouldBe` [ServingPositionCheckpoint]
+
     initial <- expectStore store (lookupProjectionGroupStatus Catalog.mainGroupId)
     initial ^? _Just . #servingPositionBasis
       `shouldBe` Just ServingPositionCheckpoint
@@ -191,7 +197,10 @@
     Right first <- expectStore store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 Nothing))
     Right second <- expectStore store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 (first ^. #next)))
     Right third <- expectStore store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 (second ^. #next)))
-    (first ^. #items) <> (second ^. #items) <> (third ^. #items) `shouldBe` statuses
+    let rows = (first ^. #items) <> (second ^. #items) <> (third ^. #items)
+    map (^. #status) rows `shouldBe` statuses
+    map (^. #feedingSubscriptions) (filter ((== Catalog.mainGroupId) . (^. #status . #groupId)) rows)
+      `shouldBe` [[]]
     third ^. #next `shouldBe` Nothing
     expectStore store listProjectionGroupStatuses `shouldReturn` statuses
     for_ [0, -1, 501] $ \size ->
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -6,6 +6,7 @@
 import CatalogEvolutionSpec qualified
 import CatalogOperationsSpec qualified
 import CatalogSpec qualified
+import ConsumerInspectionSpec qualified
 import Contravariant.Extras (contrazip2, contrazip3, contrazip4, contrazip5, contrazip6)
 import Control.Concurrent (forkIO, killThread, threadDelay)
 import Control.Concurrent.MVar (MVar, modifyMVar, newEmptyMVar, newMVar, putMVar, readMVar, takeMVar, tryPutMVar)
@@ -208,6 +209,7 @@
 import Keiro.ReadModel.Rebuild qualified as Rebuild
 import Keiro.ReplayAudit qualified as ReplayAudit
 import Keiro.Snapshot.Policy (shouldSnapshot, shouldSnapshotSpan)
+import Keiro.Snapshot.Schema qualified as SnapshotSchema
 import Keiro.Stream qualified as Stream
 import Keiro.Subscription.Shard
   ( LeaseState (..),
@@ -453,6 +455,7 @@
 main :: IO ()
 main = withMigratedSuite $ \fixture -> hspec $ do
   WorkflowInspectionSpec.spec fixture
+  ConsumerInspectionSpec.spec
   CatalogSpec.spec
   PreimageSpec.spec
   CatalogEvolutionSpec.spec fixture
@@ -2459,10 +2462,19 @@
       forM_ [2, 3, 4] $ \amount -> do
         Right (Right _) <- Store.runStoreIO store (runCommand inspectionRunCommandOptions snapshotCounterEventStream target (Add amount))
         pure ()
+      Right (Just infoBefore) <- Store.runStoreIO store (Store.getStream (StreamName "inspectioncounter-one"))
+      Right (Just storedBefore) <- Store.runStoreIO store (SnapshotSchema.lookupSnapshotRow (infoBefore ^. #id))
+      (storedBefore ^. #streamVersion) `shouldBe` StreamVersion 2
+      (storedBefore ^. #state) `shouldBe` (codec ^. #encode) (Counting, RCons (Proxy @"lastAmount") (3 :: Int) RNil)
       Right (Right view) <- Store.runStoreIO store (Aggregate.inspectAggregate inspector "one")
       Right (Right full) <- Store.runStoreIO store (hydrateFull inspectionRunCommandOptions (unvalidated snapshotCounterEventStream) target)
       (view ^. #state) `shouldBe` Just ((codec ^. #encode) (full ^. #state, full ^. #registers))
       (view ^. #streamVersion) `shouldBe` (full ^. #streamVersion)
+      (view ^. #streamVersion) `shouldBe` StreamVersion 3
+      (view ^. #state) `shouldBe` Just ((codec ^. #encode) (Counting, RCons (Proxy @"lastAmount") (4 :: Int) RNil))
+      case Aggregate.aggregateViewJson view of
+        Aeson.Object fields -> KeyMap.lookup "state_version" fields `shouldBe` Just (Aeson.toJSON (3 :: Int))
+        _ -> expectationFailure "expected aggregate JSON object"
       let mappedDefinition :: SnapshotCounterEventStream
           mappedDefinition = (unvalidated snapshotCounterEventStream) & #resolveStreamName .~ (\typed -> case Stream.streamName typed of StreamName raw -> StreamName ("inspectioncounter-" <> raw))
           mappedStream = mkEventStreamOrThrow "inspection-mapped-counter" mappedDefinition
@@ -2474,14 +2486,35 @@
       Aggregate.aggregateIdOf mappedInspector (StreamName "inspectioncounter-one") `shouldBe` Just "one"
       (view ^. #hydration) `shouldBe` Just (FromSnapshot (StreamVersion 2))
       fmap (^. #compatibility) (view ^. #snapshot . #stored) `shouldBe` Just Aggregate.Compatible
+      Store.runStoreIO store (SnapshotSchema.lookupSnapshotRow (infoBefore ^. #id)) `shouldReturn` Right (Just storedBefore)
+      Store.runStoreIO store (Store.getStream (StreamName "inspectioncounter-one")) `shouldReturn` Right (Just infoBefore)
       Right _ <- Store.runStoreIO store (Store.softDeleteStream (StreamName "inspectioncounter-one"))
       Right (Right deleted) <- Store.runStoreIO store (Aggregate.inspectAggregate inspector "one")
       (deleted ^. #state) `shouldBe` Nothing
       (deleted ^. #hydration) `shouldBe` Nothing
       (deleted ^. #streamVersion) `shouldBe` StreamVersion 3
+      Store.runStoreIO store (SnapshotSchema.lookupSnapshotRow (infoBefore ^. #id)) `shouldReturn` Right (Just storedBefore)
       Aggregate.aggregateStream inspector " " `shouldBe` Left (Aggregate.AggregateIdRejected " ")
       missing <- Store.runStoreIO store (Aggregate.inspectAggregate inspector "missing")
       missing `shouldBe` Right (Left (Aggregate.AggregateNotFound (StreamName "inspectioncounter-missing")))
+
+    it "reports incompatible snapshot identifiers independently while replaying current values" $ \store -> do
+      let codec = defaultStateCodec @SnapshotCounterRegs @CounterState 1
+          inspector = Aggregate.aggregateInspector "counter" snapshotCounterEventStream (Stream.categoryUnsafe "inspectioncounter") (Aggregate.displayViaStateCodec codec)
+          target = Stream.entityStream (Stream.categoryUnsafe "inspectioncounter") "incompatible"
+      forM_ [2, 3, 4] $ \amount -> do
+        Right (Right _) <- Store.runStoreIO store (runCommand inspectionRunCommandOptions snapshotCounterEventStream target (Add amount))
+        pure ()
+      Right (Just info) <- Store.runStoreIO store (Store.getStream (StreamName "inspectioncounter-incompatible"))
+      let payload = object ["legacy" Aeson..= True]
+      Right () <- Store.runStoreIO store (SnapshotSchema.writeSnapshotRow (SnapshotSchema.SnapshotWrite (info ^. #id) (StreamVersion 2) payload 999 "old-registers" "old-fold"))
+      Right before <- Store.runStoreIO store (SnapshotSchema.lookupSnapshotRow (info ^. #id))
+      Right (Right view) <- Store.runStoreIO store (Aggregate.inspectAggregate inspector "incompatible")
+      (view ^. #streamVersion) `shouldBe` StreamVersion 3
+      (view ^. #state) `shouldBe` Just ((codec ^. #encode) (Counting, RCons (Proxy @"lastAmount") (4 :: Int) RNil))
+      fmap (^. #compatibility) (view ^. #snapshot . #stored) `shouldBe` Just (Aggregate.Incompatible ["state_codec_version", "regfile_shape_hash", "state_shape_hash"])
+      Store.runStoreIO store (SnapshotSchema.lookupSnapshotRow (info ^. #id)) `shouldReturn` Right before
+      Store.runStoreIO store (Store.getStream (StreamName "inspectioncounter-incompatible")) `shouldReturn` Right (Just info)
 
     it "advances one raw page through rejected rows without losing the look-ahead row" $ \store -> do
       let base = Aggregate.aggregateInspector "counter" counterEventStream (Stream.categoryUnsafe "inspectionpage") (\state _ -> Aeson.String (Text.pack (show state)))
diff --git a/test/WorkflowInspectionSpec.hs b/test/WorkflowInspectionSpec.hs
--- a/test/WorkflowInspectionSpec.hs
+++ b/test/WorkflowInspectionSpec.hs
@@ -4,6 +4,7 @@
 
 import Control.Monad (forM, forM_)
 import Data.Aeson (Value (..))
+import Data.Aeson qualified as Aeson
 import Data.List (nub, sort)
 import Data.List.NonEmpty (NonEmpty (..))
 import Data.Text (Text)
@@ -45,6 +46,30 @@
       forM_ ["not-a-cursor", "", Text.replicate 4097 "a", "===="] $ \bad ->
         (decodeCursor "wf-key2" (InspectionCursor bad) :: Either CursorError Value) `shouldSatisfy` (\case Left (MalformedCursor _) -> True; _ -> False)
   around (withFreshStore fixture) $ do
+    it "encodes every count as a JSON integer beyond 32 bits" $ \_ -> do
+      let counts = WorkflowStatusCounts 4294967296 0 0 0 0 0 0
+      Aeson.toJSON counts `shouldBe` Aeson.object ["running" Aeson..= (4294967296 :: Integer), "retrying" Aeson..= (0 :: Int), "sleeping" Aeson..= (0 :: Int), "awaiting" Aeson..= (0 :: Int), "completed" Aeson..= (0 :: Int), "cancelled" Aeson..= (0 :: Int), "failed" Aeson..= (0 :: Int)]
+    it "counts empty selections and exact names with one database observation" $ \store -> do
+      empty <- expectStore store (countWorkflowInstancesByStatus (WorkflowCountsQuery Nothing))
+      empty.counts `shouldBe` WorkflowStatusCounts 0 0 0 0 0 0 0
+      empty.observedAt `shouldSatisfy` (> read "2026-01-01 00:00:00 UTC")
+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('one','支払_%''\\ λ','running'),('two','支払_other','failed')"
+      exact <- expectStore store (countWorkflowInstancesByStatus (WorkflowCountsQuery (Just "支払_%'\\ λ")))
+      exact.counts `shouldBe` WorkflowStatusCounts 1 0 0 0 0 0 0
+      absent <- expectStore store (countWorkflowInstancesByStatus (WorkflowCountsQuery (Just "absent")))
+      absent.counts `shouldBe` empty.counts
+    it "counts every class beyond one list page and ignores lease and retry clocks" $ \store -> do
+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status,attempts,wake_after,next_attempt_at,leased_by,lease_expires_at,generation) SELECT 'counts-'||lpad(n::text,4,'0'),'counts',CASE n%7 WHEN 0 THEN 'running' WHEN 1 THEN 'running' WHEN 2 THEN 'suspended' WHEN 3 THEN 'suspended' WHEN 4 THEN 'completed' WHEN 5 THEN 'cancelled' ELSE 'failed' END,CASE WHEN n%7=1 THEN -1 ELSE 0 END,CASE WHEN n%7=2 THEN '2020-01-01+00'::timestamptz END,'2099-01-01+00'::timestamptz,'worker','2099-01-01+00'::timestamptz,5 FROM generate_series(1,700) n"
+      result <- expectStore store (countWorkflowInstancesByStatus (WorkflowCountsQuery Nothing))
+      result.counts `shouldBe` WorkflowStatusCounts 100 100 100 100 100 100 100
+      pages <- drain store defaultWorkflowListQuery {pageSize = 500}
+      let rows = concatMap (.items) pages
+      length rows `shouldBe` 700
+      forM_ [minBound .. maxBound] $ \cls ->
+        length (filter ((== cls) . statusClassOf) rows) `shouldBe` 100
+      sql store "UPDATE keiro.keiro_workflows SET attempts=0 WHERE status='running'; UPDATE keiro.keiro_workflows SET wake_after=NULL WHERE status='suspended'"
+      transitioned <- expectStore store (countWorkflowInstancesByStatus (WorkflowCountsQuery Nothing))
+      transitioned.counts `shouldBe` WorkflowStatusCounts 200 0 0 200 100 100 100
     it "rechecks every header field against its exact image, including old and far-future dates" $ \store -> do
       let fields =
             [ ("workflow_id", "'two'"),
@@ -213,7 +238,10 @@
       length steps.items `shouldBe` 2
       other <- expectStore store (loadStepIndexPage (WorkflowName "test") (WorkflowId "different") (DetailPageQuery Nothing 2 steps.nextCursor))
       other `shouldSatisfy` (\case Left (InvalidCursor _) -> True; _ -> False)
+      beforeRotation <- expectStore store (countWorkflowInstancesByStatus (WorkflowCountsQuery Nothing))
       _ <- expectStore store (runWorkflow (WorkflowName "test") (WorkflowId "owner") (continueAsNew () >> pure ()))
+      afterRotation <- expectStore store (countWorkflowInstancesByStatus (WorkflowCountsQuery Nothing))
+      afterRotation.counts `shouldBe` beforeRotation.counts
       rotated <- expectStore store (loadStepIndexPage (WorkflowName "test") (WorkflowId "owner") (DetailPageQuery Nothing 2 steps.nextCursor))
       rotated `shouldBe` Left (CursorGenerationMismatch 1 0)
     it "pages children and directly allocated awakeables through their owning tables" $ \store -> do