keiro 0.21.0.0 → 0.22.0.0

raw patch · 31 files changed

+3994/−368 lines, 31 filesdep +base64-bytestringdep ~keiro-coredep ~keiro-migrationsdep ~keiro-test-supportPVP ok

version bump matches the API change (PVP)

Dependencies added: base64-bytestring

Dependency ranges changed: keiro-core, keiro-migrations, keiro-test-support, kiroku-store, shibuya-core, shibuya-kiroku-adapter

API changes (from Hackage documentation)

Files

• CHANGELOG.md +62/−0
• bench/BenchSetup.hs +72/−0
• bench/CommandProbe.hs +149/−0
• bench/InboxDelegatedBench.hs +69/−38
• bench/Main.hs +337/−153
• bench/ProducerIdentityBench.hs +230/−44
• bench/ReadModelBench.hs +97/−41
• keiro.cabal +28/−19
• retention/Retention/Fixture.hs +3/−2
• src/Keiro/Command.hs +91/−36
• src/Keiro/Inspection/Aggregate.hs +424/−0
• src/Keiro/Inspection/Cursor.hs +61/−0
• src/Keiro/ProcessManager.hs +28/−2
• src/Keiro/ProcessManager/Inspect.hs +441/−0
• src/Keiro/ProcessManager/Provenance.hs +130/−0
• src/Keiro/ReadModel/Rebuild.hs +6/−0
• src/Keiro/ReadModel/Rebuild/Status.hs +76/−0
• src/Keiro/Snapshot/Schema.hs +22/−0
• src/Keiro/Subscription/Shard.hs +58/−13
• src/Keiro/Subscription/Shard/Schema.hs +93/−0
• src/Keiro/Timer.hs +11/−0
• src/Keiro/Timer/Schema.hs +144/−2
• src/Keiro/Workflow/Awakeable/Schema.hs +17/−0
• src/Keiro/Workflow/Child/Schema.hs +21/−0
• src/Keiro/Workflow/Inspection.hs +353/−0
• src/Keiro/Workflow/Inspection/Query.hs +67/−0
• src/Keiro/Workflow/Instance.hs +239/−15
• src/Keiro/Workflow/Schema.hs +22/−0
• test/GroupRebuildSpec.hs +12/−0
• test/Main.hs +372/−3
• test/WorkflowInspectionSpec.hs +259/−0

@@ -6,6 +6,68 @@  ## [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.+- Require Shibuya Core `^>=0.12.0.0` from+  mori://shinzui/shibuya/packages/shibuya-core; direct `Ingested` records supply+  `deliveryStatus` (`pure DeliveryCurrent` preserves prior behavior), and+  `AppConfig` should be constructed from `defaultAppConfig`.++- Direct `RunCommandOptions` records must supply `inspectionProvenance`; updates of defaults remain disabled. Direct `ProcessManagerInspector` records must supply `provenanceMode` (`ReadLegacyProvenance` or `NoReactionProvenance`).++### New Features++- Add pure mounted process-manager discovery with sorted catalog lookup names,+  snapshot configuration, and explicitly unknown timer configuration; discovery+  performs no store reads or application callback evaluation.+  Summary JSON supports equivalent canonical values and direct encodings.++- 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.+++- Explicit, bounded legacy manager provenance through+  `RunCommandOptions.inspectionProvenance` (default `NoInspectionProvenance`).+  Capture records intended source/target evidence under `keiro.pm`, capped at+  64 targets, 64 timer ids and 16 KiB, with an explicit truncation marker.+  Existing updates of defaultRunCommandOptions remain disabled.++- `Keiro.Inspection.Aggregate` exposes folded aggregate state, snapshot health and cursor pages through explicit application descriptors. `hydrateWithSource` reports the existing hydration algorithm’s source; ordinary hydration constructs no diagnostic result. `observeSnapshotRow` adds a database observation clock without changing runtime snapshot lookup.++- Exact, bounded conditional workflow header sessions and reusable single-SELECT point reads.++- Bounded workflow inspection pages, canonical status classes and JSON, pure validation, and opaque cursors.+++- `reactiveProcessManagerInspector` adapts `ReactiveProcessManager.sagaHandler` for state, live input candidates, pending timers and journal listing. Persisted reaction events report unavailable/null provenance without interpreting legacy metadata or probing invented targets. Legacy positional capture remains opt-in and scheduled for retirement with Languages 1–5 before 1.0.++### Bug Fixes++- Fail oversized workflow continuations explicitly instead of returning a cursor+  that the next request cannot decode.+++- Unfiltered workflow pages range over the existing `(workflow_id, workflow_name)` primary key with `wf-key2` cursors. Remove the withdrawn-index wide split, `InspectionWorkLimitExceeded`, `maxIndexedWorkflowKeyBytes`, `maxWideWorkflowCandidates` and `maxWideWorkflowKeyBytes`. Valid wide keys retain byte-capped continuations. The released legacy listing and `wf-created` contract are unchanged; arbitrary filters retain measured deadlines/cost limits.++### Other Changes++- Keep workflow classification and browsing on the read side. No inspection-specific generated column, index, or migration changes ordinary writer work.+ ## 0.21.0.0 — 2026-10-14  ### Other Changes

@@ -0,0 +1,72 @@+-- | Selection-aware fixture diagnostics shared by the benchmark groups.+module BenchSetup (selected, selectedWithCleanup, memoSetup, setupTrace, freshCounter, nextIteration, nextBoundedIteration, selectedCounter, selectedCounterWithCleanup) where++import Control.Concurrent.MVar (modifyMVar, newMVar)+import Control.DeepSeq (NFData (..))+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)+import System.Environment (lookupEnv)+import System.IO (hPutStrLn, stderr)+import Test.Tasty.Bench (Benchmark, envWithCleanup)+import Prelude++-- Fixture handles contain mutable state; force the handle, never its contents.+data Handle a = Handle !a++instance NFData (Handle a) where+  rnf (Handle value) = value `seq` ()++selected :: IO a -> (a -> Benchmark) -> Benchmark+selected acquire = selectedWithCleanup acquire (const (pure ()))++selectedWithCleanup :: IO a -> (a -> IO ()) -> (a -> Benchmark) -> Benchmark+selectedWithCleanup acquire cleanup build =+  envWithCleanup (Handle <$> acquire) (\(Handle value) -> cleanup value) $ \handle ->+    build (case handle of Handle value -> value)++memoSetup :: IO a -> IO (IO a)+memoSetup acquire = do+  cell <- newMVar Nothing+  pure $ modifyMVar cell $ \current -> case current of+    Just value -> pure (current, value)+    Nothing -> do+      value <- acquire+      pure (Just value, value)++setupTrace :: String -> String -> String -> IO ()+setupTrace group scenario detail = do+  enabled <- lookupEnv "KEIRO_BENCH_SETUP_TRACE"+  if enabled == Just "1"+    then hPutStrLn stderr ("BENCH_SETUP group=" <> group <> " scenario=" <> scenario <> " " <> detail)+    else pure ()++freshCounter :: String -> Int -> IO (IORef Int)+freshCounter label units = do+  setupTrace "command" label ("counter budget_iterations=" <> show (iterationBudget units) <> " units_per_iteration=" <> show units)+  newIORef 0++iterationBudget :: Int -> Int+iterationBudget units = 100000 `div` max 1 units++nextIteration :: String -> Int -> IORef Int -> IO Int+nextIteration label units = nextBoundedIteration label (iterationBudget units)++nextBoundedIteration :: String -> Int -> IORef Int -> IO Int+nextBoundedIteration label budget counter = do+  next <- atomicModifyIORef' counter (\current -> let n = current + 1 in (n, n))+  if next > budget+    then fail ("benchmark fixture growth budget exceeded: " <> label <> " iterations=" <> show next)+    else pure next++selectedCounter :: String -> Int -> IO (IORef Int) -> (IORef Int -> Benchmark) -> Benchmark+selectedCounter label units acquire = selectedCounterWithCleanup label units acquire (const (pure ()))++selectedCounterWithCleanup :: String -> Int -> IO (IORef Int) -> (Int -> IO ()) -> (IORef Int -> Benchmark) -> Benchmark+selectedCounterWithCleanup label units acquire cleanup build =+  envWithCleanup+    acquire+    ( \counter -> do+        n <- readIORef counter+        setupTrace "counts" label ("iterations=" <> show n <> " units=" <> show (n * units))+        cleanup n+    )+    build

@@ -0,0 +1,149 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Inspect the actual stream-read prepared plan on a disposable benchmark store.+module CommandProbe (commandPlanProbe, delegatedPlanProbe, tableResetProbe, workCountProbe, rebuildCountProbe, maintenanceProbe) where++import Control.Monad (replicateM)+import Data.Aeson qualified as Aeson+import Data.ByteString.Lazy.Char8 qualified as Lazy+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text+import Data.UUID qualified as UUID+import GHC.Clock (getMonotonicTimeNSec)+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (preparable)+import Keiro.Prelude+import Kiroku.Store qualified as Store+import Kiroku.Store.SQL qualified as SQL+import System.Environment (lookupEnv)+import System.IO (stderr)+import "hasql-transaction" Hasql.Transaction qualified as Tx++commandPlanProbe :: Store.KirokuStore -> Text -> IO ()+commandPlanProbe store label = do+  outcome <- Store.runStoreIO store $ Store.runTransaction $ do+    _ <- Tx.statement ("bench-probe-absent", 0, 100) SQL.readStreamForwardStmt+    entries <-+      Tx.statement () $+        preparable+          "SELECT name FROM pg_prepared_statements WHERE statement LIKE 'SELECT e.event_id,%' AND statement LIKE '%FROM stream_events se%' AND statement LIKE '%s.stream_name = $1%' AND statement LIKE '%se.stream_version > $2%'"+          E.noParams+          (D.rowList (D.column (D.nonNullable D.text)))+    name <- case entries of+      [name] -> pure name+      _ -> error ("expected actual stream read statement: " <> show entries)+    plan <-+      Tx.statement () $+        preparable+          ("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) EXECUTE \"" <> Text.replace "\"" "\"\"" name <> "\"('bench-probe-absent',0,100)")+          E.noParams+          (D.singleRow (D.column (D.nonNullable D.json)))+    counts <-+      Tx.statement () $+        preparable+          "SELECT jsonb_build_object('streams',(SELECT count(*) FROM streams),'events',(SELECT count(*) FROM events),'stream_events',(SELECT count(*) FROM stream_events),'relation_bytes',pg_total_relation_size('stream_events'),'server_version',current_setting('server_version'),'fsync',current_setting('fsync'),'synchronous_commit',current_setting('synchronous_commit'),'full_page_writes',current_setting('full_page_writes'),'autovacuum',current_setting('autovacuum'))"+          E.noParams+          (D.singleRow (D.column (D.nonNullable D.jsonb)))+    pure (Aeson.object ["label" Aeson..= label, "counts" Aeson..= (counts :: Aeson.Value), "plan" Aeson..= (plan :: Aeson.Value)])+  either (fail . show) (Lazy.putStrLn . Aeson.encode) outcome++-- Reset alternatives use the same populated row count on isolated group stores.+tableResetProbe :: Store.KirokuStore -> Text -> Int -> IO () -> IO ()+tableResetProbe store table expected seed = do+  for_ ["TRUNCATE", "DELETE FROM"] $ \verb -> do+    samples <- replicateM 5 $ do+      seed+      before <- info+      unless (fst before == fromIntegral expected) (fail "reset probe seed count differs")+      started <- getMonotonicTimeNSec+      checked (Tx.sql (Text.encodeUtf8 (verb <> " " <> table)))+      ended <- getMonotonicTimeNSec+      after <- info+      unless (fst after == 0) (fail "reset probe retained rows")+      pure (Aeson.object ["elapsed_ns" Aeson..= (ended - started), "rows_before" Aeson..= fst before, "bytes_before" Aeson..= snd before, "rows_after" Aeson..= fst after, "bytes_after" Aeson..= snd after])+    Lazy.putStrLn (Aeson.encode (Aeson.object ["table" Aeson..= table, "reset" Aeson..= verb, "samples" Aeson..= samples]))+  where+    checked :: Tx.Transaction a -> IO a+    checked action = Store.runStoreIO store (Store.runTransaction action) >>= either (fail . show) pure+    info =+      checked+        ( Tx.statement+            ()+            ( preparable+                ("SELECT count(*), pg_total_relation_size('" <> table <> "') FROM " <> table)+                E.noParams+                (D.singleRow ((,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8)))+            )+        )++delegatedPlanProbe :: Store.KirokuStore -> Text -> IO ()+delegatedPlanProbe store label = do+  result <- Store.runStoreIO store $ Store.runTransaction $ do+    _ <- Tx.statement ("bench-probe-absent", UUID.nil) SQL.eventExistsInStreamStmt+    names <-+      Tx.statement () $+        preparable+          "SELECT name FROM pg_prepared_statements WHERE statement LIKE 'SELECT EXISTS (%' AND statement LIKE '%FROM stream_events se%' AND statement LIKE '%stream_name = $1%'"+          E.noParams+          (D.rowList (D.column (D.nonNullable D.text)))+    name <- case names of [name] -> pure name; _ -> error ("expected delegated prepared statement: " <> show names)+    plan <-+      Tx.statement () $+        preparable+          ("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) EXECUTE \"" <> Text.replace "\"" "\"\"" name <> "\"('bench-probe-absent','00000000-0000-0000-0000-000000000000'::uuid)")+          E.noParams+          (D.singleRow (D.column (D.nonNullable D.json)))+    pure (Aeson.object ["label" Aeson..= label, "plan" Aeson..= (plan :: Aeson.Value)])+  either (fail . show) (Lazy.hPutStrLn stderr . Aeson.encode) result++-- Run only in diagnostic invocations: these reads must not change timed plans.+workCountProbe :: Store.KirokuStore -> Text -> Text -> Int -> Int -> IO ()+workCountProbe store label prefix expectedStreams expectedEvents = do+  enabled <- lookupEnv "KEIRO_BENCH_WORK_COUNTS"+  when (enabled == Just "1") $ do+    result <-+      Store.runStoreIO store $+        Store.runTransaction $+          Tx.statement prefix $+            preparable+              "SELECT count(DISTINCT s.stream_id), count(se.event_id), pg_total_relation_size('streams'), pg_total_relation_size('events'), pg_total_relation_size('stream_events') FROM streams s LEFT JOIN stream_events se USING (stream_id) WHERE left(s.stream_name, length($1)) = $1"+              (E.param (E.nonNullable E.text))+              (D.singleRow ((,,,,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8)))+    (streams, events, streamBytes, eventBytes, linkBytes) <- either (fail . show) pure result+    unless (streams == fromIntegral expectedStreams && events == fromIntegral expectedEvents) $+      fail ("benchmark work-count mismatch: " <> Text.unpack label <> " expected=" <> show (expectedStreams, expectedEvents) <> " actual=" <> show (streams, events))+    Lazy.hPutStrLn stderr (Aeson.encode (Aeson.object ["label" Aeson..= label, "prefix" Aeson..= prefix, "streams" Aeson..= streams, "events" Aeson..= events, "streams_bytes" Aeson..= streamBytes, "events_bytes" Aeson..= eventBytes, "stream_events_bytes" Aeson..= linkBytes]))++rebuildCountProbe :: Store.KirokuStore -> Int -> IO ()+rebuildCountProbe store expectedRuns = do+  enabled <- lookupEnv "KEIRO_BENCH_WORK_COUNTS"+  when (enabled == Just "1") $ do+    result <-+      Store.runStoreIO store $+        Store.runTransaction $+          Tx.statement () $+            preparable+              "SELECT (SELECT count(*) FROM keiro.keiro_projection_rebuild_runs), (SELECT count(*) FROM keiro.keiro_projection_rebuild_sources), (SELECT count(*) FROM events), jsonb_build_object('runs_bytes',pg_total_relation_size('keiro.keiro_projection_rebuild_runs'),'sources_bytes',pg_total_relation_size('keiro.keiro_projection_rebuild_sources'),'targets',(SELECT count(*) FROM keiro.keiro_projection_rebuild_run_targets),'adapters',(SELECT count(*) FROM keiro.keiro_projection_rebuild_adapters),'verifications',(SELECT count(*) FROM keiro.keiro_projection_rebuild_verifications))"+              E.noParams+              (D.singleRow ((,,,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.jsonb)))+    (runs, sources, events, bookkeeping) <- either (fail . show) pure result+    unless (runs == fromIntegral expectedRuns && sources == fromIntegral (3 * expectedRuns) && events == 600) $+      fail ("rebuild work-count mismatch: " <> show (runs, sources, events))+    Lazy.hPutStrLn stderr (Aeson.encode (Aeson.object ["label" Aeson..= ("rebuild" :: Text), "runs" Aeson..= runs, "sources" Aeson..= sources, "events" Aeson..= events, "bookkeeping" Aeson..= (bookkeeping :: Aeson.Value)]))++-- Cluster maintenance is shared across database clones. Record it outside timing.+maintenanceProbe :: Store.KirokuStore -> Text -> IO ()+maintenanceProbe store label = do+  enabled <- lookupEnv "KEIRO_BENCH_MAINTENANCE_PROBE"+  when (enabled == Just "1") $ do+    result <-+      Store.runStoreIO store $+        Store.runTransaction $+          Tx.statement () $+            preparable+              "SELECT jsonb_build_object('at',clock_timestamp(),'checkpoint',(SELECT to_jsonb(c) FROM pg_stat_checkpointer c),'wal',(SELECT to_jsonb(w) FROM pg_stat_wal w),'lsn',pg_current_wal_lsn(),'max_wal_size',current_setting('max_wal_size'),'checkpoint_timeout',current_setting('checkpoint_timeout'),'shared_buffers',current_setting('shared_buffers'),'activity',(SELECT jsonb_agg(jsonb_build_object('type',backend_type,'database',datname,'state',state,'wait_type',wait_event_type,'wait',wait_event)) FROM pg_stat_activity WHERE backend_type IN ('checkpointer','autovacuum worker','background writer','walwriter','io worker')),'tables',(SELECT jsonb_agg(jsonb_build_object('table',relname,'live',n_live_tup,'dead',n_dead_tup,'vacuum',vacuum_count,'autovacuum',autovacuum_count,'last_autovacuum',last_autovacuum,'analyze',analyze_count,'last_autoanalyze',last_autoanalyze)) FROM pg_stat_user_tables WHERE relname IN ('streams','stream_events','events','keiro_inbox','keiro_inbox_delegated_bench_effect')))"+              E.noParams+              (D.singleRow (D.column (D.nonNullable D.jsonb)))+    snapshot <- either (fail . show) pure result+    Lazy.hPutStrLn stderr (Aeson.encode (Aeson.object ["maintenance" Aeson..= label, "snapshot" Aeson..= (snapshot :: Aeson.Value)]))

@@ -2,16 +2,19 @@ {-# LANGUAGE OverloadedRecordDot #-}  module InboxDelegatedBench-  ( prepareInboxDelegatedBenchmarks,+  ( inboxDelegatedBenchmarks,     runInboxDelegatedExplainIfRequested,   ) where +import BenchSetup (nextIteration, selectedWithCleanup, setupTrace)+import CommandProbe (delegatedPlanProbe, maintenanceProbe, workCountProbe) import Control.DeepSeq (NFData (..))+import Control.Monad (replicateM_) import Data.Aeson qualified as Aeson import Data.Bifunctor (first) import Data.ByteString qualified as ByteString-import Data.IORef (IORef, atomicModifyIORef', newIORef)+import Data.IORef (IORef, newIORef, readIORef) import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map import Data.Text qualified as Text@@ -42,6 +45,7 @@ import Keiro.Prelude import Keiro.Stream (Stream, stream) import Keiro.Telemetry (KeiroMetrics)+import Keiro.Test.Postgres (checkpointBenchmarkStore) import Kiroku.Store qualified as Store import Kiroku.Store.Effect (PreparedEvent, Store) import Kiroku.Store.SQL qualified as StoreSQL@@ -57,7 +61,7 @@ import System.Directory (createDirectoryIfMissing) import System.Environment (lookupEnv) import System.FilePath (takeDirectory)-import Test.Tasty.Bench (Benchmark, bench, bgroup, env, nfIO)+import Test.Tasty.Bench (Benchmark, bench, bgroup, nfIO) import "hasql-transaction" Hasql.Transaction qualified as Tx import Prelude @@ -109,22 +113,21 @@       Map.size downstreamRun.workByMessageId `seq`         () -prepareInboxDelegatedBenchmarks :: Store.KirokuStore -> KeiroMetrics -> IO [Benchmark]-prepareInboxDelegatedBenchmarks store metrics = do-  runStoreChecked store $ Store.runTransaction (Tx.sql businessTableSql)-  scenarios <- concat <$> traverse (prepareScenario store metrics) scenarioInputs-  traverse_ (runScenario store) [scenario | scenario <- scenarios, scenario.traffic == FreshTraffic]-  pure-    ( [ bgroup-          (Text.unpack scenario.name)-          [ bgroup-              (trafficName scenario.traffic)-              [bench (metricsName scenario.metrics) (nfIO (runScenario store scenario))]+inboxDelegatedBenchmarks :: Store.KirokuStore -> KeiroMetrics -> [Benchmark]+inboxDelegatedBenchmarks store metrics =+  [ bgroup+      (Text.unpack name)+      [ bgroup+          (trafficName traffic)+          [ selectedWithCleanup (prepareScenario store name mode chunkSize traffic mMetrics) (checkScenarioCounts store) $ \scenario ->+              bench (metricsName mMetrics) (nfIO (scenario `seq` runScenario store scenario))+          | mMetrics <- [Nothing, Just metrics]           ]-      | scenario <- scenarios+      | traffic <- [FreshTraffic, RepeatedTraffic, DuplicateTraffic]       ]-        <> [bgroup "delegated-history-duplicate" (historyBenchmark store <$> [10, 1000, 100000])]-    )+  | (name, mode, chunkSize) <- scenarioInputs+  ]+    <> [bgroup "delegated-history-duplicate" (historyBenchmark store <$> [10, 1000, 100000])]  runInboxDelegatedExplainIfRequested :: Store.KirokuStore -> IO () runInboxDelegatedExplainIfRequested store =@@ -179,30 +182,56 @@     ("delegated-batch-1000", DelegatedMode, 1000)   ] -prepareScenario :: Store.KirokuStore -> KeiroMetrics -> (Text, IntakeMode, Int) -> IO [Scenario]-prepareScenario store metrics (name, mode, chunkSize) =-  traverse make [(traffic, mMetrics) | traffic <- [FreshTraffic, RepeatedTraffic, DuplicateTraffic], mMetrics <- [Nothing, Just metrics]]-  where-    make (traffic, mMetrics) = do-      runNumber <- newIORef 0-      duplicateRun <--        case traffic of-          DuplicateTraffic -> do-            prepared <- prepareDownstreamRun (scenarioPrefix name traffic mMetrics <> "-seed") RepeatedTraffic-            seedDuplicateRun store mode prepared-            pure (Just prepared)-          _ -> pure Nothing-      pure Scenario {name, mode, chunkSize, traffic, metrics = mMetrics, runNumber, duplicateRun}+prepareScenario :: Store.KirokuStore -> Text -> IntakeMode -> Int -> Traffic -> Maybe KeiroMetrics -> IO Scenario+prepareScenario store name mode chunkSize traffic metrics = do+  let label = Text.unpack (scenarioPrefix name traffic metrics)+  setupTrace "delegated-inbox" label "schema"+  runStoreChecked store $ Store.runTransaction (Tx.sql businessTableSql)+  runNumber <- newIORef 0+  duplicateRun <- case traffic of+    DuplicateTraffic -> do+      setupTrace "delegated-inbox" label "duplicate-seed deliveries=2000"+      prepared <- prepareDownstreamRun (scenarioPrefix name traffic metrics <> "-seed") RepeatedTraffic+      seedDuplicateRun store mode prepared+      pure (Just prepared)+    _ -> pure Nothing+  let scenario = Scenario {name, mode, chunkSize, traffic, metrics, runNumber, duplicateRun}+  setupTrace "delegated-inbox" label "warm-up invocations=3"+  runScenario store scenario+  probe <- lookupEnv "KEIRO_BENCH_INBOX_PLAN_PROBE"+  when (probe == Just "1") (delegatedPlanProbe store (Text.pack label <> "-before-analyze"))+  setupTrace "delegated-inbox" label "analyze=after-first-warm-up remaining_warm_up=2"+  runStoreChecked store $ Store.runTransaction (Tx.sql "ANALYZE streams; ANALYZE stream_events; ANALYZE events; ANALYZE keiro.keiro_inbox")+  when (probe == Just "1") (delegatedPlanProbe store (Text.pack label <> "-after-analyze"))+  setupTrace "delegated-inbox" label "checkpoint=before-final-two-warm-ups"+  checkpointBenchmarkStore store+  replicateM_ 2 (runScenario store scenario)+  maintenanceProbe store (Text.pack label <> "-ready")+  pure scenario  runScenario :: Store.KirokuStore -> Scenario -> IO () runScenario store scenario = do   downstreamRun <- case scenario.duplicateRun of-    Just prepared -> pure prepared+    Just prepared -> do+      _ <- nextIteration (Text.unpack scenario.name) 0 scenario.runNumber+      pure prepared     Nothing -> do-      invocation <- atomicModifyIORef' scenario.runNumber (\current -> let next = current + 1 in (next, next))+      invocation <- nextIteration (Text.unpack scenario.name) (if scenario.traffic == FreshTraffic then deliveryCount else 1) scenario.runNumber       prepareDownstreamRun (scenarioPrefix scenario.name scenario.traffic scenario.metrics <> "-" <> Text.pack (show invocation)) scenario.traffic   runStoreChecked store (runDownstream scenario downstreamRun) +checkScenarioCounts :: Store.KirokuStore -> Scenario -> IO ()+checkScenarioCounts store scenario = do+  n <- readIORef scenario.runNumber+  let prefix = scenarioPrefix scenario.name scenario.traffic scenario.metrics+      expected = case scenario.traffic of+        FreshTraffic -> n * deliveryCount+        RepeatedTraffic -> n+        DuplicateTraffic -> 1+  setupTrace "counts" (Text.unpack prefix) ("iterations=" <> show n <> " expected_events=" <> show expected)+  workCountProbe store prefix ("benchInbox-" <> prefix <> "-") expected expected+  maintenanceProbe store (prefix <> "-finished")+ runDownstream :: (IOE :> es, Store :> es) => Scenario -> DownstreamRun -> Eff es () runDownstream scenario downstreamRun =   case scenario.mode of@@ -358,12 +387,12 @@ prepareDownstreamRun :: Text -> Traffic -> IO DownstreamRun prepareDownstreamRun prefix traffic = do   uniqueWorks <- traverse (prepareWork prefix) uniqueIndexes-  let selected = case (traffic, uniqueWorks) of+  let selectedWorks = case (traffic, uniqueWorks) of         (FreshTraffic, _) -> uniqueWorks         (RepeatedTraffic, work : _) -> Prelude.replicate deliveryCount work         (DuplicateTraffic, work : _) -> Prelude.replicate deliveryCount work         _ -> error "prepareDownstreamRun: traffic generated no work"-      deliveries = [(work.event, Just work.kafka) | work <- selected]+      deliveries = [(work.event, Just work.kafka) | work <- selectedWorks]       workByMessageId = Map.fromList [(work.event.messageId, work) | work <- uniqueWorks]   pure DownstreamRun {deliveries, workByMessageId}   where@@ -398,14 +427,16 @@  historyBenchmark :: Store.KirokuStore -> Int -> Benchmark historyBenchmark store historySize =-  env (prepareHistoryRun store historySize) $ \downstreamRun ->+  selectedWithCleanup (prepareHistoryRun store historySize) (const (workCountProbe store ("history-" <> Text.pack (show historySize)) ("benchInbox-history-" <> Text.pack (show historySize) <> "-") 1 historySize)) $ \downstreamRun ->     bench ("events-" <> show historySize) $       nfIO $-        runStoreChecked store $-          traverse_ (runDelegatedSingle Nothing downstreamRun.workByMessageId) downstreamRun.deliveries+        downstreamRun `seq`+          runStoreChecked store (traverse_ (runDelegatedSingle Nothing downstreamRun.workByMessageId) downstreamRun.deliveries)  prepareHistoryRun :: Store.KirokuStore -> Int -> IO DownstreamRun prepareHistoryRun store historySize = do+  runStoreChecked store $ Store.runTransaction (Tx.sql businessTableSql)+  setupTrace "delegated-inbox" ("events-" <> show historySize) ("history source_events=" <> show historySize)   downstreamRun <- prepareDownstreamRun ("history-" <> Text.pack (show historySize)) DuplicateTraffic   seedDuplicateRun store DelegatedMode downstreamRun   case Map.elems downstreamRun.workByMessageId of

@@ -6,19 +6,24 @@   ) where +import BenchSetup (memoSetup, nextBoundedIteration, nextIteration, selected, selectedCounterWithCleanup, selectedWithCleanup, setupTrace)+import CommandProbe (commandPlanProbe, maintenanceProbe, rebuildCountProbe, tableResetProbe, workCountProbe) import Control.Concurrent (threadDelay)+import Control.Monad (replicateM_) import Data.Aeson qualified as Aeson import Data.ByteString qualified as BS-import Data.IORef (IORef, atomicModifyIORef', newIORef)+import Data.IORef (IORef, newIORef) import Data.Int (Int32) import Data.Text qualified as Text import Data.Time (UTCTime (..), secondsToDiffTime) import Data.Time.Calendar (Day (ModifiedJulianDay)) import Data.UUID qualified as UUID+import Data.Vector qualified as Vector import Effectful (Eff, IOE, (:>)) import Effectful.Error.Static (Error)+import GHC.Clock (getMonotonicTimeNSec) import GuardedCatalogBench (runGuardedCatalogEvidence)-import InboxDelegatedBench (prepareInboxDelegatedBenchmarks, runInboxDelegatedExplainIfRequested)+import InboxDelegatedBench (inboxDelegatedBenchmarks, runInboxDelegatedExplainIfRequested) import Keiki.Core   ( Edge (..),     HsPred,@@ -75,11 +80,10 @@     startCatalogRebuild,   ) import Keiro.Telemetry qualified as Telemetry-import Keiro.Test.Postgres (StoreRunner (..), withFreshResourceStore, withMigratedSuite)+import Keiro.Test.Postgres (StoreRunner (..), checkpointBenchmarkStore, withFreshResourceStore, withMigratedBenchmarkSuite) import Kiroku.Store qualified as Store import Kiroku.Store.Effect (Store) import Kiroku.Store.Effect.Resource (KirokuStoreResource)-import Kiroku.Store.Lifecycle qualified as Lifecycle import Kiroku.Store.Types   ( CategoryName (..),     EventData (..),@@ -94,18 +98,21 @@ import OpenTelemetry.MeterProvider (createMeterProvider, defaultSdkMeterProviderOptions) import OpenTelemetry.Metric.Core (getMeter) import OpenTelemetry.Resource (emptyMaterializedResources)-import ProducerIdentityBench (producerIdentityBenchmarks)+import ProducerIdentityBench (producerIdentityBenchmarks, runProducerIdentityProbe) import ReadModelBench-  ( readModelBenchmarks,+  ( probeReadModelCounts,+    readModelBenchmarks,     runReadModelExplainEvidenceIfRequested,     runReadModelLatencyEvidenceIfRequested,     setupReadModelBench,   ) import Shibuya.Adapter (Adapter (..)) import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..)) import Shibuya.Core.Types (Envelope (..)) import Streamly.Data.Stream qualified as Streamly+import System.Environment (lookupEnv)+import System.IO (hFlush, stdout) import Test.Tasty.Bench (Benchmark, bcompareWithin, bench, bgroup, defaultMain, nfIO) import "hasql-transaction" Hasql.Transaction qualified as Tx import Prelude@@ -152,88 +159,108 @@  ordinaryMain :: IO () ordinaryMain =-  withMigratedSuite \fixture ->-    withFreshResourceStore fixture \(store, runner) -> do-      withFreshResourceStore fixture \(readModelStore, readModelRunner) -> do-        (provider, _env) <--          createMeterProvider-            emptyMaterializedResources-            defaultSdkMeterProviderOptions-        meter <- getMeter provider Telemetry.keiroInstrumentationLibrary-        metrics <- Telemetry.newKeiroMetrics meter-        rebuildRunCounter <- newIORef 0-        runStoreChecked store (Store.runTransaction (Tx.sql rebuildBenchSql))-        readModelFixture <- setupReadModelBench readModelStore readModelRunner-        runReadModelExplainEvidenceIfRequested readModelFixture-        runReadModelLatencyEvidenceIfRequested readModelFixture-        delegatedInboxBenchmarks <- prepareInboxDelegatedBenchmarks store metrics-        runInboxDelegatedExplainIfRequested store-        defaultMain (benchmarks store runner metrics rebuildRunCounter delegatedInboxBenchmarks <> readModelBenchmarks readModelFixture <> producerIdentityBenchmarks store rebuildRunCounter)+  withMigratedBenchmarkSuite \fixture ->+    withFreshResourceStore fixture \(store, runner) ->+      withFreshResourceStore fixture \(outboxStore, _) ->+        withFreshResourceStore fixture \(inboxStore, _) ->+          withFreshResourceStore fixture \(delegatedStore, _) ->+            withFreshResourceStore fixture \(producerStore, _) ->+              withFreshResourceStore fixture \(rebuildStore, _) ->+                withFreshResourceStore fixture \(readModelStore, readModelRunner) -> do+                  (provider, _env) <- createMeterProvider emptyMaterializedResources defaultSdkMeterProviderOptions+                  meter <- getMeter provider Telemetry.keiroInstrumentationLibrary+                  metrics <- Telemetry.newKeiroMetrics meter+                  producerCounter <- newIORef 0+                  readModelSetup <- memoSetup do+                    setupTrace "read-model" "shared" "schema history=25000"+                    model <- setupReadModelBench readModelStore readModelRunner+                    runReadModelExplainEvidenceIfRequested model+                    runReadModelLatencyEvidenceIfRequested model+                    pure model+                  runInboxDelegatedExplainIfRequested delegatedStore+                  producerProbe <- lookupEnv "KEIRO_BENCH_PRODUCER_PROBE"+                  resetProbe <- lookupEnv "KEIRO_BENCH_RESET_PROBE"+                  probe <- lookupEnv "KEIRO_BENCH_COMMAND_PROBE"+                  if producerProbe == Just "1"+                    then runProducerIdentityProbe producerStore+                    else+                      if probe == Just "1"+                        then runCommandGrowthProbe store+                        else+                          if resetProbe == Just "1"+                            then runResetProbes outboxStore inboxStore producerStore+                            else+                              defaultMain+                                ( benchmarks store runner outboxStore inboxStore rebuildStore metrics (inboxDelegatedBenchmarks delegatedStore metrics)+                                    <> [selectedWithCleanup readModelSetup probeReadModelCounts (\model -> readModelBenchmarks model !! groupIndex) | groupIndex <- [0 .. 2]]+                                    <> producerIdentityBenchmarks producerStore producerCounter+                                ) -benchmarks :: Store.KirokuStore -> StoreRunner -> Telemetry.KeiroMetrics -> IORef Int -> [Benchmark] -> [Benchmark]-benchmarks store runner metrics rebuildRunCounter delegatedInboxBenchmarks =+benchmarks :: Store.KirokuStore -> StoreRunner -> Store.KirokuStore -> Store.KirokuStore -> Store.KirokuStore -> Telemetry.KeiroMetrics -> [Benchmark] -> [Benchmark]+benchmarks store runner outboxStore inboxStore rebuildStore metrics delegatedInboxBenchmarks =   [ bgroup       "outbox"-      [ scenarioBench store hotKey,-        scenarioBench store hotKeyNoLatency,-        scenarioBench store multiKey+      [ scenarioBench outboxStore hotKey,+        scenarioBench outboxStore hotKeyNoLatency,+        scenarioBench outboxStore multiKey       ],     bgroup       "inbox"-      [ inboxScenarioBench store (singleFull metrics),-        inboxScenarioBench store singleNoMetrics,-        inboxScenarioBench store batch100,-        inboxScenarioBench store singleSlim,+      [ inboxScenarioBench inboxStore (singleFull metrics),+        inboxScenarioBench inboxStore singleNoMetrics,+        inboxScenarioBench inboxStore batch100,+        inboxScenarioBench inboxStore singleSlim,         bgroup "downstream" delegatedInboxBenchmarks       ],-    bgroup-      "command"-      [ bgroup-          "legacy"-          [commandScenarioBench store "accepted-1" legacyAcceptedOneStream legacyAcceptedOneTarget EmitOne],-        bgroup-          "control"-          [commandScenarioBench store "accepted-1" legacyAcceptedOneStream legacyAcceptedOneTarget EmitOne],-        bgroup-          "domain-warmup"-          [domainCommandScenarioBench store "accepted-1" domainAcceptedOneHandler domainAcceptedOneTarget EmitOne],-        bgroup-          "domain"-          [ bcompareWithin 0 1.25 acceptedOneControlPattern $-              domainCommandScenarioBench store "accepted-1" domainAcceptedOneHandler domainAcceptedOneTarget EmitOne-          ],-        bgroup-          "legacy"-          [commandScenarioBench store "accepted-large" legacyAcceptedLargeStream legacyAcceptedLargeTarget EmitLarge],-        bgroup-          "domain"-          [ bcompareWithin 0 1.25 legacyAcceptedLargePattern $-              domainCommandScenarioBench store "accepted-large" domainAcceptedLargeHandler domainAcceptedLargeTarget EmitLarge-          ],-        bgroup-          "legacy"-          [commandScenarioBench store "no-op" legacyNoOpStream legacyNoOpTarget SelectNoOp],-        bgroup-          "domain"-          [ bcompareWithin 0 1.25 legacyNoOpPattern $-              domainCommandScenarioBench store "rejected" domainRejectedHandler domainRejectedTarget SelectNoOp,-            bcompareWithin 0 1.25 legacyNoOpPattern $-              domainCommandScenarioBench store "no-op" domainNoOpHandler domainNoOpTarget SelectNoOp,-            bgroup-              "router-fanout"-              [ domainRouterFanoutBench runner fanout-              | fanout <- coordinatorFanouts-              ],-            bgroup-              "process-manager-fanout"-              [ domainProcessManagerFanoutBench runner fanout-              | fanout <- coordinatorFanouts-              ]-          ]-      ],+    selected (prepareCommandStore store) $ \_ ->+      bgroup+        "command"+        [ bgroup+            "legacy"+            [commandScenarioBench store "accepted-1" legacyAcceptedOneStream legacyAcceptedOneTarget EmitOne],+          bgroup+            "control"+            [commandScenarioBench store "accepted-1" legacyAcceptedOneStream (stream "bench-command-control") EmitOne],+          bgroup+            "domain-warmup"+            [domainCommandScenarioBench store "accepted-1" domainAcceptedOneHandler (stream "bench-command-domain-warmup") EmitOne],+          bgroup+            "domain"+            [ bcompareWithin 0 1.25 acceptedOneControlPattern $+                domainCommandScenarioBench store "accepted-1" domainAcceptedOneHandler domainAcceptedOneTarget EmitOne+            ],+          bgroup+            "legacy"+            [commandScenarioBench store "accepted-large" legacyAcceptedLargeStream legacyAcceptedLargeTarget EmitLarge],+          bgroup+            "domain"+            [ bcompareWithin 0 1.25 legacyAcceptedLargePattern $+                domainCommandScenarioBench store "accepted-large" domainAcceptedLargeHandler domainAcceptedLargeTarget EmitLarge+            ],+          bgroup+            "legacy"+            [commandScenarioBench store "no-op" legacyNoOpStream legacyNoOpTarget SelectNoOp],+          bgroup+            "domain"+            [ bcompareWithin 0 1.25 legacyNoOpPattern $+                domainCommandScenarioBench store "rejected" domainRejectedHandler domainRejectedTarget SelectNoOp,+              bcompareWithin 0 1.25 legacyNoOpPattern $+                domainCommandScenarioBench store "no-op" domainNoOpHandler domainNoOpTarget SelectNoOp,+              bgroup+                "router-fanout"+                [ domainRouterFanoutBench store runner fanout+                | fanout <- coordinatorFanouts+                ],+              bgroup+                "process-manager-fanout"+                [ domainProcessManagerFanoutBench store runner fanout+                | fanout <- coordinatorFanouts+                ]+            ]+        ],     bgroup       "rebuild"-      [rebuildScenarioBench store rebuildRunCounter]+      [rebuildScenarioBench rebuildStore]   ]   where     hotKey =@@ -289,9 +316,16 @@  scenarioBench :: Store.KirokuStore -> OutboxScenario -> Benchmark scenarioBench store scenario =-  bench (Text.unpack scenario.scenarioName) $-    nfIO (runScenario store scenario)+  selected (warmAction store "outbox" (Text.unpack scenario.scenarioName) (runScenario store scenario)) $ \_ ->+    bench (Text.unpack scenario.scenarioName) $ nfIO (runScenario store scenario) +warmAction :: Store.KirokuStore -> String -> String -> IO () -> IO ()+warmAction store group name action = do+  setupTrace group name "checkpoint=before-warm-up"+  checkpointBenchmarkStore store+  setupTrace group name "warm-up invocations=3 timed_reset=TRUNCATE"+  replicateM_ 3 action+ runScenario :: Store.KirokuStore -> OutboxScenario -> IO () runScenario store scenario = do   runStoreChecked store do@@ -301,8 +335,8 @@  inboxScenarioBench :: Store.KirokuStore -> InboxScenario -> Benchmark inboxScenarioBench store scenario =-  bench (Text.unpack scenario.inboxScenarioName) $-    nfIO (runInboxScenario store scenario)+  selected (warmAction store "inbox" (Text.unpack scenario.inboxScenarioName) (runInboxScenario store scenario)) $ \_ ->+    bench (Text.unpack scenario.inboxScenarioName) $ nfIO (runInboxScenario store scenario)  runInboxScenario :: Store.KirokuStore -> InboxScenario -> IO () runInboxScenario store scenario = do@@ -614,24 +648,94 @@       upcasters = []     } +-- Seed a representative stream population before any prepared lookup is planned.+-- Empty-table statistics otherwise keep sequential scans as fresh samples grow.+prepareCommandStore :: Store.KirokuStore -> IO ()+prepareCommandStore store = do+  setupTrace "command" "shared" "schema seed_streams=5000 seed_events=5000 analyze=once-before-warm-up"+  for_ [1 .. 5000 :: Int] $ \n -> runStoreChecked store $ do+    void+      ( Store.appendToStream+          (StreamName ("benchseed-" <> Text.pack (show n)))+          NoStream+          [EventData Nothing (EventType "BenchSeed") Aeson.Null Nothing Nothing Nothing]+      )+  runStoreChecked store (Store.runTransaction (Tx.sql "ANALYZE streams; ANALYZE stream_events; ANALYZE events"))+ commandScenarioBench :: Store.KirokuStore -> String -> ValidatedBenchEventStream -> Stream BenchEventStream -> BenchCommand -> Benchmark commandScenarioBench store benchmarkName validatedStream target command =-  bench benchmarkName $ nfIO $ runStoreChecked store do-    void (Lifecycle.hardDeleteStream (streamName target))-    result <- runCommand defaultRunCommandOptions validatedStream target command-    case result of-      Right _ -> pure ()-      Left err -> liftIO (fail ("unexpected command benchmark result: " <> show err))+  selectedCounterWithCleanup label units (prepareCommand store label units (commandBudget command) action) checkCounts $ \counter ->+    bench benchmarkName $ nfIO (action counter)+  where+    label = show (streamName target) <> "/legacy/" <> benchmarkName+    checkCounts n = do+      maintenanceProbe store (Text.pack label <> "-finished")+      let StreamName prefix = streamName (tagTarget target "legacy")+      workCountProbe store (Text.pack label) (prefix <> "-iteration-") (if units == 0 then 0 else n) (n * units)+    units = commandUnits command+    action counter = do+      n <- nextBoundedIteration label (commandBudget command) counter+      let fresh = iterationTarget (tagTarget target "legacy") n+      runStoreChecked store do+        result <- runCommand defaultRunCommandOptions validatedStream fresh command+        case result of+          Right value | value ^. #eventsAppended == eventCount command -> pure ()+          other -> liftIO (fail ("unexpected command benchmark result: " <> show other))  domainCommandScenarioBench :: Store.KirokuStore -> String -> DomainCommandHandler (HsPred '[] BenchCommand) '[] BenchState BenchCommand BenchEvent Text Text -> Stream BenchEventStream -> BenchCommand -> Benchmark domainCommandScenarioBench store benchmarkName handler target command =-  bench benchmarkName $ nfIO $ runStoreChecked store do-    void (Lifecycle.hardDeleteStream (streamName target))-    result <- runDomainCommand defaultRunCommandOptions handler target command-    case result of-      Right DomainCommandOutcome {} -> pure ()-      Left err -> liftIO (fail ("unexpected typed command benchmark result: " <> show err))+  selectedCounterWithCleanup label units (prepareCommand store label units (commandBudget command) action) checkCounts $ \counter ->+    bench benchmarkName $ nfIO (action counter)+  where+    label = show (streamName target) <> "/domain/" <> benchmarkName+    checkCounts n = do+      maintenanceProbe store (Text.pack label <> "-finished")+      let StreamName prefix = streamName (tagTarget target ("domain-" <> Text.pack benchmarkName))+      workCountProbe store (Text.pack label) (prefix <> "-iteration-") (if units == 0 then 0 else n) (n * units)+    units = commandUnits command+    action counter = do+      n <- nextBoundedIteration label (commandBudget command) counter+      let fresh = iterationTarget (tagTarget target ("domain-" <> Text.pack benchmarkName)) n+      runStoreChecked store do+        result <- runDomainCommand defaultRunCommandOptions handler fresh command+        case result of+          Right value | value ^. #result . #eventsAppended == eventCount command -> pure ()+          other -> liftIO (fail ("unexpected typed command benchmark result: " <> show other)) +commandUnits :: BenchCommand -> Int+commandUnits = eventCount++commandBudget :: BenchCommand -> Int+commandBudget EmitOne = 100000+commandBudget EmitLarge = 10000+commandBudget SelectNoOp = 1000000++eventCount :: BenchCommand -> Int+eventCount EmitOne = 1+eventCount EmitLarge = largeCommandBatchSize+eventCount SelectNoOp = 0++tagTarget :: Stream eventStream -> Text -> Stream eventStream+tagTarget target tag =+  let StreamName name = streamName target+   in stream (name <> "-" <> tag)++iterationTarget :: Stream eventStream -> Int -> Stream eventStream+iterationTarget target n =+  let StreamName name = streamName target+   in stream (name <> "-iteration-" <> Text.pack (show n))++prepareCommand :: Store.KirokuStore -> String -> Int -> Int -> (IORef Int -> IO ()) -> IO (IORef Int)+prepareCommand store label units budget action = do+  setupTrace "command" label ("counter budget_iterations=" <> show budget <> " events_per_iteration=" <> show units)+  counter <- newIORef 0+  setupTrace "command" label "checkpoint=before-warm-up"+  checkpointBenchmarkStore store+  setupTrace "command" label "warm-up invocations=3"+  replicateM_ 3 (action counter)+  maintenanceProbe store (Text.pack label <> "-ready")+  pure counter+ -- * Projection rebuild benchmark fixtures ---------------------------------  rebuildEventsPerCategory :: Int@@ -640,32 +744,44 @@ rebuildPageSize :: Int32 rebuildPageSize = 16 -rebuildScenarioBench :: Store.KirokuStore -> IORef Int -> Benchmark-rebuildScenarioBench store runCounter =-  bench "three-categories-200" $ nfIO $ runStoreChecked store do-    traverse_ (void . Lifecycle.hardDeleteStream) rebuildStreams-    traverse_ seedRebuildStream rebuildStreams-    registered <- registerProjectionCatalog rebuildCatalog-    case registered of-      Left err -> liftIO (fail ("unexpected rebuild catalog registration result: " <> show err))-      Right _ -> pure ()-    runNumber <- liftIO $ atomicModifyIORef' runCounter (\current -> let next = current + 1 in (next, next))-    let request =-          RebuildRequest-            { rebuildRunId = benchmarkRebuildRunId runNumber,-              requestedBy = "keiro-bench",-              requestReason = "measure buffered projection replay",-              replayFrom = GlobalPosition 0-            }-        options =-          (defaultRebuildOptions request)-            { replayPageSize = rebuildPageSize-            }-    startCatalogRebuild rebuildCatalog benchmarkRebuildGroupId options >>= \case-      Left err -> liftIO (fail ("unexpected rebuild benchmark result: " <> show err))-      Right report-        | report ^. #runStatus == RebuildRunPromoted -> pure ()-        | otherwise -> liftIO (fail ("rebuild benchmark did not promote: " <> show (report ^. #runStatus)))+rebuildScenarioBench :: Store.KirokuStore -> Benchmark+rebuildScenarioBench store =+  selectedCounterWithCleanup "rebuild" 600 setup (rebuildCountProbe store) $ \runCounter -> bench "three-categories-200" $ nfIO (runRebuild False runCounter)+  where+    setup = do+      setupTrace "rebuild" "three-categories-200" "schema source_events=600 bookkeeping_run_budget=10000"+      runStoreChecked store do+        Store.runTransaction (Tx.sql rebuildBenchSql)+        traverse_ seedRebuildStream rebuildStreams+        registerProjectionCatalog rebuildCatalog >>= \case+          Left err -> liftIO (fail (show err))+          Right _ -> pure ()+      counter <- newIORef 0+      checkpointBenchmarkStore store+      setupTrace "rebuild" "three-categories-200" "checkpoint=before-warm-up warm-up invocations=3 source_events_per_invocation=600"+      replicateM_ 3 (runRebuild True counter)+      probe <- lookupEnv "KEIRO_BENCH_REBUILD_PROBE"+      when (probe == Just "1") (replicateM_ 3 (runRebuild True counter))+      pure counter+    runRebuild probe runCounter = do+      runNumber <- nextBoundedIteration "rebuild" 10000 runCounter+      runStoreChecked store do+        let request =+              RebuildRequest+                { rebuildRunId = benchmarkRebuildRunId runNumber,+                  requestedBy = "keiro-bench",+                  requestReason = "measure buffered projection replay",+                  replayFrom = GlobalPosition 0+                }+            options = (defaultRebuildOptions request) {replayPageSize = rebuildPageSize}+        startCatalogRebuild rebuildCatalog benchmarkRebuildGroupId options >>= \case+          Left err -> liftIO (fail ("unexpected rebuild benchmark result: " <> show err))+          Right report+            | report ^. #runStatus == RebuildRunPromoted -> do+                let sourceEvents = sum [sourceProgress ^. #eventCount | sourceProgress <- report ^. #sources]+                unless (sourceEvents == 600) (liftIO (fail ("rebuild source count changed: " <> show sourceEvents)))+                when probe (liftIO (setupTrace "rebuild" (show runNumber) ("source_events=" <> show sourceEvents)))+            | otherwise -> liftIO (fail ("rebuild benchmark did not promote: " <> show (report ^. #runStatus)))  seedRebuildStream :: (Store :> es) => StreamName -> Eff es () seedRebuildStream streamName' =@@ -909,33 +1025,55 @@ coordinatorFanouts :: [Int] coordinatorFanouts = [10, 100, 1000] -domainRouterFanoutBench :: StoreRunner -> Int -> Benchmark-domainRouterFanoutBench runner fanout =-  bench (show fanout) $ nfIO $ do-    let correlationId = "router-" <> Text.pack (show fanout)-        input = FanoutInput correlationId fanout-    runResourceStoreChecked runner do-      resetFanoutTargets "router" correlationId fanout-      runDomainRouterWorker-        defaultRunCommandOptions-        fanoutDomainRouter-        (fanoutAdapter input)-        (\message -> Just (fanoutSourceEvent, message))+domainRouterFanoutBench :: Store.KirokuStore -> StoreRunner -> Int -> Benchmark+domainRouterFanoutBench store runner fanout =+  selectedCounterWithCleanup label fanout (prepareCommand store label fanout (100000 `div` fanout) action) checkCounts $ \counter -> bench (show fanout) $ nfIO (action counter)+  where+    label = "router-" <> show fanout+    checkCounts n = do+      maintenanceProbe store (Text.pack label <> "-finished")+      workCountProbe store (Text.pack label) ("bench-command-domain-router:" <> Text.pack label <> "-") (n * fanout) (n * fanout)+    action counter = do+      n <- nextIteration label fanout counter+      let correlationId = Text.pack label <> "-" <> Text.pack (show n)+          input = FanoutInput correlationId fanout+      runResourceStoreChecked runner do+        runDomainRouterWorker+          defaultRunCommandOptions+          fanoutDomainRouter+          (fanoutAdapter input)+          (\message -> Just (iterationSourceEvent n, message))+        when (n <= 3) (verifyFanout "router" correlationId fanout) -domainProcessManagerFanoutBench :: StoreRunner -> Int -> Benchmark-domainProcessManagerFanoutBench runner fanout =-  bench (show fanout) $ nfIO $ do-    let correlationId = "process-manager-" <> Text.pack (show fanout)-        input = FanoutInput correlationId fanout-    runResourceStoreChecked runner do-      resetFanoutTargets "process-manager" correlationId fanout-      void (Lifecycle.hardDeleteStream (StreamName ("bench-command-domain-process-manager:" <> correlationId)))-      runDomainProcessManagerWorker-        defaultRunCommandOptions-        fanoutDomainProcessManager-        (fanoutAdapter input)-        (\message -> Just (fanoutSourceEvent, message))+domainProcessManagerFanoutBench :: Store.KirokuStore -> StoreRunner -> Int -> Benchmark+domainProcessManagerFanoutBench store runner fanout =+  selectedCounterWithCleanup label fanout (prepareCommand store label fanout (100000 `div` fanout) action) checkCounts $ \counter -> bench (show fanout) $ nfIO (action counter)+  where+    label = "process-manager-" <> show fanout+    checkCounts n = do+      maintenanceProbe store (Text.pack label <> "-finished")+      workCountProbe store (Text.pack label) ("bench-command-domain-process-manager:" <> Text.pack label <> "-") (n * fanout) (n * fanout)+    action counter = do+      n <- nextIteration label fanout counter+      let correlationId = Text.pack label <> "-" <> Text.pack (show n)+          input = FanoutInput correlationId fanout+      runResourceStoreChecked runner do+        runDomainProcessManagerWorker+          defaultRunCommandOptions+          fanoutDomainProcessManager+          (fanoutAdapter input)+          (\message -> Just (iterationSourceEvent n, message))+        when (n <= 3) (verifyFanout "process-manager" correlationId fanout) +verifyFanout :: (IOE :> es, Store :> es) => Text -> Text -> Int -> Eff es ()+verifyFanout coordinator correlationId fanout =+  for_ [0 .. fanout - 1] \targetIndex -> do+    rows <- Store.readStreamForward (StreamName (fanoutTargetName coordinator correlationId targetIndex)) (StreamVersion 0) 2+    unless (Vector.length rows == 1) (liftIO (fail "fan-out benchmark did not append exactly one event per fresh target"))++iterationSourceEvent :: Int -> RecordedEvent+iterationSourceEvent n = fanoutSourceEvent & #eventId .~ EventId (UUID.fromWords 0 0 1 (fromIntegral n))+ fanoutDomainRouter ::   DomainRouter     FanoutInput@@ -1005,12 +1143,6 @@     <> ":"     <> Text.pack (show targetIndex) -resetFanoutTargets :: (Store :> es) => Text -> Text -> Int -> Eff es ()-resetFanoutTargets coordinator correlationId fanout =-  traverse_-    (\targetIndex -> void (Lifecycle.hardDeleteStream (StreamName (fanoutTargetName coordinator correlationId targetIndex))))-    [0 .. fanout - 1]- fanoutSourceEvent :: RecordedEvent fanoutSourceEvent =   RecordedEvent@@ -1047,7 +1179,8 @@                       payload = input                     },                 ack = AckHandle (\_ -> pure ()),-                lease = Nothing+                lease = Nothing,+                deliveryStatus = pure DeliveryCurrent               }           ],       shutdown = pure ()@@ -1066,3 +1199,54 @@   case result of     Left err -> fail (show err)     Right value -> pure value++-- A plain-IO diagnostic: it never contributes a tasty-bench timing or baseline.+runCommandGrowthProbe :: Store.KirokuStore -> IO ()+runCommandGrowthProbe store = do+  sample "empty" 0+  for_ [1 .. 5000 :: Int] $ \n -> append n+  sample "grown-5000-before-analyze" 5000+  runStoreChecked store (Store.runTransaction (Tx.sql "ANALYZE streams; ANALYZE stream_events; ANALYZE events"))+  sample "grown-5000-after-analyze" 5000+  for_ [5001 .. 10000 :: Int] $ \n -> append n+  sample "grown-10000" 10000+  for_ [10001 .. 100000 :: Int] $ \n -> append n+  sample "grown-100000" 100000+  for_ [100001 .. 3005000 :: Int] $ \n -> append n+  sample "grown-3005000" 3005000+  where+    append n = appendTarget (stream ("bench-probe-" <> Text.pack (show n)))+    appendTarget target = runStoreChecked store $ do+      result <- runDomainCommand defaultRunCommandOptions domainAcceptedOneHandler target EmitOne+      case result of+        Right value | value ^. #result . #eventsAppended == 1 -> pure ()+        other -> liftIO (fail (show other))+    sample :: Text -> Int -> IO ()+    sample label n = do+      commandPlanProbe store label+      for_ [1 .. 5 :: Int] $ \block -> do+        started <- getMonotonicTimeNSec+        replicateM_ 100 $ runStoreChecked store $ do+          result <- runDomainCommand defaultRunCommandOptions domainNoOpHandler (stream "bench-probe-absent") SelectNoOp+          case result of+            Right value | value ^. #result . #eventsAppended == 0 -> pure ()+            other -> liftIO (fail (show other))+        stopped <- getMonotonicTimeNSec+        freshStarted <- getMonotonicTimeNSec+        for_ [1 .. 100 :: Int] $ \sampleIndex ->+          appendTarget (stream (Text.intercalate "-" ["bench-probe-diagnostic", label, Text.pack (show block), Text.pack (show sampleIndex)]))+        freshStopped <- getMonotonicTimeNSec+        putStrLn ("FRESH_PROBE label=" <> Text.unpack label <> " block=" <> show block <> " mean_ns=" <> show ((freshStopped - freshStarted) `div` 100))+        putStrLn ("NOOP_PROBE label=" <> Text.unpack label <> " block=" <> show block <> " events=" <> show n <> " mean_ns=" <> show ((stopped - started) `div` 100))+      hFlush stdout++runResetProbes :: Store.KirokuStore -> Store.KirokuStore -> Store.KirokuStore -> IO ()+runResetProbes outboxStore inboxStore producerStore = do+  checkpointBenchmarkStore outboxStore+  let outboxSeed store count = do+        runStoreChecked store (Store.runTransaction (Tx.sql "TRUNCATE keiro.keiro_outbox"))+        seedOutbox store (take count (scenarioMessages (const Nothing)))+      inboxSeed = runInboxScenario inboxStore (InboxScenario "probe" Nothing PersistFullEnvelope Nothing inboxScenarioMessages)+  tableResetProbe outboxStore "keiro.keiro_outbox" 2000 (outboxSeed outboxStore 2000)+  tableResetProbe inboxStore "keiro.keiro_inbox" 2000 inboxSeed+  tableResetProbe producerStore "keiro.keiro_outbox" 1000 (outboxSeed producerStore 1000)

@@ -1,19 +1,30 @@ -- | Compare the pre-plan-164 fresh-ID path with deterministic producer writes.-module ProducerIdentityBench (producerIdentityBenchmarks) where+module ProducerIdentityBench (producerIdentityBenchmarks, runProducerIdentityProbe) where -import Control.Monad (forM, forM_)+import BenchSetup (selected, setupTrace)+import Control.Exception (evaluate)+import Control.Monad (forM, forM_, replicateM_) import Data.Aeson qualified as Aeson import Data.ByteString qualified as BS-import Data.IORef (IORef, atomicModifyIORef')+import Data.ByteString.Lazy.Char8 qualified as Lazy+import Data.IORef (IORef, atomicModifyIORef', newIORef)+import Data.List (sort, sortOn)+import Data.Text qualified as Text import Data.Time (UTCTime (..), secondsToDiffTime) import Data.Time.Calendar (Day (ModifiedJulianDay)) import Data.UUID qualified as UUID+import GHC.Clock (getMonotonicTimeNSec)+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (preparable) import Keiro.Integration.Event import Keiro.Outbox+import Keiro.Outbox.Schema qualified as Schema import Keiro.Prelude+import Keiro.Test.Postgres (checkpointBenchmarkStore) import Kiroku.Store qualified as Store import Kiroku.Store.Types (EventId (..), EventType (..), GlobalPosition (..), RecordedEvent (..), StreamId (..), StreamVersion (..))-import Test.Tasty.Bench (Benchmark, bcompareWithin, bench, bgroup, nf, nfIO)+import Test.Tasty.Bench (Benchmark, bench, bgroup, nf, nfIO) import "hasql-transaction" Hasql.Transaction qualified as Tx  producerIdentityBenchmarks :: Store.KirokuStore -> IORef Int -> [Benchmark]@@ -21,50 +32,59 @@   [ bgroup       "producer-identity"       [ bench "derive-v1" $ nf identityText (ProducerEventKey (EventId (UUID.fromWords 0 0 0 1)) 0),-        bench "legacy-fresh-1000" $ nfIO (run True False False),-        bcompareWithin 0 1.10 "$0 == \"All.producer-identity.legacy-fresh-1000\"" $ bench "deterministic-fresh-1000" $ nfIO (run True True False),-        bench "deterministic-fresh-and-replay-1000" $ nfIO (run True True True),-        bench "legacy-single-tx-per-event-1000" $ nfIO (run False False False),-        bcompareWithin 0 1.10 "$0 == \"All.producer-identity.legacy-single-tx-per-event-1000\"" $ bench "deterministic-single-tx-per-event-1000" $ nfIO (run False True False)+        sqlBench "legacy-fresh-1000" (run True False False),+        freshBench True "deterministic-fresh-1000" (run True True False),+        sqlBench "deterministic-fresh-and-replay-1000" (run True True True),+        sqlBench "legacy-single-tx-per-event-1000" (run False False False),+        freshBench False "deterministic-single-tx-per-event-1000" (run False True False)       ]   ]   where-    run batch deterministic replay = do-      generation <- atomicModifyIORef' sequenceRef (\n -> (n + 1, n + 1))-      result <- Store.runStoreIO store $ do-        Store.runTransaction (Tx.sql "TRUNCATE keiro.keiro_outbox")-        let recorded = [sourceEvent generation i | i <- [1 .. 1000]]-        if deterministic+    sqlBench name action = selected (setupTrace "producer-identity" name "checkpoint=before-warm-up warm-up invocations=3 timed_reset=TRUNCATE" >> checkpointBenchmarkStore store >> replicateM_ 3 action) $ \_ -> bench name (nfIO action)+    freshBench batch name action =+      selected (checkFreshRatio store sequenceRef batch) $ \_ -> sqlBench name action+    run = runProducerWorkload store sequenceRef++-- The diagnostic pairs call precisely the normal benchmark workload, including+-- timed reset, preparation, insert-result checks and transaction boundaries.+runProducerWorkload :: Store.KirokuStore -> IORef Int -> Bool -> Bool -> Bool -> IO ()+runProducerWorkload store sequenceRef batch deterministic replay = do+  generation <- atomicModifyIORef' sequenceRef (\n -> (n + 1, n + 1))+  result <- Store.runStoreIO store $ do+    Store.runTransaction (Tx.sql "TRUNCATE keiro.keiro_outbox")+    let recorded = [sourceEvent generation i | i <- [1 .. 1000]]+    if deterministic+      then do+        let actions = fmap (\event -> enqueueProducerEventTx producer event 0 draft) recorded+        if batch           then do-            let actions = fmap (\event -> enqueueProducerEventTx producer event 0 draft) recorded-            if batch-              then do-                first <- Store.runTransaction (sequence actions)-                unless (all isInserted first) (error "producer benchmark failed to insert")-              else forM_ actions $ \action -> do-                outcome <- Store.runTransaction action-                unless (isInserted outcome) (error "producer benchmark failed to insert")-            when replay $ do-              second <- Store.runTransaction (sequence actions)-              unless (all isDuplicate second) (error "producer benchmark failed to replay")-          else do-            -- Match the old helper's preparation boundary: bulk preparation-            -- for bulk SQL, and preparation immediately before each individual-            -- transaction for the normal subscription path.-            let prepare event = do-                  oid <- freshOutboxId-                  envelope <- freshIntegrationEvent producer (draft & #sourceEventId ?~ event ^. #eventId & #sourceGlobalPosition ?~ event ^. #globalPosition)-                  pure (oid, envelope)-            if batch-              then do-                messages <- forM recorded prepare-                Store.runTransaction (traverse_ (uncurry enqueueIntegrationEventTx) messages)-              else forM_ recorded $ \event -> do-                message <- prepare event-                Store.runTransaction (uncurry enqueueIntegrationEventTx message)-      case result of-        Left err -> fail (show err)-        Right () -> pure ()+            first <- Store.runTransaction (sequence actions)+            unless (all isInserted first) (error "producer benchmark failed to insert")+          else forM_ actions $ \action -> do+            outcome <- Store.runTransaction action+            unless (isInserted outcome) (error "producer benchmark failed to insert")+        when replay $ do+          second <- Store.runTransaction (sequence actions)+          unless (all isDuplicate second) (error "producer benchmark failed to replay")+      else do+        -- Match the old helper's preparation boundary: bulk preparation+        -- for bulk SQL, and preparation immediately before each individual+        -- transaction for the normal subscription path.+        let prepare event = do+              oid <- freshOutboxId+              envelope <- freshIntegrationEvent producer (draft & #sourceEventId ?~ event ^. #eventId & #sourceGlobalPosition ?~ event ^. #globalPosition)+              pure (oid, envelope)+        if batch+          then do+            messages <- forM recorded prepare+            Store.runTransaction (traverse_ (uncurry enqueueIntegrationEventTx) messages)+          else forM_ recorded $ \event -> do+            message <- prepare event+            Store.runTransaction (uncurry enqueueIntegrationEventTx message)+  case result of+    Left err -> fail (show err)+    Right () -> pure ()+  where     isInserted ProducerInserted {} = True     isInserted _ = False     isDuplicate ProducerDuplicateIdentical {} = True@@ -113,3 +133,169 @@ identityText key =   let identity = deriveProducerIdentity producer key    in (UUID.toText (unOutboxId (identity ^. #outboxId)), identity ^. #messageId)++-- | Opt-in attribution on the existing isolated producer fixture. Normal+-- full workload and the 1.10 ceiling are shared with the paired guard.+runProducerIdentityProbe :: Store.KirokuStore -> IO ()+runProducerIdentityProbe store = do+  checkpointBenchmarkStore store+  let recorded = [sourceEvent 9001 i | i <- [1 .. 1000]]+      factory = traverse (\event -> enqueueProducerEventTx producer event 0 draft) recorded+      reset = probeTx store (Tx.sql "TRUNCATE keiro.keiro_outbox")+      emit label value = Lazy.putStrLn (Aeson.encode (Aeson.object ["phase" Aeson..= (label :: Text), "data" Aeson..= value]))+      countPhase label expected action = do+        observed <- probeTx store $ do+          before <- producerStatementCounts+          action+          after <- producerStatementCounts+          rows <- producerRows+          unless (rows == 1000) (error "producer probe row count differs")+          pure (subtractCounts after before)+        unless (observed == expected) (fail ("producer statement-count mismatch: " <> show observed))+        emit label (Aeson.toJSON observed)+  reset+  countPhase "fresh-factory-statements-legacy-producer-lookup" (0, 1000, 0) $ do+    outcomes <- factory+    unless (all inserted outcomes) (error "producer probe fresh insert failed")+  stored <- Store.runStoreIO store (Schema.listOutbox "bench.outbox") >>= either (fail . show) pure+  let ordered = sortOn (\row -> (row ^. #event) ^. #sourceGlobalPosition) stored+  template <- case ordered of+    row : _ -> pure (row ^. #event)+    [] -> fail "producer probe has no stored envelope"+  let prepare event =+        let identity = deriveProducerIdentity producer (ProducerEventKey (event ^. #eventId) 0)+            envelope = normalizeProducerEvent (template & #messageId .~ (identity ^. #messageId) & #sourceEventId ?~ (event ^. #eventId) & #sourceGlobalPosition ?~ (event ^. #globalPosition))+         in (identity, envelope)+      prepared = map prepare recorded+  unless (length ordered == 1000 && and [identity ^. #outboxId == row ^. #outboxId && envelope == row ^. #event | ((identity, envelope), row) <- zip prepared ordered]) $+    fail "producer probe equivalent preparation differs from the actual factory"+  _ <- evaluate (forceProducerInputs [(identity ^. #outboxId, event) | (identity, event) <- prepared])+  nativeLegacy <- probeLegacyInputs store recorded+  _ <- evaluate (forceProducerInputs nativeLegacy)+  countPhase "replay-factory-statements-legacy-producer-lookup" (0, 1000, 1000) $ do+    outcomes <- factory+    unless (all duplicate outcomes) (error "producer probe replay failed")+  reset+  countPhase "legacy-same-envelope-statements-legacy-producer-lookup" (1000, 0, 0) $+    traverse_ (\(identity, event) -> Schema.enqueueOutboxTx (OutboxMessage (identity ^. #outboxId) event)) prepared+  for_ [1 .. 5 :: Int] $ \sample -> do+    let inputs = [sourceEvent (9100 + sample) i | i <- [1 .. 1000]]+        deterministicPreparation = evaluate (forceProducerInputs [(identity ^. #outboxId, event) | (identity, event) <- map prepare inputs])+        legacyPreparation = do+          messages <- probeLegacyInputs store inputs+          evaluate (forceProducerInputs messages)+    for_ (if even sample then [("deterministic-equivalent-preparation", deterministicPreparation), ("legacy-actual-preparation", legacyPreparation)] else [("legacy-actual-preparation", legacyPreparation), ("deterministic-equivalent-preparation", deterministicPreparation)]) $ \(label, action) -> do+      started <- getMonotonicTimeNSec+      checksum <- action+      ended <- getMonotonicTimeNSec+      emit label (Aeson.object ["sample" Aeson..= sample, "events" Aeson..= (1000 :: Int), "elapsed_ns" Aeson..= (ended - started), "checksum" Aeson..= checksum])+    for_ [True, False] $ \identicalEnvelopes ->+      for_ [True, False] $ \batch ->+        for_ (if even sample then [True, False] else [False, True]) $ \deterministic -> do+          reset+          let actions+                | deterministic =+                    [ do+                        outcome <- Schema.enqueueProducerOutboxTx identity event+                        unless (inserted outcome) (error "prepared producer probe did not insert")+                    | (identity, event) <- prepared+                    ]+                | identicalEnvelopes = [Schema.enqueueOutboxTx (OutboxMessage (identity ^. #outboxId) event) | (identity, event) <- prepared]+                | otherwise = [Schema.enqueueOutboxTx (OutboxMessage oid event) | (oid, event) <- nativeLegacy]+          started <- getMonotonicTimeNSec+          if batch then probeTx store (sequence_ actions) else traverse_ (probeTx store) actions+          ended <- getMonotonicTimeNSec+          rows <- probeTx store producerRows+          unless (rows == 1000) (fail "prepared producer probe retained wrong rows")+          emit "prepared-enqueue" (Aeson.object ["sample" Aeson..= sample, "identical_envelopes" Aeson..= identicalEnvelopes, "batch" Aeson..= batch, "deterministic" Aeson..= deterministic, "events" Aeson..= rows, "elapsed_ns" Aeson..= (ended - started)])+  counter <- newIORef 0+  for_ [True, False] $ \batch -> do+    pairs <- measureFreshPairs store counter batch+    for_ (zip [1 .. freshPairSamples] pairs) $ \(sample, (legacy, deterministic)) ->+      for_ [(False, legacy), (True, deterministic)] $ \(flag, elapsed) ->+        emit "full-factory" (Aeson.object ["sample" Aeson..= sample, "batch" Aeson..= batch, "deterministic" Aeson..= flag, "events" Aeson..= (1000 :: Int), "elapsed_ns" Aeson..= elapsed])+  where+    inserted ProducerInserted {} = True+    inserted _ = False+    duplicate ProducerDuplicateIdentical {} = True+    duplicate _ = False++-- Compare adjacent full actions in alternating order. Both include their normal+-- reset/preparation/transaction boundaries; all fixed samples remain in the log.+measureFreshPairs :: Store.KirokuStore -> IORef Int -> Bool -> IO [(Double, Double)]+measureFreshPairs store counter batch = do+  checkpointBenchmarkStore store+  for_ [False, True] $ \deterministic ->+    replicateM_ 3 (runProducerWorkload store counter batch deterministic False)+  forM [1 .. freshPairSamples] $ \sample -> do+    values <- forM (if even sample then [True, False] else [False, True]) $ \deterministic -> do+      started <- getMonotonicTimeNSec+      runProducerWorkload store counter batch deterministic False+      ended <- getMonotonicTimeNSec+      rows <- probeTx store producerRows+      unless (rows == 1000 && ended > started) (fail "paired producer workload/count mismatch")+      pure (deterministic, fromIntegral (ended - started))+    case (lookup False values, lookup True values) of+      (Just legacy, Just deterministic) -> pure (legacy, deterministic)+      _ -> fail "paired producer measurement is incomplete"++checkFreshRatio :: Store.KirokuStore -> IORef Int -> Bool -> IO ()+checkFreshRatio store counter batch = do+  pairs <- measureFreshPairs store counter batch+  let ratios = [deterministic / legacy | (legacy, deterministic) <- pairs]+      medianRatio = sort ratios !! (freshPairSamples `div` 2)+  Lazy.putStrLn (Aeson.encode (Aeson.object ["phase" Aeson..= ("producer-fresh-ratio" :: Text), "batch" Aeson..= batch, "pairs_ns_legacy_deterministic" Aeson..= pairs, "median_ratio" Aeson..= medianRatio, "ceiling" Aeson..= freshRatioCeiling]))+  unless (medianRatio <= freshRatioCeiling) $+    fail ("producer fresh median paired ratio " <> show medianRatio <> " exceeds " <> show freshRatioCeiling)++freshPairSamples :: Int+freshPairSamples = 15++freshRatioCeiling :: Double+freshRatioCeiling = 1.10++probeLegacyInputs :: Store.KirokuStore -> [RecordedEvent] -> IO [(OutboxId, IntegrationEvent)]+probeLegacyInputs store inputs =+  Store.runStoreIO+    store+    ( forM inputs $ \event -> do+        oid <- freshOutboxId+        envelope <- freshIntegrationEvent producer (draft & #sourceEventId ?~ (event ^. #eventId) & #sourceGlobalPosition ?~ (event ^. #globalPosition))+        pure (oid, envelope)+    )+    >>= either (fail . show) pure++probeTx :: Store.KirokuStore -> Tx.Transaction a -> IO a+probeTx store action = Store.runStoreIO store (Store.runTransaction action) >>= either (fail . show) pure++producerRows :: Tx.Transaction Int64+producerRows = Tx.statement () (preparable "SELECT count(*) FROM keiro.keiro_outbox" E.noParams (D.singleRow (D.column (D.nonNullable D.int8))))++producerStatementCounts :: Tx.Transaction (Int64, Int64, Int64)+producerStatementCounts = do+  rows <- Tx.statement () (preparable "SELECT statement, generic_plans + custom_plans FROM pg_prepared_statements" E.noParams (D.rowList ((,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.int8))))+  let inserts = [(sql, n) | (sql, n) <- rows, "INSERT INTO keiro.keiro_outbox" `Text.isInfixOf` sql]+      count fragment = sum [n | (sql, n) <- inserts, fragment `Text.isInfixOf` sql]+      lookupCount = sum [n | (sql, n) <- rows, "WHERE outbox_id = $1 OR" `Text.isInfixOf` sql, "FOR UPDATE" `Text.isInfixOf` sql]+  pure (count "ON CONFLICT (source, message_id) DO NOTHING", count "ON CONFLICT DO NOTHING", lookupCount)++subtractCounts :: (Int64, Int64, Int64) -> (Int64, Int64, Int64) -> (Int64, Int64, Int64)+subtractCounts (a, b, c) (x, y, z) = (a - x, b - y, c - z)++-- Force the dynamic identity, provenance and timestamp fields without adding+-- envelope serialization to the preparation estimate.+{-# NOINLINE forceProducerInputs #-}+forceProducerInputs :: [(OutboxId, IntegrationEvent)] -> Int+forceProducerInputs = foldl' step 0+  where+    step total (oid, event) = case event ^. #occurredAt of+      UTCTime day time ->+        day `seq`+          time `seq`+            forceProvenance event `seq`+              (total + Text.length (UUID.toText (unOutboxId oid)) + Text.length (event ^. #messageId) + BS.length (event ^. #payloadBytes))++forceProvenance :: IntegrationEvent -> ()+forceProvenance event =+  (case event ^. #sourceEventId of Nothing -> (); Just (EventId uuid) -> uuid `seq` ()) `seq`+    (case event ^. #sourceGlobalPosition of Nothing -> (); Just (GlobalPosition position) -> position `seq` ())

@@ -7,14 +7,18 @@     runReadModelExplainEvidenceIfRequested,     runReadModelLatencyEvidenceIfRequested,     setupReadModelBench,+    probeReadModelCounts,   ) where +import BenchSetup (selectedWithCleanup, setupTrace)+import CommandProbe (commandPlanProbe, workCountProbe) import Control.Monad (forM, replicateM_) import Data.Aeson qualified as Aeson import Data.ByteString qualified as ByteString+import Data.ByteString.Lazy.Char8 qualified as Lazy import Data.Foldable (toList)-import Data.IORef (IORef, atomicModifyIORef', newIORef)+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) import Data.List qualified as List import Data.Map.Strict qualified as Map import Data.Maybe (listToMaybe)@@ -79,7 +83,7 @@     resumeVersionedRebuild,   ) import Keiro.Stream qualified as Keiro.Stream-import Keiro.Test.Postgres (StoreRunner (..))+import Keiro.Test.Postgres (StoreRunner (..), checkpointBenchmarkStore) import Kiroku.Store qualified as Store import Kiroku.Store.Effect (Store) import Kiroku.Store.Effect.Resource (KirokuStoreResource)@@ -102,6 +106,7 @@     StreamVersion (..),   ) import System.Environment (lookupEnv)+import System.IO (stderr) import Test.Tasty.Bench (Benchmark, bcompareWithin, bench, bgroup, nfIO) import "hasql-transaction" Hasql.Transaction qualified as Tx import Prelude@@ -185,6 +190,12 @@   runStoreChecked store (Store.runTransaction seedStatusScale)   seedStabilizingHistory store   seedRepairStreams store+  probe <- lookupEnv "KEIRO_BENCH_READ_MODEL_PLAN_PROBE"+  when (probe == Just "1") (commandPlanProbe store "read-model-before-analyze")+  setupTrace "read-model" "shared" "seed_streams=5000 seed_events=25000 analyze=once-before-warm-up"+  runStoreChecked store (Store.runTransaction (Tx.sql "ANALYZE"))+  when (probe == Just "1") (commandPlanProbe store "read-model-after-analyze")+  checkpointBenchmarkStore store   commandCounter <- newIORef 0   asyncCounter <- newIORef 0   runCounter <- newIORef 0@@ -207,23 +218,14 @@       "projection"       [ bgroup           "inline"-          [ bgroup-              "legacy"-              [ inlineScenarioBench fixture "one-target" "bench-legacy-one",-                inlineScenarioBench fixture "three-target" "bench-legacy-three",-                inlineScenarioBench fixture "three-groups" "bench-legacy-groups",-                inlineScenarioBench fixture "many-revisions" "bench-legacy-many-revisions",-                inlineScenarioBench fixture "mixed-delivery" "bench-legacy-mixed"-              ],-            bgroup-              "versioned"-              [ comparedInline "one-target" "bench-versioned-one",-                comparedInline "three-target" "bench-versioned-three",-                comparedInline "three-groups" "bench-versioned-groups",-                comparedInline "many-revisions" "bench-versioned-many-revisions",-                comparedInline "mixed-delivery" "bench-versioned-mixed"+          ( concat+              [ inlinePair "one-target" "one",+                inlinePair "three-target" "three",+                inlinePair "three-groups" "groups",+                inlinePair "many-revisions" "many-revisions",+                inlinePair "mixed-delivery" "mixed"               ]-          ],+          ),         bgroup           "async"           [ bgroup "legacy" [asyncScenarioBench fixture "mixed-delivery" "bench-legacy-mixed"],@@ -265,6 +267,12 @@       ]   ]   where+    -- Compare adjacent populations rather than five early legacy cases with+    -- five later versioned cases after the shared event store has grown.+    inlinePair benchmarkName suffix =+      [ bgroup "legacy" [inlineScenarioBench fixture benchmarkName ("bench-legacy-" <> suffix)],+        bgroup "versioned" [comparedInline benchmarkName ("bench-versioned-" <> suffix)]+      ]     comparedInline benchmarkName scenarioKey =       bcompareWithin 0 projectionDiagnosticTimeBudget (inlineLegacyPattern benchmarkName) $         inlineScenarioBench fixture benchmarkName scenarioKey@@ -328,12 +336,13 @@  inlineScenarioBench :: ReadModelBenchFixture -> String -> Text -> Benchmark inlineScenarioBench fixture benchmarkName scenarioKey =-  bench benchmarkName $ nfIO (runInlineScenario fixture scenarioKey)+  selectedWithCleanup (warmScenario fixture.store scenarioKey (runInlineScenario fixture scenarioKey)) (const (probeScenarioPlans fixture scenarioKey)) $ \_ ->+    bench benchmarkName $ nfIO (runInlineScenario fixture scenarioKey)  runInlineScenario :: ReadModelBenchFixture -> Text -> IO () runInlineScenario fixture scenarioKey = do   scenario <- requireScenario fixture scenarioKey-  invocation <- nextCounter fixture.commandCounter+  invocation <- nextCounter "read-model command events" 500000 fixture.commandCounter   let target = stream ("bench-projection-command-" <> scenarioKey <> "-" <> Text.pack (show invocation))   result <-     runResourceStoreChecked fixture.runner $@@ -350,14 +359,15 @@  asyncScenarioBench :: ReadModelBenchFixture -> String -> Text -> Benchmark asyncScenarioBench fixture benchmarkName scenarioKey =-  bench benchmarkName $ nfIO (runAsyncScenario fixture scenarioKey)+  selectedWithCleanup (warmScenario fixture.store (scenarioKey <> "-async") (runAsyncScenario fixture scenarioKey)) (const (probeScenarioPlans fixture scenarioKey)) $ \_ ->+    bench benchmarkName $ nfIO (runAsyncScenario fixture scenarioKey)  runAsyncScenario :: ReadModelBenchFixture -> Text -> IO () runAsyncScenario fixture scenarioKey = do   scenario <- requireScenario fixture scenarioKey   (projectionId, projection) <-     maybe (fail ("scenario has no async delivery: " <> Text.unpack scenarioKey)) pure scenario.asyncDelivery-  invocation <- nextCounter fixture.asyncCounter+  invocation <- nextCounter "read-model async operations" 1000000 fixture.asyncCounter   outcome <-     runStoreChecked fixture.store $       Store.runTransaction@@ -395,7 +405,7 @@ runPromotionScenario :: ReadModelBenchFixture -> IO () runPromotionScenario fixture = do   scenario <- requireScenario fixture "bench-versioned-three"-  runNumber <- nextCounter fixture.runCounter+  runNumber <- nextCounter "read-model promotion runs" 10000 fixture.runCounter   promoteToV1 <- atomicModifyIORef' fixture.promotionRevision (\current -> (not current, current))   let group = firstOrError "promotion scenario group" scenario.scenarioGroups       revisionIds = map (^. #revisionId) group.revisions@@ -1128,21 +1138,22 @@  seedStabilizingHistory :: Store.KirokuStore -> IO () seedStabilizingHistory store =-  runStoreChecked store $-    void $-      Store.appendToStream-        (StreamName "benchseed-history")-        NoStream-        [ EventData-            { eventId = Nothing,-              eventType = EventType "BenchSeedEvent",-              payload = Aeson.Null,-              metadata = Nothing,-              causationId = Nothing,-              correlationId = Nothing-            }-        | _ <- [1 .. stabilizingHistoryEventCount]-        ]+  for_ [1 .. 5000 :: Int] $ \n ->+    runStoreChecked store $+      void $+        Store.appendToStream+          (StreamName ("benchseed-history-" <> Text.pack (show n)))+          NoStream+          [ EventData+              { eventId = Nothing,+                eventType = EventType "BenchSeedEvent",+                payload = Aeson.Null,+                metadata = Nothing,+                causationId = Nothing,+                correlationId = Nothing+              }+          | _ <- [1 .. stabilizingHistoryEventCount `div` 5000]+          ]  repairStreamName :: Int -> StreamName repairStreamName count = StreamName ("benchrepair-" <> Text.pack (show count))@@ -1427,9 +1438,27 @@ requireScenario fixture scenarioKey =   maybe (fail ("unknown read-model benchmark scenario: " <> Text.unpack scenarioKey)) pure (Map.lookup scenarioKey fixture.scenarios) -nextCounter :: IORef Int64 -> IO Int64-nextCounter counter = atomicModifyIORef' counter (\current -> let next = current + 1 in (next, next))+nextCounter :: String -> Int64 -> IORef Int64 -> IO Int64+nextCounter label budget counter = do+  next <- atomicModifyIORef' counter (\current -> let n = current + 1 in (n, n))+  when (next > budget) (fail ("benchmark fixture growth budget exceeded: " <> label <> " iterations=" <> show next))+  pure next +warmScenario :: Store.KirokuStore -> Text -> IO () -> IO ()+warmScenario store label action = do+  setupTrace "read-model" (Text.unpack label) "checkpoint=before-warm-up"+  checkpointBenchmarkStore store+  setupTrace "read-model" (Text.unpack label) "warm-up invocations=3"+  replicateM_ 3 action++probeReadModelCounts :: ReadModelBenchFixture -> IO ()+probeReadModelCounts fixture = do+  commands <- readIORef fixture.commandCounter+  asyncOperations <- readIORef fixture.asyncCounter+  promotionRuns <- readIORef fixture.runCounter+  setupTrace "counts" "read-model" ("command_events=" <> show commands <> " command_budget=500000 async_operations=" <> show asyncOperations <> " async_budget=1000000 promotion_runs=" <> show promotionRuns <> " promotion_budget=10000")+  workCountProbe fixture.store "read-model" "bench-projection-command-" (fromIntegral commands) (fromIntegral commands)+ fixedOccurredAt :: UTCTime fixedOccurredAt = UTCTime (ModifiedJulianDay 61000) (secondsToDiffTime 0) @@ -1439,4 +1468,31 @@  runStoreChecked :: Store.KirokuStore -> Eff '[Store, Error Store.StoreError, IOE] value -> IO value runStoreChecked store action =-  Store.runStoreIO store action >>= either (fail . show) pure+  store `seq` Store.runStoreIO store action >>= either (fail . show) pure++-- Inspect the same prepared group-admission query used by the completed case.+-- This is diagnostic only and never runs during accepted timing samples.+probeScenarioPlans :: ReadModelBenchFixture -> Text -> IO ()+probeScenarioPlans fixture scenarioKey = do+  enabled <- lookupEnv "KEIRO_BENCH_READ_MODEL_PLAN_PROBE"+  when (enabled == Just "1") $ do+    commandPlanProbe fixture.store (scenarioKey <> "-after-case")+    scenario <- requireScenario fixture scenarioKey+    for_ scenario.scenarioGroups $ \group -> do+      let groupText = rebuildGroupIdText group.groupId+      plan <- runStoreChecked fixture.store $ Store.runTransaction $ do+        names <-+          Tx.statement () $+            preparable+              "SELECT name FROM pg_prepared_statements WHERE statement LIKE 'SELECT locked_group.status,%'"+              E.noParams+              (D.rowList (D.column (D.nonNullable D.text)))+        name <- case names of+          [name] -> pure name+          _ -> error ("expected actual projection group query: " <> show names)+        Tx.statement () $+          preparable+            ("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) EXECUTE \"" <> Text.replace "\"" "\"\"" name <> "\"('" <> Text.replace "'" "''" groupText <> "')")+            E.noParams+            (D.singleRow (D.column (D.nonNullable D.json)))+      Lazy.hPutStrLn stderr (Aeson.encode (Aeson.object ["scenario" Aeson..= scenarioKey, "group" Aeson..= groupText, "plan" Aeson..= (plan :: Aeson.Value)]))

@@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro-version: 0.21.0.0+version: 0.22.0.0 synopsis: Event sourcing framework and workflow engine description:   A library that composes kiroku, keiki, and shibuya into an@@ -20,8 +20,7 @@   CHANGELOG.md   README.md -extra-source-files:-  bench/GuardedCatalogBaselineMain.hs+extra-source-files: bench/GuardedCatalogBaselineMain.hs  source-repository head   type: git@@ -71,12 +70,15 @@     Keiro.Inbox.Kafka     Keiro.Inbox.Schema     Keiro.Inbox.Types+    Keiro.Inspection.Aggregate+    Keiro.Inspection.Cursor     Keiro.Outbox     Keiro.Outbox.Identity     Keiro.Outbox.Kafka     Keiro.Outbox.Schema     Keiro.Outbox.Types     Keiro.ProcessManager+    Keiro.ProcessManager.Inspect     Keiro.ProcessManager.Reaction     Keiro.Projection     Keiro.Projection.Catalog@@ -107,6 +109,7 @@     Keiro.Workflow.Child     Keiro.Workflow.Child.Schema     Keiro.Workflow.Gc+    Keiro.Workflow.Inspection     Keiro.Workflow.Instance     Keiro.Workflow.Resume     Keiro.Workflow.Schema@@ -119,6 +122,7 @@   other-modules:     Keiro.Command.Domain     Keiro.Outbox.Rejection+    Keiro.ProcessManager.Provenance     Keiro.Projection.Types     Keiro.ReadModel.Rebuild.Group     Keiro.ReadModel.Rebuild.Runner@@ -128,6 +132,7 @@     Keiro.ReplayDigest     Keiro.Workflow.Awakeable.Internal.Identity     Keiro.Workflow.Child.Cancel+    Keiro.Workflow.Inspection.Query     Keiro.Workflow.Instance.Schema     Keiro.Workflow.Journal     Paths_keiro@@ -154,6 +159,7 @@     aeson-casing >=0.2 && <0.3,     base >=4.21 && <5,     base16-bytestring >=1.0.2 && <1.1,+    base64-bytestring >=1.2 && <1.3,     bytestring >=0.11 && <0.13,     containers >=0.6 && <0.8,     contravariant-extras >=0.3 && <0.4,@@ -170,13 +176,13 @@     hs-opentelemetry-semantic-conventions >=1.40 && <2,     keiki >=0.9 && <0.10,     keiki-codec-json >=0.9 && <0.10,-    keiro-core ^>=0.21.0.0,-    kiroku-store ^>=0.11.0.0,+    keiro-core ^>=0.22.0.0,+    kiroku-store ^>=0.12.0.0,     lens >=5.2 && <5.4,     mmzk-typeid >=0.7 && <0.8,     random >=1.2.1 && <1.4,     scientific >=0.3 && <0.4,-    shibuya-core ^>=0.10.0.1,+    shibuya-core ^>=0.12.0.0,     stm >=2.5 && <2.6,     streamly >=0.11 && <0.12,     streamly-core >=0.3 && <0.4,@@ -213,6 +219,7 @@     ReplayCompatibilitySpec     VersionedRebuildSpec     VersionedTargetPostgresSpec+    WorkflowInspectionSpec     WorkflowReplayCompatibilitySpec    ghc-options:@@ -240,11 +247,11 @@     keiki,     keiki-codec-json,     keiro,-    keiro-migrations ^>=0.21.0.0,-    keiro-test-support ^>=0.21.0.0,-    kiroku-store ^>=0.11.0.0,+    keiro-migrations ^>=0.22.0.0,+    keiro-test-support ^>=0.22.0.0,+    kiroku-store ^>=0.12.0.0,     process >=1.6 && <1.7,-    shibuya-core ^>=0.10.0.1,+    shibuya-core ^>=0.12.0.0,     stm >=2.5 && <2.6,     streamly-core >=0.3 && <0.4,     text >=2.1 && <2.2,@@ -277,11 +284,11 @@     keiki >=0.9 && <0.10,     keiki-codec-json >=0.9 && <0.10,     keiro,-    keiro-core ^>=0.21.0.0,-    keiro-test-support ^>=0.21.0.0,-    kiroku-store ^>=0.11.0.0,-    shibuya-core ^>=0.10.0.1,-    shibuya-kiroku-adapter ^>=0.6.0.1,+    keiro-core ^>=0.22.0.0,+    keiro-test-support ^>=0.22.0.0,+    kiroku-store ^>=0.12.0.0,+    shibuya-core ^>=0.12.0.0,+    shibuya-kiroku-adapter ^>=0.7.0.0,     stm >=2.5 && <2.6,     streamly-core >=0.3 && <0.4,     text >=2.1 && <2.2,@@ -297,6 +304,8 @@    main-is: Main.hs   other-modules:+    BenchSetup+    CommandProbe     GuardedCatalogBench     GuardedCatalogExample     GuardedCatalogFixture@@ -325,11 +334,11 @@     hs-opentelemetry-sdk >=1.0 && <1.1,     keiki >=0.9 && <0.10,     keiro,-    keiro-core ^>=0.21.0.0,-    keiro-test-support ^>=0.21.0.0,-    kiroku-store ^>=0.11.0.0,+    keiro-core ^>=0.22.0.0,+    keiro-test-support ^>=0.22.0.0,+    kiroku-store ^>=0.12.0.0,     process >=1.6 && <1.7,-    shibuya-core ^>=0.10.0.1,+    shibuya-core ^>=0.12.0.0,     streamly-core >=0.3 && <0.4,     tasty-bench >=0.4,     text >=2.1 && <2.2,

@@ -85,7 +85,7 @@ import Shibuya.Core.Ack (AckDecision (..)) import Shibuya.Core.Ack qualified as Ack import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..)) import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), MessageId (..)) import Streamly.Data.Stream qualified as Streamly @@ -420,7 +420,8 @@       ack = AckHandle \case         AckHalt _ -> liftIO cancel         decision -> liftIO $ atomically $ void $ tryPutTMVar reply (toSubscriptionResult attempt decision),-      lease = Nothing+      lease = Nothing,+      deliveryStatus = pure DeliveryCurrent     }  toSubscriptionResult :: Word -> AckDecision -> Sub.SubscriptionResult

@@ -47,7 +47,8 @@ -- post-commit snapshot failures are likewise swallowed and counted. Every runner -- accepts a tracer for optional OpenTelemetry spans. ----- The additive hydration primitives are also consumed by "Keiro.ReplayAudit".+-- The source-reporting hydration variant supports inspection without command+-- side effects. The additive hydration primitives are also consumed by "Keiro.ReplayAudit". -- The audit deliberately calls the seeded and full variants separately so the -- public command-serving fallback cannot hide a stale or unreplayable seed. module Keiro.Command@@ -64,6 +65,7 @@      -- * Options     RunCommandOptions (..),+    InspectionProvenancePolicy (..),     defaultRunCommandOptions,      -- * Running commands@@ -82,6 +84,10 @@      -- * Hydration primitives (replay audit)     Hydrated (..),+    HydrationSource (..),+    FullReplayReason (..),+    inspectionRunCommandOptions,+    hydrateWithSource,     hydrate,     hydrateFull,     hydrateSeeded,@@ -330,6 +336,11 @@     HydrationTruncatedChain   deriving stock (Generic, Eq, Show) +-- | Legacy reaction source metadata is explicit application opt-in. Mounting+-- an inspection application never enables it.+data InspectionProvenancePolicy = NoInspectionProvenance | CaptureInspectionProvenance+  deriving stock (Eq, Show, Generic)+ -- | Knobs controlling a single command invocation. -- -- * 'retryLimit' — how many times to rehydrate-and-replay after an@@ -357,6 +368,8 @@ -- * 'seedVerifyInFlightLimit' — maximum concurrent sampled verifications in --   this process. A sampled verification is skipped when the limit is full; --   @0@ removes the limit.+-- * 'inspectionProvenance' — explicit bounded legacy reaction metadata;+--   disabled by default and independent of mounting an inspection server. data RunCommandOptions = RunCommandOptions   { retryLimit :: !Int,     pageSize :: !Int32,@@ -380,14 +393,16 @@     --     keys here are merged on top (see 'Keiro.Codec.metadataFor'). When     --     'Nothing', events carry only the schema-version marker, exactly as     --     before this field existed.-    metadata :: !(Maybe Value)+    metadata :: !(Maybe Value),+    -- | Bounded legacy process-manager provenance; disabled on ordinary writes.+    inspectionProvenance :: !InspectionProvenancePolicy   }   deriving stock (Generic)  -- | Sensible defaults: 3 retries, 256-event read pages, no caller-assigned -- event ids, a no-op pre-append hook, 5ms retry backoff, no metrics, post-append -- replay verification enabled, one sampled snapshot-seed verification per 1000--- snapshot hits, no tracer, and no extra metadata.+-- snapshot hits, no tracer, no extra metadata, and inspection provenance off. defaultRunCommandOptions :: RunCommandOptions defaultRunCommandOptions =   RunCommandOptions@@ -401,7 +416,8 @@       seedVerifySampleRate = 1000,       seedVerifyInFlightLimit = 1,       tracer = Nothing,-      metadata = Nothing+      metadata = Nothing,+      inspectionProvenance = NoInspectionProvenance     }  data Hydrated rs s = Hydrated@@ -411,6 +427,37 @@   }   deriving stock (Generic) +-- | How the ordinary hydration algorithm recovered this state.+data HydrationSource+  = FromSnapshot !StreamVersion+  | FullReplay !FullReplayReason+  deriving stock (Eq, Show, Generic)++data FullReplayReason+  = NoStateCodec+  | SnapshotMissed !SnapshotMissReason+  | SeededReplayFailed !CommandError+  deriving stock (Eq, Show, Generic)++-- | Read-only diagnostics never schedule seed verification or emit telemetry.+inspectionRunCommandOptions :: RunCommandOptions+inspectionRunCommandOptions =+  defaultRunCommandOptions+    { seedVerifySampleRate = 0,+      metrics = Nothing,+      tracer = Nothing,+      verifyReplayOnAppend = False+    }++hydrateWithSource ::+  forall phi rs s ci co es.+  (HasCallStack, IOE :> es, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>+  RunCommandOptions ->+  EventStream phi rs s ci co ->+  Stream (EventStream phi rs s ci co) ->+  Eff es (Either CommandError (Hydrated rs s, HydrationSource))+hydrateWithSource = hydrateUsing (\source -> fmap (\hydrated -> (hydrated, source)))+ data DomainCommandPlan target rs s co rejection noOp   = DomainCommandSilent !(SilentDomainDecision rejection noOp) !(CommandResult target)   | DomainCommandAppend !(Hydrated rs s) !(NonEmpty co) ![EventData]@@ -451,39 +498,47 @@   EventStream phi rs s ci co ->   Stream (EventStream phi rs s ci co) ->   Eff es (Either CommandError (Hydrated rs s))-hydrateAfterProbe options eventStream targetStream =-  snapshotSeed >>= \case-    Nothing -> hydrateFull options eventStream targetStream-    Just seed -> do-      replayed <--        hydrateSeeded-          options-          eventStream-          targetStream-          (seed ^. #state)-          (seed ^. #registers)-          (seed ^. #streamVersion)-      case replayed of-        Left _ -> hydrateFull options eventStream targetStream-        Right hydrated -> do-          for_ (eventStream ^. #stateCodec) $ \codec ->-            scheduleSeedVerification options eventStream targetStream codec seed-          pure (Right hydrated)+hydrateAfterProbe = hydrateUsing (\_ result -> result)++-- Inline at both entry points: the ordinary continuation ignores its lazy+-- source argument, so it constructs neither diagnostics nor a result pair.+{-# INLINE hydrateUsing #-}+hydrateUsing ::+  forall phi rs s ci co es result.+  (HasCallStack, IOE :> es, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>+  (HydrationSource -> Either CommandError (Hydrated rs s) -> result) ->+  RunCommandOptions ->+  EventStream phi rs s ci co ->+  Stream (EventStream phi rs s ci co) ->+  Eff es result+hydrateUsing finish options eventStream targetStream =+  case eventStream ^. #stateCodec of+    Nothing -> fullReplay NoStateCodec+    Just codec ->+      lookupSnapshotSeed ((eventStream ^. #resolveStreamName) targetStream) codec >>= \case+        SnapshotUnavailable reason -> do+          recordSnapshotReadMisses (options ^. #metrics) 1+          case reason of+            SnapshotDecodeFailed _ -> recordSnapshotDecodeFailures (options ^. #metrics) 1+            _ -> pure ()+          fullReplay (SnapshotMissed reason)+        SnapshotHit seed -> do+          recordSnapshotReadHits (options ^. #metrics) 1+          replayed <-+            hydrateSeeded+              options+              eventStream+              targetStream+              (seed ^. #state)+              (seed ^. #registers)+              (seed ^. #streamVersion)+          case replayed of+            Left failure -> fullReplay (SeededReplayFailed failure)+            Right hydrated -> do+              scheduleSeedVerification options eventStream targetStream codec seed+              pure (finish (FromSnapshot (seed ^. #streamVersion)) (Right hydrated))   where-    snapshotSeed =-      case eventStream ^. #stateCodec of-        Nothing -> pure Nothing-        Just codec -> do-          lookupSnapshotSeed ((eventStream ^. #resolveStreamName) targetStream) codec >>= \case-            SnapshotHit seed -> do-              recordSnapshotReadHits (options ^. #metrics) 1-              pure (Just seed)-            SnapshotUnavailable reason -> do-              recordSnapshotReadMisses (options ^. #metrics) 1-              case reason of-                SnapshotDecodeFailed _ -> recordSnapshotDecodeFailures (options ^. #metrics) 1-                _ -> pure ()-              pure Nothing+    fullReplay reason = finish (FullReplay reason) <$> hydrateFull options eventStream targetStream  hydrateFull ::   forall phi rs s ci co es.

@@ -0,0 +1,424 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedRecordDot #-}++-- | Application-owned aggregate descriptions and read-only inspection.+-- Raw history belongs to Kiroku; these reads return folded state and references.+module Keiro.Inspection.Aggregate+  ( AggregateInspector (..),+    SomeAggregateInspector (..),+    AggregateInspectors,+    aggregateInspector,+    inspectorFromAuditTarget,+    displayViaStateCodec,+    aggregateInspectors,+    inspectorName,+    AggregateTypeSummary (..),+    listAggregateTypes,+    AggregateInspectionError (..),+    aggregateStream,+    aggregateIdOf,+    SnapshotCompatibility (..),+    SnapshotConfig (..),+    StoredSnapshot (..),+    SnapshotStatus (..),+    HistoryReference (..),+    AggregateView (..),+    inspectAggregate,+    snapshotStatusFor,+    AggregateListRequest (..),+    AggregateSummary (..),+    AggregateListPage (..),+    maxAggregatePageSize,+    aggregateListCursorKind,+    listAggregates,+    aggregateTypeJson,+    hydrationSourceJson,+    snapshotStatusJson,+    aggregateViewJson,+    aggregateSummaryJson,+    aggregateListPageJson,+    inspectionErrorText,+  )+where++import Data.Aeson (object, (.=))+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Maybe (mapMaybe)+import Data.Text qualified as Text+import Data.Time (diffUTCTime)+import Data.Vector qualified as Vector+import Effectful (Eff, IOE, (:>))+import Keiki.Core (BoolAlg, RegFile)+import Keiro.Command+import Keiro.EventStream (EventStream, SnapshotPolicy (..), StateCodec)+import Keiro.EventStream.Validate (ValidatedEventStream, unvalidated)+import Keiro.Inspection.Cursor+import Keiro.Prelude hiding ((.=))+import Keiro.ReplayAudit (AuditTarget, streamInCategory)+import Keiro.Snapshot (SnapshotMissReason (..))+import Keiro.Snapshot.Schema qualified as Snapshot+import Keiro.Stream (CategoryError, Stream, StreamCategory, categoryName, categoryText, entityStream)+import Keiro.Stream qualified as Stream+import Kiroku.Store.Effect (Store)+import Kiroku.Store.Read (getStream, listStreams)+import Kiroku.Store.Types (StreamInfo (..), StreamName (..), StreamVersion (..), mkBrowsePageSize)++data AggregateInspector phi rs s ci co = AggregateInspector+  { name :: !Text,+    eventStream :: !(ValidatedEventStream phi rs s ci co),+    category :: !(StreamCategory (EventStream phi rs s ci co)),+    mkStream :: !(StreamName -> Maybe (Stream (EventStream phi rs s ci co))),+    displayState :: !(s -> RegFile rs -> Value),+    stateLabel :: !(s -> Text)+  }+  deriving stock (Generic)++aggregateInspector :: (Show s) => Text -> ValidatedEventStream phi rs s ci co -> StreamCategory (EventStream phi rs s ci co) -> (s -> RegFile rs -> Value) -> AggregateInspector phi rs s ci co+aggregateInspector name eventStream cat displayState =+  AggregateInspector name eventStream cat (streamInCategory (categoryText cat)) displayState (Text.pack . show)++inspectorFromAuditTarget :: (Show s) => Text -> AuditTarget phi rs s ci co -> (s -> RegFile rs -> Value) -> Either CategoryError (AggregateInspector phi rs s ci co)+inspectorFromAuditTarget name target display = do+  cat <- Stream.category (target ^. #category)+  pure ((aggregateInspector name (target ^. #eventStream) cat display) {mkStream = target ^. #mkStream})++displayViaStateCodec :: StateCodec (s, RegFile rs) -> s -> RegFile rs -> Value+displayViaStateCodec codec state registers = (codec ^. #encode) (state, registers)++data SomeAggregateInspector where+  SomeAggregateInspector :: (BoolAlg phi (RegFile rs, ci), Eq co) => AggregateInspector phi rs s ci co -> SomeAggregateInspector++type AggregateInspectors = Map Text SomeAggregateInspector++inspectorName :: SomeAggregateInspector -> Text+inspectorName (SomeAggregateInspector inspector) = inspector.name++aggregateInspectors :: [SomeAggregateInspector] -> Either Text AggregateInspectors+aggregateInspectors = go Map.empty+  where+    go acc [] = Right acc+    go acc (inspector : rest)+      | Map.member (inspectorName inspector) acc = Left ("duplicate aggregate inspector: " <> inspectorName inspector)+      | otherwise = go (Map.insert (inspectorName inspector) inspector acc) rest++data AggregateTypeSummary = AggregateTypeSummary+  {name :: !Text, category :: !Text, snapshotsConfigured :: !Bool, snapshotPolicy :: !Text}+  deriving stock (Eq, Show, Generic)++policyLabel :: SnapshotPolicy s -> Text+policyLabel = \case Never -> "never"; Every n -> "every:" <> Text.pack (show n); OnTerminal -> "on_terminal"; Custom _ -> "custom"++listAggregateTypes :: AggregateInspectors -> [AggregateTypeSummary]+listAggregateTypes = map describe . Map.elems+  where+    describe :: SomeAggregateInspector -> AggregateTypeSummary+    describe (SomeAggregateInspector inspector) =+      let definition = unvalidated inspector.eventStream+       in AggregateTypeSummary inspector.name (categoryText inspector.category) (isJust (definition ^. #stateCodec)) (policyLabel (definition ^. #snapshotPolicy))++data AggregateInspectionError+  = AggregateNotFound !StreamName+  | AggregateIdRejected !Text+  | AggregateHydrationFailed !StreamName !CommandError+  | InvalidPageSize !Int+  | InvalidCursor !CursorError+  | ResultTooLarge+  deriving stock (Eq, Show, Generic)++aggregateStream :: AggregateInspector phi rs s ci co -> Text -> Either AggregateInspectionError (Stream (EventStream phi rs s ci co))+aggregateStream inspector aggregateId+  | Text.null (Text.strip aggregateId) = Left (AggregateIdRejected aggregateId)+  | otherwise =+      let target = entityStream inspector.category aggregateId+       in case inspector.mkStream (Stream.streamName target) of+            Nothing -> Left (AggregateIdRejected aggregateId)+            Just accepted | inspectorStreamName inspector accepted == Stream.streamName target -> Right accepted+            _ -> Left (AggregateIdRejected aggregateId)++-- Metadata, membership and hydration must name the same physical stream even+-- when an audit target maps a physical name to a different typed handle.+inspectorStreamName :: AggregateInspector phi rs s ci co -> Stream (EventStream phi rs s ci co) -> StreamName+inspectorStreamName inspector target = (unvalidated inspector.eventStream ^. #resolveStreamName) target++aggregateIdOf :: AggregateInspector phi rs s ci co -> StreamName -> Maybe Text+aggregateIdOf inspector name@(StreamName raw) = do+  accepted <- inspector.mkStream name+  if inspectorStreamName inspector accepted /= name+    then Nothing+    else do+      suffix <- Text.stripPrefix (categoryText inspector.category <> "-") raw+      if Text.null (Text.strip suffix) then Nothing else Just suffix++data SnapshotCompatibility = Compatible | Incompatible ![Text] | Unused+  deriving stock (Eq, Show, Generic)++data SnapshotConfig = SnapshotConfig+  {policy :: !Text, stateCodecVersion :: !Int, regfileShapeHash :: !Text, stateShapeHash :: !Text}+  deriving stock (Eq, Show, Generic)++data StoredSnapshot = StoredSnapshot+  { aheadOfObservedHead :: !Bool,+    streamVersion :: !StreamVersion,+    createdAt :: !UTCTime,+    updatedAt :: !UTCTime,+    ageSeconds :: !Double,+    eventsSince :: !Int64,+    stateCodecVersion :: !Int,+    regfileShapeHash :: !Text,+    stateShapeHash :: !Text,+    compatibility :: !SnapshotCompatibility+  }+  deriving stock (Eq, Show, Generic)++data SnapshotStatus = SnapshotStatus+  {observedAt :: !UTCTime, configured :: !(Maybe SnapshotConfig), stored :: !(Maybe StoredSnapshot)}+  deriving stock (Eq, Show, Generic)++data HistoryReference = HistoryReference+  {owner :: !Text, streamName :: !StreamName, headVersion :: !StreamVersion}+  deriving stock (Eq, Show, Generic)++data AggregateView = AggregateView+  { name :: !Text,+    category :: !Text,+    aggregateId :: !Text,+    streamName :: !StreamName,+    streamVersion :: !StreamVersion,+    createdAt :: !UTCTime,+    deletedAt :: !(Maybe UTCTime),+    truncateBefore :: !StreamVersion,+    hydration :: !(Maybe HydrationSource),+    stateLabel :: !(Maybe Text),+    state :: !(Maybe Value),+    snapshot :: !SnapshotStatus,+    history :: !HistoryReference+  }+  deriving stock (Eq, Show, Generic)++snapshotStatusOf :: (Store :> es) => AggregateInspector phi rs s ci co -> StreamInfo -> Eff es SnapshotStatus+snapshotStatusOf inspector info = do+  observation <- Snapshot.observeSnapshotRow info.id+  let definition = unvalidated inspector.eventStream+      configured = fmap (\codec -> SnapshotConfig (policyLabel (definition ^. #snapshotPolicy)) (codec ^. #stateCodecVersion) (codec ^. #shapeHash) (codec ^. #stateShapeHash)) (definition ^. #stateCodec)+      stored =+        fmap+          ( \row ->+              StoredSnapshot+                (row.streamVersion > info.version)+                row.streamVersion+                row.createdAt+                row.updatedAt+                (realToFrac (diffUTCTime observation.observedAt row.updatedAt))+                (max 0 (versionInt info.version - versionInt row.streamVersion))+                row.stateCodecVersion+                row.regfileShapeHash+                row.stateShapeHash+                (compatibilityOf configured row)+          )+          observation.row+  pure SnapshotStatus {observedAt = observation.observedAt, configured, stored}+  where+    compatibilityOf :: Maybe SnapshotConfig -> Snapshot.SnapshotRow -> SnapshotCompatibility+    compatibilityOf Nothing _ = Unused+    compatibilityOf (Just config) row =+      case [key | (key, differs) <- [("state_codec_version", config.stateCodecVersion /= row.stateCodecVersion), ("regfile_shape_hash", config.regfileShapeHash /= row.regfileShapeHash), ("state_shape_hash", config.stateShapeHash /= row.stateShapeHash)], differs] of+        [] -> Compatible+        keys -> Incompatible keys++snapshotStatusFor :: (Store :> es) => AggregateInspector phi rs s ci co -> Text -> Eff es (Either AggregateInspectionError (StreamName, StreamVersion, SnapshotStatus))+snapshotStatusFor inspector aggregateId =+  case aggregateStream inspector aggregateId of+    Left err -> pure (Left err)+    Right target -> do+      info <- getStream (inspectorStreamName inspector target)+      case info of+        Nothing -> pure (Left (AggregateNotFound (inspectorStreamName inspector target)))+        Just found -> do+          status <- snapshotStatusOf inspector found+          pure (Right (found.name, found.version, status))++inspectAggregate :: (IOE :> es, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) => AggregateInspector phi rs s ci co -> Text -> Eff es (Either AggregateInspectionError AggregateView)+inspectAggregate inspector aggregateId =+  case aggregateStream inspector aggregateId of+    Left err -> pure (Left err)+    Right target ->+      getStream (inspectorStreamName inspector target) >>= \case+        Nothing -> pure (Left (AggregateNotFound (inspectorStreamName inspector target)))+        Just info -> do+          folded <- if isJust info.deletedAt then pure (Right Nothing) else fmap (fmap Just) (hydrateWithSource inspectionRunCommandOptions (unvalidated inspector.eventStream) target)+          case folded of+            Left err -> pure (Left (AggregateHydrationFailed info.name err))+            Right result -> do+              snapshot <- snapshotStatusOf inspector info+              pure+                ( Right+                    AggregateView+                      { name = inspector.name,+                        category = categoryText inspector.category,+                        aggregateId,+                        streamName = info.name,+                        streamVersion = maybe info.version (\(hydrated, _) -> hydrated ^. #streamVersion) result,+                        createdAt = info.createdAt,+                        deletedAt = info.deletedAt,+                        truncateBefore = info.truncateBefore,+                        hydration = fmap snd result,+                        stateLabel = fmap (\(hydrated, _) -> inspector.stateLabel (hydrated ^. #state)) result,+                        state = fmap (\(hydrated, _) -> inspector.displayState (hydrated ^. #state) (hydrated ^. #registers)) result,+                        snapshot,+                        history = HistoryReference "kiroku" info.name info.version+                      }+                )++data AggregateListRequest = AggregateListRequest {pageSize :: !Int, after :: !(Maybe InspectionCursor)}+  deriving stock (Eq, Show, Generic)++data AggregateSummary = AggregateSummary+  { aggregateId :: !Text,+    streamName :: !StreamName,+    streamVersion :: !StreamVersion,+    createdAt :: !UTCTime,+    deletedAt :: !(Maybe UTCTime),+    truncateBefore :: !StreamVersion+  }+  deriving stock (Eq, Show, Generic)++data AggregateListPage = AggregateListPage+  {name :: !Text, category :: !Text, items :: ![AggregateSummary], skipped :: !Int, next :: !(Maybe InspectionCursor)}+  deriving stock (Eq, Show, Generic)++maxAggregatePageSize :: Int+maxAggregatePageSize = 500++aggregateListCursorKind :: Text+aggregateListCursorKind = "aggregate-list"++listAggregates :: (Store :> es) => AggregateInspector phi rs s ci co -> AggregateListRequest -> Eff es (Either AggregateInspectionError AggregateListPage)+listAggregates inspector request =+  case parameters of+    Left err -> pure (Left err)+    Right (after, limit) -> do+      raw <- listStreams (Just (categoryName inspector.category)) (Just (categoryText inspector.category <> "-")) after limit+      let examined = Vector.toList (Vector.take request.pageSize raw)+          items = mapMaybe summary examined+          continuation =+            if Vector.length raw > request.pageSize+              then case reverse examined of+                lastRow : _ -> case lastRow.name of StreamName lastName -> maybe (Left ResultTooLarge) (Right . Just) (encodeCursorBounded aggregateListCursorKind (inspector.name, categoryText inspector.category, lastName))+                [] -> Right Nothing+              else Right Nothing+      pure $ fmap (\next -> AggregateListPage inspector.name (categoryText inspector.category) items (length examined - length items) next) continuation+  where+    parameters+      | request.pageSize < 1 || request.pageSize > maxAggregatePageSize = Left (InvalidPageSize request.pageSize)+      | otherwise = do+          limit <- either (const (Left (InvalidPageSize request.pageSize))) Right (mkBrowsePageSize (request.pageSize + 1))+          after <- traverse decodeAfter request.after+          pure (after, limit)+    decodeAfter :: InspectionCursor -> Either AggregateInspectionError StreamName+    decodeAfter cursor = do+      (name, cat, lastName) <- either (Left . InvalidCursor) Right (decodeCursor aggregateListCursorKind cursor)+      if name == inspector.name && cat == categoryText inspector.category && Text.isPrefixOf (cat <> "-") lastName+        then Right (StreamName lastName)+        else Left (InvalidCursor (MalformedCursor "cursor belongs to another aggregate inspector"))+    summary :: StreamInfo -> Maybe AggregateSummary+    summary info = do+      aggregateId <- aggregateIdOf inspector info.name+      pure (AggregateSummary aggregateId info.name info.version info.createdAt info.deletedAt info.truncateBefore)++versionInt :: StreamVersion -> Int64+versionInt (StreamVersion n) = n++nameText :: StreamName -> Text+nameText (StreamName name) = name++aggregateTypeJson :: AggregateTypeSummary -> Value+aggregateTypeJson summary = object ["name" .= summary.name, "category" .= summary.category, "snapshots_configured" .= summary.snapshotsConfigured, "snapshot_policy" .= summary.snapshotPolicy]++hydrationSourceJson :: HydrationSource -> Value+hydrationSourceJson = \case+  FromSnapshot version -> object ["source" .= ("snapshot" :: Text), "snapshot_version" .= versionInt version]+  FullReplay reason ->+    object+      ( ["source" .= ("full_replay" :: Text)] <> case reason of+          NoStateCodec -> ["reason" .= ("no_state_codec" :: Text)]+          SnapshotMissed miss ->+            ["reason" .= ("snapshot_missed" :: Text)] <> case miss of+              SnapshotNoStream -> ["miss" .= ("no_stream" :: Text)]+              SnapshotNotFound -> ["miss" .= ("not_found" :: Text)]+              SnapshotDecodeFailed detail -> ["miss" .= ("decode_failed" :: Text), "detail" .= detail]+          SeededReplayFailed err -> ["reason" .= ("seeded_replay_failed" :: Text), "error" .= commandErrorClass err, "detail" .= Text.pack (show err)]+      )++snapshotStatusJson :: SnapshotStatus -> Value+snapshotStatusJson status =+  object+    [ "status" .= (if isJust status.stored then "present" else "absent" :: Text),+      "observed_at" .= status.observedAt,+      "configured" .= fmap configJson status.configured,+      "stored" .= fmap storedJson status.stored+    ]+  where+    configJson :: SnapshotConfig -> Value+    configJson config = object ["policy" .= config.policy, "state_codec_version" .= config.stateCodecVersion, "regfile_shape_hash" .= config.regfileShapeHash, "state_shape_hash" .= config.stateShapeHash]+    storedJson :: StoredSnapshot -> Value+    storedJson stored =+      object+        [ "stream_version" .= versionInt stored.streamVersion,+          "created_at" .= stored.createdAt,+          "updated_at" .= stored.updatedAt,+          "age_seconds" .= stored.ageSeconds,+          "events_since" .= stored.eventsSince,+          "ahead_of_observed_head" .= stored.aheadOfObservedHead,+          "state_codec_version" .= stored.stateCodecVersion,+          "regfile_shape_hash" .= stored.regfileShapeHash,+          "state_shape_hash" .= stored.stateShapeHash,+          "compatibility" .= (case stored.compatibility of Compatible -> "compatible"; Incompatible _ -> "incompatible"; Unused -> "unused" :: Text),+          "mismatched" .= (case stored.compatibility of Incompatible fields -> fields; _ -> [])+        ]++aggregateViewJson :: AggregateView -> Value+aggregateViewJson view =+  object+    [ "name" .= view.name,+      "category" .= view.category,+      "aggregate_id" .= view.aggregateId,+      "stream_name" .= nameText view.streamName,+      "stream_version" .= versionInt view.streamVersion,+      "state_version" .= (if isJust view.state then Just (versionInt view.streamVersion) else Nothing),+      "created_at" .= view.createdAt,+      "deleted_at" .= view.deletedAt,+      "truncate_before" .= versionInt view.truncateBefore,+      "hydration" .= fmap hydrationSourceJson view.hydration,+      "state_label" .= view.stateLabel,+      "state" .= view.state,+      "snapshot" .= snapshotStatusJson view.snapshot,+      "history" .= object ["owner" .= view.history.owner, "stream_name" .= nameText view.history.streamName, "head_version" .= versionInt view.history.headVersion]+    ]++aggregateSummaryJson :: AggregateSummary -> Value+aggregateSummaryJson summary =+  object+    [ "aggregate_id" .= summary.aggregateId,+      "stream_name" .= nameText summary.streamName,+      "stream_version" .= versionInt summary.streamVersion,+      "created_at" .= summary.createdAt,+      "deleted_at" .= summary.deletedAt,+      "truncate_before" .= versionInt summary.truncateBefore+    ]++aggregateListPageJson :: AggregateListPage -> Value+aggregateListPageJson page =+  object+    ( ["name" .= page.name, "category" .= page.category, "items" .= map aggregateSummaryJson page.items, "skipped" .= page.skipped]+        <> maybe [] (\cursor -> ["next_cursor" .= cursor.cursorText]) page.next+    )++inspectionErrorText :: AggregateInspectionError -> Text+inspectionErrorText = \case+  AggregateNotFound name -> "aggregate not found: " <> nameText name+  AggregateIdRejected aggregateId -> "aggregate id rejected: " <> aggregateId+  AggregateHydrationFailed name err -> "aggregate hydration failed: " <> nameText name <> ": " <> Text.pack (show err)+  InvalidPageSize n -> "page size must be 1..500: " <> Text.pack (show n)+  InvalidCursor err -> "invalid aggregate cursor: " <> Text.pack (show err)+  ResultTooLarge -> "aggregate continuation exceeds the cursor byte limit"

@@ -0,0 +1,61 @@+module Keiro.Inspection.Cursor+  ( InspectionCursor (..),+    CursorError (..),+    maxCursorBytes,+    encodeCursor,+    encodeCursorBounded,+    decodeCursor,+  )+where++import Data.Aeson (FromJSON, ToJSON, eitherDecodeStrict', encode, object, parseJSON, withObject, (.:), (.=))+import Data.Aeson.Types (parseEither)+import Data.ByteString qualified as ByteString+import Data.ByteString.Base64.URL qualified as Base64+import Data.ByteString.Lazy qualified as LazyByteString+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text+import GHC.Generics (Generic)++-- | An opaque, kind-bound continuation. Clients only echo it to the same read.+newtype InspectionCursor = InspectionCursor {cursorText :: Text}+  deriving stock (Eq, Show, Generic)++data CursorError = MalformedCursor !Text | CursorKindMismatch !Text !Text+  deriving stock (Eq, Show, Generic)++maxCursorBytes :: Int+maxCursorBytes = 4096++encodeCursor :: (ToJSON payload) => Text -> payload -> InspectionCursor+encodeCursor kind payload = encodedCursor (cursorJson kind payload)++-- | Page producers must not emit a continuation the decoder would reject.+-- Demand only the maximum representable JSON prefix before base64 encoding.+encodeCursorBounded :: (ToJSON payload) => Text -> payload -> Maybe InspectionCursor+encodeCursorBounded kind payload =+  let decodedLimit = maxCursorBytes * 3 `div` 4+      prefix = LazyByteString.take (fromIntegral decodedLimit + 1) (cursorJson kind payload)+   in if LazyByteString.length prefix > fromIntegral decodedLimit then Nothing else Just (encodedCursor prefix)++cursorJson :: (ToJSON payload) => Text -> payload -> LazyByteString.ByteString+cursorJson kind payload = encode (object ["k" .= kind, "v" .= payload])++encodedCursor :: LazyByteString.ByteString -> InspectionCursor+encodedCursor = InspectionCursor . Text.decodeUtf8 . Base64.encodeUnpadded . LazyByteString.toStrict++decodeCursor :: (FromJSON payload) => Text -> InspectionCursor -> Either CursorError payload+decodeCursor expected (InspectionCursor token)+  -- Check character count before UTF-8 encoding, then the actual encoded bytes.+  | Text.length token > maxCursorBytes = Left (MalformedCursor "cursor exceeds 4096 encoded bytes")+  | ByteString.length bytes > maxCursorBytes = Left (MalformedCursor "cursor exceeds 4096 encoded bytes")+  | otherwise = do+      decoded <- malformed (Base64.decodeUnpadded bytes)+      value <- malformed (eitherDecodeStrict' decoded)+      (kind, payload) <- malformed (parseEither (withObject "inspection cursor" (\obj -> (,) <$> obj .: "k" <*> obj .: "v")) value)+      if kind /= expected then Left (CursorKindMismatch expected kind) else malformed (parseEither parseJSONValue payload)+  where+    bytes = Text.encodeUtf8 token+    malformed = either (Left . MalformedCursor . Text.pack) Right+    parseJSONValue = parseJSON

@@ -156,6 +156,7 @@     CommandResult,     DomainCommandHandler,     DomainCommandOutcome,+    InspectionProvenancePolicy (..),     RunCommandOptions,     commandErrorClass,     runCommandWithSql,@@ -165,6 +166,7 @@ import Keiro.EventStream (EventStream) import Keiro.EventStream.Validate (ValidatedEventStream, unvalidated) import Keiro.Prelude+import Keiro.ProcessManager.Provenance import Keiro.Projection (InlineProjection, runCommandWithProjections, runDomainCommandWithProjections) import Keiro.Stream (Stream) import Keiro.Telemetry (KeiroMetrics, recordDispatchDeadLettered, recordDispatchDuplicate, recordDispatchFailed, recordDispatchPoison)@@ -520,6 +522,30 @@       Text.pack (show emitIndex)     ] +-- Inline so the disabled branch retains the original options update without+-- constructing provenance, traversing lists or resolving any target names.+{-# INLINE managerCommandOptions #-}+managerCommandOptions :: RunCommandOptions -> EventId -> RecordedEvent -> ProcessManagerAction ci targetCi -> (Stream targetCi -> StoreTypes.StreamName) -> RunCommandOptions+managerCommandOptions options managerEventId source action resolveTarget =+  case options ^. #inspectionProvenance of+    NoInspectionProvenance -> original+    CaptureInspectionProvenance ->+      withProvenanceMetadata+        ( encodeReactionProvenance+            ReactionProvenance+              { sourceEventId = source ^. #eventId,+                sourceGlobalPosition = source ^. #globalPosition,+                sourceStreamId = source ^. #originalStreamId,+                dispatched = [(index, resolveTarget (command ^. #target)) | (index, command) <- zip [0 ..] (action ^. #commands)],+                timerIds = fmap (^. #timerId) (action ^. #timers),+                family = legacyProvenanceFamily,+                truncated = False+              }+        )+        original+  where+    original = options & #eventIds .~ [managerEventId]+ -- | React to a single source event: advance the manager's state, dispatch -- its target commands, and schedule its timers — each under a deterministic, -- idempotent write id.@@ -554,7 +580,7 @@       managerStream = (manager ^. #streamFor) correlationId       managerProbes = deterministicCommandIdProbes (manager ^. #name) correlationId (sourceEvent ^. #eventId) (-1)       managerEventId = NonEmpty.head managerProbes-      managerOptions = options & #eventIds .~ [managerEventId]+      managerOptions = managerCommandOptions options managerEventId sourceEvent action (\target -> ((unvalidated (manager ^. #targetEventStream)) ^. #resolveStreamName) (coerce target))       managerStreamName = ((unvalidated (manager ^. #eventStream)) ^. #resolveStreamName) managerStream   existingManagerId <- firstExistingEventId options managerStreamName managerProbes   case existingManagerId of@@ -694,7 +720,7 @@       managerStream = (manager ^. #streamFor) correlationId       managerProbes = deterministicCommandIdProbes (manager ^. #name) correlationId (sourceEvent ^. #eventId) (-1)       managerEventId = NonEmpty.head managerProbes-      managerOptions = options & #eventIds .~ [managerEventId]+      managerOptions = managerCommandOptions options managerEventId sourceEvent action (\target -> ((unvalidated ((manager ^. #targetHandler) ^. #eventStream)) ^. #resolveStreamName) (coerce target))       managerStreamName = ((unvalidated (manager ^. #eventStream)) ^. #resolveStreamName) managerStream       finish managerResult = pure (Right (correlationId, managerResult, action))   existingManagerId <- firstExistingEventId options managerStreamName managerProbes

@@ -0,0 +1,441 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedRecordDot #-}++-- | Application-owned descriptors and observations of persisted manager state.+-- Inspection uses the runtime hydration algorithm; raw history remains in Kiroku.+module Keiro.ProcessManager.Inspect+  ( ProcessManagerInspector (..),+    SomeProcessManagerInspector (..),+    ProcessManagerInspectors,+    ProcessManagerTypeSummary (..),+    listProcessManagerTypes,+    processManagerTypeJson,+    processManagerInspector,+    domainProcessManagerInspector,+    reactiveProcessManagerInspector,+    ProcessManagerProvenanceMode (..),+    EnabledInput (..),+    ReactionProvenance (..),+    encodeReactionProvenance,+    decodeReactionProvenance,+    DispatchWitness (..),+    DispatchedCommandStatus (..),+    LastReaction (..),+    InstanceViewRequest (..),+    ProcessManagerInspectionError (..),+    ProcessManagerInstanceView (..),+    inspectProcessManagerInstance,+    processManagerViewJson,+    InstanceListRequest (..),+    InstanceListError (..),+    InstanceSummary (..),+    InstancePage (..),+    instanceListCursorKind,+    listProcessManagerInstances,+    instancePageJson,+  )+where++import Data.Aeson (object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.Encoding qualified as Encoding+import Data.Aeson.KeyMap qualified as KeyMap+import Data.List.NonEmpty qualified as NonEmpty+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Maybe (mapMaybe)+import Data.Text qualified as Text+import Data.UUID qualified as UUID+import Data.Vector qualified as Vector+import Effectful (Eff, IOE, (:>))+import Keiki.Core (BoolAlg, HsPred, RegFile)+import Keiki.Core qualified as Keiki+import Keiki.Render.Mermaid (edgeInputName)+import Keiro.Command+import Keiro.EventStream (EventStream (..), SnapshotPolicy (..))+import Keiro.EventStream.Validate (ValidatedEventStream, unvalidated)+import Keiro.Inspection.Aggregate (hydrationSourceJson)+import Keiro.Inspection.Cursor+import Keiro.Prelude hiding ((.=))+import Keiro.ProcessManager (DomainProcessManager, ProcessManager, deterministicCommandIdProbes, firstExistingEventId)+import Keiro.ProcessManager.Provenance+import Keiro.ProcessManager.Reaction (ReactiveProcessManager)+import Keiro.Stream (Stream, StreamCategory, categoryName, categoryText)+import Keiro.Timer+import Kiroku.Store.Effect (Store)+import Kiroku.Store.Read (getStream, listStreams, lookupStreamName, readStreamForward)+import Kiroku.Store.Types hiding (categoryName)++-- | Select the persisted evidence family explicitly; state inspection alone+-- does not authorize interpreting legacy metadata as modern reaction evidence.+data ProcessManagerProvenanceMode = ReadLegacyProvenance | NoReactionProvenance+  deriving stock (Generic, Eq, Show)++-- | Encoders and membership mappings belong to the application. A custom+-- mapping must round-trip through streamFor and resolveStreamName.+data ProcessManagerInspector phi rs s ci co = ProcessManagerInspector+  { name :: !Text,+    eventStream :: !(ValidatedEventStream phi rs s ci co),+    streamFor :: !(Text -> Stream (EventStream phi rs s ci co)),+    category :: !(StreamCategory (EventStream phi rs s ci co)),+    correlationOf :: !(StreamName -> Maybe Text),+    displayState :: !(s -> RegFile rs -> Value),+    stateLabel :: !(s -> Text),+    enabledInputs :: !(s -> [EnabledInput]),+    provenanceMode :: !ProcessManagerProvenanceMode+  }+  deriving stock (Generic)++data EnabledInput = EnabledInput {input :: !(Maybe Text), targetState :: !Text}+  deriving stock (Generic, Eq, Show)++mkInspector :: (Show s) => ProcessManagerProvenanceMode -> Text -> ValidatedEventStream (HsPred rs ci) rs s ci co -> (Text -> Stream (EventStream (HsPred rs ci) rs s ci co)) -> StreamCategory (EventStream (HsPred rs ci) rs s ci co) -> (s -> RegFile rs -> Value) -> ProcessManagerInspector (HsPred rs ci) rs s ci co+mkInspector provenanceMode name definition streamFor category displayState =+  ProcessManagerInspector+    name+    definition+    streamFor+    category+    (\(StreamName raw) -> Text.stripPrefix (categoryText category <> "-") raw)+    displayState+    (Text.pack . show)+    ( \state ->+        [ EnabledInput (edgeInputName edge) (Text.pack (show (Keiki.target edge)))+        | edge <- Keiki.edgesOut (unvalidated definition ^. #transducer) state,+          Keiki.mode edge == Keiki.Live+        ]+    )+    provenanceMode++processManagerInspector :: (Show s) => ProcessManager input (HsPred rs ci) rs s ci co tPhi tRs tS tCi tCo -> StreamCategory (EventStream (HsPred rs ci) rs s ci co) -> (s -> RegFile rs -> Value) -> ProcessManagerInspector (HsPred rs ci) rs s ci co+processManagerInspector manager = mkInspector ReadLegacyProvenance (manager ^. #name) (manager ^. #eventStream) (manager ^. #streamFor)++domainProcessManagerInspector :: (Show s) => DomainProcessManager input (HsPred rs ci) rs s ci co tPhi tRs tS tCi tCo rej noOp -> StreamCategory (EventStream (HsPred rs ci) rs s ci co) -> (s -> RegFile rs -> Value) -> ProcessManagerInspector (HsPred rs ci) rs s ci co+domainProcessManagerInspector manager = mkInspector ReadLegacyProvenance (manager ^. #name) (manager ^. #eventStream) (manager ^. #streamFor)++-- | State inspection for the surviving reaction API. Persisted events do not+-- retain enough input/source information to infer modern dispatch provenance.+reactiveProcessManagerInspector :: (Show s) => ReactiveProcessManager input (HsPred rs ci) rs s ci co tPhi tRs tS tCi tCo rej noOp -> StreamCategory (EventStream (HsPred rs ci) rs s ci co) -> (s -> RegFile rs -> Value) -> ProcessManagerInspector (HsPred rs ci) rs s ci co+reactiveProcessManagerInspector manager = mkInspector NoReactionProvenance (manager ^. #name) (manager ^. #sagaHandler . #eventStream) (manager ^. #streamFor)++data SomeProcessManagerInspector where+  SomeProcessManagerInspector :: (BoolAlg phi (RegFile rs, ci), Eq co) => ProcessManagerInspector phi rs s ci co -> SomeProcessManagerInspector++type ProcessManagerInspectors = Map Text SomeProcessManagerInspector++-- | Mounted metadata, independent of persisted instances. Names are catalog+-- lookup keys, including aliases. Timer capability is not currently declared.+data ProcessManagerTypeSummary = ProcessManagerTypeSummary+  { name :: !Text,+    category :: !Text,+    snapshotsConfigured :: !Bool,+    snapshotPolicy :: !Text,+    timersConfigured :: !(Maybe Bool)+  }+  deriving stock (Eq, Show, Generic)++-- | Enumerate only descriptor fields; never evaluate a policy predicate.+-- Lookup names retain the map's ascending order.+listProcessManagerTypes :: ProcessManagerInspectors -> [ProcessManagerTypeSummary]+listProcessManagerTypes = map describe . Map.toAscList+  where+    describe (lookupName, SomeProcessManagerInspector inspector) =+      case unvalidated inspector.eventStream of+        EventStream {stateCodec = codec, snapshotPolicy = policy} ->+          ProcessManagerTypeSummary lookupName (categoryText inspector.category) (isJust codec) (policyLabel policy) Nothing++-- Keep the constructor match separate from the existential map traversal.+-- GHC 9.12.4 otherwise mislabels Custom as Never in the optimized callback+-- sentinel test; OPAQUE also prevents worker/wrapper from removing this boundary.+{-# OPAQUE policyLabel #-}+policyLabel :: SnapshotPolicy s -> Text+policyLabel = \case+  Never -> "never"+  Every n -> "every:" <> Text.pack (show n)+  OnTerminal -> "on_terminal"+  Custom _ -> "custom"++processManagerTypeJson :: ProcessManagerTypeSummary -> Value+processManagerTypeJson summary =+  object+    [ "name" .= summary.name,+      "category" .= summary.category,+      "snapshots_configured" .= summary.snapshotsConfigured,+      "snapshot_policy" .= summary.snapshotPolicy,+      "timers_configured" .= summary.timersConfigured+    ]++-- Direct encoding avoids building an intermediate JSON tree on polling reads.+-- Keep its meaning equal to processManagerTypeJson (covered by parity tests).+instance Aeson.ToJSON ProcessManagerTypeSummary where+  toJSON = processManagerTypeJson+  toEncoding summary =+    Encoding.pairs+      ( "category" .= summary.category+          <> "name" .= summary.name+          <> "snapshot_policy" .= summary.snapshotPolicy+          <> "snapshots_configured" .= summary.snapshotsConfigured+          <> "timers_configured" .= summary.timersConfigured+      )++data DispatchWitness = EventPresent | NotObserved deriving stock (Generic, Eq, Show)++data DispatchedCommandStatus = DispatchedCommandStatus+  {emitIndex :: !Int, targetStreamName :: !StreamName, eventId :: !EventId, witness :: !DispatchWitness}+  deriving stock (Generic, Eq, Show)++data LastReaction = LastReaction+  { eventId :: !EventId,+    eventType :: !EventType,+    streamVersion :: !StreamVersion,+    globalPosition :: !GlobalPosition,+    recordedAt :: !UTCTime,+    provenance :: !(Maybe ReactionProvenance),+    sourceStreamName :: !(Maybe StreamName),+    dispatchedCommands :: ![DispatchedCommandStatus]+  }+  deriving stock (Generic, Eq, Show)++data InstanceViewRequest = InstanceViewRequest {timerPageSize :: !Int}+  deriving stock (Generic, Eq, Show)++data ProcessManagerInspectionError+  = ProcessManagerHydrationFailed !CommandError+  | InvalidInstanceViewRequest !Text+  | ProcessManagerTimerReadFailed !PendingTimerReadError+  deriving stock (Generic, Eq, Show)++data ProcessManagerInstanceView = ProcessManagerInstanceView+  { name :: !Text,+    correlationId :: !Text,+    journalStreamName :: !StreamName,+    streamVersion :: !StreamVersion,+    observedHeadVersion :: !StreamVersion,+    createdAt :: !UTCTime,+    deletedAt :: !(Maybe UTCTime),+    truncateBefore :: !StreamVersion,+    hydration :: !(Maybe HydrationSource),+    stateLabel :: !(Maybe Text),+    state :: !(Maybe Value),+    lastReaction :: !(Maybe LastReaction),+    waitingFor :: ![EnabledInput],+    pendingTimers :: !PendingTimerPage,+    observedAt :: !UTCTime+  }+  deriving stock (Generic, Eq, Show)++physicalName :: ProcessManagerInspector phi rs s ci co -> Text -> StreamName+physicalName inspector correlation = (unvalidated inspector.eventStream ^. #resolveStreamName) (inspector.streamFor correlation)++inspectProcessManagerInstance :: (IOE :> es, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) => ProcessManagerInspector phi rs s ci co -> InstanceViewRequest -> Text -> Eff es (Either ProcessManagerInspectionError (Maybe ProcessManagerInstanceView))+inspectProcessManagerInstance inspector request correlation+  | Text.null (Text.strip correlation) = pure (Left (InvalidInstanceViewRequest "correlation id must not be blank"))+  | Left err <- validatePendingTimerRequest timerFilter timerRequest = pure (Left (ProcessManagerTimerReadFailed err))+  | otherwise = do+      metadata <- getStream streamName+      case metadata of+        Nothing -> pure (Right Nothing)+        Just info -> do+          result <-+            if isJust info.deletedAt+              then pure (Right Nothing)+              else fmap (fmap Just) (hydrateWithSource inspectionRunCommandOptions (unvalidated inspector.eventStream) (inspector.streamFor correlation))+          case result of+            Left err -> pure (Left (ProcessManagerHydrationFailed err))+            Right hydrated -> do+              let version = maybe info.version (\(value, _) -> value ^. #streamVersion) hydrated+              lastReaction <- case hydrated of+                Just _ | version > StreamVersion 0 -> do+                  let StreamVersion n = version+                  events <- readStreamForward streamName (StreamVersion (n - 1)) 1+                  case Vector.toList events of+                    event : _ | event.streamVersion == version -> Just <$> observeReaction event+                    _ -> pure Nothing+                _ -> pure Nothing+              timers <- findPendingTimers timerFilter timerRequest+              pure $ case timers of+                Left err -> Left (ProcessManagerTimerReadFailed err)+                Right page ->+                  Right+                    ( Just+                        ( ProcessManagerInstanceView+                            inspector.name+                            correlation+                            streamName+                            version+                            info.version+                            info.createdAt+                            info.deletedAt+                            info.truncateBefore+                            (snd <$> hydrated)+                            (fmap (\(value, _) -> inspector.stateLabel (value ^. #state)) hydrated)+                            (fmap (\(value, _) -> inspector.displayState (value ^. #state) (value ^. #registers)) hydrated)+                            lastReaction+                            (maybe [] (\(value, _) -> inspector.enabledInputs (value ^. #state)) hydrated)+                            page+                            page.observedAt+                        )+                    )+  where+    streamName = physicalName inspector correlation+    timerFilter = PendingTimerFilter (Just inspector.name) (Just correlation)+    timerRequest = PendingTimerPageRequest request.timerPageSize Nothing+    observeReaction event = case inspector.provenanceMode of+      NoReactionProvenance -> pure (LastReaction event.eventId event.eventType event.streamVersion event.globalPosition event.createdAt Nothing Nothing [])+      ReadLegacyProvenance -> observeLegacyReaction event+    observeLegacyReaction event = do+      let provenance = event.metadata >>= (\case Aeson.Object fields -> KeyMap.lookup "keiro.pm" fields; _ -> Nothing) >>= decodeReactionProvenance+      sourceName <- maybe (pure Nothing) (lookupStreamName . (.sourceStreamId)) provenance+      statuses <- maybe (pure []) (\record -> traverse (probe record) record.dispatched) provenance+      pure (LastReaction event.eventId event.eventType event.streamVersion event.globalPosition event.createdAt provenance sourceName statuses)+    probe record (index, target) = do+      let probes = deterministicCommandIdProbes inspector.name correlation record.sourceEventId index+      existing <- firstExistingEventId inspectionRunCommandOptions target probes+      pure (DispatchedCommandStatus index target (fromMaybe (NonEmpty.head probes) existing) (maybe NotObserved (const EventPresent) existing))++data InstanceListRequest = InstanceListRequest {pageSize :: !Int, after :: !(Maybe InspectionCursor)}+  deriving stock (Generic, Eq, Show)++data InstanceListError = InvalidInstancePageSize !Int | InvalidInstanceCursor !CursorError | InstanceResultTooLarge+  deriving stock (Generic, Eq, Show)++data InstanceSummary = InstanceSummary+  { correlationId :: !Text,+    streamName :: !StreamName,+    streamVersion :: !StreamVersion,+    createdAt :: !UTCTime,+    deletedAt :: !(Maybe UTCTime),+    truncateBefore :: !StreamVersion+  }+  deriving stock (Generic, Eq, Show)++data InstancePage = InstancePage {items :: ![InstanceSummary], skipped :: !Int, next :: !(Maybe InspectionCursor)}+  deriving stock (Generic, Eq, Show)++instanceListCursorKind :: Text+instanceListCursorKind = "pm-list"++-- | Exactly one raw store page. Rejected membership rows count as examined;+-- even an empty displayed page advances from the last examined raw row.+listProcessManagerInstances :: (Store :> es) => ProcessManagerInspector phi rs s ci co -> InstanceListRequest -> Eff es (Either InstanceListError InstancePage)+listProcessManagerInstances inspector request = case parameters of+  Left err -> pure (Left err)+  Right (after, limit) -> do+    raw <- listStreams (Just (categoryName inspector.category)) (Just (categoryText inspector.category <> "-")) after limit+    let examined = Vector.toList (Vector.take request.pageSize raw)+        items = mapMaybe summary examined+        continuation =+          if Vector.length raw > request.pageSize+            then case reverse examined of+              row : _ -> maybe (Left InstanceResultTooLarge) (Right . Just) (encodeCursorBounded instanceListCursorKind (inspector.name, categoryText inspector.category, nameText row.name))+              [] -> Right Nothing+            else Right Nothing+    pure (InstancePage items (length examined - length items) <$> continuation)+  where+    parameters+      | request.pageSize < 1 || request.pageSize > 500 = Left (InvalidInstancePageSize request.pageSize)+      | otherwise = do+          limit <- either (const (Left (InvalidInstancePageSize request.pageSize))) Right (mkBrowsePageSize (request.pageSize + 1))+          after <- traverse decodeAfter request.after+          pure (after, limit)+    decodeAfter cursor = do+      (name, category, lastName) <- either (Left . InvalidInstanceCursor) Right (decodeCursor instanceListCursorKind cursor)+      if name == inspector.name && category == categoryText inspector.category && Text.isPrefixOf (category <> "-") lastName+        then Right (StreamName lastName)+        else Left (InvalidInstanceCursor (MalformedCursor "cursor belongs to another process-manager inspector"))+    summary info = do+      correlation <- inspector.correlationOf info.name+      if Text.null (Text.strip correlation) || physicalName inspector correlation /= info.name+        then Nothing+        else Just (InstanceSummary correlation info.name info.version info.createdAt info.deletedAt info.truncateBefore)++-- | Library-owned representation shared by transport adapters.+processManagerViewJson :: ProcessManagerInstanceView -> Value+processManagerViewJson view =+  object+    [ "name" .= view.name,+      "correlation_id" .= view.correlationId,+      "journal_stream_name" .= nameText view.journalStreamName,+      "stream_version" .= versionNumber view.streamVersion,+      "state_version" .= fmap (const (versionNumber view.streamVersion)) view.state,+      "observed_head_version" .= versionNumber view.observedHeadVersion,+      "created_at" .= view.createdAt,+      "deleted_at" .= view.deletedAt,+      "truncate_before" .= versionNumber view.truncateBefore,+      "observed_at" .= view.observedAt,+      "hydration" .= fmap hydrationSourceJson view.hydration,+      "state_label" .= view.stateLabel,+      "state" .= view.state,+      "last_reaction" .= fmap reactionJson view.lastReaction,+      "pending_timers" .= pendingTimerPageJson view.pendingTimers,+      "waiting_for" .= [object ["input" .= candidate.input, "target_state" .= candidate.targetState] | candidate <- view.waitingFor]+    ]++reactionJson :: LastReaction -> Value+reactionJson reaction =+  object+    [ "event_id" .= eventText reaction.eventId,+      "event_type" .= eventTypeText reaction.eventType,+      "stream_version" .= versionNumber reaction.streamVersion,+      "global_position" .= positionNumber reaction.globalPosition,+      "recorded_at" .= reaction.recordedAt,+      "provenance_status" .= (case reaction.provenance of Nothing -> "unavailable"; Just p | p.truncated -> "truncated"; _ -> "available" :: Text),+      "provenance" .= fmap provenanceJson reaction.provenance+    ]+  where+    provenanceJson record =+      object+        [ "family" .= record.family,+          "truncated" .= record.truncated,+          "source_event_id" .= eventText record.sourceEventId,+          "source_global_position" .= positionNumber record.sourceGlobalPosition,+          "source_stream_id" .= streamNumber record.sourceStreamId,+          "source_stream_name" .= fmap nameText reaction.sourceStreamName,+          "dispatched_commands"+            .= [ object+                   [ "emit_index" .= status.emitIndex,+                     "target_stream_name" .= nameText status.targetStreamName,+                     "event_id" .= eventText status.eventId,+                     "witness" .= (case status.witness of EventPresent -> "event_present"; NotObserved -> "not_observed" :: Text)+                   ]+               | status <- reaction.dispatchedCommands+               ],+          "timer_ids" .= [UUID.toText uuid | TimerId uuid <- record.timerIds]+        ]++instancePageJson :: InstancePage -> Value+instancePageJson page =+  object+    ( [ "items"+          .= [ object+                 [ "correlation_id" .= item.correlationId,+                   "stream_name" .= nameText item.streamName,+                   "stream_version" .= versionNumber item.streamVersion,+                   "created_at" .= item.createdAt,+                   "deleted_at" .= item.deletedAt,+                   "truncate_before" .= versionNumber item.truncateBefore+                 ]+             | item <- page.items+             ],+        "skipped" .= page.skipped+      ]+        <> maybe [] (\cursor -> ["next_cursor" .= cursor.cursorText]) page.next+    )++nameText :: StreamName -> Text+nameText (StreamName value) = value++versionNumber :: StreamVersion -> Int64+versionNumber (StreamVersion value) = value++positionNumber :: GlobalPosition -> Int64+positionNumber (GlobalPosition value) = value++streamNumber :: StreamId -> Int64+streamNumber (StreamId value) = value++eventText :: EventId -> Text+eventText (EventId value) = UUID.toText value++eventTypeText :: EventType -> Text+eventTypeText (EventType value) = value

@@ -0,0 +1,130 @@+{-# LANGUAGE OverloadedRecordDot #-}++-- | The bounded, explicitly enabled legacy reaction metadata contract.+-- Kept below the runner and inspector to avoid a dependency cycle.+module Keiro.ProcessManager.Provenance+  ( ReactionProvenance (..),+    legacyProvenanceFamily,+    encodeReactionProvenance,+    decodeReactionProvenance,+    withProvenanceMetadata,+  )+where++import Control.Monad (unless)+import Data.Aeson (Value (..), object, withObject, withText, (.:), (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KeyMap+import Data.Aeson.Types (Parser, parseEither)+import Data.ByteString.Lazy qualified as Lazy+import Data.Text qualified as Text+import Data.UUID qualified as UUID+import Data.Vector qualified as Vector+import Keiro.Command (RunCommandOptions)+import Keiro.Prelude hiding ((.=))+import Keiro.Timer.Types (TimerId (..))+import Kiroku.Store.Types (EventId (..), GlobalPosition (..), StreamId (..), StreamName (..))++data ReactionProvenance = ReactionProvenance+  { sourceEventId :: !EventId,+    sourceGlobalPosition :: !GlobalPosition,+    sourceStreamId :: !StreamId,+    dispatched :: ![(Int, StreamName)],+    timerIds :: ![TimerId],+    family :: !Text,+    truncated :: !Bool+  }+  deriving stock (Eq, Show, Generic)++legacyProvenanceFamily :: Text+legacyProvenanceFamily = "legacy-positional-v1"++maxEntries, maxBytes :: Int+maxEntries = 64+maxBytes = 16 * 1024++-- | Encode the inner @keiro.pm@ value. Only the supported identity family can+-- be written. Oversized identifiers or frames retain source-only evidence with+-- @truncated=true@; the prefix walks never demand the full follow-up lists.+encodeReactionProvenance :: ReactionProvenance -> Value+encodeReactionProvenance original+  | original.family /= legacyProvenanceFamily = Null+  | any oversized targets = frame sourceOnly+  | not (withinByteBudget value) = frame sourceOnly+  | otherwise = value+  where+    (targets, moreTargets) = prefix original.dispatched+    (timers, moreTimers) = prefix original.timerIds+    bounded = original {dispatched = targets, timerIds = timers, truncated = original.truncated || moreTargets || moreTimers}+    sourceOnly = original {dispatched = [], timerIds = [], truncated = True}+    value = frame bounded+    oversized (_, StreamName name) = Text.length (Text.take (maxBytes + 1) name) > maxBytes+    prefix xs = let (selected, rest) = splitAt maxEntries xs in (selected, not (null rest))++withinByteBudget :: Value -> Bool+withinByteBudget value = Lazy.length (Lazy.take (fromIntegral maxBytes + 1) (Aeson.encode value)) <= fromIntegral maxBytes++frame :: ReactionProvenance -> Value+frame provenance =+  object+    [ "v" .= (1 :: Int),+      "family" .= provenance.family,+      "truncated" .= provenance.truncated,+      "source"+        .= object+          [ "eventId" .= eventText provenance.sourceEventId,+            "globalPosition" .= positionNumber provenance.sourceGlobalPosition,+            "streamId" .= streamNumber provenance.sourceStreamId+          ],+      "dispatched" .= [object ["i" .= index, "target" .= name] | (index, StreamName name) <- provenance.dispatched],+      "timers" .= [UUID.toText timer | TimerId timer <- provenance.timerIds]+    ]+  where+    eventText (EventId uuid) = UUID.toText uuid+    positionNumber (GlobalPosition n) = n+    streamNumber (StreamId n) = n++-- | Decode an inner frame. Missing/unknown versions or identity families are+-- unavailable, never interpreted through another family's deterministic IDs.+decodeReactionProvenance :: Value -> Maybe ReactionProvenance+decodeReactionProvenance = either (const Nothing) Just . parseEither parseFrame+  where+    parseFrame = withObject "legacy reaction provenance" $ \value -> do+      version <- value .: "v" :: Parser Int+      family <- value .: "family"+      unless (version == 1 && family == legacyProvenanceFamily) (fail "unsupported reaction provenance")+      unless (withinByteBudget (Object value)) (fail "provenance exceeds its byte bound")+      (eventId, position, streamId) <-+        value .: "source"+          >>= withObject+            "source"+            ( \source ->+                (,,)+                  <$> (EventId <$> (source .: "eventId" >>= uuidValue))+                  <*> (GlobalPosition <$> source .: "globalPosition")+                  <*> (StreamId <$> source .: "streamId")+            )+      dispatched <-+        value .: "dispatched"+          >>= boundedArray+            ( withObject "dispatch" $ \entry -> do+                index <- entry .: "i"+                unless (index >= (0 :: Int)) (fail "negative emit index")+                (index,) . StreamName <$> entry .: "target"+            )+      timerIds <- value .: "timers" >>= boundedArray (fmap TimerId . uuidValue)+      truncated <- value .: "truncated"+      pure (ReactionProvenance eventId position streamId dispatched timerIds family truncated)+    uuidValue = withText "UUID" (maybe (fail "invalid UUID") pure . UUID.fromText)+    boundedArray parse = \case+      Array values | Vector.length values <= maxEntries -> traverse parse (Vector.toList values)+      _ -> fail "provenance array exceeds its bound"++-- Keep non-object caller values intact for the existing codec rejection. The+-- inner merge remains lazy until encoding, so silent/rejected decisions do not+-- demand provenance or follow-ups merely because capture is enabled.+withProvenanceMetadata :: Value -> RunCommandOptions -> RunCommandOptions+withProvenanceMetadata provenance options =+  options & #metadata .~ case options ^. #metadata of+    Nothing -> Just (object ["keiro.pm" .= provenance])+    Just value -> Just (case value of Object fields -> Object (KeyMap.insert "keiro.pm" provenance fields); _ -> value)

@@ -75,6 +75,12 @@     ProjectionGroupStatusV1 (..),     listProjectionGroupStatuses,     lookupProjectionGroupStatus,+    ProjectionStatusPageRequest (..),+    ProjectionStatusReadError (..),+    ProjectionStatusPage (..),+    projectionGroupsCursorKind,+    validateProjectionStatusRequest,+    listProjectionGroupStatusesPage,     beginGroupRebuild,     resetDeclaredSubscriptions,     insertProjectionDedupBatchStmt,

@@ -7,14 +7,22 @@     ProjectionGroupStatusV1 (..),     listProjectionGroupStatuses,     lookupProjectionGroupStatus,+    ProjectionStatusPageRequest (..),+    ProjectionStatusReadError (..),+    ProjectionStatusPage (..),+    projectionGroupsCursorKind,+    validateProjectionStatusRequest,+    listProjectionGroupStatusesPage,   ) where +import Contravariant.Extras (contrazip2) import Data.Text qualified as Text import Effectful (Eff, (:>)) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Statement (Statement, preparable)+import Keiro.Inspection.Cursor import Keiro.Prelude import Keiro.Projection.Catalog   ( ProjectionRevisionId,@@ -63,6 +71,74 @@     failureDetail :: !(Maybe Text)   }   deriving stock (Eq, Show, Generic)++-- | A bounded observation of the frozen v1 relation. The existing unbounded+-- compatibility API below retains its behavior.+data ProjectionStatusPageRequest = ProjectionStatusPageRequest+  {pageSize :: !Int, after :: !(Maybe InspectionCursor)}+  deriving stock (Eq, Show, Generic)++data ProjectionStatusReadError+  = InvalidProjectionStatusPageSize !Int+  | InvalidProjectionStatusCursor !CursorError+  | ProjectionStatusResultTooLarge+  deriving stock (Eq, Show, Generic)++data ProjectionStatusPage = ProjectionStatusPage+  {observedAt :: !UTCTime, items :: ![ProjectionGroupStatusV1], next :: !(Maybe InspectionCursor)}+  deriving stock (Eq, Show, Generic)++projectionGroupsCursorKind :: Text+projectionGroupsCursorKind = "projection-groups"++validateProjectionStatusRequest :: ProjectionStatusPageRequest -> Either ProjectionStatusReadError (Maybe Text)+validateProjectionStatusRequest request+  | request ^. #pageSize < 1 || request ^. #pageSize > 500 = Left (InvalidProjectionStatusPageSize (request ^. #pageSize))+  | otherwise = traverse decode (request ^. #after)+  where+    decode cursor = do+      (contract, groupId) <- either (Left . InvalidProjectionStatusCursor) Right (decodeCursor projectionGroupsCursorKind cursor)+      if contract /= ("projection_group_status_v1" :: Text)+        then invalid "cursor belongs to another status contract"+        else either (const (invalid "invalid group id")) (Right . rebuildGroupIdText) (mkRebuildGroupId groupId)+    invalid = Left . InvalidProjectionStatusCursor . MalformedCursor++-- | Page in database group-id order, with exclusive continuation and one+-- look-ahead row. Select Keiro-owned group keys first, then use correlated+-- lookups through the frozen owner relation. The OFFSET 0 boundary preserves+-- parameterized lookups instead of flattening the view into fleet-wide joins.+-- A group's own model/checkpoint cardinality remains part of its read cost.+listProjectionGroupStatusesPage :: (Store :> es) => ProjectionStatusPageRequest -> Eff es (Either ProjectionStatusReadError ProjectionStatusPage)+listProjectionGroupStatusesPage request = case validateProjectionStatusRequest request of+  Left err -> pure (Left err)+  Right after -> do+    (observed, rows) <- runTransaction $ do+      observed <- Tx.statement () (preparable "SELECT now()" mempty (D.singleRow (D.column (D.nonNullable D.timestamptz))))+      rows <- Tx.statement (after, fromIntegral size + 1) (pageProjectionGroupStatusesStmt (isJust after))+      pure (observed, rows)+    let selected = take size rows+        continuation = case (drop size rows, reverse selected) of+          (_ : _, row : _) ->+            maybe+              (Left ProjectionStatusResultTooLarge)+              (Right . Just)+              (encodeCursorBounded projectionGroupsCursorKind ("projection_group_status_v1" :: Text, rebuildGroupIdText (row ^. #groupId)))+          _ -> Right Nothing+    pure (ProjectionStatusPage observed selected <$> continuation)+  where+    size = request ^. #pageSize++pageProjectionGroupStatusesStmt :: Bool -> Statement (Maybe Text, Int64) [ProjectionGroupStatusV1]+pageProjectionGroupStatusesStmt after =+  preparable+    ( "WITH inspection_groups AS MATERIALIZED (SELECT group_id FROM keiro.keiro_projection_rebuild_groups"+        <> (if after then " WHERE group_id > $1" else " WHERE $1::text IS NULL")+        <> " ORDER BY group_id LIMIT $2) SELECT status.* FROM inspection_groups page CROSS JOIN LATERAL ("+        <> statusSelect+        <> " WHERE group_id = page.group_id OFFSET 0) status ORDER BY status.group_id"+    )+    (contrazip2 (E.param (E.nullable E.text)) (E.param (E.nonNullable E.int8)))+    (D.rowList projectionGroupStatusDecoder)  -- | List the complete public status relation in stable group-id order. listProjectionGroupStatuses ::

@@ -21,11 +21,13 @@ module Keiro.Snapshot.Schema   ( -- * Rows     SnapshotRow (..),+    SnapshotObservation (..),     SnapshotWrite (..),      -- * Storage     lookupSnapshot,     lookupSnapshotRow,+    observeSnapshotRow,     deleteSnapshotRow,     writeSnapshotRow,   )@@ -59,6 +61,14 @@   }   deriving stock (Generic, Eq, Show) +-- | A snapshot and its database observation clock. This is an inspection read;+-- it does not change the lookup used by ordinary hydration.+data SnapshotObservation = SnapshotObservation+  { row :: !(Maybe SnapshotRow),+    observedAt :: !UTCTime+  }+  deriving stock (Generic, Eq, Show)+ -- | The fields needed to write a snapshot — 'SnapshotRow' minus the -- database-managed timestamps. data SnapshotWrite = SnapshotWrite@@ -99,6 +109,18 @@     $ Tx.statement       (streamIdToInt streamId)       lookupSnapshotRowStmt++-- | Observe a row and database time in one transaction. This is observational:+-- ReadCommitted does not freeze concurrent updates between statements.+observeSnapshotRow :: (Store :> es) => StreamId -> Eff es SnapshotObservation+observeSnapshotRow streamId = runTransaction $ do+  row <- Tx.statement (streamIdToInt streamId) lookupSnapshotRowStmt+  observedAt <- Tx.statement () observationTimeStmt+  pure SnapshotObservation {row, observedAt}++observationTimeStmt :: Statement () UTCTime+observationTimeStmt =+  preparable "SELECT now()" E.noParams (D.singleRow (D.column (D.nonNullable D.timestamptz)))  -- | Delete a stream's advisory snapshot. Returns 'True' when a row existed. -- Event history is untouched, so the next hydration falls back to full replay.

@@ -49,6 +49,15 @@     ownershipSnapshot,     ownershipSnapshotFor,     shardCountSnapshot,+    LeaseState (..),+    ShardOwnershipRow (..),+    ShardOwnershipView (..),+    ShardedSubscriptionSummary (..),+    ShardedSubscriptionList (..),+    leaseStateAt,+    leaseAgeAt,+    shardOwnershipView,+    listShardedSubscriptions,      -- * Fair-share helper     fairShareTarget,@@ -57,18 +66,23 @@  import Data.Set (Set) import Data.Set qualified as Set-import Data.Time (NominalDiffTime)+import Data.Time (NominalDiffTime, diffUTCTime) import Data.Typeable (cast) import Data.UUID.V4 qualified as UUIDv4 import Effectful (Eff, IOE, (:>)) import Effectful.Exception (Exception (..), throwIO) import Keiro.Prelude import Keiro.Subscription.Shard.Schema-  ( WorkerId (..),+  ( ShardOwnershipRow (..),+    ShardedSubscriptionSummary (..),+    WorkerId (..),     claimShardsTx,+    currentDatabaseTime,     ensureShardRows,     listShardCounts,     listShardOwnership,+    listShardOwnershipDetail,+    listShardedSubscriptionSummaries,     lockShardNameTx,     lockShardRowsTx,     releaseShardsTx,@@ -81,6 +95,37 @@ import Kiroku.Store.Transaction (runTransaction) import "hasql-transaction" Hasql.Transaction qualified as Tx +-- | Interpret lease state at the database observation time. Equality remains+-- live, matching the runtime's strict-less-than expiry predicate.+data LeaseState = LeaseLive | LeaseExpired | LeaseUnowned | LeaseInvalid+  deriving stock (Generic, Eq, Show)++data ShardOwnershipView = ShardOwnershipView+  {subscriptionName :: !SubscriptionName, observedAt :: !UTCTime, buckets :: ![ShardOwnershipRow]}+  deriving stock (Generic, Eq, Show)++data ShardedSubscriptionList = ShardedSubscriptionList+  {observedAt :: !UTCTime, subscriptions :: ![ShardedSubscriptionSummary]}+  deriving stock (Generic, Eq, Show)++leaseStateAt :: UTCTime -> ShardOwnershipRow -> LeaseState+leaseStateAt now row = case (row ^. #owner, row ^. #leaseExpiresAt) of+  (Nothing, _) -> LeaseUnowned+  (Just _, Nothing) -> LeaseInvalid+  (Just _, Just expires) | expires < now -> LeaseExpired+  _ -> LeaseLive++leaseAgeAt :: UTCTime -> ShardOwnershipRow -> Maybe NominalDiffTime+leaseAgeAt now row = case row ^. #owner of+  Nothing -> Nothing+  Just _ -> diffUTCTime now <$> row ^. #heartbeatAt++shardOwnershipView :: (Store :> es) => SubscriptionName -> Eff es ShardOwnershipView+shardOwnershipView name = runTransaction (ShardOwnershipView name <$> currentDatabaseTime <*> listShardOwnershipDetail name)++listShardedSubscriptions :: (Store :> es) => Eff es ShardedSubscriptionList+listShardedSubscriptions = runTransaction (ShardedSubscriptionList <$> currentDatabaseTime <*> listShardedSubscriptionSummaries)+ -- | Mint a fresh per-process 'WorkerId' (a random UUID). freshWorkerId :: (IOE :> es) => Eff es WorkerId freshWorkerId = WorkerId <$> liftIO UUIDv4.nextRandom@@ -125,17 +170,17 @@ ensureShards :: (Store :> es) => ShardLease -> Eff es () ensureShards lease = do   outcome <- runTransaction $ do-    lockShardNameTx (subscriptionName lease)-    counts <- listShardCounts (subscriptionName lease)-    let configured = shardCount lease+    lockShardNameTx (lease ^. #subscriptionName)+    counts <- listShardCounts (lease ^. #subscriptionName)+    let configured = lease ^. #shardCount         found = [n | (n, _) <- counts, n /= configured]     if null found-      then ensureShardRows (subscriptionName lease) configured >> pure (Right ())+      then ensureShardRows (lease ^. #subscriptionName) configured >> pure (Right ())       else         pure           ( Left               ShardCountMismatch-                { mismatchSubscriptionName = case subscriptionName lease of SubscriptionName name -> name,+                { mismatchSubscriptionName = case lease ^. #subscriptionName of SubscriptionName name -> name,                   mismatchConfigured = configured,                   mismatchFound = found                 }@@ -214,13 +259,13 @@ acquireOwnedBuckets :: (IOE :> es, Store :> es) => ShardLease -> Int -> Eff es (Set Int) acquireOwnedBuckets lease liveWorkers = do   now <- liftIO getCurrentTime-  let target = fairShareTarget (shardCount lease) liveWorkers+  let target = fairShareTarget (lease ^. #shardCount) liveWorkers   runTransaction $ do-    held <- renewLeaseTx (subscriptionName lease) (workerId lease) now (leaseTtl lease)+    held <- renewLeaseTx (lease ^. #subscriptionName) (workerId lease) now (leaseTtl lease)     -- Claim at most one bucket per pass (see the note above on convergence).     claimed <-       if length held < target-        then claimShardsTx (subscriptionName lease) (workerId lease) 1 now (leaseTtl lease)+        then claimShardsTx (lease ^. #subscriptionName) (workerId lease) 1 now (leaseTtl lease)         else pure []     pure (Set.fromList held <> Set.fromList claimed) @@ -229,14 +274,14 @@ renewOwnedBuckets :: (IOE :> es, Store :> es) => ShardLease -> Eff es (Set Int) renewOwnedBuckets lease = do   now <- liftIO getCurrentTime-  held <- runTransaction (renewLeaseTx (subscriptionName lease) (workerId lease) now (leaseTtl lease))+  held <- runTransaction (renewLeaseTx (lease ^. #subscriptionName) (workerId lease) now (leaseTtl lease))   pure (Set.fromList held)  -- | Graceful release of the given buckets (clean shutdown), so they are -- claimable immediately instead of after lease expiry. relinquish :: (Store :> es) => ShardLease -> Set Int -> Eff es () relinquish lease buckets =-  runTransaction (releaseShardsTx (subscriptionName lease) (workerId lease) (Set.toList buckets))+  runTransaction (releaseShardsTx (lease ^. #subscriptionName) (workerId lease) (Set.toList buckets))  -- | Read every bucket's @(bucket, owner, lease_expires_at)@ for this -- subscription. The worker uses it to estimate @liveWorkers@ (count of distinct@@ -244,7 +289,7 @@ ownershipSnapshot ::   (Store :> es) => ShardLease -> Eff es [(Int, Maybe WorkerId, Maybe UTCTime)] ownershipSnapshot lease =-  runTransaction (listShardOwnership (subscriptionName lease))+  runTransaction (listShardOwnership (lease ^. #subscriptionName))  -- | Read ownership for a subscription without constructing a worker lease. -- Intended for operator and observability surfaces.

@@ -23,6 +23,11 @@ module Keiro.Subscription.Shard.Schema   ( -- * Worker identity     WorkerId (..),+    ShardOwnershipRow (..),+    ShardedSubscriptionSummary (..),+    listShardOwnershipDetail,+    listShardedSubscriptionSummaries,+    currentDatabaseTime,      -- * Lease statements (composable within a transaction)     ensureShardRows,@@ -39,6 +44,7 @@  import Contravariant.Extras (contrazip2, contrazip3, contrazip4, contrazip5) import Data.Int (Int32)+import Data.Map.Strict qualified as Map import Data.Time (NominalDiffTime, addUTCTime) import Data.UUID (UUID) import Hasql.Decoders qualified as D@@ -54,6 +60,93 @@ -- dead process's leases — it claims afresh once the old leases expire. newtype WorkerId = WorkerId UUID   deriving stock (Eq, Ord, Show)++-- | Detailed lease observation. No ownership or lock is acquired by reads.+data ShardOwnershipRow = ShardOwnershipRow+  { bucket :: !Int,+    shardCount :: !Int,+    owner :: !(Maybe WorkerId),+    leaseExpiresAt :: !(Maybe UTCTime),+    heartbeatAt :: !(Maybe UTCTime),+    updatedAt :: !UTCTime+  }+  deriving stock (Generic, Eq, Show)++data ShardedSubscriptionSummary = ShardedSubscriptionSummary+  { subscriptionName :: !SubscriptionName,+    shardCounts :: ![(Int, Int)],+    bucketsTotal :: !Int,+    bucketsLive :: !Int,+    bucketsExpired :: !Int,+    bucketsUnowned :: !Int,+    liveOwners :: !Int+  }+  deriving stock (Generic, Eq, Show)++currentDatabaseTime :: Tx.Transaction UTCTime+currentDatabaseTime = Tx.statement () (preparable "SELECT now()" mempty (D.singleRow (D.column (D.nonNullable D.timestamptz))))++listShardOwnershipDetail :: SubscriptionName -> Tx.Transaction [ShardOwnershipRow]+listShardOwnershipDetail (SubscriptionName name) =+  Tx.statement name $+    preparable+      "SELECT bucket,shard_count,owner_worker_id,lease_expires_at,heartbeat_at,updated_at FROM keiro.keiro_subscription_shards WHERE subscription_name=$1 ORDER BY bucket"+      (E.param (E.nonNullable E.text))+      ( D.rowList+          ( ShardOwnershipRow+              <$> (fromIntegral <$> D.column (D.nonNullable D.int4))+              <*> (fromIntegral <$> D.column (D.nonNullable D.int4))+              <*> (fmap WorkerId <$> D.column (D.nullable D.uuid))+              <*> D.column (D.nullable D.timestamptz)+              <*> D.column (D.nullable D.timestamptz)+              <*> D.column (D.nonNullable D.timestamptz)+          )+      )++listShardedSubscriptionSummaries :: Tx.Transaction [ShardedSubscriptionSummary]+listShardedSubscriptionSummaries = do+  rows <-+    Tx.statement () $+      preparable+        """+        WITH owners AS (+          SELECT subscription_name, count(DISTINCT owner_worker_id) AS live_owners+          FROM keiro.keiro_subscription_shards+          WHERE owner_worker_id IS NOT NULL AND lease_expires_at >= now()+          GROUP BY subscription_name+        )+        SELECT s.subscription_name,s.shard_count,count(*),+          count(*) FILTER (WHERE owner_worker_id IS NOT NULL AND lease_expires_at >= now()),+          count(*) FILTER (WHERE owner_worker_id IS NOT NULL AND lease_expires_at < now()),+          count(*) FILTER (WHERE owner_worker_id IS NULL),coalesce(o.live_owners,0)+        FROM keiro.keiro_subscription_shards s LEFT JOIN owners o USING (subscription_name)+        GROUP BY s.subscription_name,s.shard_count,o.live_owners+        ORDER BY s.subscription_name COLLATE "C",s.shard_count+        """+        mempty+        ( D.rowList+            ( (\name count total live expired unowned owners -> ShardedSubscriptionSummary (SubscriptionName name) [(count, total)] total live expired unowned owners)+                <$> D.column (D.nonNullable D.text)+                <*> (fromIntegral <$> D.column (D.nonNullable D.int4))+                <*> number+                <*> number+                <*> number+                <*> number+                <*> number+            )+        )+  pure (Map.elems (Map.fromListWith merge [(row ^. #subscriptionName, row) | row <- rows]))+  where+    number = fromIntegral <$> D.column (D.nonNullable D.int8)+    merge a b =+      ShardedSubscriptionSummary+        (a ^. #subscriptionName)+        (b ^. #shardCounts <> a ^. #shardCounts)+        (a ^. #bucketsTotal + b ^. #bucketsTotal)+        (a ^. #bucketsLive + b ^. #bucketsLive)+        (a ^. #bucketsExpired + b ^. #bucketsExpired)+        (a ^. #bucketsUnowned + b ^. #bucketsUnowned)+        (a ^. #liveOwners)  -- | Idempotently insert the @N@ rows @(name, bucket = 0..N-1, shard_count = N)@ -- with @owner_worker_id@ left @NULL@. @ON CONFLICT DO NOTHING@ makes calling it

@@ -37,6 +37,17 @@     DeadTimerPage (..),     lookupTimerInspection,     findDeadTimers,+    PendingTimerFilter (..),+    anyPendingTimer,+    PendingTimerPageRequest (..),+    PendingTimerReadError (..),+    PendingTimerPage (..),+    pendingTimerStatuses,+    pendingTimerCursorKind,+    validatePendingTimerRequest,+    findPendingTimers,+    pendingTimerItemJson,+    pendingTimerPageJson,      -- * Guarded foreground resume     DeadTimerClaimRequest (..),

@@ -28,6 +28,17 @@     DeadTimerPage (..),     lookupTimerInspection,     findDeadTimers,+    PendingTimerFilter (..),+    anyPendingTimer,+    PendingTimerPageRequest (..),+    PendingTimerReadError (..),+    PendingTimerPage (..),+    pendingTimerStatuses,+    pendingTimerCursorKind,+    validatePendingTimerRequest,+    findPendingTimers,+    pendingTimerItemJson,+    pendingTimerPageJson,      -- * Guarded foreground resume     DeadTimerClaimRequest (..),@@ -66,15 +77,19 @@ where  import Contravariant.Extras (contrazip2, contrazip5, contrazip6)+import Data.Aeson (object, (.=)) import Data.Int (Int32)-import Data.Time (NominalDiffTime, addUTCTime)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Time (NominalDiffTime, addUTCTime, diffUTCTime) import Data.UUID (UUID)+import Data.UUID qualified as UUID import Data.UUID.V4 qualified as UUIDv4 import Effectful (Eff, IOE, (:>)) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Statement (Statement, preparable)-import Keiro.Prelude+import Keiro.Inspection.Cursor+import Keiro.Prelude hiding ((.=)) import Keiro.Timer.Types (TimerId (..), TimerRequest (..)) import Kiroku.Store.Effect (Store) import Kiroku.Store.Transaction (runTransaction)@@ -205,6 +220,133 @@       ReasonAbsent -> (1, Nothing)       ReasonExact value -> (2, Just value)       ReasonPrefix value -> (3, Just value)++-- | Exact optional owner/correlation filters. These reads reserve no work.+data PendingTimerFilter = PendingTimerFilter+  { processManagerName :: !(Maybe Text),+    correlationId :: !(Maybe Text)+  }+  deriving stock (Generic, Eq, Show)++anyPendingTimer :: PendingTimerFilter+anyPendingTimer = PendingTimerFilter Nothing Nothing++data PendingTimerPageRequest = PendingTimerPageRequest+  { pageSize :: !Int,+    after :: !(Maybe InspectionCursor)+  }+  deriving stock (Generic, Eq, Show)++data PendingTimerReadError+  = InvalidPendingTimerPageSize !Int+  | InvalidPendingTimerCursor !CursorError+  | PendingTimerResultTooLarge+  deriving stock (Generic, Eq, Show)++-- | Live eligibility, ordered by (fire_at,timer_id), not a frozen traversal.+-- Limits bound returned rows, not search work. In particular filtered reads+-- can examine the pending ledger; automatic callers must apply their budget.+data PendingTimerPage = PendingTimerPage+  { observedAt :: !UTCTime,+    timers :: ![TimerInspection],+    next :: !(Maybe InspectionCursor)+  }+  deriving stock (Generic, Eq, Show)++pendingTimerStatuses :: NonEmpty TimerStatus+pendingTimerStatuses = Scheduled :| [Firing]++pendingTimerCursorKind :: Text+pendingTimerCursorKind = "timer-pending"++-- | Pure validation, before borrowing a connection or hydrating an instance.+-- Continuations bind both filters as well as the exclusive database key.+validatePendingTimerRequest :: PendingTimerFilter -> PendingTimerPageRequest -> Either PendingTimerReadError (Maybe (UTCTime, UUID))+validatePendingTimerRequest timerFilter request+  | request ^. #pageSize < 1 || request ^. #pageSize > 500 = Left (InvalidPendingTimerPageSize (request ^. #pageSize))+  | otherwise = traverse decode (request ^. #after)+  where+    invalid = Left . InvalidPendingTimerCursor+    decode cursor = case decodeCursor pendingTimerCursorKind cursor of+      Left err -> invalid err+      Right (manager, correlation, fireAt, timerText)+        | manager /= timerFilter ^. #processManagerName || correlation /= timerFilter ^. #correlationId -> invalid (MalformedCursor "cursor belongs to another timer filter")+        | otherwise -> case UUID.fromText timerText of+            Nothing -> invalid (MalformedCursor "invalid timer UUID")+            Just timerId -> Right (fireAt, timerId)++-- | Read pending timers and database time in one observational transaction.+-- No locks, claims, cleanup or mutation. Statement shapes avoid nullable OR+-- filters so custom and generic plans can use the available ordered indexes.+findPendingTimers :: (Store :> es) => PendingTimerFilter -> PendingTimerPageRequest -> Eff es (Either PendingTimerReadError PendingTimerPage)+findPendingTimers timerFilter request = case validatePendingTimerRequest timerFilter request of+  Left err -> pure (Left err)+  Right position -> do+    (observed, rows) <- runTransaction $ do+      observed <- Tx.statement () (preparable "SELECT now()" mempty (D.singleRow (D.column (D.nonNullable D.timestamptz))))+      rows <-+        Tx.statement+          (timerFilter ^. #processManagerName, timerFilter ^. #correlationId, fst <$> position, snd <$> position, fromIntegral size + 1)+          (findPendingTimersStmt (isJust (timerFilter ^. #processManagerName)) (isJust (timerFilter ^. #correlationId)) (isJust position))+      pure (observed, rows)+    let selected = take size rows+        continuation = case (drop size rows, reverse selected) of+          (_ : _, lastRow : _) ->+            let timer = lastRow ^. #timer+             in Just+                  <$> maybe+                    (Left PendingTimerResultTooLarge)+                    Right+                    ( encodeCursorBounded+                        pendingTimerCursorKind+                        (timerFilter ^. #processManagerName, timerFilter ^. #correlationId, timer ^. #fireAt, UUID.toText (timerIdToUuid (timer ^. #timerId)))+                    )+          _ -> Right Nothing+    pure (PendingTimerPage observed selected <$> continuation)+  where+    size = request ^. #pageSize++-- | Shared representation for pending pages and process-manager details.+pendingTimerItemJson :: UTCTime -> TimerInspection -> Value+pendingTimerItemJson observed inspection =+  let timer = inspection ^. #timer+   in object+        [ "timer_id" .= UUID.toText (timerIdToUuid (timer ^. #timerId)),+          "process_manager_name" .= (timer ^. #processManagerName),+          "correlation_id" .= (timer ^. #correlationId),+          "fire_at" .= (timer ^. #fireAt),+          "payload" .= (timer ^. #payload),+          "status" .= statusToText (timer ^. #status),+          "attempts" .= (timer ^. #attempts),+          "fired_event_id" .= fmap (UUID.toText . eventIdToUuid) (timer ^. #firedEventId),+          "last_error" .= (inspection ^. #lastError),+          "due_in_seconds" .= (realToFrac (diffUTCTime (timer ^. #fireAt) observed) :: Double)+        ]++pendingTimerPageJson :: PendingTimerPage -> Value+pendingTimerPageJson page =+  object+    ( ["observed_at" .= (page ^. #observedAt), "items" .= map (pendingTimerItemJson (page ^. #observedAt)) (page ^. #timers)]+        <> maybe [] (\cursor -> ["next_cursor" .= cursorText cursor]) (page ^. #next)+    )++findPendingTimersStmt :: Bool -> Bool -> Bool -> Statement (Maybe Text, Maybe Text, Maybe UTCTime, Maybe UUID, Int64) [TimerInspection]+findPendingTimersStmt manager correlation cursor =+  preparable+    ( "SELECT timer_id,process_manager_name,correlation_id,fire_at,payload,status,attempts,fired_event_id,last_error FROM keiro.keiro_timers WHERE status IN ('scheduled','firing')"+        <> (if manager then " AND process_manager_name=$1" else " AND $1::text IS NULL")+        <> (if correlation then " AND correlation_id=$2" else " AND $2::text IS NULL")+        <> (if cursor then " AND (fire_at,timer_id)>($3,$4)" else " AND $3::timestamptz IS NULL AND $4::uuid IS NULL")+        <> " ORDER BY fire_at,timer_id LIMIT $5"+    )+    ( contrazip5+        (E.param (E.nullable E.text))+        (E.param (E.nullable E.text))+        (E.param (E.nullable E.timestamptz))+        (E.param (E.nullable E.uuid))+        (E.param (E.nonNullable E.int8))+    )+    (D.rowList timerInspectionDecoder)  -- | Exact dead-row guards and an explicit total attempt ceiling. Lease seconds -- must be between 1 and 2147483647, avoiding interval conversion overflow.

@@ -32,11 +32,14 @@      -- * Read-only lookups     lookupAwakeable,+    lookupAwakeablesOfOwnerPage,     countPendingAwakeables,   ) where  import Contravariant.Extras (contrazip3)+import Data.Functor.Contravariant ((>$<))+import Data.Int (Int32) import Data.UUID (UUID) import Effectful (Eff, (:>)) import Hasql.Decoders qualified as D@@ -46,6 +49,20 @@ import Kiroku.Store.Effect (Store) import Kiroku.Store.Transaction (runTransaction) import "hasql-transaction" Hasql.Transaction qualified as Tx++-- | One exclusive creation-key page, including allocations not yet journaled.+lookupAwakeablesOfOwnerPage :: (Store :> es) => Text -> Text -> Maybe (UTCTime, UUID) -> Int -> Eff es [AwakeableRow]+lookupAwakeablesOfOwnerPage name wid after count = runTransaction (Tx.statement () statement)+  where+    prefix = (const name >$< E.param (E.nonNullable E.text)) <> (const wid >$< E.param (E.nonNullable E.text))+    (cursorSql, cursorParams, limitParam) = case after of+      Nothing -> ("", mempty, "$3")+      Just (created, ident) -> (" AND (created_at,awakeable_id)>($3,$4)", (const created >$< E.param (E.nonNullable E.timestamptz)) <> (const ident >$< E.param (E.nonNullable E.uuid)), "$5")+    statement =+      preparable+        ("SELECT awakeable_id, owner_workflow_name, owner_workflow_id, status, payload, created_at, updated_at, completed_at FROM keiro.keiro_awakeables WHERE owner_workflow_name=$1 AND owner_workflow_id=$2" <> cursorSql <> " ORDER BY created_at,awakeable_id LIMIT " <> limitParam)+        (prefix <> cursorParams <> (const (fromIntegral count :: Int32) >$< E.param (E.nonNullable E.int4)))+        (D.rowList awakeableRowDecoder)  -- | An awakeable's lifecycle state. --

@@ -40,6 +40,7 @@     -- * Read-only lookups     lookupChild,     lookupChildrenOfParent,+    lookupChildrenOfParentPage,     childStatus,     countActiveChildren,     findRunningChildIds,@@ -47,6 +48,8 @@ where  import Contravariant.Extras (contrazip2, contrazip3, contrazip4, contrazip5)+import Data.Functor.Contravariant ((>$<))+import Data.Int (Int32) import Effectful (Eff, (:>)) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E@@ -55,6 +58,24 @@ import Kiroku.Store.Effect (Store) import Kiroku.Store.Transaction (runTransaction) import "hasql-transaction" Hasql.Transaction qualified as Tx++-- | One exclusive creation-key page; size validation belongs to inspection.+lookupChildrenOfParentPage :: (Store :> es) => Text -> Text -> Maybe (UTCTime, Text, Text) -> Int -> Eff es [ChildRow]+lookupChildrenOfParentPage pid pname after count = runTransaction (Tx.statement () statement)+  where+    prefix = (const pid >$< E.param (E.nonNullable E.text)) <> (const pname >$< E.param (E.nonNullable E.text))+    (cursorSql, cursorParams, limitParam) = case after of+      Nothing -> ("", mempty, "$3")+      Just (created, cid, cname) ->+        ( " AND (created_at,child_id,child_name)>($3,$4,$5)",+          (const created >$< E.param (E.nonNullable E.timestamptz)) <> (const cid >$< E.param (E.nonNullable E.text)) <> (const cname >$< E.param (E.nonNullable E.text)),+          "$6"+        )+    statement =+      preparable+        ("SELECT child_id, child_name, parent_id, parent_name, await_step, status, result, failure_reason, created_at, updated_at, completed_at FROM keiro.keiro_workflow_children WHERE parent_id=$1 AND parent_name=$2" <> cursorSql <> " ORDER BY created_at,child_id,child_name LIMIT " <> limitParam)+        (prefix <> cursorParams <> (const (fromIntegral count :: Int32) >$< E.param (E.nonNullable E.int4)))+        (D.rowList childRowDecoder)  -- | A child workflow's lifecycle, as seen from its parent. --

@@ -0,0 +1,353 @@+{-# LANGUAGE OverloadedRecordDot #-}++-- | Bounded observations of workflow state. Cursors bind immutable ordering+-- keys, read kinds, owners, and (for journal/steps) generations. Each request+-- observes then-current rows; it does not hold a snapshot across pages.+module Keiro.Workflow.Inspection+  ( WorkflowStatusClass (..),+    statusClassToText,+    statusClassFromText,+    statusClassOf,+    classStatus,+    InspectionError (..),+    maxInspectionPageSize,+    WorkflowListOrdering (..),+    WorkflowListQuery (..),+    defaultWorkflowListQuery,+    WorkflowListPage (..),+    listWorkflowInstancesPage,+    DetailPageQuery (..),+    OwnerPageQuery (..),+    WorkflowJournalEntry (..),+    WorkflowJournalPage (..),+    WorkflowStepPage (..),+    WorkflowChildPage (..),+    WorkflowAwakeablePage (..),+    readWorkflowJournalPage,+    loadStepIndexPage,+    listWorkflowChildrenPage,+    listWorkflowAwakeablesPage,+    validateWorkflowListQuery,+    validateJournalQuery,+    validateStepQuery,+    validateChildQuery,+    validateAwakeableQuery,+    instanceToJson,+    childToJson,+    awakeableToJson,+    stepToJson,+    journalEntryToJson,+  )+where++import Data.Aeson (FromJSON, ToJSON (..), Value, object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.Types (Pair)+import Data.Int (Int32, Int64)+import Data.Maybe (isJust)+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Time (UTCTime)+import Data.UUID (UUID)+import Data.UUID qualified as UUID+import Data.Vector qualified as Vector+import Effectful (Eff, IOE, (:>))+import Effectful.Exception (throwIO)+import GHC.Generics (Generic)+import Keiro.Codec (decodeRecorded)+import Keiro.Inspection.Cursor+import Keiro.Workflow.Awakeable.Schema qualified as Awakeable+import Keiro.Workflow.Child.Schema qualified as Child+import Keiro.Workflow.Inspection.Query+import Keiro.Workflow.Instance (WorkflowInstanceRow, WorkflowStatus (..))+import Keiro.Workflow.Instance qualified as Instance+import Keiro.Workflow.Schema (WorkflowStepRow)+import Keiro.Workflow.Schema qualified as Schema+import Keiro.Workflow.Types+import Kiroku.Store.Effect (Store)+import Kiroku.Store.Read (readStreamForward)+import Kiroku.Store.Types++maxInspectionPageSize :: Int+maxInspectionPageSize = 500++data InspectionError = InvalidPageSize !Int | InvalidCursor !CursorError | CursorGenerationMismatch !Int !Int | CursorResultTooLarge+  deriving stock (Eq, Show, Generic)++statusClassOf :: WorkflowInstanceRow -> WorkflowStatusClass+statusClassOf row = case row.status of+  WfRunning | row.attempts == 0 -> ClassRunning+  WfRunning -> ClassRetrying+  WfSuspended | isJust row.wakeAfter -> ClassSleeping+  WfSuspended -> ClassAwaiting+  WfCompleted -> ClassCompleted+  WfCancelled -> ClassCancelled+  WfFailed -> ClassFailed++data WorkflowListPage = WorkflowListPage+  {items :: ![WorkflowInstanceRow], nextCursor :: !(Maybe InspectionCursor), observedAt :: !UTCTime}+  deriving stock (Eq, Show, Generic)++validateSize :: Int -> Either InspectionError ()+validateSize size+  | size < 1 || size > maxInspectionPageSize = Left (InvalidPageSize size)+  | otherwise = Right ()++decode :: (FromJSON a) => Text -> InspectionCursor -> Either InspectionError a+decode kind token = either (Left . InvalidCursor) Right (decodeCursor kind token)++invalid :: Text -> Either InspectionError a+invalid = Left . InvalidCursor . MalformedCursor++nextFor :: (ToJSON key) => Text -> Int -> [a] -> (a -> key) -> Either InspectionError (Maybe InspectionCursor)+nextFor kind count rows key+  | length rows <= count = Right Nothing+  | otherwise = case reverse (take count rows) of+      row : _ ->+        maybe (Left CursorResultTooLarge) (Right . Just) (encodeCursorBounded kind (key row))+      [] -> Right Nothing++listWorkflowInstancesPage :: (IOE :> es, Store :> es) => WorkflowListQuery -> Eff es (Either InspectionError WorkflowListPage)+listWorkflowInstancesPage query = case validated of+  Left err -> pure (Left err)+  Right position -> do+    (observedAt, rows) <- Instance.inspectionInstanceRows query position (query.pageSize + 1)+    pure $ do+      nextCursor <- case query.ordering of+        ByKey -> nextFor "wf-key2" query.pageSize rows (\r -> (r.workflowId, r.workflowName))+        ByCreatedDesc -> nextFor "wf-created" query.pageSize rows (\r -> (r.createdAt, r.workflowName, r.workflowId))+      pure (WorkflowListPage (take query.pageSize rows) nextCursor observedAt)+  where+    validated = do+      validateSize query.pageSize+      traverse+        ( \token -> case query.ordering of+            ByKey -> uncurry AfterWorkflowKey <$> decode "wf-key2" token+            ByCreatedDesc -> (\(created, name, wid) -> AfterWorkflowCreated created name wid) <$> decode "wf-created" token+        )+        query.cursor++-- | Pure validation for front ends before they admit work or verify schemas.+-- The reads repeat validation so direct library callers retain the same guard.+validateWorkflowListQuery :: WorkflowListQuery -> Either InspectionError ()+validateWorkflowListQuery query = do+  validateSize query.pageSize+  case query.cursor of+    Nothing -> Right ()+    Just token -> case query.ordering of+      ByKey -> () <$ (decode "wf-key2" token :: Either InspectionError (Text, Text))+      ByCreatedDesc -> () <$ (decode "wf-created" token :: Either InspectionError (UTCTime, Text, Text))++data DetailPageQuery = DetailPageQuery+  {generation :: !(Maybe Int), pageSize :: !Int, cursor :: !(Maybe InspectionCursor)}+  deriving stock (Eq, Show, Generic)++data OwnerPageQuery = OwnerPageQuery+  {pageSize :: !Int, cursor :: !(Maybe InspectionCursor)}+  deriving stock (Eq, Show, Generic)++data WorkflowJournalEntry = WorkflowJournalEntry+  { eventId :: !UUID,+    eventType :: !Text,+    streamVersion :: !Int64,+    globalPosition :: !Int64,+    stepName :: !Text,+    recordedAt :: !UTCTime,+    payload :: !Value,+    event :: !WorkflowJournalEvent+  }+  deriving stock (Eq, Show, Generic)++data WorkflowJournalPage = WorkflowJournalPage+  {generation :: !Int, items :: ![WorkflowJournalEntry], nextCursor :: !(Maybe InspectionCursor), observedAt :: !UTCTime}+  deriving stock (Eq, Show, Generic)++data WorkflowStepPage = WorkflowStepPage+  {generation :: !Int, items :: ![WorkflowStepRow], nextCursor :: !(Maybe InspectionCursor), observedAt :: !UTCTime}+  deriving stock (Eq, Show, Generic)++data WorkflowChildPage = WorkflowChildPage+  {items :: ![Child.ChildRow], nextCursor :: !(Maybe InspectionCursor), observedAt :: !UTCTime}+  deriving stock (Eq, Show, Generic)++data WorkflowAwakeablePage = WorkflowAwakeablePage+  {items :: ![Awakeable.AwakeableRow], nextCursor :: !(Maybe InspectionCursor), observedAt :: !UTCTime}+  deriving stock (Eq, Show, Generic)++validGeneration :: Int -> Bool+validGeneration gen = gen >= 0 && toInteger gen <= toInteger (maxBound :: Int32)++prepareDetail :: forall key es. (FromJSON key, Store :> es) => Text -> WorkflowName -> WorkflowId -> DetailPageQuery -> (key -> Bool) -> Eff es (Either InspectionError (Int, Maybe key))+prepareDetail kind name@(WorkflowName ownerName) wid@(WorkflowId ownerId) query validKey =+  case validateDetail kind ownerName ownerId query validKey of+    Left err -> pure (Left err)+    Right position -> do+      gen <- maybe (Schema.currentGeneration name wid) pure query.generation+      pure $ case position of+        Just (cursorGen, _) | gen /= cursorGen -> Left (CursorGenerationMismatch gen cursorGen)+        _ -> Right (gen, snd <$> position)++validateDetail :: forall key. (FromJSON key) => Text -> Text -> Text -> DetailPageQuery -> (key -> Bool) -> Either InspectionError (Maybe (Int, key))+validateDetail kind ownerName ownerId query validKey = do+  validateSize query.pageSize+  case query.generation of Just gen | not (validGeneration gen) -> invalid "generation is outside the supported range"; _ -> Right ()+  traverse+    ( \token -> do+        (cursorOwner, cursorId, gen, key) <- decode kind token :: Either InspectionError (Text, Text, Int, key)+        if cursorOwner /= ownerName || cursorId /= ownerId+          then invalid "cursor belongs to another workflow"+          else+            if not (validGeneration gen) || not (validKey key)+              then invalid "cursor key is outside the supported range"+              else Right (gen, key)+    )+    query.cursor++validateJournalQuery :: WorkflowName -> WorkflowId -> DetailPageQuery -> Either InspectionError ()+validateJournalQuery (WorkflowName name) (WorkflowId wid) query = () <$ validateDetail "wf-journal" name wid query (> (0 :: Int64))++validateStepQuery :: WorkflowName -> WorkflowId -> DetailPageQuery -> Either InspectionError ()+validateStepQuery (WorkflowName name) (WorkflowId wid) query = () <$ validateDetail "wf-steps" name wid query (const True :: Text -> Bool)++validateChildQuery :: WorkflowName -> WorkflowId -> OwnerPageQuery -> Either InspectionError ()+validateChildQuery name wid query = () <$ (prepareOwner "wf-children" name wid query :: Either InspectionError (Maybe (UTCTime, Text, Text)))++validateAwakeableQuery :: WorkflowName -> WorkflowId -> OwnerPageQuery -> Either InspectionError ()+validateAwakeableQuery name wid query = () <$ (prepareOwner "wf-awakeables" name wid query :: Either InspectionError (Maybe (UTCTime, UUID)))++prepareOwner :: forall key. (FromJSON key) => Text -> WorkflowName -> WorkflowId -> OwnerPageQuery -> Either InspectionError (Maybe key)+prepareOwner kind (WorkflowName ownerName) (WorkflowId ownerId) query = do+  validateSize query.pageSize+  traverse+    ( \token -> do+        (cursorOwner, cursorId, key) <- decode kind token :: Either InspectionError (Text, Text, key)+        if cursorOwner == ownerName && cursorId == ownerId then Right key else invalid "cursor belongs to another workflow"+    )+    query.cursor++readWorkflowJournalPage :: (IOE :> es, Store :> es) => WorkflowName -> WorkflowId -> DetailPageQuery -> Eff es (Either InspectionError WorkflowJournalPage)+readWorkflowJournalPage name@(WorkflowName ownerName) wid@(WorkflowId ownerId) query = do+  prepared <- prepareDetail "wf-journal" name wid query (> (0 :: Int64))+  case prepared of+    Left err -> pure (Left err)+    Right (gen, after) -> do+      observedAt <- Schema.inspectionObservedAt+      recorded <- Vector.toList <$> readStreamForward (workflowGenerationStreamName name wid gen) (StreamVersion (maybe 0 (\value -> value) after)) (fromIntegral (query.pageSize + 1))+      entries <- either (throwIO . WorkflowJournalDecodeError) pure (traverse decodeJournalEntry (take query.pageSize recorded))+      pure $ (\nextCursor -> WorkflowJournalPage gen entries nextCursor observedAt) <$> nextFor "wf-journal" query.pageSize recorded (\r -> (ownerName, ownerId, gen, case r.streamVersion of StreamVersion v -> v))++loadStepIndexPage :: (IOE :> es, Store :> es) => WorkflowName -> WorkflowId -> DetailPageQuery -> Eff es (Either InspectionError WorkflowStepPage)+loadStepIndexPage name@(WorkflowName ownerName) wid@(WorkflowId ownerId) query = do+  prepared <- prepareDetail "wf-steps" name wid query (const True)+  case prepared of+    Left err -> pure (Left err)+    Right (gen, after) -> do+      observedAt <- Schema.inspectionObservedAt+      rows <- Schema.loadStepIndexPage name wid gen after (query.pageSize + 1)+      pure $ (\nextCursor -> WorkflowStepPage gen (take query.pageSize rows) nextCursor observedAt) <$> nextFor "wf-steps" query.pageSize rows (\r -> (ownerName, ownerId, gen, r.stepName))++listWorkflowChildrenPage :: (IOE :> es, Store :> es) => WorkflowName -> WorkflowId -> OwnerPageQuery -> Eff es (Either InspectionError WorkflowChildPage)+listWorkflowChildrenPage name@(WorkflowName ownerName) wid@(WorkflowId ownerId) query =+  case prepareOwner "wf-children" name wid query of+    Left err -> pure (Left err)+    Right after -> do+      observedAt <- Schema.inspectionObservedAt+      rows <- Child.lookupChildrenOfParentPage ownerId ownerName after (query.pageSize + 1)+      pure $ (\nextCursor -> WorkflowChildPage (take query.pageSize rows) nextCursor observedAt) <$> nextFor "wf-children" query.pageSize rows (\r -> (ownerName, ownerId, (r.createdAt, r.childId, r.childName)))++listWorkflowAwakeablesPage :: (IOE :> es, Store :> es) => WorkflowName -> WorkflowId -> OwnerPageQuery -> Eff es (Either InspectionError WorkflowAwakeablePage)+listWorkflowAwakeablesPage name@(WorkflowName ownerName) wid@(WorkflowId ownerId) query =+  case prepareOwner "wf-awakeables" name wid query of+    Left err -> pure (Left err)+    Right after -> do+      observedAt <- Schema.inspectionObservedAt+      rows <- Awakeable.lookupAwakeablesOfOwnerPage ownerName ownerId after (query.pageSize + 1)+      pure $ (\nextCursor -> WorkflowAwakeablePage (take query.pageSize rows) nextCursor observedAt) <$> nextFor "wf-awakeables" query.pageSize rows (\r -> (ownerName, ownerId, (r.createdAt, r.awakeableId)))++decodeJournalEntry :: RecordedEvent -> Either Text WorkflowJournalEntry+decodeJournalEntry recorded = do+  event <- either (Left . Text.pack . show) Right (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 gen timestamp -> (continuedAsNewStepName, timestamp, Aeson.toJSON gen)+  pure+    WorkflowJournalEntry+      { eventId = case recorded.eventId of EventId value -> 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,+        event+      }++stepToJson :: WorkflowStepRow -> Value+stepToJson row = object ["step" .= row.stepName, "result" .= row.result, "recorded_at" .= row.recordedAt]++journalEntryToJson :: WorkflowJournalEntry -> Value+journalEntryToJson row = object ["event_id" .= UUID.toText row.eventId, "event_type" .= row.eventType, "stream_version" .= row.streamVersion, "global_position" .= row.globalPosition, "step_name" .= row.stepName, "recorded_at" .= row.recordedAt, "payload" .= row.payload]++pageFields :: UTCTime -> Maybe InspectionCursor -> [Value] -> [Pair]+pageFields clock next items = ["observed_at" .= clock, "items" .= items] <> maybe [] (\token -> ["next_cursor" .= token.cursorText]) next++instance ToJSON WorkflowListPage where toJSON page = object (pageFields page.observedAt page.nextCursor (map instanceToJson page.items))++instance ToJSON WorkflowJournalPage where toJSON page = object ("generation" .= page.generation : pageFields page.observedAt page.nextCursor (map journalEntryToJson page.items))++instance ToJSON WorkflowStepPage where toJSON page = object ("generation" .= page.generation : pageFields page.observedAt page.nextCursor (map stepToJson page.items))++instance ToJSON WorkflowChildPage where toJSON page = object (pageFields page.observedAt page.nextCursor (map childToJson page.items))++instance ToJSON WorkflowAwakeablePage where toJSON page = object (pageFields page.observedAt page.nextCursor (map awakeableToJson page.items))++instanceToJson :: Instance.WorkflowInstanceRow -> Value+instanceToJson row =+  object+    [ "workflow_id" .= row.workflowId,+      "workflow_name" .= row.workflowName,+      "generation" .= row.generation,+      "status" .= Instance.statusToText row.status,+      "status_class" .= statusClassToText (statusClassOf row),+      "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+    ]++childToJson :: Child.ChildRow -> Value+childToJson 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+    ]++awakeableToJson :: Awakeable.AwakeableRow -> Value+awakeableToJson 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+    ]

@@ -0,0 +1,67 @@+-- | Inspection-only query types. Kept separate to let the owning instance+-- schema consume them without depending on the rendering/page facade.+module Keiro.Workflow.Inspection.Query where++import Data.List.NonEmpty (NonEmpty)+import Data.Text (Text)+import Data.Time (UTCTime)+import GHC.Generics (Generic)+import Keiro.Inspection.Cursor+import Keiro.Workflow.Instance.Schema (WorkflowStatus (..))++data WorkflowStatusClass+  = ClassRunning+  | ClassRetrying+  | ClassSleeping+  | ClassAwaiting+  | ClassCompleted+  | ClassCancelled+  | ClassFailed+  deriving stock (Eq, Ord, Show, Bounded, Enum, Generic)++statusClassToText :: WorkflowStatusClass -> Text+statusClassToText = \case+  ClassRunning -> "running"+  ClassRetrying -> "retrying"+  ClassSleeping -> "sleeping"+  ClassAwaiting -> "awaiting"+  ClassCompleted -> "completed"+  ClassCancelled -> "cancelled"+  ClassFailed -> "failed"++statusClassFromText :: Text -> Maybe WorkflowStatusClass+statusClassFromText value = lookup value [(statusClassToText c, c) | c <- [minBound .. maxBound]]++classStatus :: WorkflowStatusClass -> WorkflowStatus+classStatus = \case+  ClassRunning -> WfRunning+  ClassRetrying -> WfRunning+  ClassSleeping -> WfSuspended+  ClassAwaiting -> WfSuspended+  ClassCompleted -> WfCompleted+  ClassCancelled -> WfCancelled+  ClassFailed -> WfFailed++data WorkflowListOrdering = ByKey | ByCreatedDesc+  deriving stock (Eq, Show, Generic)++data WorkflowListQuery = WorkflowListQuery+  { classes :: !(Maybe (NonEmpty WorkflowStatusClass)),+    workflowName :: !(Maybe Text),+    createdSince :: !(Maybe UTCTime),+    createdBefore :: !(Maybe UTCTime),+    updatedSince :: !(Maybe UTCTime),+    updatedBefore :: !(Maybe UTCTime),+    ordering :: !WorkflowListOrdering,+    pageSize :: !Int,+    cursor :: !(Maybe InspectionCursor)+  }+  deriving stock (Eq, Show, Generic)++defaultWorkflowListQuery :: WorkflowListQuery+defaultWorkflowListQuery = WorkflowListQuery Nothing Nothing Nothing Nothing Nothing Nothing ByKey 100 Nothing++-- | Validated exclusive database keys, decoded by the inspection facade.+-- The default key tuple is (workflow_id, workflow_name).+data WorkflowListPosition = AfterWorkflowKey !Text !Text | AfterWorkflowCreated !UTCTime !Text !Text+  deriving stock (Eq, Show)

@@ -1,3 +1,5 @@+{-# LANGUAGE OverloadedRecordDot #-}+ -- | Durable workflow instance summaries. -- -- The journal stream and @keiro_workflow_steps@ index remain the source of truth@@ -17,7 +19,13 @@     upsertInstanceTx,     markInstanceSuspendedAwaiting,     lookupInstance,+    lookupInstanceSession,+    WorkflowHeaderKey,+    workflowHeaderKeyBytes,+    WorkflowHeaderObservation (..),+    lookupInstanceConditionalSession,     listWorkflowInstances,+    inspectionInstanceRows,     cancelWorkflow,     claimInstance,     renewInstanceLeaseTx,@@ -32,17 +40,27 @@ where  import Contravariant.Extras (contrazip2, contrazip3, contrazip4, contrazip5)+import Data.Aeson ((.:))+import Data.Aeson qualified as Aeson+import Data.Aeson.Types (parseEither)+import Data.ByteString qualified as ByteString+import Data.ByteString.Lazy qualified as LazyByteString+import Data.Functor.Compose (Compose (..))+import Data.Functor.Contravariant ((>$<)) import Data.Int (Int32)+import Data.List (nub) import Data.List.NonEmpty qualified as NonEmpty import Data.Map.Strict qualified as Map+import Data.Maybe (catMaybes, mapMaybe) import Data.Text qualified as Text-import Data.Time (NominalDiffTime, addUTCTime)+import Data.Time (NominalDiffTime, UTCTime (..), addDays, addUTCTime, fromGregorian) import Data.UUID (UUID) import Data.UUID qualified as UUID import Effectful (Eff, IOE, (:>)) import Effectful.Exception (throwIO) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E+import Hasql.Session qualified as Session import Hasql.Statement (Statement, preparable) import Keiro.Prelude import Keiro.Workflow.Awakeable.Schema@@ -55,6 +73,7 @@     lookupChild,     reviveFailedChildTx,   )+import Keiro.Workflow.Inspection.Query import Keiro.Workflow.Instance.Schema   ( WorkflowStatus (..),     statusFromText,@@ -213,6 +232,47 @@ lookupInstance (WorkflowName name) (WorkflowId wid) =   runTransaction (Tx.statement (wid, name) lookupInstanceStmt) +-- | Read one header in the caller's session. A single SELECT already observes+-- one database snapshot; inspection hosts need no transaction-control round+-- trips around it. This reuses the ordinary lookup's statement and decoder.+lookupInstanceSession :: WorkflowName -> WorkflowId -> Session.Session (Maybe WorkflowInstanceRow)+lookupInstanceSession (WorkflowName name) (WorkflowId wid) =+  Session.statement (wid, name) lookupInstanceStmt++-- | An exact image of all header fields, produced only by the owning query.+-- It is a representation key, never a timestamp, hash or transaction id.+data WorkflowHeaderKey = WorkflowHeaderKey !ByteString.ByteString !Text !Text+  deriving stock (Eq)++instance Show WorkflowHeaderKey where+  show _ = "<workflow header key>"++workflowHeaderKeyBytes :: WorkflowHeaderKey -> Int+workflowHeaderKeyBytes (WorkflowHeaderKey bytes _ _) = ByteString.length bytes++data WorkflowHeaderObservation+  = HeaderUnchanged+  | HeaderChanged !WorkflowHeaderKey !WorkflowInstanceRow+  | HeaderImageTooLarge+  deriving stock (Eq, Show)++-- | Recheck the complete current row in one SELECT. An equal image permits+-- reuse of its pure rendering; every request still observes the database.+-- The byte limit bounds the returned image before client decoding. Missing+-- rows, changed rows and oversized images remain distinct.+lookupInstanceConditionalSession :: Int -> Maybe WorkflowHeaderKey -> WorkflowName -> WorkflowId -> Session.Session (Maybe WorkflowHeaderObservation)+lookupInstanceConditionalSession limit known (WorkflowName name) (WorkflowId wid) =+  Session.statement (image, fromIntegral limit) lookupInstanceConditionalStmt+  where+    -- A complete prior image already contains the exact lookup identity. Reuse+    -- its encoded bytes, avoiding separate serialization of both key strings.+    -- A foreign identity starts with an incomplete image and cannot compare+    -- equal to the complete database header.+    image = case known of+      Just (WorkflowHeaderKey bytes previousName previousId)+        | name == previousName && wid == previousId -> bytes+      _ -> LazyByteString.toStrict (Aeson.encode (Aeson.object ["workflow_name" Aeson..= name, "workflow_id" Aeson..= wid]))+ -- | Enumerate workflow instance summaries using stable keyset pagination. -- -- Status and name filters are exact. Rows inserted before the supplied cursor@@ -464,6 +524,95 @@     )     (D.rowMaybe instanceRowDecoder) +lookupInstanceConditionalStmt :: Statement (ByteString.ByteString, Int64) (Maybe WorkflowHeaderObservation)+{-# NOINLINE lookupInstanceConditionalStmt #-}+lookupInstanceConditionalStmt =+  preparable+    """+    WITH header AS MATERIALIZED (+      SELECT workflow_id, workflow_name, generation, status, attempts,+             last_error,+             CASE WHEN next_attempt_at IS NOT NULL THEN CASE WHEN isfinite(next_attempt_at) THEN jsonb_build_array(+               (next_attempt_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',+               extract(epoch FROM (next_attempt_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(next_attempt_at::text) END END AS next_attempt_at,+             CASE WHEN wake_after IS NOT NULL THEN CASE WHEN isfinite(wake_after) THEN jsonb_build_array(+               (wake_after AT TIME ZONE 'UTC')::date - DATE '2000-01-01',+               extract(epoch FROM (wake_after AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(wake_after::text) END END AS wake_after,+             leased_by,+             CASE WHEN lease_expires_at IS NOT NULL THEN CASE WHEN isfinite(lease_expires_at) THEN jsonb_build_array(+               (lease_expires_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',+               extract(epoch FROM (lease_expires_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(lease_expires_at::text) END END AS lease_expires_at,+             CASE WHEN isfinite(created_at) THEN jsonb_build_array((created_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',+               extract(epoch FROM (created_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(created_at::text) END AS created_at,+             CASE WHEN isfinite(updated_at) THEN jsonb_build_array((updated_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',+               extract(epoch FROM (updated_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(updated_at::text) END AS updated_at,+             CASE WHEN completed_at IS NOT NULL THEN CASE WHEN isfinite(completed_at) THEN jsonb_build_array(+               (completed_at AT TIME ZONE 'UTC')::date - DATE '2000-01-01',+               extract(epoch FROM (completed_at AT TIME ZONE 'UTC')::time)) ELSE to_jsonb(completed_at::text) END END AS completed_at+      FROM keiro.keiro_workflows+      WHERE workflow_id = ($1::jsonb ->> 'workflow_id')+        AND workflow_name = ($1::jsonb ->> 'workflow_name')+    ), image AS MATERIALIZED (+      SELECT CASE+        WHEN $2 <= 0 OR octet_length(workflow_id)::bigint+             + octet_length(workflow_name)::bigint+             + coalesce(octet_length(last_error)::bigint,0)+             + coalesce(octet_length(leased_by)::bigint,0) > $2+          THEN NULL+        ELSE to_jsonb(header)+      END AS value FROM header+    ), checked AS MATERIALIZED (+      SELECT value, value IS NULL OR octet_length(value::text) > $2 AS oversized+      FROM image+    )+    SELECT oversized,+           CASE WHEN oversized OR value = $1::jsonb THEN NULL ELSE value END+    FROM checked+    """+    ( contrazip2+        (E.param (E.nonNullable E.jsonbBytes))+        (E.param (E.nonNullable E.int8))+    )+    (D.rowMaybe (observation <$> D.column (D.nonNullable D.bool) <*> D.column (D.nullable (D.jsonbBytes decodeImage))))+  where+    observation True _ = HeaderImageTooLarge+    observation False Nothing = HeaderUnchanged+    observation False (Just (key, row)) = HeaderChanged key row+    decodeImage bytes = do+      value <- either (Left . Text.pack) Right (Aeson.eitherDecodeStrict' bytes)+      row <- either (Left . Text.pack) Right (parseEither parseHeader value)+      pure (WorkflowHeaderKey bytes (row ^. #workflowName) (row ^. #workflowId), row)+    parseHeader = Aeson.withObject "workflow header image" $ \row ->+      WorkflowInstanceRow+        <$> row .: "workflow_id"+        <*> row .: "workflow_name"+        <*> row .: "generation"+        <*> (statusFromText <$> row .: "status")+        <*> row .: "attempts"+        <*> row .: "last_error"+        <*> (row .: "next_attempt_at" >>= traverse timestampImage)+        <*> (row .: "wake_after" >>= traverse timestampImage)+        <*> row .: "leased_by"+        <*> (row .: "lease_expires_at" >>= traverse timestampImage)+        <*> (row .: "created_at" >>= timestampImage)+        <*> (row .: "updated_at" >>= timestampImage)+        <*> (row .: "completed_at" >>= traverse timestampImage)+    -- Separate days from seconds so epoch division cannot round a distant+    -- timestamp's microseconds. BC dates also avoid textual JSON date parsing.+    -- Match the existing binary decoder's Int64 microsecond interpretation+    -- of PostgreSQL's infinity sentinels, rather than changing legacy JSON.+    timestampImage (Aeson.String "infinity") = pure (sentinelTime (maxBound :: Int64))+    timestampImage (Aeson.String "-infinity") = pure (sentinelTime (minBound :: Int64))+    timestampImage value = do+      (days, seconds) <- Aeson.parseJSON value+      Aeson.withScientific+        "timestamp seconds"+        (\s -> pure (UTCTime (addDays days (fromGregorian 2000 1 1)) (fromRational (toRational s))))+        seconds+    sentinelTime microseconds =+      let (days, remainder) = divMod (toInteger microseconds) (86400 * 1000000)+       in UTCTime (addDays days (fromGregorian 2000 1 1)) (fromRational (toRational remainder / 1000000))+ listWorkflowInstancesStmt :: Statement (Maybe [Text], Maybe Text, Maybe Text, Maybe Text, Int32) [WorkflowInstanceRow] listWorkflowInstancesStmt =   preparable@@ -678,18 +827,93 @@     ((> 0) <$> D.rowsAffected)  instanceRowDecoder :: D.Row WorkflowInstanceRow-instanceRowDecoder =+instanceRowDecoder = instanceFields (D.column . D.nonNullable) (D.column . D.nullable)++-- Shared by ordinary reads and the nullable row of an inspection outer join.+-- The ordinary decoder retains the exact same columns and nullability.+instanceFields :: (Applicative f) => (forall a. D.Value a -> f a) -> (forall a. D.Value a -> f (Maybe a)) -> f WorkflowInstanceRow+instanceFields required optional =   WorkflowInstanceRow-    <$> D.column (D.nonNullable D.text)-    <*> D.column (D.nonNullable D.text)-    <*> D.column (D.nonNullable D.int4)-    <*> (statusFromText <$> D.column (D.nonNullable D.text))-    <*> D.column (D.nonNullable D.int4)-    <*> D.column (D.nullable D.text)-    <*> D.column (D.nullable D.timestamptz)-    <*> D.column (D.nullable D.timestamptz)-    <*> D.column (D.nullable D.text)-    <*> D.column (D.nullable D.timestamptz)-    <*> D.column (D.nonNullable D.timestamptz)-    <*> D.column (D.nonNullable D.timestamptz)-    <*> D.column (D.nullable D.timestamptz)+    <$> required D.text+    <*> required D.text+    <*> required D.int4+    <*> (statusFromText <$> required D.text)+    <*> required D.int4+    <*> optional D.text+    <*> optional D.timestamptz+    <*> optional D.timestamptz+    <*> optional D.text+    <*> optional D.timestamptz+    <*> required D.timestamptz+    <*> required D.timestamptz+    <*> optional D.timestamptz++-- | Inspection-only parameterized statement shapes. Absent filters are omitted+-- from SQL rather than hidden behind nullable ORs that defeat generic plans.+-- The facade validates the size and cursor before calling this primitive.+-- The outer join returns the database observation clock even for an empty page.+inspectionInstanceRows :: (Store :> es) => WorkflowListQuery -> Maybe WorkflowListPosition -> Int -> Eff es (UTCTime, [WorkflowInstanceRow])+inspectionInstanceRows query position count = do+  rows <- runTransaction (Tx.statement () (inspectionInstanceStatement query position count))+  case rows of+    (clock, _) : _ -> pure (clock, mapMaybe snd rows)+    [] -> error "inspection clock outer join returned no row"++inspectionInstanceStatement :: WorkflowListQuery -> Maybe WorkflowListPosition -> Int -> Statement () [(UTCTime, Maybe WorkflowInstanceRow)]+inspectionInstanceStatement query position count = preparable sql params decoder+  where+    number = Text.pack . show+    parameter :: Text -> E.Value a -> a -> (Int -> Text, Int, E.Params ())+    parameter template value argument = (\n -> Text.replace "?" ("$" <> number n) template, 1, const argument >$< E.param (E.nonNullable value))+    optional template value = fmap (parameter template value)+    selectedClasses = nub (maybe [] NonEmpty.toList query.classes)+    filters =+      [(const ("(" <> Text.intercalate " OR " ["(" <> classPredicate status <> ")" | status <- selectedClasses] <> ")"), 0, mempty) | query.classes /= Nothing]+        <> catMaybes+          [ optional "workflow_name = ?" E.text query.workflowName,+            optional "created_at >= ?" E.timestamptz query.createdSince,+            optional "created_at < ?" E.timestamptz query.createdBefore,+            optional "updated_at >= ?" E.timestamptz query.updatedSince,+            optional "updated_at < ?" E.timestamptz query.updatedBefore+          ]+        <> cursorParts+    cursorParts = case position of+      Nothing -> []+      Just (AfterWorkflowKey wid name) ->+        [ ( \n -> "(workflow_id, workflow_name) > ($" <> number n <> ",$" <> number (n + 1) <> ")",+            2,+            (const wid >$< E.param (E.nonNullable E.text)) <> (const name >$< E.param (E.nonNullable E.text))+          )+        ]+      Just (AfterWorkflowCreated created name wid) ->+        [ ( \n -> "(created_at, workflow_name, workflow_id) < ($" <> number n <> ",$" <> number (n + 1) <> ",$" <> number (n + 2) <> ")",+            3,+            (const created >$< E.param (E.nonNullable E.timestamptz)) <> (const name >$< E.param (E.nonNullable E.text)) <> (const wid >$< E.param (E.nonNullable E.text))+          )+        ]+    (next, predicates, filterParams) = foldl' (\(index, clauses, encoders) (render, width, encoder) -> (index + width, clauses <> [render index], encoders <> encoder)) (1, [], mempty) filters+    order = case query.ordering of ByKey -> "workflow_id, workflow_name"; ByCreatedDesc -> "created_at DESC, workflow_name DESC, workflow_id DESC"+    columns = "workflow_id, workflow_name, generation, status, attempts, last_error, next_attempt_at, wake_after, leased_by, lease_expires_at, created_at, updated_at, completed_at"+    -- Classification is inspection-only. Reuse existing status/key access;+    -- no generated column or inspection index adds work to ordinary writes.+    classPredicate = \case+      ClassRunning -> "status = 'running' AND attempts = 0"+      ClassRetrying -> "status = 'running' AND attempts <> 0"+      ClassSleeping -> "status = 'suspended' AND wake_after IS NOT NULL"+      ClassAwaiting -> "status = 'suspended' AND wake_after IS NULL"+      ClassCompleted -> "status = 'completed'"+      ClassCancelled -> "status = 'cancelled'"+      ClassFailed -> "status = 'failed'"+    sql =+      "/* keiro-inspection-page-v2 */ WITH inspection_page AS MATERIALIZED (SELECT "+        <> columns+        <> " FROM keiro.keiro_workflows WHERE "+        <> Text.intercalate " AND " ("true" : predicates)+        <> " ORDER BY "+        <> order+        <> " LIMIT $"+        <> number next+        <> ") SELECT now(), inspection_page.* FROM (VALUES (1)) AS clock(x) LEFT JOIN inspection_page ON true ORDER BY "+        <> order+    params = filterParams <> (const (fromIntegral count :: Int32) >$< E.param (E.nonNullable E.int4))+    decoder = D.rowList ((,) <$> D.column (D.nonNullable D.timestamptz) <*> getCompose (instanceFields (Compose . D.column . D.nullable) (Compose . fmap Just . D.column . D.nullable)))

@@ -27,6 +27,8 @@      -- * Read-only lookups     loadStepIndex,+    loadStepIndexPage,+    inspectionObservedAt,     lookupStepResult,     stepExists,     terminalMarkers,@@ -36,6 +38,7 @@ where  import Contravariant.Extras (contrazip2, contrazip3, contrazip4, contrazip5, contrazip6)+import Data.Functor.Contravariant ((>$<)) import Data.Int (Int32) import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map@@ -56,6 +59,25 @@ import Kiroku.Store.Effect (Store) import Kiroku.Store.Transaction (runTransaction) import "hasql-transaction" Hasql.Transaction qualified as Tx++-- | Database clock for an inspection page. This is an observation timestamp,+-- not a claim that separate ReadCommitted statements share a row snapshot.+inspectionObservedAt :: (Store :> es) => Eff es UTCTime+inspectionObservedAt = runTransaction (Tx.statement () (preparable "SELECT now()" E.noParams (D.singleRow (D.column (D.nonNullable D.timestamptz)))))++-- | Bounded step rows in immutable name order. The caller validates the size.+loadStepIndexPage :: (Store :> es) => WorkflowName -> WorkflowId -> Int -> Maybe Text -> Int -> Eff es [WorkflowStepRow]+loadStepIndexPage (WorkflowName name) (WorkflowId wid) gen after count = runTransaction (Tx.statement () statement)+  where+    prefix = (const wid >$< E.param (E.nonNullable E.text)) <> (const name >$< E.param (E.nonNullable E.text)) <> (const (fromIntegral gen :: Int32) >$< E.param (E.nonNullable E.int4))+    (cursorSql, cursorParams, limitParam) = case after of+      Nothing -> ("", mempty, "$4")+      Just key -> (" AND step_name > $4", const key >$< E.param (E.nonNullable E.text), "$5")+    statement =+      preparable+        ("SELECT workflow_id, workflow_name, generation, step_name, result, recorded_at FROM keiro.keiro_workflow_steps WHERE workflow_id=$1 AND workflow_name=$2 AND generation=$3" <> cursorSql <> " ORDER BY step_name LIMIT " <> limitParam)+        (prefix <> cursorParams <> (const (fromIntegral count :: Int32) >$< E.param (E.nonNullable E.int4)))+        (D.rowList (WorkflowStepRow <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text) <*> (fromIntegral <$> D.column (D.nonNullable D.int4)) <*> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.jsonb) <*> D.column (D.nonNullable D.timestamptz)))  -- | A row of the @keiro_workflow_steps@ index: the workflow instance and -- name, the step name, the step's JSON result, and when it was recorded. The

@@ -20,6 +20,7 @@ import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Statement (Statement, preparable)+import Keiro.Inspection.Cursor qualified as Cursor import Keiro.Prelude import Keiro.Projection   ( CatalogAsyncApplyOutcome (..),@@ -187,6 +188,17 @@      statuses <- expectStore store listProjectionGroupStatuses     length statuses `shouldBe` 251+    Right first <- expectStore store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 Nothing))+    Right second <- expectStore store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 (first ^. #next)))+    Right third <- expectStore store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 (second ^. #next)))+    (first ^. #items) <> (second ^. #items) <> (third ^. #items) `shouldBe` statuses+    third ^. #next `shouldBe` Nothing+    expectStore store listProjectionGroupStatuses `shouldReturn` statuses+    for_ [0, -1, 501] $ \size ->+      validateProjectionStatusRequest (ProjectionStatusPageRequest size Nothing) `shouldBe` Left (InvalidProjectionStatusPageSize size)+    for_ [Cursor.InspectionCursor "broken", Cursor.encodeCursor "timer-pending" ("other" :: Text)] $ \cursor ->+      validateProjectionStatusRequest (ProjectionStatusPageRequest 1 (Just cursor)) `shouldSatisfy` isLeft+     map (^. #groupId) statuses `shouldSatisfy` strictlyAscending    it "registers an additive catalog without disturbing the existing group slice" $ \store -> do

@@ -18,6 +18,7 @@ import Data.Aeson.Types (parseEither) import Data.ByteString (ByteString) import Data.ByteString qualified as ByteString+import Data.ByteString.Lazy qualified as LazyByteString import Data.Char (isDigit) import Data.IORef (IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef, writeIORef) import Data.Int (Int32)@@ -102,6 +103,7 @@   ) import Keiro.DeterministicId (deterministicIdProbes, identitySeedBytes, legacySeedBytes) import Keiro.EventStream (Terminality (..))+import Keiro.EventStream qualified as EventStream import Keiro.EventStream.Validate   ( EventStreamWarning (..),     ValidatedEventStream,@@ -141,6 +143,8 @@     delegatedFromPMCommand,   ) import Keiro.Inbox.Kafka qualified as InboxKafka+import Keiro.Inspection.Aggregate qualified as Aggregate+import Keiro.Inspection.Cursor qualified as Cursor import Keiro.Integration.Event   ( IntegrationContentType (..),     IntegrationEvent (..),@@ -197,6 +201,7 @@ import Keiro.Outbox.Schema (markOutboxFailedTx, markOutboxRejectedTx) import Keiro.Prelude import Keiro.ProcessManager+import Keiro.ProcessManager.Inspect qualified as PMInspection import Keiro.ProcessManager.Reaction qualified as Reaction import Keiro.Projection import Keiro.ReadModel@@ -205,15 +210,20 @@ import Keiro.Snapshot.Policy (shouldSnapshot, shouldSnapshotSpan) import Keiro.Stream qualified as Stream import Keiro.Subscription.Shard-  ( ShardCountMismatch (..),+  ( LeaseState (..),+    ShardCountMismatch (..),     ShardCountResizeReport (..),     ShardLease (..),     ShardResizeActiveLeases (..),     WorkerId (..),     ensureShards,     fairShareTarget,+    leaseAgeAt,+    leaseStateAt,+    listShardedSubscriptions,     resizeShardCount,     resizeShardCountTx,+    shardOwnershipView,   ) import Keiro.Subscription.Shard.Schema   ( claimShardsTx,@@ -426,7 +436,7 @@ import Shibuya.Adapter (Adapter (..)) import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..), RetryDelay (..), deadLetterCodeText, deadLetterReasonCode, deadLetterReasonDetail, renderDeadLetterReason) import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..)) import Shibuya.Core.Types (Envelope (..)) import Streamly.Data.Stream qualified as Streamly import System.Exit (ExitCode (..))@@ -435,11 +445,13 @@ import Test.Hspec import VersionedRebuildSpec qualified import VersionedTargetPostgresSpec qualified+import WorkflowInspectionSpec qualified import WorkflowReplayCompatibilitySpec qualified import "hasql-transaction" Hasql.Transaction qualified as Tx  main :: IO () main = withMigratedSuite $ \fixture -> hspec $ do+  WorkflowInspectionSpec.spec fixture   CatalogSpec.spec   PreimageSpec.spec   CatalogEvolutionSpec.spec fixture@@ -2438,6 +2450,57 @@         for_ callbackRecordeds $ \recorded ->           recorded ^. #metadata `shouldSatisfy` hasMarker +  describe "Keiro.Inspection.Aggregate" $ around (withFreshStore fixture) $ do+    it "inspects the command fold and snapshot, while deleted streams have no folded state" $ \store -> do+      let codec = defaultStateCodec @SnapshotCounterRegs @CounterState 1+          inspector = Aggregate.aggregateInspector "counter" snapshotCounterEventStream (Stream.categoryUnsafe "inspectioncounter") (Aggregate.displayViaStateCodec codec)+          target = Stream.entityStream (Stream.categoryUnsafe "inspectioncounter") "one"+      forM_ [2, 3, 4] $ \amount -> do+        Right (Right _) <- Store.runStoreIO store (runCommand inspectionRunCommandOptions snapshotCounterEventStream target (Add amount))+        pure ()+      Right (Right view) <- Store.runStoreIO store (Aggregate.inspectAggregate inspector "one")+      Right (Right full) <- Store.runStoreIO store (hydrateFull inspectionRunCommandOptions (unvalidated snapshotCounterEventStream) target)+      (view ^. #state) `shouldBe` Just ((codec ^. #encode) (full ^. #state, full ^. #registers))+      (view ^. #streamVersion) `shouldBe` (full ^. #streamVersion)+      let mappedDefinition :: SnapshotCounterEventStream+          mappedDefinition = (unvalidated snapshotCounterEventStream) & #resolveStreamName .~ (\typed -> case Stream.streamName typed of StreamName raw -> StreamName ("inspectioncounter-" <> raw))+          mappedStream = mkEventStreamOrThrow "inspection-mapped-counter" mappedDefinition+          auditTarget = ReplayAudit.AuditTarget mappedStream "inspectioncounter" (\(StreamName physical) -> stream <$> Text.stripPrefix "inspectioncounter-" physical)+      Right mappedInspector <- pure (Aggregate.inspectorFromAuditTarget "counter" auditTarget (Aggregate.displayViaStateCodec codec))+      Right (Right mappedView) <- Store.runStoreIO store (Aggregate.inspectAggregate mappedInspector "one")+      (mappedView ^. #state) `shouldBe` (view ^. #state)+      (mappedView ^. #streamName) `shouldBe` StreamName "inspectioncounter-one"+      Aggregate.aggregateIdOf mappedInspector (StreamName "inspectioncounter-one") `shouldBe` Just "one"+      (view ^. #hydration) `shouldBe` Just (FromSnapshot (StreamVersion 2))+      fmap (^. #compatibility) (view ^. #snapshot . #stored) `shouldBe` Just Aggregate.Compatible+      Right _ <- Store.runStoreIO store (Store.softDeleteStream (StreamName "inspectioncounter-one"))+      Right (Right deleted) <- Store.runStoreIO store (Aggregate.inspectAggregate inspector "one")+      (deleted ^. #state) `shouldBe` Nothing+      (deleted ^. #hydration) `shouldBe` Nothing+      (deleted ^. #streamVersion) `shouldBe` StreamVersion 3+      Aggregate.aggregateStream inspector " " `shouldBe` Left (Aggregate.AggregateIdRejected " ")+      missing <- Store.runStoreIO store (Aggregate.inspectAggregate inspector "missing")+      missing `shouldBe` Right (Left (Aggregate.AggregateNotFound (StreamName "inspectioncounter-missing")))++    it "advances one raw page through rejected rows without losing the look-ahead row" $ \store -> do+      let base = Aggregate.aggregateInspector "counter" counterEventStream (Stream.categoryUnsafe "inspectionpage") (\state _ -> Aeson.String (Text.pack (show state)))+          inspector = base & #mkStream .~ (\name@(StreamName raw) -> if Text.isSuffixOf "keep" raw then (base ^. #mkStream) name else Nothing)+      forM_ ["a-reject", "b-reject", "c-keep"] $ \suffix -> do+        Right (Right _) <- Store.runStoreIO store (runCommand inspectionRunCommandOptions counterEventStream (Stream.entityStream (Stream.categoryUnsafe "inspectionpage") suffix) (Add 1))+        pure ()+      Right (Right first) <- Store.runStoreIO store (Aggregate.listAggregates inspector (Aggregate.AggregateListRequest 2 Nothing))+      (first ^. #items) `shouldBe` []+      (first ^. #skipped) `shouldBe` 2+      (first ^. #next) `shouldSatisfy` isJust+      Right (Right second) <- Store.runStoreIO store (Aggregate.listAggregates inspector (Aggregate.AggregateListRequest 2 (first ^. #next)))+      map (^. #aggregateId) (second ^. #items) `shouldBe` ["c-keep"]+      (second ^. #next) `shouldBe` Nothing+      let other = inspector & #name .~ ("another-counter" :: Text)+      Right foreignPage <- Store.runStoreIO store (Aggregate.listAggregates other (Aggregate.AggregateListRequest 2 (first ^. #next)))+      foreignPage `shouldSatisfy` (\case Left (Aggregate.InvalidCursor _) -> True; _ -> False)+      Right malformed <- Store.runStoreIO store (Aggregate.listAggregates inspector (Aggregate.AggregateListRequest 2 (Just (Cursor.InspectionCursor "bad"))))+      malformed `shouldSatisfy` (\case Left (Aggregate.InvalidCursor _) -> True; _ -> False)+   describe "Keiro.Snapshot" $ around (withFreshStore fixture) $ do     it "reports an ErrorCall when strict encoding reaches an empty register slot" $ \_storeHandle -> do       result <-@@ -2550,6 +2613,55 @@       lookup "keiro.snapshot.encode.failures" scalars `shouldBe` Just (IntNumber 1)       lookup "keiro.snapshot.write.failures" scalars `shouldBe` Nothing +    it "reports snapshot and corrupt-seed sources without changing the folded state" $ \storeHandle -> do+      let target = stream "inspection-hydration-source" :: Stream SnapshotCounterEventStream+          options = inspectionRunCommandOptions+          definition = unvalidated snapshotCounterEventStream+          codec = defaultStateCodec @SnapshotCounterRegs @CounterState 1+          image hydrated = (hydrated ^. #streamVersion, (codec ^. #encode) (hydrated ^. #state, hydrated ^. #registers))+      forM_ [2, 3, 4] $ \amount -> do+        Right (Right _) <- Store.runStoreIO storeHandle (runCommand options snapshotCounterEventStream target (Add amount))+        pure ()+      Right (Right (fromSeed, source)) <- Store.runStoreIO storeHandle (hydrateWithSource options definition target)+      Right (Right ordinary) <- Store.runStoreIO storeHandle (hydrate options definition target)+      Right (Right full) <- Store.runStoreIO storeHandle (hydrateFull options definition target)+      source `shouldBe` FromSnapshot (StreamVersion 2)+      image fromSeed `shouldBe` image full+      image ordinary `shouldBe` image full+      Right () <-+        Store.runStoreIO storeHandle $+          Store.runTransaction $+            Tx.statement ("inspection-hydration-source", Aeson.String "bad") corruptSnapshotStateStmt+      Right (Right (fallback, fallbackSource)) <- Store.runStoreIO storeHandle (hydrateWithSource options definition target)+      fallbackSource `shouldSatisfy` (\case FullReplay (SnapshotMissed (SnapshotDecodeFailed _)) -> True; _ -> False)+      image fallback `shouldBe` image full++    it "distinguishes absent snapshot streams and no codec during inspection hydration" $ \storeHandle -> do+      Right (Right (_, noCodec)) <-+        Store.runStoreIO storeHandle $+          hydrateWithSource inspectionRunCommandOptions (unvalidated counterEventStream) (stream "inspection-no-codec")+      noCodec `shouldBe` FullReplay NoStateCodec+      Right (Right (_, absent)) <-+        Store.runStoreIO storeHandle $+          hydrateWithSource inspectionRunCommandOptions (unvalidated snapshotCounterEventStream) (stream "inspection-absent-seed")+      absent `shouldBe` FullReplay (SnapshotMissed SnapshotNoStream)+      (inspectionRunCommandOptions ^. #seedVerifySampleRate) `shouldBe` 0+      (inspectionRunCommandOptions ^. #verifyReplayOnAppend) `shouldBe` False++    it "observes absent and present snapshots with a database clock" $ \storeHandle -> do+      let target = stream "inspection-snapshot-clock" :: Stream SnapshotCounterEventStream+      Right (Right _) <- Store.runStoreIO storeHandle (runCommand inspectionRunCommandOptions snapshotCounterEventStream target (Add 2))+      Right (Just sid) <- Store.runStoreIO storeHandle (Store.lookupStreamId (StreamName "inspection-snapshot-clock"))+      Right absent <- Store.runStoreIO storeHandle (observeSnapshotRow sid)+      (absent ^. #row) `shouldBe` Nothing+      Right (Right _) <- Store.runStoreIO storeHandle (runCommand inspectionRunCommandOptions snapshotCounterEventStream target (Add 3))+      Right observed <- Store.runStoreIO storeHandle (observeSnapshotRow sid)+      case (observed ^. #row) of+        Nothing -> expectationFailure "expected snapshot observation"+        Just row -> do+          (row ^. #streamVersion) `shouldBe` StreamVersion 2+          (observed ^. #observedAt) `shouldSatisfy` (>= (row ^. #updatedAt))+     it "hydrates from snapshot and replays only the tail" $ \storeHandle -> do       (exporter, metricsRef) <- inMemoryMetricExporter       (provider, _env) <-@@ -5205,6 +5317,198 @@           Prelude.sum persistedCounts `shouldBe` fanOut    describe "Keiro.ProcessManager" $ around (withFreshResourceStore fixture) $ do+    it "discovers sorted lookup names without evaluating instance callbacks" $ \_ -> do+      let unused = error "discovery invoked an instance callback"+          legacy =+            (PMInspection.processManagerInspector pmSnapshotProcessManager (categoryUnsafe "pm:counter") (\_ _ -> unused))+              { PMInspection.streamFor = \_ -> unused,+                PMInspection.correlationOf = \_ -> unused,+                PMInspection.stateLabel = \_ -> unused,+                PMInspection.enabledInputs = \_ -> unused+              }+          definition :: SnapshotPolicy (CounterState, RegFile SnapshotCounterRegs) -> ValidatedSnapshotCounterEventStream+          definition policy = mkEventStreamOrThrow "discovery" (pmSnapshotCounterEventStreamDef {EventStream.snapshotPolicy = policy, EventStream.resolveStreamName = \_ -> unused})+          domain = PMInspection.domainProcessManagerInspector domainProcessManager (categoryUnsafe "domain") (\_ _ -> unused)+          reactive = PMInspection.reactiveProcessManagerInspector counterReactionManager (categoryUnsafe "reactive") (\_ _ -> unused)+          catalog =+            Map.fromList+              [ ("z-alias", PMInspection.SomeProcessManagerInspector legacy),+                ("a-custom", PMInspection.SomeProcessManagerInspector (legacy {PMInspection.eventStream = definition (Custom (\_ _ _ -> unused))})),+                ("b-never", PMInspection.SomeProcessManagerInspector (legacy {PMInspection.eventStream = definition Never})),+                ("c-terminal", PMInspection.SomeProcessManagerInspector (legacy {PMInspection.eventStream = definition OnTerminal})),+                ("domain", PMInspection.SomeProcessManagerInspector domain),+                ("reactive", PMInspection.SomeProcessManagerInspector reactive)+              ]+          summaries = PMInspection.listProcessManagerTypes catalog+      PMInspection.listProcessManagerTypes Map.empty `shouldBe` []+      map (^. #name) summaries `shouldBe` ["a-custom", "b-never", "c-terminal", "domain", "reactive", "z-alias"]+      map (^. #snapshotPolicy) summaries `shouldBe` ["custom", "never", "on_terminal", "never", "never", "every:2"]+      map (^. #snapshotsConfigured) summaries `shouldBe` [True, True, True, False, False, True]+      map (^. #timersConfigured) summaries `shouldBe` replicate 6 Nothing+      map (^. #category) summaries `shouldBe` ["pm:counter", "pm:counter", "pm:counter", "domain", "reactive", "pm:counter"]+      encoded <- evaluate (LazyByteString.length (Aeson.encode (map PMInspection.processManagerTypeJson summaries)))+      encoded `shouldSatisfy` (> 0)+      forM_ (summaries <> [PMInspection.ProcessManagerTypeSummary "known-timers" "timer-metadata" False "never" (Just True), PMInspection.ProcessManagerTypeSummary "no-timers" "timer-metadata" False "never" (Just False), PMInspection.ProcessManagerTypeSummary "escaped-\"\\\n界" "category-界" True "custom" Nothing]) $ \summary ->+        Aeson.eitherDecode (Aeson.encode summary) `shouldBe` Right (PMInspection.processManagerTypeJson summary)+      Aeson.eitherDecode (Aeson.encode summaries) `shouldBe` Right (Aeson.toJSON (map PMInspection.processManagerTypeJson summaries))+    it "instance view equals the runtime's own rebuild" $ \(storeHandle, StoreRunner runner) -> do+      let options = defaultRunCommandOptions & #inspectionProvenance .~ CaptureInspectionProvenance+          definition = unvalidated pmSnapshotCounterEventStream+          Just codec = definition ^. #stateCodec+          display state registers = codec ^. #encode $ (state, registers)+          inspector = PMInspection.processManagerInspector pmSnapshotProcessManager (categoryUnsafe "pm:counter") display+          inspect value correlation = runner (PMInspection.inspectProcessManagerInstance value (PMInspection.InstanceViewRequest 20) correlation)+      forM_ [1 .. 3] $ \n -> do+        let source = recordedFromEventId (EventId (UUID.fromWords 0 0 0 n)) (CounterAdded (fromIntegral n))+        Right (Right _) <- runner (runProcessManagerOnce options pmSnapshotProcessManager source (CounterAdded (fromIntegral n)))+        pure ()+      Right (Right rebuilt) <- runner (hydrateFull inspectionRunCommandOptions definition ((pmSnapshotProcessManager ^. #streamFor) "order-1"))+      Right (Right (Just view)) <- inspect inspector "order-1"+      view ^. #state `shouldBe` Just (display (rebuilt ^. #state) (rebuilt ^. #registers))+      view ^. #streamVersion `shouldBe` rebuilt ^. #streamVersion+      view ^. #hydration `shouldBe` Just (FromSnapshot (StreamVersion 2))+      view ^. #waitingFor `shouldSatisfy` (not . null)+      let noCodec = inspector & #eventStream .~ mkEventStreamOrThrow "inspection-no-codec" definition {stateCodec = Nothing, snapshotPolicy = Never}+      Right (Right (Just replayed)) <- inspect noCodec "order-1"+      replayed ^. #state `shouldBe` view ^. #state+      replayed ^. #hydration `shouldBe` Just (FullReplay NoStateCodec)+      Right absent <- inspect inspector "missing"+      absent `shouldBe` Right Nothing+      Right _ <- Store.runStoreIO storeHandle (Store.softDeleteStream (StreamName "pm:counter-snap-order-1"))+      Right (Right (Just deleted)) <- inspect inspector "order-1"+      deleted ^. #state `shouldBe` Nothing+      deleted ^. #hydration `shouldBe` Nothing+      deleted ^. #deletedAt `shouldSatisfy` isJust++    it "advances manager listings across rejected raw rows" $ \(_, StoreRunner runner) -> do+      let inspector =+            (PMInspection.processManagerInspector counterProcessManager (categoryUnsafe "pm:counter") (\state _ -> Aeson.toJSON (show state)))+              & #correlationOf+              .~ (\(StreamName raw) -> if raw == "pm:counter-c" then Just ("c" :: Text) else Nothing)+          readPage after = runner (PMInspection.listProcessManagerInstances inspector (PMInspection.InstanceListRequest 2 after))+      forM_ ["a", "b", "c"] $ \key -> do+        Right (Right _) <- runner (runCommand defaultRunCommandOptions counterEventStream (stream ("pm:counter-" <> key)) (Add 1))+        pure ()+      Right (Right first) <- readPage Nothing+      first ^. #items `shouldBe` []+      first ^. #skipped `shouldBe` 2+      first ^. #next `shouldSatisfy` isJust+      Right (Right second) <- readPage (first ^. #next)+      map (^. #correlationId) (second ^. #items) `shouldBe` ["c"]+      second ^. #next `shouldBe` Nothing++    it "inspects modern reaction state without inventing input-dependent dispatch provenance" $ \(storeHandle, StoreRunner runner) -> do+      let manager = counterReactionManager {Reaction.name = "inspection-reaction", Reaction.streamFor = \cid -> stream ("inspectionreaction-" <> cid)}+          inspector = PMInspection.reactiveProcessManagerInspector manager (categoryUnsafe "inspectionreaction") (\state _ -> Aeson.toJSON (show state))+          source = recordedFromEventId (EventId sampleUuid) (CounterAdded 5)+          target = stream "inspection-target-dependent" :: Stream CounterCommand+          timer = counterTimerRequest & #processManagerName .~ "inspection-reaction" & #correlationId .~ "one"+          legacy = PMInspection.ReactionProvenance (source ^. #eventId) (source ^. #globalPosition) (source ^. #originalStreamId) [(0, StreamName "fictional-target")] [] "legacy-positional-v1" False+          options = defaultRunCommandOptions & #metadata ?~ object ["keiro.pm" Aeson..= PMInspection.encodeReactionProvenance legacy]+          plan = Reaction.AdvanceReaction (Add 5) [] [Reaction.FollowDispatch (PMCommand target (Add 2)), Reaction.FollowDispatch (PMCommand target (Add 3)), Reaction.FollowSchedule Reaction.Once timer]+          inspect cid = runner (PMInspection.inspectProcessManagerInstance inspector (PMInspection.InstanceViewRequest 20) cid)+      Right (Right _) <- runner (Reaction.runReactiveProcessManagerOnce options manager source ("one", plan))+      Right targetEvents <- runner (Store.readStreamForward (StreamName "inspection-target-dependent") (StreamVersion 0) 10)+      Vector.length targetEvents `shouldBe` 2+      Right (Right rebuilt) <- runner (hydrateFull inspectionRunCommandOptions (unvalidated multiCounterEventStream) ((manager ^. #streamFor) "one"))+      Right (Right (Just view)) <- inspect "one"+      view ^. #state `shouldBe` Just (Aeson.toJSON (show (rebuilt ^. #state)))+      view ^. #streamVersion `shouldBe` StreamVersion 2+      view ^. #waitingFor `shouldSatisfy` (not . null)+      length (view ^. #pendingTimers . #timers) `shouldBe` 1+      Just last <- pure (view ^. #lastReaction)+      last ^. #streamVersion `shouldBe` StreamVersion 2+      last ^. #provenance `shouldBe` Nothing+      last ^. #sourceStreamName `shouldBe` Nothing+      last ^. #dispatchedCommands `shouldBe` []+      -- Even a valid legacy-looking frame on the latest persisted event is ignored.+      Right events <- runner (Store.readStreamForward (StreamName "inspectionreaction-one") (StreamVersion 0) 10)+      (Vector.last events ^. #metadata) `shouldSatisfy` isJust+      let json = PMInspection.processManagerViewJson view+      case json of+        Aeson.Object fields -> case KeyMap.lookup "last_reaction" fields of+          Just (Aeson.Object reaction) -> do+            KeyMap.lookup "provenance_status" reaction `shouldBe` Just (Aeson.String "unavailable")+            KeyMap.lookup "provenance" reaction `shouldBe` Just Aeson.Null+          other -> expectationFailure (show other)+        other -> expectationFailure (show other)+      Right (Right _) <- runner (Reaction.runReactiveProcessManagerOnce defaultRunCommandOptions manager (recordedFromEventId (EventId sampleUuid2) (CounterAdded 1)) ("one", Reaction.NoAdvance []))+      Right (Right (Just unchanged)) <- inspect "one"+      unchanged ^. #lastReaction `shouldBe` view ^. #lastReaction+      Right (Right _) <- runner (Reaction.runReactiveProcessManagerOnce defaultRunCommandOptions manager (recordedFromEventId (EventId sampleUuid3) (CounterAdded 1)) ("two", Reaction.AdvanceReaction (Add 1) [] []))+      Right (Right page) <- runner (PMInspection.listProcessManagerInstances inspector (PMInspection.InstanceListRequest 1 Nothing))+      map (^. #correlationId) (page ^. #items) `shouldBe` ["one"]+      Right (Right next) <- runner (PMInspection.listProcessManagerInstances inspector (PMInspection.InstanceListRequest 1 (page ^. #next)))+      map (^. #correlationId) (next ^. #items) `shouldBe` ["two"]+      let silentManager = silentReactionManager {Reaction.name = "inspection-silent", Reaction.streamFor = \cid -> stream ("inspectionsilent-" <> cid)}+          silentInspector = PMInspection.reactiveProcessManagerInspector silentManager (categoryUnsafe "inspectionsilent") (\state _ -> Aeson.toJSON (show state))+      Right (Right _) <- runner (Reaction.runReactiveProcessManagerOnce defaultRunCommandOptions silentManager source ("one", Reaction.AdvanceReaction NoOpSilently [] []))+      Right silent <- runner (PMInspection.inspectProcessManagerInstance silentInspector (PMInspection.InstanceViewRequest 20) "one")+      silent `shouldBe` Right Nothing+      Right absent <- inspect "missing"+      absent `shouldBe` Right Nothing+      Right _ <- Store.runStoreIO storeHandle (Store.softDeleteStream (StreamName "inspectionreaction-one"))+      Right (Right (Just deleted)) <- inspect "one"+      deleted ^. #state `shouldBe` Nothing+      deleted ^. #lastReaction `shouldBe` Nothing++    it "records source provenance for explicit inspection capture" $ \(_, StoreRunner runner) -> do+      let source = recordedFromEventId (EventId sampleUuid) (CounterAdded 9)+      let options = defaultRunCommandOptions & #inspectionProvenance .~ CaptureInspectionProvenance & #metadata ?~ Aeson.object ["actor" Aeson..= ("ops" :: Text)]+      Right (Right _) <- runner (runProcessManagerOnce options counterProcessManager source (CounterAdded 9))+      Right events <- runner (Store.readStreamForward (StreamName "pm:counter-order-1") (StreamVersion 0) 1)+      Vector.length events `shouldBe` 1+      case Vector.head events ^. #metadata of+        Just (Aeson.Object fields) -> do+          KeyMap.lookup "actor" fields `shouldBe` Just (Aeson.String "ops")+          Just frame <- pure (KeyMap.lookup "keiro.pm" fields)+          Just provenance <- pure (PMInspection.decodeReactionProvenance frame)+          (provenance ^. #sourceEventId) `shouldBe` (source ^. #eventId)+          (provenance ^. #sourceGlobalPosition) `shouldBe` (source ^. #globalPosition)+          (provenance ^. #sourceStreamId) `shouldBe` (source ^. #originalStreamId)+          (provenance ^. #dispatched) `shouldBe` [(0, StreamName "counter-target-order-1")]+          (provenance ^. #truncated) `shouldBe` False+          LazyByteString.length (Aeson.encode frame) `shouldSatisfy` (<= 16384)+          let many = provenance & #dispatched .~ [(i, StreamName ("target-" <> Text.pack (show i))) | i <- [0 .. 99]] & #timerIds .~ replicate 100 (TimerId sampleUuid)+          Just prefix <- pure (PMInspection.decodeReactionProvenance (PMInspection.encodeReactionProvenance many))+          length (prefix ^. #dispatched) `shouldBe` 64+          length (prefix ^. #timerIds) `shouldBe` 64+          (prefix ^. #truncated) `shouldBe` True+          let escaped = many & #dispatched .~ [(i, StreamName (Text.replicate 400 "\"")) | i <- [0 .. 63]]+              limited = PMInspection.encodeReactionProvenance escaped+          Just sourceOnly <- pure (PMInspection.decodeReactionProvenance limited)+          (sourceOnly ^. #dispatched) `shouldBe` []+          (sourceOnly ^. #truncated) `shouldBe` True+          LazyByteString.length (Aeson.encode limited) `shouldSatisfy` (<= 16384)+          case frame of+            Aeson.Object body -> do+              PMInspection.decodeReactionProvenance (Aeson.Object (KeyMap.insert "family" (Aeson.String "other-family") body)) `shouldBe` Nothing+              let tooLarge = Aeson.toJSON [Aeson.object ["i" Aeson..= (0 :: Int), "target" Aeson..= Text.replicate 20000 "x"]]+              PMInspection.decodeReactionProvenance (Aeson.Object (KeyMap.insert "dispatched" tooLarge body)) `shouldBe` Nothing+            _ -> expectationFailure "expected provenance object"+        _ -> expectationFailure "expected source provenance metadata"+      (Vector.head events ^. #eventId) `shouldBe` deterministicCommandId "counter-pm" "order-1" (source ^. #eventId) (-1)+      let inspector = PMInspection.processManagerInspector counterProcessManager (categoryUnsafe "pm:counter") (\state _ -> Aeson.toJSON (show state))+      Right (Right (Just observed)) <- runner (PMInspection.inspectProcessManagerInstance inspector (PMInspection.InstanceViewRequest 20) "order-1")+      Just last <- pure (observed ^. #lastReaction)+      map (^. #witness) (last ^. #dispatchedCommands) `shouldBe` [PMInspection.EventPresent]+      map (^. #eventId) (last ^. #dispatchedCommands) `shouldBe` [deterministicCommandId "counter-pm" "order-1" (source ^. #eventId) 0]+      Right _ <- runner (Store.softDeleteStream (StreamName "counter-target-order-1"))+      Right (Right (Just afterDelete)) <- runner (PMInspection.inspectProcessManagerInstance inspector (PMInspection.InstanceViewRequest 20) "order-1")+      Just missingTarget <- pure (afterDelete ^. #lastReaction)+      map (^. #witness) (missingTarget ^. #dispatchedCommands) `shouldBe` [PMInspection.NotObserved]+      let another = recordedFromEventId (EventId sampleUuid2) (CounterAdded 9)+      Right invalid <- runner (runProcessManagerOnce (options & #metadata ?~ Aeson.String "invalid") counterProcessManager another (CounterAdded 9))+      invalid `shouldSatisfy` (\case Left (EncodeFailed (NonObjectCallerMetadata _)) -> True; _ -> False)+      Right (Right _) <- runner (runDomainProcessManagerOnce options domainProcessManager source (DomainDispatchInput "capture" [CoordinatorAccept 9]))+      Right domainEvents <- runner (Store.readStreamForward (StreamName "domain-pm:capture") (StreamVersion 0) 1)+      case Vector.head domainEvents ^. #metadata of+        Just (Aeson.Object fields) -> do+          Just domainFrame <- pure (KeyMap.lookup "keiro.pm" fields)+          Just provenance <- pure (PMInspection.decodeReactionProvenance domainFrame)+          (provenance ^. #dispatched) `shouldBe` [(0, StreamName "domain-pm-target:capture:0")]+        _ -> expectationFailure "expected domain manager provenance"+     it "advances manager state, emits a deterministic target command once, and schedules a timer" $ \(_storeHandle, StoreRunner _runner) -> do       let sourceEvent = recordedFromEventId (EventId sampleUuid) (CounterAdded 9)       result <-@@ -5229,6 +5533,7 @@         _runner $           Store.readStreamForward (StreamName "counter-target-order-1") (StreamVersion 0) 10       Vector.length managerEvents `shouldBe` 1+      (Vector.head managerEvents ^. #metadata) `shouldBe` Just (Aeson.object ["schemaVersion" Aeson..= (1 :: Int)])       Vector.length targetEvents `shouldBe` 1       timer <-         _runner $@@ -6625,6 +6930,47 @@         noRecovery `shouldBe` Right 0    describe "Keiro.Timer" $ around (withFreshStore fixture) $ do+    it "pages pending timers in due order without mutating rows" $ \storeHandle -> do+      let requests =+            [ counterTimerRequest+                & #timerId+                .~ TimerId (UUID.fromWords 0 0 0 n)+                & #processManagerName+                .~ manager+                & #correlationId+                .~ correlation+            | (n, manager, correlation) <- [(1, "a", "x"), (2, "b", "x"), (3, "a", "y"), (4, "b", "y")]+            ]+          readPage timerFilter size after = Store.runStoreIO storeHandle (Timer.findPendingTimers timerFilter (Timer.PendingTimerPageRequest size after))+          ids page = map (^. #timer . #timerId) (page ^. #timers)+          snapshot = Store.runStoreIO storeHandle (Store.runTransaction (Tx.statement () timerReadSnapshotStmt))+      Right () <- Store.runStoreIO storeHandle $ Store.runTransaction $ traverse_ scheduleTimerTx requests+      Right before <- snapshot+      Right (Right first) <- readPage Timer.anyPendingTimer 2 Nothing+      ids first `shouldBe` map (^. #timerId) (take 2 requests)+      Right (Right second) <- readPage Timer.anyPendingTimer 2 (first ^. #next)+      ids second `shouldBe` map (^. #timerId) (drop 2 requests)+      second ^. #next `shouldBe` Nothing+      forM_ [(Timer.PendingTimerFilter (Just "a") Nothing, [0, 2]), (Timer.PendingTimerFilter Nothing (Just "y"), [2, 3]), (Timer.PendingTimerFilter (Just "b") (Just "x"), [1])] $ \(timerFilter, indices) -> do+        Right (Right page) <- readPage timerFilter 10 Nothing+        ids page `shouldBe` [requests !! index ^. #timerId | index <- indices]+      snapshot `shouldReturn` Right before+      Right (Left foreignFilter) <- readPage (Timer.PendingTimerFilter (Just "a") Nothing) 2 (first ^. #next)+      foreignFilter `shouldSatisfy` (\case Timer.InvalidPendingTimerCursor _ -> True; _ -> False)+      Right (Just claimed) <- Store.runStoreIO storeHandle (claimDueTimer dueTimerTime)+      Right True <- Store.runStoreIO storeHandle (markTimerFired (claimed ^. #timerId) (EventId sampleUuid2))+      Right True <- Store.runStoreIO storeHandle (deadLetterTimer (requests !! 1 ^. #timerId) "excluded")+      Right (Right remaining) <- readPage Timer.anyPendingTimer 10 Nothing+      ids remaining `shouldBe` map (^. #timerId) (drop 2 requests)++    it "rejects invalid pending timer requests before database work" $ \_ -> do+      forM_ [0, -1, 501] $ \size ->+        Timer.validatePendingTimerRequest Timer.anyPendingTimer (Timer.PendingTimerPageRequest size Nothing)+          `shouldBe` Left (Timer.InvalidPendingTimerPageSize size)+      forM_ [Cursor.InspectionCursor "not-a-cursor", Cursor.encodeCursor "wf-key" ("key" :: Text)] $ \cursor ->+        Timer.validatePendingTimerRequest Timer.anyPendingTimer (Timer.PendingTimerPageRequest 1 (Just cursor))+          `shouldSatisfy` (\case Left (Timer.InvalidPendingTimerCursor _) -> True; _ -> False)+     it "guards dead resume ownership and retains attempts when parked" $ \storeHandle -> do       let tid = counterTimerRequest ^. #timerId           request = DeadTimerClaimRequest tid (counterTimerRequest ^. #processManagerName) "deferred" 2 60@@ -11857,6 +12203,28 @@         t0 = UTCTime (ModifiedJulianDay 60000) (secondsToDiffTime 0)         tExpired = addUTCTime 60 t0 -- past A's 30 s lease         shardOpts = defaultShardedWorkerOptions (Category (CategoryName "orders")) 4+    it "observes ownership failover and lease ages without changing workers" $ \store -> do+      now <- getCurrentTime+      Right _ <- Store.runStoreIO store $ Store.runTransaction $ do+        ensureShardRows subName 4+        ensureShardRows (SubscriptionName "other-shard") 2+        claimShardsTx subName wA 4 now ttl+      Right first <- Store.runStoreIO store (shardOwnershipView subName)+      map (^. #owner) (first ^. #buckets) `shouldBe` replicate 4 (Just wA)+      map (leaseStateAt (first ^. #observedAt)) (first ^. #buckets) `shouldBe` replicate 4 LeaseLive+      map (leaseAgeAt (first ^. #observedAt)) (first ^. #buckets) `shouldBe` replicate 4 (Just (diffUTCTime (first ^. #observedAt) now))+      Right _ <- Store.runStoreIO store $ Store.runTransaction $ claimShardsTx subName wB 4 (addUTCTime 60 now) ttl+      Right second <- Store.runStoreIO store (shardOwnershipView subName)+      map (^. #owner) (second ^. #buckets) `shouldBe` replicate 4 (Just wB)+      Right listing <- Store.runStoreIO store listShardedSubscriptions+      let summaries = listing ^. #subscriptions+      map (^. #bucketsTotal) summaries `shouldBe` [4, 2]+      map (^. #bucketsLive) summaries `shouldBe` [4, 0]+      map (^. #liveOwners) summaries `shouldBe` [1, 0]+      Right () <- Store.runStoreIO store $ Store.runTransaction $ releaseShardsTx subName wB [0]+      Right released <- Store.runStoreIO store (shardOwnershipView subName)+      map (leaseAgeAt (released ^. #observedAt)) (take 1 (released ^. #buckets)) `shouldBe` [Nothing]+     it "validates sharded worker options before startup" $ \_store -> do       shouldBeRight_ (mkShardedWorkerOptions shardOpts)       mkShardedWorkerOptions (shardOpts & #shardCount .~ (fromIntegral (maxBound :: Int32) + 1))@@ -18492,7 +18860,8 @@       Ingested         { envelope = routerTestEnvelope message,           ack = AckHandle (\decision -> liftIO (modifyIORef' decisionsRef (<> [decision]))),-          lease = Nothing+          lease = Nothing,+          deliveryStatus = pure DeliveryCurrent         }  routerTestEnvelope :: msg -> Envelope msg

@@ -0,0 +1,259 @@+{-# LANGUAGE OverloadedRecordDot #-}++module WorkflowInspectionSpec (spec) where++import Control.Monad (forM, forM_)+import Data.Aeson (Value (..))+import Data.List (nub, sort)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding (encodeUtf8)+import Data.Time (UTCTime)+import Data.UUID qualified as UUID+import Effectful (Eff, IOE)+import Effectful.Error.Static (Error)+import Hasql.Pool qualified as Pool+import Keiro.Inspection.Cursor+import Keiro.Test.Postgres+import Keiro.Workflow (appendJournalEntry, continueAsNew, runWorkflow)+import Keiro.Workflow.Awakeable.Schema qualified as Awakeable+import Keiro.Workflow.Child.Schema qualified as Child+import Keiro.Workflow.Inspection+import Keiro.Workflow.Instance qualified as Instance+import Keiro.Workflow.Schema qualified as Schema+import Keiro.Workflow.Types+import Kiroku.Store.Append (appendToStream)+import Kiroku.Store.Connection (KirokuStore (..))+import Kiroku.Store.Effect (Store, runStoreIO)+import Kiroku.Store.Error (StoreError)+import Kiroku.Store.Transaction (runTransaction)+import Kiroku.Store.Types (EventData (..), EventType (..), ExpectedVersion (..))+import Test.Hspec+import "hasql-transaction" Hasql.Transaction qualified as Tx++spec :: Fixture -> Spec+spec fixture = describe "Keiro.Workflow.Inspection" $ do+  describe "opaque cursors" $ do+    it "round-trips Unicode and exact timestamps, rejects foreign kinds and oversized tokens" $ do+      let payload = ("承認" :: Text, "workflow/1" :: Text, read "2026-10-20 01:02:03.123456 UTC" :: UTCTime)+          token = encodeCursor "wf-key2" payload+      decodeCursor "wf-key2" token `shouldBe` Right payload+      encodeCursorBounded "wf-key2" payload `shouldBe` Just token+      encodeCursorBounded "wf-key2" (Text.replicate 100000 "界") `shouldBe` Nothing+      (decodeCursor "wf-created" token :: Either CursorError Value) `shouldBe` Left (CursorKindMismatch "wf-created" "wf-key2")+      forM_ ["not-a-cursor", "", Text.replicate 4097 "a", "===="] $ \bad ->+        (decodeCursor "wf-key2" (InspectionCursor bad) :: Either CursorError Value) `shouldSatisfy` (\case Left (MalformedCursor _) -> True; _ -> False)+  around (withFreshStore fixture) $ do+    it "rechecks every header field against its exact image, including old and far-future dates" $ \store -> do+      let fields =+            [ ("workflow_id", "'two'"),+              ("workflow_name", "'other'"),+              ("generation", "1"),+              ("status", "'failed'"),+              ("attempts", "0"),+              ("last_error", "'revised'"),+              ("next_attempt_at", "NULL"),+              ("wake_after", "NULL"),+              ("leased_by", "NULL"),+              ("lease_expires_at", "NULL"),+              ("created_at", "'0002-01-01 00:00:00 BC'"),+              ("updated_at", "'294276-01-01 00:00:00.123456+00'"),+              ("completed_at", "NULL"),+              ("created_at", "'infinity'"),+              ("updated_at", "'-infinity'"),+              ("next_attempt_at", "'infinity'"),+              ("wake_after", "'-infinity'"),+              ("lease_expires_at", "'infinity'"),+              ("completed_at", "'-infinity'")+            ]+          readHeader known name wid =+            Pool.use store.pool (Instance.lookupInstanceConditionalSession 4096 known (WorkflowName name) (WorkflowId wid)) >>= either (fail . show) pure+      forM_ fields $ \(field, value) -> do+        sql store "DELETE FROM keiro.keiro_workflows; INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status,attempts,last_error,next_attempt_at,wake_after,leased_by,lease_expires_at,created_at,updated_at,completed_at) VALUES ('one','header','running',2,'問題','1910-01-01 01:02:03.123456+00','2050-01-01+00','worker','2070-01-01+00','0001-01-01 00:00:00 BC','1970-01-01 00:00:00.123456+00','2026-01-01+00')"+        initial <- readHeader Nothing "header" "one"+        (key, original) <- case initial of+          Just (Instance.HeaderChanged key row) -> pure (key, row)+          other -> fail (show other)+        shouldBe (Just original) =<< expectStore store (Instance.lookupInstance (WorkflowName "header") (WorkflowId "one"))+        shouldReturn (readHeader (Just key) "header" "one") (Just Instance.HeaderUnchanged)+        sql store ("UPDATE keiro.keiro_workflows SET " <> field <> "=" <> value)+        let name = if field == "workflow_name" then "other" else "header"+            wid = if field == "workflow_id" then "two" else "one"+        changed <- readHeader (Just key) name wid+        case changed of+          Just (Instance.HeaderChanged next row) -> do+            shouldBe (Just row) =<< expectStore store (Instance.lookupInstance (WorkflowName name) (WorkflowId wid))+            shouldReturn (readHeader (Just next) name wid) (Just Instance.HeaderUnchanged)+          other -> fail ("did not observe changed " <> Text.unpack field <> ": " <> show other)+        shouldReturn (readHeader (Just key) "absent" "one") Nothing+        sql store "DELETE FROM keiro.keiro_workflows"+        shouldReturn (readHeader (Just key) name wid) Nothing+    it "rejects header images before returning oversized text or escaped JSON" $ \store -> do+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status,last_error) VALUES ('one','header','running',repeat('x',5000))"+      let readHeader = Pool.use store.pool (Instance.lookupInstanceConditionalSession 512 Nothing (WorkflowName "header") (WorkflowId "one"))+      shouldReturn readHeader (Right (Just Instance.HeaderImageTooLarge))+      sql store "UPDATE keiro.keiro_workflows SET last_error=repeat(chr(1),100)"+      shouldReturn readHeader (Right (Just Instance.HeaderImageTooLarge))+    it "reads fresh, owner-scoped headers through a reused session action" $ \store -> do+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('one','header','running')"+      let readHeader = Pool.use store.pool (Instance.lookupInstanceSession (WorkflowName "header") (WorkflowId "one")) >>= either (fail . show) pure+      initial <- readHeader+      shouldBe initial =<< expectStore store (Instance.lookupInstance (WorkflowName "header") (WorkflowId "one"))+      sql store "UPDATE keiro.keiro_workflows SET generation=1,status='failed',last_error='new' WHERE workflow_id='one' AND workflow_name='header'"+      refreshed <- readHeader+      shouldBe (fmap (\row -> (row.generation, row.status, row.lastError)) refreshed) (Just (1, Instance.WfFailed, Just "new"))+      shouldReturn (Pool.use store.pool (Instance.lookupInstanceSession (WorkflowName "other") (WorkflowId "one"))) (Right Nothing)+    it "rejects an outgoing cursor that clients could not resume within the cursor limit" $ \store -> do+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('a',repeat('界',2000),'running'),('b',repeat('界',2000),'running')"+      result <- expectStore store (listWorkflowInstancesPage defaultWorkflowListQuery {pageSize = 1})+      result `shouldBe` Left CursorResultTooLarge+    it "preserves large ordinary keys and includes them in both list orderings" $ \store -> do+      sql store "WITH bytes AS (SELECT string_agg(md5(n::text),'') value FROM generate_series(1,120) n) INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) SELECT left(value,2680),'a','running' FROM bytes UNION ALL SELECT 'b',left(value,2680),'completed' FROM bytes"+      original <- expectStore store (Instance.listWorkflowInstances Instance.defaultWorkflowInstanceFilter)+      length original `shouldBe` 2+      forM_ [ByKey, ByCreatedDesc] $ \ordering -> do+        pages <- drain store defaultWorkflowListQuery {pageSize = 1, ordering = ordering}+        sort (map (.workflowId) (concatMap (.items) pages)) `shouldBe` sort (map (.workflowId) original)+        map (length . (.items)) pages `shouldBe` [1, 1]+      running <- inspect store defaultWorkflowListQuery {classes = Just (ClassRunning :| [])}+      map statusClassOf running.items `shouldBe` [ClassRunning]+    it "pages wide keys without an obsolete exceptional-population limit" $ \store -> do+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) SELECT 'wide-'||lpad(n::text,3,'0'),repeat('x',2700),'running' FROM generate_series(1,65) n; INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('normal','normal','running')"+      forM_ [ByKey, ByCreatedDesc] $ \ordering -> do+        pages <- drain store defaultWorkflowListQuery {pageSize = 10, ordering}+        length (concatMap (.items) pages) `shouldBe` 66+      absent <- inspect store defaultWorkflowListQuery {workflowName = Just "absent"}+      absent.items `shouldBe` []+      normal <- inspect store defaultWorkflowListQuery {workflowName = Just "normal"}+      map (.workflowId) normal.items `shouldBe` ["normal"]+      sql store "DELETE FROM keiro.keiro_workflows; INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) SELECT 'bytes-'||n,repeat('x',131072),'running' FROM generate_series(1,2) n"+      expectStore store (listWorkflowInstancesPage defaultWorkflowListQuery {pageSize = 1}) `shouldReturn` Left CursorResultTooLarge+    it "orders by id before name and refuses the old key cursor kind" $ \store -> do+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('a','z','running'),('z','a','running')"+      pages <- drain store defaultWorkflowListQuery {pageSize = 1}+      map (\row -> (row.workflowId, row.workflowName)) (concatMap (.items) pages) `shouldBe` [("a", "z"), ("z", "a")]+      legacy <- expectStore store (Instance.listWorkflowInstances Instance.defaultWorkflowInstanceFilter)+      map (.workflowId) legacy `shouldBe` ["z", "a"]+      let old = encodeCursor "wf-key" ("a" :: Text, "z" :: Text)+      validateWorkflowListQuery defaultWorkflowListQuery {cursor = Just old} `shouldBe` Left (InvalidCursor (CursorKindMismatch "wf-key2" "wf-key"))+      expectStore store (listWorkflowInstancesPage defaultWorkflowListQuery {cursor = Just old}) `shouldReturn` Left (InvalidCursor (CursorKindMismatch "wf-key2" "wf-key"))+    it "classifies every stored row identically in Haskell and SQL without modifying rows" $ \store -> do+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id, workflow_name, status, attempts, wake_after) VALUES ('running','test','running',0,NULL),('retrying','test','running',2,NULL),('sleeping','test','suspended',0,now()+interval '1 hour'),('awaiting','test','suspended',0,NULL),('completed','test','completed',0,NULL),('cancelled','test','cancelled',0,NULL),('failed','test','failed',0,NULL)"+      sql store "UPDATE keiro.keiro_workflows SET status='running' WHERE workflow_id='failed'"+      _ <- expectStore store (Instance.claimInstance "stuck-worker" 30 (WorkflowName "test") (WorkflowId "failed"))+      expectStore store (runTransaction (Instance.upsertInstanceTx "failed" "test" 0 Instance.WfFailed (Just "exhausted")))+      original <- expectStore store (Instance.listWorkflowInstances Instance.defaultWorkflowInstanceFilter)+      pages <- forM [minBound .. maxBound] $ \status -> do+        page <- inspect store defaultWorkflowListQuery {classes = Just (status :| [])}+        map statusClassOf page.items `shouldBe` [status]+        if status == ClassFailed then map (.leasedBy) page.items `shouldBe` [Just "stuck-worker"] else pure ()+        pure page.items+      sort (map (.workflowId) (concat pages)) `shouldBe` sort (map (.workflowId) original)+      afterRows <- expectStore store (Instance.listWorkflowInstances Instance.defaultWorkflowInstanceFilter)+      afterRows `shouldBe` original+    it "pages key order completely, and omits continuation on a full final page" $ \store -> do+      seed store 23+      sql store "UPDATE keiro.keiro_workflows SET status=CASE WHEN right(workflow_id,1) IN ('1','4','7') THEN 'failed' WHEN right(workflow_id,1) IN ('2','5','8') THEN 'completed' ELSE status END"+      pages <- drain store defaultWorkflowListQuery {pageSize = 5}+      map (length . (.items)) pages `shouldBe` [5, 5, 5, 5, 3]+      let keys = map (\row -> (row.workflowId, row.workflowName)) (concatMap (.items) pages)+      length (nub keys) `shouldBe` 23+      keys `shouldBe` sort keys+      filtered <- drain store defaultWorkflowListQuery {pageSize = 3, classes = Just (ClassFailed :| [ClassCompleted, ClassFailed])}+      concatMap (.items) filtered `shouldBe` filter (\row -> statusClassOf row `elem` [ClassFailed, ClassCompleted]) (concatMap (.items) pages)+      sql store "DELETE FROM keiro.keiro_workflows WHERE workflow_id > 'w-020'"+      full <- drain store defaultWorkflowListQuery {pageSize = 5}+      map (length . (.items)) full `shouldBe` [5, 5, 5, 5]+      map (.nextCursor) full `shouldSatisfy` (\tokens -> case reverse tokens of Nothing : _ -> True; _ -> False)+    it "pages newest creation first including tied timestamps and observes an empty page clock" $ \store -> do+      seed store 12+      pages <- drain store defaultWorkflowListQuery {pageSize = 5, ordering = ByCreatedDesc}+      map (length . (.items)) pages `shouldBe` [5, 5, 2]+      let keys = map (\row -> (row.createdAt, row.workflowName, row.workflowId)) (concatMap (.items) pages)+      keys `shouldBe` reverse (sort keys)+      empty <- inspect store defaultWorkflowListQuery {workflowName = Just "absent"}+      empty.items `shouldBe` []+      empty.nextCursor `shouldBe` Nothing+      empty.observedAt `shouldSatisfy` (> read "2026-01-01 00:00:00 UTC")+    it "rejects sizes and malformed or cross-order cursors before reading" $ \store -> do+      forM_ [0, 501, maxBound] $ \size ->+        expectStore store (listWorkflowInstancesPage defaultWorkflowListQuery {pageSize = size}) `shouldReturn` Left (InvalidPageSize size)+      malformed <- expectStore store (listWorkflowInstancesPage defaultWorkflowListQuery {cursor = Just (InspectionCursor "not-a-cursor")})+      malformed `shouldSatisfy` (\case Left (InvalidCursor (MalformedCursor _)) -> True; _ -> False)+      seed store 2+      first <- inspect store defaultWorkflowListQuery {pageSize = 1}+      crossOrder <- expectStore store (listWorkflowInstancesPage defaultWorkflowListQuery {ordering = ByCreatedDesc, cursor = first.nextCursor})+      crossOrder `shouldSatisfy` (\case Left (InvalidCursor (CursorKindMismatch "wf-created" "wf-key2")) -> True; _ -> False)+    it "applies half-open creation and update windows" $ \store -> do+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status,created_at,updated_at) SELECT 'window-'||n,'window','running','2026-10-01 00:00:00+00'::timestamptz+n*interval '1 hour','2026-10-01 00:00:00+00'::timestamptz+n*interval '1 hour' FROM generate_series(0,2) n"+      let boundary = read "2026-10-01 01:00:00 UTC" :: UTCTime+      forM_ [defaultWorkflowListQuery {createdSince = Just boundary}, defaultWorkflowListQuery {updatedSince = Just boundary}] $ \query -> do+        page <- inspect store query+        length page.items `shouldBe` 2+      forM_ [defaultWorkflowListQuery {createdBefore = Just boundary}, defaultWorkflowListQuery {updatedBefore = Just boundary}] $ \query -> do+        page <- inspect store query+        length page.items `shouldBe` 1+    it "observes then-current inserts strictly beyond the cursor" $ \store -> do+      seed store 3+      first <- inspect store defaultWorkflowListQuery {pageSize = 1}+      sql store "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('a','a','running'),('z','z','running')"+      second <- inspect store defaultWorkflowListQuery {cursor = first.nextCursor}+      map (.workflowId) second.items `shouldSatisfy` elem "z"+      map (.workflowId) second.items `shouldSatisfy` not . elem "a"+    it "independently pages journal and steps and rejects cross-owner and rotated generation cursors" $ \store -> do+      clock <- expectStore store Schema.inspectionObservedAt+      forM_ ["a", "b", "c"] $ \step -> expectStore store (appendJournalEntry (WorkflowName "test") (WorkflowId "owner") (StepRecorded step Null clock))+      first <- expectStore store (readWorkflowJournalPage (WorkflowName "test") (WorkflowId "owner") (DetailPageQuery Nothing 2 Nothing)) >>= right+      length first.items `shouldBe` 2+      second <- expectStore store (readWorkflowJournalPage (WorkflowName "test") (WorkflowId "owner") (DetailPageQuery Nothing 2 first.nextCursor)) >>= right+      length second.items `shouldBe` 1+      second.nextCursor `shouldBe` Nothing+      steps <- expectStore store (loadStepIndexPage (WorkflowName "test") (WorkflowId "owner") (DetailPageQuery Nothing 2 Nothing)) >>= right+      length steps.items `shouldBe` 2+      other <- expectStore store (loadStepIndexPage (WorkflowName "test") (WorkflowId "different") (DetailPageQuery Nothing 2 steps.nextCursor))+      other `shouldSatisfy` (\case Left (InvalidCursor _) -> True; _ -> False)+      _ <- expectStore store (runWorkflow (WorkflowName "test") (WorkflowId "owner") (continueAsNew () >> pure ()))+      rotated <- expectStore store (loadStepIndexPage (WorkflowName "test") (WorkflowId "owner") (DetailPageQuery Nothing 2 steps.nextCursor))+      rotated `shouldBe` Left (CursorGenerationMismatch 1 0)+    it "pages children and directly allocated awakeables through their owning tables" $ \store -> do+      expectStore store $ runTransaction $ do+        Child.registerChildTx "child-a" "child" "owner" "test" "await-a"+        Child.registerChildTx "child-b" "child" "owner" "test" "await-b"+        Awakeable.registerAwakeableTx UUID.nil "test" "owner"+        Awakeable.registerAwakeableTx (maybe (error "UUID fixture") (\x -> x) (UUID.fromString "00000000-0000-0000-0000-000000000001")) "test" "owner"+      first <- expectStore store (listWorkflowChildrenPage (WorkflowName "test") (WorkflowId "owner") (OwnerPageQuery 1 Nothing)) >>= right+      second <- expectStore store (listWorkflowChildrenPage (WorkflowName "test") (WorkflowId "owner") (OwnerPageQuery 1 first.nextCursor)) >>= right+      map (.childId) (first.items <> second.items) `shouldBe` ["child-a", "child-b"]+      second.nextCursor `shouldBe` Nothing+      awake <- expectStore store (listWorkflowAwakeablesPage (WorkflowName "test") (WorkflowId "owner") (OwnerPageQuery 1 Nothing)) >>= right+      following <- expectStore store (listWorkflowAwakeablesPage (WorkflowName "test") (WorkflowId "owner") (OwnerPageQuery 1 awake.nextCursor)) >>= right+      length (awake.items <> following.items) `shouldBe` 2+      following.nextCursor `shouldBe` Nothing+      foreignPage <- expectStore store (listWorkflowAwakeablesPage (WorkflowName "test") (WorkflowId "different") (OwnerPageQuery 1 awake.nextCursor))+      foreignPage `shouldSatisfy` (\case Left (InvalidCursor _) -> True; _ -> False)+    it "raises the runtime's decode error instead of returning a partial corrupt journal" $ \store -> do+      _ <- expectStore store (appendToStream (workflowGenerationStreamName (WorkflowName "corrupt") (WorkflowId "owner") 0) AnyVersion [EventData Nothing (EventType "StepRecorded") (String "invalid") Nothing Nothing Nothing])+      expectStore store (readWorkflowJournalPage (WorkflowName "corrupt") (WorkflowId "owner") (DetailPageQuery Nothing 2 Nothing))+        `shouldThrow` (\case WorkflowJournalDecodeError _ -> True; _ -> False)++seed :: KirokuStore -> Int -> IO ()+seed store count = sql store ("INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status,created_at,updated_at) SELECT 'w-'||lpad(n::text,3,'0'),'name-'||(n%3),'running','2026-10-01 00:00:00+00'::timestamptz+(n/2)*interval '1 minute','2026-10-01 00:00:00+00'::timestamptz FROM generate_series(1," <> Text.pack (show count) <> ") n")++sql :: KirokuStore -> Text -> IO ()+sql store text = expectStore store (runTransaction (Tx.sql (encodeUtf8 text)))++expectStore :: KirokuStore -> Eff '[Store, Error StoreError, IOE] a -> IO a+expectStore store action = runStoreIO store action >>= either (fail . show) pure++right :: (Show err) => Either err a -> IO a+right = either (fail . show) pure++inspect :: KirokuStore -> WorkflowListQuery -> IO WorkflowListPage+inspect store query = expectStore store (listWorkflowInstancesPage query) >>= right++drain :: KirokuStore -> WorkflowListQuery -> IO [WorkflowListPage]+drain store query = do+  page <- inspect store query+  rest <- maybe (pure []) (\token -> drain store query {cursor = Just token}) page.nextCursor+  pure (page : rest)