keiro 0.23.0.0 → 0.24.0.0

raw patch · 15 files changed

+430/−23 lines, 15 filesdep ~keiro-coredep ~keiro-migrationsdep ~keiro-test-supportPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: keiro-core, keiro-migrations, keiro-test-support

API changes (from Hackage documentation)

Files

• CHANGELOG.md +14/−0
• keiro.cabal +10/−8
• src/Keiro/Inspection/Consumers.hs +99/−0
• src/Keiro/Projection/Catalog/Operations.hs +57/−0
• src/Keiro/ReadModel/Rebuild.hs +1/−0
• src/Keiro/ReadModel/Rebuild/Status.hs +24/−5
• src/Keiro/Workflow/Inspection.hs +16/−0
• src/Keiro/Workflow/Inspection/Query.hs +20/−0
• src/Keiro/Workflow/Instance.hs +30/−8
• test/CatalogOperationsSpec.hs +48/−1
• test/CatalogSpec.hs +1/−0
• test/ConsumerInspectionSpec.hs +39/−0
• test/GroupRebuildSpec.hs +10/−1
• test/Main.hs +33/−0
• test/WorkflowInspectionSpec.hs +28/−0

@@ -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

@@ -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,

@@ -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

@@ -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

@@ -78,6 +78,7 @@     ProjectionStatusPageRequest (..),     ProjectionStatusReadError (..),     ProjectionStatusPage (..),+    ProjectionGroupStatusRow (..),     projectionGroupsCursorKind,     validateProjectionStatusRequest,     listProjectionGroupStatusesPage,

@@ -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 ::

@@ -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)

@@ -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).

@@ -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)

@@ -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)

@@ -2,6 +2,7 @@   ( spec,     CatalogEvent (..),     validCatalog,+    twoSourceCatalog,     bridgeCatalog,     bridgeRevisionV1,     bridgeRevisionV2,

@@ -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` []

@@ -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 ->

@@ -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)))

@@ -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