keiro-ops-http (empty) → 0.22.0.0

raw patch · 26 files changed

+4152/−0 lines, 26 filesdep +aesondep +asyncdep +base

Dependencies added: aeson, async, base, bytestring, containers, directory, effectful, filepath, hasql, hasql-pool, hspec, http-client, http-types, jitsurei, keiro, keiro-migrations, keiro-ops, keiro-ops-http, keiro-pgmq, keiro-test-support, kiroku-metrics, kiroku-store, network, network-uri, optparse-applicative, pgmq-migration, process, stm, text, time, uuid, vector, wai, wai-extra, wai-websockets, warp, websockets

Files

• CHANGELOG.md +63/−0
• LICENSE +28/−0
• app/Main.hs +50/−0
• bench/InspectionClients.hs +141/−0
• bench/InspectionInterference.hs +343/−0
• example/Compose.hs +56/−0
• keiro-ops-http.cabal +228/−0
• src/Keiro/Ops/Http.hs +23/−0
• src/Keiro/Ops/Http/Application.hs +264/−0
• src/Keiro/Ops/Http/Budget.hs +96/−0
• src/Keiro/Ops/Http/Compose.hs +110/−0
• src/Keiro/Ops/Http/Config.hs +56/−0
• src/Keiro/Ops/Http/Cors.hs +67/−0
• src/Keiro/Ops/Http/Feed.hs +235/−0
• src/Keiro/Ops/Http/Feed/Driver.hs +258/−0
• src/Keiro/Ops/Http/Feed/Queue.hs +158/−0
• src/Keiro/Ops/Http/Pool.hs +40/−0
• src/Keiro/Ops/Http/Protocol.hs +106/−0
• src/Keiro/Ops/Http/Route.hs +94/−0
• src/Keiro/Ops/Http/Server.hs +71/−0
• src/Keiro/Ops/Http/Views.hs +147/−0
• src/Keiro/Ops/Http/WebSocket.hs +184/−0
• test/Main.hs +802/−0
• test/Test/ComposeSpec.hs +137/−0
• test/Test/FeedSpec.hs +228/−0
• test/Test/WebSocketSpec.hs +167/−0

@@ -0,0 +1,63 @@+# Changelog++## [Unreleased]++## 0.22.0.0 — 2026-10-27++### Breaking Changes++- Require the Kiroku Store 0.12 cohort from+  mori://shinzui/kiroku/packages/kiroku-store; apply migration 0016 from+  mori://shinzui/kiroku/packages/kiroku-store-migrations with affected workers+  stopped, then restart new binaries before using namespace targets. Update+  exhaustive subscription target/event matches and low-level record constructors.++- Direct `InspectionLimits` records must supply `deliveryDeadlineSeconds`; the default is 30 seconds. Nonempty confirmation bodies must be application/json, with media-type parameters permitted.++### New Features++- Expose mounted process-manager discovery as `GET /pm/types` through the shared+  command tree, preserving schema verification and transport resource limits.+  Encode discovery on demand with the same byte/deadline bounds and no response cache.++- Compose allocated inspection applications under stable `/keiro`, `/kiroku`,+  `/shibuya` and `/pgmq` prefixes with preserved raw paths, shared CORS policy+  and a flag-guarded example using released Kiroku metrics.++- Add shared `/ws` first-page feeds for workflow, timer and projection status,+  backed by the same bounded reads and encoders as polls. Drivers share the+  HTTP budget; idle handles own no wake cursor. Count/byte-bounded queues,+  in-flight accounting, overflow recovery, upgrade policy checks and scoped+  cleanup protect runtime resources. The standalone HTTP server mounts feeds.++- 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.+++- Bounded workflow routes and typed cursor/size/generation failures; pure validation precedes schema scans.+++- Bounded HTTP transport for the supported operational command tree.++### Bug Fixes++- Fail oversized workflow continuations explicitly instead of returning a cursor+  that the next request cannot decode.+++- Refuse malformed, null, multiple and foreign mutation origins before body, schema or command effects with `origin_not_allowed` (403). HTTP, CORS and WebSocket comparisons normalize scheme/host/effective port. No-Origin clients and empty previews remain supported; TLS proxies explicitly allow-list HTTPS.+- Nonempty unsupported confirmation media types, including chunked bodies, return `unsupported_media_type` (415).+- Bound delivery independently of computation; expire stalled response callbacks, return permits and terminate incomplete connections without a second response. Shared health admission may refuse during saturation and recovers after cancellation.+- Retain per-refresh schema verification and fresh snapshot presentation without clock-only deltas. Return `view_retiring` while zero-reference cleanup joins. Composition preserves the complete child origin authorization policy and one CORS header.+- Follow primary-key workflow pages with `wf-key2`, preserving continuation byte caps and honest filtered-query deadlines.++### Other Changes++- Reuse at most 32 header presentations / 1 MiB charged data after exact current-row checks, retaining live schema verification and host response caps.++- Reuse bounded pure read parsing and the RTS deadline manager. Retain a fresh+  live schema check for every admitted operation and hold admission through+  response delivery.+

@@ -0,0 +1,28 @@+BSD 3-Clause License++Copyright (c) 2026, Nadeem Bitar++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice, this+   list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright notice,+   this list of conditions and the following disclaimer in the documentation+   and/or other materials provided with the distribution.++3. Neither the name of the copyright holder nor the names of its contributors+   may be used to endorse or promote products derived from this software without+   specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES+INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS+OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT INCLUDING+NEGLIGENCE OR OTHERWISE ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN+IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,50 @@+module Main (main) where++import Control.Concurrent.Async (wait)+import Data.Foldable (traverse_)+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.IO qualified as Text+import Hasql.Connection.Settings qualified as Settings+import Keiro.Migrations.SchemaCheck+import Keiro.Ops (emptyAppHooks)+import Keiro.Ops.Env (resolveConnectionString)+import Keiro.Ops.Http+import Keiro.Ops.Http.Config qualified as Config+import Kiroku.Store.Connection (ConnectionSettingsM (..), KirokuStore (..), defaultConnectionSettings, withStore)+import Options.Applicative+import System.IO (stderr)++data Options = Options !(Maybe Text) !OpsHttpConfig !OpsHttpServerConfig++options :: Parser Options+options =+  Options+    <$> optional (textOption (long "database-url" <> metavar "URL"))+    <*> ( (\enabled drift origins -> defaultOpsHttpConfig {mutations = if enabled then MutationsEnabled else MutationsDisabled, schemaDrift = if drift then AllowDrift else RefuseOnDrift, Config.allowedOrigins = origins})+            <$> switch (long "enable-mutations")+            <*> switch (long "allow-schema-drift")+            <*> many (textOption (long "allow-origin" <> metavar "ORIGIN"))+        )+    <*> (OpsHttpServerConfig <$> textOption (long "bind" <> value "127.0.0.1") <*> option auto (long "port" <> value 9092) <*> option auto (long "request-timeout" <> value 300))+  where+    textOption modifiers = Text.pack <$> strOption modifiers++main :: IO ()+main = do+  Options database config serverConfig <- execParser (info (options <**> helper) (fullDesc <> progDesc "Serve Keiro inspection on a trusted network or behind an authenticating proxy"))+  connectionString <- resolveConnectionString database+  verified <- verifyExpectedSchema (Settings.connectionString connectionString)+  case verified of+    Left err -> ioError (userError (show err))+    Right drift -> traverse_ (Text.hPutStrLn stderr . ("warning: " <>) . renderSchemaDrift) drift+  -- This store belongs only to inspection. All its pooled connections inherit+  -- the SQL deadline; command execution never borrows the runtime's pool.+  let settings = (defaultConnectionSettings connectionString) {poolSize = config.inspectionLimits.maxConcurrentReads, statementTimeout = Just 2}+  withStore settings $ \baseStore -> withOpsInspectionPool config.inspectionLimits settings $ \pool -> do+    let store = baseStore {pool}+    withFeedHandle config defaultFeedConfig store $ \feeds -> do+      app <- opsApplicationWithFeeds config feeds emptyAppHooks store+      withOpsHttpServer serverConfig app $ \server -> do+        Text.hPutStrLn stderr ("keiro-ops-http: listening on " <> serverConfig.bindHost <> ":" <> Text.pack (show server.serverPort) <> " (mutations " <> (if config.mutations == MutationsEnabled then "enabled" else "disabled") <> ")")+        wait server.serverThread

@@ -0,0 +1,141 @@+-- External load process: client allocation never enters runtime RTS counters.+module InspectionClients (runClients) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async+import Control.Concurrent.MVar+import Control.Exception (bracket, onException, throwIO, try)+import Control.Monad (forM, forM_, forever, unless, void)+import Data.Aeson (Value (..), encode, object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KeyMap+import Data.ByteString.Lazy qualified as Lazy+import Data.ByteString.Lazy.Char8 qualified as Char8+import Data.IORef+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as Text+import Network.HTTP.Client qualified as HTTP+import Network.HTTP.Types (statusCode)+import Network.WebSockets qualified as WS+import System.IO (hFlush, hPutStrLn, stderr, stdout)+import System.Timeout (timeout)++runClients :: [String] -> IO ()+runClients [portArg, clientsArg, _, secondsArg, mode]+  | mode `elem` ["pm-types", "pm-types-many"] =+      runManagerClients (read portArg) (read clientsArg) (read secondsArg) (mode == "pm-types-many")+runClients [portArg, clientsArg, viewsArg, secondsArg, mode] = do+  let port = read portArg+      clients = read clientsArg+      views = read viewsArg+      seconds = read secondsArg+  go <- newEmptyMVar+  counters <- newIORef Map.empty+  let count code = atomicModifyIORef' counters (\counts -> (Map.insertWith (+) code (1 :: Int) counts, ()))+      subscribe index+        | views == 32 && index `mod` views == 31 = encode (object ["type" .= ("subscribe" :: Text), "feed" .= ("timers" :: Text)])+        | views == 32 && index `mod` views == 30 = encode (object ["type" .= ("subscribe" :: Text), "feed" .= ("projection_groups" :: Text)])+        | otherwise = encode (object ["type" .= ("subscribe" :: Text), "feed" .= ("workflows" :: Text), "workflow_name" .= (if index `mod` views == 0 then "index-evidence" else "seed-" <> show (index `mod` views - 1)), "limit" .= (100 :: Int)])+      frame connection = do+        raw <- WS.receiveData connection :: IO Lazy.ByteString+        case Aeson.eitherDecode raw of+          Right (Object fields) -> case KeyMap.lookup "type" fields of+            Just (String "snapshot") -> count "snapshot" >> pure "snapshot"+            Just (String "update") -> count "update" >> pure "update"+            Just (String "error") -> count ("feed_" <> show (KeyMap.lookup "code" fields)) >> pure "error"+            _ -> pure "control"+          _ -> fail "invalid feed JSON"+      worker index ready = WS.runClient "127.0.0.1" port "/keiro/ws" $ \connection -> do+        WS.sendTextData connection (subscribe index)+        initial <- frame connection+        unless (initial == ("snapshot" :: Text)) (fail "client did not receive a real initial snapshot")+        putMVar ready (Right ())+        readMVar go+        let consume = if mode == "slow-reader" then threadDelay (seconds * 1000000) else forever (frame connection)+            churn = forever $ do+              threadDelay 1000000+              WS.sendTextData connection (encode (object ["type" .= ("unsubscribe" :: Text), "feed" .= ("workflows" :: Text)]))+              WS.sendTextData connection (subscribe index)+        if mode == "churn" then race_ consume churn else consume+      start index = do+        ready <- newEmptyMVar+        task <- async (worker index ready `onException` void (tryPutMVar ready (Left (userError "feed client failed before readiness"))))+        status <- timeout 10000000 (takeMVar ready)+        case status of+          Just (Right ()) -> pure task+          Just (Left err) -> cancel task >> throwIO err+          Nothing -> cancel task >> fail "feed client readiness timeout"+  -- Sequential handshake avoids benchmarking a startup admission race. All+  -- sockets remain live together during the measured window.+  bracket (forM [0 .. clients - 1] start) (mapM_ cancel) $ \tasks -> do+    manager <- HTTP.newManager HTTP.defaultManagerSettings {HTTP.managerConnCount = clients}+    -- Match the established header-cache experiment: warm the exact 32 poll+    -- keys outside the timed writer window, without counting these requests.+    forM_ [1 .. 32 :: Int] $ \index -> do+      request <- HTTP.parseRequest ("http://127.0.0.1:" <> show port <> "/keiro/wf/inspect/header/index-evidence/warm-" <> show index)+      response <- HTTP.httpLbs request manager+      unless (statusCode (HTTP.responseStatus response) == 200) (fail "HTTP view warm-up refused")+    let poll index = do+          request <- HTTP.parseRequest ("http://127.0.0.1:" <> show port <> "/keiro/wf/inspect/header/index-evidence/warm-" <> show (1 + index `mod` 32))+          readMVar go+          threadDelay (index * 1000000 `div` clients)+          forever $ do+            result <- try (HTTP.httpLbs request manager) :: IO (Either HTTP.HttpException (HTTP.Response Lazy.ByteString))+            either (const (count "http_exception")) (count . ("http_" <>) . show . statusCode . HTTP.responseStatus) result+            threadDelay 1000000+    bracket (forM [0 .. clients - 1] (async . poll)) (mapM_ cancel) $ \pollers -> do+      putStrLn "ready"+      hFlush stdout+      instruction <- getLine+      if instruction /= "go" then fail "missing measurement control" else pure ()+      putMVar go ()+      -- Surface unexpected client failures instead of measuring a lost load.+      race_ (threadDelay (seconds * 1000000)) (mapConcurrently_ wait (tasks <> pollers))+    HTTP.closeManager manager+  counts <- readIORef counters+  hPutStrLn stderr ("inspection clients: " <> show counts)+  let good = sum [n | (key, n) <- Map.toList counts, key `elem` ["http_200", "snapshot", "update"]]+      bad = sum [n | (key, n) <- Map.toList counts, key `notElem` ["http_200", "snapshot", "update"]]+  Char8.putStrLn (encode (good, bad))+runClients _ = fail "expected PORT CLIENTS VIEWS SECONDS MODE"++-- HTTP-only discovery load: no WebSocket subscriptions or workflow header warmup.+-- Finite staggered schedules retain the intended four requests per second even+-- when request time varies; a failed request or late schedule fails the trial.+runManagerClients :: Int -> Int -> Int -> Bool -> IO ()+runManagerClients port clients seconds many = do+  unless (clients == 4 && seconds > 0) (fail "expected four manager discovery clients")+  let expectedNames = if many then ["alias-" <> Text.justifyRight 4 '0' (Text.pack (show n)) | n <- [1 .. 1000 :: Int]] else ["jitsurei-fulfillment"]+      expected = object ["items" .= [object ["name" .= name, "category" .= ("fulfillment" :: Text), "snapshots_configured" .= False, "snapshot_policy" .= ("never" :: Text), "timers_configured" .= Null] | name <- expectedNames]]+  bracket (HTTP.newManager HTTP.defaultManagerSettings {HTTP.managerConnCount = clients}) HTTP.closeManager $ \manager -> do+    request <- HTTP.parseRequest ("http://127.0.0.1:" <> show port <> "/keiro/pm/types")+    let fetch = do+          response <- HTTP.httpLbs request manager+          unless+            (statusCode (HTTP.responseStatus response) == 200 && Aeson.eitherDecode (HTTP.responseBody response) == Right expected)+            (fail "manager discovery poll was refused or returned incorrect metadata")+    fetch+    go <- newEmptyMVar+    successes <- newIORef (0 :: Int)+    let poll index = do+          readMVar go+          threadDelay (index * 1000000 `div` clients)+          forM_ [1 .. seconds] $ \second -> do+            -- Sleep until the next tick independently of request duration.+            withAsync (threadDelay 1000000) $ \tick -> do+              fetch+              atomicModifyIORef' successes (\n -> (n + 1, ()))+              unless (second == seconds) (wait tick)+    bracket (forM [0 .. clients - 1] (async . poll)) (mapM_ cancel) $ \tasks -> do+      putStrLn "ready"+      hFlush stdout+      instruction <- getLine+      unless (instruction == "go") (fail "missing measurement control")+      putMVar go ()+      finished <- timeout ((seconds + 5) * 1000000) (mapConcurrently_ wait tasks)+      unless (finished == Just ()) (fail "manager polling did not sustain the intended load")+    good <- readIORef successes+    unless (good == clients * seconds) (fail "manager polling load disappeared")+    hPutStrLn stderr ("manager discovery clients: successes=" <> show good <> " failures=0")+    Char8.putStrLn (encode (good, 0 :: Int))

@@ -0,0 +1,343 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE OverloadedStrings #-}++-- Durable writer profile. KEIRO_INSPECTION_BASELINE compiles the same writer+-- against pre-inspection packages without requiring the new HTTP package.+-- Supply OUTPUT_DIRECTORY [saturation|OPS_PER_SECOND] [TRIALS] [MODE].+-- Run with +RTS -N2 -T -RTS; this is not the full domain workload matrix.+import Control.Concurrent (threadDelay)+import Control.Exception (bracket)+import Control.Monad (forM, forM_, unless, void, when)+import Data.Aeson (Value, object, (.=))+import Data.Aeson qualified as Aeson+import Data.ByteString.Lazy qualified as Lazy+import Data.ByteString.Lazy.Char8 qualified as LazyChar8Portable+#ifndef KEIRO_INSPECTION_BASELINE+import InspectionClients (runClients)+import Data.Map.Strict qualified as Map+import Jitsurei.Inspection (jitsureiProcessManagerInspectors)+#endif+import Data.ByteString qualified as Bytes+import Data.List (sort)+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text+import Data.Time (getCurrentTime)+import Data.UUID qualified as UUID+import Data.UUID.V5 qualified as UUIDv5+import Effectful (Eff, IOE)+import Effectful.Error.Static (Error)+import GHC.Clock (getMonotonicTimeNSec)+import GHC.Conc (ThreadStatus (..), getAllocationCounter, listThreads, threadStatus)+import GHC.Stats (GCDetails (..), RTSStats (..), getRTSStats, getRTSStatsEnabled)+import Hasql.Connection qualified as Connection+import Hasql.Connection.Settings qualified as Settings+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Hasql.Statement qualified as Statement+#ifndef KEIRO_INSPECTION_BASELINE+import Keiro.Ops qualified as Ops+import Keiro.Ops.Http+#endif+import Keiro.ReadModel (registerReadModel)+import Keiro.Test.Postgres+import Keiro.Timer qualified as Timer+import Keiro.Workflow+import Keiro.Workflow.Instance qualified as Instance+import Kiroku.Store.Connection+import Kiroku.Store.Effect (Store, runStoreIO)+import Kiroku.Store.Error (StoreError)+import Kiroku.Store.Transaction (runTransaction)+import Kiroku.Store.Types (EventId (..))+import System.Directory (createDirectoryIfMissing)+import System.Environment (getArgs, getExecutablePath, lookupEnv, withArgs)+import System.IO (IOMode (WriteMode), hFlush, withFile)+#ifndef KEIRO_INSPECTION_BASELINE+import System.Exit (ExitCode (..))+import System.IO (hClose, hGetLine, hPutStrLn)+#endif+import System.FilePath ((</>))+import System.Mem (performMajorGC)+#ifndef KEIRO_INSPECTION_BASELINE+import System.Process (CreateProcess (..), StdStream (..), createProcess, proc, readProcessWithExitCode, terminateProcess, waitForProcess)+#endif+import Text.Read (readMaybe)++main :: IO ()+main = do+  args <- getArgs+#ifndef KEIRO_INSPECTION_BASELINE+  case args of+    "--clients" : rest -> runClients rest+    _ -> withArgs (translate args) benchmark+#else+  withArgs (translate args) benchmark+#endif++-- Named options for cabal bench, retaining positional arguments for matched runs.+translate :: [String] -> [String]+translate args+  | "--mode" `elem` args = [option "--output" "inspection-results", option "--rate" "100", option "--trials" "1", option "--mode" "disabled"]+  | otherwise = args+  where+    option key fallback = case dropWhile (/= key) args of _ : value : _ -> value; _ -> fallback++benchmark :: IO ()+benchmark = do+  (output, loadArgument, trialsArgument, modeArgument) <-+    getArgs >>= \case+      [path] -> pure (path, "50", "5", "all")+      [path, load] -> pure (path, load, "5", "all")+      [path, load, trials] -> pure (path, load, trials, "all")+      [path, load, trials, mode] -> pure (path, load, trials, mode)+      _ -> fail "expected OUTPUT_DIRECTORY [saturation|OPS_PER_SECOND] [TRIALS] [MODE]"+  offeredRate <- case loadArgument of+    "saturation" -> pure Nothing+    raw | Just rate <- readMaybe raw, rate > 0, rate <= (1000 :: Int) -> pure (Just rate)+    _ -> fail "offered rate must be 1..1000 or saturation"+  trialCount <- case readMaybe trialsArgument of+    Just count | count > 0, count <= (100 :: Int) -> pure count+    _ -> fail "trials must be 1..100"+  selectedModes <- case modeArgument of+    "all" -> pure inspectionModes+    mode | Text.pack mode `elem` inspectionModes -> pure [Text.pack mode]+    _ -> fail "mode is unavailable in this binary"+  retentionSeconds <-+    lookupEnv "KEIRO_INSPECTION_RETENTION_SECONDS" >>= \case+      Nothing -> pure Nothing+      Just raw | Just seconds <- readMaybe raw, seconds >= 60, seconds <= (3600 :: Int), mod seconds 5 == 0 -> pure (Just seconds)+      _ -> fail "KEIRO_INSPECTION_RETENTION_SECONDS must be 60..3600 and divisible by 5"+  unless (retentionSeconds == Nothing || offeredRate /= Nothing) (fail "retention mode requires a paced offered rate")+  configuredSeconds <- lookupEnv "KEIRO_INSPECTION_SECONDS"+  let duration = maybe (maybe 20 id (configuredSeconds >>= readMaybe)) id retentionSeconds+      sampleCount = maybe 1000 (* duration) offeredRate+      modes = maybe ["disabled"] (const selectedModes) offeredRate+  enabled <- getRTSStatsEnabled+  unless enabled (fail "run with +RTS -T")+  createDirectoryIfMissing True output+  withMigratedBenchmarkSuite $ \fixture ->+    forM_ [1 .. trialCount] $ \trial ->+      forM_ modes $ \mode -> withFreshDatabase fixture $ \url -> do+        execute url "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) SELECT 'seed-'||n,'seed-'||(n%31),'running' FROM generate_series(1,100000) n; ANALYZE keiro.keiro_workflows"+        execute url "UPDATE keiro.keiro_workflows SET last_error=repeat('x',65536) WHERE workflow_id=(SELECT workflow_id FROM keiro.keiro_workflows WHERE workflow_name='seed-0' ORDER BY workflow_id LIMIT 1)"+        execute url "INSERT INTO keiro.keiro_timers(timer_id,process_manager_name,correlation_id,fire_at,payload,status) SELECT ('00000000-0000-0000-0000-'||lpad(n::text,12,'0'))::uuid,'seed-timer','corr-'||(n%100),'2100-01-01'::timestamptz+n*interval '1 second',CASE WHEN n=2 THEN to_jsonb(repeat('x',65536)) ELSE jsonb_build_object('n',n) END,CASE WHEN n%4=0 THEN 'fired' WHEN n%8=1 THEN 'firing' ELSE 'scheduled' END FROM generate_series(1,100000) n; ANALYZE keiro.keiro_timers"+        withStore (defaultConnectionSettings url) {poolSize = 1} $ \store -> withMount store url mode $ \app -> do+          _ <- runStoreIO store (registerReadModel "inspection-benchmark" 1 "shape") >>= right+          settings <- use store (Session.statement () (Statement.preparable "SELECT jsonb_build_object('fsync',current_setting('fsync'),'synchronous_commit',current_setting('synchronous_commit'),'full_page_writes',current_setting('full_page_writes'),'autovacuum',current_setting('autovacuum'),'server_version',current_setting('server_version'))" E.noParams (D.singleRow (D.column (D.nonNullable D.jsonb)))))+          durable <- use store (Session.statement () (Statement.preparable "SELECT current_setting('fsync')='on' AND current_setting('synchronous_commit')='on' AND current_setting('full_page_writes')='on' AND current_setting('autovacuum')='on'" E.noParams (D.singleRow (D.column (D.nonNullable D.bool)))))+          unless durable (fail "index experiment requires normal durable settings")+          -- Finish the 100k-row setup's dirty-page writes before warm-up.+          -- Both package sets execute this same explicit checkpoint; normal+          -- durable settings and autovacuum stay enabled during measurement.+          use store (Session.script "CHECKPOINT")+          forM_ [1 .. 50] $ \n -> operation store ("warm-" <> number n) >> pure ()+          lifecycleBefore <- if retentionSeconds /= Nothing then Just <$> heapSample "mounted-before-pollers" 0 else pure Nothing+          ((samples, statsBefore, started, before, writerFinished), pollSuccess, pollFailure) <- withPollers app mode duration $ \startPolling -> do+            -- Let warm writer notifications settle in every mode, including+            -- the baseline, before capturing counters and the measurement clock.+            threadDelay 1000000+            flush store+            before <- counters store+            performMajorGC+            statsBefore <- getRTSStats+            started <- getMonotonicTimeNSec+            startPolling+            let measured n = do+                  now <- getMonotonicTimeNSec+                  let due = maybe now (\rate -> started + fromIntegral (n - 1) * div 1000000000 (fromIntegral rate)) offeredRate+                  when (due > now) (threadDelay (fromIntegral (div (due - now) 1000)))+                  admitted <- getMonotonicTimeNSec+                  (schedule, claim, resume, timer, allocation) <- operation store ("measured-" <> Text.justifyRight 9 '0' (number (sampleCount - n)))+                  let waiting = if offeredRate /= Nothing && admitted > due then fromIntegral (admitted - due) / 1000 else 0+                  pure (waiting, schedule, claim, resume, timer, allocation)+            samples <- case retentionSeconds of+              Nothing -> forM [1 .. sampleCount] measured+              Just _ -> withFile (output </> (show trial <> "-" <> Text.unpack mode <> "-heap.jsonl")) WriteMode $ \handle -> do+                let emit phase n = do+                      value <- heapSample phase n+                      LazyChar8Portable.hPutStrLn handle (Aeson.encode value)+                      hFlush handle+                emit "before-running" 0+                -- Discard every per-operation timing. Retaining a latency list+                -- would itself create a false positive live-heap trend.+                -- A range shared across trials/modes floats outside their+                -- callbacks at O1 and retains 40 bytes per completed item.+                -- Keep this counter strict and allocate no enumeration list.+                let loop n+                      | n > sampleCount = pure ()+                      | otherwise = do+                          void (measured n)+                          when (mod n (maybe 1 (* 5) offeredRate) == 0) (emit "running" n)+                          loop (n + 1)+                loop 1+                pure []+            writerFinished <- getMonotonicTimeNSec+            pure (samples, statsBefore, started, before, writerFinished)+          finished <- getMonotonicTimeNSec+          performMajorGC+          statsAfter <- getRTSStats+          flush store+          after <- counters store+          resourcesAfter <- traverse (\(_, sample) -> sample True) app+          lifecycleAfter <- case retentionSeconds of+            Nothing -> pure Nothing+            Just _ -> do+              threadDelay 12000000+              Just <$> heapSample "mounted-after-poller-cleanup-and-drain" sampleCount+          let elapsed = fromIntegral (writerFinished - started) / 1e9 :: Double+              cleanupSeconds = fromIntegral (finished - writerFinished) / 1e9 :: Double+              durations = map (\(q, a, b, c, d, _) -> q + a + b + c + d) samples+              percentile p = if null durations then Nothing else Just (sort durations !! min (sampleCount - 1) (floor (p * fromIntegral sampleCount)))+              (views, clients) = pollingShape mode+              artifact = object ["resources_after_clients" .= resourcesAfter, "trial" .= trial, "retention_seconds" .= retentionSeconds, "lifecycle_before" .= lifecycleBefore, "lifecycle_after" .= lifecycleAfter, "mode" .= mode, "views" .= views, "clients" .= clients, "poll_success" .= pollSuccess, "poll_failure" .= pollFailure, "writer_thread_allocated_bytes_per_operation" .= (if null samples then Nothing else Just (sum [allocated | (_, _, _, _, _, allocated) <- samples] / fromIntegral sampleCount)), "allocated_bytes" .= (statsAfter.allocated_bytes - statsBefore.allocated_bytes), "cpu_ns" .= (statsAfter.cpu_ns - statsBefore.cpu_ns), "settings" .= (settings :: Value), "offered_ops_per_second" .= offeredRate, "elapsed_seconds" .= elapsed, "client_cleanup_seconds" .= cleanupSeconds, "minor_and_major_gcs" .= (statsAfter.gcs - statsBefore.gcs), "major_gcs" .= (statsAfter.major_gcs - statsBefore.major_gcs), "gc_cpu_ns" .= (statsAfter.gc_cpu_ns - statsBefore.gc_cpu_ns), "gc_elapsed_ns" .= (statsAfter.gc_elapsed_ns - statsBefore.gc_elapsed_ns), "throughput" .= (fromIntegral sampleCount / elapsed), "p50_us" .= percentile 0.5, "p95_us" .= percentile 0.95, "p99_us" .= percentile 0.99, "before" .= before, "after" .= after, "samples_wait_schedule_claim_resume_timer_us_writer_bytes" .= samples]+              path = output </> (show trial <> "-" <> Text.unpack mode <> ".json")+          Lazy.writeFile path (Aeson.encode artifact)+          putStrLn path++-- Equal operation-block samples after major GC, including active threads and+-- large objects. The JSONL output is streamed and never retained in a list.+heapSample :: Text -> Int -> IO Value+heapSample phase completed = do+  performMajorGC+  stats <- getRTSStats+  statuses <- listThreads >>= traverse threadStatus+  let active ThreadFinished = False+      active ThreadDied = False+      active _ = True+  pure (object ["phase" .= phase, "operations" .= completed, "live_bytes" .= stats.gc.gcdetails_live_bytes, "large_object_bytes" .= stats.gc.gcdetails_large_objects_bytes, "active_threads" .= length (filter active statuses)])++-- Returns schedule/claim/resume phase durations. The resume performs a real+-- named step and completion, then releases its advance lease.+operation :: KirokuStore -> Text -> IO (Double, Double, Double, Double, Double)+operation store wid = do+  let name = WorkflowName "index-evidence"+      identity = WorkflowId wid+      checked :: Eff '[Store, Error StoreError, IOE] a -> IO a+      checked action = runStoreIO store action >>= right+  allocationBefore <- getAllocationCounter+  a <- getMonotonicTimeNSec+  checked (runTransaction (Instance.upsertInstanceTx wid "index-evidence" 0 Instance.WfRunning Nothing))+  b <- getMonotonicTimeNSec+  claim <- checked (Instance.claimInstance "index-evidence" 30 name identity)+  unless (claim == Instance.ClaimAcquired) (fail (show claim))+  c <- getMonotonicTimeNSec+  outcome <- checked (runWorkflow name identity (step (StepName "work") (pure (1 :: Int))))+  case outcome of Completed 1 -> pure (); _ -> fail (show outcome)+  checked (Instance.releaseInstance "index-evidence" True name identity)+  d <- getMonotonicTimeNSec+  now <- getCurrentTime+  let timerId = Timer.TimerId (UUIDv5.generateNamed UUIDv5.namespaceURL (Bytes.unpack (Text.encodeUtf8 wid)))+  checked (runTransaction (Timer.scheduleTimerTx (Timer.TimerRequest timerId "index-evidence" wid now Aeson.Null)))+  claimed <- checked (Timer.claimDueTimer now)+  unless (maybe False (\row -> row.timerId == timerId) claimed) (fail "timer was not claimed")+  fired <- checked (Timer.markTimerFired timerId (EventId UUID.nil))+  unless fired (fail "timer was not fired")+  e <- getMonotonicTimeNSec+  allocationAfter <- getAllocationCounter+  pure (micros (b - a), micros (c - b), micros (d - c), micros (e - d), fromIntegral (allocationBefore - allocationAfter))+  where+    micros ns = fromIntegral ns / 1000++flush :: KirokuStore -> IO ()+flush store = use store (Session.script "SELECT pg_stat_force_next_flush(); SELECT pg_stat_clear_snapshot()")++counters :: KirokuStore -> IO (Text, Int, Int)+counters store = use store (Session.statement () (Statement.preparable "SELECT pg_current_wal_insert_lsn()::text,n_tup_upd,n_tup_hot_upd FROM pg_stat_user_tables WHERE schemaname='keiro' AND relname='keiro_workflows'" E.noParams (D.singleRow ((,,) <$> D.column (D.nonNullable D.text) <*> (fromIntegral <$> D.column (D.nonNullable D.int8)) <*> (fromIntegral <$> D.column (D.nonNullable D.int8))))))++number :: Int -> Text+number = Text.pack . show++use :: KirokuStore -> Session.Session a -> IO a+use store session = Pool.use store.pool session >>= right++right :: (Show e) => Either e a -> IO a+right = either (fail . show) pure++execute :: Text -> Text -> IO ()+execute url sql = bracket (Connection.acquire (Settings.connectionString url) >>= right) Connection.release $ \connection ->+  Connection.use connection (Session.script sql) >>= right++withMount :: KirokuStore -> Text -> Text -> (Maybe (Int, Bool -> IO Value) -> IO a) -> IO a+#ifdef KEIRO_INSPECTION_BASELINE+withMount _ _ _ action = action Nothing++inspectionModes :: [Text]+inspectionModes = ["disabled"]++withPollers :: Maybe (Int, Bool -> IO Value) -> Text -> Int -> (IO () -> IO a) -> IO (a, Int, Int)+withPollers _ _ _ action = do+  result <- action (pure ())+  pure (result, 0, 0)+#else+withMount store url mode action+  | mode == "disabled" = action Nothing+  | otherwise = withOpsInspectionPool defaultInspectionLimits (defaultConnectionSettings url) $ \pool -> do+      let inspected = store {pool = pool}+      withFeedHandle defaultOpsHttpConfig defaultFeedConfig inspected $ \feeds -> do+        let catalog+              | mode `elem` ["pm-types-many", "pm-types-many-idle"] = Map.fromList [("alias-" <> Text.justifyRight 4 '0' (number n), snd (Map.findMin jitsureiProcessManagerInspectors)) | n <- [1 .. 1000]]+              | otherwise = jitsureiProcessManagerInspectors+            hooks+              | "pm-types" `Text.isPrefixOf` mode = Ops.emptyAppHooks {Ops.processManagers = Just catalog}+              | otherwise = Ops.emptyAppHooks+        child <- opsApplicationWithFeeds defaultOpsHttpConfig feeds hooks inspected+        app <- either (fail . show) pure (composeSurfaces defaultComposeConfig [MountedSurface keiroPrefix child])+        withOpsHttpServer defaultOpsHttpServerConfig {port = 0} app $ \server -> action (Just (server.serverPort, resourceSample feeds))++resourceSample :: FeedHandle -> Bool -> IO Value+resourceSample handle awaitCleanup = loop (if awaitCleanup then 200 else 0 :: Int)+  where+    loop attempts = do+      stats <- feedStats handle+      if awaitCleanup && (stats.activeViews /= 0 || stats.activeConnections /= 0 || stats.wakePumpRunning) && attempts > 0+        then threadDelay 10000 >> loop (attempts - 1)+        else do+          when (awaitCleanup && (stats.activeViews /= 0 || stats.activeConnections /= 0 || stats.wakePumpRunning)) (fail "feed resources did not return to idle")+          pure (object ["views" .= stats.activeViews, "connections" .= stats.activeConnections, "wake_pump" .= stats.wakePumpRunning])++withPollers :: Maybe (Int, Bool -> IO Value) -> Text -> Int -> (IO () -> IO a) -> IO (a, Int, Int)+withPollers mount mode seconds action = case mount of+  Just (serverPort, _) | clients > 0 -> do+    executable <- getExecutablePath+    let child = (proc executable ["--clients", show serverPort, show clients, show views, show seconds, Text.unpack mode]) {std_in = CreatePipe, std_out = CreatePipe}+        cleanup (input, output, errors, process) = do+          terminateProcess process+          _ <- waitForProcess process+          mapM_ (maybe (pure ()) hClose) [input, output, errors]+    bracket (createProcess child) cleanup $ \case+      (Just input, Just output, _, process) -> do+        ready <- hGetLine output+        unless (ready == "ready") (fail "poll client did not finish warming its views")+        result <- action (hPutStrLn input "go" >> hFlush input)+        counts <- hGetLine output+        exit <- waitForProcess process+        unless (exit == ExitSuccess) (fail "poll client failed")+        (good, bad) <- right (Aeson.eitherDecode (LazyChar8Portable.pack counts))+        pure (result, good, bad)+      _ -> fail "missing poll client control pipes"+  _ -> do+    result <- action (pure ())+    pure (result, 0, 0)+  where+    (views, clients) = pollingShape mode++inspectionModes :: [Text]+inspectionModes = ["disabled", "idle", "active-shared", "active-distinct", "slow-reader", "churn", "pm-types-idle", "pm-types", "pm-types-many-idle", "pm-types-many"]++#endif++-- Distinct views and connection counts are independent: a single client can+-- poll all views, while many clients can share one view's authoritative route.+pollingShape :: Text -> (Int, Int)+pollingShape = \case+  "pm-types" -> (0, 4)+  "pm-types-many" -> (0, 4)+  "active-shared" -> (1, 1)+  "active-distinct" -> (32, 256)+  "slow-reader" -> (32, 256)+  "churn" -> (32, 32)+  _ -> (0, 0)

@@ -0,0 +1,56 @@+module Main (main) where++import Control.Concurrent.Async (wait)+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.IO qualified as Text+import Keiro.Ops (emptyAppHooks)+import Keiro.Ops.Env (resolveConnectionString)+import Keiro.Ops.Http+import Keiro.Ops.Http.Application (opsApplicationWithBudget)+import Keiro.Ops.Http.Config qualified as Config+import Keiro.Ops.Http.WebSocket (feedsMiddleware)+import Kiroku.Metrics qualified as Metrics+import Kiroku.Metrics.Config qualified as MetricsConfig+import Kiroku.Store.Connection (ConnectionSettingsM (..), KirokuStore (..), defaultConnectionSettings, withStore)+import Options.Applicative+import System.IO (stderr)++data Options = Options !(Maybe Text) !Text !Int ![Text]++options :: Parser Options+options =+  Options+    <$> optional (textOption (long "database-url" <> metavar "URL"))+    <*> textOption (long "bind" <> value "127.0.0.1")+    <*> option auto (long "port" <> value 9092)+    <*> many (textOption (long "allow-origin" <> metavar "ORIGIN"))+  where+    textOption modifiers = Text.pack <$> strOption modifiers++main :: IO ()+main = do+  Options database host port origins <- execParser (info (options <**> helper) (fullDesc <> progDesc "Compose inspection surfaces on a trusted network or behind an authenticating proxy"))+  connectionString <- resolveConnectionString database+  parsedOrigins <- either (fail . show) pure (traverse Metrics.allowedOrigin origins)+  let cors = Metrics.corsAllowOrigins parsedOrigins+  let config = defaultOpsHttpConfig {Config.allowedOrigins = origins}+      serverConfig = defaultOpsHttpServerConfig {bindHost = host, port}+      settings = (defaultConnectionSettings connectionString) {poolSize = config.inspectionLimits.maxConcurrentReads, statementTimeout = Just 2}+  -- An inspection-owned store and pool; the application never borrows the+  -- runtime writer's reserved connections. Each constructor runs once.+  withStore settings $ \baseStore -> withOpsInspectionPool config.inspectionLimits settings $ \pool -> do+    let store = baseStore {pool}+        metricsConfig = Metrics.defaultConfig {MetricsConfig.cors = cors}+    metrics <- Metrics.newKirokuMetrics store+    providers <- Metrics.storeServerProviders metricsConfig metrics store+    let kiroku = Metrics.combinedAppWithProviders metricsConfig metrics [] providers+    withFeedHandle config defaultFeedConfig store $ \feeds -> do+      http <- opsApplicationWithBudget config (handleBudget feeds) emptyAppHooks store+      let keiro = feedsMiddleware config feeds http+      app <-+        either (fail . Text.unpack . renderComposeError) pure $+          composeSurfaces (ComposeConfig origins) [MountedSurface keiroPrefix keiro, MountedSurface kirokuPrefix kiroku]+      withOpsHttpServer serverConfig app $ \server -> do+        Text.hPutStrLn stderr ("keiro-ops-http-compose-example: listening on " <> host <> ":" <> Text.pack (show server.serverPort) <> "; surfaces /keiro /kiroku")+        wait server.serverThread

@@ -0,0 +1,228 @@+cabal-version: 3.0+name: keiro-ops-http+version: 0.22.0.0+synopsis:+  Bounded HTTP transport for the Keiro operational command tree++description:+  An embeddable WAI application, Warp runner, and database-only executable.+  Exposes the supported command tree with bounded reads and confirmed mutations.++license: BSD-3-Clause+license-file: LICENSE+author: Nadeem Bitar+maintainer: nadeem@gmail.com+copyright: 2026 Nadeem Bitar+category: Operations+homepage: https://github.com/shinzui/keiro#readme+bug-reports: https://github.com/shinzui/keiro/issues+build-type: Simple+tested-with: ghc >=9.12 && <9.13+extra-doc-files: CHANGELOG.md++source-repository head+  type: git+  location: https://github.com/shinzui/keiro.git++common warnings+  ghc-options:+    -Wall+    -Wcompat+    -Widentities+    -Wincomplete-record-updates+    -Wincomplete-uni-patterns+    -Wpartial-fields+    -Wredundant-constraints++common shared+  default-language: GHC2024+  default-extensions:+    BlockArguments+    DeriveAnyClass+    DuplicateRecordFields+    ImportQualifiedPost+    LambdaCase+    OverloadedLabels+    OverloadedRecordDot+    OverloadedStrings++library+  import: warnings, shared+  hs-source-dirs: src+  exposed-modules:+    Keiro.Ops.Http+    Keiro.Ops.Http.Application+    Keiro.Ops.Http.Budget+    Keiro.Ops.Http.Compose+    Keiro.Ops.Http.Config+    Keiro.Ops.Http.Cors+    Keiro.Ops.Http.Feed+    Keiro.Ops.Http.Pool+    Keiro.Ops.Http.Protocol+    Keiro.Ops.Http.Route+    Keiro.Ops.Http.Server+    Keiro.Ops.Http.Views+    Keiro.Ops.Http.WebSocket++  other-modules:+    Keiro.Ops.Http.Feed.Driver+    Keiro.Ops.Http.Feed.Queue++  build-depends:+    aeson >=2.2.2 && <2.3,+    async >=2.2 && <2.3,+    base >=4.21 && <5,+    bytestring >=0.12 && <0.13,+    containers >=0.6 && <0.8,+    effectful >=2.6 && <2.8,+    hasql >=1.10 && <1.11,+    hasql-pool >=1.2 && <1.5,+    http-types >=0.12 && <0.13,+    keiro ^>=0.22.0.0,+    keiro-migrations ^>=0.22.0.0,+    keiro-ops ^>=0.22.0.0,+    kiroku-store ^>=0.12.0.0,+    network >=3.1 && <3.3,+    network-uri >=2.6.4.2 && <2.7,+    optparse-applicative >=0.19 && <0.20,+    stm >=2.5 && <2.6,+    text >=2.1 && <2.2,+    time >=1.12 && <1.15,+    uuid >=1.3 && <1.4,+    wai >=3.2 && <3.3,+    wai-websockets >=3.0 && <3.1,+    warp >=3.4 && <3.5,+    websockets >=0.13 && <0.14,++executable keiro-ops-http+  import: warnings, shared+  hs-source-dirs: app+  main-is: Main.hs+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N++  build-depends:+    async >=2.2 && <2.3,+    base >=4.21 && <5,+    hasql >=1.10 && <1.11,+    keiro-migrations ^>=0.22.0.0,+    keiro-ops ^>=0.22.0.0,+    keiro-ops-http,+    kiroku-store ^>=0.12.0.0,+    optparse-applicative >=0.19 && <0.20,+    text >=2.1 && <2.2,++flag compose-example+  description: Build the composed inspection example with the sibling metrics surface.+  default: False+  manual: True++executable keiro-ops-http-compose-example+  import: warnings, shared+  hs-source-dirs: example+  main-is: Compose.hs+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N++  if !flag(compose-example)+    buildable: False+  build-depends:+    async >=2.2 && <2.3,+    base >=4.21 && <5,+    keiro-ops ^>=0.22.0.0,+    keiro-ops-http,+    kiroku-metrics ^>=0.4.0.0,+    kiroku-store ^>=0.12.0.0,+    optparse-applicative >=0.19 && <0.20,+    text >=2.1 && <2.2,++benchmark inspection-interference+  import: warnings, shared+  type: exitcode-stdio-1.0+  hs-source-dirs: bench+  main-is: InspectionInterference.hs+  other-modules: InspectionClients+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N8++  build-depends:+    aeson >=2.2 && <2.3,+    async >=2.2 && <2.3,+    base >=4.21 && <5,+    bytestring >=0.12 && <0.13,+    containers >=0.6 && <0.8,+    directory >=1.3 && <1.4,+    effectful >=2.6 && <2.8,+    filepath >=1.4 && <1.6,+    hasql >=1.10 && <1.11,+    hasql-pool >=1.2 && <1.5,+    http-client >=0.7 && <0.8,+    http-types >=0.12 && <0.13,+    jitsurei,+    keiro ^>=0.22.0.0,+    keiro-ops ^>=0.22.0.0,+    keiro-ops-http,+    keiro-test-support ^>=0.22.0.0,+    kiroku-store ^>=0.12.0.0,+    process >=1.6 && <1.7,+    text >=2.1 && <2.2,+    time >=1.12 && <1.15,+    uuid >=1.3 && <1.4,+    websockets >=0.13 && <0.14,++test-suite keiro-ops-http-test+  import: warnings, shared+  type: exitcode-stdio-1.0+  hs-source-dirs: test+  main-is: Main.hs+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N++  other-modules:+    Test.ComposeSpec+    Test.FeedSpec+    Test.WebSocketSpec++  build-tool-depends: keiro-ops:keiro-ops+  build-depends:+    aeson >=2.2.2 && <2.3,+    async >=2.2 && <2.3,+    base >=4.21 && <5,+    bytestring >=0.12 && <0.13,+    containers >=0.6 && <0.8,+    effectful >=2.6 && <2.8,+    hasql >=1.10 && <1.11,+    hasql-pool >=1.2 && <1.5,+    hspec >=2.11,+    http-client >=0.7 && <0.8,+    http-types >=0.12 && <0.13,+    jitsurei,+    keiro ^>=0.22.0.0,+    keiro-migrations ^>=0.22.0.0,+    keiro-ops ^>=0.22.0.0,+    keiro-ops-http,+    keiro-pgmq ^>=0.22.0.0,+    keiro-test-support ^>=0.22.0.0,+    kiroku-metrics ^>=0.4.0.0,+    kiroku-store ^>=0.12.0.0,+    network >=3.1 && <3.3,+    optparse-applicative >=0.19 && <0.20,+    pgmq-migration >=0.7.0.0 && <0.8,+    process >=1.6 && <1.7,+    stm >=2.5 && <2.6,+    text >=2.1 && <2.2,+    time >=1.12 && <1.15,+    uuid >=1.3 && <1.4,+    vector >=0.13 && <0.14,+    wai >=3.2 && <3.3,+    wai-extra >=3.1 && <3.2,+    warp >=3.4 && <3.5,+    websockets >=0.13 && <0.14,

@@ -0,0 +1,23 @@+module Keiro.Ops.Http+  ( opsApplication,+    opsApplicationWithFeeds,+    module Keiro.Ops.Http.Feed,+    module Keiro.Ops.Http.Compose,+    OpsHttpError (..),+    errorResponse,+    module Keiro.Ops.Http.Config,+    module Keiro.Ops.Http.Server,+    withOpsInspectionPool,+    corsMiddleware,+  )+where++import Keiro.Ops.Http.Application+import Keiro.Ops.Http.Compose+import Keiro.Ops.Http.Config+import Keiro.Ops.Http.Cors+import Keiro.Ops.Http.Pool+import Keiro.Ops.Http.Server++import Keiro.Ops.Http.Feed+import Keiro.Ops.Http.WebSocket (opsApplicationWithFeeds)

@@ -0,0 +1,264 @@+module Keiro.Ops.Http.Application+  ( opsApplication,+    opsApplicationWithBudget,+    OpsHttpError (..),+    errorResponse,+    OpsHttpOperations (..),+    opsApplicationWithOperations,+  )+where++import Control.Exception (SomeAsyncException, SomeException, displayException, evaluate, fromException, throwIO, try)+import Data.Aeson (Value (..), object, (.=))+import Data.Aeson qualified as Aeson+import Data.ByteString qualified as ByteString+import Data.ByteString.Char8 qualified as Char8+import Data.ByteString.Lazy qualified as LazyByteString+import Data.Char (toLower)+import Data.IORef+import Data.Map.Strict qualified as Map+import Data.Sequence qualified as Seq+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text+import Hasql.Pool qualified as Pool+import Keiro.Migrations.SchemaCheck (renderSchemaDrift, verifyExpectedSchemaSession)+import Keiro.Ops (AppHooks, Command, commandParser, isMutation, runCommandWithReadEncoding, validateCommand)+import Keiro.Ops.Env (OpsEnv (..), OutputMode (Json))+import Keiro.Ops.Http.Budget+import Keiro.Ops.Http.Config+import Keiro.Ops.Http.Cors+import Keiro.Ops.Http.Route+import Keiro.Ops.Render+import Keiro.Ops.WorkflowHeaderCache (newWorkflowHeaderCache)+import Kiroku.Store.Connection (KirokuStore (..))+import Network.HTTP.Types+import Network.Wai+import System.Exit (ExitCode (..))++data ResponsePlan = Immediate Response | Inspected Bool (IO Response)++data OpsHttpError = OpsHttpError+  { status :: !Status,+    code :: !Text,+    message :: !Text,+    details :: !(Maybe Value)+  }++jsonHeaders :: ResponseHeaders+jsonHeaders = [(hContentType, "application/json; charset=utf-8")]++errorResponse :: OpsHttpError -> Response+errorResponse err = responseLBS err.status jsonHeaders (Aeson.encode (errorValue err))++errorValue :: OpsHttpError -> Value+errorValue err = object ["error" .= object (["code" .= err.code, "message" .= err.message] <> maybe [] (\value -> ["details" .= value]) err.details)]++failure :: Status -> Text -> Text -> Response+failure status code message = errorResponse (OpsHttpError status code message Nothing)++-- | A host/test seam for supported verification and execution, never a route+-- registry. Production mounting uses 'opsApplication'.+data OpsHttpOperations = OpsHttpOperations+  { verifySchema :: IO (Either Text [Text]),+    executeCommand :: OpsEnv -> Command -> IO OpsOutcome+  }++-- | Allocate once per mount. The supplied store must have statement_timeout+-- configured on EVERY connection, at or below inspectionLimits. Use a dedicated+-- pool; changing a borrowed connection here cannot constrain another handler's+-- borrow and would change writer policy on a shared pool.+opsApplication :: OpsHttpConfig -> AppHooks -> KirokuStore -> IO Application+opsApplication config hooks store = do+  budget <- newInspectionBudget config.inspectionLimits+  opsApplicationWithBudget config budget hooks store++-- | Internal feed integration: HTTP and refresh work consume the same budget.+opsApplicationWithBudget :: OpsHttpConfig -> InspectionBudget -> AppHooks -> KirokuStore -> IO Application+opsApplicationWithBudget config budget hooks store = do+  cache <- newWorkflowHeaderCache config.inspectionLimits.maxResponseBytes $ \value ->+    fmap LazyByteString.toStrict <$> encodeBounded config.inspectionLimits.maxResponseBytes value+  let verify = do+        result <- Pool.use store.pool verifyExpectedSchemaSession+        pure $ case result of+          Left err -> Left (Text.pack (show err))+          Right (Left err) -> Left (Text.pack (show err))+          Right (Right drifts) -> Right (map renderSchemaDrift drifts)+  let encodeDirect encoding = fmap LazyByteString.toStrict <$> encodeBoundedEncoding config.inspectionLimits.maxResponseBytes encoding+  opsApplicationPrepared config budget hooks store verify (runCommandWithReadEncoding encodeDirect cache hooks)++opsApplicationWithOperations :: OpsHttpConfig -> InspectionBudget -> AppHooks -> KirokuStore -> OpsHttpOperations -> IO Application+opsApplicationWithOperations config budget hooks store operations =+  opsApplicationPrepared config budget hooks store operations.verifySchema $ \env command -> do+    outcome <- operations.executeCommand env command+    pure (outcome, Nothing)++-- The prepared bytes are paired with the exact successful read outcome by the+-- owning runner. They never change the public OpsResult or mutation protocol.+opsApplicationPrepared :: OpsHttpConfig -> InspectionBudget -> AppHooks -> KirokuStore -> IO (Either Text [Text]) -> (OpsEnv -> Command -> IO (OpsOutcome, Maybe ByteString.ByteString)) -> IO Application+opsApplicationPrepared config budget hooks store verify execute = do+  parsedReads <- newIORef (Map.empty, Seq.empty)+  let parser = commandParser hooks+      flags = flagNames parser+      parse arguments = do+        (known, _) <- readIORef parsedReads+        case Map.lookup arguments known of+          Just command -> pure (Parsed command)+          Nothing -> case parseArgumentsWithParser parser arguments of+            Parsed command+              | not (isMutation command),+                Right () <- validateCommand command -> do+                  -- Cache only pure parsing, never observations or authorization.+                  -- FIFO capacity and the request-byte cap bound retained inputs.+                  atomicModifyIORef' parsedReads $ \(commands, order) ->+                    if Map.member arguments commands+                      then ((commands, order), ())+                      else+                        let (kept, queue) = case Seq.viewl order of+                              oldest Seq.:< rest | Map.size commands >= 64 -> (Map.delete oldest commands, rest)+                              _ -> (commands, order)+                         in ((Map.insert arguments command kept, queue Seq.|> arguments), ())+                  pure (Parsed command)+            outcome -> pure outcome+  -- Initialize static command metadata during mounting, before requests arrive.+  _ <- evaluate (flags [])+  pure $ corsMiddleware config.allowedOrigins $ \request respond -> do+    result <- trySync (dispatch parse flags request)+    case result of+      Left _ -> respond internalError+      Right (Immediate response) -> respond response+      Right (Inspected mutation action) -> do+        let deadline = if mutation then limits.mutationDeadlineSeconds else limits.readDeadlineSeconds+        delivered <- withInspectionAdmission budget $ do+          computed <- withInspectionDeadline deadline (trySync action)+          -- Network exceptions must propagate; never attempt a second response.+          withInspectionDelivery limits.deliveryDeadlineSeconds $ respond $ case computed of+            Left problem -> budgetFailure mutation problem+            Right (Left _) -> internalError+            Right (Right response) -> response+        case delivered of+          Left problem -> respond (budgetFailure mutation problem)+          Right sent -> pure sent+  where+    limits = config.inspectionLimits+    dispatch parse flags request+      | inputBytes request > limits.maxRequestBytes = immediate (failure status413 "inspection_request_too_large" "request path and query exceed the host limit")+      | pathInfo request == [] = immediate (failure status404 "not_found" "no inspection command at this path")+      | pathInfo request == ["health"] =+          if requestMethod request == methodGet+            then pure (Inspected False health)+            else immediate (methodFailure methodGet)+      | otherwise = case translate flags request of+          Left message -> immediate (failure status400 "invalid_request" message)+          Right arguments -> do+            parsed <- parse arguments+            case parsed of+              Rejected message -> immediate (errorResponse (OpsHttpError status400 "invalid_request" message (Just (object ["arguments" .= arguments]))))+              Parsed command | Left (code, message) <- validateCommand command -> immediate (failure status400 code message)+              Parsed command -> do+                let mutation = isMutation command+                    method = if mutation then methodPost else methodGet+                if requestMethod request /= method+                  then immediate (methodFailure method)+                  else+                    if mutation && (any (\(key, value) -> key == "Origin" && ByteString.length value > limits.maxRequestBytes) (requestHeaders request) || not (originAllowed config.allowedOrigins (isSecure request) (requestHeaders request)))+                      then immediate (failure status403 "origin_not_allowed" "mutation origin is not allowed")+                      else+                        if mutation && config.mutations == MutationsDisabled+                          then immediate (failure status403 "mutations_disabled" "this surface was mounted without mutations")+                          else budgetedCommand request command mutation+    translate flags request = do+      query <- either (Left . Text.pack . displayException) Right (traverse decodeQuery (queryString request))+      either (Left . renderRouteError) Right (translateRequest (flags (pathInfo request)) (pathInfo request) query)+    decodeQuery (key, value) = (,) <$> Text.decodeUtf8' key <*> traverse Text.decodeUtf8' value+    inputBytes request = ByteString.length (rawPathInfo request) + ByteString.length (rawQueryString request)+    methodFailure method = mapResponseHeaders ((hAllow, method) :) (failure status405 "method_not_allowed" "method does not match the command classification")+    budgetedCommand request command mutation =+      pure $ Inspected mutation $ do+        confirmation <- if mutation then readConfirmation request else pure (Right False)+        case confirmation of+          Left err -> pure err+          Right confirmed -> do+            verified <- verify+            case verified of+              Left message -> pure (failure status500 "schema_verification_failed" message)+              Right drift+                | not (null drift) && config.schemaDrift == RefuseOnDrift ->+                    boundedJson status409 [] (errorValue (OpsHttpError status409 "schema_drift" "the live schema differs from this binary; refusing to serve until they agree" (Just (object ["drift" .= drift]))))+                | otherwise -> do+                    let env = OpsEnv store Json (mutation && confirmed) drift (config.schemaDrift == AllowDrift)+                    (outcome, prepared) <- execute env command+                    renderOutcome mutation drift prepared outcome+    immediate = pure . Immediate+    internalError = failure status500 "internal_error" "inspection failed"+    budgetFailure mutation = \case+      InspectionBusy -> failure status503 "inspection_busy" "all inspection slots are in use; retry later"+      InspectionTimedOut+        | mutation -> failure status504 "mutation_outcome_unknown" "deadline exceeded; the mutation may have committed; inspect state before deciding whether to retry"+        | otherwise -> failure status504 "inspection_timeout" "inspection exceeded the host deadline"+    health = do+      result <- verify+      case result of+        Left message -> boundedJson status503 [] (object ["status" .= ("unavailable" :: Text), "message" .= message])+        Right drift ->+          boundedJson+            (if null drift || config.schemaDrift == AllowDrift then status200 else status503)+            (driftHeaders drift)+            (object ["status" .= (if null drift then "ok" else "schema_drift" :: Text), "schema_drift" .= drift, "mutations_enabled" .= (config.mutations == MutationsEnabled)])+    driftHeaders drift = [("Keiro-Ops-Schema-Drift", Char8.pack (show (length drift))) | not (null drift) && config.schemaDrift == AllowDrift]+    boundedJson status headers value = do+      encoded <- encodeBounded limits.maxResponseBytes value+      pure $ case encoded of+        Nothing -> failure status413 "inspection_result_too_large" "encoded result exceeds the host limit; narrow the read"+        Just body -> responseLBS status (jsonHeaders <> headers) body+    renderOutcome mutation drift prepared = \case+      InspectionFailed code message ->+        let status+              | code == "cursor_generation_mismatch" = status409+              | code == "inspection_result_too_large" = status413+              | otherwise = status400+         in boundedJson status [] (errorValue (OpsHttpError status code message Nothing))+      Failed message -> boundedJson status422 [] (errorValue (OpsHttpError status422 "command_failed" message Nothing))+      PreviewRequired result invocation -> boundedJson status200 (driftHeaders drift) (object ["outcome" .= ("preview" :: Text), "result" .= result.jsonValue, "cli_reinvocation" .= invocation])+      Succeeded result -> success result ExitSuccess+      SucceededWithExit result exitCode -> success result exitCode+      where+        success result exitCode =+          let exitFields = case exitCode of ExitSuccess -> []; ExitFailure n -> ["exit_code" .= n]+              exitHeaders = case exitCode of ExitSuccess -> []; ExitFailure n -> [("Keiro-Ops-Exit-Code", Char8.pack (show n))]+              value = if mutation then object (["outcome" .= ("executed" :: Text), "result" .= result.jsonValue] <> exitFields) else result.jsonValue+           in case (mutation, prepared) of+                (False, Just body)+                  | ByteString.length body <= limits.maxResponseBytes ->+                      pure (responseLBS status200 (jsonHeaders <> driftHeaders drift <> exitHeaders) (LazyByteString.fromStrict body))+                  | otherwise ->+                      pure (failure status413 "inspection_result_too_large" "encoded result exceeds the host limit; narrow the read")+                _ -> boundedJson status200 (driftHeaders drift <> if mutation then [] else exitHeaders) value++readConfirmation :: Request -> IO (Either Response Bool)+readConfirmation request = go 0 []+  where+    go size chunks = do+      chunk <- getRequestBodyChunk request+      let nextSize = size + ByteString.length chunk+      if nextSize > 1024+        then pure (Left (failure status413 "inspection_request_too_large" "confirmation body exceeds 1 KiB"))+        else+          if ByteString.null chunk+            then pure (parse (LazyByteString.fromChunks (reverse chunks)))+            else go nextSize (chunk : chunks)+    jsonContentType = case [value | (key, value) <- requestHeaders request, key == hContentType] of+      [value] -> Char8.map toLower (Char8.strip (Char8.takeWhile (/= ';') value)) == "application/json"+      _ -> False+    parse body+      | LazyByteString.null body = Right False+      | not jsonContentType = Left (failure status415 "unsupported_media_type" "a nonempty confirmation body requires application/json")+      | Aeson.decode body == Just (object ["confirm" .= ("execute" :: Text)]) = Right True+      | otherwise = Left (failure status400 "invalid_confirmation" "expected an empty body or exactly {\"confirm\":\"execute\"}")++trySync :: IO a -> IO (Either SomeException a)+trySync action = do+  result <- try action+  case result of+    Left exception | Just (_ :: SomeAsyncException) <- fromException exception -> throwIO exception+    _ -> pure result

@@ -0,0 +1,96 @@+module Keiro.Ops.Http.Budget+  ( InspectionBudget,+    BudgetFailure (..),+    newInspectionBudget,+    withInspectionAdmission,+    withInspectionDeadline,+    withInspectionDelivery,+    withInspectionBudget,+    encodeBounded,+    encodeBoundedEncoding,+  )+where++import Control.Concurrent.STM+import Control.Exception (Exception, bracket, evaluate, mask, throwIO)+import Data.Aeson (Value, encode)+import Data.Aeson.Encoding (Encoding, encodingToLazyByteString)+import Data.ByteString qualified as ByteString+import Data.ByteString.Lazy qualified as LazyByteString+import Keiro.Ops.Http.Config+import System.Timeout (timeout)++-- | One mount's admission counter; feed refreshes share this exact handle.+data InspectionBudget = InspectionBudget !Int !(TVar Int)++data BudgetFailure = InspectionBusy | InspectionTimedOut+  deriving stock (Eq, Show)++newInspectionBudget :: InspectionLimits -> IO InspectionBudget+newInspectionBudget limits+  | any (<= 0) values || any (> maxBound `div` 1000000) deadlines =+      ioError (userError "inspection limits must be positive and deadlines representable in microseconds")+  | otherwise = InspectionBudget limits.maxConcurrentReads <$> newTVarIO 0+  where+    deadlines = [limits.readDeadlineSeconds, limits.mutationDeadlineSeconds, limits.deliveryDeadlineSeconds]+    values = deadlines <> [limits.maxConcurrentReads, limits.statementTimeoutMillis, limits.maxResponseBytes, limits.maxRequestBytes]++-- | No waiting queue. Timed-out or cancelled work unwinds before releasing the+-- permit. Exceptions from work are preserved.+withInspectionBudget :: InspectionBudget -> Int -> IO a -> IO (Either BudgetFailure a)+withInspectionBudget budget seconds action = do+  result <- withInspectionAdmission budget (withInspectionDeadline seconds action)+  pure (either Left id result)++-- | Keep the permit through response delivery as well as computation, so slow+-- clients cannot retain an unbounded number of encoded response bodies.+withInspectionAdmission :: InspectionBudget -> IO a -> IO (Either BudgetFailure a)+withInspectionAdmission (InspectionBudget capacity active) action = mask $ \restore -> do+  admitted <- atomically $ do+    count <- readTVar active+    if count >= capacity then pure False else writeTVar active (count + 1) >> pure True+  if not admitted+    then pure (Left InspectionBusy)+    else bracket (pure ()) (const (atomically (modifyTVar' active (subtract 1)))) $ \_ -> do+      Right <$> restore action++-- | Bound computation separately from delivery: once headers are sent, a+-- computation timeout response can no longer be substituted safely.+withInspectionDeadline :: Int -> IO a -> IO (Either BudgetFailure a)+withInspectionDeadline seconds action = do+  -- The threaded RTS timer manager delivers an async exception on expiry;+  -- successful reads reuse the request thread rather than allocating two Async+  -- threads and fresh stacks per request. timeout owns timer cleanup as well as+  -- unwinding the action before returning Nothing.+  outcome <- timeout (seconds * 1000000) action+  pure (maybe (Left InspectionTimedOut) Right outcome)++-- Private termination signal: after delivery starts no second response is safe.+data DeliveryExpired = DeliveryExpired deriving stock (Show)++instance Exception DeliveryExpired++withInspectionDelivery :: Int -> IO a -> IO a+withInspectionDelivery seconds action = do+  result <- timeout (seconds * 1000000) action+  maybe (throwIO DeliveryExpired) pure result++-- | Demand at most the cap plus one byte of Aeson's lazy output. Force the+-- bounded body before leaving the deadline; no partial result is served.+encodeBounded :: Int -> Value -> IO (Maybe LazyByteString.ByteString)+encodeBounded limit = boundBytes limit . encode++-- | The same byte boundary for owner-provided direct JSON encodings.+encodeBoundedEncoding :: Int -> Encoding -> IO (Maybe LazyByteString.ByteString)+encodeBoundedEncoding limit = boundBytes limit . encodingToLazyByteString++boundBytes :: Int -> LazyByteString.ByteString -> IO (Maybe LazyByteString.ByteString)+boundBytes limit encoded = do+  let prefix = LazyByteString.take (fromIntegral limit + 1) encoded+  size <- evaluate (LazyByteString.length prefix)+  if size > fromIntegral limit+    then pure Nothing+    else do+      body <- evaluate (LazyByteString.toStrict prefix)+      _ <- evaluate (ByteString.length body)+      pure (Just (LazyByteString.fromStrict body))

@@ -0,0 +1,110 @@+-- | Mount already allocated applications. This router owns no runtime resources.+module Keiro.Ops.Http.Compose+  ( SurfacePrefix,+    mkSurfacePrefix,+    surfacePrefixText,+    keiroPrefix,+    kirokuPrefix,+    shibuyaPrefix,+    pgmqPrefix,+    MountedSurface (..),+    ComposeConfig (..),+    defaultComposeConfig,+    ComposeError (..),+    renderComposeError,+    composeSurfaces,+    stripSurfacePrefix,+    composedCorsMiddleware,+  )+where++import Data.Aeson (encode, object, (.=))+import Data.ByteString qualified as Bytes+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as Text+import Keiro.Ops.Http.Application (OpsHttpError (..), errorResponse)+import Keiro.Ops.Http.Cors (corsMiddleware)+import Network.HTTP.Types+import Network.Wai++newtype SurfacePrefix = SurfacePrefix Text+  deriving stock (Eq, Ord, Show)++mkSurfacePrefix :: Text -> Either ComposeError SurfacePrefix+mkSurfacePrefix value+  | not (Text.null value), not (Text.isPrefixOf "-" value), Text.all valid value = Right (SurfacePrefix value)+  | otherwise = Left (InvalidPrefix value)+  where+    valid c = c >= 'a' && c <= 'z' || c >= '0' && c <= '9' || c == '-'++surfacePrefixText :: SurfacePrefix -> Text+surfacePrefixText (SurfacePrefix value) = value++keiroPrefix, kirokuPrefix, shibuyaPrefix, pgmqPrefix :: SurfacePrefix+keiroPrefix = SurfacePrefix "keiro"+kirokuPrefix = SurfacePrefix "kiroku"+shibuyaPrefix = SurfacePrefix "shibuya"+pgmqPrefix = SurfacePrefix "pgmq"++data MountedSurface = MountedSurface+  { prefix :: !SurfacePrefix,+    application :: !Application+  }++newtype ComposeConfig = ComposeConfig {allowedOrigins :: [Text]}+  deriving stock (Eq, Show)++defaultComposeConfig :: ComposeConfig+defaultComposeConfig = ComposeConfig []++data ComposeError = InvalidPrefix !Text | DuplicatePrefix !Text | NoSurfaces+  deriving stock (Eq, Show)++renderComposeError :: ComposeError -> Text+renderComposeError = \case+  InvalidPrefix value -> "invalid surface prefix: " <> value+  DuplicatePrefix value -> "duplicate surface prefix: " <> value+  NoSurfaces -> "at least one inspection surface must be mounted"++composeSurfaces :: ComposeConfig -> [MountedSurface] -> Either ComposeError Application+composeSurfaces _ [] = Left NoSurfaces+composeSurfaces config surfaces = do+  mounts <- foldl add (Right Map.empty) surfaces+  let names = map (surfacePrefixText . (.prefix)) surfaces+      index =+        responseLBS status200 [(hContentType, "application/json; charset=utf-8")] $+          encode (object ["surfaces" .= map (\name -> object ["prefix" .= name, "path" .= ("/" <> name)]) names])+      router req respond = case pathInfo req of+        [] | requestMethod req == methodGet -> respond index+        [] -> respond $ mapResponseHeaders ((hAllow, "GET") :) $ errorResponse (OpsHttpError status405 "method_not_allowed" "the composed root answers GET only" Nothing)+        name : _ -> case Map.lookup name mounts of+          Just surface -> case stripSurfacePrefix surface.prefix req of+            Just relative -> surface.application relative respond+            Nothing -> respond (errorResponse (OpsHttpError status400 "invalid_path" "inconsistent encoded request path" Nothing))+          Nothing -> respond (errorResponse (OpsHttpError status404 "surface_not_mounted" ("no surface is mounted at /" <> name) (Just (object ["mounted" .= names]))))+  pure (composedCorsMiddleware config.allowedOrigins router)+  where+    add result surface = do+      mounts <- result+      let name = surfacePrefixText surface.prefix+      if Map.member name mounts then Left (DuplicatePrefix name) else Right (Map.insert name surface mounts)++-- | Strip exactly one raw segment, preserving every remaining byte and query.+-- Only the mount root's trailing slash is normalized. Inconsistent synthetic+-- WAI requests are rejected rather than routed to a different target.+stripSurfacePrefix :: SurfacePrefix -> Request -> Maybe Request+stripSurfacePrefix prefix req = case pathInfo req of+  name : rest+    | name == surfacePrefixText prefix,+      Bytes.isPrefixOf "/" (rawPathInfo req),+      decodePathSegments (rawPathInfo req) == pathInfo req ->+        let remainder = Bytes.dropWhile (/= 47) (Bytes.drop 1 (rawPathInfo req))+            raw = if Bytes.null remainder then "/" else remainder+            segments = if rest == [""] then [] else rest+         in Just req {pathInfo = segments, rawPathInfo = raw}+  _ -> Nothing++-- | Use the same exact-origin, duplicate-safe policy as standalone HTTP.+composedCorsMiddleware :: [Text] -> Middleware+composedCorsMiddleware = corsMiddleware

@@ -0,0 +1,56 @@+module Keiro.Ops.Http.Config+  ( MutationPolicy (..),+    SchemaDriftPolicy (..),+    InspectionLimits (..),+    defaultInspectionLimits,+    OpsHttpConfig (..),+    defaultOpsHttpConfig,+    OpsHttpServerConfig (..),+    defaultOpsHttpServerConfig,+  )+where++import Data.Text (Text)++data MutationPolicy = MutationsDisabled | MutationsEnabled+  deriving stock (Eq, Show)++data SchemaDriftPolicy = RefuseOnDrift | AllowDrift+  deriving stock (Eq, Show)++-- | All limits must be positive. Deadlines cover verification, execution, and+-- encoding. A mutation timeout has an unknown commit outcome and is not retried.+data InspectionLimits = InspectionLimits+  { maxConcurrentReads :: !Int,+    readDeadlineSeconds :: !Int,+    mutationDeadlineSeconds :: !Int,+    deliveryDeadlineSeconds :: !Int,+    statementTimeoutMillis :: !Int,+    maxResponseBytes :: !Int,+    maxRequestBytes :: !Int+  }+  deriving stock (Eq, Show)++defaultInspectionLimits :: InspectionLimits+defaultInspectionLimits = InspectionLimits 4 5 30 30 2000 1048576 16384++data OpsHttpConfig = OpsHttpConfig+  { mutations :: !MutationPolicy,+    schemaDrift :: !SchemaDriftPolicy,+    allowedOrigins :: ![Text],+    inspectionLimits :: !InspectionLimits+  }+  deriving stock (Eq, Show)++defaultOpsHttpConfig :: OpsHttpConfig+defaultOpsHttpConfig = OpsHttpConfig MutationsDisabled RefuseOnDrift [] defaultInspectionLimits++data OpsHttpServerConfig = OpsHttpServerConfig+  { bindHost :: !Text,+    port :: !Int,+    requestTimeoutSeconds :: !Int+  }+  deriving stock (Eq, Show)++defaultOpsHttpServerConfig :: OpsHttpServerConfig+defaultOpsHttpServerConfig = OpsHttpServerConfig "127.0.0.1" 9092 300

@@ -0,0 +1,67 @@+module Keiro.Ops.Http.Cors (corsMiddleware, originAllowed) where++import Control.Monad (guard)+import Data.ByteString (ByteString)+import Data.ByteString.Char8 qualified as Bytes+import Data.Char (isAlphaNum, isAscii, toLower)+import Data.Text (Text)+import Data.Text.Encoding (encodeUtf8)+import Network.HTTP.Types+import Network.URI (URI (..), URIAuth (..), parseURI)+import Network.Wai+import Text.Read (readMaybe)++data Origin = Origin String String Int deriving stock (Eq)++-- A serialized HTTP origin contains only scheme, host and optional port.+-- No user information, path, query, fragment, opaque/null or origin list.+parseOrigin :: ByteString -> Maybe Origin+parseOrigin bytes = do+  uri <- parseURI (Bytes.unpack bytes)+  let scheme = map toLower uri.uriScheme+  defaultPort <- case scheme of "http:" -> Just 80; "https:" -> Just 443; _ -> Nothing+  authority <- uri.uriAuthority+  guard (null uri.uriPath && null uri.uriQuery && null uri.uriFragment && null authority.uriUserInfo)+  let host = map toLower authority.uriRegName+  guard (not (null host) && all (\c -> isAscii c && (isAlphaNum c || c `elem` (".-:[]" :: String))) host)+  port <- case authority.uriPort of+    "" -> Just defaultPort+    ':' : digits | not (null digits) && all (\c -> c >= '0' && c <= '9') digits -> readMaybe digits+    _ -> Nothing+  guard (port >= 0 && port <= 65535)+  pure (Origin scheme host port)++allowedCorsOrigin :: [Text] -> RequestHeaders -> Maybe ByteString+allowedCorsOrigin origins headers = do+  [raw] <- pure [value | (key, value) <- headers, key == "Origin"]+  origin <- parseOrigin raw+  guard (any ((== Just origin) . parseOrigin . encodeUtf8) origins)+  pure raw++-- HTTP mutations and WebSocket upgrades use exactly the same origin policy.+-- Forwarded headers are deliberately excluded: TLS proxies allow-list HTTPS.+originAllowed :: [Text] -> Bool -> RequestHeaders -> Bool+originAllowed origins secure headers = case [value | (key, value) <- headers, key == "Origin"] of+  [] -> True+  [raw] -> case parseOrigin raw of+    Nothing -> False+    Just origin ->+      any ((== Just origin) . parseOrigin . encodeUtf8) origins+        || case [host | (key, host) <- headers, key == hHost] of+          [host] -> parseOrigin ((if secure then "https://" else "http://") <> host) == Just origin+          _ -> False+  _ -> False++-- No credentials. Composition preserves already supplied surface CORS headers.+corsMiddleware :: [Text] -> Middleware+corsMiddleware [] = id+corsMiddleware origins = \app request respond ->+  case allowedCorsOrigin origins (requestHeaders request) of+    Just origin ->+      let add headers+            | any ((== "Access-Control-Allow-Origin") . fst) headers = headers+            | otherwise = ("Access-Control-Allow-Origin", origin) : ("Vary", "Origin") : headers+       in if requestMethod request == methodOptions && any ((== "Access-Control-Request-Method") . fst) (requestHeaders request)+            then respond (responseLBS status204 (add [("Access-Control-Allow-Methods", "GET, POST, OPTIONS"), ("Access-Control-Allow-Headers", maybe "Content-Type" id (lookup "Access-Control-Request-Headers" (requestHeaders request))), ("Access-Control-Max-Age", "600")]) "")+            else app request (respond . mapResponseHeaders add)+    Nothing -> app request respond

@@ -0,0 +1,235 @@+-- | Shared, bounded inspection drivers. A mounted idle handle owns no wake+-- cursor and performs no recurring reads. Close the handle before its store.+module Keiro.Ops.Http.Feed+  ( FeedConfig (..),+    defaultFeedConfig,+    Driver,+    DriverFailure (..),+    Subscriber,+    newDriver,+    stopDriver,+    subscribe,+    unsubscribe,+    nextFrame,+    driverSnapshot,+    FeedAcquireError (..),+    FeedHandle,+    FeedStats (..),+    feedStats,+    newFeedHandle,+    closeFeedHandle,+    withFeedHandle,+    acquireDriver,+    releaseDriver,+    handleHttpConfig,+    handleFeedConfig,+    handleBudget,+    handleClosed,+    withFeedConnection,+    FeedSchemaCheck (..),+    checkFeedSchema,+  )+where++import Control.Concurrent.Async (Async, asyncWithUnmask, uninterruptibleCancel, wait)+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Exception (bracket, finally, mask, onException, uninterruptibleMask_)+import Control.Monad (forever, when)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Word (Word64)+import Hasql.Pool qualified as Pool+import Keiro.Migrations.SchemaCheck (verifyExpectedSchemaSession)+import Keiro.Ops.Http.Budget+import Keiro.Ops.Http.Config+import Keiro.Ops.Http.Feed.Driver+import Keiro.Ops.Http.Protocol qualified as Protocol+import Keiro.Ops.Http.Views+import Keiro.Wake+import Kiroku.Store.Connection (KirokuStore (..))+import Kiroku.Store.Effect (runStoreIO)++data FeedAcquireError = TooManyViews | InitialReadFailed !Text | FeedClosed | FeedBusy | FeedRetiring | InitialResultTooLarge+  deriving stock (Eq, Show)++data Entry = Entry {token :: !Int, driver :: !Driver, references :: !Int}++data Registry = Registry+  {entries :: !(Map.Map Text Entry), nextToken :: !Int, pump :: !(Maybe (Async ())), closing :: !(Maybe (Async ()))}++data FeedHandle = FeedHandle+  { handleHttpConfig :: !OpsHttpConfig,+    handleFeedConfig :: !FeedConfig,+    store :: !KirokuStore,+    handleBudget :: !InspectionBudget,+    registry :: !(MVar Registry),+    epoch :: !(TVar Word64),+    closed :: !(TVar Bool),+    connections :: !(TVar Int)+  }++data FeedStats = FeedStats {activeViews :: !Int, activeConnections :: !Int, wakePumpRunning :: !Bool}+  deriving stock (Eq, Show)++newFeedHandle :: OpsHttpConfig -> FeedConfig -> KirokuStore -> IO FeedHandle+newFeedHandle http config store = do+  let effective =+        config+          { maxItems = min 500 config.maxItems,+            maxFrameBytes = min config.maxFrameBytes http.inspectionLimits.maxResponseBytes,+            readTimeout = min config.readTimeout (fromIntegral http.inspectionLimits.readDeadlineSeconds)+          }+  validateFeedConfig effective+  budget <- newInspectionBudget http.inspectionLimits+  FeedHandle http effective store budget+    <$> newMVar (Registry Map.empty 0 Nothing Nothing)+    <*> newTVarIO 0+    <*> newTVarIO False+    <*> newTVarIO 0++withFeedHandle :: OpsHttpConfig -> FeedConfig -> KirokuStore -> (FeedHandle -> IO a) -> IO a+withFeedHandle http config store = bracket (newFeedHandle http config store) closeFeedHandle++handleClosed :: FeedHandle -> STM Bool+handleClosed handle = readTVar handle.closed++-- | Reserve upgrading sockets too, before any schema IO. All exits, including+-- failed accept/send and cancellation, return this connection slot.+withFeedConnection :: FeedHandle -> IO a -> IO (Maybe a)+withFeedConnection handle action = mask $ \restore -> do+  admitted <- atomically $ do+    closed <- handleClosed handle+    active <- readTVar handle.connections+    if closed || active >= handle.handleFeedConfig.maxConnections+      then pure False+      else writeTVar handle.connections (active + 1) >> pure True+  if not admitted+    then pure Nothing+    else+      Just <$> restore action `finally` atomically (modifyTVar' handle.connections (subtract 1))++data FeedSchemaCheck = FeedSchemaMatches | FeedSchemaDrift | FeedSchemaReadFailed+  deriving stock (Eq, Show)++-- Call inside the shared admission/deadline scope, both on upgrade and refresh.+checkFeedSchema :: FeedHandle -> IO FeedSchemaCheck+checkFeedSchema handle = do+  result <- Pool.use handle.store.pool verifyExpectedSchemaSession+  pure $ case result of+    Right (Right []) -> FeedSchemaMatches+    Right (Right _) -> FeedSchemaDrift+    _ -> FeedSchemaReadFailed++readView :: FeedHandle -> View -> IO (Either Text ReadResult)+readView handle view = do+  schema <- checkFeedSchema handle+  if schema == FeedSchemaReadFailed || (schema == FeedSchemaDrift && handle.handleHttpConfig.schemaDrift == RefuseOnDrift)+    then pure (Left "schema verification failed")+    else do+      result <- runStoreIO handle.store view.readItems+      pure (either (const (Left "inspection read failed")) Right result)++-- | One initialization task/result per canonical view. Entries retiring at zero+-- references keep their slot until cancellation has joined the driver.+acquireDriver :: FeedHandle -> View -> IO (Either FeedAcquireError Driver)+acquireDriver handle view = mask $ \restore -> do+  reserved <- modifyMVar handle.registry $ \state -> do+    closed <- readTVarIO handle.closed+    if closed+      then pure (state, Left FeedClosed)+      else case Map.lookup view.viewId state.entries of+        Just entry | entry.references == 0 -> pure (state, Left FeedRetiring)+        Just entry -> pure (state {entries = Map.insert view.viewId entry {references = entry.references + 1} state.entries}, Right entry.driver)+        Nothing | Map.size state.entries >= handle.handleFeedConfig.maxViews -> pure (state, Left TooManyViews)+        Nothing -> do+          pump <- case state.pump of+            Just pump -> pure pump+            Nothing -> startPump handle+          seen <- readTVarIO handle.epoch >>= newTVarIO+          driver <-+            startDriver+              view.feedName+              handle.handleFeedConfig+              handle.handleBudget+              handle.handleHttpConfig.inspectionLimits.readDeadlineSeconds+              (awaitEpoch handle seen)+              (readView handle view)+              `onException` (case state.pump of Nothing -> uninterruptibleCancel pump; Just _ -> pure ())+          let entry = Entry state.nextToken driver 1+          pure (state {entries = Map.insert view.viewId entry state.entries, nextToken = state.nextToken + 1, pump = Just pump}, Right driver)+  case reserved of+    Left problem -> pure (Left problem)+    Right driver -> do+      result <- restore (awaitDriver driver) `onException` releaseDriver handle view+      case result of+        Right () -> do+          closed <- atomically ((||) <$> handleClosed handle <*> driverClosed driver)+          if closed then releaseDriver handle view >> pure (Left FeedClosed) else pure (Right driver)+        Left (DriverFailure code message) -> do+          releaseDriver handle view+          pure+            ( Left+                ( case code of+                    Protocol.InspectionBusy -> FeedBusy+                    Protocol.InspectionResultTooLarge -> InitialResultTooLarge+                    _ -> InitialReadFailed message+                )+            )++releaseDriver :: FeedHandle -> View -> IO ()+releaseDriver handle view = uninterruptibleMask_ $ do+  retiring <- modifyMVar handle.registry $ \state -> case Map.lookup view.viewId state.entries of+    Nothing -> pure (state, Nothing)+    Just entry | entry.references <= 0 -> pure (state, Nothing)+    Just entry | entry.references > 1 -> pure (state {entries = Map.insert view.viewId entry {references = entry.references - 1} state.entries}, Nothing)+    Just entry -> pure (state {entries = Map.insert view.viewId entry {references = 0} state.entries}, Just entry)+  case retiring of+    Nothing -> pure ()+    Just entry -> do+      stopDriver entry.driver+      modifyMVar_ handle.registry $ \state -> case Map.lookup view.viewId state.entries of+        Just current | current.token == entry.token -> do+          let remaining = Map.delete view.viewId state.entries+          if Map.null remaining+            then do+              -- Join the old cursor owner before admitting a replacement pump.+              mapM_ uninterruptibleCancel state.pump+              pure state {entries = remaining, pump = Nothing}+            else pure state {entries = remaining}+        _ -> pure state++startPump :: FeedHandle -> IO (Async ())+startPump handle = do+  signal <- wakeSignalFromStore handle.store+  asyncWithUnmask $ \unmask -> unmask $ forever $ do+    reason <- waitForWake signal 1000000+    when (reason == WokenByNotify) (atomically (modifyTVar' handle.epoch (+ 1)))++awaitEpoch :: FeedHandle -> TVar Word64 -> IO WakeReason+awaitEpoch handle seen = do+  timer <- registerDelay (ceiling (handle.handleFeedConfig.fallbackInterval * 1000000))+  atomically $+    (do current <- readTVar handle.epoch; previous <- readTVar seen; check (current /= previous); writeTVar seen current; pure WokenByNotify)+      `orElse` (readTVar timer >>= check >> pure WokenByTimeout)++-- An owned closing job finishes even if a caller waiting on close is cancelled.+-- Later close calls join the same job; it is never restarted or abandoned.+closeFeedHandle :: FeedHandle -> IO ()+closeFeedHandle handle = mask $ \restore -> do+  closing <- uninterruptibleMask_ $ modifyMVar handle.registry $ \state -> case state.closing of+    Just task -> pure (state, task)+    Nothing -> do+      atomically (writeTVar handle.closed True)+      task <- asyncWithUnmask $ \unmask -> unmask $ do+        mapM_ (stopDriver . (.driver)) (Map.elems state.entries)+        mapM_ uninterruptibleCancel state.pump+        atomically (readTVar handle.connections >>= check . (== 0))+      pure (state {entries = Map.empty, pump = Nothing, closing = Just task}, task)+  restore (wait closing)++feedStats :: FeedHandle -> IO FeedStats+feedStats handle = do+  state <- readMVar handle.registry+  active <- readTVarIO handle.connections+  pure (FeedStats (Map.size state.entries) active (maybe False (const True) state.pump))

@@ -0,0 +1,258 @@+-- | One read loop per canonical view. Only the registry creates production+-- drivers; the injected constructor keeps fallback/diff checks small.+module Keiro.Ops.Http.Feed.Driver+  ( FeedConfig (..),+    defaultFeedConfig,+    validateFeedConfig,+    Driver,+    Subscriber,+    DriverFailure (..),+    ReadResult,+    newDriver,+    startDriver,+    awaitDriver,+    stopDriver,+    subscribe,+    unsubscribe,+    nextFrame,+    subscribeOnQueue,+    unsubscribeSTM,+    driverClosed,+    driverSnapshot,+    enqueueSnapshot,+  )+where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (Async, asyncWithUnmask, uninterruptibleCancel)+import Control.Concurrent.STM+import Control.Exception (Exception, SomeAsyncException, SomeException, finally, fromException, mask, onException, throwIO, try)+import Control.Monad (forM_, unless, void)+import Data.Aeson (Value)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Time (NominalDiffTime, UTCTime)+import Data.Word (Word64)+import GHC.Clock (getMonotonicTimeNSec)+import GHC.Generics (Generic)+import Keiro.Ops.Http.Budget qualified as Budget+import Keiro.Ops.Http.Config (InspectionLimits (..), defaultInspectionLimits)+import Keiro.Ops.Http.Feed.Queue qualified as Queue+import Keiro.Ops.Http.Protocol qualified as P+import Keiro.Ops.Http.Views+import Keiro.Wake (WakeReason (..))+import System.Timeout (timeout)++data FeedConfig = FeedConfig+  { fallbackInterval :: !NominalDiffTime,+    minRefreshInterval :: !NominalDiffTime,+    maxViews :: !Int,+    maxFrameBytes :: !Int,+    maxQueuedBytes :: !Int,+    readTimeout :: !NominalDiffTime,+    queueCapacity :: !Int,+    idlePingSeconds :: !Int,+    maxConnections :: !Int,+    maxItems :: !Int+  }+  deriving stock (Eq, Show, Generic)++defaultFeedConfig :: FeedConfig+defaultFeedConfig = FeedConfig 1 1 32 1048576 2097152 5 256 30 256 500++validateFeedConfig :: FeedConfig -> IO ()+validateFeedConfig config = unless valid (ioError (userError "feed limits must be positive, representable, and hold control frames (256 bytes)"))+  where+    periods = [config.fallbackInterval, config.minRefreshInterval, config.readTimeout, fromIntegral config.idlePingSeconds]+    valid =+      all (> 0) [config.maxViews, config.queueCapacity, config.maxConnections, config.maxItems]+        && config.maxFrameBytes >= 256+        && config.maxQueuedBytes >= 256+        && all (\period -> period > 0 && ceiling (period * 1000000) <= toInteger (maxBound :: Int)) periods++type ReadResult = (UTCTime, [Item], Maybe Value)++data DriverFailure = DriverFailure !P.SocketErrorCode !Text+  deriving stock (Eq, Show)++instance Exception DriverFailure++data DriverState = DriverState+  { currentSequence :: !Word64,+    rows :: ![Item],+    extra :: !(Maybe Value),+    snapshot :: !(Maybe Queue.Outgoing),+    subscribers :: !(Map.Map Int Queue.OutboundQueue),+    nextId :: !Int,+    closed :: !Bool,+    lastError :: !(Maybe P.SocketErrorCode)+  }++data Driver = Driver+  { name :: !P.FeedName,+    config :: !FeedConfig,+    state :: !(TVar DriverState),+    ready :: !(TMVar (Either DriverFailure ())),+    worker :: !(Async ())+  }++data Subscriber = Subscriber !Int !Queue.OutboundQueue++-- | Production supplies the shared mount budget. The worker, including its+-- initial read, is owned before any caller waits; cancelled waiters cannot+-- strand an initialization task or lose another waiter's shared result.+startDriver :: P.FeedName -> FeedConfig -> Budget.InspectionBudget -> Int -> IO WakeReason -> IO (Either Text ReadResult) -> IO Driver+startDriver name config budget hostDeadline awaitChange readItems = mask $ \_ -> do+  validateFeedConfig config+  state <- newTVarIO (DriverState 0 [] Nothing Nothing Map.empty 0 False Nothing)+  ready <- newEmptyTMVarIO+  worker <- asyncWithUnmask $ \unmask ->+    unmask (run state ready)+      `finally` atomically+        ( do+            modifyTVar' state (\value -> value {closed = True, rows = [], extra = Nothing, snapshot = Nothing, subscribers = Map.empty})+            void (tryPutTMVar ready (Left (DriverFailure P.ReadFailed "feed driver closed")))+        )+  pure (Driver name config state ready worker)+  where+    maximumBytes = min config.maxFrameBytes config.maxQueuedBytes+    deadline = ceiling (min config.readTimeout (fromIntegral hostDeadline) * 1000000)+    cadence = if name == P.Timers then max 5 config.minRefreshInterval else config.minRefreshInterval+    run state ready = do+      firstStart <- getMonotonicTimeNSec+      initial <- bounded (refresh state P.Subscribed)+      atomically (void (tryPutTMVar ready (fmap (const ()) initial)))+      case initial of Left _ -> pure (); Right _ -> loop state firstStart+    loop state previousStart = do+      active <- atomically $ do+        value <- readTVar state+        check (value.closed || not (Map.null value.subscribers))+        pure (not value.closed)+      if not active+        then pure ()+        else do+          reason <- awaitChange+          now <- getMonotonicTimeNSec+          let elapsed = toInteger (now - previousStart) `div` 1000+              remaining = ceiling (cadence * 1000000) - elapsed+          if remaining > 0 then threadDelay (fromInteger remaining) else pure ()+          value <- readTVarIO state+          if value.closed+            then pure ()+            else+              if Map.null value.subscribers+                then loop state previousStart+                else do+                  started <- getMonotonicTimeNSec+                  result <- bounded (refresh state (case reason of WokenByNotify -> P.Notify; WokenByTimeout -> P.Timeout))+                  case result of Left problem -> publishError state problem; Right _ -> pure ()+                  loop state started+    bounded action = do+      admitted <- Budget.withInspectionAdmission budget (trySync (timeout deadline action))+      pure $ case admitted of+        Left _ -> Left (DriverFailure P.InspectionBusy "inspection read capacity is busy")+        Right (Left problem) -> Left problem+        Right (Right Nothing) -> Left (DriverFailure P.ReadFailed "inspection read deadline exceeded")+        Right (Right (Just value)) -> Right value+    refresh state trigger = do+      result <- readItems+      (stamp, items, extra) <- either (const (throwIO (DriverFailure P.ReadFailed "inspection read failed"))) pure result+      previous <- readTVarIO state+      let old = Map.fromList [(item.key, item) | item <- previous.rows]+          current = Map.fromList [(item.key, item) | item <- items]+          changed = [item.body | item <- items, maybe True ((/= stableItemBody name item) . stableItemBody name) (Map.lookup item.key old)]+          removed = [item.keyJson | item <- previous.rows, Map.notMember item.key current]+          different = not (null changed && null removed) || extra /= previous.extra+          sequence = previous.currentSequence + if trigger /= P.Subscribed && different then 1 else 0+      snapshot <- prepare (P.Snapshot name sequence P.Subscribed stamp (map (.body) items) extra)+      update <-+        if trigger /= P.Subscribed && different+          then Just <$> prepare (P.Update name sequence trigger stamp changed removed extra)+          else pure Nothing+      atomically $ do+        value <- readTVar state+        unless value.closed $ do+          writeTVar state value {currentSequence = sequence, rows = items, extra = extra, snapshot = Just snapshot, lastError = Nothing}+          forM_ update $ \frame -> forM_ (Map.elems value.subscribers) (\queue -> void (Queue.enqueueOutgoing queue frame))+    prepare frame =+      Queue.prepareOutgoing maximumBytes frame+        >>= maybe+          (throwIO (DriverFailure P.InspectionResultTooLarge "inspection frame exceeds the host byte limit"))+          pure+    publishError state (DriverFailure code message) = do+      frame <- Queue.prepareOutgoing maximumBytes (P.ErrorFrame code message (Just name))+      atomically $ do+        value <- readTVar state+        unless (value.closed || value.lastError == Just code) $ do+          writeTVar state value {lastError = Just code}+          forM_ frame $ \outgoing -> forM_ (Map.elems value.subscribers) (\queue -> void (Queue.enqueueOutgoing queue outgoing))++trySync :: IO a -> IO (Either DriverFailure a)+trySync action =+  try action >>= \case+    Right value -> pure (Right value)+    Left (err :: SomeException) -> case fromException err :: Maybe SomeAsyncException of+      Just _ -> throwIO err+      Nothing -> pure (Left (maybe (DriverFailure P.ReadFailed "inspection read failed") id (fromException err)))++newDriver :: P.FeedName -> FeedConfig -> IO WakeReason -> IO (Either Text ReadResult) -> IO Driver+newDriver name config awaitChange readItems = mask $ \restore -> do+  budget <- Budget.newInspectionBudget defaultInspectionLimits {maxConcurrentReads = 1}+  driver <- startDriver name config budget (max 1 (ceiling config.readTimeout)) awaitChange readItems+  outcome <- restore (awaitDriver driver) `onException` stopDriver driver+  case outcome of Left problem -> stopDriver driver >> throwIO problem; Right () -> pure driver++awaitDriver :: Driver -> IO (Either DriverFailure ())+awaitDriver driver = atomically (readTMVar driver.ready)++driverClosed :: Driver -> STM Bool+driverClosed driver = (.closed) <$> readTVar driver.state++stopDriver :: Driver -> IO ()+stopDriver driver = mask $ \_ -> do+  atomically $ modifyTVar' driver.state (\value -> value {closed = True, rows = [], extra = Nothing, snapshot = Nothing, subscribers = Map.empty})+  uninterruptibleCancel driver.worker++subscribe :: Driver -> IO Subscriber+subscribe driver = do+  queue <- Queue.newOutboundQueue driver.config.queueCapacity driver.config.maxQueuedBytes driver.config.maxFrameBytes+  atomically (subscribeOnQueue driver queue)++subscribeOnQueue :: Driver -> Queue.OutboundQueue -> STM Subscriber+subscribeOnQueue driver queue = do+  value <- readTVar driver.state+  if value.closed || Map.size value.subscribers >= driver.config.maxConnections+    then throwSTM (DriverFailure P.ReadFailed "driver is closed or subscriber capacity is full")+    else case value.snapshot of+      Nothing -> throwSTM (DriverFailure P.ReadFailed "driver initialization is incomplete")+      Just snapshot -> do+        let identifier = value.nextId+        writeTVar driver.state value {subscribers = Map.insert identifier queue value.subscribers, nextId = identifier + 1}+        void (Queue.enqueueOutgoing queue snapshot)+        pure (Subscriber identifier queue)++unsubscribeSTM :: Driver -> Subscriber -> STM ()+unsubscribeSTM driver (Subscriber identifier queue) = do+  modifyTVar' driver.state (\value -> value {subscribers = Map.delete identifier value.subscribers})+  Queue.purgeFeed queue driver.name++unsubscribe :: Driver -> Subscriber -> IO ()+unsubscribe driver subscriber = atomically (unsubscribeSTM driver subscriber)++nextFrame :: Subscriber -> STM P.ServerFrame+nextFrame (Subscriber _ queue) = Queue.nextFrame queue++-- | The last successful observation, used for resubscription without a read.+driverSnapshot :: Driver -> STM P.ServerFrame+driverSnapshot driver = do+  value <- readTVar driver.state+  case value.snapshot of+    Just outgoing | not value.closed -> pure outgoing.frame+    _ -> throwSTM (DriverFailure P.ReadFailed "driver has no current snapshot")++enqueueSnapshot :: Driver -> Queue.OutboundQueue -> STM ()+enqueueSnapshot driver queue = do+  value <- readTVar driver.state+  case value.snapshot of+    Just outgoing | not value.closed -> void (Queue.enqueueOutgoing queue outgoing)+    _ -> throwSTM (DriverFailure P.ReadFailed "driver has no current snapshot")

@@ -0,0 +1,158 @@+-- | One bounded outbound queue per socket, including the writer's in-flight+-- frame. Broadcast bytes are shared; overflow is signalled before later deltas.+module Keiro.Ops.Http.Feed.Queue+  ( OutboundQueue,+    Outgoing (..),+    newOutboundQueue,+    prepareOutgoing,+    enqueueOutgoing,+    nextOutgoing,+    finishOutgoing,+    nextFrame,+    purgeFeed,+    closeQueue,+  )+where++import Control.Concurrent.STM+import Data.Aeson (toJSON)+import Data.ByteString qualified as Bytes+import Data.ByteString.Lazy qualified as Lazy+import Data.Map.Strict qualified as Map+import Data.Sequence (Seq, ViewL (..), (|>))+import Data.Sequence qualified as Seq+import Data.Set qualified as Set+import Keiro.Ops.Http.Budget (encodeBounded)+import Keiro.Ops.Http.Protocol++-- | Keep the frame for typed consumers and its single shared wire encoding.+data Outgoing = Outgoing {frame :: !ServerFrame, bytes :: !Bytes.ByteString, owner :: !(Maybe FeedName)}++data QueueState = QueueState+  { pending :: !(Seq Outgoing),+    usedBytes :: !Int,+    usedFrames :: !Int,+    overflow :: !(Set.Set (Maybe FeedName)),+    closed :: !Bool+  }++data OutboundQueue = OutboundQueue !Int !Int !(Map.Map (Maybe FeedName) Outgoing) !(TVar QueueState)++newOutboundQueue :: Int -> Int -> Int -> IO OutboundQueue+newOutboundQueue capacity byteLimit frameLimit+  | capacity <= 0 || byteLimit < 256 || frameLimit < 256 = ioError (userError "feed queues must hold a control frame (256 bytes) and at least one frame")+  | otherwise = do+      errors <- mapM makeError [Nothing, Just Workflows, Just Timers, Just ProjectionGroups]+      OutboundQueue capacity byteLimit (Map.fromList errors) <$> newTVarIO (QueueState Seq.empty 0 0 Set.empty False)+  where+    makeError feed = do+      let value = ErrorFrame QueueOverflow "frames were dropped for this feed; re-subscribe or re-read" feed+      prepared <- prepareOutgoing (min frameLimit byteLimit) value+      maybe (ioError (userError "feed byte limits cannot hold an overflow frame")) (pure . (feed,)) prepared++prepareOutgoing :: Int -> ServerFrame -> IO (Maybe Outgoing)+prepareOutgoing limit value = do+  encoded <- encodeBounded limit (toJSON value)+  pure (fmap (\bytes -> Outgoing value (Lazy.toStrict bytes) (frameOwner value)) encoded)++frameOwner :: ServerFrame -> Maybe FeedName+frameOwner = \case+  Snapshot feed _ _ _ _ _ -> Just feed+  Update feed _ _ _ _ _ _ -> Just feed+  Unsubscribed feed -> Just feed+  ErrorFrame _ _ feed -> feed+  _ -> Nothing++size :: Outgoing -> Int+size value = Bytes.length value.bytes++-- Drop oldest queued frames until both limits fit. An in-flight frame cannot+-- be retracted; if it occupies the budget, discard the incoming frame instead.+roomFor :: Int -> Int -> Outgoing -> QueueState -> QueueState+roomFor capacity byteLimit incoming state+  | state.usedFrames < capacity && toInteger state.usedBytes + toInteger (size incoming) <= toInteger byteLimit = state+  | otherwise = case Seq.viewl state.pending of+      EmptyL -> state+      dropped :< rest ->+        roomFor+          capacity+          byteLimit+          incoming+          state+            { pending = rest,+              usedFrames = state.usedFrames - 1,+              usedBytes = state.usedBytes - size dropped,+              overflow = Set.insert dropped.owner state.overflow+            }++enqueueOutgoing :: OutboundQueue -> Outgoing -> STM Bool+enqueueOutgoing (OutboundQueue capacity byteLimit _ variable) value = do+  initial <- readTVar variable+  if initial.closed+    then pure False+    else do+      let state = roomFor capacity byteLimit value initial+          fits = state.usedFrames < capacity && toInteger state.usedBytes + toInteger (size value) <= toInteger byteLimit+          result =+            if fits+              then state {pending = state.pending |> value, usedFrames = state.usedFrames + 1, usedBytes = state.usedBytes + size value}+              else state {overflow = Set.insert value.owner state.overflow}+      writeTVar variable result+      pure fits++-- | The caller must finish this frame in a masked/bracketed writer before+-- dequeuing again. Its bytes/count stay charged while network delivery blocks.+nextOutgoing :: OutboundQueue -> STM (Maybe Outgoing)+nextOutgoing (OutboundQueue capacity byteLimit errors variable) = do+  state <- readTVar variable+  if state.closed+    then pure Nothing+    else case Set.lookupMin state.overflow of+      Just feed -> do+        let value = errors Map.! feed+            withRoom = roomFor capacity byteLimit value state+        check (withRoom.usedFrames < capacity && toInteger withRoom.usedBytes + toInteger (size value) <= toInteger byteLimit)+        writeTVar+          variable+          withRoom+            { overflow = Set.delete feed withRoom.overflow,+              usedFrames = withRoom.usedFrames + 1,+              usedBytes = withRoom.usedBytes + size value+            }+        pure (Just value)+      Nothing -> case Seq.viewl state.pending of+        EmptyL -> retry+        value :< rest -> do+          writeTVar variable state {pending = rest}+          pure (Just value)++finishOutgoing :: OutboundQueue -> Outgoing -> STM ()+finishOutgoing (OutboundQueue _ _ _ variable) value = modifyTVar' variable $ \state ->+  state {usedFrames = state.usedFrames - 1, usedBytes = state.usedBytes - size value}++-- | Typed in-process consumption releases the queue's reference immediately.+nextFrame :: OutboundQueue -> STM ServerFrame+nextFrame queue =+  nextOutgoing queue >>= \case+    Nothing -> pure Goodbye+    Just value -> finishOutgoing queue value >> pure value.frame++purgeFeed :: OutboundQueue -> FeedName -> STM ()+purgeFeed (OutboundQueue _ _ _ variable) feed = modifyTVar' variable $ \state ->+  let removed = Seq.filter ((== Just feed) . (.owner)) state.pending+   in state+        { pending = Seq.filter ((/= Just feed) . (.owner)) state.pending,+          usedFrames = state.usedFrames - Seq.length removed,+          usedBytes = state.usedBytes - sum (fmap size removed),+          overflow = Set.delete (Just feed) state.overflow+        }++closeQueue :: OutboundQueue -> STM ()+closeQueue (OutboundQueue _ _ _ variable) = modifyTVar' variable $ \state ->+  state+    { pending = Seq.empty,+      usedFrames = state.usedFrames - Seq.length state.pending,+      usedBytes = state.usedBytes - sum (fmap size state.pending),+      overflow = Set.empty,+      closed = True+    }

@@ -0,0 +1,40 @@+module Keiro.Ops.Http.Pool (withOpsInspectionPool) where++import Control.Exception (bracket)+import Data.Text qualified as Text+import Hasql.Connection.Settings qualified as Settings+import Hasql.Pool (Pool)+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session qualified as Session+import Keiro.Ops.Http.Budget (newInspectionBudget)+import Keiro.Ops.Http.Config+import Kiroku.Store.Connection (ConnectionSettings)+import Kiroku.Store.Connection qualified as Store++-- | Create an inspection-owned pool, preserving the host's connection string+-- and search path. Substitute this pool on the store passed to the HTTP/feed+-- surface. Every connection gets the same statement timeout, including after+-- cancellation/reconnection. No writer connection is modified.+withOpsInspectionPool :: InspectionLimits -> ConnectionSettings -> (Pool -> IO a) -> IO a+withOpsInspectionPool limits settings action = do+  _ <- newInspectionBudget limits+  bracket (Pool.acquire config) Pool.release action+  where+    quote name = "\"" <> Text.replace "\"" "\"\"" name <> "\""+    initScript =+      "SET search_path TO "+        <> Text.intercalate ", " (map quote (settings.schema : settings.extraSearchPath))+        <> ", pg_catalog; SET statement_timeout = '"+        <> Text.pack (show limits.statementTimeoutMillis)+        <> "ms'; SET idle_in_transaction_session_timeout = '"+        <> Text.pack (show settings.idleInTransactionTimeout)+        <> "s'"+    config =+      PoolConfig.settings+        ( [ PoolConfig.staticConnectionSettings (Settings.connectionString settings.connString),+            PoolConfig.size limits.maxConcurrentReads,+            PoolConfig.initSession (Session.script initScript)+          ]+            <> maybe [] (\observe -> [PoolConfig.observationHandler observe]) settings.observationHandler+        )

@@ -0,0 +1,106 @@+-- | The inspection socket dialect. Frames describe observations; clients recover+-- missed deltas by re-subscribing or reading the paired HTTP page.+module Keiro.Ops.Http.Protocol+  ( FeedName (..),+    feedNameText,+    parseFeedName,+    ClientFrame (..),+    ServerFrame (..),+    Trigger (..),+    SocketErrorCode (..),+    socketErrorText,+  )+where++import Data.Aeson (FromJSON (..), ToJSON (..), Value (..), object, withObject, withText, (.:), (.=))+import Data.Aeson.KeyMap qualified as KeyMap+import Data.Text (Text)+import Data.Time (UTCTime)+import Data.Word (Word64)+import GHC.Generics (Generic)++-- | Exactly the three supported first-page views.+data FeedName = Workflows | Timers | ProjectionGroups deriving stock (Eq, Ord, Show, Generic)++feedNameText :: FeedName -> Text+feedNameText = \case Workflows -> "workflows"; Timers -> "timers"; ProjectionGroups -> "projection_groups"++parseFeedName :: Text -> Maybe FeedName+parseFeedName = \case "workflows" -> Just Workflows; "timers" -> Just Timers; "projection_groups" -> Just ProjectionGroups; _ -> Nothing++instance ToJSON FeedName where toJSON = String . feedNameText++instance FromJSON FeedName where parseJSON = withText "feed" (maybe (fail "unknown_feed") pure . parseFeedName)++-- Unknown names remain a typed subscription refusal, independently of malformed+-- JSON or an unknown frame type. The filter object is validated by the view.+data ClientFrame = Subscribe !FeedName !Value | UnknownSubscription !Text | Unsubscribe !FeedName | Ping+  deriving stock (Eq, Show, Generic)++instance FromJSON ClientFrame where+  parseJSON = withObject "client frame" $ \fields -> do+    kind <- fields .: "type"+    case (kind :: Text) of+      "subscribe" -> do+        name <- fields .: "feed"+        pure (maybe (UnknownSubscription name) (\feed -> Subscribe feed (Object (KeyMap.delete "type" (KeyMap.delete "feed" fields)))) (parseFeedName name))+      "unsubscribe" -> do+        name <- fields .: "feed"+        pure (maybe (UnknownSubscription name) Unsubscribe (parseFeedName name))+      "ping" -> pure Ping+      _ -> fail "unknown frame type"++instance ToJSON ClientFrame where+  toJSON = \case+    Subscribe feed filters -> object (["type" .= ("subscribe" :: Text), "feed" .= feed] <> filterFields filters)+    UnknownSubscription feed -> object ["type" .= ("subscribe" :: Text), "feed" .= feed]+    Unsubscribe feed -> object ["type" .= ("unsubscribe" :: Text), "feed" .= feed]+    Ping -> object ["type" .= ("ping" :: Text)]+    where+      filterFields (Object fields) = [(key, value) | (key, value) <- KeyMap.toList fields, key /= "type", key /= "feed"]; filterFields _ = []++data Trigger = Subscribed | Notify | Timeout deriving stock (Eq, Show, Generic)++instance ToJSON Trigger where toJSON = String . \case Subscribed -> "subscribe"; Notify -> "notify"; Timeout -> "timeout"++data SocketErrorCode = InvalidFrame | UnknownFeed | InvalidSubscription | QueueOverflow | ReadFailed | TooManyViews | InspectionBusy | ViewRetiring | InspectionResultTooLarge+  deriving stock (Eq, Show, Generic)++socketErrorText :: SocketErrorCode -> Text+socketErrorText = \case+  InvalidFrame -> "invalid_frame"+  UnknownFeed -> "unknown_feed"+  InvalidSubscription -> "invalid_subscription"+  QueueOverflow -> "queue_overflow"+  ReadFailed -> "read_failed"+  TooManyViews -> "too_many_views"+  InspectionBusy -> "inspection_busy"+  ViewRetiring -> "view_retiring"+  InspectionResultTooLarge -> "inspection_result_too_large"++instance ToJSON SocketErrorCode where toJSON = String . socketErrorText++data ServerFrame+  = Snapshot {feed :: !FeedName, sequence :: !Word64, trigger :: !Trigger, readAt :: !UTCTime, items :: ![Value], extra :: !(Maybe Value)}+  | Update {feed :: !FeedName, sequence :: !Word64, trigger :: !Trigger, readAt :: !UTCTime, changed :: ![Value], removed :: ![Value], extra :: !(Maybe Value)}+  | Unsubscribed !FeedName+  | Pong+  | ErrorFrame {code :: !SocketErrorCode, message :: !Text, errorFeed :: !(Maybe FeedName)}+  | Goodbye+  deriving stock (Eq, Show, Generic)++instance ToJSON ServerFrame where+  toJSON = \case+    Snapshot feed sequence trigger stamp items extra ->+      object+        (["type" .= ("snapshot" :: Text), "feed" .= feed, "sequence" .= sequence, "trigger" .= trigger, "observed_at" .= stamp, "items" .= items] <> extraFields extra)+    Update feed sequence trigger stamp changed removed extra ->+      object+        (["type" .= ("update" :: Text), "feed" .= feed, "sequence" .= sequence, "trigger" .= trigger, "observed_at" .= stamp, "changed" .= changed, "removed" .= removed] <> extraFields extra)+    Unsubscribed feed -> object ["type" .= ("unsubscribed" :: Text), "feed" .= feed]+    Pong -> object ["type" .= ("pong" :: Text)]+    ErrorFrame code message feed -> object (["type" .= ("error" :: Text), "code" .= code, "message" .= message] <> maybe [] (\value -> ["feed" .= value]) feed)+    Goodbye -> object ["type" .= ("goodbye" :: Text)]+    where+      extraFields (Just (Object fields)) = [(key, value) | (key, value) <- KeyMap.toList fields, key `notElem` ["type", "feed", "sequence", "trigger", "observed_at", "items", "changed", "removed"]]+      extraFields _ = []

@@ -0,0 +1,94 @@+module Keiro.Ops.Http.Route+  ( RouteError (..),+    renderRouteError,+    flagNames,+    translateRequest,+    ParseOutcome (..),+    parseArguments,+    parseArgumentsWithParser,+  )+where++import Data.Char (isAsciiLower, isDigit)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Set (Set)+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Text qualified as Text+import Keiro.Ops (AppHooks, Command, commandParser)+import Options.Applicative qualified as Options+import Options.Applicative.Common (mapParser)+import Options.Applicative.Types qualified as Types++data RouteError+  = EmptyPathSegment+  | OptionLikePathSegment !Text+  | InvalidQueryKey !Text+  | InvalidFlagValue !Text !Text+  | ReservedQueryKey !Text+  deriving stock (Eq, Show)++renderRouteError :: RouteError -> Text+renderRouteError = Text.pack . show++data Grammar = Grammar !(Set Text) !(Map Text Grammar)++-- | Partially apply at startup: metadata is shared across requests. Sibling+-- commands may use the same long option with different arities.+flagNames :: Options.Parser a -> [Text] -> Set Text+flagNames parser = select Set.empty (grammar parser)+  where+    select inherited (Grammar flags commands) path =+      let current = inherited <> flags+       in case path of+            word : rest | Just child <- Map.lookup word commands -> select current child rest+            _ -> current++grammar :: Options.Parser a -> Grammar+grammar parser = foldr combine (Grammar Set.empty Map.empty) (mapParser optionGrammar parser)+  where+    combine (Grammar a b) (Grammar c d) = Grammar (a <> c) (b <> d)+    optionGrammar :: Types.ArgumentReachability -> Types.Option x -> Grammar+    optionGrammar _ option = case Types.optMain option of+      Types.FlagReader names _ -> Grammar (Set.fromList [Text.pack name | Types.OptLong name <- names]) Map.empty+      Types.CmdReader _ commands ->+        Grammar Set.empty (Map.fromList [(Text.pack name, grammar (Types.infoParser child)) | (name, child) <- commands])+      _ -> Grammar Set.empty Map.empty++translateRequest :: Set Text -> [Text] -> [(Text, Maybe Text)] -> Either RouteError [Text]+translateRequest flags path query = do+  traverse_ checkSegment path+  options <- concat <$> traverse translate query+  pure (path <> options)+  where+    traverse_ f = foldr ((>>) . f) (pure ())+    checkSegment segment+      | Text.null segment = Left EmptyPathSegment+      | "-" `Text.isPrefixOf` segment = Left (OptionLikePathSegment segment)+      | otherwise = Right ()+    translate (key, maybeValue)+      | key `elem` ["json", "force", "database_url", "allow_schema_drift", "help"] = Left (ReservedQueryKey key)+      | Text.null key || not (Text.all (\c -> isAsciiLower c || isDigit c || c == '_') key) = Left (InvalidQueryKey key)+      | Set.member name flags = case value of+          "true" -> Right ["--" <> name]+          "false" -> Right []+          _ -> Left (InvalidFlagValue key value)+      | otherwise = Right ["--" <> name <> "=" <> value]+      where+        name = Text.replace "_" "-" key+        value = maybe "" id maybeValue++data ParseOutcome = Parsed !Command | Rejected !Text++parseArguments :: AppHooks -> [Text] -> ParseOutcome+parseArguments hooks = parseArgumentsWithParser (commandParser hooks)++-- | Share the immutable command grammar across requests at one mount. This+-- does not cache command results or bypass any owning-library validation.+parseArgumentsWithParser :: Options.Parser Command -> [Text] -> ParseOutcome+parseArgumentsWithParser parser arguments =+  case Options.execParserPure Options.defaultPrefs (Options.info parser Options.fullDesc) (map Text.unpack arguments) of+    Options.Success command -> Parsed command+    Options.Failure failure -> Rejected (Text.pack (fst (Options.renderFailure failure "keiro-ops")))+    Options.CompletionInvoked _ -> Rejected "shell completion is not available over HTTP"

@@ -0,0 +1,71 @@+module Keiro.Ops.Http.Server+  ( OpsHttpServer (..),+    startOpsHttpServer,+    stopOpsHttpServer,+    withOpsHttpServer,+  )+where++import Control.Concurrent.Async (Async, async, asyncThreadId, asyncWithUnmask, cancel, link)+import Control.Concurrent.MVar+import Control.Exception (bracket, bracketOnError, finally, mask, onException)+import Data.Map.Strict qualified as Map+import Data.Text qualified as Text+import Keiro.Ops.Http.Config+import Network.Socket qualified as Socket+import Network.Wai (Application)+import Network.Wai.Handler.Warp qualified as Warp++data OpsHttpServer = OpsHttpServer+  { serverThread :: !(Async ()),+    serverPort :: !Int+  }++startOpsHttpServer :: OpsHttpServerConfig -> Application -> IO OpsHttpServer+startOpsHttpServer config app+  | config.port < 0 || config.port > 65535 || config.requestTimeoutSeconds <= 0 = ioError (userError "invalid inspection server port or timeout")+  | otherwise = mask $ \restore -> do+      -- The runner owns the listener and every connection worker. Closing only+      -- the accept loop leaves keep-alive clients and their handlers alive.+      socket <- openSocket config+      actualPort <- fromIntegral <$> Socket.socketPort socket+      ready <- newEmptyMVar+      clients <- newMVar Map.empty+      let spawn :: ((forall a. IO a -> IO a) -> IO ()) -> IO ()+          spawn action = mask $ \_ -> do+            registered <- newEmptyMVar+            client <- asyncWithUnmask $ \unmask ->+              -- Warp must install its accepted-socket cleanup before waiting+              -- for registration. Its first unmask occurs inside that bracket.+              action (\io -> readMVar registered >> unmask io) `finally` do+                tid <- readMVar registered+                modifyMVar_ clients (\connections -> pure $! Map.delete tid connections)+            let tid = asyncThreadId client+            (modifyMVar_ clients (\connections -> pure $! Map.insert tid client connections) >> putMVar registered tid)+              `onException` (tryPutMVar registered tid >> cancel client)+          stopClients = do+            live <- modifyMVar clients (\connections -> pure (Map.empty, Map.elems connections))+            mapM_ cancel live+          settings = Warp.setFork spawn (Warp.setBeforeMainLoop (putMVar ready ()) (Warp.setTimeout config.requestTimeoutSeconds Warp.defaultSettings))+      worker <- async (restore (Warp.runSettingsSocket settings socket app) `finally` (Socket.close socket >> stopClients)) `onException` Socket.close socket+      (link worker >> restore (takeMVar ready)) `onException` (cancel worker >> Socket.close socket)+      pure (OpsHttpServer worker actualPort)++stopOpsHttpServer :: OpsHttpServer -> IO ()+stopOpsHttpServer server = cancel server.serverThread++withOpsHttpServer :: OpsHttpServerConfig -> Application -> (OpsHttpServer -> IO a) -> IO a+withOpsHttpServer config app = bracket (startOpsHttpServer config app) stopOpsHttpServer++openSocket :: OpsHttpServerConfig -> IO Socket.Socket+openSocket config = do+  let hints = Socket.defaultHints {Socket.addrSocketType = Socket.Stream, Socket.addrFlags = [Socket.AI_PASSIVE]}+      host = if config.bindHost == "*" then Nothing else Just (Text.unpack config.bindHost)+  addresses <- Socket.getAddrInfo (Just hints) host (Just (show config.port))+  case addresses of+    [] -> ioError (userError "inspection bind host has no addresses")+    address : _ -> bracketOnError (Socket.socket (Socket.addrFamily address) Socket.Stream Socket.defaultProtocol) Socket.close $ \socket -> do+      Socket.setSocketOption socket Socket.ReuseAddr 1+      Socket.bind socket (Socket.addrAddress address)+      Socket.listen socket Socket.maxListenQueue+      pure socket

@@ -0,0 +1,147 @@+-- | Canonical first-page views over the same reads and encoders as the CLI.+module Keiro.Ops.Http.Views+  ( Item (..),+    View (..),+    workflowsView,+    timersView,+    projectionGroupsView,+    stableItemBody,+  )+where++import Control.Monad (unless)+import Data.Aeson (Value (..), encode, object, withObject, (.:?), (.=))+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KeyMap+import Data.Aeson.Types (Parser, parseEither)+import Data.ByteString.Lazy qualified as Lazy+import Data.List.NonEmpty qualified as NonEmpty+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text+import Data.Time (UTCTime)+import Data.UUID qualified as UUID+import Effectful (Eff, IOE)+import Effectful.Error.Static (Error)+import Effectful.Exception (throwIO)+import Keiro.Ops.Http.Protocol (FeedName (..), feedNameText)+import Keiro.Ops.Projection (projectionGroupStatusJson)+import Keiro.Projection.Catalog (rebuildGroupIdText)+import Keiro.ReadModel (storeHeadPosition)+import Keiro.ReadModel.Rebuild+import Keiro.Timer+import Keiro.Workflow.Inspection qualified as Workflow+import Keiro.Workflow.Instance (WorkflowInstanceRow (..))+import Kiroku.Store.Effect (Store)+import Kiroku.Store.Error (StoreError)+import Kiroku.Store.Types (GlobalPosition (..))++-- | keyJson is the removal identity; body is the shared poll representation.+data Item = Item {key :: !Text, keyJson :: !Value, body :: !Value}+  deriving stock (Eq, Show)++data View = View+  { viewId :: !Text,+    feedName :: !FeedName,+    readItems :: Eff '[Store, Error StoreError, IOE] (UTCTime, [Item], Maybe Value)+  }++canonical :: FeedName -> Value -> Text+canonical feed filters = feedNameText feed <> ":" <> Text.decodeUtf8 (Lazy.toStrict (encode filters))++parseFilters :: [Text] -> (KeyMap.KeyMap Value -> Parser a) -> Value -> Either Text a+parseFilters known parser =+  either (Left . Text.pack) Right+    . parseEither+      ( withObject+          "subscription"+          ( \fields -> do+              case filter (`notElem` known) (map Key.toText (KeyMap.keys fields)) of+                field : _ -> fail ("unknown subscription field: " <> Text.unpack field)+                [] -> pure ()+              parser fields+          )+      )++pageLimit :: Int -> KeyMap.KeyMap Value -> Parser Int+pageLimit bound fields = do+  supplied <- fields .:? "limit"+  let maximumItems = min 500 bound+      count = maybe (min 100 maximumItems) id supplied+  unless (maximumItems > 0 && count > 0 && count <= maximumItems) (fail "limit must be between 1 and the host maximum (at most 500)")+  pure count++workflowsView :: Int -> Value -> Either Text View+workflowsView bound filters = do+  (statuses, name, limit) <- parseFilters ["status_class", "workflow_name", "limit"] parse filters+  let query = Workflow.defaultWorkflowListQuery {Workflow.classes = NonEmpty.nonEmpty statuses, Workflow.workflowName = name, Workflow.pageSize = limit}+      identity = object ["status_class" .= map Workflow.statusClassToText statuses, "workflow_name" .= name, "limit" .= limit]+  pure+    ( View (canonical Workflows identity) Workflows $ do+        page <- Workflow.listWorkflowInstancesPage query >>= require+        pure (page.observedAt, map workflowItem page.items, Nothing)+    )+  where+    parse fields = do+      raw <- fields .:? "status_class"+      statuses <- case raw of+        Nothing -> pure []+        Just names -> do+          unless (not (null (names :: [Text]))) (fail "status_class must not be empty")+          parsed <- traverse (maybe (fail "unknown status_class") pure . Workflow.statusClassFromText) names+          let selected = Set.fromList parsed+          pure (if selected == Set.fromList [minBound .. maxBound] then [] else Set.toAscList selected)+      (statuses,,) <$> fields .:? "workflow_name" <*> pageLimit bound fields++workflowItem :: WorkflowInstanceRow -> Item+workflowItem row =+  Item+    (row.workflowName <> "\NUL" <> row.workflowId)+    (object ["workflow_name" .= row.workflowName, "workflow_id" .= row.workflowId])+    (Workflow.instanceToJson row)++timersView :: Int -> Value -> Either Text View+timersView bound filters = do+  (manager, correlation, limit) <-+    parseFilters+      ["manager", "correlation_id", "limit"]+      (\fields -> (,,) <$> fields .:? "manager" <*> fields .:? "correlation_id" <*> pageLimit bound fields)+      filters+  let identity = object ["manager" .= manager, "correlation_id" .= correlation, "limit" .= limit]+  pure+    ( View (canonical Timers identity) Timers $ do+        page <- findPendingTimers (PendingTimerFilter manager correlation) (PendingTimerPageRequest limit Nothing) >>= require+        pure+          ( page.observedAt,+            [ let TimerId uuid = inspection.timer.timerId+                  timerId = UUID.toText uuid+               in Item timerId (object ["timer_id" .= timerId]) (pendingTimerItemJson page.observedAt inspection)+            | inspection <- page.timers+            ],+            Nothing+          )+    )++projectionGroupsView :: Int -> Value -> Either Text View+projectionGroupsView bound filters = do+  limit <- parseFilters ["limit"] (pageLimit bound) filters+  pure+    ( View (canonical ProjectionGroups (object ["limit" .= limit])) ProjectionGroups $ do+        page <- listProjectionGroupStatusesPage (ProjectionStatusPageRequest limit Nothing) >>= require+        GlobalPosition visibleHead <- storeHeadPosition+        pure+          ( page.observedAt,+            [let groupId = rebuildGroupIdText row.groupId in Item groupId (object ["group_id" .= groupId]) (projectionGroupStatusJson row) | row <- page.items],+            Just (object ["visible_store_head" .= visibleHead])+          )+    )++require :: (Show err) => Either err value -> Eff '[Store, Error StoreError, IOE] value+require = either (throwIO . userError . show) pure++-- Database time changes the presentation countdown, not the timer's state.+-- Drivers retain each fresh body/read time even when this comparison is equal.+stableItemBody :: FeedName -> Item -> Value+stableItemBody Timers item = case item.body of Object fields -> Object (KeyMap.delete "due_in_seconds" fields); value -> value+stableItemBody _ item = item.body

@@ -0,0 +1,184 @@+-- | One socket writer and a multiplexed, bounded connection queue. Notifications+-- remain hints; all frames are produced by supported inspection reads.+module Keiro.Ops.Http.WebSocket (feedServerApp, feedsMiddleware, opsApplicationWithFeeds) where++import Control.Concurrent.Async (race_)+import Control.Concurrent.STM+import Control.Exception (bracket, catch, finally, mask, onException, throwIO)+import Control.Monad (forM_, void, when)+import Data.Aeson (object, (.=))+import Data.Aeson qualified as Aeson+import Data.ByteString qualified as Bytes+import Data.ByteString.Char8 qualified as Char8+import Data.ByteString.Lazy qualified as Lazy+import Data.IORef+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text.Encoding qualified as Text+import Keiro.Ops (AppHooks)+import Keiro.Ops.Http.Application (opsApplicationWithBudget)+import Keiro.Ops.Http.Budget qualified as Budget+import Keiro.Ops.Http.Config+import Keiro.Ops.Http.Cors qualified as Cors+import Keiro.Ops.Http.Feed qualified as Feed+import Keiro.Ops.Http.Feed.Driver qualified as Driver+import Keiro.Ops.Http.Feed.Queue qualified as Queue+import Keiro.Ops.Http.Protocol qualified as P+import Keiro.Ops.Http.Views+import Kiroku.Store.Connection (KirokuStore)+import Network.Wai (Application, Middleware, pathInfo)+import Network.Wai.Handler.WebSockets qualified as WaiWS+import Network.WebSockets qualified as WS+import System.Timeout (timeout)++opsApplicationWithFeeds :: OpsHttpConfig -> Feed.FeedHandle -> AppHooks -> KirokuStore -> IO Application+opsApplicationWithFeeds config handle hooks store+  | config /= Feed.handleHttpConfig handle = ioError (userError "HTTP and feeds must share the same inspection policy and budget")+  | otherwise = feedsMiddleware config handle <$> opsApplicationWithBudget config (Feed.handleBudget handle) hooks store++feedsMiddleware :: OpsHttpConfig -> Feed.FeedHandle -> Middleware+feedsMiddleware config handle application request respond+  | pathInfo request == ["ws"] = WaiWS.websocketsOr options (feedServerApp config handle) application request respond+  | otherwise = application request respond+  where+    maximum = fromIntegral config.inspectionLimits.maxRequestBytes+    options = WS.defaultConnectionOptions {WS.connectionFramePayloadSizeLimit = WS.SizeLimit maximum, WS.connectionMessageDataSizeLimit = WS.SizeLimit maximum}++feedServerApp :: OpsHttpConfig -> Feed.FeedHandle -> WS.ServerApp+feedServerApp config handle pending+  | config /= Feed.handleHttpConfig handle = reject 500 "invalid_feed_policy" "HTTP and feeds have different host policies"+  | Char8.takeWhile (/= '?') request.requestPath /= "/ws" = reject 404 "not_found" "no inspection socket at this path"+  | Bytes.length request.requestPath > config.inspectionLimits.maxRequestBytes = reject 413 "inspection_request_too_large" "socket path exceeds the host limit"+  | not (originAllowed config request) = reject 403 "origin_not_allowed" "socket origin is not allowed"+  | otherwise = do+      admitted <- Feed.withFeedConnection handle $ do+        checked <- Budget.withInspectionBudget (Feed.handleBudget handle) config.inspectionLimits.readDeadlineSeconds (Feed.checkFeedSchema handle)+        case checked of+          Left Budget.InspectionBusy -> reject 503 "inspection_busy" "inspection read capacity is busy"+          Left Budget.InspectionTimedOut -> reject 504 "inspection_timeout" "schema verification exceeded the read deadline"+          Right Feed.FeedSchemaReadFailed -> reject 500 "schema_check_failed" "schema verification failed"+          Right Feed.FeedSchemaDrift | config.schemaDrift == RefuseOnDrift -> reject 409 "schema_drift" "database schema does not match this binary"+          Right _ -> do+            closed <- atomically (Feed.handleClosed handle)+            if closed+              then reject 503 "feed_closed" "inspection feeds are closed"+              else WS.acceptRequest pending >>= runConnection handle+      case admitted of+        Just () -> pure ()+        Nothing -> reject 503 "too_many_connections" "socket connection capacity is full"+  where+    request = WS.pendingRequest pending+    reject code reason message =+      WS.rejectRequestWith+        pending+        WS.defaultRejectRequest+          { WS.rejectCode = code,+            WS.rejectMessage = "Inspection request refused",+            WS.rejectHeaders = [("Content-Type", "application/json")],+            WS.rejectBody = Lazy.toStrict (Aeson.encode (object ["error" .= object ["code" .= (reason :: Text), "message" .= (message :: Text)]]))+          }++originAllowed :: OpsHttpConfig -> WS.RequestHead -> Bool+originAllowed config request =+  all (\(key, value) -> key /= "Origin" || Bytes.length value <= config.inspectionLimits.maxRequestBytes) request.requestHeaders+    && Cors.originAllowed config.allowedOrigins request.requestSecure request.requestHeaders++type Subscription = (View, Feed.Driver, Feed.Subscriber)++runConnection :: Feed.FeedHandle -> WS.Connection -> IO ()+runConnection handle connection = do+  let config = Feed.handleFeedConfig handle+  queue <- Queue.newOutboundQueue config.queueCapacity config.maxQueuedBytes config.maxFrameBytes+  subscriptions <- newIORef Map.empty+  let cleanup = do+        atomically (Queue.closeQueue queue)+        current <- atomicModifyIORef' subscriptions (\values -> (Map.empty, values))+        forM_ (Map.elems current) $ \(view, driver, subscriber) -> do+          Feed.unsubscribe driver subscriber+          Feed.releaseDriver handle view+      sendControl frame = do+        prepared <- Queue.prepareOutgoing (min config.maxFrameBytes config.maxQueuedBytes) frame+        case prepared of+          Nothing -> throwIO (userError "socket control frame exceeds host limits")+          Just outgoing -> void (atomically (Queue.enqueueOutgoing queue outgoing))+      failure code message feed = sendControl (P.ErrorFrame code message feed)+      receive = do+        bytes <- WS.receiveData connection :: IO Lazy.ByteString+        case Aeson.eitherDecode bytes of+          Left _ -> failure P.InvalidFrame "invalid socket frame" Nothing+          Right P.Ping -> sendControl P.Pong+          Right (P.UnknownSubscription _) -> failure P.UnknownFeed "unknown feed" Nothing+          Right (P.Unsubscribe name) -> mask $ \_ -> do+            current <- readIORef subscriptions+            forM_ (Map.lookup name current) $ \(view, driver, subscriber) -> do+              atomically (Driver.unsubscribeSTM driver subscriber)+              writeIORef subscriptions (Map.delete name current)+              Feed.releaseDriver handle view+            sendControl (P.Unsubscribed name)+          Right (P.Subscribe name filters) -> case makeView name filters of+            Left message -> failure P.InvalidSubscription message (Just name)+            Right view -> replaceSubscription handle queue subscriptions name view failure+        receive+      send = do+        again <-+          bracket+            ( timeout+                (config.idlePingSeconds * 1000000)+                ( atomically $ do+                    closed <- Feed.handleClosed handle+                    if closed then Queue.closeQueue queue >> pure Nothing else Queue.nextOutgoing queue+                )+            )+            (\case Just (Just outgoing) -> atomically (Queue.finishOutgoing queue outgoing); _ -> pure ())+            ( \case+                Nothing -> boundedSend (WS.sendPing connection ("" :: Bytes.ByteString)) >> pure True+                Just Nothing -> do+                  boundedSend (WS.sendTextData connection (Aeson.encode P.Goodbye))+                  boundedSend (WS.sendClose connection ("inspection feeds closed" :: Text))+                  pure False+                Just (Just outgoing) -> boundedSend (WS.sendTextData connection outgoing.bytes) >> pure True+            )+        when again send+      boundedSend action =+        timeout (ceiling (config.readTimeout * 1000000)) action+          >>= maybe+            (throwIO (userError "socket send deadline exceeded"))+            pure+      makeView = \case P.Workflows -> workflowsView config.maxItems; P.Timers -> timersView config.maxItems; P.ProjectionGroups -> projectionGroupsView config.maxItems+  (race_ receive send `catch` (\(_ :: WS.ConnectionException) -> pure ())) `finally` cleanup++-- Acquiring a replacement does not disturb the old subscription on refusal.+-- The queue purge, detach and attach commit atomically so old-view deltas cannot+-- arrive after the new snapshot. Balanced registry releases cannot be cancelled.+replaceSubscription :: Feed.FeedHandle -> Queue.OutboundQueue -> IORef (Map.Map P.FeedName Subscription) -> P.FeedName -> View -> (P.SocketErrorCode -> Text -> Maybe P.FeedName -> IO ()) -> IO ()+replaceSubscription handle queue subscriptions name view failure = mask $ \restore -> do+  current <- readIORef subscriptions+  case Map.lookup name current of+    Just (old, driver, _) | old.viewId == view.viewId -> atomically $ do+      Queue.purgeFeed queue name+      Driver.enqueueSnapshot driver queue+    previous -> do+      acquired <- Feed.acquireDriver handle view+      case acquired of+        Left problem ->+          restore+            ( failure+                ( case problem of+                    Feed.TooManyViews -> P.TooManyViews+                    Feed.FeedBusy -> P.InspectionBusy+                    Feed.FeedRetiring -> P.ViewRetiring+                    Feed.InitialResultTooLarge -> P.InspectionResultTooLarge+                    _ -> P.ReadFailed+                )+                "subscription could not be initialized"+                (Just name)+            )+        Right driver -> do+          subscriber <-+            ( atomically $ do+                forM_ previous (\(_, old, subscriber) -> Driver.unsubscribeSTM old subscriber)+                Driver.subscribeOnQueue driver queue+            )+              `onException` Feed.releaseDriver handle view+          writeIORef subscriptions (Map.insert name (view, driver, subscriber) current)+          forM_ previous (\(old, _, _) -> Feed.releaseDriver handle old)

@@ -0,0 +1,802 @@+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async+import Control.Concurrent.Chan (newChan, readChan, writeChan)+import Control.Concurrent.MVar+import Control.Exception (IOException, SomeException, bracket, finally, try)+import Control.Monad (forM_, replicateM)+import Data.Aeson (Value (..), object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KeyMap+import Data.ByteString.Char8 qualified as Char8+import Data.ByteString.Lazy qualified as LazyByteString+import Data.IORef+import Data.Map.Strict qualified as Map+import Data.Maybe (isJust)+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text+import Data.Time (addUTCTime, getCurrentTime)+import Data.UUID qualified as UUID+import Hasql.Connection qualified as Connection+import Hasql.Connection.Settings qualified as Settings+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Pool.Observation qualified as Observation+import Hasql.Session qualified as Session+import Hasql.Statement qualified as Statement+import Jitsurei.Inspection (jitsureiProcessManagerInspectors)+import Keiro.Ops qualified as Ops+import Keiro.Ops.Env+import Keiro.Ops.Http+import Keiro.Ops.Http.Application+import Keiro.Ops.Http.Budget+import Keiro.Ops.Http.Config qualified+import Keiro.Ops.Http.Config qualified as Config+import Keiro.Ops.Http.Cors qualified as Cors+import Keiro.Ops.Http.Protocol qualified as Protocol+import Keiro.Ops.Http.Route+import Keiro.Ops.Http.Views qualified as Views+import Keiro.Ops.Render+import Keiro.Ops.ReplayAudit qualified as Audit+import Keiro.Ops.Workflow qualified as Workflow+import Keiro.Ops.WorkflowHeaderCache+import Keiro.PGMQ+import Keiro.Projection.Catalog qualified as Catalog+import Keiro.Projection.Catalog.Operations qualified as CatalogOperations+import Keiro.Test.Postgres+import Keiro.Timer qualified as Timer+import Keiro.Workflow (WorkflowId (..), WorkflowJournalEvent (..), WorkflowName (..), appendJournalEntry, continueAsNew, runWorkflow)+import Keiro.Workflow.Resume (defaultWorkflowResumeOptions)+import Kiroku.Store.Append (appendToStream)+import Kiroku.Store.Connection+import Kiroku.Store.Effect (runStoreIO)+import Kiroku.Store.Read (getStream)+import Kiroku.Store.Transaction (runTransaction)+import Kiroku.Store.Types hiding (id)+import Network.HTTP.Client qualified as Client+import Network.HTTP.Types+import Network.Socket qualified as Socket+import Network.Socket.ByteString qualified as Socket+import Network.Wai+import Network.Wai.Test hiding (request)+import Options.Applicative qualified as Options+import Options.Applicative.Common (mapParser)+import Options.Applicative.Types qualified as Types+import Pgmq.Migration qualified as Pgmq+import System.Exit (ExitCode (..))+import System.Process (readProcessWithExitCode)+import System.Timeout (timeout)+import Test.ComposeSpec qualified as ComposeSpec+import Test.FeedSpec qualified as FeedSpec+import Test.Hspec+import Test.WebSocketSpec qualified as WebSocketSpec++main :: IO ()+main = do+  pgmq <- either (fail . show) pure Pgmq.pgmqMigrations+  withMigratedSuiteWith [pgmq] $ \fixture -> hspec (spec fixture)++spec :: Fixture -> Spec+spec fixture = do+  ComposeSpec.spec fixture+  FeedSpec.spec fixture+  WebSocketSpec.spec fixture+  describe "live view parameters" $ do+    it "canonicalizes equivalent status sets and bounds subscription inputs" $ do+      Right one <- pure (Views.workflowsView 500 (object ["status_class" .= (["running", "sleeping", "running"] :: [Text])]))+      Right two <- pure (Views.workflowsView 500 (object ["status_class" .= (["sleeping", "running"] :: [Text])]))+      one.viewId `shouldBe` two.viewId+      either (const True) (const False) (Views.timersView 20 (object ["limit" .= (21 :: Int)])) `shouldBe` True+      case Views.timersView 20 (object ["unexpected_filter" .= True]) of+        Left message -> message `shouldSatisfy` Text.isInfixOf "unexpected_filter"+        Right _ -> expectationFailure "unknown filter was accepted"+    it "compares timer state independently of presentation countdowns" $ do+      let one = Views.Item "timer" (object ["timer_id" .= ("timer" :: Text)]) (object ["status" .= ("scheduled" :: Text), "due_in_seconds" .= (10 :: Int)])+          two = one {Views.body = object ["status" .= ("scheduled" :: Text), "due_in_seconds" .= (9 :: Int)]}+      Views.stableItemBody Protocol.Timers one `shouldBe` Views.stableItemBody Protocol.Timers two+      one.body `shouldNotBe` two.body++  describe "request translation" $ do+    it "preserves path and repeated option order, converting snake_case" $ do+      translateRequest Set.empty ["wf", "show", "approval", "wf-1"] [] `shouldBe` Right ["wf", "show", "approval", "wf-1"]+      translateRequest Set.empty ["wf", "list"] [("status", Just "running"), ("status", Just "failed"), ("min_age", Just "5m")]+        `shouldBe` Right ["wf", "list", "--status=running", "--status=failed", "--min-age=5m"]+    it "translates flags using only the selected command grammar" $ do+      let parser =+            Options.hsubparser+              ( Options.command "flag" (Options.info (Options.switch (Options.long "shared")) mempty)+                  <> Options.command "option" (Options.info ((== ("yes" :: String)) <$> Options.strOption (Options.long "shared")) mempty)+              )+          flags = flagNames parser+      flags ["flag"] `shouldBe` Set.singleton "shared"+      flags ["option"] `shouldBe` Set.empty+      translateRequest (flags ["flag"]) ["flag"] [("shared", Just "true")] `shouldBe` Right ["flag", "--shared"]+      translateRequest (flags ["flag"]) ["flag"] [("shared", Just "false")] `shouldBe` Right ["flag"]+      translateRequest (flags ["option"]) ["option"] [("shared", Just "yes")] `shouldBe` Right ["option", "--shared=yes"]+      translateRequest (flags ["flag"]) ["flag"] [("shared", Just "yes")] `shouldBe` Left (InvalidFlagValue "shared" "yes")+    it "rejects global options, malformed keys, empty and option-like segments" $ do+      forM_ ["json", "force", "database_url", "allow_schema_drift", "help"] $ \key ->+        translateRequest Set.empty ["wf"] [(key, Nothing)] `shouldBe` Left (ReservedQueryKey key)+      forM_ ["Bad", "with-hyphen", "", "x y"] $ \key ->+        translateRequest Set.empty ["wf"] [(key, Nothing)] `shouldBe` Left (InvalidQueryKey key)+      translateRequest Set.empty [""] [] `shouldBe` Left EmptyPathSegment+      translateRequest Set.empty ["--force"] [] `shouldBe` Left (OptionLikePathSegment "--force")+    it "uses the real parser including the legacy two-value after option" $ do+      let parsed path query = either (const False) parses (translateRequest (flagNames (Ops.commandParser Ops.emptyAppHooks) path) path query)+          parses args = case parseArguments Ops.emptyAppHooks args of Parsed _ -> True; Rejected _ -> False+      parsed ["wf", "list", "instance-1"] [("after", Just "approval")] `shouldBe` True+      parsed ["wf", "list"] [("limit", Just "abc")] `shouldBe` False+      case parseArguments Ops.emptyAppHooks ["wf", "list"] of+        Parsed command -> Ops.isMutation command `shouldBe` False+        _ -> expectationFailure "read did not parse"+      case parseArguments Ops.emptyAppHooks ["outbox", "maintenance-pass"] of+        Parsed command -> Ops.isMutation command `shouldBe` True+        _ -> expectationFailure "mutation did not parse"+  describe "budget" $ do+    it "rejects excess work immediately and releases a cancelled permit after cleanup" $ do+      budget <- newInspectionBudget defaultInspectionLimits+      entered <- newChanCounter+      gate <- newEmptyMVar+      let blocked = withInspectionBudget budget 10 (signalCounter entered >> takeMVar gate)+      bracket (replicateM 4 (async blocked)) (mapM_ cancel) $ \jobs -> do+        awaitCounter entered 4+        withInspectionBudget budget 1 (pure ()) `shouldReturn` Left InspectionBusy+        case jobs of first : _ -> cancel first; [] -> expectationFailure "missing jobs"+        withInspectionBudget budget 1 (pure ()) `shouldReturn` Right ()+    it "joins timed-out work and keeps the permit reusable" $ do+      budget <- newInspectionBudget defaultInspectionLimits+      cleaned <- newEmptyMVar+      withInspectionBudget budget 1 (threadDelay 3000000 `finally` putMVar cleaned ()) `shouldReturn` Left InspectionTimedOut+      tryTakeMVar cleaned `shouldReturn` Just ()+      withInspectionBudget budget 1 (pure ()) `shouldReturn` Right ()+    it "bounds encoding and rejects invalid host limits" $ do+      encodeBounded 10 (String (Text.replicate 10000 "x")) `shouldReturn` Nothing+      encodeBounded 4 (String "hi") `shouldReturn` Just "\"hi\""+      encodeBoundedEncoding 4 (Aeson.toEncoding ("hi" :: Text)) `shouldReturn` Just "\"hi\""+      encodeBoundedEncoding 10 (Aeson.toEncoding (replicate 100000 ("x" :: Text))) `shouldReturn` Nothing+      newInspectionBudget defaultInspectionLimits {maxConcurrentReads = 0} `shouldThrow` anyIOException+      forM_ [0, -1, maxBound] $ \seconds ->+        newInspectionBudget defaultInspectionLimits {deliveryDeadlineSeconds = seconds} `shouldThrow` anyIOException+  around (withFreshStore fixture) $ do+    describe "mounted process-manager discovery" $ do+      it "serves Jitsurei metadata without journal instances and agrees with the parsed CLI" $ \store -> do+        let mounted = Ops.emptyAppHooks {Ops.processManagers = Just jitsureiProcessManagerInspectors}+            expected = object ["items" .= [object ["name" .= ("jitsurei-fulfillment" :: Text), "category" .= ("fulfillment" :: Text), "snapshots_configured" .= False, "snapshot_policy" .= ("never" :: Text), "timers_configured" .= Null]]]+        app <- opsApplication defaultOpsHttpConfig mounted store+        response <- call app methodGet "/pm/types" ""+        simpleStatus response `shouldBe` status200+        body response `shouldBe` expected+        case parseArguments mounted ["pm", "types"] of+          Parsed command -> do+            Succeeded result <- Ops.runCommand mounted (OpsEnv store Json False [] False) command+            result.jsonValue `shouldBe` body response+          _ -> expectationFailure "CLI pm types did not parse"+        refused <- call app methodPost "/pm/types" ""+        assertError refused status405 "method_not_allowed"+        named <- call app methodGet "/pm/types?name=jitsurei-fulfillment" ""+        assertError named status400 "invalid_request"+      it "distinguishes absent and empty hooks and bounds large catalogs" $ \store -> do+        absent <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+        forM_ ["/pm/types", "/pm/list?name=missing"] $ \path -> do+          response <- call absent methodGet path ""+          assertError response status400 "invalid_request"+        empty <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks {Ops.processManagers = Just Map.empty} store+        response <- call empty methodGet "/pm/types" ""+        simpleStatus response `shouldBe` status200+        body response `shouldBe` object ["items" .= ([] :: [Value])]+        let descriptor = snd (Map.findMin jitsureiProcessManagerInspectors)+            catalog = Map.fromList [("alias-" <> Text.pack (show n), descriptor) | n <- [1 .. 1000 :: Int]]+            mounted = Ops.emptyAppHooks {Ops.processManagers = Just catalog}+            config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxResponseBytes = 1024}}+        bounded <- opsApplication config mounted store+        oversized <- call bounded methodGet "/pm/types" ""+        assertError oversized status413 "inspection_result_too_large"+      it "keeps fresh schema refusal for metadata reads" $ \store -> do+        app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks {Ops.processManagers = Just jitsureiProcessManagerInspectors} store+        Right () <- Pool.use store.pool (Session.script "ALTER TABLE keiro.keiro_timers ADD COLUMN pm_types_drift text")+        response <- call app methodGet "/pm/types" ""+        assertError response status409 "schema_drift"+    describe "workflow page routes" $ do+      it "rejects an unrepresentable outgoing cursor instead of returning an unusable token" $ \store -> do+        result <- Pool.use store.pool (Session.script "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('a',repeat('界',2000),'running'),('b',repeat('界',2000),'running')")+        either (fail . show) pure result+        app <- opsApplication defaultOpsHttpConfig hooks store+        response <- call app methodGet "/wf/inspect/list?limit=1" ""+        assertError response status413 "inspection_result_too_large"+      it "returns wide workflow keys within the continuation and response caps" $ \store -> do+        result <- Pool.use store.pool (Session.script "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) SELECT 'wide-'||n,repeat('x',2700),'running' FROM generate_series(1,65) n")+        either (fail . show) pure result+        app <- opsApplication defaultOpsHttpConfig hooks store+        response <- call app methodGet "/wf/inspect/list?limit=1" ""+        simpleStatus response `shouldBe` status200+        field "next_cursor" (body response) `shouldSatisfy` isJust+      it "pages supported reads through the same parser and maps typed failures" $ \store -> do+        seedInspection store+        now <- getCurrentTime+        forM_ ["wf-2", "wf-3"] $ \wid ->+          runStoreIO store (appendJournalEntry (WorkflowName "approval") (WorkflowId wid) (StepRecorded "received" Null now)) >>= either (fail . show) pure+        app <- opsApplication defaultOpsHttpConfig hooks store+        let pages cursor = do+              response <- call app methodGet ("/wf/inspect/list?limit=1" <> maybe "" ("&after=" <>) cursor) ""+              simpleStatus response `shouldBe` status200+              field "observed_at" (body response) `shouldSatisfy` maybe False (\case String _ -> True; _ -> False)+              let items = case field "items" (body response) of Just (Array rows) -> foldr (:) [] rows; _ -> []+              length items `shouldBe` 1+              rest <- case field "next_cursor" (body response) of Just (String token) -> pages (Just token); _ -> pure []+              pure (items <> rest)+        rows <- pages Nothing+        map (field "workflow_id") rows `shouldBe` map (Just . String) ["wf-1", "wf-2", "wf-3"]+        header <- call app methodGet "/wf/inspect/header/approval/wf-1" ""+        legacy <- call app methodGet "/wf/show/approval/wf-1" ""+        field "instance" (body legacy) `shouldBe` Just (body header)+        forM_ ["header", "journal", "steps", "children", "awakeables"] $ \command -> do+          response <- call app methodGet ("/wf/inspect/" <> command <> "/approval/wf-1") ""+          simpleStatus response `shouldBe` status200+        malformed <- call app methodGet "/wf/inspect/list?after=not-a-cursor" ""+        assertError malformed status400 "invalid_cursor"+        oversized <- call app methodGet "/wf/inspect/list?limit=501" ""+        assertError oversized status400 "invalid_page_size"+        runStoreIO store (appendJournalEntry (WorkflowName "approval") (WorkflowId "wf-1") (StepRecorded "second" Null now)) >>= either (fail . show) pure+        first <- call app methodGet "/wf/inspect/steps/approval/wf-1?limit=1" ""+        token <- case field "next_cursor" (body first) of Just (String value) -> pure value; _ -> fail "missing step cursor"+        _ <- runStoreIO store (runWorkflow (WorkflowName "approval") (WorkflowId "wf-1") (continueAsNew () >> pure ())) >>= either (fail . show) pure+        rotated <- call app methodGet ("/wf/inspect/steps/approval/wf-1?after=" <> token) ""+        assertError rotated status409 "cursor_generation_mismatch"+        freshHeader <- call app methodGet "/wf/inspect/header/approval/wf-1" ""+        field "generation" (body freshHeader) `shouldBe` Just (Number 1)+    describe "fresh header presentation reuse" $ do+      it "preserves canonical bytes and observes edits, deletion and recreation" $ \store -> do+        seedInspection store+        app <- opsApplication defaultOpsHttpConfig hooks store+        let readHeader = call app methodGet "/wf/inspect/header/approval/wf-1" ""+            sql statement = Pool.use store.pool (Session.script statement) >>= either (fail . show) pure+        first <- readHeader+        second <- readHeader+        shouldBe (simpleBody second) (simpleBody first)+        sql "UPDATE keiro.keiro_workflows SET attempts=7,last_error='changed',leased_by='new-owner' WHERE workflow_name='approval' AND workflow_id='wf-1'"+        changed <- readHeader+        shouldBe (field "attempts" (body changed)) (Just (Number 7))+        shouldBe (field "last_error" (body changed)) (Just (String "changed"))+        shouldBe (field "leased_by" (body changed)) (Just (String "new-owner"))+        legacy <- call app methodGet "/wf/show/approval/wf-1" ""+        shouldBe (field "instance" (body legacy)) (Just (body changed))+        sql "DELETE FROM keiro.keiro_workflows WHERE workflow_name='approval' AND workflow_id='wf-1'"+        missing <- readHeader+        assertError missing status422 "command_failed"+        sql "INSERT INTO keiro.keiro_workflows(workflow_name,workflow_id,status) VALUES ('approval','wf-1','failed')"+        recreated <- readHeader+        shouldBe (field "status" (body recreated)) (Just (String "failed"))+      it "enforces byte and entry limits through churn, oversized changes and misses" $ \store -> do+        let sql statement = Pool.use store.pool (Session.script statement) >>= either (fail . show) pure+            encode value = fmap LazyByteString.toStrict <$> encodeBounded (1024 * 1024) value+        sql "INSERT INTO keiro.keiro_workflows(workflow_name,workflow_id,status) SELECT 'cache','id-'||n,'running' FROM generate_series(1,96) n"+        cache <- newWorkflowHeaderCache (1024 * 1024) encode+        let readHeader wid = Ops.runCommandWithHeaderCache cache hooks (OpsEnv store Json False [] False) (Ops.Workflow (Workflow.Inspect (Workflow.InspectHeader (Workflow.WorkflowRef "cache" wid))))+            bounded = do+              (count, bytes) <- workflowHeaderCacheUsage cache+              shouldSatisfy count (<= 32)+              shouldSatisfy bytes (<= 1024 * 1024)+        forM_ [1 .. (96 :: Int)] $ \n -> do+          (outcome, encoded) <- readHeader ("id-" <> Text.pack (show n))+          shouldSatisfy outcome (\case Succeeded _ -> True; _ -> False)+          shouldSatisfy encoded isJust+          bounded+        shouldReturn (fmap fst (workflowHeaderCacheUsage cache)) 32+        sql "UPDATE keiro.keiro_workflows SET last_error=repeat('x',20000)"+        forM_ [1 .. (96 :: Int)] $ \n -> readHeader ("id-" <> Text.pack (show n)) >> bounded+        (largeCount, _) <- workflowHeaderCacheUsage cache+        shouldSatisfy largeCount (< 32)+        sql "DELETE FROM keiro.keiro_workflows"+        forM_ [1 .. (96 :: Int)] $ \n -> readHeader ("id-" <> Text.pack (show n)) >> bounded+        shouldReturn (workflowHeaderCacheUsage cache) (0, 0)+      it "fails oversized changes explicitly and reapplies schema policy to warm headers" $ \store -> do+        seedInspection store+        let sql statement = Pool.use store.pool (Session.script statement) >>= either (fail . show) pure+            config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxResponseBytes = 1024}}+        app <- opsApplication config hooks store+        let readHeader = call app methodGet "/wf/inspect/header/approval/wf-1" ""+        warm <- readHeader+        shouldBe (simpleStatus warm) status200+        sql "UPDATE keiro.keiro_workflows SET last_error=repeat('x',100000) WHERE workflow_id='wf-1'"+        tooLarge <- readHeader+        assertError tooLarge status413 "inspection_result_too_large"+        sql "UPDATE keiro.keiro_workflows SET last_error=NULL WHERE workflow_id='wf-1'"+        restored <- readHeader+        shouldBe (body restored) (body warm)+        sql "ALTER TABLE keiro.keiro_timers ADD COLUMN header_test_drift text"+        refused <- readHeader+        assertError refused status409 "schema_drift"+        allowed <- opsApplication config {Keiro.Ops.Http.Config.schemaDrift = AllowDrift} hooks store+        first <- call allowed methodGet "/wf/inspect/header/approval/wf-1" ""+        second <- call allowed methodGet "/wf/inspect/header/approval/wf-1" ""+        shouldBe (simpleBody second) (simpleBody first)+        shouldBe (lookup "Keiro-Ops-Schema-Drift" (simpleHeaders second)) (Just "1")+    describe "HTTP boundary" $ do+      it "shares the supported read/write classification across every mounted domain" $ \store -> do+        seedInspection store+        app <- opsApplication defaultOpsHttpConfig hooks store+        let domainWords = Set.fromList (concat (mapParser (\_ option -> case Types.optMain option of Types.CmdReader _ commands -> map (Text.pack . fst) commands; _ -> []) (Ops.commandParser hooks)))+        domainWords `shouldBe` Set.fromList ("pgmq" : "replay-audit" : map (Text.takeWhile (/= '/') . Text.drop 1 . fst) commandPairs)+        forM_ commandPairs $ \(readPath, mutationPath) -> do+          response <- call app methodGet readPath ""+          (readPath, simpleStatus response, body response) `shouldSatisfy` (\(_, status, _) -> status == status200)+          readPost <- call app methodPost readPath ""+          simpleStatus readPost `shouldBe` status405+          lookup hAllow (simpleHeaders readPost) `shouldBe` Just methodGet+          mutationGet <- call app methodGet mutationPath ""+          simpleStatus mutationGet `shouldBe` status405+          lookup hAllow (simpleHeaders mutationGet) `shouldBe` Just methodPost+          disabled <- call app methodPost mutationPath "{\"confirm\":\"execute\"}"+          assertError disabled status403 "mutations_disabled"+      it "renders a real preview and applies the confirmed operation only" $ \store -> do+        seedInspection store+        app <- opsApplication defaultOpsHttpConfig {mutations = MutationsEnabled} hooks store+        preview <- call app methodPost "/stream/soft-delete/order-1" ""+        simpleStatus preview `shouldBe` status200+        field "outcome" (body preview) `shouldBe` Just (String "preview")+        beforeStream <- runStoreIO store (getStream (StreamName "order-1")) >>= either (fail . show) pure+        fmap (isJust . (.deletedAt)) beforeStream `shouldBe` Just False+        direct <- case parseArguments hooks ["stream", "soft-delete", "order-1"] of+          Parsed command -> Ops.runCommand hooks (OpsEnv store Json False [] False) command+          Rejected err -> fail (Text.unpack err)+        case direct of+          PreviewRequired result invocation -> do+            field "result" (body preview) `shouldBe` Just result.jsonValue+            field "cli_reinvocation" (body preview) `shouldBe` Just (String invocation)+          _ -> expectationFailure "expected supported preview"+        confirmed <- call app methodPost "/stream/soft-delete/order-1" "{\"confirm\":\"execute\"}"+        simpleStatus confirmed `shouldBe` status200+        field "outcome" (body confirmed) `shouldBe` Just (String "executed")+        afterStream <- runStoreIO store (getStream (StreamName "order-1")) >>= either (fail . show) pure+        fmap (isJust . (.deletedAt)) afterStream `shouldBe` Just True+        forM_ ["{\"confirm\":true}", "{\"confirm\":\"yes\"}", "{\"confirm\":\"execute\",\"extra\":1}", "[]"] $ \bad -> do+          response <- call app methodPost "/stream/soft-delete/order-1" bad+          assertError response status400 "invalid_confirmation"+      it "requires preview and confirmation for DLQ reads that reserve messages" $ \store -> do+        seedInspection store+        readOnly <- opsApplication defaultOpsHttpConfig hooks store+        denied <- call readOnly methodPost "/pgmq/dlq/read?queue=http_test" ""+        assertError denied status403 "mutations_disabled"+        getResponse <- call readOnly methodGet "/pgmq/dlq/read?queue=http_test" ""+        assertError getResponse status405 "method_not_allowed"+        app <- opsApplication defaultOpsHttpConfig {mutations = MutationsEnabled} hooks store+        preview <- call app methodPost "/pgmq/dlq/read?queue=http_test" ""+        field "outcome" (body preview) `shouldBe` Just (String "preview")+        response <- call app methodPost "/pgmq/dlq/read?queue=http_test" "{\"confirm\":\"execute\"}"+        field "outcome" (body response) `shouldBe` Just (String "executed")+        field "result" (body response) `shouldSatisfy` maybe False (\case Array items -> not (null items); _ -> False)+        next <- call app methodPost "/pgmq/dlq/read?queue=http_test" "{\"confirm\":\"execute\"}"+        field "result" (body next) `shouldBe` Just (Array mempty)+      it "rejects invalid requests and has structured errors" $ \store -> do+        app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+        forM_ ["/wf/list?limit=abc", "/wf", "/nope", "/wf/list?force=true", "/wf/list?%FF=x", "/wf/list?name=%FF", "/wf/--force"] $ \path -> do+          response <- call app methodGet path ""+          assertError response status400 "invalid_request"+        response <- call app methodGet "/" ""+        assertError response status404 "not_found"+      it "normalizes parsed origins and rejects malformed or repeated origins" $ \_ -> do+        let allowed headers = Cors.originAllowed ["https://console.example"] False headers+        forM_ [[], [("Origin", "http://EXAMPLE.test:80"), (hHost, "example.test")], [("Origin", "HTTPS://CONSOLE.EXAMPLE:443")]] $ \headers ->+          allowed headers `shouldBe` True+        forM_ [[("Origin", "null")], [("Origin", "https://console.example/path")], [("Origin", "https://console.example#fragment")], [("Origin", "https://user@console.example")], [("Origin", "https://console.example https://other.example")], [("Origin", "https://console.example"), ("Origin", "https://console.example")], [("Origin", "http://console.example")], [("Origin", "https://console.example:444")], [("Origin", "https://evil.example"), ("X-Forwarded-Host", "evil.example")]] $ \headers ->+          allowed headers `shouldBe` False+      it "rejects foreign mutations before reading their body or verifying schema" $ \store -> do+        count <- newIORef (0 :: Int)+        budget <- newInspectionBudget defaultInspectionLimits+        app <-+          opsApplicationWithOperations+            defaultOpsHttpConfig {mutations = MutationsEnabled}+            budget+            hooks+            store+            (OpsHttpOperations (modifyIORef' count (+ 1) >> pure (Right [])) (\_ _ -> modifyIORef' count (+ 1) >> pure (Succeeded emptyResult)))+        let foreignApp request respond = app (request {requestBody = fail "foreign request body was read"}) respond+            req = (setPath defaultRequest "/stream/soft-delete/order-1") {requestMethod = methodPost, requestHeaders = [("Origin", "https://evil.example"), (hContentType, "application/json")]}+        refused <- runSession (srequest (SRequest req "{\"confirm\":\"execute\"}")) foreignApp+        assertError refused status403 "origin_not_allowed"+        readIORef count `shouldReturn` 0+      it "requires JSON for nonempty bodies including chunked confirmation, and permits empty previews" $ \store -> do+        app <- opsApplication defaultOpsHttpConfig {mutations = MutationsEnabled} hooks store+        let post headers payload chunked =+              let req = (setPath defaultRequest "/stream/soft-delete/order-1") {requestMethod = methodPost, requestHeaders = headers}+                  wrapped request respond = app (if chunked then request {requestBodyLength = ChunkedBody} else request) respond+               in runSession (srequest (SRequest req payload)) wrapped+        forM_ [False, True] $ \chunked -> do+          forM_ [[], [(hContentType, "text/plain")]] $ \headers ->+            post headers "{\"confirm\":\"execute\"}" chunked >>= \response -> assertError response status415 "unsupported_media_type"+          preview <- post [] "" chunked+          simpleStatus preview `shouldBe` status200+        forM_ [[("Origin", "http://example.test:80"), (hHost, "EXAMPLE.test")], [("Origin", "https://console.example")], []] $ \origins -> do+          allowed <- opsApplication defaultOpsHttpConfig {mutations = MutationsEnabled, Config.allowedOrigins = ["https://console.example"]} hooks store+          response <- runSession (srequest (SRequest ((setPath defaultRequest "/outbox/maintenance-pass") {requestMethod = methodPost, requestHeaders = (hContentType, "Application/JSON; charset=utf-8") : origins}) "{\"confirm\":\"execute\"}")) allowed+          simpleStatus response `shouldBe` status200+      it "applies exact-origin CORS without credentials or duplicate headers" $ \store -> do+        app <- opsApplication defaultOpsHttpConfig {Config.allowedOrigins = ["http://ui.example"]} Ops.emptyAppHooks store+        let request origin method = (setPath defaultRequest "/wf/list") {requestMethod = method, requestHeaders = [("Origin", origin), ("Access-Control-Request-Method", "GET")]}+        allowed <- runSession (srequest (SRequest (request "http://ui.example" methodGet) "")) app+        lookup "Access-Control-Allow-Origin" (simpleHeaders allowed) `shouldBe` Just "http://ui.example"+        lookup "Access-Control-Allow-Credentials" (simpleHeaders allowed) `shouldBe` Nothing+        preflight <- runSession (srequest (SRequest (request "http://ui.example" methodOptions) "")) app+        simpleStatus preflight `shouldBe` status204+        denied <- runSession (srequest (SRequest (request "http://other.example" methodGet) "")) app+        lookup "Access-Control-Allow-Origin" (simpleHeaders denied) `shouldBe` Nothing+        disabledApp <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+        disabled <- runSession (srequest (SRequest (request "http://ui.example" methodGet) "")) disabledApp+        lookup "Access-Control-Allow-Origin" (simpleHeaders disabled) `shouldBe` Nothing+      it "bounds requests and results before responding" $ \store -> do+        let config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxRequestBytes = 30, maxResponseBytes = 2}}+        app <- opsApplication config Ops.emptyAppHooks store+        tooLarge <- call app methodGet ("/wf/list?name=" <> Text.replicate 100 "x") ""+        assertError tooLarge status413 "inspection_request_too_large"+        health <- call app methodGet "/health" ""+        assertError health status413 "inspection_result_too_large"+        enabled <- opsApplication defaultOpsHttpConfig {mutations = MutationsEnabled} hooks store+        oversized <- call enabled methodPost "/outbox/maintenance-pass" (LazyByteString.replicate 1025 32)+        assertError oversized status413 "inspection_request_too_large"+      it "keeps hook-dependent commands absent unless mounted and preserves supported failures" $ \store -> do+        standalone <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+        absent <- call standalone methodGet "/rebuild/list" ""+        assertError absent status400 "invalid_request"+        app <- opsApplication defaultOpsHttpConfig hooks store+        audited <- call app methodGet "/replay-audit?full=true" ""+        assertError audited status422 "command_failed"+        listed <- call app methodGet "/rebuild/list" ""+        direct <- case parseArguments hooks ["rebuild", "list"] of+          Parsed command -> Ops.runCommand hooks (OpsEnv store Json False [] False) command+          Rejected err -> fail (Text.unpack err)+        case direct of+          Succeeded result -> body listed `shouldBe` result.jsonValue+          _ -> expectationFailure "expected supported catalog read"+      it "refuses excess requests before verification and cancels schema work on deadline" $ \store -> do+        count <- newIORef (0 :: Int)+        entered <- newEmptyMVar+        let config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxConcurrentReads = 1, readDeadlineSeconds = 1}}+            operations = OpsHttpOperations (modifyIORef' count (+ 1) >> putMVar entered () >> threadDelay 3000000 >> pure (Right [])) (\_ _ -> pure (Succeeded emptyResult))+        budget <- newInspectionBudget config.inspectionLimits+        app <- opsApplicationWithOperations config budget Ops.emptyAppHooks store operations+        withAsync (call app methodGet "/wf/list" "") $ \job -> do+          takeMVar entered+          refused <- call app methodGet "/wf/list" ""+          assertError refused status503 "inspection_busy"+          readIORef count `shouldReturn` 1+          response <- wait job+          assertError response status504 "inspection_timeout"+      it "holds admission during response delivery and releases it when the client cancels" $ \store -> do+        entered <- newEmptyMVar+        gate <- newEmptyMVar+        count <- newIORef (0 :: Int)+        let config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxConcurrentReads = 1}}+            operations = OpsHttpOperations (modifyIORef' count (+ 1) >> pure (Right [])) (\_ _ -> pure (Succeeded emptyResult))+        budget <- newInspectionBudget config.inspectionLimits+        app <- opsApplicationWithOperations config budget Ops.emptyAppHooks store operations+        let slow request respond = app request (\response -> putMVar entered () >> takeMVar gate >> respond response)+        withAsync (call slow methodGet "/health" "") $ \job -> do+          takeMVar entered+          refused <- call app methodGet "/health" ""+          assertError refused status503 "inspection_busy"+          readIORef count `shouldReturn` 1+          cancel job+        recovered <- call app methodGet "/health" ""+        simpleStatus recovered `shouldBe` status200+      it "expires four blocked deliveries once, returns all permits, and recovers health" $ \store -> do+        entered <- newChan+        blocked <- newEmptyMVar+        responses <- newIORef (0 :: Int)+        cancelled <- newIORef (0 :: Int)+        let config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxConcurrentReads = 4, deliveryDeadlineSeconds = 1}}+        budget <- newInspectionBudget config.inspectionLimits+        app <- opsApplicationWithOperations config budget Ops.emptyAppHooks store (OpsHttpOperations (pure (Right [])) (\_ _ -> pure (Succeeded emptyResult)))+        let slow request respond =+              app+                request+                ( \response -> do+                    atomicModifyIORef' responses (\n -> (n + 1, ()))+                    writeChan entered ()+                    (takeMVar blocked >> respond response) `finally` atomicModifyIORef' cancelled (\n -> (n + 1, ()))+                )+        jobs <- replicateM 4 (async (try (call slow methodGet "/health" "") :: IO (Either SomeException SResponse)))+        timeout 1000000 (replicateM 4 (readChan entered)) `shouldReturn` Just [(), (), (), ()]+        refused <- call app methodGet "/health" ""+        assertError refused status503 "inspection_busy"+        finished <- timeout 3000000 (mapM wait jobs)+        case finished of+          Just outcomes -> forM_ outcomes $ \outcome -> outcome `shouldSatisfy` either (const True) (const False)+          Nothing -> mapM_ cancel jobs >> expectationFailure "delivery did not terminate"+        readIORef responses `shouldReturn` 4+        readIORef cancelled `shouldReturn` 4+        call app methodGet "/health" "" >>= \response -> simpleStatus response `shouldBe` status200+      it "mounts idly without recurring verification and skips SQL for disabled/invalid requests" $ \store -> do+        count <- newIORef (0 :: Int)+        budget <- newInspectionBudget defaultInspectionLimits+        app <-+          opsApplicationWithOperations+            defaultOpsHttpConfig+            budget+            Ops.emptyAppHooks+            store+            (OpsHttpOperations (modifyIORef' count (+ 1) >> pure (Right [])) (\_ _ -> pure (Succeeded emptyResult)))+        threadDelay 100000+        _ <- call app methodGet "/wf/list?force=true" ""+        _ <- call app methodGet "/wf/inspect/list?after=not-a-cursor" ""+        _ <- call app methodGet "/wf/inspect/list?limit=501" ""+        _ <- call app methodPost "/outbox/maintenance-pass" ""+        readIORef count `shouldReturn` 0+      it "serves through a loopback socket and releases its port" $ \store -> do+        app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+        manager <- Client.newManager Client.defaultManagerSettings+        port <- withOpsHttpServer defaultOpsHttpServerConfig {port = 0} app $ \server -> do+          request <- Client.parseRequest ("http://127.0.0.1:" <> show server.serverPort <> "/health")+          response <- Client.httpLbs request manager+          Client.responseStatus response `shouldBe` status200+          pure server.serverPort+        request <- Client.parseRequest ("http://127.0.0.1:" <> show port <> "/health")+        freshManager <- Client.newManager Client.defaultManagerSettings+        Client.httpLbs request manager `shouldThrow` (\(_ :: Client.HttpException) -> True)+        Client.httpLbs request freshManager `shouldThrow` (\(_ :: Client.HttpException) -> True)+      it "refuses foreign origins and non-JSON confirmation over a real HTTP socket" $ \store -> do+        app <- opsApplication defaultOpsHttpConfig {mutations = MutationsEnabled} hooks store+        manager <- Client.newManager Client.defaultManagerSettings+        withOpsHttpServer defaultOpsHttpServerConfig {port = 0} app $ \server -> do+          base <- Client.parseRequest ("http://127.0.0.1:" <> show server.serverPort <> "/outbox/maintenance-pass")+          forM_ [([("Origin", "https://evil.example"), (hContentType, "application/json")], status403), ([(hContentType, "text/plain")], status415)] $ \(headers, status) -> do+            let request = base {Client.method = methodPost, Client.requestHeaders = headers, Client.requestBody = Client.RequestBodyLBS "{\"confirm\":\"execute\"}"}+            response <- Client.httpLbs request manager+            Client.responseStatus response `shouldBe` status+      it "terminates a backpressured Warp delivery and recovers admission" $ \store -> do+        entered <- newEmptyMVar+        delivered <- newEmptyMVar+        let bodyBytes = 32 * 1024 * 1024+            config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxConcurrentReads = 1, deliveryDeadlineSeconds = 1, maxResponseBytes = 64 * 1024 * 1024}}+            large = emptyResult {jsonValue = String (Text.replicate bodyBytes "x")}+        budget <- newInspectionBudget config.inspectionLimits+        app <- opsApplicationWithOperations config budget Ops.emptyAppHooks store (OpsHttpOperations (pure (Right [])) (\_ _ -> pure (Succeeded large)))+        let watched request respond =+              app+                request+                ( \response -> do+                    putMVar entered ()+                    respond response `finally` putMVar delivered ()+                )+        withOpsHttpServer defaultOpsHttpServerConfig {port = 0} watched $ \server ->+          bracket (Socket.socket Socket.AF_INET Socket.Stream Socket.defaultProtocol) Socket.close $ \client -> do+            Socket.setSocketOption client Socket.RecvBuffer 4096+            Socket.connect client (Socket.SockAddrInet (fromIntegral server.serverPort) (Socket.tupleToHostAddress (127, 0, 0, 1)))+            Socket.sendAll client "GET /wf/list HTTP/1.1\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n"+            timeout 5000000 (takeMVar entered) `shouldReturn` Just ()+            tryTakeMVar delivered `shouldReturn` Nothing+            -- The 32-MiB response exceeds socket buffers; no response bytes+            -- have been read. The callback is still inside actual Warp delivery.+            refused <- call app methodGet "/health" ""+            assertError refused status503 "inspection_busy"+            timeout 3000000 (takeMVar delivered) `shouldReturn` Just ()+            recovered <- call app methodGet "/health" ""+            simpleStatus recovered `shouldBe` status200+            Socket.setSocketOption client Socket.RecvBuffer (1024 * 1024)+            let drain total = do+                  chunk <- Socket.recv client 65536+                  if Char8.null chunk then pure total else drain (total + Char8.length chunk)+            terminated <- timeout 5000000 (try (drain 0) :: IO (Either IOException Int))+            case terminated of+              Just (Right count) -> count `shouldSatisfy` (< bodyBytes)+              Just (Left _) -> pure ()+              Nothing -> expectationFailure "expired Warp connection remained open"+      it "joins idle accepted connections across repeated start/stop churn" $ \store -> do+        app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+        forM_ [1 .. (20 :: Int)] $ \_ ->+          withOpsHttpServer defaultOpsHttpServerConfig {port = 0} app $ \server ->+            bracket (Socket.socket Socket.AF_INET Socket.Stream Socket.defaultProtocol) Socket.close $ \client -> do+              Socket.connect client (Socket.SockAddrInet (fromIntegral server.serverPort) (Socket.tupleToHostAddress (127, 0, 0, 1)))+              stopOpsHttpServer server+              result <- timeout 1000000 (try (Socket.recv client 1))+              case result of+                Just (Left (_ :: IOException)) -> pure ()+                Just (Right bytes) | Char8.null bytes -> pure ()+                _ -> expectationFailure "stopped runner retained an idle connection"+  around (withFreshDatabase fixture) $ do+    describe "process-manager discovery database boundary" $ do+      it "performs only aggregate-types verifier work and rejects parsing before borrowing" $ \url ->+        withStore (defaultConnectionSettings url) $ \store -> do+          calls <- newIORef (0 :: Int)+          let observe = \case Observation.ConnectionObservation _ Observation.InUseConnectionStatus -> atomicModifyIORef' calls (\n -> (n + 1, ())); _ -> pure ()+              settings = (defaultConnectionSettings url) {observationHandler = Just observe}+              mounted = Ops.emptyAppHooks {Ops.processManagers = Just jitsureiProcessManagerInspectors, Ops.aggregates = Just Map.empty}+          withOpsInspectionPool defaultInspectionLimits settings $ \pool -> do+            app <- opsApplication defaultOpsHttpConfig mounted store {pool}+            before <- readIORef calls+            invalid <- call app methodGet "/pm/types?name=invalid" ""+            assertError invalid status400 "invalid_request"+            readIORef calls `shouldReturn` before+            aggregate <- call app methodGet "/aggregate/types" ""+            simpleStatus aggregate `shouldBe` status200+            afterAggregate <- readIORef calls+            manager <- call app methodGet "/pm/types" ""+            simpleStatus manager `shouldBe` status200+            afterManager <- readIORef calls+            afterAggregate `shouldSatisfy` (> before)+            afterManager - afterAggregate `shouldBe` afterAggregate - before+            repeated <- call app methodGet "/pm/types" ""+            body repeated `shouldBe` body manager+            afterRepeated <- readIORef calls+            afterRepeated - afterManager `shouldBe` afterAggregate - before+    describe "bounded workflow header" $ do+      it "uses one supported instance read even with 100000 children" $ \url ->+        withStore (defaultConnectionSettings url) $ \store -> do+          seedInspection store+          executeSql url "INSERT INTO keiro.keiro_workflow_children(child_id,child_name,parent_id,parent_name,await_step) SELECT 'child-'||n,'child','wf-1','approval','await-'||n FROM generate_series(1,100000) n"+          calls <- newIORef (0 :: Int)+          let observe = \case Observation.ConnectionObservation _ Observation.InUseConnectionStatus -> atomicModifyIORef' calls (\n -> (n + 1, ())); _ -> pure ()+              settings = (defaultConnectionSettings url) {observationHandler = Just observe}+          withOpsInspectionPool defaultInspectionLimits settings $ \pool -> do+            app <- opsApplication defaultOpsHttpConfig hooks store {pool}+            response <- call app methodGet "/wf/inspect/header/approval/wf-1" ""+            simpleStatus response `shouldBe` status200+            field "workflow_id" (body response) `shouldBe` Just (String "wf-1")+            field "children" (body response) `shouldBe` Nothing+            -- One schema-verification session and one instance-read session.+            readIORef calls `shouldReturn` 2+    describe "schema handshake" $ do+      it "bounds SQL on every inspection connection and releases pool capacity on cancellation" $ \url -> do+        let limits = defaultInspectionLimits {statementTimeoutMillis = 100, maxConcurrentReads = 1}+        withOpsInspectionPool limits (defaultConnectionSettings url) $ \pool -> do+          slow <- Pool.use pool (Session.script "SELECT pg_sleep(1)")+          slow `shouldSatisfy` either (const True) (const False)+          result <- timeout 1000000 (Pool.use pool (Session.script "SELECT 1"))+          result `shouldBe` Just (Right ())+        withOpsInspectionPool defaultInspectionLimits {maxConcurrentReads = 1} (defaultConnectionSettings url) $ \pool ->+          bracket (Connection.acquire (Settings.connectionString url) >>= either (fail . show) pure) Connection.release $ \control ->+            withAsync (Pool.use pool (Session.script "SELECT pg_sleep(10)")) $ \job -> do+              let active = do+                    result <- Connection.use control (Session.statement () (Statement.preparable "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND query='SELECT pg_sleep(10)' AND state='active')" E.noParams (D.singleRow (D.column (D.nonNullable D.bool)))))+                    case result of Right True -> pure (); Right False -> threadDelay 1000 >> active; Left err -> fail (show err)+              timeout 1000000 active `shouldReturn` Just ()+              cancel job+              timeout 1000000 (Pool.use pool (Session.script "SELECT 1")) `shouldReturn` Just (Right ())+        -- The host's ordinary connection was never configured by the pool.+        executeSql url "DO $$ BEGIN IF current_setting('statement_timeout') <> '0' THEN RAISE EXCEPTION 'writer timeout changed'; END IF; END $$"+      it "detects drift after construction, refuses previews/confirmation, and supports explicit policy" $ \url ->+        withStore (defaultConnectionSettings url) $ \store -> do+          app <- opsApplication defaultOpsHttpConfig {mutations = MutationsEnabled} Ops.emptyAppHooks store+          healthy <- call app methodGet "/health" ""+          simpleStatus healthy `shouldBe` status200+          executeSql url "ALTER TABLE keiro.keiro_timers ADD COLUMN http_test_drift text"+          forM_ [(methodGet, "/wf/list", ""), (methodPost, "/outbox/maintenance-pass", ""), (methodPost, "/outbox/maintenance-pass", "{\"confirm\":\"execute\"}")] $ \(method, path, payload) -> do+            response <- call app method path payload+            assertError response status409 "schema_drift"+          health <- call app methodGet "/health" ""+          simpleStatus health `shouldBe` status503+          allowed <- opsApplication defaultOpsHttpConfig {Keiro.Ops.Http.Config.schemaDrift = AllowDrift} Ops.emptyAppHooks store+          response <- call allowed methodGet "/wf/list" ""+          simpleStatus response `shouldBe` status200+          lookup "Keiro-Ops-Schema-Drift" (simpleHeaders response) `shouldBe` Just "1"+    describe "CLI agreement" $ do+      it "agrees with executable JSON across the database-only command domains" $ \url ->+        withStore (defaultConnectionSettings url) $ \store -> do+          seedInspection store+          executable <- opsExecutable+          app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+          forM_ cliReads $ \(path, args) -> do+            (exit, out, err) <- readProcessWithExitCode executable (["--database-url", Text.unpack url] <> args <> ["--json"]) ""+            (path, exit, out, err) `shouldSatisfy` (\(_, code, _, _) -> code == ExitSuccess)+            response <- call app methodGet path ""+            simpleStatus response `shouldBe` status200+            -- These are two observations at different times. Compare stable+            -- fields while retaining the timestamp/age shape checks in ops.+            fmap stableObservation (Aeson.eitherDecode (simpleBody response)) `shouldBe` (fmap stableObservation (Aeson.eitherDecode (LazyByteString.fromStrict (Char8.pack out))) :: Either String Value)+            err `shouldBe` ""++stableObservation :: Value -> Value+stableObservation (Object fields) = Object (KeyMap.map stableObservation (foldr KeyMap.delete fields ["observed_at", "lease_age_seconds", "due_in_seconds"]))+stableObservation (Array items) = Array (fmap stableObservation items)+stableObservation value = value++hooks :: Ops.AppHooks+hooks =+  Ops.emptyAppHooks+    { Ops.workflowResume = Just (Map.empty, defaultWorkflowResumeOptions),+      Ops.timerFire = Just (\_ -> pure Nothing),+      Ops.replayAudit = Just (Audit.OpsAuditConfig []),+      Ops.projectionCatalog =+        Just+          ( case Catalog.validateProjectionCatalog Catalog.emptyProjectionCatalog of+              Catalog.Success catalog -> CatalogOperations.projectionCatalogOperations catalog+              Catalog.Failure errors -> error (show errors)+          )+    }++commandPairs :: [(Text, Text)]+commandPairs =+  [ ("/wf/list", "/wf/cancel/missing/missing"),+    ("/timer/stuck/list", "/timer/cancel/00000000-0000-0000-0000-000000000001"),+    ("/outbox/backlog", "/outbox/maintenance-pass"),+    ("/inbox/backlog", "/inbox/gc"),+    ("/projection/position?subscription=http_test", "/projection/prune-dedup?projection=http_test&before=2026-01-01T00%3A00%3A00Z"),+    ("/shard/status?subscription=http_test", "/shard/relinquish?subscription=http_test&worker=00000000-0000-0000-0000-000000000001"),+    ("/snapshot/show?stream=missing", "/snapshot/delete?stream=missing"),+    ("/stream/subscriptions", "/stream/soft-delete/missing"),+    ("/rebuild/list", "/rebuild/start/missing?run_id=test&requested_by=test&reason=test")+  ]++cliReads :: [(Text, [String])]+cliReads =+  [ ("/wf/list", ["wf", "list"]),+    ("/timer/pending/list?limit=1", ["timer", "pending", "list", "--limit", "1"]),+    ("/shard/list", ["shard", "list"]),+    ("/timer/stuck/list?min_age=0s", ["timer", "stuck", "list", "--min-age", "0s"]),+    ("/outbox/backlog", ["outbox", "backlog"]),+    ("/inbox/backlog", ["inbox", "backlog"]),+    ("/projection/position?subscription=http_test", ["projection", "position", "--subscription", "http_test"]),+    ("/projection/status?limit=1", ["projection", "status", "--limit", "1"]),+    ("/shard/status?subscription=http_test", ["shard", "status", "--subscription", "http_test"]),+    ("/snapshot/show?stream=missing", ["snapshot", "show", "--stream", "missing"]),+    ("/stream/show/order-1", ["stream", "show", "order-1"]),+    ("/stream/subscriptions", ["stream", "subscriptions"])+  ]++call :: Application -> Method -> Text -> LazyByteString.ByteString -> IO SResponse+call app method path payload = runSession (srequest (SRequest ((setPath defaultRequest (Text.encodeUtf8 path)) {requestMethod = method, requestHeaders = [(hContentType, "application/json")]}) payload)) app++body :: SResponse -> Value+body response = either error id (Aeson.eitherDecode (simpleBody response))++field :: Aeson.Key -> Value -> Maybe Value+field key (Object value) = KeyMap.lookup key value+field _ _ = Nothing++assertError :: SResponse -> Status -> Text -> Expectation+assertError response expected code = do+  simpleStatus response `shouldBe` expected+  (field "error" (body response) >>= field "code") `shouldBe` Just (String code)++executeSql :: Text -> Text -> IO ()+executeSql url sql = bracket acquire Connection.release $ \connection -> do+  result <- Connection.use connection (Session.script sql)+  either (fail . show) pure result+  where+    acquire = Connection.acquire (Settings.connectionString url) >>= either (fail . show) pure++opsExecutable :: IO FilePath+opsExecutable = do+  (exit, out, err) <- readProcessWithExitCode "cabal" ["list-bin", "exe:keiro-ops"] ""+  if exit == ExitSuccess then pure (Text.unpack (Text.strip (Text.pack out))) else fail err++newChanCounter :: IO (MVar Int)+newChanCounter = newMVar 0++signalCounter :: MVar Int -> IO ()+signalCounter counter = modifyMVar_ counter (pure . (+ 1))++awaitCounter :: MVar Int -> Int -> IO ()+awaitCounter counter expected = do+  result <- timeout 2000000 loop+  result `shouldBe` Just ()+  where+    loop = do count <- readMVar counter; if count == expected then pure () else threadDelay 1000 >> loop++seedInspection :: KirokuStore -> IO ()+seedInspection store = do+  now <- getCurrentTime+  _ <-+    runStoreIO+      store+      ( appendToStream+          (StreamName "order-1")+          AnyVersion+          [ EventData Nothing (EventType "order.created") (object ["order" .= (1 :: Int)]) Nothing Nothing Nothing,+            EventData Nothing (EventType "order.confirmed") (object ["order" .= (1 :: Int)]) Nothing Nothing Nothing+          ]+      )+      >>= either (fail . show) pure+  runStoreIO store (appendJournalEntry (WorkflowName "approval") (WorkflowId "wf-1") (StepRecorded "received" Null now)) >>= either (fail . show) pure+  let timer = Timer.TimerRequest (Timer.TimerId UUID.nil) "billing" "invoice-1" (addUTCTime (-60) now) Null+  runStoreIO store (runTransaction (Timer.scheduleTimerTx timer)) >>= either (fail . show) pure+  let job = Job {jobName = "http_test", jobQueue = queueRef "http_test", jobCodec = aesonJobCodec, jobOrdering = Unordered, jobPolicy = defaultRetryPolicy} :: Job Value+  result <- runJobEff (JobRuntime store.pool Nothing) $ do+    ensureJobQueue job+    _ <- enqueue job (object ["poison" .= True])+    _ <- runJobOnce 1 job (\_ -> pure (Dead "test poison"))+    pure ()+  either (fail . show) pure result

@@ -0,0 +1,137 @@+module Test.ComposeSpec (spec) where++import Control.Monad (forM_, void)+import Data.Aeson (Value (..), encode, object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KeyMap+import Data.ByteString.Builder qualified as Builder+import Data.ByteString.Lazy qualified as Lazy+import Data.Text (Text)+import Data.Time (getCurrentTime)+import Effectful (Eff, IOE)+import Effectful.Error.Static (Error)+import Keiro.Ops (emptyAppHooks)+import Keiro.Ops.Http+import Keiro.Ops.Http.Application (opsApplicationWithBudget)+import Keiro.Ops.Http.Config qualified as Config+import Keiro.Ops.Http.WebSocket (feedsMiddleware)+import Keiro.Test.Postgres (Fixture, withFreshStore)+import Keiro.Workflow (WorkflowId (..), WorkflowJournalEvent (..), WorkflowName (..), appendJournalEntry)+import Kiroku.Metrics qualified as Metrics+import Kiroku.Store.Connection (KirokuStore)+import Kiroku.Store.Effect (Store, runStoreIO)+import Kiroku.Store.Error (StoreError)+import Network.HTTP.Types+import Network.Wai+import Network.Wai.Test qualified as Wai+import Network.WebSockets qualified as WS+import System.Timeout (timeout)+import Test.Hspec++spec :: Fixture -> Spec+spec fixture = describe "Compose" $ do+  it "strips exactly one raw segment and preserves route spelling, queries and streaming bodies" $ do+    let app = composed defaultComposeConfig [MountedSurface keiroPrefix echo]+    forM_+      [ ("/keiro", "/"),+        ("/keiro/", "/"),+        ("/k%65iro/ws?from=3", "/ws"),+        ("/keiro/a%2Fb/%25/%E6%97%A5/", "/a%2Fb/%25/%E6%97%A5/"),+        ("/keiro/x//y/", "/x//y/")+      ]+      $ \(path, relative) -> do+        let req = Wai.setPath defaultRequest path+        Just stripped <- pure (stripSurfacePrefix keiroPrefix req)+        rawPathInfo stripped `shouldBe` relative+        rawQueryString stripped `shouldBe` rawQueryString req+        queryString stripped `shouldBe` queryString req+        response <- get app req+        Wai.simpleStatus response `shouldBe` status200+        Wai.simpleBody response `shouldBe` Lazy.fromStrict relative+    stripSurfacePrefix keiroPrefix ((Wai.setPath defaultRequest "/keiro/one") {rawPathInfo = "/keiro/two"}) `shouldSatisfy` maybe True (const False)+  it "validates mounts, lists them in order, refuses absent surfaces and applies CORS once" $ do+    forM_ ["", "-x", "UPPER", "x/y"] $ \prefix -> mkSurfacePrefix prefix `shouldBe` Left (InvalidPrefix prefix)+    case composeSurfaces defaultComposeConfig [] of Left NoSurfaces -> pure (); _ -> expectationFailure "empty mount accepted"+    case composeSurfaces defaultComposeConfig [MountedSurface keiroPrefix echo, MountedSurface keiroPrefix echo] of+      Left (DuplicatePrefix "keiro") -> pure ()+      _ -> expectationFailure "duplicate mount accepted"+    let app = composed (ComposeConfig ["http://ui.test"]) [MountedSurface keiroPrefix echo, MountedSurface kirokuPrefix existingCors]+    index <- get app (Wai.setPath defaultRequest "/")+    decode (Wai.simpleBody index) `shouldBe` object ["surfaces" .= [object ["prefix" .= ("keiro" :: Text), "path" .= ("/keiro" :: Text)], object ["prefix" .= ("kiroku" :: Text), "path" .= ("/kiroku" :: Text)]]]+    refused <- get app (Wai.setPath defaultRequest "/pgmq/queues")+    Wai.simpleStatus refused `shouldBe` status404+    method <- get app ((Wai.setPath defaultRequest "/") {requestMethod = methodPost})+    lookup hAllow (Wai.simpleHeaders method) `shouldBe` Just "GET"+    forM_ [("/keiro/x", "http://ui.test"), ("/kiroku/x", "http://already.test")] $ \(path, origin) -> do+      response <- get app ((Wai.setPath defaultRequest path) {requestHeaders = [(hOrigin, "http://ui.test")]})+      filter ((== "Access-Control-Allow-Origin") . fst) (Wai.simpleHeaders response) `shouldBe` [("Access-Control-Allow-Origin", origin)]+      lookup "Access-Control-Allow-Credentials" (Wai.simpleHeaders response) `shouldBe` Nothing+    let preflight = (Wai.setPath defaultRequest "/keiro/x") {requestMethod = methodOptions, requestHeaders = [(hOrigin, "http://ui.test"), ("Access-Control-Request-Method", "GET"), ("Access-Control-Request-Headers", "X-Request-Id")]}+    response <- get app preflight+    Wai.simpleStatus response `shouldBe` status204+    lookup "Access-Control-Allow-Headers" (Wai.simpleHeaders response) `shouldBe` Just "X-Request-Id"+    forM_ [composed defaultComposeConfig [MountedSurface keiroPrefix echo], app] $ \candidate -> do+      denied <- get candidate ((Wai.setPath defaultRequest "/keiro/x") {requestHeaders = [(hOrigin, "http://other.test")]})+      lookup "Access-Control-Allow-Origin" (Wai.simpleHeaders denied) `shouldBe` Nothing+  around (withFreshStore fixture) $ it "mounts real Keiro and Kiroku applications and both socket protocols on one port" $ \store -> do+    now <- getCurrentTime+    void (stored store (appendJournalEntry (WorkflowName "compose") (WorkflowId "one") (StepRecorded "start" Null now)))+    let config = defaultOpsHttpConfig {Config.allowedOrigins = ["http://ui.test"], mutations = MutationsEnabled}+        metricsConfig = Metrics.defaultConfig {Metrics.wsPushIntervalUs = 30000000}+    metrics <- Metrics.newKirokuMetrics store+    providers <- Metrics.storeServerProviders metricsConfig metrics store+    let kiroku = Metrics.combinedAppWithProviders metricsConfig metrics [] providers+    withFeedHandle config defaultFeedConfig store $ \handle -> do+      http <- opsApplicationWithBudget config (handleBudget handle) emptyAppHooks store+      let keiro = feedsMiddleware config handle http+          app = composed (ComposeConfig config.allowedOrigins) [MountedSurface keiroPrefix keiro, MountedSurface kirokuPrefix kiroku]+      forM_ [(keiro, "/health", "/keiro/health"), (kiroku, "/capabilities", "/kiroku/capabilities")] $ \(standalone, relative, mounted) -> do+        plain <- get standalone (Wai.setPath defaultRequest relative)+        through <- get app ((Wai.setPath defaultRequest mounted) {requestHeaders = [(hOrigin, "http://ui.test")]})+        Wai.simpleStatus through `shouldBe` Wai.simpleStatus plain+        Wai.simpleBody through `shouldBe` Wai.simpleBody plain+        filter ((== "Access-Control-Allow-Origin") . fst) (Wai.simpleHeaders through) `shouldBe` [("Access-Control-Allow-Origin", "http://ui.test")]+      forM_ [("http://ui.test", status200), ("http://evil.test", status403)] $ \(origin, status) -> do+        let request = (Wai.setPath defaultRequest "/keiro/outbox/maintenance-pass") {requestMethod = methodPost, requestHeaders = [(hOrigin, origin)]}+        response <- Wai.runSession (Wai.srequest (Wai.SRequest request "")) app+        Wai.simpleStatus response `shouldBe` status+      withOpsHttpServer defaultOpsHttpServerConfig {port = 0} app $ \server -> within $ do+        WS.runClient "127.0.0.1" server.serverPort "/kiroku/ws/metrics" $ \connection -> do+          receive connection >>= hasType "snapshot"+          WS.sendTextData connection (encode (object ["type" .= ("ping" :: Text)]))+          receive connection >>= hasType "pong"+        WS.runClient "127.0.0.1" server.serverPort "/keiro/ws?from=3" $ \connection -> do+          WS.sendTextData connection (encode (object ["type" .= ("subscribe" :: Text), "feed" .= ("workflows" :: Text)]))+          receive connection >>= hasType "snapshot"+          void (stored store (appendJournalEntry (WorkflowName "compose") (WorkflowId "one") (WorkflowCompleted now)))+          receive connection >>= hasType "update"+          closeFeedHandle handle+          receive connection >>= hasType "goodbye"+      feedStats handle `shouldReturn` FeedStats 0 0 False++composed :: ComposeConfig -> [MountedSurface] -> Application+composed config = either (error . show) id . composeSurfaces config++echo :: Application+echo req respond = respond (responseStream status200 [] (\write flush -> write (Builder.byteString (rawPathInfo req)) >> flush))++existingCors :: Application+existingCors req respond = echo req (respond . mapResponseHeaders (("Access-Control-Allow-Origin", "http://already.test") :))++get :: Application -> Request -> IO Wai.SResponse+get app req = Wai.runSession (Wai.srequest (Wai.SRequest req "")) app++decode :: Lazy.ByteString -> Value+decode = either error id . Aeson.eitherDecode++receive :: WS.Connection -> IO Value+receive connection = decode <$> (WS.receiveData connection :: IO Lazy.ByteString)++hasType :: Text -> Value -> Expectation+hasType kind = \case Object fields -> KeyMap.lookup "type" fields `shouldBe` Just (String kind); value -> expectationFailure (show value)++within :: IO a -> IO a+within action = timeout 15000000 action >>= maybe (fail "composed socket deadline") pure++stored :: KirokuStore -> Eff '[Store, Error StoreError, IOE] a -> IO a+stored store action = runStoreIO store action >>= either (fail . show) pure

@@ -0,0 +1,228 @@+module Test.FeedSpec (spec) where++import Control.Concurrent.Async (async, cancel, wait)+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Exception (bracket, finally)+import Control.Monad (forM_)+import Data.Aeson (Value, object, (.=))+import Data.IORef+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Time (addUTCTime, getCurrentTime)+import Data.Word (Word64)+import Effectful (liftIO)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Keiro.Ops.Http.Config+import Keiro.Ops.Http.Feed qualified as Feed+import Keiro.Ops.Http.Protocol qualified as P+import Keiro.Ops.Http.Views+import Keiro.Test.Postgres (Fixture, withFreshStore)+import Keiro.Wake (WakeReason (..), neverWake, waitForWake)+import Kiroku.Store.Connection (KirokuStore (..))+import System.Timeout (timeout)+import Test.Hspec++spec :: Fixture -> Spec+spec fixture = do+  describe "shared feed driver" $ do+    it "diffs fallback truth, extra-only changes, and recovery without duplicate reads per subscriber" $ do+      let removedItem = Item "two" (object ["id" .= ("two" :: Text)]) (body (item 2))+      source <- newIORef (Right ([item 1, removedItem], Nothing))+      let readItems = do+            value <- readIORef source+            stamp <- getCurrentTime+            pure (fmap (\(rows, extra) -> (stamp, rows, extra)) value)+      bracket (Feed.newDriver P.Workflows quick (waitForWake neverWake 20000) readItems) Feed.stopDriver $ \driver -> do+        one <- Feed.subscribe driver+        two <- Feed.subscribe driver+        first <- receive one+        receive two `shouldReturn` first+        snapshotSequence first `shouldBe` Just 0+        writeIORef source (Right ([item 3], Nothing))+        changed <- receive one+        receive two `shouldReturn` changed+        case changed of+          P.Update P.Workflows 1 P.Timeout _ rows removed _ -> do+            rows `shouldBe` [body (item 3)]+            removed `shouldBe` [keyJson removedItem]+          _ -> expectationFailure (show changed)+        timeout 100000 (atomically (Feed.nextFrame one)) `shouldReturn` Nothing+        writeIORef source (Left "read unavailable")+        forM_ [one, two] $ \subscriber ->+          receive subscriber >>= \case+            P.ErrorFrame P.ReadFailed _ _ -> pure ()+            frame -> expectationFailure (show frame)+        writeIORef source (Right ([item 3], Just (object ["visible_store_head" .= (9 :: Int)])))+        recovered <- receive one+        receive two `shouldReturn` recovered+        case recovered of+          P.Update P.Workflows 2 _ _ rows removed _ -> (rows, removed) `shouldBe` ([], [])+          _ -> expectationFailure (show recovered)+    it "signals overflow before retained deltas and resubscribes from the current sequence" $ do+      source <- newIORef [item 1]+      calls <- newIORef (0 :: Int)+      wake <- newEmptyMVar+      let readItems = do+            modifyIORef' calls (+ 1)+            rows <- readIORef source+            stamp <- getCurrentTime+            pure (Right (stamp, rows, Nothing))+      bracket (Feed.newDriver P.Workflows quick {Feed.queueCapacity = 2} (takeMVar wake >> pure WokenByNotify) readItems) Feed.stopDriver $ \driver -> do+        subscriber <- Feed.subscribe driver+        _ <- receive subscriber+        forM_ [1 .. 5] $ \sequence -> do+          writeIORef source [item (fromIntegral sequence + 1)]+          putMVar wake ()+          reached <- timeout 1000000 $ atomically $ do+            frame <- Feed.driverSnapshot driver+            check (snapshotSequence frame == Just sequence)+          reached `shouldBe` Just ()+        receive subscriber >>= \case+          P.ErrorFrame P.QueueOverflow _ _ -> pure ()+          frame -> expectationFailure (show frame)+        -- The overflow control frame also consumes the strict queue budget;+        -- retaining it can evict one more old delta. Recovery uses a snapshot.+        receive subscriber >>= \case+          P.Update P.Workflows 5 _ _ rows _ _ -> rows `shouldBe` [body (item 6)]+          frame -> expectationFailure (show frame)+        recovered <- Feed.subscribe driver >>= receive+        snapshotSequence recovered `shouldBe` Just 5+        readIORef calls `shouldReturn` 6+    it "retains fresh timer presentation without broadcasting a clock-only delta" $ do+      firstStamp <- getCurrentTime+      let secondStamp = addUTCTime 1 firstStamp+          timer countdown = Item "timer" (object ["timer_id" .= ("timer" :: Text)]) (object ["status" .= ("scheduled" :: Text), "due_in_seconds" .= (countdown :: Int)])+      source <- newIORef (firstStamp, [timer 10])+      wake <- newEmptyMVar+      let readItems = do+            (stamp, rows) <- readIORef source+            pure (Right (stamp, rows, Nothing))+      bracket (Feed.newDriver P.Timers quick (takeMVar wake >> pure WokenByNotify) readItems) Feed.stopDriver $ \driver -> do+        one <- Feed.subscribe driver+        _ <- receive one+        writeIORef source (secondStamp, [timer 9])+        putMVar wake ()+        refreshed <- timeout 7000000 $ atomically $ do+          frame <- Feed.driverSnapshot driver+          case frame of+            P.Snapshot _ 0 _ stamp rows _ -> check (stamp == secondStamp && rows == [body (timer 9)])+            _ -> retry+        refreshed `shouldBe` Just ()+        timeout 100000 (atomically (Feed.nextFrame one)) `shouldReturn` Nothing+        fresh <- Feed.subscribe driver >>= receive+        case fresh of+          P.Snapshot P.Timers 0 _ stamp rows _ -> (stamp, rows) `shouldBe` (secondStamp, [body (timer 9)])+          _ -> expectationFailure (show fresh)+  describe "feed registry" $ around (withFreshStore fixture) $ do+    it "reserves initialization, joins cancellation, shares a ready view, and drops the idle wake cursor" $ \store ->+      Feed.withFeedHandle defaultOpsHttpConfig quick {Feed.maxViews = 1, Feed.maxFrameBytes = 512} store $ \handle -> do+        Feed.feedStats handle `shouldReturn` Feed.FeedStats 0 0 False+        entered <- newEmptyMVar+        gate <- newEmptyMVar+        cleaned <- newEmptyMVar+        let blocked = View "blocked" P.Workflows (liftIO ((putMVar entered () >> takeMVar gate) `finally` putMVar cleaned ()))+            other = View "other" P.Workflows (liftIO (getCurrentTime >>= \stamp -> pure (stamp, [], Nothing)))+        request <- async (Feed.acquireDriver handle blocked)+        timeout 1000000 (takeMVar entered) `shouldReturn` Just ()+        refused <- Feed.acquireDriver handle other+        either (== Feed.TooManyViews) (const False) refused `shouldBe` True+        cancel request+        tryTakeMVar cleaned `shouldReturn` Just ()+        Feed.feedStats handle `shouldReturn` Feed.FeedStats 0 0 False+        let large = View "large" P.Workflows $ liftIO $ do+              stamp <- getCurrentTime+              pure (stamp, [Item "large" (object []) (object ["text" .= Text.replicate 1000 "x"])], Nothing)+        oversized <- Feed.acquireDriver handle large+        either (== Feed.InitialResultTooLarge) (const False) oversized `shouldBe` True+        Feed.feedStats handle `shouldReturn` Feed.FeedStats 0 0 False+        calls <- newIORef (0 :: Int)+        let shared = View "shared" P.Workflows $ liftIO $ do+              modifyIORef' calls (+ 1)+              stamp <- getCurrentTime+              pure (stamp, [item 1], Nothing)+        Right first <- Feed.acquireDriver handle shared+        Right second <- Feed.acquireDriver handle shared+        readIORef calls `shouldReturn` 1+        atomically (Feed.driverSnapshot first) >>= \frame -> atomically (Feed.driverSnapshot second) `shouldReturn` frame+        Feed.releaseDriver handle shared+        Feed.feedStats handle `shouldReturn` Feed.FeedStats 1 0 True+        Feed.releaseDriver handle shared+        Feed.feedStats handle `shouldReturn` Feed.FeedStats 0 0 False+        Feed.closeFeedHandle handle+        Feed.closeFeedHandle handle+    it "refuses a retiring view distinctly until blocked cleanup has joined" $ \store ->+      Feed.withFeedHandle defaultOpsHttpConfig quick {Feed.maxViews = 1} store $ \handle -> do+        block <- newIORef False+        entered <- newEmptyMVar+        cleanupEntered <- newEmptyMVar+        cleanupGate <- newEmptyMVar+        never <- newEmptyMVar+        let view = View "retiring" P.Workflows $ liftIO $ do+              shouldBlock <- readIORef block+              if shouldBlock+                then (putMVar entered () >> takeMVar never) `finally` (putMVar cleanupEntered () >> takeMVar cleanupGate)+                else pure ()+              stamp <- getCurrentTime+              pure (stamp, [item 1], Nothing)+        Right driver <- Feed.acquireDriver handle view+        _ <- Feed.subscribe driver >>= receive+        writeIORef block True+        timeout 2000000 (takeMVar entered) `shouldReturn` Just ()+        release <- async (Feed.releaseDriver handle view)+        ( do+            timeout 1000000 (takeMVar cleanupEntered) `shouldReturn` Just ()+            refused <- Feed.acquireDriver handle view+            either (== Feed.FeedRetiring) (const False) refused `shouldBe` True+            Feed.feedStats handle `shouldReturn` Feed.FeedStats 1 0 True+            writeIORef block False+          )+          `finally` (tryPutMVar cleanupGate () >> pure ())+        timeout 1000000 (wait release) `shouldReturn` Just ()+        Feed.feedStats handle `shouldReturn` Feed.FeedStats 0 0 False+        Right _ <- Feed.acquireDriver handle view+        Feed.releaseDriver handle view+    it "rechecks schema for each active view and recovers only after drift is removed" $ \store ->+      Feed.withFeedHandle defaultOpsHttpConfig quick store $ \handle -> do+        firstCalls <- newTVarIO (0 :: Int)+        secondCalls <- newTVarIO (0 :: Int)+        let view name calls = View name P.Workflows $ liftIO $ do+              atomically (modifyTVar' calls (+ 1))+              stamp <- getCurrentTime+              pure (stamp, [item 1], Nothing)+            firstView = view "schema-first" firstCalls+            secondView = view "schema-second" secondCalls+            sql text = Pool.use store.pool (Session.script text) >>= either (fail . show) pure+        Right first <- Feed.acquireDriver handle firstView+        one <- Feed.subscribe first+        _ <- receive one+        Right second <- Feed.acquireDriver handle secondView+        two <- Feed.subscribe second+        _ <- receive two+        sql "ALTER TABLE keiro.keiro_workflows ADD COLUMN feed_drift bigint"+        forM_ [one, two] $ \subscriber ->+          receive subscriber >>= \case+            P.ErrorFrame P.ReadFailed _ _ -> pure ()+            frame -> expectationFailure (show frame)+        before <- atomically ((,) <$> readTVar firstCalls <*> readTVar secondCalls)+        sql "ALTER TABLE keiro.keiro_workflows DROP COLUMN feed_drift"+        recovered <- timeout 2000000 $ atomically $ do+          after <- (,) <$> readTVar firstCalls <*> readTVar secondCalls+          check (fst after > fst before && snd after > snd before)+        recovered `shouldBe` Just ()+        forM_ [first, second] $ \driver -> Feed.subscribe driver >>= receive >>= \frame -> snapshotSequence frame `shouldBe` Just 0+        Feed.releaseDriver handle firstView+        Feed.releaseDriver handle secondView++quick :: Feed.FeedConfig+quick = Feed.defaultFeedConfig {Feed.fallbackInterval = 0.02, Feed.minRefreshInterval = 0.001, Feed.readTimeout = 2}++item :: Int -> Item+item value = Item "one" (object ["id" .= ("one" :: Text)]) (object ["value" .= value])++snapshotSequence :: P.ServerFrame -> Maybe Word64+snapshotSequence = \case P.Snapshot _ sequence _ _ _ _ -> Just sequence; _ -> Nothing++receive :: Feed.Subscriber -> IO P.ServerFrame+receive subscriber = timeout 2000000 (atomically (Feed.nextFrame subscriber)) >>= maybe (fail "feed frame deadline exceeded") pure

@@ -0,0 +1,167 @@+module Test.WebSocketSpec (spec) where++import Control.Monad (forM_, void)+import Data.Aeson (Value (..), object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KeyMap+import Data.ByteString qualified as Bytes+import Data.ByteString.Char8 qualified as Char8+import Data.ByteString.Lazy qualified as Lazy+import Data.Text (Text)+import Data.Time (UTCTime, addUTCTime, getCurrentTime)+import Data.UUID qualified as UUID+import Effectful (Eff, IOE)+import Effectful.Error.Static (Error)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Keiro.Ops (emptyAppHooks)+import Keiro.Ops.Http hiding (subscribe)+import Keiro.Ops.Projection (projectionGroupStatusJson)+import Keiro.ReadModel (registerReadModel)+import Keiro.ReadModel.Rebuild+import Keiro.Test.Postgres (Fixture, withFreshStore)+import Keiro.Timer qualified as Timer+import Keiro.Workflow (WorkflowId (..), WorkflowJournalEvent (..), WorkflowName (..), appendJournalEntry)+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 (EventId (..))+import Network.HTTP.Client qualified as HTTP+import Network.HTTP.Types+import Network.Wai (Application, defaultRequest)+import Network.Wai.Test qualified as Wai+import Network.WebSockets qualified as WS+import System.Timeout (timeout)+import Test.Hspec++spec :: Fixture -> Spec+spec fixture = describe "inspection sockets" $ around (withFreshStore fixture) $ do+  it "shares workflow truth, retains a subscription on refusal, and releases each sent frame" $ \store -> do+    now <- getCurrentTime+    void (stored store (appendJournalEntry (WorkflowName "approval") (WorkflowId "one") (StepRecorded "start" Null now)))+    serve store defaultOpsHttpConfig socketConfig {maxViews = 1} $ \handle app port ->+      within $ WS.runClient "127.0.0.1" port "/ws?client=one" $ \one ->+        WS.runClient "127.0.0.1" port "/ws" $ \two -> do+          forM_ [one, two] $ \connection -> send connection (subscribe "workflows") >> receive connection >>= assertType "snapshot"+          feedStats handle `shouldReturn` FeedStats 1 2 True+          -- With two queue slots, five sequential sends detect unreleased+          -- in-flight frames or a writer recursing inside its bracket.+          forM_ [1 .. 5 :: Int] $ \_ -> send one (object ["type" .= ("ping" :: Text)]) >> receive one >>= assertType "pong"+          send one (object ["type" .= ("subscribe" :: Text), "feed" .= ("workflows" :: Text), "workflow_name" .= ("other" :: Text)])+          refused <- receive one+          field "code" refused `shouldBe` Just (String "too_many_views")+          forM_+            [ (object ["type" .= ("subscribe" :: Text), "feed" .= ("unknown" :: Text)], "unknown_feed"),+              (object ["type" .= ("subscribe" :: Text), "feed" .= ("timers" :: Text), "bad_filter" .= True], "invalid_subscription")+            ]+            $ \(frame, code) -> do+              send one frame+              response <- receive one+              field "code" response `shouldBe` Just (String code)+          WS.sendTextData one ("{broken" :: Text)+          malformed <- receive one+          field "code" malformed `shouldBe` Just (String "invalid_frame")++          void (stored store (appendJournalEntry (WorkflowName "approval") (WorkflowId "one") (WorkflowCompleted now)))+          update <- receive one+          receive two `shouldReturn` update+          assertType "update" update+          field "trigger" update `shouldBe` Just (String "notify")+          field "sequence" update `shouldBe` Just (Aeson.toJSON (1 :: Int))+          response <- Wai.runSession (Wai.srequest (Wai.SRequest (Wai.setPath defaultRequest "/wf/inspect/list") "")) app+          field "items" (decode (Wai.simpleBody response)) `shouldBe` field "changed" update+          forM_ [one, two] $ \connection -> do+            send connection (object ["type" .= ("unsubscribe" :: Text), "feed" .= ("workflows" :: Text)])+            receive connection >>= assertType "unsubscribed"+          feedStats handle `shouldReturn` FeedStats 0 2 False+  it "streams timer fallback truth and projection extra-only changes using the poll encoders" $ \store ->+    serve store defaultOpsHttpConfig socketConfig $ \_ _ port -> within $+      WS.runClient "127.0.0.1" port "/ws" $ \connection -> do+        forM_ ["timers", "projection_groups"] $ \feed -> send connection (subscribe feed) >> receive connection >>= assertType "snapshot"+        now <- getCurrentTime+        let timer = Timer.TimerRequest (Timer.TimerId UUID.nil) "manager" "one" (addUTCTime (-60) now) Null+        stored store (runTransaction (Timer.scheduleTimerTx timer))+        added <- receive connection+        field "feed" added `shouldBe` Just (String "timers")+        field "trigger" added `shouldBe` Just (String "timeout")+        Just observed <- pure (field "observed_at" added)+        Aeson.Success stamp <- pure (Aeson.fromJSON observed :: Aeson.Result UTCTime)+        Just inspection <- stored store (Timer.lookupTimerInspection timer.timerId)+        field "changed" added `shouldBe` Just (Aeson.toJSON [Timer.pendingTimerItemJson stamp inspection])+        Just _ <- stored store (Timer.claimDueTimer now)+        stored store (Timer.markTimerFired timer.timerId (EventId UUID.nil)) `shouldReturn` True+        removed <- receive connection+        field "removed" removed `shouldBe` Just (Aeson.toJSON [object ["timer_id" .= UUID.toText UUID.nil]])+        _ <- stored store (registerReadModel "socket-model" 1 "shape")+        projection <- receive connection+        field "feed" projection `shouldBe` Just (String "projection_groups")+        Right page <- stored store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 Nothing))+        field "changed" projection `shouldBe` Just (Aeson.toJSON (map projectionGroupStatusJson page.items))+        void (stored store (appendJournalEntry (WorkflowName "head") (WorkflowId "one") (StepRecorded "head" Null now)))+        headOnly <- receive connection+        field "feed" headOnly `shouldBe` Just (String "projection_groups")+        field "changed" headOnly `shouldBe` Just (Aeson.toJSON ([] :: [Value]))+        field "removed" headOnly `shouldBe` Just (Aeson.toJSON ([] :: [Value]))+        field "visible_store_head" headOnly `shouldSatisfy` (\case Just (Number position) -> position > 0; _ -> False)+  it "checks upgrade origin, capacity and schema policy and closes active sockets" $ \store -> do+    manager <- HTTP.newManager HTTP.defaultManagerSettings+    serve store defaultOpsHttpConfig socketConfig {maxConnections = 1} $ \handle _ port -> do+      forbidden <- upgrade manager port [(hOrigin, "http://evil.test")]+      HTTP.responseStatus forbidden `shouldBe` status403+      let origin = "http://127.0.0.1:" <> show port+      within $ WS.runClientWith "127.0.0.1" port "/ws" WS.defaultConnectionOptions [("Origin", fromStringBytes origin)] $ \connection -> do+        full <- upgrade manager port []+        HTTP.responseStatus full `shouldBe` status503+        closeFeedHandle handle+        receive connection >>= assertType "goodbye"+      feedStats handle `shouldReturn` FeedStats 0 0 False+    result <- Pool.use store.pool (Session.script "ALTER TABLE keiro.keiro_timers ADD COLUMN socket_test_drift text")+    either (fail . show) pure result+    serve store defaultOpsHttpConfig socketConfig $ \_ _ port -> do+      refused <- upgrade manager port []+      HTTP.responseStatus refused `shouldBe` status409+    let allow = defaultOpsHttpConfig {schemaDrift = AllowDrift}+    serve store allow socketConfig $ \_ _ port -> within $ WS.runClient "127.0.0.1" port "/ws" $ \connection -> do+      send connection (subscribe "timers")+      receive connection >>= assertType "snapshot"++socketConfig :: FeedConfig+socketConfig = defaultFeedConfig {fallbackInterval = 5, minRefreshInterval = 0.01, readTimeout = 2, queueCapacity = 2}++serve :: KirokuStore -> OpsHttpConfig -> FeedConfig -> (FeedHandle -> Application -> Int -> IO a) -> IO a+serve store config feeds action = withFeedHandle config feeds store $ \handle -> do+  app <- opsApplicationWithFeeds config handle emptyAppHooks store+  withOpsHttpServer defaultOpsHttpServerConfig {port = 0} app (\server -> action handle app server.serverPort)++subscribe :: Text -> Value+subscribe feed = object ["type" .= ("subscribe" :: Text), "feed" .= feed]++send :: WS.Connection -> Value -> IO ()+send connection = WS.sendTextData connection . Aeson.encode++receive :: WS.Connection -> IO Value+receive connection = timeout 7000000 (WS.receiveData connection :: IO Lazy.ByteString) >>= maybe (fail "socket frame deadline exceeded") (pure . decode)++decode :: Lazy.ByteString -> Value+decode = either error id . Aeson.eitherDecode++field :: Aeson.Key -> Value -> Maybe Value+field key = \case Object fields -> KeyMap.lookup key fields; _ -> Nothing++assertType :: Text -> Value -> IO ()+assertType expected value = field "type" value `shouldBe` Just (String expected)++stored :: KirokuStore -> Eff '[Store, Error StoreError, IOE] a -> IO a+stored store action = runStoreIO store action >>= either (fail . show) pure++within :: IO a -> IO a+within action = timeout 25000000 action >>= maybe (fail "socket client deadline exceeded") pure++upgrade :: HTTP.Manager -> Int -> RequestHeaders -> IO (HTTP.Response Lazy.ByteString)+upgrade manager port headers = do+  request <- HTTP.parseRequest ("http://127.0.0.1:" <> show port <> "/ws")+  HTTP.httpLbs request {HTTP.requestHeaders = headers <> [(hConnection, "Upgrade"), (hUpgrade, "websocket"), ("Sec-WebSocket-Version", "13"), ("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")]} manager++fromStringBytes :: String -> Bytes.ByteString+fromStringBytes = Char8.pack