keiro-ops 0.21.0.0 → 0.22.0.0

raw patch · 17 files changed

+1164/−239 lines, 17 filesdep +hasql-pooldep +keikidep ~keirodep ~keiro-migrationsdep ~keiro-pgmqPVP ok

version bump matches the API change (PVP)

Dependencies added: hasql-pool, keiki

Dependency ranges changed: keiro, keiro-migrations, keiro-pgmq, keiro-test-support, kiroku-store, pgmq-migration

API changes (from Hackage documentation)

Files

• CHANGELOG.md +51/−0
• keiro-ops.cabal +16/−10
• src/Keiro/Ops.hs +70/−2
• src/Keiro/Ops/Aggregate.hs +121/−0
• src/Keiro/Ops/Embed.hs +10/−2
• src/Keiro/Ops/Parse.hs +8/−1
• src/Keiro/Ops/Pgmq.hs +5/−3
• src/Keiro/Ops/ProcessManager.hs +136/−0
• src/Keiro/Ops/Projection.hs +78/−3
• src/Keiro/Ops/Render.hs +1/−0
• src/Keiro/Ops/Shard.hs +59/−34
• src/Keiro/Ops/Snapshot.hs +23/−1
• src/Keiro/Ops/Timer.hs +74/−7
• src/Keiro/Ops/Workflow.hs +143/−175
• src/Keiro/Ops/WorkflowHeaderCache.hs +103/−0
• test/Main.hs +96/−1
• test/ReactionInspectionSpec.hs +170/−0

@@ -6,6 +6,57 @@  ## [Unreleased] +## 0.22.0.0 — 2026-10-27++### Breaking Changes++- Require the Kiroku Store 0.12 cohort from+  mori://shinzui/kiroku/packages/kiroku-store; apply migration 0016 from+  mori://shinzui/kiroku/packages/kiroku-store-migrations with affected workers+  stopped, then restart new binaries before using namespace targets. Update+  exhaustive subscription target/event matches and low-level record constructors.++- Direct `AppHooks` records must supply `aggregates` and `processManagers`; `emptyAppHooks` leaves them disabled. `OpsOutcome` adds `InspectionFailed`, requiring exhaustive consumers to handle structured codes.+- `wf show/steps/journal` return `InspectionFailed` for typed inspection errors. `pgmq dlq read` requires `--force` to reserve rows; an unforced invocation previews and exits 1.++### New Features++- Add embedded `pm types` with shared JSON and human metadata output, including+  empty catalogs and aliases, without requiring a manager name or store reads.+  Hosts may supply a bounded direct encoder through `runCommandWithReadEncoding`.++- Add bounded projection-status pages and `projection status`, preserving the+  frozen v1 relation and complete-list compatibility API. Page keys precede+  correlated status reads; no writer index is added. HTTP feed protocol and+  canonical view definitions are available as foundations for live feeds.++- Add process-manager persisted-state inspection and bounded raw stream listing,+  pending-timer pages with filter-bound cursors, and database-clock shard+  ownership/listing. Reads add no worker queries, polling tasks or migrations.+  Embedded `pm show/list` uses the additive `AppHooks.processManagers` field;+  `emptyAppHooks` leaves this optional catalog off.+++- An optional `AppHooks.aggregates` catalog mounts read-only `aggregate types|list|show|snapshot` commands. Snapshot show adds `observed_at` and `age_seconds`.++- Bounded header presentation reuse through an alternative runner that checks the exact current row before returning prepared JSON bytes.++- Bounded `wf inspect` commands and typed inspection errors; shared workflow rendering and paged compatibility reads.+++- Export the authoritative command grammar, classifier, runner, and invocation fields for alternative front ends.+++- `pm show/list` accepts reaction-state descriptors with explicitly unavailable provenance. Capture and `legacy-positional-v1` remain legacy positional features scheduled for retirement before 1.0.+- Workflow list/show JSON includes `status_class`; step rows include `recorded_at`. Show uses owner-table creation-key ordering for children and awakeables. Shard status includes database `observed_at`, `heartbeat_at`, `updated_at` and `lease_age_seconds`.++### Bug Fixes++- Treat `pgmq dlq read` as a mutation because it reserves messages for 30 seconds. It now previews by default and requires `--force` to read; the HTTP surface uses POST with explicit confirmation.+++- Default workflow inspection consumes primary-key `wf-key2` pages; old default tokens fail closed and the obsolete exceptional-key refusal is removed. Filtered and creation-order reads retain host deadlines.+ ## 0.21.0.0 — 2026-10-14  ### Bug Fixes

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro-ops-version: 0.21.0.0+version: 0.22.0.0 synopsis: Operational command-line interface for Keiro deployments description:   A standalone database-only console and embeddable command tree for inspecting@@ -49,12 +49,14 @@   hs-source-dirs: src   exposed-modules:     Keiro.Ops+    Keiro.Ops.Aggregate     Keiro.Ops.Embed     Keiro.Ops.Env     Keiro.Ops.Inbox     Keiro.Ops.Outbox     Keiro.Ops.Parse     Keiro.Ops.Pgmq+    Keiro.Ops.ProcessManager     Keiro.Ops.Projection     Keiro.Ops.Rebuild     Keiro.Ops.Render@@ -64,6 +66,7 @@     Keiro.Ops.Stream     Keiro.Ops.Timer     Keiro.Ops.Workflow+    Keiro.Ops.WorkflowHeaderCache    build-depends:     aeson >=2.2.2 && <2.3,@@ -73,10 +76,11 @@     effectful >=2.6 && <2.8,     effectful-core >=2.6 && <2.7 || >=2.7.1.1 && <2.8,     hasql >=1.10 && <1.11,-    keiro ^>=0.21.0.0,-    keiro-migrations ^>=0.21.0.0,-    keiro-pgmq ^>=0.21.0.0,-    kiroku-store ^>=0.11.0.0,+    hasql-pool >=1.4 && <1.5,+    keiro ^>=0.22.0.0,+    keiro-migrations ^>=0.22.0.0,+    keiro-pgmq ^>=0.22.0.0,+    kiroku-store ^>=0.12.0.0,     optparse-applicative >=0.18 && <0.20,     text >=2.1 && <2.2,     time >=1.12 && <1.15,@@ -106,6 +110,7 @@     -rtsopts     -with-rtsopts=-N +  other-modules: ReactionInspectionSpec   build-tool-depends: keiro-ops:keiro-ops   build-depends:     aeson >=2.2 && <2.3,@@ -117,13 +122,14 @@     hasql >=1.10 && <1.11,     hasql-transaction >=1.1 && <1.3,     hspec >=2.11,-    keiro ^>=0.21.0.0,+    keiki >=0.9.1 && <0.10,+    keiro ^>=0.22.0.0,     keiro-ops,-    keiro-pgmq ^>=0.21.0.0,-    keiro-test-support ^>=0.21.0.0,-    kiroku-store ^>=0.11.0.0,+    keiro-pgmq ^>=0.22.0.0,+    keiro-test-support ^>=0.22.0.0,+    kiroku-store ^>=0.12.0.0,     optparse-applicative >=0.18 && <0.20,-    pgmq-migration >=0.6.1.1 && <0.7,+    pgmq-migration >=0.7.0.0 && <0.8,     process >=1.6 && <1.7,     text >=2.1 && <2.2,     time >=1.12 && <1.15,

@@ -2,26 +2,39 @@   ( main,     mainWithHooks,     AppHooks (..),+    AggregateInspectors,+    ProcessManagerInspectors,     OpsAuditConfig (..),     emptyAppHooks,-    OpsInvocation,+    OpsInvocation (..),+    Command (..),     opsCommandTree,+    commandParser,+    validateCommand,+    isMutation,+    runCommand,+    runCommandWithHeaderCache,+    runCommandWithReadEncoding,     runOpsInvocation,   ) where  import Control.Exception (SomeException, displayException, fromException, try)+import Data.Aeson.Encoding (Encoding)+import Data.ByteString (ByteString) import Data.Foldable (traverse_) import Data.Maybe (isJust) import Data.Text qualified as Text import Data.Text.IO qualified as Text.IO import Hasql.Connection.Settings qualified as Settings import Keiro.Migrations.SchemaCheck (renderSchemaDrift, verifyExpectedSchema)+import Keiro.Ops.Aggregate qualified as Aggregate import Keiro.Ops.Embed import Keiro.Ops.Env import Keiro.Ops.Inbox qualified as Inbox import Keiro.Ops.Outbox qualified as Outbox import Keiro.Ops.Pgmq qualified as Pgmq+import Keiro.Ops.ProcessManager qualified as ProcessManager import Keiro.Ops.Projection qualified as Projection import Keiro.Ops.Rebuild qualified as Rebuild import Keiro.Ops.Render@@ -31,6 +44,7 @@ import Keiro.Ops.Stream qualified as Stream import Keiro.Ops.Timer qualified as Timer import Keiro.Ops.Workflow qualified as Workflow+import Keiro.Ops.WorkflowHeaderCache (WorkflowHeaderCache, runHeaderCached) import Kiroku.Store.Connection (defaultConnectionSettings, withStore) import Options.Applicative import System.Exit qualified as Exit@@ -60,8 +74,12 @@     opsCommand :: !Command   } +-- | Programmatic front-end boundary. Alternative front ends must verify the+-- live schema, use JSON output mode, and respect 'isMutation'. data Command-  = Workflow Workflow.Command+  = ProcessManager ProcessManager.Command+  | Aggregate Aggregate.Command+  | Workflow Workflow.Command   | Timer Timer.Command   | Outbox Outbox.Command   | Inbox Inbox.Command@@ -85,6 +103,7 @@ invocationParser :: AppHooks -> Parser OpsInvocation invocationParser hooks = OpsInvocation <$> globalOptionsParser <*> commandParser hooks +-- | The authoritative command grammar, including only mounted hooks. commandParser :: AppHooks -> Parser Command commandParser hooks =   hsubparser@@ -123,8 +142,17 @@           (info (Stream <$> Stream.commandParser) (progDesc "Inspect and operate Kiroku streams"))         <> replayAuditCommand         <> rebuildCommand+        <> aggregateCommand+        <> processManagerCommand     )   where+    processManagerCommand = case hooks.processManagers of+      Nothing -> mempty+      Just _ -> command "pm" (info (ProcessManager <$> ProcessManager.commandParser) (progDesc "Inspect mounted process managers"))+    aggregateCommand =+      case hooks.aggregates of+        Nothing -> mempty+        Just _ -> command "aggregate" (info (Aggregate <$> Aggregate.commandParser) (progDesc "Inspect mounted aggregates"))     replayAuditCommand =       case hooks.replayAudit of         Nothing -> mempty@@ -165,8 +193,20 @@                   }           runCommand hooks env opsCommand >>= finishOutcome env +-- | Validate supported command arguments before admission or schema work.+validateCommand :: Command -> Either (Text.Text, Text.Text) ()+validateCommand (ProcessManager command) = ProcessManager.validateCommand command+validateCommand (Aggregate command) = Aggregate.validateCommand command+validateCommand (Workflow command) = Workflow.validateCommand command+validateCommand (Timer command) = Timer.validateCommand command+validateCommand (Projection command) = Projection.validateCommand command+validateCommand _ = Right ()++-- | Authoritative read/write classification for every front end. isMutation :: Command -> Bool isMutation = \case+  ProcessManager command -> ProcessManager.isMutation command+  Aggregate command -> Aggregate.isMutation command   Workflow workflowCommand -> Workflow.isMutation workflowCommand   Timer timerCommand -> Timer.isMutation timerCommand   Outbox outboxCommand -> Outbox.isMutation outboxCommand@@ -179,8 +219,35 @@   ReplayAudit _ -> False   Rebuild rebuildCommand -> Rebuild.isMutation rebuildCommand +-- | Reuse only pure header presentation after an exact fresh database check.+-- The optional bytes encode this outcome's successful read result. Other+-- commands use the ordinary authoritative runner without a cache lookup.+runCommandWithHeaderCache :: WorkflowHeaderCache -> AppHooks -> OpsEnv -> Command -> IO (OpsOutcome, Maybe ByteString)+runCommandWithHeaderCache cache _ env (Workflow (Workflow.Inspect (Workflow.InspectHeader ref))) = runHeaderCached cache env ref+runCommandWithHeaderCache _ hooks env command = do+  outcome <- runCommand hooks env command+  pure (outcome, Nothing)++-- | Prepare supported direct encodings on demand using the host's bounded+-- encoder. Schema verification and admission remain the front end's job.+-- The callback must return the supplied encoding's exact bytes or refuse size.+-- Existing command/header behavior is delegated without changing its cache.+runCommandWithReadEncoding :: (Encoding -> IO (Maybe ByteString)) -> WorkflowHeaderCache -> AppHooks -> OpsEnv -> Command -> IO (OpsOutcome, Maybe ByteString)+runCommandWithReadEncoding encode cache hooks env command = case command of+  ProcessManager ProcessManager.Types | Just catalog <- hooks.processManagers -> do+    let (result, encoding) = ProcessManager.prepareTypes catalog+    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+-- JSON mode and enable force only after explicit mutation confirmation. runCommand :: AppHooks -> OpsEnv -> Command -> IO OpsOutcome runCommand hooks env = \case+  ProcessManager command -> maybe (pure (Failed "process-manager inspection hook is not mounted")) (\catalog -> ProcessManager.runCommand catalog env command) hooks.processManagers+  Aggregate command -> maybe (pure (Failed "aggregate inspection hook is not mounted")) (\catalog -> Aggregate.runCommand env catalog command) hooks.aggregates   Workflow workflowCommand -> Workflow.runCommandWithResume hooks.workflowResume env workflowCommand   Timer timerCommand -> Timer.runCommandWithFire hooks.timerFire env timerCommand   Outbox outboxCommand -> Outbox.runCommand env outboxCommand@@ -210,6 +277,7 @@     Text.IO.hPutStrLn stderr ("preview only; re-run with --force: " <> reinvocation)     pure (Exit.ExitFailure 1)   Failed message -> operationalFailure message+  InspectionFailed code message -> operationalFailure (code <> ": " <> message)  operationalFailure :: Text.Text -> IO Exit.ExitCode operationalFailure message = do

@@ -0,0 +1,121 @@+-- | Read-only adapters over the application's aggregate catalog.+module Keiro.Ops.Aggregate (Command (..), ShowOptions (..), ListOptions (..), commandParser, validateCommand, isMutation, runCommand) where++import Data.Aeson (object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KeyMap+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as Text+import Effectful (Eff, IOE)+import Effectful.Error.Static (Error)+import Keiro.Inspection.Aggregate qualified as Aggregate+import Keiro.Inspection.Cursor (InspectionCursor, decodeCursor)+import Keiro.Ops.Env (OpsEnv (..))+import Keiro.Ops.Parse (cursorReader, positiveIntReader)+import Keiro.Ops.Render+import Kiroku.Store.Effect (Store, runStoreIO)+import Kiroku.Store.Error (StoreError)+import Kiroku.Store.Types (StreamName (..), StreamVersion (..))+import Options.Applicative hiding (action, command, info, value)+import Options.Applicative qualified as Opt++data ShowOptions = ShowOptions {name :: !Text, aggregateId :: !Text}+  deriving stock (Eq, Show)++data ListOptions = ListOptions {name :: !Text, limit :: !Int, after :: !(Maybe InspectionCursor)}+  deriving stock (Eq, Show)++data Command = Types | Show !ShowOptions | Snapshot !ShowOptions | List !ListOptions+  deriving stock (Eq, Show)++commandParser :: Parser Command+commandParser =+  hsubparser+    ( Opt.command "types" (Opt.info (pure Types) (progDesc "List mounted aggregate kinds"))+        <> Opt.command "show" (Opt.info (Show <$> showOptions) (progDesc "Inspect folded aggregate state"))+        <> Opt.command "snapshot" (Opt.info (Snapshot <$> showOptions) (progDesc "Inspect aggregate snapshot health"))+        <> Opt.command "list" (Opt.info (List <$> listOptions) (progDesc "Read one aggregate page"))+    )+  where+    textOption key label = Text.pack <$> strOption (long key <> metavar label)+    nameOption = textOption "name" "NAME"+    showOptions = ShowOptions <$> nameOption <*> textOption "id" "ID"+    listOptions = ListOptions <$> nameOption <*> option positiveIntReader (long "limit" <> Opt.value 50 <> metavar "N") <*> optional (option cursorReader (long "after" <> metavar "CURSOR"))++isMutation :: Command -> Bool+isMutation = const False++-- | Reject invalid input before a transport borrows a schema-check connection.+-- The owning page read additionally checks the catalog's exact category.+validateCommand :: Command -> Either (Text, Text) ()+validateCommand = \case+  Types -> Right ()+  Show options -> validateId options+  Snapshot options -> validateId options+  List options+    | options.limit < 1 || options.limit > Aggregate.maxAggregatePageSize -> Left ("invalid_page_size", "page size must be 1..500")+    | otherwise -> case options.after of+        Nothing -> Right ()+        Just cursor -> case decodeCursor Aggregate.aggregateListCursorKind cursor of+          Right (name, category, after :: Text)+            | name == options.name && Text.isPrefixOf (category <> "-") after -> Right ()+          _ -> Left ("invalid_cursor", "invalid cursor or cursor belongs to another aggregate inspector")+  where+    validateId options+      | Text.null (Text.strip options.aggregateId) = Left ("invalid_aggregate_id", "aggregate id must not be blank")+      | otherwise = Right ()++runCommand :: OpsEnv -> Aggregate.AggregateInspectors -> Command -> IO OpsOutcome+runCommand env catalog command = case command of+  Types ->+    pure+      ( Succeeded+          ( OpsResult+              ["name", "category", "snapshots_configured", "snapshot_policy"]+              [[summary.name, summary.category, if summary.snapshotsConfigured then "true" else "false", summary.snapshotPolicy] | summary <- Aggregate.listAggregateTypes catalog]+              (object ["items" .= map Aggregate.aggregateTypeJson (Aggregate.listAggregateTypes catalog)])+          )+      )+  Show options -> case Map.lookup options.name catalog of+    Nothing -> pure (Failed ("unknown aggregate kind: " <> options.name))+    Just (Aggregate.SomeAggregateInspector inspector) ->+      runAction env (Aggregate.inspectAggregate inspector options.aggregateId) $ \view ->+        OpsResult+          ["name", "aggregate_id", "stream_version", "state"]+          [[view.name, view.aggregateId, versionText view.streamVersion, maybe "" id view.stateLabel]]+          (Aggregate.aggregateViewJson view)+  Snapshot options -> case Map.lookup options.name catalog of+    Nothing -> pure (Failed ("unknown aggregate kind: " <> options.name))+    Just (Aggregate.SomeAggregateInspector inspector) ->+      runAction env (Aggregate.snapshotStatusFor inspector options.aggregateId) $ \(StreamName streamName, StreamVersion version, status) ->+        let value = case Aggregate.snapshotStatusJson status of+              Aeson.Object fields -> Aeson.Object (fields <> KeyMap.fromList [("name", Aeson.toJSON options.name), ("aggregate_id", Aeson.toJSON options.aggregateId), ("stream_name", Aeson.toJSON streamName), ("stream_version", Aeson.toJSON version)])+              other -> other+         in OpsResult ["name", "aggregate_id"] [[options.name, options.aggregateId]] value+  List options -> case Map.lookup options.name catalog of+    Nothing -> pure (Failed ("unknown aggregate kind: " <> options.name))+    Just (Aggregate.SomeAggregateInspector inspector) ->+      runAction env (Aggregate.listAggregates inspector (Aggregate.AggregateListRequest options.limit options.after)) $ \page ->+        OpsResult+          ["aggregate_id", "stream_name", "stream_version"]+          [[item.aggregateId, nameText item.streamName, versionText item.streamVersion] | item <- page.items]+          (Aggregate.aggregateListPageJson page)++runAction :: OpsEnv -> Eff '[Store, Error StoreError, IOE] (Either Aggregate.AggregateInspectionError a) -> (a -> OpsResult) -> IO OpsOutcome+runAction env action render = do+  result <- runStoreIO env.store action+  pure $ case result of+    Left err -> Failed (Text.pack (show err))+    Right (Left err) -> case err of+      Aggregate.InvalidPageSize _ -> InspectionFailed "invalid_page_size" (Aggregate.inspectionErrorText err)+      Aggregate.InvalidCursor _ -> InspectionFailed "invalid_cursor" (Aggregate.inspectionErrorText err)+      Aggregate.ResultTooLarge -> InspectionFailed "inspection_result_too_large" (Aggregate.inspectionErrorText err)+      _ -> Failed (Aggregate.inspectionErrorText err)+    Right (Right value) -> Succeeded (render value)++versionText :: StreamVersion -> Text+versionText (StreamVersion version) = Text.pack (show version)++nameText :: StreamName -> Text+nameText (StreamName name) = name

@@ -6,20 +6,26 @@ module Keiro.Ops.Embed   ( AppHooks (..),     OpsAuditConfig (..),+    AggregateInspectors,+    ProcessManagerInspectors,     emptyAppHooks,   ) where +import Keiro.Inspection.Aggregate (AggregateInspectors) import Keiro.Ops.ReplayAudit (OpsAuditConfig (..)) import Keiro.Ops.Timer (TimerFire) import Keiro.Ops.Workflow (ResumeHook)+import Keiro.ProcessManager.Inspect (ProcessManagerInspectors) import Keiro.Projection.Catalog.Operations (ProjectionCatalogOperations)  data AppHooks = AppHooks   { workflowResume :: !(Maybe ResumeHook),     timerFire :: !(Maybe TimerFire),     replayAudit :: !(Maybe OpsAuditConfig),-    projectionCatalog :: !(Maybe ProjectionCatalogOperations)+    projectionCatalog :: !(Maybe ProjectionCatalogOperations),+    aggregates :: !(Maybe AggregateInspectors),+    processManagers :: !(Maybe ProcessManagerInspectors)   }  emptyAppHooks :: AppHooks@@ -28,5 +34,7 @@     { workflowResume = Nothing,       timerFire = Nothing,       replayAudit = Nothing,-      projectionCatalog = Nothing+      projectionCatalog = Nothing,+      aggregates = Nothing,+      processManagers = Nothing     }

@@ -5,13 +5,20 @@     positiveIntReader,     nonNegativeReader,     nonNegativeIntReader,+    cursorReader,   ) where  import Data.Char (toLower)+import Data.Text qualified as Text import Data.Time (NominalDiffTime)-import Options.Applicative (ReadM, eitherReader)+import Keiro.Inspection.Cursor (InspectionCursor (..))+import Options.Applicative (ReadM, eitherReader, str) import Text.Read qualified as Read++-- | The owning read validates the opaque token's kind, owner, and payload.+cursorReader :: ReadM InspectionCursor+cursorReader = InspectionCursor . Text.pack <$> str  durationReader :: ReadM NominalDiffTime durationReader = eitherReader parseDuration

@@ -46,7 +46,7 @@ dlqParser :: Parser DlqCommand dlqParser =   hsubparser-    ( command "read" (info readParser (progDesc "Read and decode visible DLQ entries"))+    ( command "read" (info readParser (progDesc "Preview or reserve and decode visible DLQ entries for 30 seconds"))         <> command "redrive" (info redriveParser (progDesc "Preview or move DLQ entries back to the main queue"))         <> command "archive" (info archiveParser (progDesc "Preview or archive DLQ entries for retention"))         <> command "purge" (info purgeParser (progDesc "Preview or permanently purge a DLQ"))@@ -75,14 +75,16 @@  isMutation :: Command -> Bool isMutation (Dlq dlqCommand) = case dlqCommand of-  Read {} -> False+  Read {} -> True   Redrive {} -> True   Archive {} -> True   Purge {} -> True  runCommand :: OpsEnv -> Command -> IO OpsOutcome runCommand env (Dlq dlqCommand) = case dlqCommand of-  Read queue limit -> handlePgmq (runJobEff (pgmqRuntime env) (readDlq (rawJob queue) (fromIntegral limit))) (Succeeded . dlqListResult queue)+  Read queue limit+    | env.force -> handlePgmq (runJobEff (pgmqRuntime env) (readDlq (rawJob queue) (fromIntegral limit))) (Succeeded . dlqListResult queue)+    | otherwise -> previewFromDepth env "read" queue (Just limit) ["pgmq", "dlq", "read", "--queue", queue, "--limit", showText limit]   Redrive queue limit -> runRedrive env queue limit   Archive queue entry limit -> runArchive env queue entry limit   Purge queue -> runPurge env queue

@@ -0,0 +1,136 @@+-- | Thin read-only adapters over application-owned manager inspectors.+module Keiro.Ops.ProcessManager (Command (..), ShowOptions (..), ListOptions (..), commandParser, validateCommand, isMutation, runCommand, prepareTypes) where++import Data.Aeson qualified as Aeson+import Data.Aeson.Encoding (Encoding)+import Data.Aeson.Encoding qualified as Encoding+import Data.Aeson.KeyMap qualified as KeyMap+import Data.Map.Lazy qualified as LazyMap+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as Text+import Keiro.Inspection.Cursor (InspectionCursor, decodeCursor)+import Keiro.Ops.Env (OpsEnv (..))+import Keiro.Ops.Parse (cursorReader, positiveIntReader)+import Keiro.Ops.Render+import Keiro.ProcessManager.Inspect qualified as PM+import Kiroku.Store.Effect (runStoreIO)+import Options.Applicative hiding (command, info, value)+import Options.Applicative qualified as Opt++data ShowOptions = ShowOptions {name :: !Text, correlationId :: !Text, timerLimit :: !Int}+  deriving stock (Eq, Show)++data ListOptions = ListOptions {name :: !Text, after :: !(Maybe InspectionCursor), limit :: !Int}+  deriving stock (Eq, Show)++data Command = Types | Show !ShowOptions | List !ListOptions deriving stock (Eq, Show)++commandParser :: Parser Command+commandParser =+  hsubparser+    ( Opt.command "types" (Opt.info (pure Types) (progDesc "List mounted process managers"))+        <> Opt.command+          "show"+          ( Opt.info+              ( Show+                  <$> ( ShowOptions+                          <$> name+                          <*> textOption "correlation" "ID"+                          <*> option positiveIntReader (long "timer-limit" <> Opt.value 20 <> metavar "N")+                      )+              )+              (progDesc "Observe persisted manager state and intended dispatch witnesses")+          )+        <> Opt.command+          "list"+          ( Opt.info+              ( List+                  <$> ( ListOptions+                          <$> name+                          <*> optional (option cursorReader (long "after" <> metavar "CURSOR"))+                          <*> option positiveIntReader (long "limit" <> Opt.value 50 <> metavar "N")+                      )+              )+              (progDesc "Read one page of manager journal streams")+          )+    )+  where+    name = textOption "name" "NAME"+    textOption key label = Text.pack <$> strOption (long key <> metavar label)++isMutation :: Command -> Bool+isMutation = const False++validateCommand :: Command -> Either (Text, Text) ()+validateCommand Types = Right ()+validateCommand (Show options)+  | Text.null (Text.strip options.correlationId) = Left ("invalid_correlation_id", "correlation id must not be blank")+  | options.timerLimit < 1 || options.timerLimit > 500 = Left ("invalid_page_size", "timer page size must be 1..500")+  | otherwise = Right ()+validateCommand (List options)+  | options.limit < 1 || options.limit > 500 = Left ("invalid_page_size", "page size must be 1..500")+  | otherwise = case options.after of+      Nothing -> Right ()+      Just cursor -> case decodeCursor PM.instanceListCursorKind cursor of+        Right (owner, category, after :: Text) | owner == options.name && Text.isPrefixOf (category <> "-") after -> Right ()+        _ -> Left ("invalid_cursor", "cursor belongs to another manager inspector or is malformed")++runCommand :: PM.ProcessManagerInspectors -> OpsEnv -> Command -> IO OpsOutcome+runCommand catalog env command = case validateCommand command of+  Left (code, detail) -> pure (InspectionFailed code detail)+  Right () -> case command of+    Types -> pure (Succeeded (fst (prepareTypes catalog)))+    Show options -> case Map.lookup options.name catalog of+      Nothing -> unknown options.name+      Just (PM.SomeProcessManagerInspector inspector) -> do+        result <- runStoreIO env.store (PM.inspectProcessManagerInstance inspector (PM.InstanceViewRequest options.timerLimit) options.correlationId)+        pure $ case result of+          Left err -> Failed (Text.pack (show err))+          Right (Left err) -> Failed (Text.pack (show err))+          Right (Right value) ->+            Succeeded+              ( OpsResult+                  ["name", "correlation_id", "state"]+                  (maybe [] (\view -> [[view.name, view.correlationId, maybe "" id view.stateLabel]]) value)+                  (maybe Aeson.Null PM.processManagerViewJson value)+              )+    List options -> case Map.lookup options.name catalog of+      Nothing -> unknown options.name+      Just (PM.SomeProcessManagerInspector inspector) -> do+        result <- runStoreIO env.store (PM.listProcessManagerInstances inspector (PM.InstanceListRequest options.limit options.after))+        pure $ case result of+          Left err -> Failed (Text.pack (show err))+          Right (Left err) -> uncurry InspectionFailed (listError err)+          Right (Right page) ->+            Succeeded+              ( OpsResult+                  ["correlation_id", "stream_version"]+                  [[item.correlationId, Text.pack (show item.streamVersion)] | item <- page.items]+                  (PM.instancePageJson page)+              )+  where+    unknown owner = pure (Failed ("unknown process manager: " <> owner))+    listError err =+      ( case err of+          PM.InvalidInstancePageSize _ -> "invalid_page_size"+          PM.InvalidInstanceCursor _ -> "invalid_cursor"+          PM.InstanceResultTooLarge -> "inspection_result_too_large",+        Text.pack (show err)+      )++-- | Canonical presentation and an equivalent direct encoding from one traversal.+-- The host must consume the encoding inside its own byte/deadline budget.+-- Neither representation is retained between requests.+prepareTypes :: PM.ProcessManagerInspectors -> (OpsResult, Encoding)+prepareTypes catalog =+  let summaries = PM.listProcessManagerTypes catalog+      -- Defer the canonical array until a caller actually consumes JSON. HTTP+      -- can consume the direct encoding while retaining the same OpsResult API.+      value = Aeson.Object (KeyMap.fromMap (LazyMap.singleton "items" (Aeson.toJSON summaries)))+   in ( OpsResult+          ["name", "category", "snapshots_configured", "snapshot_policy", "timers_configured"]+          [[summary.name, summary.category, if summary.snapshotsConfigured then "true" else "false", summary.snapshotPolicy, maybe "unknown" (\configured -> if configured then "true" else "false") summary.timersConfigured] | summary <- summaries]+          value,+        Encoding.pairs (Encoding.pair "items" (Aeson.toEncoding summaries))+      )

@@ -4,6 +4,9 @@ -- with ADR 28. module Keiro.Ops.Projection   ( Command (..),+    ProjectionStatusOptions (..),+    projectionGroupStatusJson,+    validateCommand,     commandParser,     isMutation,     runCommand,@@ -18,10 +21,14 @@ import Data.Vector qualified as Vector import Effectful (Eff, IOE) import Effectful.Error.Static (Error)+import Keiro.Inspection.Cursor (InspectionCursor (..)) 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.ReadModel (storeHeadPosition)+import Keiro.ReadModel.Rebuild import Kiroku.Store.Effect (Store, runStoreIO) import Kiroku.Store.Error (StoreError) import Kiroku.Store.Subscription@@ -31,18 +38,23 @@     subscriptionCheckpointInventory,   ) import Kiroku.Store.Types (GlobalPosition (..))-import Options.Applicative hiding (action, value)+import Options.Applicative hiding (action) import Prelude +data ProjectionStatusOptions = ProjectionStatusOptions {limit :: !Int, after :: !(Maybe InspectionCursor)}+  deriving stock (Eq, Show)+ data Command-  = Position !Text+  = Status !ProjectionStatusOptions+  | Position !Text   | PruneDedup !Text !UTCTime   deriving stock (Eq, Show)  commandParser :: Parser Command commandParser =   hsubparser-    ( command "position" (info positionParser (progDesc "Show one subscription's durable member checkpoints and floor"))+    ( 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"))+        <> 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@@ -62,13 +74,49 @@     Right     (parseTimeM True defaultTimeLocale "%Y-%m-%dT%H:%M:%S%QZ" raw) +statusInput :: ProjectionStatusOptions -> ProjectionStatusPageRequest+statusInput options = ProjectionStatusPageRequest options.limit options.after++validateCommand :: Command -> Either (Text, Text) ()+validateCommand (Status options) = either (Left . statusError) (const (Right ())) (validateProjectionStatusRequest (statusInput options))+validateCommand _ = Right ()++statusError :: ProjectionStatusReadError -> (Text, Text)+statusError err =+  ( case err of+      InvalidProjectionStatusPageSize _ -> "invalid_page_size"+      InvalidProjectionStatusCursor _ -> "invalid_cursor"+      ProjectionStatusResultTooLarge -> "inspection_result_too_large",+    Text.pack (show err)+  )+ isMutation :: Command -> Bool isMutation = \case+  Status {} -> False   Position {} -> False   PruneDedup {} -> True  runCommand :: OpsEnv -> Command -> IO OpsOutcome runCommand env = \case+  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]+            ( object+                ( ["observed_at" .= page.observedAt, "visible_store_head" .= positionInt head, "items" .= map projectionGroupStatusJson page.items]+                    <> maybe [] (\cursor -> ["next_cursor" .= cursor.cursorText]) page.next+                )+            )+        )+    where+      action = do+        page <- listProjectionGroupStatusesPage (statusInput options)+        case page of+          Left err -> pure (Left err)+          Right value -> Right . (value,) <$> storeHeadPosition   Position subscription -> runPosition env subscription   PruneDedup projection before     | env.force ->@@ -79,6 +127,33 @@           PreviewRequired             (pruneResult True projection before affected)             (forceInvocation env ["projection", "prune-dedup", "--projection", projection, "--before", utcText before])++-- | 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+    ]++basisText :: ServingPositionBasis -> Text+basisText = \case ServingPositionAppend -> "append"; ServingPositionCheckpoint -> "checkpoint"; ServingPositionUnmanaged -> "unmanaged"  runPosition :: OpsEnv -> Text -> IO OpsOutcome runPosition env subscription =

@@ -34,6 +34,7 @@   | SucceededWithExit !OpsResult !ExitCode   | PreviewRequired !OpsResult !Text   | Failed !Text+  | InspectionFailed !Text !Text   deriving stock (Eq, Show)  emptyResult :: OpsResult

@@ -10,11 +10,12 @@   ) where -import Data.Aeson (object, (.=))+import Data.Aeson (Value, object, (.=))+import Data.Map.Strict qualified as Map import Data.Set qualified as Set import Data.Text (Text) import Data.Text qualified as Text-import Data.Time (UTCTime, getCurrentTime)+import Data.Time (UTCTime) import Data.UUID (UUID) import Data.UUID qualified as UUID import Effectful (Eff, IOE, (:>))@@ -28,14 +29,16 @@ import Options.Applicative hiding (action, value)  data Command-  = Status !Text+  = List+  | Status !Text   | Relinquish !Text !WorkerId   deriving stock (Eq, Show)  commandParser :: Parser Command commandParser =   hsubparser-    ( command "status" (info statusParser (progDesc "Show shard counts and ownership"))+    ( command "list" (info (pure List) (progDesc "List sharded subscriptions and live ownership counts"))+        <> command "status" (info statusParser (progDesc "Show shard counts and ownership"))         <> command "relinquish" (info relinquishParser (progDesc "Preview or release every bucket owned by one worker"))     )   where@@ -50,22 +53,16 @@  isMutation :: Command -> Bool isMutation = \case+  List -> False   Status {} -> False   Relinquish {} -> True  runCommand :: OpsEnv -> Command -> IO OpsOutcome runCommand env = \case-  Status name -> do-    now <- getCurrentTime-    runAction env (statusAction name) (Succeeded . uncurry (statusResult now name))+  List -> runAction env listShardedSubscriptions (Succeeded . listResult)+  Status name -> runAction env (shardOwnershipView (SubscriptionName name)) (Succeeded . statusResult)   Relinquish name worker -> runRelinquish env name worker -statusAction :: (Store :> es) => Text -> Eff es ([(Int, Maybe WorkerId, Maybe UTCTime)], [(Int, Int)])-statusAction name = do-  ownership <- ownershipSnapshotFor (SubscriptionName name)-  counts <- shardCountSnapshot (SubscriptionName name)-  pure (ownership, counts)- runRelinquish :: OpsEnv -> Text -> WorkerId -> IO OpsOutcome runRelinquish env name worker   | not env.force =@@ -94,38 +91,66 @@   result <- runStoreIO env.store action   pure $ either (Failed . Text.pack . show) onSuccess result -statusResult :: UTCTime -> Text -> [(Int, Maybe WorkerId, Maybe UTCTime)] -> [(Int, Int)] -> OpsResult-statusResult now name ownership counts =+statusResult :: ShardOwnershipView -> OpsResult+statusResult view =   OpsResult     { headers = ["subscription", "bucket", "owner", "lease_expires_at", "lease_state", "configured_shards"],-      rows = map rowText ownership,-      jsonValue = object ["subscription" .= name, "shard_counts" .= map countJson counts, "ownership" .= map ownershipJson ownership]+      rows = map rowText view.buckets,+      jsonValue =+        object+          [ "subscription" .= name,+            "observed_at" .= view.observedAt,+            "shard_counts" .= map countJson counts,+            "ownership" .= map ownershipJson view.buckets+          ]     }   where+    SubscriptionName name = view.subscriptionName+    counts = Map.toAscList (Map.fromListWith (+) [(row.shardCount, 1 :: Int) | row <- view.buckets])     configured = Text.intercalate "," [showText shardCount <> " (" <> showText rowCount <> " rows)" | (shardCount, rowCount) <- counts]-    rowText (bucket, owner, expiresAt) =+    rowText row =       [ name,-        showText bucket,-        maybe "unowned" workerText owner,-        maybe "" timeText expiresAt,-        leaseState now owner expiresAt,+        showText row.bucket,+        maybe "unowned" workerText row.owner,+        maybe "" timeText row.leaseExpiresAt,+        leaseStateText (leaseStateAt view.observedAt row),         configured       ]-    countJson (shardCount, rowCount) = object ["shard_count" .= shardCount, "rows" .= rowCount]-    ownershipJson (bucket, owner, expiresAt) =+    ownershipJson row =       object-        [ "bucket" .= bucket,-          "owner" .= fmap workerText owner,-          "lease_expires_at" .= expiresAt,-          "lease_state" .= leaseState now owner expiresAt+        [ "bucket" .= row.bucket,+          "owner" .= fmap workerText row.owner,+          "lease_expires_at" .= row.leaseExpiresAt,+          "lease_state" .= leaseStateText (leaseStateAt view.observedAt row),+          "heartbeat_at" .= row.heartbeatAt,+          "updated_at" .= row.updatedAt,+          "lease_age_seconds" .= fmap (\age -> realToFrac age :: Double) (leaseAgeAt view.observedAt row)         ] -leaseState :: UTCTime -> Maybe WorkerId -> Maybe UTCTime -> Text-leaseState _ Nothing _ = "unowned"-leaseState now (Just _) (Just expiry)-  | expiry < now = "expired"-  | otherwise = "live"-leaseState _ (Just _) Nothing = "invalid"+countJson :: (Int, Int) -> Value+countJson (shardCount, rowCount) = object ["shard_count" .= shardCount, "rows" .= rowCount]++listResult :: ShardedSubscriptionList -> OpsResult+listResult view =+  OpsResult+    ["subscription", "buckets_total", "buckets_live", "live_owners"]+    [[nameText row.subscriptionName, showText row.bucketsTotal, showText row.bucketsLive, showText row.liveOwners] | row <- view.subscriptions]+    (object ["observed_at" .= view.observedAt, "items" .= map summaryJson view.subscriptions])+  where+    nameText (SubscriptionName name) = name+    summaryJson row =+      object+        [ "subscription" .= nameText row.subscriptionName,+          "shard_counts" .= map countJson row.shardCounts,+          "buckets_total" .= row.bucketsTotal,+          "buckets_live" .= row.bucketsLive,+          "buckets_expired" .= row.bucketsExpired,+          "buckets_unowned" .= row.bucketsUnowned,+          "live_owners" .= row.liveOwners+        ]++leaseStateText :: LeaseState -> Text+leaseStateText = \case LeaseLive -> "live"; LeaseExpired -> "expired"; LeaseUnowned -> "unowned"; LeaseInvalid -> "invalid"  relinquishResult :: Bool -> Text -> WorkerId -> [Int] -> OpsResult relinquishResult preview name worker buckets =

@@ -15,10 +15,12 @@  import Data.Aeson (Value, object, (.=)) import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KeyMap import Data.ByteString.Lazy qualified as LazyByteString import Data.Int (Int64) import Data.Text (Text) import Data.Text qualified as Text+import Data.Time (diffUTCTime) import Effectful (Eff, IOE, (:>)) import Effectful.Error.Static (Error) import Keiro.Ops.Env (OpsEnv (..), OutputMode (..))@@ -92,7 +94,7 @@  runCommand :: OpsEnv -> Command -> IO OpsOutcome runCommand env = \case-  Show name -> runAction env (lookupByName name) (Succeeded . snapshotResult name)+  Show name -> runAction env (observationByName name) (Succeeded . snapshotObservedResult name)   Delete name -> runDelete env name   TruncationPreflight name before suppliedExpected ->     runAction env (preflightFor name before suppliedExpected) (Succeeded . preflightResult)@@ -104,6 +106,11 @@     Nothing -> pure Nothing     Just info -> lookupSnapshotRow info.id +observationByName :: (Store :> es) => Text -> Eff es (Maybe SnapshotObservation)+observationByName name = do+  stream <- getStream (StreamName name)+  traverse (observeSnapshotRow . (.id)) stream+ runDelete :: OpsEnv -> Text -> IO OpsOutcome runDelete env name   | not env.force =@@ -193,6 +200,21 @@       rows = maybe [] (pure . snapshotRowText name) row,       jsonValue = maybe Aeson.Null (snapshotJson name) row     }++snapshotObservedResult :: Text -> Maybe SnapshotObservation -> OpsResult+snapshotObservedResult name observation =+  case observation of+    Just observed+      | Just row <- observed.row ->+          let age = realToFrac (diffUTCTime observed.observedAt row.updatedAt) :: Double+              original = snapshotResult name (Just row)+              fields = KeyMap.fromList [("observed_at", Aeson.toJSON observed.observedAt), ("age_seconds", Aeson.toJSON age)]+           in original+                { headers = original.headers <> ["observed_at", "age_seconds"],+                  rows = map (<> [showText observed.observedAt, showText age]) original.rows,+                  jsonValue = case original.jsonValue of Aeson.Object value -> Aeson.Object (value <> fields); value -> value+                }+    _ -> snapshotResult name Nothing  snapshotRowText :: Text -> SnapshotRow -> [Text] snapshotRowText name row =

@@ -2,6 +2,9 @@   ( Command (..),     DrainOptions (..),     StuckListOptions (..),+    PendingListOptions (..),+    pendingTimerItemJson,+    validateCommand,     TimerFire,     commandParser,     commandParserWithDrain,@@ -20,8 +23,9 @@ import Data.UUID qualified as UUID import Effectful (Eff, IOE) import Effectful.Error.Static (Error)+import Keiro.Inspection.Cursor (InspectionCursor (..)) import Keiro.Ops.Env (OpsEnv (..), OutputMode (..))-import Keiro.Ops.Parse (durationReader, nonNegativeIntReader, positiveIntReader)+import Keiro.Ops.Parse (cursorReader, durationReader, nonNegativeIntReader, positiveIntReader) import Keiro.Ops.Render import Keiro.Timer import Kiroku.Store.Effect (Store, runStoreIO)@@ -36,6 +40,10 @@   }   deriving stock (Eq, Show) +data PendingListOptions = PendingListOptions+  {manager :: !(Maybe Text), correlation :: !(Maybe Text), after :: !(Maybe InspectionCursor), limit :: !Int}+  deriving stock (Eq, Show)+ data DrainOptions = DrainOptions   { limit :: !Int   }@@ -45,6 +53,7 @@  data Command   = StuckList !StuckListOptions+  | PendingList !PendingListOptions   | Requeue !TimerId   | Cancel !TimerId   | DeadLetter !TimerId !Text@@ -57,13 +66,14 @@ commandParserWithDrain :: Bool -> Parser Command commandParserWithDrain includeDrain =   hsubparser-    ( command-        "stuck"-        ( info-            stuckCommandParser-            (progDesc "Find timers left firing by an interrupted worker")-        )+    ( command "pending" (info pendingCommandParser (progDesc "Read a bounded page of scheduled and firing timers"))         <> command+          "stuck"+          ( info+              stuckCommandParser+              (progDesc "Find timers left firing by an interrupted worker")+          )+        <> command           "requeue"           ( info               (Requeue <$> timerIdArgument)@@ -97,6 +107,42 @@             )       | otherwise = mempty +pendingCommandParser :: Parser Command+pendingCommandParser =+  hsubparser+    ( command+        "list"+        ( info+            ( PendingList+                <$> ( PendingListOptions+                        <$> optional (Text.pack <$> strOption (long "manager" <> metavar "NAME"))+                        <*> optional (Text.pack <$> strOption (long "correlation" <> metavar "ID"))+                        <*> optional (option cursorReader (long "after" <> metavar "CURSOR"))+                        <*> option positiveIntReader (long "limit" <> metavar "N" <> Optparse.value 100)+                    )+            )+            (progDesc "Observe pending timers without claiming them")+        )+    )++pendingInput :: PendingListOptions -> (PendingTimerFilter, PendingTimerPageRequest)+pendingInput options = (PendingTimerFilter options.manager options.correlation, PendingTimerPageRequest options.limit options.after)++validateCommand :: Command -> Either (Text, Text) ()+validateCommand (PendingList options) =+  let (timerFilter, request) = pendingInput options+   in either (Left . pendingError) (const (Right ())) (validatePendingTimerRequest timerFilter request)+validateCommand _ = Right ()++pendingError :: PendingTimerReadError -> (Text, Text)+pendingError err =+  ( case err of+      InvalidPendingTimerPageSize _ -> "invalid_page_size"+      InvalidPendingTimerCursor _ -> "invalid_cursor"+      PendingTimerResultTooLarge -> "inspection_result_too_large",+    Text.pack (show err)+  )+ stuckCommandParser :: Parser Command stuckCommandParser =   hsubparser@@ -123,6 +169,7 @@ isMutation :: Command -> Bool isMutation = \case   StuckList {} -> False+  PendingList {} -> False   Requeue {} -> True   Cancel {} -> True   DeadLetter {} -> True@@ -134,10 +181,30 @@ runCommandWithFire :: Maybe TimerFire -> OpsEnv -> Command -> IO OpsOutcome runCommandWithFire timerFire env = \case   StuckList options -> runStuckList env options+  PendingList options -> runPendingList env options   Requeue timerId -> runMutation env RequeueOperation timerId Nothing   Cancel timerId -> runMutation env CancelOperation timerId Nothing   DeadLetter timerId reason -> runMutation env DeadLetterOperation timerId (Just reason)   DrainOnce options -> runDrainOnce timerFire env options++runPendingList :: OpsEnv -> PendingListOptions -> IO OpsOutcome+runPendingList env options = do+  let (timerFilter, request) = pendingInput options+  runAction env (findPendingTimers timerFilter request) $ \case+    Left err -> uncurry InspectionFailed (pendingError err)+    Right page ->+      Succeeded+        ( OpsResult+            ["id", "manager", "correlation", "fire_at", "status", "attempts", "payload"]+            (map (timerRow . (.timer)) page.timers)+            ( object+                ( ["observed_at" .= page.observedAt, "items" .= map (pendingTimerItemJson page.observedAt) page.timers]+                    <> maybe [] (\manager -> ["manager" .= manager]) options.manager+                    <> maybe [] (\correlation -> ["correlation_id" .= correlation]) options.correlation+                    <> maybe [] (\cursor -> ["next_cursor" .= cursor.cursorText]) page.next+                )+            )+        )  runDrainOnce :: Maybe TimerFire -> OpsEnv -> DrainOptions -> IO OpsOutcome runDrainOnce Nothing _ _ = pure (Failed "timer fire hook is not mounted")

@@ -1,6 +1,7 @@ module Keiro.Ops.Workflow   ( AwakeableCommand (..),     Command (..),+    InspectionCommand (..),     GcOptions (..),     InspectOptions (..),     ListOptions (..),@@ -9,39 +10,37 @@     ResumeOptions (..),     WorkflowRef (..),     commandParser,+    validateCommand,     commandParserWithResume,     isMutation,     runCommand,     runCommandWithResume,+    headerOutcome,   ) where  import Data.Aeson (Value, object, (.=)) import Data.Aeson qualified as Aeson-import Data.Aeson.Types qualified as AesonTypes-import Data.Int (Int64) import Data.List.NonEmpty qualified as NonEmpty-import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map-import Data.Maybe (catMaybes) import Data.Set qualified as Set import Data.Text (Text) import Data.Text qualified as Text import Data.Text.Encoding qualified as Text.Encoding-import Data.Time (NominalDiffTime, UTCTime, getCurrentTime)+import Data.Time (NominalDiffTime, UTCTime, defaultTimeLocale, getCurrentTime, parseTimeM) import Data.UUID (UUID) import Data.UUID qualified as UUID-import Data.Vector qualified as Vector import Effectful (Eff, IOE, (:>)) import Effectful.Error.Static (Error)-import Keiro.Codec (decodeRecorded)+import Keiro.Inspection.Cursor (InspectionCursor, maxCursorBytes) import Keiro.Ops.Env (OpsEnv (..), OutputMode (..))-import Keiro.Ops.Parse (durationReader, nonNegativeReader, positiveIntReader)+import Keiro.Ops.Parse (cursorReader, durationReader, nonNegativeReader, positiveIntReader) import Keiro.Ops.Render import Keiro.Workflow.Awakeable (AwakeableId (..), cancelAwakeable, signalAwakeable) import Keiro.Workflow.Awakeable.Schema qualified as Awakeable import Keiro.Workflow.Child.Schema qualified as Child import Keiro.Workflow.Gc qualified as Gc+import Keiro.Workflow.Inspection qualified as Inspection import Keiro.Workflow.Instance qualified as Instance import Keiro.Workflow.Resume   ( ResumeSummary (..),@@ -52,20 +51,10 @@ import Keiro.Workflow.Schema qualified as WorkflowSchema import Keiro.Workflow.Types   ( WorkflowId (..),-    WorkflowJournalEvent (..),     WorkflowName (..),-    awakeableAllocStepPrefix,-    cancelledStepName,-    completedStepName,-    continuedAsNewStepName,-    failedStepName,-    workflowGenerationStreamName,-    workflowJournalCodec,   ) import Kiroku.Store.Effect (Store, runStoreIO) import Kiroku.Store.Error (StoreError)-import Kiroku.Store.Read qualified as StoreRead-import Kiroku.Store.Types import Options.Applicative hiding (action, value) import Options.Applicative qualified as Optparse @@ -119,6 +108,7 @@  data Command   = List !ListOptions+  | Inspect !InspectionCommand   | Show !WorkflowRef   | Steps !InspectOptions   | Journal !InspectOptions@@ -130,6 +120,15 @@   | ResumeOnce !ResumeOptions   deriving stock (Eq, Show) +data InspectionCommand+  = InspectList !Inspection.WorkflowListQuery+  | InspectHeader !WorkflowRef+  | InspectJournal !WorkflowRef !Inspection.DetailPageQuery+  | InspectSteps !WorkflowRef !Inspection.DetailPageQuery+  | InspectChildren !WorkflowRef !Inspection.OwnerPageQuery+  | InspectAwakeables !WorkflowRef !Inspection.OwnerPageQuery+  deriving stock (Eq, Show)+ commandParser :: Parser Command commandParser = commandParserWithResume False @@ -137,6 +136,7 @@ commandParserWithResume includeResume =   hsubparser     ( command "list" (info (List <$> listOptionsParser) (progDesc "List workflow instances with stable keyset paging"))+        <> command "inspect" (info (Inspect <$> inspectionParser) (progDesc "Bounded workflow observation pages"))         <> command "show" (info (Show <$> workflowRefParser) (progDesc "Show an instance, its children, and its awakeables"))         <> command "steps" (info (Steps <$> inspectOptionsParser) (progDesc "Show the derived step index for one generation"))         <> command "journal" (info (Journal <$> inspectOptionsParser) (progDesc "Decode one workflow journal generation in order"))@@ -180,6 +180,40 @@     <$> workflowRefParser     <*> optional (option (nonNegativeReader "expected a non-negative generation") (long "generation" <> metavar "N" <> help "Journal generation; defaults to current")) +inspectionParser :: Parser InspectionCommand+inspectionParser =+  hsubparser+    ( command "list" (info (InspectList <$> listQuery) (progDesc "Page workflow instances"))+        <> 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"))+        <> command "children" (info (InspectChildren <$> workflowRefParser <*> ownerQuery) (progDesc "Page child links"))+        <> command "awakeables" (info (InspectAwakeables <$> workflowRefParser <*> ownerQuery) (progDesc "Page allocated external promises"))+    )+  where+    limit = option positiveIntReader (long "limit" <> metavar "N" <> Optparse.value 100 <> help "Page size, at most 500")+    after = optional (option cursorReader (long "after" <> metavar "CURSOR"))+    detailQuery =+      Inspection.DetailPageQuery+        <$> optional (option (nonNegativeReader "expected a non-negative generation") (long "generation" <> metavar "N"))+        <*> limit+        <*> after+    ownerQuery = Inspection.OwnerPageQuery <$> limit <*> after+    listQuery =+      Inspection.WorkflowListQuery+        <$> (NonEmpty.nonEmpty <$> many (option statusClassReader (long "status-class" <> metavar "CLASS")))+        <*> optional (Text.pack <$> strOption (long "workflow-name" <> metavar "NAME"))+        <*> optional (option utcReader (long "created-since" <> metavar "UTC"))+        <*> optional (option utcReader (long "created-before" <> metavar "UTC"))+        <*> optional (option utcReader (long "updated-since" <> metavar "UTC"))+        <*> optional (option utcReader (long "updated-before" <> metavar "UTC"))+        <*> option orderingReader (long "ordering" <> metavar "key|created_desc" <> Optparse.value Inspection.ByKey)+        <*> limit+        <*> after+    statusClassReader = eitherReader $ \raw -> maybe (Left "expected running, retrying, sleeping, awaiting, completed, cancelled, or failed") Right (Inspection.statusClassFromText (Text.pack raw))+    orderingReader = eitherReader $ \case "key" -> Right Inspection.ByKey; "created_desc" -> Right Inspection.ByCreatedDesc; _ -> Left "expected key or created_desc"+    utcReader = eitherReader $ \raw -> maybe (Left "expected a UTC timestamp such as 2026-10-20T01:02:03Z") Right (parseTimeM True defaultTimeLocale "%Y-%m-%dT%H:%M:%S%QZ" raw)+ workflowRefParser :: Parser WorkflowRef workflowRefParser =   WorkflowRef@@ -245,6 +279,7 @@  isMutation :: Command -> Bool isMutation = \case+  Inspect {} -> False   List {} -> False   Show {} -> False   Steps {} -> False@@ -263,6 +298,7 @@  runCommandWithResume :: Maybe ResumeHook -> OpsEnv -> Command -> IO OpsOutcome runCommandWithResume resumeHook env = \case+  Inspect inspection -> runInspection env inspection   List options -> runList env options   Show ref -> runShow env ref   Steps options -> runSteps env options@@ -274,6 +310,53 @@   GcRunOnce options -> runGc env options   ResumeOnce options -> runResumeOnce resumeHook env options +runInspection :: OpsEnv -> InspectionCommand -> IO OpsOutcome+runInspection env = \case+  InspectList query -> page (Inspection.listWorkflowInstancesPage query)+  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)+  InspectChildren ref query -> page (Inspection.listWorkflowChildrenPage (refName ref) (refId ref) query)+  InspectAwakeables ref query -> page (Inspection.listWorkflowAwakeablesPage (refName ref) (refId ref) query)+  where+    page :: (Aeson.ToJSON a) => Eff '[Store, Error StoreError, IOE] (Either Inspection.InspectionError a) -> IO OpsOutcome+    page action = runAction env action (either inspectionFailure (Succeeded . jsonResult . Aeson.toJSON))+    jsonResult value = OpsResult ["result"] [[jsonText value]] value++-- | Canonical CLI and HTTP outcome for a freshly observed header.+headerOutcome :: WorkflowRef -> Maybe Instance.WorkflowInstanceRow -> OpsOutcome+headerOutcome ref = \case+  Nothing -> Failed (workflowLabel ref <> " was not found")+  Just row ->+    let value = Inspection.instanceToJson row+     in Succeeded (OpsResult ["result"] [[jsonText value]] value)++inspectionFailure :: Inspection.InspectionError -> OpsOutcome+inspectionFailure err = uncurry InspectionFailed (inspectionError err)++inspectionError :: Inspection.InspectionError -> (Text, Text)+inspectionError err = (code, message)+  where+    message = case err of+      Inspection.CursorResultTooLarge -> "the continuation key exceeds the " <> Text.pack (show maxCursorBytes) <> "-byte cursor limit"+      _ -> Text.pack (show err)+    code = case err of+      Inspection.InvalidPageSize _ -> "invalid_page_size"+      Inspection.InvalidCursor _ -> "invalid_cursor"+      Inspection.CursorGenerationMismatch _ _ -> "cursor_generation_mismatch"+      Inspection.CursorResultTooLarge -> "inspection_result_too_large"++-- | Validate the owning library's inspection grammar without any SQL.+validateCommand :: Command -> Either (Text, Text) ()+validateCommand (Inspect command) = either (Left . inspectionError) Right $ case command of+  InspectList query -> Inspection.validateWorkflowListQuery query+  InspectHeader _ -> Right ()+  InspectJournal ref query -> Inspection.validateJournalQuery (refName ref) (refId ref) query+  InspectSteps ref query -> Inspection.validateStepQuery (refName ref) (refId ref) query+  InspectChildren ref query -> Inspection.validateChildQuery (refName ref) (refId ref) query+  InspectAwakeables ref query -> Inspection.validateAwakeableQuery (refName ref) (refId ref) query+validateCommand _ = Right ()+ runResumeOnce :: Maybe ResumeHook -> OpsEnv -> ResumeOptions -> IO OpsOutcome runResumeOnce Nothing _ _ =   pure (Failed "workflow resume hook is not mounted")@@ -360,41 +443,49 @@         options.limit  runShow :: OpsEnv -> WorkflowRef -> IO OpsOutcome-runShow env ref =-  runAction env action $ \case-    Nothing -> Failed (workflowLabel ref <> " was not found")-    Just details -> Succeeded (workflowDetailsResult details)+runShow env ref = runAction env action $ \case+  Left err -> inspectionFailure err+  Right Nothing -> Failed (workflowLabel ref <> " was not found")+  Right (Just details) -> Succeeded (workflowDetailsResult details)   where     action = do       instanceRow <- Instance.lookupInstance (refName ref) (refId ref)       case instanceRow of-        Nothing -> pure Nothing+        Nothing -> pure (Right Nothing)         Just row -> do-          children <- Child.lookupChildrenOfParent ref.workflowId ref.workflowName-          awakeables <- lookupWorkflowAwakeables ref-          pure (Just (row, children, awakeables))+          children <- drainPages (\cursor -> Inspection.listWorkflowChildrenPage (refName ref) (refId ref) (Inspection.OwnerPageQuery 500 cursor)) (.items) (.nextCursor)+          awakeables <- drainPages (\cursor -> Inspection.listWorkflowAwakeablesPage (refName ref) (refId ref) (Inspection.OwnerPageQuery 500 cursor)) (.items) (.nextCursor)+          pure ((\cs as -> Just (row, cs, as)) <$> children <*> awakeables)  runSteps :: OpsEnv -> InspectOptions -> IO OpsOutcome-runSteps env options =-  runAction env action (Succeeded . uncurry stepsResult)+runSteps env options = runAction env action $ \(gen, rows) -> either inspectionFailure (Succeeded . stepsResult gen) rows   where     action = do-      selectedGeneration <- resolveGeneration options-      steps <- WorkflowSchema.loadStepIndex (refName options.target) (refId options.target) selectedGeneration-      pure (selectedGeneration, steps)+      gen <- resolveGeneration options+      rows <- drainPages (\cursor -> Inspection.loadStepIndexPage (refName options.target) (refId options.target) (Inspection.DetailPageQuery (Just gen) 500 cursor)) (.items) (.nextCursor)+      pure (gen, rows)  runJournal :: OpsEnv -> InspectOptions -> IO OpsOutcome-runJournal env options =-  runAction env action $ \(selectedGeneration, recorded) ->-    case traverse decodeJournalView recorded of-      Left err -> Failed err-      Right views -> Succeeded (journalResult selectedGeneration views)+runJournal env options = runAction env action $ \(gen, rows) -> either inspectionFailure (Succeeded . journalResult gen) rows   where     action = do-      selectedGeneration <- resolveGeneration options-      recorded <- readJournalEvents (workflowGenerationStreamName (refName options.target) (refId options.target) selectedGeneration)-      pure (selectedGeneration, recorded)+      gen <- resolveGeneration options+      rows <- drainPages (\cursor -> Inspection.readWorkflowJournalPage (refName options.target) (refId options.target) (Inspection.DetailPageQuery (Just gen) 500 cursor)) (.items) (.nextCursor)+      pure (gen, rows) +-- Compatibility commands intentionally drain pages. Bounded inspect commands+-- above invoke one library page only, and the HTTP deadline still bounds these.+drainPages :: (Maybe InspectionCursor -> Eff es (Either Inspection.InspectionError page)) -> (page -> [a]) -> (page -> Maybe InspectionCursor) -> Eff es (Either Inspection.InspectionError [a])+drainPages load items next = go Nothing []+  where+    go cursor chunks = do+      result <- load cursor+      case result of+        Left err -> pure (Left err)+        Right page -> case next page of+          Nothing -> pure (Right (concat (reverse (items page : chunks))))+          Just token -> go (Just token) (items page : chunks)+ runAwakeable :: OpsEnv -> AwakeableCommand -> IO OpsOutcome runAwakeable env = \case   AwakeableShow awakeableId ->@@ -514,76 +605,12 @@     pure     options.generation -lookupWorkflowAwakeables :: (Store :> es) => WorkflowRef -> Eff es [Awakeable.AwakeableRow]-lookupWorkflowAwakeables ref = do-  current <- WorkflowSchema.currentGeneration (refName ref) (refId ref)-  stepIndexes <- traverse (WorkflowSchema.loadStepIndex (refName ref) (refId ref)) [0 .. current]-  catMaybes <$> traverse Awakeable.lookupAwakeable (awakeableIds stepIndexes)--awakeableIds :: [Map Text Value] -> [UUID]-awakeableIds indexes =-  Set.toAscList . Set.fromList $ do-    index <- indexes-    (stepName, value) <- Map.toList index-    if awakeableAllocStepPrefix `Text.isPrefixOf` stepName-      then case Aeson.fromJSON value of-        AesonTypes.Success (AwakeableId awakeableId) -> [awakeableId]-        AesonTypes.Error _ -> []-      else []--readJournalEvents :: (Store :> es) => StreamName -> Eff es [RecordedEvent]-readJournalEvents streamName = go (StreamVersion 0) []-  where-    pageSize = 256-    go cursor pages = do-      page <- StoreRead.readStreamForward streamName cursor pageSize-      if Vector.null page-        then pure (concat (reverse pages))-        else-          let nextCursor = (Vector.last page).streamVersion-           in go nextCursor (Vector.toList page : pages)--data JournalView = JournalView-  { eventId :: !Text,-    eventType :: !Text,-    streamVersion :: !Int64,-    globalPosition :: !Int64,-    stepName :: !Text,-    recordedAt :: !UTCTime,-    payload :: !Value-  }--decodeJournalView :: RecordedEvent -> Either Text JournalView-decodeJournalView recorded = do-  event <- firstShow (decodeRecorded workflowJournalCodec recorded)-  let (stepName, recordedAt, payload) = case event of-        StepRecorded name value timestamp -> (name, timestamp, value)-        WorkflowCompleted timestamp -> (completedStepName, timestamp, Aeson.Null)-        WorkflowCancelled timestamp -> (cancelledStepName, timestamp, Aeson.Null)-        WorkflowFailed reason timestamp -> (failedStepName, timestamp, Aeson.toJSON reason)-        WorkflowContinuedAsNew generation timestamp -> (continuedAsNewStepName, timestamp, Aeson.toJSON generation)-  pure-    JournalView-      { eventId = case recorded.eventId of EventId value -> UUID.toText value,-        eventType = case recorded.eventType of EventType value -> value,-        streamVersion = case recorded.streamVersion of StreamVersion value -> value,-        globalPosition = case recorded.globalPosition of GlobalPosition value -> value,-        stepName,-        recordedAt,-        payload-      }--firstShow :: (Show err) => Either err value -> Either Text value-firstShow = \case-  Left err -> Left ("workflow journal decode failed: " <> Text.pack (show err))-  Right value -> Right value- workflowListResult :: [Instance.WorkflowInstanceRow] -> OpsResult workflowListResult instances =   OpsResult     { headers = ["name", "id", "generation", "status", "attempts", "lease", "wake_after", "updated_at"],       rows = map workflowListRow instances,-      jsonValue = Aeson.toJSON (map workflowInstanceJson instances)+      jsonValue = Aeson.toJSON (map Inspection.instanceToJson instances)     }  workflowListRow :: Instance.WorkflowInstanceRow -> [Text]@@ -598,24 +625,6 @@     timeText row.updatedAt   ] -workflowInstanceJson :: Instance.WorkflowInstanceRow -> Value-workflowInstanceJson row =-  object-    [ "workflow_id" .= row.workflowId,-      "workflow_name" .= row.workflowName,-      "generation" .= row.generation,-      "status" .= Instance.statusToText row.status,-      "attempts" .= row.attempts,-      "last_error" .= row.lastError,-      "next_attempt_at" .= row.nextAttemptAt,-      "wake_after" .= row.wakeAfter,-      "leased_by" .= row.leasedBy,-      "lease_expires_at" .= row.leaseExpiresAt,-      "created_at" .= row.createdAt,-      "updated_at" .= row.updatedAt,-      "completed_at" .= row.completedAt-    ]- workflowDetailsResult :: (Instance.WorkflowInstanceRow, [Child.ChildRow], [Awakeable.AwakeableRow]) -> OpsResult workflowDetailsResult (row, children, awakeables) =   OpsResult@@ -634,41 +643,25 @@         ],       jsonValue =         object-          [ "instance" .= workflowInstanceJson row,-            "children" .= map childJson children,-            "awakeables" .= map awakeableJson awakeables+          [ "instance" .= Inspection.instanceToJson row,+            "children" .= map Inspection.childToJson children,+            "awakeables" .= map Inspection.awakeableToJson awakeables           ]     } -childJson :: Child.ChildRow -> Value-childJson row =-  object-    [ "child_id" .= row.childId,-      "child_name" .= row.childName,-      "parent_id" .= row.parentId,-      "parent_name" .= row.parentName,-      "await_step" .= row.awaitStep,-      "status" .= Child.statusToText row.status,-      "result" .= row.result,-      "failure_reason" .= row.failureReason,-      "created_at" .= row.createdAt,-      "updated_at" .= row.updatedAt,-      "completed_at" .= row.completedAt-    ]--stepsResult :: Int -> Map Text Value -> OpsResult+stepsResult :: Int -> [WorkflowSchema.WorkflowStepRow] -> OpsResult stepsResult generation steps =   OpsResult     { headers = ["step", "result"],-      rows = [[name, truncateCell 120 (jsonText value)] | (name, value) <- Map.toAscList steps],+      rows = [[row.stepName, truncateCell 120 (jsonText row.result)] | row <- steps],       jsonValue =         object           [ "generation" .= generation,-            "steps" .= [object ["step" .= name, "result" .= value] | (name, value) <- Map.toAscList steps]+            "steps" .= map Inspection.stepToJson steps           ]     } -journalResult :: Int -> [JournalView] -> OpsResult+journalResult :: Int -> [Inspection.WorkflowJournalEntry] -> OpsResult journalResult generation views =   OpsResult     { headers = ["version", "event_type", "step", "recorded_at", "payload"],@@ -681,21 +674,9 @@           ]         | view <- views         ],-      jsonValue = object ["generation" .= generation, "events" .= map journalViewJson views]+      jsonValue = object ["generation" .= generation, "events" .= map Inspection.journalEntryToJson views]     } -journalViewJson :: JournalView -> Value-journalViewJson view =-  object-    [ "event_id" .= view.eventId,-      "event_type" .= view.eventType,-      "stream_version" .= view.streamVersion,-      "global_position" .= view.globalPosition,-      "step_name" .= view.stepName,-      "recorded_at" .= view.recordedAt,-      "payload" .= view.payload-    ]- awakeableResult :: Awakeable.AwakeableRow -> OpsResult awakeableResult row =   OpsResult@@ -709,22 +690,9 @@             timeText row.updatedAt           ]         ],-      jsonValue = awakeableJson row+      jsonValue = Inspection.awakeableToJson row     } -awakeableJson :: Awakeable.AwakeableRow -> Value-awakeableJson row =-  object-    [ "awakeable_id" .= UUID.toText row.awakeableId,-      "owner_workflow_name" .= row.ownerWorkflowName,-      "owner_workflow_id" .= row.ownerWorkflowId,-      "status" .= Awakeable.statusToText row.status,-      "payload" .= row.payload,-      "created_at" .= row.createdAt,-      "updated_at" .= row.updatedAt,-      "completed_at" .= row.completedAt-    ]- awakeablePreviewResult :: Text -> UUID -> Maybe Awakeable.AwakeableRow -> OpsResult awakeablePreviewResult operation awakeableId row =   OpsResult@@ -735,7 +703,7 @@           [ "preview" .= True,             "operation" .= operation,             "disposition" .= disposition,-            "awakeable" .= fmap awakeableJson row+            "awakeable" .= fmap Inspection.awakeableToJson row           ]     }   where@@ -756,7 +724,7 @@           [ "operation" .= operation,             "outcome" .= outcome,             "transitioned" .= transitioned,-            "awakeable" .= fmap awakeableJson row+            "awakeable" .= fmap Inspection.awakeableToJson row           ]     }   where@@ -775,7 +743,7 @@             "operation" .= operation,             "disposition" .= disposition,             "target" .= object ["workflow_name" .= ref.workflowName, "workflow_id" .= ref.workflowId],-            "instance" .= fmap workflowInstanceJson row+            "instance" .= fmap Inspection.instanceToJson row           ]     } 

@@ -0,0 +1,103 @@+-- | Bounded reuse of pure header presentation. Every lookup still reads the+-- complete current header in PostgreSQL; callers must verify live schema first.+module Keiro.Ops.WorkflowHeaderCache+  ( WorkflowHeaderCache,+    newWorkflowHeaderCache,+    workflowHeaderCacheUsage,+    runHeaderCached,+  )+where++import Data.Aeson (Value)+import Data.ByteString (ByteString)+import Data.ByteString qualified as ByteString+import Data.IORef+import Data.Map.Strict qualified as Map+import Data.Sequence qualified as Seq+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text+import Hasql.Pool qualified as Pool+import Keiro.Ops.Env (OpsEnv (..))+import Keiro.Ops.Render+import Keiro.Ops.Workflow (WorkflowRef (..), headerOutcome)+import Keiro.Workflow.Instance qualified as Instance+import Keiro.Workflow.Types (WorkflowId (..), WorkflowName (..))+import Kiroku.Store.Connection (KirokuStore (..))+import Kiroku.Store.Error (mapGenericUsageError)++type View = (Text, Text)++data Entry = Entry !Instance.WorkflowHeaderKey !OpsResult !ByteString !Integer++data Entries = Entries !(Map.Map View Entry) !(Seq.Seq View) !Integer++data WorkflowHeaderCache = WorkflowHeaderCache+  { entries :: !(IORef Entries),+    imageLimit :: !Int,+    encode :: Value -> IO (Maybe ByteString)+  }++-- | At most 32 presentations, charged at most 1 MiB in total. Charge includes+-- input keys, exact row images, JSON values, rendered cells and encoded bodies.+-- A larger valid response can still be served without retaining its rendering.+-- The encoder must enforce the host's response cap and return exact JSON bytes.+newWorkflowHeaderCache :: Int -> (Value -> IO (Maybe ByteString)) -> IO WorkflowHeaderCache+newWorkflowHeaderCache responseLimit encode+  | responseLimit <= 0 = ioError (userError "header response limit must be positive")+  | otherwise = do+      entries <- newIORef (Entries Map.empty Seq.empty 0)+      -- PostgreSQL's image has field names and date arrays independent of the+      -- public wire format. Allow its fixed overhead without an Int overflow.+      let imageLimit = fromInteger (min (toInteger (maxBound :: Int)) (2 * toInteger responseLimit + 4096))+      pure WorkflowHeaderCache {entries, imageLimit, encode}++-- | Charged retained bytes and entry count, for host diagnostics.+workflowHeaderCacheUsage :: WorkflowHeaderCache -> IO (Int, Integer)+workflowHeaderCacheUsage cache = do+  Entries known _ bytes <- readIORef cache.entries+  pure (Map.size known, bytes)++runHeaderCached :: WorkflowHeaderCache -> OpsEnv -> WorkflowRef -> IO (OpsOutcome, Maybe ByteString)+runHeaderCached cache env ref = do+  Entries known _ _ <- readIORef cache.entries+  let view = (ref.workflowName, ref.workflowId)+      previous = Map.lookup view known+      key = case previous of Just (Entry image _ _ _) -> Just image; Nothing -> Nothing+      forget = atomicModifyIORef' cache.entries (\entries -> (remove view entries, ()))+      failure code message = forget >> pure (InspectionFailed code message, Nothing)+  observed <- Pool.use env.store.pool (Instance.lookupInstanceConditionalSession cache.imageLimit key (WorkflowName ref.workflowName) (WorkflowId ref.workflowId))+  case observed of+    Left err -> forget >> pure (Failed (Text.pack (show (mapGenericUsageError err))), Nothing)+    Right Nothing -> forget >> pure (headerOutcome ref Nothing, Nothing)+    Right (Just Instance.HeaderImageTooLarge) -> failure "inspection_result_too_large" "workflow header image exceeds the host limit"+    Right (Just Instance.HeaderUnchanged) -> case previous of+      Just (Entry _ result body _) -> pure (Succeeded result, Just body)+      Nothing -> failure "inspection_header_inconsistent" "header comparison returned no initial image"+    Right (Just (Instance.HeaderChanged image row)) -> case headerOutcome ref (Just row) of+      Succeeded result -> do+        encoded <- cache.encode result.jsonValue+        case encoded of+          Nothing -> failure "inspection_result_too_large" "encoded result exceeds the host limit; narrow the read"+          Just body -> do+            let charge = 4096 + toInteger (ByteString.length (Text.encodeUtf8 ref.workflowName)) + toInteger (ByteString.length (Text.encodeUtf8 ref.workflowId)) + 3 * toInteger (Instance.workflowHeaderKeyBytes image) + 6 * toInteger (ByteString.length body)+                entry = Entry image result body charge+            atomicModifyIORef' cache.entries (\entries -> (insert view entry entries, ()))+            pure (Succeeded result, Just body)+      outcome -> forget >> pure (outcome, Nothing)++remove :: View -> Entries -> Entries+remove view (Entries known order bytes) = case Map.lookup view known of+  Nothing -> Entries known order bytes+  Just (Entry _ _ _ charge) -> Entries (Map.delete view known) (Seq.filter (/= view) order) (bytes - charge)++insert :: View -> Entry -> Entries -> Entries+insert view entry@(Entry _ _ _ charge) entries =+  let without = remove view entries+   in if charge > 1024 * 1024 then without else fit without+  where+    fit (Entries known order bytes)+      | Map.size known < 32 && bytes + charge <= 1024 * 1024 = Entries (Map.insert view entry known) (order Seq.|> view) (bytes + charge)+      | otherwise = case Seq.viewl order of+          oldest Seq.:< _ -> fit (remove oldest (Entries known order bytes))+          Seq.EmptyL -> Entries Map.empty Seq.empty 0

@@ -29,6 +29,7 @@ import Hasql.Transaction qualified as Tx import Keiro.DeadLetter import Keiro.Inbox qualified as Inbox+import Keiro.Inspection.Cursor (InspectionCursor (..)) import Keiro.Integration.Event import Keiro.Ops (AppHooks (..)) import Keiro.Ops qualified as Ops@@ -74,6 +75,7 @@ import Kiroku.Store.Types import Options.Applicative qualified as Optparse import Pgmq.Migration qualified as PgmqMigration+import ReactionInspectionSpec qualified import System.Exit (ExitCode (..)) import System.Process (readProcessWithExitCode) import Test.Hspec@@ -89,7 +91,9 @@     { workflowResume = Just (Map.empty, defaultWorkflowResumeOptions),       timerFire = Just (\_ -> pure Nothing),       replayAudit = Just (OpsReplayAudit.OpsAuditConfig []),-      projectionCatalog = Just emptyCatalogOperations+      projectionCatalog = Just emptyCatalogOperations,+      aggregates = Just Map.empty,+      processManagers = Just Map.empty     }  emptyCatalogOperations :: CatalogOperations.ProjectionCatalogOperations@@ -111,8 +115,12 @@  spec :: Fixture -> Spec spec fixture = do+  ReactionInspectionSpec.spec fixture   describe "embedded command tree" do     it "omits code-dependent commands from the standalone tree" do+      isParseSuccess (parseOps Ops.emptyAppHooks ["pm", "show", "--name", "test", "--correlation", "one"]) `shouldBe` False+      isParseSuccess (parseOps Ops.emptyAppHooks ["pm", "types"]) `shouldBe` False+      isParseSuccess (parseOps Ops.emptyAppHooks ["aggregate", "types"]) `shouldBe` False       isParseFailure (parseOps Ops.emptyAppHooks ["wf", "resume-once"]) `shouldBe` True       isParseFailure (parseOps Ops.emptyAppHooks ["timer", "drain-once"]) `shouldBe` True       isParseFailure (parseOps Ops.emptyAppHooks ["replay-audit", "--full"]) `shouldBe` True@@ -124,6 +132,14 @@       isParseFailure (parseOps Ops.emptyAppHooks ["rebuild", "reproject-stream", "ops-group", "ops-projection", "orders-1"]) `shouldBe` True      it "mounts every code-dependent command from typed application hooks" do+      isParseSuccess (parseOps embeddedHooks ["pm", "show", "--name", "test", "--correlation", "one"]) `shouldBe` True+      isParseSuccess (parseOps embeddedHooks ["pm", "list", "--name", "test"]) `shouldBe` True+      isParseSuccess (parseOps embeddedHooks ["pm", "types"]) `shouldBe` True+      isParseSuccess (parseOps embeddedHooks ["pm", "types", "--name", "test"]) `shouldBe` False+      isParseSuccess (parseOps embeddedHooks ["aggregate", "types"]) `shouldBe` True+      isParseSuccess (parseOps embeddedHooks ["aggregate", "show", "--name", "counter", "--id", "one"]) `shouldBe` True+      isParseSuccess (parseOps embeddedHooks ["aggregate", "snapshot", "--name", "counter", "--id", "one"]) `shouldBe` True+      isParseSuccess (parseOps embeddedHooks ["aggregate", "list", "--name", "counter", "--limit", "2"]) `shouldBe` True       isParseSuccess (parseOps embeddedHooks ["wf", "resume-once"]) `shouldBe` True       isParseSuccess (parseOps embeddedHooks ["timer", "drain-once"]) `shouldBe` True       isParseSuccess (parseOps embeddedHooks ["replay-audit", "--full"]) `shouldBe` True@@ -977,6 +993,34 @@       workflowStatus store ref `shouldReturn` Nothing    describe "timer handlers" $ around (withFreshStore fixture) do+    it "renders pending timer pages with reusable cursors and due time" $ \store -> do+      now <- getCurrentTime+      let request = timerRequest "018f5f43-8a70-7b9a-9a9b-59d391a76720" (addUTCTime (-60) now)+          another = timerRequest "018f5f43-8a70-7b9a-9a9b-59d391a76721" (addUTCTime (-60) now)+          options = OpsTimer.PendingListOptions Nothing Nothing Nothing 1+      expectStore store (runTransaction (Timer.scheduleTimerTx request >> Timer.scheduleTimerTx another))+      first <- OpsTimer.runCommand (opsEnv False store) (OpsTimer.PendingList options)+      case first of+        Succeeded result -> case result.jsonValue of+          Aeson.Object fields -> do+            Just (Aeson.String cursor) <- pure (KeyMap.lookup "next_cursor" fields)+            Just (Aeson.Array items) <- pure (KeyMap.lookup "items" fields)+            Vector.length items `shouldBe` 1+            case Vector.head items of+              Aeson.Object item -> KeyMap.lookup "due_in_seconds" item `shouldSatisfy` (\case Just (Aeson.Number n) -> n < 0; _ -> False)+              _ -> expectationFailure "expected timer item"+            second <- OpsTimer.runCommand (opsEnv False store) (OpsTimer.PendingList (options {OpsTimer.after = Just (InspectionCursor cursor)}))+            case second of+              Succeeded rendered -> case rendered.jsonValue of+                Aeson.Object page -> KeyMap.lookup "next_cursor" page `shouldBe` Nothing+                _ -> expectationFailure "expected timer page"+              _ -> expectationFailure (show second)+          _ -> expectationFailure "expected timer page"+        _ -> expectationFailure (show first)+      filtered <- OpsTimer.runCommand (opsEnv False store) (OpsTimer.PendingList (options {OpsTimer.manager = Just "missing"}))+      filtered `shouldSatisfy` isSucceeded+      timerStatus store request.timerId `shouldReturn` Just Timer.Scheduled+     it "previews and dispatches one bounded due-timer pass through the mounted hook" $ \store -> do       now <- getCurrentTime       let request = timerRequest "018f5f43-8a70-7b9a-9a9b-59d391a76722" (addUTCTime (-60) now)@@ -1253,6 +1297,22 @@       dlqDepth `shouldReturn` 0    describe "projection handlers" $ around (withFreshStore fixture) do+    it "renders bounded projection status through the owning public relation" $ \store -> do+      expectStore store (runTransaction (Tx.sql "CREATE SCHEMA app; CREATE TABLE app.ops_catalog (id bigint PRIMARY KEY)"))+      catalog <- expectValidatedCatalog (opsCatalog "status-codec")+      _ <- expectStore store (Rebuild.registerProjectionCatalog catalog)+      Right page <- expectStore store (Rebuild.listProjectionGroupStatusesPage (Rebuild.ProjectionStatusPageRequest 1 Nothing))+      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.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+          _ -> expectationFailure "expected projection status page"+        _ -> expectationFailure (show shown)+     it "prunes only the named dedup rows" $ \store -> do       _ <- seedKirokuEvent store "projection-source" "018f5f43-8a70-7b9a-9a9b-59d391a76810" Nothing       events <- expectStore store (readStreamForward (StreamName "projection-source") (StreamVersion 0) 1)@@ -1285,6 +1345,32 @@       status <- OpsShard.runCommand (opsEnv False store) (OpsShard.Status "ops-shards")       resultArrayLengthFromObject "ownership" status `shouldBe` Just 2 +      detailed <- expectStore store (Shard.shardOwnershipView subscription)+      case status of+        Succeeded result -> case result.jsonValue of+          Aeson.Object fields -> do+            KeyMap.member "observed_at" fields `shouldBe` True+            Just (Aeson.Array ownership) <- pure (KeyMap.lookup "ownership" fields)+            Vector.length ownership `shouldBe` length detailed.buckets+            case Vector.head ownership of+              Aeson.Object row -> do+                KeyMap.member "heartbeat_at" row `shouldBe` True+                KeyMap.member "lease_age_seconds" row `shouldBe` True+              _ -> expectationFailure "expected ownership row"+          _ -> expectationFailure "expected shard status"+        _ -> expectationFailure (show status)+      listed <- OpsShard.runCommand (opsEnv False store) OpsShard.List+      case listed of+        Succeeded result -> case result.jsonValue of+          Aeson.Object fields -> do+            Just (Aeson.Array items) <- pure (KeyMap.lookup "items" fields)+            Vector.length items `shouldBe` 1+            case Vector.head items of+              Aeson.Object row -> KeyMap.lookup "buckets_live" row `shouldBe` Just (Aeson.toJSON (2 :: Int))+              _ -> expectationFailure "expected shard summary"+          _ -> expectationFailure "expected shard list"+        _ -> expectationFailure (show listed)+       preview <- OpsShard.runCommand (opsEnv False store) (OpsShard.Relinquish "ops-shards" worker)       preview `shouldSatisfy` isPreview       ownersBefore <- expectStore store (Shard.ownershipSnapshotFor subscription)@@ -1316,6 +1402,15 @@               regfileShapeHash = expected.regfileShapeHash,               stateShapeHash = expected.stateShapeHash             }++      shown <- OpsSnapshot.runCommand (opsEnv False store) (OpsSnapshot.Show "snapshot-ops")+      case shown of+        Succeeded result -> case result.jsonValue of+          Aeson.Object fields -> do+            KeyMap.lookup "observed_at" fields `shouldSatisfy` (\case Just (Aeson.String _) -> True; _ -> False)+            KeyMap.lookup "age_seconds" fields `shouldSatisfy` (\case Just (Aeson.Number n) -> n >= 0; _ -> False)+          _ -> expectationFailure "expected snapshot JSON object"+        _ -> expectationFailure "expected snapshot show success"        missing <- OpsSnapshot.runCommand (opsEnv False store) (OpsSnapshot.TruncationPreflight "no-snapshot" (StreamVersion 2) (Just expected))       jsonBool "passed" missing `shouldBe` Just False

@@ -0,0 +1,170 @@+module ReactionInspectionSpec (spec) where++import Control.Exception (finally)+import Control.Monad (when)+import Data.Aeson (Value (..), object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.Encoding qualified as Encoding+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KeyMap+import Data.ByteString.Lazy qualified as Lazy+import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Time (getCurrentTime)+import Data.UUID qualified as UUID+import Keiki.Core+import Keiro+import Keiro.Ops qualified as Ops+import Keiro.Ops.Env+import Keiro.Ops.ProcessManager qualified as PM+import Keiro.Ops.Render+import Keiro.Ops.WorkflowHeaderCache (newWorkflowHeaderCache, workflowHeaderCacheUsage)+import Keiro.ProcessManager.Inspect qualified as Inspect+import Keiro.ProcessManager.Reaction qualified as Reaction+import Keiro.Stream qualified as Stream+import Keiro.Test.Postgres+import Keiro.Timer qualified as Timer+import Kiroku.Store.Connection (ConnectionSettingsM (..))+import Kiroku.Store.Effect (runStoreIO)+import Kiroku.Store.Read (getStream)+import Kiroku.Store.Types hiding (id)+import Options.Applicative qualified as Options+import Test.Hspec++spec :: Fixture -> Spec+spec fixture = describe "reaction process-manager commands" $+  around (withFreshResourceStore fixture) $ do+    it "lists metadata with an unusable store and fully renders both formats" $ \_ -> do+      armed <- newIORef False+      let observe _ = readIORef armed >>= \active -> when active (fail "pm types touched the store")+      withFreshResourceStoreWith fixture (\settings -> settings {observationHandler = Just observe}) $ \(unusable, _) ->+        flip finally (writeIORef armed False) $ do+          writeIORef armed True+          let unused = error "pm types evaluated an application callback"+              manager = Reaction.ReactiveProcessManager "modern" id handler (\_ -> unused) definition (const []) (\_ -> unused)+              descriptor =+                (Inspect.reactiveProcessManagerInspector manager (Stream.categoryUnsafe "opsreaction") (\_ _ -> unused))+                  { Inspect.correlationOf = \_ -> unused,+                    Inspect.stateLabel = \_ -> unused,+                    Inspect.enabledInputs = \_ -> unused+                  }+              catalog = Map.fromList [("z-alias", Inspect.SomeProcessManagerInspector descriptor), ("a-alias", Inspect.SomeProcessManagerInspector descriptor)]+              hooks = Ops.emptyAppHooks {Ops.processManagers = Just catalog}+              env = OpsEnv unusable HumanTable False [] False+          runStoreIO unusable (getStream (StreamName "sentinel")) `shouldThrow` anyIOException+          PM.validateCommand PM.Types `shouldBe` Right ()+          PM.isMutation PM.Types `shouldBe` False+          Succeeded result <- PM.runCommand catalog env PM.Types+          result.rows `shouldBe` [["a-alias", "opsreaction", "false", "never", "unknown"], ["z-alias", "opsreaction", "false", "never", "unknown"]]+          Text.length (renderHuman result) `shouldSatisfy` (> 0)+          Text.length (jsonText result.jsonValue) `shouldSatisfy` (> 0)+          case Options.execParserPure Options.defaultPrefs (Options.info (Ops.commandParser hooks) mempty) ["pm", "types"] of+            Options.Success command -> Ops.runCommand hooks env command `shouldReturn` Succeeded result+            _ -> expectationFailure "pm types did not parse"+          Succeeded empty <- PM.runCommand Map.empty env PM.Types+          empty.jsonValue `shouldBe` object ["items" .= ([] :: [Value])]+          cache <- newWorkflowHeaderCache 1048576 (const (fail "metadata used the header cache encoder"))+          calls <- newIORef (0 :: Int)+          let encodeDirect encoding = do+                modifyIORef' calls (+ 1)+                pure (Just (Lazy.toStrict (Encoding.encodingToLazyByteString encoding)))+              command = Ops.ProcessManager PM.Types+          (Succeeded prepared, Just bytes) <- Ops.runCommandWithReadEncoding encodeDirect cache hooks env command+          prepared `shouldBe` result+          Aeson.eitherDecodeStrict bytes `shouldBe` Right result.jsonValue+          _ <- Ops.runCommandWithReadEncoding encodeDirect cache hooks env command+          readIORef calls `shouldReturn` 2+          workflowHeaderCacheUsage cache `shouldReturn` (0, 0)+          (refused, body) <- Ops.runCommandWithReadEncoding (const (pure Nothing)) cache hooks env command+          refused `shouldSatisfy` (\case InspectionFailed "inspection_result_too_large" _ -> True; _ -> False)+          body `shouldBe` Nothing+    it "mounts a handwritten reaction inspector through pm show and pm list" $ \(store, StoreRunner run) -> do+      clock <- getCurrentTime+      let source = RecordedEvent (EventId UUID.nil) (EventType "Input") (StreamVersion 1) (GlobalPosition 1) (StreamId 1) (StreamVersion 1) Null Nothing Nothing Nothing clock+          timer = Timer.TimerRequest (Timer.TimerId UUID.nil) "modern" "one" clock (object [])+          manager =+            Reaction.ReactiveProcessManager+              "modern"+              id+              handler+              (\cid -> stream ("opsreaction-" <> cid))+              definition+              (const [])+              (const (Reaction.AdvanceReaction Go [] [Reaction.FollowSchedule Reaction.Once timer]))+          descriptor = Inspect.reactiveProcessManagerInspector manager (Stream.categoryUnsafe "opsreaction") (\state _ -> String (Text.pack (show state)))+          hooks = Ops.emptyAppHooks {Ops.processManagers = Just (Map.singleton "modern" (Inspect.SomeProcessManagerInspector descriptor))}+          invoke args = case Options.execParserPure Options.defaultPrefs (Options.info (Ops.commandParser hooks) mempty) args of+            Options.Success command ->+              Ops.runCommand hooks (OpsEnv store Json False [] False) command >>= \case+                Succeeded result -> pure result.jsonValue+                other -> fail (show other)+            _ -> fail "inspection command did not parse"+      Right (Right _) <- run (Reaction.runReactiveProcessManagerOnce defaultRunCommandOptions manager source "one")+      shown <- invoke ["pm", "show", "--name", "modern", "--correlation", "one"]+      field "state" shown `shouldBe` Just (String "Finished")+      field "state_version" shown `shouldBe` Just (Number 1)+      case field "last_reaction" shown of+        Just last -> do+          field "provenance_status" last `shouldBe` Just (String "unavailable")+          field "provenance" last `shouldBe` Just Null+        Nothing -> expectationFailure "accepted saga event is missing"+      case field "pending_timers" shown >>= field "items" of+        Just (Array items) -> length items `shouldBe` 1+        _ -> expectationFailure "reaction timer is missing"+      case field "waiting_for" shown of+        Just (Array items) -> length items `shouldBe` 1+        _ -> expectationFailure "live input candidate is missing"+      listed <- invoke ["pm", "list", "--name", "modern", "--limit", "1"]+      case field "items" listed of+        Just (Array rows) -> map (field "correlation_id") (foldr (:) [] rows) `shouldBe` [Just (String "one")]+        _ -> expectationFailure "reaction journal is missing from the manager catalog"++field :: Text -> Value -> Maybe Value+field key (Object fields) = KeyMap.lookup (Key.fromText key) fields+field _ _ = Nothing++data State = Ready | Finished deriving stock (Eq, Show, Ord, Enum, Bounded)++data Cmd = Go deriving stock (Eq, Show)++data Event = Observed deriving stock (Eq, Show)++definition :: ValidatedEventStream (HsPred '[] Cmd) '[] State Cmd Event+definition =+  mkEventStreamOrThrow+    "ops-reaction-inspection"+    EventStream+      { transducer =+          SymTransducer+            { edgesOut = const [Edge {guard = matchInCtor goCtor, update = UKeep, output = [pack goCtor eventCtor oNil], target = Finished, mode = Live}],+              initial = Ready,+              initialRegs = RNil,+              isFinal = const False+            },+        initialState = Ready,+        initialRegisters = RNil,+        eventCodec =+          Codec+            { eventTypes = EventType "Observed" :| [],+              eventType = const (EventType "Observed"),+              schemaVersion = 1,+              encode = const Null,+              decode = \kind _ -> if kind == EventType "Observed" then Right Observed else Left "unknown fixture event",+              upcasters = []+            },+        resolveStreamName = Stream.streamName,+        snapshotPolicy = Never,+        stateCodec = Nothing+      }++handler :: DomainCommandHandler (HsPred '[] Cmd) '[] State Cmd Event Text Text+handler = DomainCommandHandler definition (const (SilentNoOp "unused"))++goCtor :: InCtor Cmd '[]+goCtor = unavailableInCtor "Go" (const (Just RNil)) (\RNil -> Go)++eventCtor :: WireCtor Event ()+eventCtor = unavailableWireCtor "Observed" (const (Just ())) (const Observed)