keiro-ops 0.23.0.0 → 0.24.0.0

raw patch · 8 files changed

+449/−39 lines, 8 filesdep ~keirodep ~keiro-migrationsdep ~keiro-pgmqPVP ok

version bump matches the API change (PVP)

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

API changes (from Hackage documentation)

Files

• CHANGELOG.md +16/−0
• keiro-ops.cabal +11/−8
• src/Keiro/Ops.hs +9/−2
• src/Keiro/Ops/Embed.hs +3/−0
• src/Keiro/Ops/Projection.hs +237/−28
• src/Keiro/Ops/Workflow.hs +4/−0
• test/CategoryConsumersSpec.hs +115/−0
• test/Main.hs +54/−1

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

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

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

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

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

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

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

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