plexus-synapse 3.5.0 → 3.10.1
raw patch · 21 files changed
+2984/−76 lines, 21 filesdep +fast-loggerdep +katipdep ~aesonnew-component:exe:protocol-test-demonew-component:exe:reporter-demonew-component:exe:validation-report-test
Dependencies added: fast-logger, katip
Dependency ranges changed: aeson
Files
- app/Main.hs +136/−13
- plexus-synapse.cabal +83/−2
- src/Synapse/Algebra/Navigate.hs +6/−2
- src/Synapse/Algebra/Recursion.hs +8/−4
- src/Synapse/Backend/Discovery.hs +41/−22
- src/Synapse/CLI/Help.hs +11/−1
- src/Synapse/CLI/Parse.hs +2/−10
- src/Synapse/Log.hs +106/−0
- src/Synapse/Monad.hs +30/−11
- src/Synapse/Self/Commands.hs +62/−0
- src/Synapse/Self/Debugger.hs +390/−0
- src/Synapse/Self/Protocol/Reporter.hs +322/−0
- src/Synapse/Self/Protocol/StreamTracker.hs +272/−0
- src/Synapse/Self/Protocol/TestRunner.hs +375/−0
- src/Synapse/Self/Protocol/Validator.hs +582/−0
- src/Synapse/Transport.hs +37/−11
- test/ProtocolTestDemo.hs +51/−0
- test/ReporterDemo.hs +114/−0
- test/StreamTrackerSpec.hs +101/−0
- test/ValidationReportTest.hs +99/−0
- test/WebSocketRaw.hs +156/−0
app/Main.hs view
@@ -30,12 +30,16 @@ import Options.Applicative import Data.Version (showVersion) import qualified Paths_plexus_synapse as Meta+import System.Directory (doesFileExist, getHomeDirectory) import System.Exit (exitFailure, exitSuccess)+import System.FilePath ((</>)) import System.IO (hPutStrLn, stderr, hFlush, stdout) import Synapse.Schema.Types import Synapse.Monad (SynapseM, SynapseEnv(..), SynapseError(..), BackendErrorType(..), TransportContext(..), TransportErrorCategory(..), initEnv, runSynapseM, throwNav, throwTransport, throwParse, throwBackend)+import qualified Synapse.Log as Log+import qualified Katip import Synapse.Algebra.Navigate import Synapse.Algebra.Render (renderSchema) import Synapse.CLI.Help (renderMethodHelp)@@ -76,6 +80,10 @@ , soBidirMode :: Maybe Text -- ^ Bidirectional mode override , soBidirCmd :: Maybe Text -- ^ Bidirectional subprocess command (--bidir-cmd) , soNoCache :: Bool -- ^ Disable IR caching+ , soLogLevel :: Maybe Text -- ^ Log level: info, debug, trace+ , soLogSubsystems :: [Text] -- ^ Filter logs by subsystems (empty = all)+ , soToken :: Maybe Text -- ^ JWT sent as Cookie: access_token=<jwt>+ , soTokenFile :: Maybe Text -- ^ Path to token file (overridden by --token) } deriving Show @@ -99,6 +107,56 @@ defaultPort = 4444 -- ============================================================================+-- Logging Helper+-- ============================================================================++-- | Create a logger from command-line options+-- Default: ErrorS (only show errors)+makeLoggerFromOpts :: SynapseOpts -> IO Log.Logger+makeLoggerFromOpts opts = do+ let level = case soLogLevel opts of+ Nothing -> Katip.ErrorS -- Default: only errors+ Just levelStr -> parseLogLevel levelStr+ Log.makeLogger level+ where+ parseLogLevel "error" = Katip.ErrorS+ parseLogLevel "warn" = Katip.WarningS+ parseLogLevel "info" = Katip.InfoS+ parseLogLevel "debug" = Katip.DebugS+ parseLogLevel _ = Katip.ErrorS -- Default to error++-- ============================================================================+-- Token Resolution+-- ============================================================================++-- | Resolve the auth token to use for this invocation.+--+-- Priority:+-- 1. --token <jwt> (explicit, highest priority)+-- 2. --token-file <path> (explicit file)+-- 3. ~/.plexus/tokens/<backend> (per-backend default)+--+-- Token files contain just the raw JWT, optionally with a trailing newline.+resolveToken :: SynapseOpts -> Text -> IO (Maybe Text)+resolveToken opts backend =+ case soToken opts of+ Just tok -> pure (Just tok)+ Nothing -> do+ mPath <- case soTokenFile opts of+ Just path -> pure (Just (T.unpack path))+ Nothing -> do+ home <- getHomeDirectory+ let defaultPath = home </> ".plexus" </> "tokens" </> T.unpack backend+ exists <- doesFileExist defaultPath+ pure $ if exists then Just defaultPath else Nothing+ case mPath of+ Nothing -> pure Nothing+ Just path -> do+ contents <- TIO.readFile path+ let tok = T.strip contents+ pure $ if T.null tok then Nothing else Just tok++-- ============================================================================ -- Argument Splitting -- ============================================================================ -- Main@@ -108,12 +166,24 @@ main = do args <- execParser argsInfo let opts = argOpts args++ -- Initialize logger (default: ErrorS, only show errors)+ logger <- makeLoggerFromOpts opts++ Log.logDebug logger Log.SubsystemCLI "synapse starting..."+ -- Use specified host/port for registry discovery let discovery = registryDiscovery (soHost opts) (soPort opts) -- Get the backend at the connection point (host:port) -- If it has a registry plugin, that's used for discovering other backends+ Log.logDebug logger Log.SubsystemDiscovery $+ "Checking backend at " <> soHost opts <> ":" <> T.pack (show (soPort opts))+ maybeBackend <- getBackendAt (soHost opts) (soPort opts)++ Log.logDebug logger Log.SubsystemDiscovery $+ "Backend check complete: " <> T.pack (show maybeBackend) let hostBackend = maybe "plexus" id maybeBackend -- Auto-register: if we're connecting to a non-default port, push the@@ -186,26 +256,60 @@ <> "' but " <> T.unpack (backendHost backend) <> ":" <> show (backendPort backend) <> " identifies as '" <> T.unpack name <> "'" exitFailure- _ ->- -- Match confirmed (or _info unavailable — proceed optimistically)+ Nothing -> do+ -- Protocol handshake failed when verifying backend+ logger <- makeLoggerFromOpts opts+ token <- resolveToken opts backendName+ env <- initEnv (soHost opts) (soPort opts) backendName logger token+ result <- runSynapseM env (throwBackend (ProtocolHandshakeFailed (backendHost backend) (backendPort backend)) [])+ case result of+ Left err -> do+ hPutStrLn stderr $ renderError err+ exitFailure+ Right () -> exitSuccess+ Just _ ->+ -- Match confirmed run backendName (backendHost backend) (backendPort backend) args Nothing -> do- -- Backend not found in registry - gather available backends and show error- backends <- discoverBackends discovery- backendsWithStatus <- pingBackends backends- env <- initEnv (soHost opts) (soPort opts) backendName- result <- runSynapseM env (throwBackend (BackendNotFound backendName) backendsWithStatus)- case result of- Left err -> do- hPutStrLn stderr $ renderError err- exitFailure- Right () -> exitSuccess+ -- Backend not found in registry - check if this was a direct connection attempt+ -- where the protocol handshake failed+ maybeDirectBackend <- getBackendAt (soHost opts) (soPort opts)+ logger <- makeLoggerFromOpts opts+ token <- resolveToken opts backendName+ env <- initEnv (soHost opts) (soPort opts) backendName logger token+ case maybeDirectBackend of+ Nothing -> do+ -- Protocol handshake failed at the specified host:port+ result <- runSynapseM env (throwBackend (ProtocolHandshakeFailed (soHost opts) (soPort opts)) [])+ case result of+ Left err -> do+ hPutStrLn stderr $ renderError err+ exitFailure+ Right () -> exitSuccess+ Just discoveredName | discoveredName /= backendName -> do+ -- Backend exists but has a different name+ backends <- discoverBackends discovery+ backendsWithStatus <- pingBackends backends+ result <- runSynapseM env (throwBackend (BackendNotFound backendName) backendsWithStatus)+ case result of+ Left err -> do+ hPutStrLn stderr $ renderError err+ exitFailure+ Right () -> exitSuccess+ Just _ -> do+ -- Backend discovered successfully - should not reach here, but proceed+ run backendName (soHost opts) (soPort opts) args -- | Run a Hub command with specified backend and connection details run :: Text -> Text -> Int -> Args -> IO () run backend host port args = do- env <- initEnv host port backend+ let opts = argOpts args+ logger <- makeLoggerFromOpts opts+ Log.logInfo logger Log.SubsystemCLI $ "Synapse starting: backend=" <> backend <> ", host=" <> host <> ", port=" <> T.pack (show port)+ token <- resolveToken opts backend+ env <- initEnv host port backend logger token rendererCfg <- defaultRendererConfig+ Log.logDebug logger Log.SubsystemCLI "Running dispatch..." result <- runSynapseM env (dispatch args rendererCfg) case result of Left err -> do@@ -594,6 +698,13 @@ NoBackendsAvailable -> "No backends available" <> renderBackendList backends+ ProtocolHandshakeFailed host port ->+ "Protocol handshake failed at " <> T.unpack host <> ":" <> show port <> "\n"+ <> "The _info request did not complete successfully within 2 seconds.\n"+ <> "This typically means:\n"+ <> " - The server is not responding to Plexus RPC protocol messages\n"+ <> " - The server is not sending required StreamDone messages\n"+ <> " - The connection is timing out" where showPath [] = "root" showPath p = T.unpack $ T.intercalate "." p@@ -843,4 +954,16 @@ soNoCache <- switch ( long "no-cache" <> help "Disable IR caching (force rebuild from schema)" )+ soLogLevel <- optional $ T.pack <$> strOption+ ( long "log-level" <> metavar "LEVEL"+ <> help "Log level: info, debug, trace (default: disabled)" )+ soLogSubsystems <- many $ T.pack <$> strOption+ ( long "log-subsystem" <> metavar "SUBSYSTEM"+ <> help "Filter logs by subsystem: discovery, transport, rpc, schema, cache, navigation (can specify multiple, default: all)" )+ soToken <- optional $ T.pack <$> strOption+ ( long "token" <> short 't' <> metavar "JWT"+ <> help "JWT sent as Cookie: access_token=<jwt> on WebSocket upgrade" )+ soTokenFile <- optional $ T.pack <$> strOption+ ( long "token-file" <> metavar "PATH"+ <> help "Path to token file (default lookup: ~/.plexus/tokens/<backend>)" ) pure SynapseOpts{..}
plexus-synapse.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: plexus-synapse-version: 3.5.0+version: 3.10.1 synopsis: Schema-driven CLI for Plexus RPC servers license: MIT author:@@ -17,6 +17,7 @@ Synapse.Schema.Types Synapse.Schema.Functor Synapse.Monad+ Synapse.Log Synapse.Algebra.Navigate Synapse.Algebra.Render Synapse.Algebra.Walk@@ -39,10 +40,16 @@ Synapse.Self.Template Synapse.Self.Pattern Synapse.Self.Examples+ Synapse.Self.Debugger+ Synapse.Self.Protocol.Validator+ Synapse.Self.Protocol.StreamTracker+ Synapse.Self.Protocol.Reporter+ Synapse.Self.Protocol.TestRunner build-depends: base >= 4.17 && < 5, plexus-protocol, aeson >= 2.0 && < 2.3,+ aeson-pretty >= 0.8 && < 0.9, text >= 2.0 && < 2.2, bytestring >= 0.11 && < 0.13, websockets >= 0.13 && < 0.14,@@ -65,7 +72,9 @@ regex-tdfa >= 1.3 && < 1.4, time >= 1.12 && < 1.15, array >= 0.5 && < 0.6,- process >= 1.6 && < 1.7+ process >= 1.6 && < 1.7,+ fast-logger >= 3.0 && < 3.3,+ katip >= 0.8 && < 0.9 hs-source-dirs: src default-language: GHC2021 default-extensions:@@ -92,6 +101,7 @@ containers >= 0.6 && < 0.8, directory >= 1.3 && < 1.4, filepath >= 1.4 && < 1.6,+ katip >= 0.8 && < 0.9, optparse-applicative >= 0.18 && < 0.19, prettyprinter >= 1.7 && < 1.8, time >= 1.12 && < 1.14@@ -118,6 +128,46 @@ if !flag(build-examples) buildable: False +executable reporter-demo+ main-is: ReporterDemo.hs+ build-depends:+ base >= 4.17 && < 5,+ plexus-synapse,+ text >= 2.0 && < 2.2+ hs-source-dirs: test+ default-language: GHC2021+ default-extensions:+ OverloadedStrings+ if !flag(build-examples)+ buildable: False++executable validation-report-test+ main-is: ValidationReportTest.hs+ build-depends:+ base >= 4.17 && < 5,+ plexus-synapse,+ text >= 2.0 && < 2.2,+ aeson >= 2.0 && < 2.3+ hs-source-dirs: test+ default-language: GHC2021+ default-extensions:+ OverloadedStrings+ if !flag(build-examples)+ buildable: False++executable protocol-test-demo+ main-is: ProtocolTestDemo.hs+ build-depends:+ base >= 4.17 && < 5,+ plexus-synapse,+ text >= 2.0 && < 2.2+ hs-source-dirs: test+ default-language: GHC2021+ default-extensions:+ OverloadedStrings+ if !flag(build-examples)+ buildable: False+ test-suite cli-test type: exitcode-stdio-1.0 main-is: CLISpec.hs@@ -197,6 +247,21 @@ OverloadedStrings ghc-options: -threaded +test-suite websocket-raw-test+ type: exitcode-stdio-1.0+ main-is: WebSocketRaw.hs+ build-depends:+ base >= 4.17 && < 5,+ text >= 2.0 && < 2.2,+ aeson >= 2.2 && < 2.3,+ bytestring >= 0.11 && < 0.13,+ websockets >= 0.13 && < 0.14+ hs-source-dirs: test+ default-language: GHC2021+ default-extensions:+ OverloadedStrings+ ghc-options: -threaded+ test-suite bidir-test type: exitcode-stdio-1.0 main-is: BidirSpec.hs@@ -229,4 +294,20 @@ default-extensions: OverloadedStrings RecordWildCards+ ghc-options: -threaded++test-suite stream-tracker-test+ type: exitcode-stdio-1.0+ main-is: StreamTrackerSpec.hs+ build-depends:+ base >= 4.17 && < 5,+ plexus-synapse,+ text >= 2.0 && < 2.2,+ aeson >= 2.2 && < 2.3,+ bytestring >= 0.11 && < 0.13,+ hspec >= 2.10 && < 2.12+ hs-source-dirs: test+ default-language: GHC2021+ default-extensions:+ OverloadedStrings ghc-options: -threaded
@@ -59,9 +59,13 @@ navigate :: Path -> SynapseM SchemaView navigate target = do root <- fetchSchemaAt []- -- Normalize path: strip leading root namespace if present+ -- Normalize path: strip leading root namespace if present,+ -- but only if it's NOT also a child name (otherwise the user+ -- is trying to navigate to a child that shares the root namespace)+ let hasChildWithName seg = any (\c -> csNamespace c == seg) (pluginChildren root) let normalizedTarget = case target of- (seg:rest) | seg == psNamespace root -> rest+ (seg:rest) | seg == psNamespace root+ , not (hasChildWithName seg) -> rest _ -> target withFreshVisited $ navigateFrom root [] normalizedTarget
src/Synapse/Algebra/Recursion.hs view
@@ -70,7 +70,9 @@ import Control.Monad (forM) import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (asks)+import Data.Maybe (catMaybes) import Data.Text (Text)+import System.IO (hPutStrLn, stderr) import Synapse.Schema.Types (Path, PluginSchema(..), MethodSchema(..), ChildSummary(..)) import Synapse.Schema.Functor (SchemaF(..), Fix(..), SchemaTree)@@ -181,13 +183,15 @@ env <- asks id -- Get the SynapseEnv childResults <- liftIO $ mapConcurrently (\child -> do- result <- runSynapseM env (go child) -- Fixed: env first, action second+ result <- runSynapseM env (go child) case result of- Left err -> error $ "Schema fetch failed: " ++ show err- Right val -> pure val+ Left err -> do+ hPutStrLn stderr $ "[synapse] warning: skipping plugin (fetch failed): " ++ show err+ pure Nothing+ Right val -> pure (Just val) ) children- alg (PluginF schema path childResults)+ alg (PluginF schema path (catMaybes childResults)) MethodF method ns path -> -- Methods are leaves, just apply algebra alg (MethodF method ns path)
src/Synapse/Backend/Discovery.hs view
@@ -71,6 +71,8 @@ import Plexus.Client (SubstrateConfig(..)) import qualified Plexus.Transport as ST import Plexus.Types (PlexusStreamItem(..), TransportError(..))+import qualified Synapse.Log as Log+import qualified System.IO as IO -- ============================================================================ -- Backend Type@@ -218,26 +220,39 @@ getBackendAt :: Text -> Int -> IO (Maybe Text) getBackendAt = discoverBackendName --- | Discover the backend name by calling _info+-- | Discover the backend name by calling _info (with 2000ms timeout) discoverBackendName :: Text -> Int -> IO (Maybe Text) discoverBackendName host port = do let cfg = SubstrateConfig- { substrateHost = T.unpack host- , substratePort = port- , substratePath = "/"+ { substrateHost = T.unpack host+ , substratePort = port+ , substratePath = "/" , substrateBackend = "" -- Not used for _info+ , substrateHeaders = [] }- result <- ST.rpcCallWith cfg "_info" Aeson.Null- `catch` \(_e :: SomeException) -> pure (Left $ NetworkError "Connection failed")+ timeout = do+ threadDelay 2000000 -- 2 second timeout+ pure (Left $ ConnectionTimeout host port)+ rpcCall = do+ ST.rpcCallWith cfg "_info" Aeson.Null+ `catch` \(_e :: SomeException) ->+ pure (Left $ NetworkError "Connection failed") - case result of- Left _err -> pure Nothing- Right items -> do+ -- Race the RPC call against a 2 second timeout+ raceResult <- race timeout rpcCall++ case raceResult of+ Left _err ->+ pure Nothing+ Right (Left _err) ->+ pure Nothing+ Right (Right items) -> do -- Extract the backend name from subscription response case [Aeson.fromJSON content | StreamData _ _ _ content <- items] of (Aeson.Success (Aeson.Object obj):_) | Just (Aeson.String name) <- KM.lookup "backend" obj -> pure (Just name)- _ -> pure Nothing+ _ ->+ pure Nothing -- | Query the registry for all backends queryRegistry :: Text -> Int -> Text -> IO [Backend]@@ -251,10 +266,11 @@ queryRegistryImpl :: Text -> Int -> Text -> IO [Backend] queryRegistryImpl host port backendName = do let cfg = SubstrateConfig- { substrateHost = T.unpack host- , substratePort = port- , substratePath = "/"+ { substrateHost = T.unpack host+ , substratePort = port+ , substratePath = "/" , substrateBackend = backendName+ , substrateHeaders = [] } -- Call registry.list through the backend result <- ST.invokeMethod cfg ["registry"] "list" (Aeson.object [])@@ -296,10 +312,11 @@ queryBackendImpl :: Text -> Int -> Text -> Text -> IO (Maybe Backend) queryBackendImpl host port backendName name = do let cfg = SubstrateConfig- { substrateHost = T.unpack host- , substratePort = port- , substratePath = "/"+ { substrateHost = T.unpack host+ , substratePort = port+ , substratePath = "/" , substrateBackend = backendName+ , substrateHeaders = [] } params = Aeson.object ["name" Aeson..= name] result <- ST.invokeMethod cfg ["registry"] "get" params@@ -330,10 +347,11 @@ pingBackend :: Backend -> IO Backend pingBackend backend = do let cfg = SubstrateConfig- { substrateHost = T.unpack (backendHost backend)- , substratePort = backendPort backend- , substratePath = "/"+ { substrateHost = T.unpack (backendHost backend)+ , substratePort = backendPort backend+ , substratePath = "/" , substrateBackend = "" -- _info has no namespace prefix+ , substrateHeaders = [] } timeout = threadDelay 300000 >> pure (Left $ ConnectionTimeout (backendHost backend) (backendPort backend)) rpcCall = ST.rpcCallWith cfg "_info" Aeson.Null@@ -370,10 +388,11 @@ -> IO () registerWithRegistry registryHost registryPort registryBackend name host port = do let cfg = SubstrateConfig- { substrateHost = T.unpack registryHost- , substratePort = registryPort- , substratePath = "/"+ { substrateHost = T.unpack registryHost+ , substratePort = registryPort+ , substratePath = "/" , substrateBackend = registryBackend+ , substrateHeaders = [] } params = Aeson.object [ "name" Aeson..= name
src/Synapse/CLI/Help.hs view
@@ -168,7 +168,17 @@ -- RefNamed "ConeIdentifier" -> "ConeIdentifier" renderTypeRef :: IR -> TypeRef -> Text renderTypeRef ir = \case- RefNamed qn -> qualifiedNameFull qn+ RefNamed qn ->+ -- If the named type is just a primitive or alias, unwrap it+ -- so newtypes like PaneRef(String) display as <string> not <panes.PaneRef>+ let name = qualifiedNameFull qn+ in case Map.lookup name (irTypes ir) of+ Just TypeDef{tdKind = KindPrimitive t mf}+ | Just fmt <- mf -> fmt+ | otherwise -> t+ Just TypeDef{tdKind = KindAlias target} ->+ renderTypeRef ir target+ _ -> name RefPrimitive typ mFormat | Just fmt <- mFormat -> fmt -- Use format as type (uuid, int64, etc.)
src/Synapse/CLI/Parse.hs view
@@ -203,9 +203,8 @@ else buildParamValue ir param{pdType = inner} kvs RefArray innerType ->- -- Collect all repeated --key value entries, splitting comma-separated values- let rawValues = [val | ("", val) <- kvs]- values = concatMap splitCommas rawValues+ -- Collect all repeated --key value entries (no comma splitting - use shell separation)+ let values = [val | ("", val) <- kvs] in if null values then Left $ InvalidValue (pdName param) "array requires at least one value" else Right $ Array $ V.fromList $ map (buildTypedValue ir innerType) values@@ -221,13 +220,6 @@ case lookup "" kvs of Just val -> Right $ inferValue val Nothing -> Left $ InvalidValue (pdName param) "unknown type needs value"---- | Split a string on commas, trimming whitespace. A value with no commas--- returns as a singleton list, preserving backward compatibility.-splitCommas :: Text -> [Text]-splitCommas t =- let parts = T.splitOn "," t- in map T.strip parts -- | Build value from a TypeDef buildFromTypeDef :: IR -> Text -> TypeDef -> [(Text, Text)] -> Either ParseError Value
+ src/Synapse/Log.hs view
@@ -0,0 +1,106 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Structured logging for Synapse using katip+--+-- Provides filterable, subsystem-tagged logging with severity levels:+-- - ErrorS: Errors that need attention+-- - WarningS: Warnings about potential issues+-- - InfoS: High-level operations (connections, discoveries)+-- - DebugS: Detailed flow (schema fetches, cache hits/misses)+--+-- Default log level: ErrorS (only errors shown)+module Synapse.Log+ ( -- * Types+ Severity(..)+ , Subsystem(..)+ , Logger++ -- * Logger Creation+ , makeLogger+ , closeLogger++ -- * Logging Functions+ , logError+ , logWarn+ , logInfo+ , logDebug+ ) where++import qualified Data.Text as T+import Data.Text (Text)+import Katip+import Control.Monad (void)+import Control.Monad.IO.Class (MonadIO, liftIO)+import System.IO (stdout)++-- | Subsystems for filtering (maps to katip namespace)+data Subsystem+ = SubsystemDiscovery+ | SubsystemTransport+ | SubsystemRPC+ | SubsystemSchema+ | SubsystemCache+ | SubsystemNavigation+ | SubsystemRendering+ | SubsystemCLI+ | SubsystemBidir+ deriving (Show, Eq)++-- | Logger handle+data Logger = Logger+ { logEnv :: LogEnv+ , logContext :: LogContexts+ , logNamespace :: Namespace+ }++-- | Convert subsystem to katip namespace+subsystemToNamespace :: Subsystem -> Namespace+subsystemToNamespace SubsystemDiscovery = "discovery"+subsystemToNamespace SubsystemTransport = "transport"+subsystemToNamespace SubsystemRPC = "rpc"+subsystemToNamespace SubsystemSchema = "schema"+subsystemToNamespace SubsystemCache = "cache"+subsystemToNamespace SubsystemNavigation = "navigation"+subsystemToNamespace SubsystemRendering = "rendering"+subsystemToNamespace SubsystemCLI = "cli"+subsystemToNamespace SubsystemBidir = "bidir"++-- | Create a logger with specified minimum severity+-- Default: ErrorS (only show errors)+makeLogger :: Severity -> IO Logger+makeLogger minSeverity = do+ handleScribe <- mkHandleScribe ColorIfTerminal stdout (permitItem minSeverity) V2+ env <- registerScribe "stdout" handleScribe defaultScribeSettings =<< initLogEnv "synapse" "production"+ pure Logger+ { logEnv = env+ , logContext = mempty+ , logNamespace = "synapse"+ }++-- | Close logger and flush buffers+closeLogger :: Logger -> IO ()+closeLogger logger = void $ closeScribes (logEnv logger)++-- | Log an error message+logError :: MonadIO m => Logger -> Subsystem -> Text -> m ()+logError logger subsys msg = liftIO $+ runKatipContextT (logEnv logger) (logContext logger) (logNamespace logger <> subsystemToNamespace subsys) $+ logFM ErrorS (logStr msg)++-- | Log a warning message+logWarn :: MonadIO m => Logger -> Subsystem -> Text -> m ()+logWarn logger subsys msg = liftIO $+ runKatipContextT (logEnv logger) (logContext logger) (logNamespace logger <> subsystemToNamespace subsys) $+ logFM WarningS (logStr msg)++-- | Log an info message+logInfo :: MonadIO m => Logger -> Subsystem -> Text -> m ()+logInfo logger subsys msg = liftIO $+ runKatipContextT (logEnv logger) (logContext logger) (logNamespace logger <> subsystemToNamespace subsys) $+ logFM InfoS (logStr msg)++-- | Log a debug message+logDebug :: MonadIO m => Logger -> Subsystem -> Text -> m ()+logDebug logger subsys msg = liftIO $+ runKatipContextT (logEnv logger) (logContext logger) (logNamespace logger <> subsystemToNamespace subsys) $+ logFM DebugS (logStr msg)
src/Synapse/Monad.hs view
@@ -41,6 +41,9 @@ -- * Cache Operations , lookupCache , insertCache++ -- * Logging+ , getLogger ) where import Control.Monad (when)@@ -58,6 +61,8 @@ import Synapse.Schema.Types import Synapse.Backend.Discovery (Backend(..))+import qualified Synapse.Log as Log+import qualified Katip -- | Environment for Synapse operations data SynapseEnv = SynapseEnv@@ -66,6 +71,8 @@ , seBackend :: !Text -- ^ Backend name (first CLI argument) , seCache :: !(IORef (HashMap PluginHash PluginSchema)) -- ^ Schema cache , seVisited :: !(IORef (HashSet PluginHash)) -- ^ Cycle detection+ , seLogger :: !Log.Logger -- ^ Structured logger+ , seToken :: !(Maybe Text) -- ^ JWT sent as Cookie: access_token=<jwt> } -- | Errors that can occur during Synapse operations@@ -83,6 +90,7 @@ = BackendNotFound Text | BackendUnreachable Text | NoBackendsAvailable+ | ProtocolHandshakeFailed Text Int -- host, port - _info request failed/timed out deriving stock (Show, Eq) -- | Transport error with connection details and categorization@@ -122,28 +130,31 @@ runSynapseM :: SynapseEnv -> SynapseM a -> IO (Either SynapseError a) runSynapseM env action = runReaderT (runExceptT (unSynapseM action)) env --- | Run a SynapseM action with default host/port and specified backend+-- | Run a SynapseM action with default host/port and specified backend (with error-level logger) runSynapseM' :: Text -> SynapseM a -> IO (Either SynapseError a) runSynapseM' backend action = do- env <- defaultEnv backend+ logger <- Log.makeLogger Katip.ErrorS+ env <- defaultEnv backend logger runSynapseM env action --- | Initialize environment with given host/port/backend-initEnv :: Text -> Int -> Text -> IO SynapseEnv-initEnv host port backend = do+-- | Initialize environment with given host/port/backend/logger/token+initEnv :: Text -> Int -> Text -> Log.Logger -> Maybe Text -> IO SynapseEnv+initEnv host port backend logger token = do cache <- newIORef HM.empty visited <- newIORef HS.empty pure SynapseEnv- { seHost = host- , sePort = port+ { seHost = host+ , sePort = port , seBackend = backend- , seCache = cache+ , seCache = cache , seVisited = visited+ , seLogger = logger+ , seToken = token } --- | Default environment (localhost:4444, requires backend)-defaultEnv :: Text -> IO SynapseEnv-defaultEnv = initEnv "127.0.0.1" 4444+-- | Default environment (localhost:4444, requires backend and logger)+defaultEnv :: Text -> Log.Logger -> IO SynapseEnv+defaultEnv backend logger = initEnv "127.0.0.1" 4444 backend logger Nothing -- | Throw a navigation error throwNav :: NavError -> SynapseM a@@ -206,3 +217,11 @@ insertCache hash schema = do cacheRef <- asks seCache liftIO $ modifyIORef' cacheRef (HM.insert hash schema)++-- ============================================================================+-- Logging and Config+-- ============================================================================++-- | Get the logger from the environment+getLogger :: SynapseM Log.Logger+getLogger = asks seLogger
src/Synapse/Self/Commands.hs view
@@ -20,11 +20,51 @@ import Synapse.Monad import Synapse.Backend.Discovery (getBackendAt) import qualified Synapse.Self.Template as Template+import qualified Synapse.Self.Debugger as Debugger+import Plexus.Client (SubstrateConfig(..)) -- | Dispatch to a specific _self subcommand dispatch :: Text -> [Text] -> [(Text, Text)] -> SynapseM () dispatch "template" rest params = Template.handleTemplate rest params dispatch "scan" rest _ = scanPorts rest+dispatch "debug" (hostText:portText:backendText:_) _ = do+ let host = T.unpack hostText+ let port = read (T.unpack portText) :: Int+ liftIO $ Debugger.debugConnection host port backendText+dispatch "debug" _ _ =+ throwParse "Usage: synapse _self debug <host> <port> <backend>"+dispatch "validate" (hostText:portText:backendText:_) _ = do+ let host = T.unpack hostText+ let port = read (T.unpack portText) :: Int+ liftIO $ Debugger.validateProtocol host port backendText+dispatch "validate" _ _ =+ throwParse "Usage: synapse _self validate <host> <port> <backend>"+dispatch "test" rest params = do+ -- Extract flags from params+ let allowUnknown = lookup "allow_unknown" params == Just ""+ let rawJson = lookup "raw" params++ -- Remove flags from params (keep only method params)+ let methodParams = filter (\(k, _) -> k /= "allow_unknown" && k /= "raw") params++ -- Get connection settings from environment (set by -H/-P flags)+ host <- asks seHost+ port <- asks sePort+ backend <- asks seBackend++ case rest of+ methodParts | not (null methodParts) -> do+ let method = T.intercalate "." methodParts+ let config = SubstrateConfig+ { substrateHost = T.unpack host+ , substratePort = port+ , substratePath = "/"+ , substrateBackend = backend+ , substrateHeaders = []+ }+ liftIO $ Debugger.testMethod config method methodParams allowUnknown rawJson+ _ ->+ throwParse "Usage: synapse [options] _self test [--allow-unknown] [--raw <json>] <method> [--param value ...]\n Note: Use -H/--host and -P/--port flags to set connection (default: 127.0.0.1:4444)" dispatch "--help" _ _ = showHelp dispatch "-h" _ _ = showHelp dispatch cmd _ _ =@@ -72,6 +112,24 @@ , " synapse _self scan" , " Scan ports 4440-4459 for backends (calls _info on each)" , ""+ , " synapse _self debug <host> <port> <backend>"+ , " Debug WebSocket connection and protocol issues"+ , " Tests: TCP, HTTP, WebSocket handshake, Plexus RPC"+ , ""+ , " synapse _self validate <host> <port> <backend>"+ , " Run protocol compliance validation tests"+ , " Tests: protocol_test, stream_test, error_test, metadata_test"+ , " Exits with code 0 on success, 1 on failures"+ , ""+ , " synapse [-H <host>] [-P <port>] _self test <method> [--param value ...]"+ , " Test arbitrary method with protocol validation"+ , " Connection: Uses -H/--host and -P/--port flags (defaults: 127.0.0.1:4444)"+ , " Modes:"+ , " (default) - Validate params against schema, reject unknown"+ , " --allow-unknown - Warn about unknown params but pass through"+ , " --raw <json> - Use raw JSON params, skip schema validation"+ , " Validates: StreamDone sent, metadata structure, field naming"+ , "" , " synapse _self template" , " Manage Mustache templates (CRUD operations)" , ""@@ -87,6 +145,10 @@ , "" , "Examples:" , " synapse _self scan"+ , " synapse _self debug 127.0.0.1 4444 substrate"+ , " synapse _self validate localhost 5001 substrate"+ , " synapse _self test localhost 5001 substrate echo.echo --message \"hello\""+ , " synapse _self test --allow-unknown localhost 5001 substrate echo.echo --fake \"test\"" , " synapse _self template list" , " synapse _self template show cone.chat" , " synapse _self template generate 'plexus.cone.*'"
+ src/Synapse/Self/Debugger.hs view
@@ -0,0 +1,390 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Protocol-level debugging tools for synapse+--+-- Use: synapse _self debug <host> <port> <backend>+--+-- This module provides tools to debug WebSocket connection issues,+-- protocol mismatches, and transport-level problems.+module Synapse.Self.Debugger+ ( debugConnection+ , validateProtocol+ , testMethod+ ) where++import Control.Exception (SomeException, catch, try)+import Control.Monad (unless, forM_)+import Control.Monad.IO.Class (liftIO)+import qualified Data.Aeson as Aeson+import Data.Aeson (Value(..), object, eitherDecode)+import Data.Aeson.Encode.Pretty (encodePretty)+import qualified Data.Aeson.Key as K+import qualified Data.Aeson.KeyMap as KM+import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy as LBS+import qualified Data.ByteString.Lazy.Char8 as LBS8+import Data.List (partition)+import qualified Data.Map.Strict as Map+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.IO as TIO+import qualified Data.Text.Encoding as TE+import qualified Network.Socket as NS+import qualified Network.Socket.ByteString as NSB+import qualified Network.WebSockets as WS+import Plexus.Client (SubstrateConfig(..))+import System.Exit (exitFailure, exitSuccess)+import System.IO (hFlush, stdout, stderr, hPutStrLn)+import System.Timeout (timeout)++import Synapse.CLI.Parse (parseParams)+import Synapse.IR.Builder (buildIR)+import Synapse.IR.Types (irMethods, MethodDef, mdParams, pdName)+import Synapse.Log (Logger, makeLogger)+import qualified Katip+import Synapse.Monad (SynapseEnv(..), initEnv, runSynapseM)+import qualified Synapse.Self.Protocol.TestRunner as TestRunner++-- | Debug connection to a Plexus backend+--+-- Tests:+-- 1. TCP connection+-- 2. HTTP upgrade request+-- 3. WebSocket handshake+-- 4. Protocol version compatibility+debugConnection :: String -> Int -> Text -> IO ()+debugConnection host port backend = do+ putStrLn $ "Debugging connection to " <> host <> ":" <> show port+ putStrLn $ "Backend: " <> T.unpack backend+ putStrLn ""++ -- Step 1: TCP connection+ putStr "[1/4] Testing TCP connection... "+ hFlush stdout+ tcpResult <- timeout 3000000 $ testTCPConnection host port+ case tcpResult of+ Nothing -> do+ putStrLn "TIMEOUT (3s)"+ putStrLn " TCP connection timed out. Server may be down or unreachable."+ return ()+ Just (Left err) -> do+ putStrLn $ "FAILED: " <> show err+ putStrLn " Cannot establish TCP connection. Check if server is running."+ return ()+ Just (Right ()) -> do+ putStrLn "OK"++ -- Step 2: HTTP check+ putStr "[2/4] Testing HTTP endpoint... "+ hFlush stdout+ httpResult <- timeout 3000000 $ testHTTPEndpoint host port+ case httpResult of+ Nothing -> putStrLn "TIMEOUT (3s)"+ Just (Left err) -> putStrLn $ "FAILED: " <> show err+ Just (Right response) -> do+ putStrLn "OK"+ putStrLn $ " " <> T.unpack response++ -- Step 3: WebSocket handshake+ putStr "[3/4] Testing WebSocket upgrade... "+ hFlush stdout+ wsResult <- timeout 5000000 $ testWebSocketHandshake host port+ case wsResult of+ Nothing -> do+ putStrLn "TIMEOUT (5s)"+ putStrLn ""+ putStrLn "ROOT CAUSE: WebSocket handshake hangs"+ putStrLn " The server accepts TCP connections but doesn't complete"+ putStrLn " the WebSocket upgrade. This indicates:"+ putStrLn " - Server is not a WebSocket server on this port"+ putStrLn " - Protocol version mismatch"+ putStrLn " - Server is waiting for different handshake format"+ Just (Left err) -> do+ putStrLn $ "FAILED: " <> show err+ putStrLn ""+ putStrLn "WebSocket handshake rejected by server"+ Just (Right ()) -> do+ putStrLn "OK"++ -- Step 4: Protocol test+ putStr "[4/4] Testing Plexus RPC protocol... "+ hFlush stdout+ rpcResult <- timeout 5000000 $ testPlexusRPC host port backend+ case rpcResult of+ Nothing -> do+ putStrLn "TIMEOUT (5s)"+ putStrLn ""+ putStrLn "ROOT CAUSE: Plexus RPC protocol timeout"+ putStrLn " The server accepts WebSocket connections but:"+ putStrLn " - Does not send required StreamDone messages"+ putStrLn " - Does not respond to _info requests correctly"+ putStrLn " - May not be a valid Plexus RPC server"+ Just (Left err) -> putStrLn $ "FAILED: " <> show err+ Just (Right ()) -> putStrLn "OK"++-- | Test TCP connection+testTCPConnection :: String -> Int -> IO (Either SomeException ())+testTCPConnection host port = try $ do+ addr <- NS.getAddrInfo+ (Just NS.defaultHints { NS.addrSocketType = NS.Stream })+ (Just host)+ (Just $ show port)+ case addr of+ [] -> error "No address info"+ (serverAddr:_) -> do+ sock <- NS.socket (NS.addrFamily serverAddr) NS.Stream NS.defaultProtocol+ NS.connect sock (NS.addrAddress serverAddr)+ NS.close sock++-- | Test HTTP endpoint+testHTTPEndpoint :: String -> Int -> IO (Either SomeException Text)+testHTTPEndpoint host port = try $ do+ addr <- NS.getAddrInfo+ (Just NS.defaultHints { NS.addrSocketType = NS.Stream })+ (Just host)+ (Just $ show port)+ case addr of+ [] -> error "No address info"+ (serverAddr:_) -> do+ sock <- NS.socket (NS.addrFamily serverAddr) NS.Stream NS.defaultProtocol+ NS.connect sock (NS.addrAddress serverAddr)+ let request = "GET / HTTP/1.1\r\nHost: " <> BS.pack (host <> ":" <> show port) <> "\r\n\r\n"+ NSB.sendAll sock request+ response <- NSB.recv sock 1024+ NS.close sock+ return $ T.pack $ BS.unpack $ BS.take 100 response++-- | Test WebSocket handshake+testWebSocketHandshake :: String -> Int -> IO (Either SomeException ())+testWebSocketHandshake host port = try $+ WS.runClient host port "/" $ \_conn -> return ()++-- | Test Plexus RPC protocol+testPlexusRPC :: String -> Int -> Text -> IO (Either SomeException ())+testPlexusRPC host port backend = try $+ WS.runClient host port "/" $ \conn -> do+ -- Send _info request+ let request = "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"" <> T.unpack backend <> "._info\",\"params\":{}}"+ WS.sendTextData conn (T.pack request :: Text)++ -- Wait for response (simplified - real implementation would parse)+ _response <- WS.receiveData conn :: IO Text+ return ()++-- | Validate protocol compliance of a backend+--+-- Runs the full protocol validation test suite against the specified backend+-- and reports any violations found.+--+-- Use: synapse _self validate <host> <port> <backend>+--+-- This connects to the backend and runs various protocol validation tests:+-- - Basic protocol compliance (_debug.protocol_test)+-- - Stream handling with slow data (_debug.stream_test)+-- - Progress reporting (_debug.stream_test with progress)+-- - Error handling (_debug.error_test)+-- - Metadata edge cases (_debug.metadata_test)+--+-- Exits with code 0 on success, 1 on protocol violations or errors.+validateProtocol :: String -> Int -> Text -> IO ()+validateProtocol host port backend = do+ TIO.putStrLn $ "Running protocol validation tests against " <> T.pack host <> ":" <> T.pack (show port) <> "..."+ TIO.putStrLn $ "Backend: " <> backend+ TIO.putStrLn ""++ -- Run the protocol test suite+ result <- TestRunner.runProtocolTests (T.pack host) port backend++ case result of+ Left errorReport -> do+ -- Protocol violations detected or test errors+ TIO.putStrLn errorReport+ TIO.putStrLn ""+ exitFailure++ Right successReport -> do+ -- All tests passed+ TIO.putStrLn successReport+ TIO.putStrLn ""+ exitSuccess++-- | Test an arbitrary method with protocol validation+--+-- Three modes:+-- - Mode 1 (default): Validated - uses parseParams, rejects unknown params+-- - Mode 2 (--allow-unknown): Warns about unknown params but passes them through+-- - Mode 3 (--raw <json>): Parses params as raw JSON, skips schema validation+--+-- Use: synapse _self test [--allow-unknown] [--raw <json>] <host> <port> <backend> <method> [--param value ...]+--+-- Examples:+-- synapse _self test localhost 5001 substrate echo.echo --message "hello"+-- synapse _self test --allow-unknown localhost 5001 substrate echo.echo --message "hello" --fake "test"+-- synapse _self test --raw '{"message":"hello"}' localhost 5001 substrate echo.echo+testMethod :: SubstrateConfig -> Text -> [(Text, Text)] -> Bool -> Maybe Text -> IO ()+testMethod config method cliParams allowUnknown rawJson = do+ let backend = substrateBackend config+ let host = substrateHost config+ let port = substratePort config++ -- Mode 3: Raw JSON+ case rawJson of+ Just jsonStr -> do+ case eitherDecode (LBS.fromStrict $ TE.encodeUtf8 jsonStr) of+ Left err -> do+ TIO.putStrLn $ "Error: Invalid JSON: " <> T.pack err+ exitFailure+ Right params -> runTest config method params++ -- Mode 1 & 2: Fetch schema and parse params+ Nothing -> do+ putStr "Fetching schema... "+ hFlush stdout++ -- Build IR for the method's namespace+ let methodParts = T.splitOn "." method+ let namespacePath = if length methodParts > 1 then init methodParts else []++ -- Create a SynapseM environment for buildIR+ logger <- makeLogger Katip.ErrorS -- Use error-level logger for test command+ env <- initEnv (T.pack host) port backend logger Nothing++ -- Run buildIR within SynapseM+ buildResult <- runSynapseM env (buildIR [] namespacePath)++ case buildResult of+ Left err -> do+ putStrLn "FAILED"+ TIO.putStrLn ""+ TIO.putStrLn "[SCHEMA ERROR] Failed to build IR from schema"+ TIO.putStrLn $ "Error: " <> T.pack (show err)+ TIO.putStrLn ""+ TIO.putStrLn "This usually means:"+ TIO.putStrLn " 1. Method schema is incompatible with synapse client"+ TIO.putStrLn " 2. Schema format differs from expected structure"+ TIO.putStrLn " 3. Method was generated by incompatible macro (hub_methods vs plexus-derive)"+ TIO.putStrLn ""+ TIO.putStrLn "Try testing with raw WebSocket to verify method works:"+ TIO.putStrLn $ " echo '{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"" <> method <> "\",\"params\":{...}}' | websocat ws://" <> T.pack host <> ":" <> T.pack (show port)+ TIO.putStrLn ""+ TIO.putStrLn "Or use the websocket-raw-test:"+ TIO.putStrLn " cabal test websocket-raw-test"+ exitFailure+ Right ir -> do+ putStrLn "OK"++ -- Look up method definition+ case Map.lookup method (irMethods ir) of+ Nothing -> do+ TIO.putStrLn $ "Error: Method not found in schema: " <> method+ exitFailure+ Just methodDef -> do+ -- Mode 1: Validated (strict)+ if not allowUnknown+ then do+ case parseParams ir methodDef cliParams of+ Left errors -> do+ TIO.putStrLn "Parameter validation failed:"+ mapM_ (TIO.putStrLn . T.pack . show) errors+ exitFailure+ Right params ->+ runTest config method params++ -- Mode 2: Allow unknown params+ else do+ let (knownParams, unknownParams) = partitionParams methodDef cliParams++ unless (null unknownParams) $+ TIO.putStrLn $ "⚠️ Unknown parameters: " <> T.intercalate ", " (map fst unknownParams) <> " (passing through anyway)"++ -- Parse known params with IR validation+ knownParsed <- case parseParams ir methodDef knownParams of+ Left errors -> do+ TIO.putStrLn "Error parsing known parameters:"+ mapM_ (TIO.putStrLn . T.pack . show) errors+ exitFailure+ Right p -> pure p++ -- Parse unknown params with best-effort type inference+ let unknownParsed = parseGeneric unknownParams++ -- Merge the two+ let params = mergeJson knownParsed unknownParsed+ runTest config method params++-- | Run the actual test and report results+runTest :: SubstrateConfig -> Text -> Value -> IO ()+runTest config method params = do+ TIO.putStrLn $ "Testing: " <> method+ result <- TestRunner.testEndpoint config method params++ case result of+ Left err -> do+ TIO.putStrLn "[FAIL] Test FAILED"+ TIO.putStrLn err+ exitFailure++ Right (violations, msgCount, rawMessages) -> do+ if null violations+ then do+ TIO.putStrLn $ "[PASS] Protocol VALID (" <> T.pack (show msgCount) <> " messages)"+ TIO.putStrLn " - StreamDone sent"+ TIO.putStrLn " - Metadata structure correct"+ TIO.putStrLn " - Field naming valid"+ exitSuccess+ else do+ TIO.putStrLn $ "[FAIL] Protocol violations (" <> T.pack (show (length violations)) <> ")"+ TIO.putStrLn ""+ TIO.putStrLn "Raw messages received:"+ forM_ (zip [1..] rawMessages) $ \(idx, msg) -> do+ TIO.putStrLn $ " Message " <> T.pack (show idx) <> ":"+ TIO.putStrLn $ " " <> (TE.decodeUtf8 $ LBS.toStrict $ encodePretty msg)+ TIO.putStrLn ""+ TIO.putStrLn "Violations detected:"+ mapM_ (TIO.putStrLn . (" " <>) . T.pack . show) violations+ exitFailure++-- | Partition params into known (in method def) and unknown+-- Handles dotted parameters like "identifier.type" by extracting the first segment+partitionParams :: MethodDef -> [(Text, Text)] -> ([(Text, Text)], [(Text, Text)])+partitionParams methodDef params =+ let paramNames = map pdName (mdParams methodDef)+ isKnown (key, _) =+ let baseKey = case T.breakOn "." key of+ (base, "") -> base -- No dot, use the whole key+ (base, _) -> base -- Has dot, use first segment+ in baseKey `elem` paramNames+ in partition isKnown params++-- | Parse unknown params with best-effort type inference+parseGeneric :: [(Text, Text)] -> Value+parseGeneric pairs = object+ [ (K.fromText k, inferValue v) | (k, v) <- pairs ]+ where+ inferValue :: Text -> Value+ inferValue t+ | t == "true" = Bool True+ | t == "false" = Bool False+ | t == "" = Bool True -- Flag with no value+ | Just n <- readMaybeInt t = Number (fromIntegral n)+ | Just n <- readMaybeDouble t = Number (realToFrac n)+ | otherwise = String t++ readMaybeInt :: Text -> Maybe Integer+ readMaybeInt = readMaybe . T.unpack++ readMaybeDouble :: Text -> Maybe Double+ readMaybeDouble = readMaybe . T.unpack++-- | Merge two JSON objects+mergeJson :: Value -> Value -> Value+mergeJson (Object a) (Object b) = Object (KM.union a b)+mergeJson (Object a) _ = Object a+mergeJson _ (Object b) = Object b+mergeJson a _ = a++-- | Read helper+readMaybe :: Read a => String -> Maybe a+readMaybe s = case reads s of+ [(x, "")] -> Just x+ _ -> Nothing
+ src/Synapse/Self/Protocol/Reporter.hs view
@@ -0,0 +1,322 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Synapse.Self.Protocol.Reporter+ ( -- * Main rendering functions+ renderViolations+ , renderSummary+ , renderCompact+ , renderJSON++ -- * Helper functions+ , groupBySeverity+ , formatLocation+ , severitySymbol+ , severityName+ ) where++import Data.Aeson (Value(..), object, (.=))+import qualified Data.Aeson as Aeson+import qualified Data.Aeson.Encoding as E+import qualified Data.ByteString.Lazy as BL+import Data.List (sortOn)+import Data.Ord (Down(..))+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import qualified Data.Text.Lazy as TL+import qualified Data.Text.Lazy.Builder as TB+import Synapse.Self.Protocol.Validator+ ( ProtocolViolation(..)+ , Severity(..)+ , Location(..)+ )++-- ============================================================================+-- Main Rendering Functions+-- ============================================================================++-- | Render violations in compiler-style format+--+-- Example output:+-- > error: Missing StreamDone message+-- > --> stream:subscription_42+-- > |+-- > | Stream received 5 data messages but never received StreamDone+-- > | Expected: StreamDone message to complete stream+-- > | Fix: Ensure server sends StreamDone after all data+renderViolations :: [ProtocolViolation] -> Text+renderViolations [] = "No protocol violations found.\n"+renderViolations violations =+ let grouped = groupBySeverity violations+ builder = TB.fromText "Protocol Violations:\n"+ <> TB.fromText "\n"+ <> renderGroup "Errors" Error grouped+ <> renderGroup "Warnings" Warning grouped+ <> renderGroup "Info" Info grouped+ in TL.toStrict $ TB.toLazyText builder++-- | Render a group of violations by severity+renderGroup :: Text -> Severity -> [(Severity, [ProtocolViolation])] -> TB.Builder+renderGroup label sev groups =+ case lookup sev groups of+ Nothing -> mempty+ Just [] -> mempty+ Just vs ->+ TB.fromText label+ <> TB.fromText ":\n"+ <> TB.fromText "----------------------------------------\n\n"+ <> mconcat (map renderSingleViolation vs)+ <> TB.fromText "\n"++-- | Render a single violation in compiler style+renderSingleViolation :: ProtocolViolation -> TB.Builder+renderSingleViolation ProtocolViolation{..} =+ let sevName = T.toLower $ severityName pvSeverity+ locText = formatLocation pvLocation++ -- Build the main error line+ mainLine = severitySymbol pvSeverity+ <> " "+ <> sevName+ <> ": "+ <> pvMessage+ <> "\n"++ -- Build the location line+ locLine = if T.null locText+ then mempty+ else TB.fromText " --> " <> TB.fromText locText <> TB.fromText "\n"++ -- Build the separator+ separator = if T.null locText+ then mempty+ else TB.fromText " |\n"++ -- Build the details section+ details = buildDetails pvExpected pvActual pvFix++ in TB.fromText mainLine+ <> locLine+ <> separator+ <> details+ <> TB.fromText "\n"++-- | Build the details section with expected/actual/fix+buildDetails :: Maybe Text -> Maybe Text -> Maybe Text -> TB.Builder+buildDetails expected actual fix =+ let expectedLine = case expected of+ Nothing -> mempty+ Just exp -> TB.fromText " | Expected: " <> TB.fromText exp <> TB.fromText "\n"++ actualLine = case actual of+ Nothing -> mempty+ Just act -> TB.fromText " | Actual: " <> TB.fromText act <> TB.fromText "\n"++ fixLine = case fix of+ Nothing -> mempty+ Just f -> TB.fromText " | Fix: " <> TB.fromText f <> TB.fromText "\n"++ in expectedLine <> actualLine <> fixLine++-- | Render summary of violations+--+-- Example output:+-- > Protocol Validation Results:+-- > ✗ 2 errors+-- > ⚠ 1 warning+-- > ℹ 0 info+-- >+-- > FAILED: Protocol violations detected+renderSummary :: [ProtocolViolation] -> Text+renderSummary violations =+ let grouped = groupBySeverity violations+ errorCount = maybe 0 length $ lookup Error grouped+ warningCount = maybe 0 length $ lookup Warning grouped+ infoCount = maybe 0 length $ lookup Info grouped++ status = if errorCount > 0+ then "FAILED: Protocol violations detected"+ else if warningCount > 0+ then "PASSED with warnings"+ else "PASSED: No protocol violations"++ builder = TB.fromText "Protocol Validation Results:\n"+ <> TB.fromText " " <> TB.fromText (severitySymbol Error) <> TB.fromText " "+ <> TB.fromString (show errorCount) <> TB.fromText " errors\n"+ <> TB.fromText " " <> TB.fromText (severitySymbol Warning) <> TB.fromText " "+ <> TB.fromString (show warningCount) <> TB.fromText " warnings\n"+ <> TB.fromText " " <> TB.fromText (severitySymbol Info) <> TB.fromText " "+ <> TB.fromString (show infoCount) <> TB.fromText " info\n"+ <> TB.fromText "\n"+ <> TB.fromText status+ <> TB.fromText "\n"++ in TL.toStrict $ TB.toLazyText builder++-- | Render violations in compact one-line format+--+-- Example output:+-- > [ERROR] Missing StreamDone message at stream:subscription_42: Stream received 5 data messages+renderCompact :: [ProtocolViolation] -> Text+renderCompact [] = "No protocol violations found.\n"+renderCompact violations =+ let sorted = sortOn (Down . pvSeverity) violations+ lines = map renderCompactLine sorted+ in T.intercalate "\n" lines <> "\n"++-- | Render a single violation in compact format+renderCompactLine :: ProtocolViolation -> Text+renderCompactLine ProtocolViolation{..} =+ let sevName = T.toUpper $ severityName pvSeverity+ locText = formatLocation pvLocation++ -- Build the compact line+ line = "["+ <> sevName+ <> "] "+ <> pvMessage++ -- Add location if present+ withLoc = if T.null locText+ then line+ else line <> " at " <> locText++ -- Add details if present+ withDetails = case (pvExpected, pvActual) of+ (Just exp, Just act) -> withLoc <> ": expected " <> exp <> ", got " <> act+ (Just exp, Nothing) -> withLoc <> ": expected " <> exp+ (Nothing, Just act) -> withLoc <> ": got " <> act+ (Nothing, Nothing) -> withLoc++ in withDetails++-- | Render violations in JSON format for machine consumption+--+-- Example output:+-- > {+-- > "violations": [+-- > {+-- > "message": "Missing StreamDone message",+-- > "severity": "error",+-- > "location": {"type": "stream", "method": "echo", "message_count": 5},+-- > "expected": "StreamDone message to complete stream",+-- > "actual": null,+-- > "fix": "Ensure server sends StreamDone after all data"+-- > }+-- > ],+-- > "summary": {+-- > "errors": 1,+-- > "warnings": 0,+-- > "info": 0,+-- > "passed": false+-- > }+-- > }+renderJSON :: [ProtocolViolation] -> Text+renderJSON violations =+ let grouped = groupBySeverity violations+ errorCount = maybe 0 length $ lookup Error grouped+ warningCount = maybe 0 length $ lookup Warning grouped+ infoCount = maybe 0 length $ lookup Info grouped++ violationsArray = map violationToJSON violations++ summaryObj = object+ [ "errors" .= errorCount+ , "warnings" .= warningCount+ , "info" .= infoCount+ , "passed" .= (errorCount == 0)+ ]++ rootObj = object+ [ "violations" .= violationsArray+ , "summary" .= summaryObj+ ]++ in TE.decodeUtf8 $ BL.toStrict $ Aeson.encode rootObj++-- | Convert a violation to JSON value+violationToJSON :: ProtocolViolation -> Value+violationToJSON ProtocolViolation{..} =+ object+ [ "message" .= pvMessage+ , "severity" .= severityToJSON pvSeverity+ , "location" .= locationToJSON pvLocation+ , "expected" .= pvExpected+ , "actual" .= pvActual+ , "fix" .= pvFix+ ]++-- | Convert severity to JSON string+severityToJSON :: Severity -> Text+severityToJSON Error = "error"+severityToJSON Warning = "warning"+severityToJSON Info = "info"++-- | Convert location to JSON object+locationToJSON :: Location -> Value+locationToJSON (InMessage subId msgIdx field) =+ object+ [ "type" .= ("message" :: Text)+ , "subscription_id" .= subId+ , "message_index" .= msgIdx+ , "field" .= field+ ]+locationToJSON (InStream method msgCount) =+ object+ [ "type" .= ("stream" :: Text)+ , "method" .= method+ , "message_count" .= msgCount+ ]+locationToJSON InConnection =+ object+ [ "type" .= ("connection" :: Text)+ ]++-- ============================================================================+-- Helper Functions+-- ============================================================================++-- | Group violations by severity (highest severity first)+groupBySeverity :: [ProtocolViolation] -> [(Severity, [ProtocolViolation])]+groupBySeverity violations =+ let errors = filter (\v -> pvSeverity v == Error) violations+ warnings = filter (\v -> pvSeverity v == Warning) violations+ infos = filter (\v -> pvSeverity v == Info) violations+ in [ (Error, errors)+ , (Warning, warnings)+ , (Info, infos)+ ]++-- | Format location for display+formatLocation :: Location -> Text+formatLocation (InMessage subId msgIdx field) =+ let subPart = case subId of+ Nothing -> "message"+ Just sid -> "subscription_" <> T.pack (show sid)++ msgPart = "msg_" <> T.pack (show msgIdx)++ fieldPart = case field of+ Nothing -> ""+ Just f -> "." <> f++ in subPart <> ":" <> msgPart <> fieldPart++formatLocation (InStream method msgCount) =+ "stream:" <> method <> " (after " <> T.pack (show msgCount) <> " messages)"++formatLocation InConnection =+ "connection"++-- | Get symbol for severity (for visual distinction)+severitySymbol :: Severity -> Text+severitySymbol Error = "X"+severitySymbol Warning = "!"+severitySymbol Info = "*"++-- | Get name for severity+severityName :: Severity -> Text+severityName Error = "Error"+severityName Warning = "Warning"+severityName Info = "Info"
+ src/Synapse/Self/Protocol/StreamTracker.hs view
@@ -0,0 +1,272 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Synapse.Self.Protocol.StreamTracker+ ( -- * Stream Tracker+ StreamTracker+ , StreamState(..)++ -- * Operations+ , newTracker+ , trackMessage+ , checkCompletion+ , getAllViolations++ -- * For testing+ , getStreamCount+ ) where++import Control.Monad (forM)+import Data.Aeson (Value(..), Object)+import qualified Data.Aeson.KeyMap as KM+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HM+import Data.IORef (IORef, newIORef, readIORef, modifyIORef')+import Data.Text (Text)+import qualified Data.Text as T+import Data.Maybe (fromMaybe)++import Synapse.Self.Protocol.Validator+ ( ProtocolViolation(..)+ , Severity(..)+ , Location(..)+ )++-- ============================================================================+-- Types+-- ============================================================================++-- | State for a single stream subscription+data StreamState = StreamState+ { ssSubscriptionId :: !Int+ -- ^ The subscription ID for this stream+ , ssMessages :: ![Value]+ -- ^ All messages received for this stream+ , ssHasDone :: !Bool+ -- ^ Whether StreamDone was received+ , ssViolations :: ![ProtocolViolation]+ -- ^ Violations detected for this stream+ , ssMethod :: !(Maybe Text)+ -- ^ The method name that created this subscription (if known)+ }+ deriving (Show, Eq)++-- | Stream tracker for protocol validation+-- Tracks multiple concurrent streams and detects protocol violations+data StreamTracker = StreamTracker+ { stStreams :: !(IORef (HashMap Int StreamState))+ -- ^ Map from subscription ID to stream state+ }++-- ============================================================================+-- Constructor+-- ============================================================================++-- | Create a new stream tracker+newTracker :: IO StreamTracker+newTracker = do+ ref <- newIORef HM.empty+ pure $ StreamTracker ref++-- ============================================================================+-- Message Tracking+-- ============================================================================++-- | Track a message and detect any protocol violations+--+-- This function processes incoming messages and:+-- 1. Extracts subscription information+-- 2. Updates stream state+-- 3. Detects violations (duplicate StreamDone, missing subscription ID, etc.)+trackMessage :: StreamTracker -> Value -> IO [ProtocolViolation]+trackMessage tracker msg = do+ case parseMessage msg of+ Nothing ->+ -- Not a stream notification, ignore+ pure []++ Just (subId, msgType, method) -> do+ modifyIORef' (stStreams tracker) $ \streams ->+ let currentState = HM.lookupDefault (emptyStreamState subId method) subId streams+ updatedState = updateStreamState currentState msgType msg+ in HM.insert subId updatedState streams++ -- Return violations found in this message+ streams <- readIORef (stStreams tracker)+ case HM.lookup subId streams of+ Nothing -> pure []+ Just state -> pure $ ssViolations state++-- | Parse a message to extract subscription info and message type+parseMessage :: Value -> Maybe (Int, MessageType, Maybe Text)+parseMessage (Object o) = do+ -- Check if this is a subscription notification+ method <- KM.lookup "method" o+ case method of+ String "subscription" -> do+ -- Extract params object+ params <- KM.lookup "params" o+ case params of+ Object paramsObj -> do+ -- Extract subscription ID+ subIdValue <- KM.lookup "subscription" paramsObj+ subId <- case subIdValue of+ Number n -> Just $ truncate n+ String s -> case reads (T.unpack s) of+ [(n, "")] -> Just n+ _ -> Nothing+ _ -> Nothing++ -- Extract result object to determine message type+ result <- KM.lookup "result" paramsObj+ msgType <- case result of+ Object resultObj -> do+ typeValue <- KM.lookup "type" resultObj+ case typeValue of+ String "done" -> Just StreamDoneMsg+ String "data" -> Just StreamDataMsg+ String "progress" -> Just StreamProgressMsg+ String "error" -> Just StreamErrorMsg+ String "guidance" -> Just StreamGuidanceMsg+ String "request" -> Just StreamRequestMsg+ _ -> Just UnknownMsg+ _ -> Just UnknownMsg++ pure (subId, msgType, Nothing)+ _ -> Nothing+ _ -> Nothing+parseMessage _ = Nothing++-- | Message types in the stream protocol+data MessageType+ = StreamDoneMsg+ | StreamDataMsg+ | StreamProgressMsg+ | StreamErrorMsg+ | StreamGuidanceMsg+ | StreamRequestMsg+ | UnknownMsg+ deriving (Show, Eq)++-- | Create an empty stream state+emptyStreamState :: Int -> Maybe Text -> StreamState+emptyStreamState subId method = StreamState+ { ssSubscriptionId = subId+ , ssMessages = []+ , ssHasDone = False+ , ssViolations = []+ , ssMethod = method+ }++-- | Update stream state with a new message+updateStreamState :: StreamState -> MessageType -> Value -> StreamState+updateStreamState state msgType msg =+ let newMessages = ssMessages state ++ [msg]+ newViolations = detectMessageViolations state msgType+ in case msgType of+ StreamDoneMsg ->+ if ssHasDone state+ then -- Duplicate StreamDone violation+ state+ { ssMessages = newMessages+ , ssViolations = ssViolations state ++ newViolations +++ [ ProtocolViolation+ { pvMessage = "Multiple StreamDone messages received for subscription " <> T.pack (show $ ssSubscriptionId state)+ , pvSeverity = Error+ , pvLocation = InMessage+ { lmSubscriptionId = Just (ssSubscriptionId state)+ , lmMessageIndex = length newMessages+ , lmField = Just "type"+ }+ , pvExpected = Just "Only one StreamDone per subscription"+ , pvActual = Just "Multiple StreamDone messages"+ , pvFix = Just "Server should only send StreamDone once per subscription"+ }+ ]+ }+ else -- First StreamDone, mark as complete+ state+ { ssMessages = newMessages+ , ssHasDone = True+ , ssViolations = ssViolations state ++ newViolations+ }++ _ -> -- Other message types+ if ssHasDone state+ then -- Received message after StreamDone+ state+ { ssMessages = newMessages+ , ssViolations = ssViolations state ++ newViolations +++ [ ProtocolViolation+ { pvMessage = "Received message after StreamDone for subscription " <> T.pack (show $ ssSubscriptionId state)+ , pvSeverity = Error+ , pvLocation = InMessage+ { lmSubscriptionId = Just (ssSubscriptionId state)+ , lmMessageIndex = length newMessages+ , lmField = Just "type"+ }+ , pvExpected = Just "No messages after StreamDone"+ , pvActual = Just $ "Received " <> T.pack (show msgType) <> " after StreamDone"+ , pvFix = Just "Server should not send messages after StreamDone"+ }+ ]+ }+ else+ state+ { ssMessages = newMessages+ , ssViolations = ssViolations state ++ newViolations+ }++-- | Detect violations in a specific message+detectMessageViolations :: StreamState -> MessageType -> [ProtocolViolation]+detectMessageViolations _state _msgType =+ -- For now, we don't detect per-message violations beyond what's in updateStreamState+ -- This could be extended to check message structure, metadata, etc.+ []++-- ============================================================================+-- Completion Checking+-- ============================================================================++-- | Check if all streams have completed (received StreamDone)+-- Returns violations for any incomplete streams+checkCompletion :: StreamTracker -> IO [ProtocolViolation]+checkCompletion tracker = do+ streams <- readIORef (stStreams tracker)+ let incompleteStreams = HM.filter (not . ssHasDone) streams+ pure $ map mkIncompleteViolation (HM.elems incompleteStreams)+ where+ mkIncompleteViolation :: StreamState -> ProtocolViolation+ mkIncompleteViolation state = ProtocolViolation+ { pvMessage = "Stream never received StreamDone for subscription " <> T.pack (show $ ssSubscriptionId state)+ , pvSeverity = Error+ , pvLocation = InStream+ { lsMethod = fromMaybe "unknown" (ssMethod state)+ , lsMessageCount = length (ssMessages state)+ }+ , pvExpected = Just "StreamDone message to complete the stream"+ , pvActual = Just $ T.pack (show $ length $ ssMessages state) <> " messages without StreamDone"+ , pvFix = Just "Server must send StreamDone to properly close each stream"+ }++-- ============================================================================+-- Violation Retrieval+-- ============================================================================++-- | Get all violations detected across all streams+getAllViolations :: StreamTracker -> IO [ProtocolViolation]+getAllViolations tracker = do+ streams <- readIORef (stStreams tracker)+ let allStreamViolations = concatMap ssViolations (HM.elems streams)+ completionViolations <- checkCompletion tracker+ pure $ allStreamViolations ++ completionViolations++-- ============================================================================+-- Testing Helpers+-- ============================================================================++-- | Get the number of tracked streams (for testing)+getStreamCount :: StreamTracker -> IO Int+getStreamCount tracker = do+ streams <- readIORef (stStreams tracker)+ pure $ HM.size streams
+ src/Synapse/Self/Protocol/TestRunner.hs view
@@ -0,0 +1,375 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Synapse.Self.Protocol.TestRunner+ ( -- * Main Test Runner+ runProtocolTests++ -- * Helper Functions+ , testEndpoint+ , TestResult(..)+ ) where++import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar, threadDelay)+import Control.Concurrent.MVar (MVar, newMVar, modifyMVar_)+import Control.Exception (SomeException, catch)+import Data.Aeson (Value(..), object, (.=))+import qualified Data.Aeson as Aeson+import Data.IORef (IORef, newIORef, readIORef, modifyIORef')+import Data.Text (Text)+import qualified Data.Text as T+import qualified Plexus.Transport as PT+import qualified Plexus.Types as PT+import Plexus.Client (SubstrateConfig(..))++import Synapse.Self.Protocol.StreamTracker+ ( StreamTracker+ , newTracker+ , trackMessage+ , getAllViolations+ )+import Synapse.Self.Protocol.Validator+ ( ProtocolViolation(..)+ , validateStreamItem+ )+import Synapse.Self.Protocol.Reporter+ ( renderViolations+ , renderSummary+ )++-- ============================================================================+-- Types+-- ============================================================================++-- | Result of a single test+data TestResult = TestResult+ { trTestName :: !Text+ , trViolations :: ![ProtocolViolation]+ , trMessageCount :: !Int+ , trError :: !(Maybe Text)+ }+ deriving (Show, Eq)++-- ============================================================================+-- Main Test Runner+-- ============================================================================++-- | Run protocol validation tests against a backend+--+-- Connects to the specified host/port/backend and runs a suite of+-- protocol validation tests using the debug endpoints:+-- - _debug.protocol_test (basic protocol compliance)+-- - _debug.stream_test with various scenarios (slow, progress)+-- - _debug.error_test (error handling)+-- - _debug.metadata_test (metadata edge cases)+--+-- Returns either an error message or a success report with detailed+-- violation information.+runProtocolTests :: Text -> Int -> Text -> IO (Either Text Text)+runProtocolTests host port backend = do+ let config = SubstrateConfig+ { substrateHost = T.unpack host+ , substratePort = port+ , substratePath = "/"+ , substrateBackend = backend+ , substrateHeaders = []+ }++ -- Run all test suites+ results <- runTestSuite config++ -- Aggregate all violations+ let allViolations = concatMap trViolations results+ let hasErrors = any (\r -> not (null (trViolations r))) results+ let hasTestErrors = any (\r -> case trError r of Just _ -> True; Nothing -> False) results++ -- Generate report+ if hasTestErrors+ then do+ -- Some tests failed to run+ let errorReport = T.intercalate "\n" $ map formatTestError results+ pure $ Left $ "Test execution errors:\n" <> errorReport+ else if hasErrors+ then do+ -- Protocol violations detected+ let report = renderViolations allViolations+ <> "\n"+ <> renderSummary allViolations+ <> "\n\nTest Results:\n"+ <> formatTestResults results+ pure $ Left report+ else do+ -- All tests passed+ let report = renderSummary allViolations+ <> "\n\nTest Results:\n"+ <> formatTestResults results+ pure $ Right report++-- | Format test error for display+formatTestError :: TestResult -> Text+formatTestError TestResult{..} =+ case trError of+ Nothing -> ""+ Just err -> " [" <> trTestName <> "] ERROR: " <> err++-- | Format test results summary+formatTestResults :: [TestResult] -> Text+formatTestResults results =+ T.intercalate "\n" $ map formatTestResult results++-- | Format a single test result+formatTestResult :: TestResult -> Text+formatTestResult TestResult{..} =+ let status = if null trViolations+ then "PASS"+ else "FAIL (" <> T.pack (show (length trViolations)) <> " violations)"+ in " [" <> trTestName <> "] " <> status <> " (" <> T.pack (show trMessageCount) <> " messages)"++-- ============================================================================+-- Test Suite+-- ============================================================================++-- | Run the full test suite+runTestSuite :: SubstrateConfig -> IO [TestResult]+runTestSuite config = do+ -- Test 1: Basic protocol compliance+ r1 <- testEndpointWithName config "Protocol Test" "_debug.protocol_test" Aeson.Null++ -- Test 2: Stream test with slow scenario+ r2 <- testEndpointWithName config "Stream Test (slow)" "_debug.stream_test"+ (object ["scenario" .= ("slow" :: Text)])++ -- Test 3: Stream test with progress scenario+ r3 <- testEndpointWithName config "Stream Test (progress)" "_debug.stream_test"+ (object ["scenario" .= ("progress" :: Text)])++ -- Test 4: Error test with recoverable error+ r4 <- testEndpointWithName config "Error Test (recoverable)" "_debug.error_test"+ (object ["error_type" .= ("recoverable" :: Text)])++ -- Test 5: Metadata edge cases+ r5 <- testEndpointWithName config "Metadata Test" "_debug.metadata_test" Aeson.Null++ pure [r1, r2, r3, r4, r5]++-- | Test an endpoint with a name+testEndpointWithName :: SubstrateConfig -> Text -> Text -> Value -> IO TestResult+testEndpointWithName config testName method params = do+ result <- testEndpoint config method params+ case result of+ Left err ->+ pure $ TestResult+ { trTestName = testName+ , trViolations = []+ , trMessageCount = 0+ , trError = Just err+ }+ Right (violations, msgCount, _rawMessages) ->+ pure $ TestResult+ { trTestName = testName+ , trViolations = violations+ , trMessageCount = msgCount+ , trError = Nothing+ }++-- ============================================================================+-- Endpoint Testing+-- ============================================================================++-- | Test a specific endpoint for protocol violations+--+-- Connects to the endpoint, subscribes to the stream, collects all messages,+-- validates each message, and returns all violations found.+--+-- Returns either an error message (connection/timeout issues) or a tuple of:+-- - List of protocol violations+-- - Message count+-- - Raw messages (for debugging failures)+testEndpoint :: SubstrateConfig -> Text -> Value -> IO (Either Text ([ProtocolViolation], Int, [Value]))+testEndpoint config method params = do+ -- Create tracker for this test+ tracker <- newTracker++ -- Counter for messages received+ msgCountRef <- newIORef 0++ -- Collect all violations+ violationsRef <- newIORef []++ -- Collect raw messages for debugging+ messagesRef <- newIORef []++ -- Result MVar+ resultMVar <- newEmptyMVar++ -- Build the RPC path: {backend}.call with method parameter+ let backend = substrateBackend config+ let callMethod = backend <> ".call"+ let callParams = object+ [ "method" .= method+ , "params" .= params+ ]++ -- Start the streaming call in a separate thread+ _ <- forkIO $ do+ result <- PT.rpcCallStreaming config callMethod callParams $ \item -> do+ -- Convert PlexusStreamItem to Value for validation+ let itemValue = plexusItemToValue item++ -- Store raw message for debugging+ modifyIORef' messagesRef (++ [itemValue])++ -- Track the message+ trackViolations <- trackMessage tracker itemValue+ modifyIORef' violationsRef (++ trackViolations)++ -- Validate the stream item structure+ structureViolations <- validateStreamItem itemValue+ modifyIORef' violationsRef (++ structureViolations)++ -- Increment message count+ modifyIORef' msgCountRef (+1)++ case result of+ Left transportErr ->+ putMVar resultMVar $ Left $ formatTransportError transportErr+ Right () ->+ -- Check for completion violations+ putMVar resultMVar $ Right ()++ -- Wait for completion with timeout (15 seconds)+ timeoutResult <- timeout 15000000 (takeMVar resultMVar)++ case timeoutResult of+ Nothing ->+ -- Timeout occurred+ pure $ Left "Timeout: Test did not complete within 15 seconds"++ Just (Left err) ->+ -- Transport error+ pure $ Left err++ Just (Right ()) -> do+ -- Get completion violations (streams that didn't receive StreamDone)+ completionViolations <- getAllViolations tracker++ -- Get all violations+ allViolations <- readIORef violationsRef+ let finalViolations = allViolations ++ completionViolations++ -- Get message count+ msgCount <- readIORef msgCountRef++ -- Get raw messages+ rawMessages <- readIORef messagesRef++ pure $ Right (finalViolations, msgCount, rawMessages)++-- ============================================================================+-- Helpers+-- ============================================================================++-- | Convert PlexusStreamItem to Value for validation+--+-- This extracts the JSON representation from the subscription notification+plexusItemToValue :: PT.PlexusStreamItem -> Value+plexusItemToValue (PT.StreamData hash prov ct content) =+ object+ [ "type" .= ("data" :: Text)+ , "content_type" .= ct+ , "content" .= content+ , "metadata" .= object+ [ "provenance" .= provenanceToValue prov+ , "plexus_hash" .= hash+ , "timestamp" .= (0 :: Int)+ ]+ ]+plexusItemToValue (PT.StreamProgress hash prov msg pct) =+ object+ [ "type" .= ("progress" :: Text)+ , "message" .= msg+ , "percentage" .= pct+ , "metadata" .= object+ [ "provenance" .= provenanceToValue prov+ , "plexus_hash" .= hash+ , "timestamp" .= (0 :: Int)+ ]+ ]+plexusItemToValue (PT.StreamError hash prov msg recoverable) =+ object+ [ "type" .= ("error" :: Text)+ , "message" .= msg+ , "code" .= if recoverable then Just ("recoverable" :: Text) else Nothing+ , "metadata" .= object+ [ "provenance" .= provenanceToValue prov+ , "plexus_hash" .= hash+ , "timestamp" .= (0 :: Int)+ ]+ ]+plexusItemToValue (PT.StreamDone hash prov) =+ object+ [ "type" .= ("done" :: Text)+ , "metadata" .= object+ [ "provenance" .= provenanceToValue prov+ , "plexus_hash" .= hash+ , "timestamp" .= (0 :: Int)+ ]+ ]+plexusItemToValue (PT.StreamRequest hash prov reqId reqData timeout) =+ object+ [ "type" .= ("request" :: Text)+ , "request_id" .= reqId+ , "request_data" .= reqData+ , "timeout" .= timeout+ , "metadata" .= object+ [ "provenance" .= provenanceToValue prov+ , "plexus_hash" .= hash+ , "timestamp" .= (0 :: Int)+ ]+ ]+plexusItemToValue (PT.StreamGuidance hash prov errType suggestion availMethods methodSchema) =+ object $+ [ "type" .= ("guidance" :: Text)+ , "metadata" .= object+ [ "provenance" .= provenanceToValue prov+ , "plexus_hash" .= hash+ , "timestamp" .= (0 :: Int)+ ]+ , "error_type" .= errType+ , "suggestion" .= suggestion+ ]+ ++ maybe [] (\m -> ["available_methods" .= m]) availMethods+ ++ maybe [] (\s -> ["method_schema" .= s]) methodSchema++-- | Convert Provenance to Value+provenanceToValue :: PT.Provenance -> Value+provenanceToValue (PT.Provenance servers) =+ Aeson.toJSON servers+++-- | Format transport error for display+formatTransportError :: PT.TransportError -> Text+formatTransportError (PT.ConnectionRefused h p) =+ "Connection refused to " <> h <> ":" <> T.pack (show p)+formatTransportError (PT.ConnectionTimeout h p) =+ "Connection timeout to " <> h <> ":" <> T.pack (show p)+formatTransportError (PT.ProtocolError msg) =+ "Protocol error: " <> msg+formatTransportError (PT.NetworkError msg) =+ "Network error: " <> msg++-- | Simple timeout implementation+-- Returns Nothing if timeout occurs, Just result otherwise+timeout :: Int -> IO a -> IO (Maybe a)+timeout microseconds action = do+ resultMVar <- newEmptyMVar+ -- Start the action+ _ <- forkIO $ do+ result <- action+ putMVar resultMVar (Just result)+ -- Start the timeout thread+ _ <- forkIO $ do+ threadDelay microseconds+ putMVar resultMVar Nothing+ -- Wait for either to complete+ takeMVar resultMVar
+ src/Synapse/Self/Protocol/Validator.hs view
@@ -0,0 +1,582 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Synapse.Self.Protocol.Validator+ ( -- * Types+ ProtocolViolation(..)+ , Severity(..)+ , Location(..)+ , ValidationContext(..)+ , ValidationConfig(..)++ -- * Validation functions+ , validateStreamItem+ , validateMetadata+ , validateFieldNames+ , validateDataItem+ , validateProgressItem+ , validateErrorItem+ , validateDoneItem+ ) where++import Data.Aeson (Value(..), Object)+import qualified Data.Aeson.Key as K+import qualified Data.Aeson.KeyMap as KM+import Data.Char (isDigit, isHexDigit, isLower)+import Data.Text (Text)+import qualified Data.Text as T+import Data.Scientific (isInteger, floatingOrInteger)+import Data.Vector (Vector)+import qualified Data.Vector as V++-- | Protocol violation with precise location and suggested fix+data ProtocolViolation = ProtocolViolation+ { pvMessage :: !Text+ , pvSeverity :: !Severity+ , pvLocation :: !Location+ , pvExpected :: !(Maybe Text)+ , pvActual :: !(Maybe Text)+ , pvFix :: !(Maybe Text) -- ^ Suggested fix+ }+ deriving (Show, Eq)++data Severity = Error | Warning | Info+ deriving (Show, Eq, Ord)++data Location+ = InMessage+ { lmSubscriptionId :: !(Maybe Int)+ , lmMessageIndex :: !Int+ , lmField :: !(Maybe Text)+ }+ | InStream+ { lsMethod :: !Text+ , lsMessageCount :: !Int+ }+ | InConnection+ deriving (Show, Eq)++-- | Validation context tracking stream state+data ValidationContext = ValidationContext+ { vcViolations :: ![ProtocolViolation]+ , vcConfig :: !ValidationConfig+ }+ deriving (Show, Eq)++data ValidationConfig = ValidationConfig+ { vcCheckMetadata :: !Bool+ , vcCheckStreamDone :: !Bool+ , vcCheckProvenance :: !Bool+ , vcCheckTypes :: !Bool+ , vcStrictMode :: !Bool -- ^ Fail on warnings+ }+ deriving (Show, Eq)++-- | Default validation config (check everything)+defaultValidationConfig :: ValidationConfig+defaultValidationConfig = ValidationConfig+ { vcCheckMetadata = True+ , vcCheckStreamDone = True+ , vcCheckProvenance = True+ , vcCheckTypes = True+ , vcStrictMode = False+ }++-- ============================================================================+-- Main Validation Function+-- ============================================================================++-- | Validate a stream item+--+-- Checks:+-- - Message structure (type field)+-- - Metadata structure (provenance, plexus_hash, timestamp)+-- - Field naming (snake_case for protocol fields)+-- - Type-specific validation (data, error, progress, done)+validateStreamItem :: Value -> IO [ProtocolViolation]+validateStreamItem item = pure $ case item of+ Object obj ->+ let typeViolations = validateType obj+ metadataViolations = validateMetadata item+ fieldNameViolations = validateFieldNames item+ typeSpecificViolations = validateByType obj+ in concat [typeViolations, metadataViolations, fieldNameViolations, typeSpecificViolations]+ _ ->+ [ ProtocolViolation+ { pvMessage = "Stream item must be a JSON object"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "JSON object with 'type' field"+ , pvActual = Just $ T.pack $ show item+ , pvFix = Just "Ensure stream items are JSON objects"+ }+ ]++-- ============================================================================+-- Type Field Validation+-- ============================================================================++-- | Validate that the type field exists and is valid+validateType :: Object -> [ProtocolViolation]+validateType obj = case KM.lookup "type" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'type' field in stream item"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "type field with value: data, error, progress, done, or request"+ , pvActual = Just "missing"+ , pvFix = Just "Add 'type' field to stream item"+ }+ ]+ Just (String typeStr) ->+ if typeStr `elem` ["data", "error", "progress", "done", "request"]+ then []+ else+ [ ProtocolViolation+ { pvMessage = "Invalid stream item type: " <> typeStr+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "data, error, progress, done, or request"+ , pvActual = Just typeStr+ , pvFix = Just "Use one of the valid stream item types"+ }+ ]+ Just other ->+ [ ProtocolViolation+ { pvMessage = "Stream item 'type' must be a string"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "string"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change type value to a string"+ }+ ]++-- ============================================================================+-- Metadata Validation+-- ============================================================================++-- | Validate stream metadata structure+--+-- Checks:+-- - Provenance is array of strings+-- - plexus_hash is 16-character hex string+-- - Timestamp is valid (integer, Unix seconds or ISO 8601)+validateMetadata :: Value -> [ProtocolViolation]+validateMetadata (Object obj) =+ -- Check if this item type should have metadata+ case KM.lookup "type" obj of+ Just (String "request") ->+ -- Request items don't require metadata+ []+ _ ->+ -- All other types require metadata+ case KM.lookup "metadata" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'metadata' field (required for data, error, progress, done items)"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "metadata object with provenance, plexus_hash, timestamp"+ , pvActual = Just "missing"+ , pvFix = Just "Add metadata field to stream item"+ }+ ]+ Just (Object metaObj) ->+ concat+ [ validateProvenance metaObj+ , validatePlexusHash metaObj+ , validateTimestamp metaObj+ ]+ Just other ->+ [ ProtocolViolation+ { pvMessage = "Metadata must be a JSON object"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "object"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change metadata to an object"+ }+ ]+validateMetadata _ = []++-- | Validate provenance field (must be non-empty array of strings)+validateProvenance :: Object -> [ProtocolViolation]+validateProvenance obj = case KM.lookup "provenance" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'provenance' field in metadata"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "non-empty array of strings"+ , pvActual = Just "missing"+ , pvFix = Just "Add provenance field with server names"+ }+ ]+ Just (Array arr) ->+ if V.null arr+ then+ [ ProtocolViolation+ { pvMessage = "Provenance array cannot be empty"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "non-empty array of strings"+ , pvActual = Just "empty array"+ , pvFix = Just "Add at least one server name to provenance"+ }+ ]+ else+ -- Check each element is a string+ concatMap (validateProvenanceElement arr) (zip [0..] (V.toList arr))+ Just other ->+ [ ProtocolViolation+ { pvMessage = "Provenance must be an array"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "array of strings"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change provenance to an array of server names"+ }+ ]++-- | Validate a single provenance element+validateProvenanceElement :: Vector Value -> (Int, Value) -> [ProtocolViolation]+validateProvenanceElement _arr (idx, String _) = []+validateProvenanceElement _arr (idx, other) =+ [ ProtocolViolation+ { pvMessage = "Provenance element at index " <> T.pack (show idx) <> " must be a string"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "string"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Ensure all provenance elements are strings"+ }+ ]++-- | Validate plexus_hash field (must be exactly 16 hex characters)+validatePlexusHash :: Object -> [ProtocolViolation]+validatePlexusHash obj = case KM.lookup "plexus_hash" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'plexus_hash' field in metadata"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "16-character lowercase hex string"+ , pvActual = Just "missing"+ , pvFix = Just "Add plexus_hash field with schema version hash"+ }+ ]+ Just (String hashStr) ->+ let len = T.length hashStr+ isAllHex = T.all (\c -> isHexDigit c && (isDigit c || isLower c)) hashStr+ in concat+ [ if len /= 16+ then+ [ ProtocolViolation+ { pvMessage = "plexus_hash must be exactly 16 characters (got " <> T.pack (show len) <> ")"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "16 characters"+ , pvActual = Just $ T.pack (show len) <> " characters: " <> hashStr+ , pvFix = Just "Use 16-character hex hash (SHA-256 truncated to 64 bits)"+ }+ ]+ else []+ , if not isAllHex+ then+ [ ProtocolViolation+ { pvMessage = "plexus_hash must be lowercase hexadecimal"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "lowercase hex characters (0-9, a-f)"+ , pvActual = Just hashStr+ , pvFix = Just "Use only lowercase hexadecimal characters"+ }+ ]+ else []+ ]+ Just other ->+ [ ProtocolViolation+ { pvMessage = "plexus_hash must be a string"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "string"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change plexus_hash to a string"+ }+ ]++-- | Validate timestamp field (must be positive integer)+validateTimestamp :: Object -> [ProtocolViolation]+validateTimestamp obj = case KM.lookup "timestamp" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'timestamp' field in metadata"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "positive integer (Unix seconds)"+ , pvActual = Just "missing"+ , pvFix = Just "Add timestamp field with Unix epoch seconds"+ }+ ]+ Just (Number n) ->+ case floatingOrInteger n of+ Left (_ :: Double) ->+ [ ProtocolViolation+ { pvMessage = "Timestamp must be an integer (not float)"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "integer (use Math.floor(Date.now() / 1000) in JavaScript)"+ , pvActual = Just $ T.pack $ show n+ , pvFix = Just "Convert timestamp to integer: Math.floor(Date.now() / 1000)"+ }+ ]+ Right (ts :: Integer) ->+ if ts <= 0+ then+ [ ProtocolViolation+ { pvMessage = "Timestamp must be positive"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "positive integer"+ , pvActual = Just $ T.pack $ show ts+ , pvFix = Just "Use positive Unix timestamp"+ }+ ]+ else []+ Just other ->+ [ ProtocolViolation+ { pvMessage = "Timestamp must be a number"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "integer"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change timestamp to an integer"+ }+ ]++-- ============================================================================+-- Field Naming Validation+-- ============================================================================++-- | Validate field naming conventions (snake_case for protocol fields)+validateFieldNames :: Value -> [ProtocolViolation]+validateFieldNames (Object obj) =+ let protocolFields = ["jsonrpc", "subscription_id", "method", "params", "result", "id"]+ camelCaseFields = ["subscriptionId", "plexusHash", "contentType", "requestId"]+ violations = concatMap (checkFieldNaming obj) (KM.keys obj)+ in violations+validateFieldNames _ = []++-- | Check if a field name violates naming conventions+checkFieldNaming :: Object -> K.Key -> [ProtocolViolation]+checkFieldNaming obj key =+ let keyText = K.toText key+ isCamelCase = T.any (\c -> c >= 'A' && c <= 'Z') keyText+ isProtocolField = keyText `elem` ["type", "metadata", "content_type", "content",+ "message", "percentage", "code", "plexus_hash",+ "provenance", "timestamp"]+ in if isCamelCase && isProtocolField+ then+ [ ProtocolViolation+ { pvMessage = "Protocol field '" <> keyText <> "' uses camelCase (should be snake_case)"+ , pvSeverity = Warning+ , pvLocation = InConnection+ , pvExpected = Just $ toSnakeCase keyText+ , pvActual = Just keyText+ , pvFix = Just $ "Rename to: " <> toSnakeCase keyText+ }+ ]+ else []++-- | Convert camelCase to snake_case (simple version)+toSnakeCase :: Text -> Text+toSnakeCase txt = T.pack $ go (T.unpack txt)+ where+ go [] = []+ go (c:cs)+ | c >= 'A' && c <= 'Z' = '_' : toLowerChar c : go cs -- Convert to lowercase and prepend underscore+ | otherwise = c : go cs+ toLowerChar c = toEnum (fromEnum c + 32)++-- ============================================================================+-- Type-Specific Validation+-- ============================================================================++-- | Validate based on stream item type+validateByType :: Object -> [ProtocolViolation]+validateByType obj = case KM.lookup "type" obj of+ Just (String "data") -> validateDataItem (Object obj)+ Just (String "progress") -> validateProgressItem (Object obj)+ Just (String "error") -> validateErrorItem (Object obj)+ Just (String "done") -> validateDoneItem (Object obj)+ Just (String "request") -> [] -- Request validation could be added+ _ -> []++-- | Validate data item structure+validateDataItem :: Value -> [ProtocolViolation]+validateDataItem (Object obj) = concat+ [ case KM.lookup "content_type" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'content_type' field in data item"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "content_type string (e.g., 'echo.response')"+ , pvActual = Just "missing"+ , pvFix = Just "Add content_type field"+ }+ ]+ Just (String ct) ->+ if T.null ct+ then+ [ ProtocolViolation+ { pvMessage = "content_type cannot be empty"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "non-empty string"+ , pvActual = Just "empty string"+ , pvFix = Just "Provide valid content_type"+ }+ ]+ else []+ Just other ->+ [ ProtocolViolation+ { pvMessage = "content_type must be a string"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "string"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change content_type to string"+ }+ ]+ , case KM.lookup "content" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'content' field in data item"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "content object with domain data"+ , pvActual = Just "missing"+ , pvFix = Just "Add content field"+ }+ ]+ Just _ -> [] -- Any JSON value is valid for content+ ]+validateDataItem _ = []++-- | Validate progress item structure+validateProgressItem :: Value -> [ProtocolViolation]+validateProgressItem (Object obj) = concat+ [ case KM.lookup "message" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'message' field in progress item"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "message string"+ , pvActual = Just "missing"+ , pvFix = Just "Add message field"+ }+ ]+ Just (String _) -> []+ Just other ->+ [ ProtocolViolation+ { pvMessage = "Progress message must be a string"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "string"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change message to string"+ }+ ]+ , case KM.lookup "percentage" obj of+ Nothing -> [] -- percentage is optional+ Just Null -> [] -- null is allowed+ Just (Number n) ->+ case floatingOrInteger n of+ Left (_ :: Double) -> [] -- Allow float percentages+ Right (pct :: Integer) ->+ if pct < 0 || pct > 100+ then+ [ ProtocolViolation+ { pvMessage = "Progress percentage must be between 0 and 100"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "0-100"+ , pvActual = Just $ T.pack $ show pct+ , pvFix = Just "Use percentage value between 0 and 100"+ }+ ]+ else []+ Just other ->+ [ ProtocolViolation+ { pvMessage = "Progress percentage must be a number or null"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "number or null"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change percentage to number or null"+ }+ ]+ ]+validateProgressItem _ = []++-- | Validate error item structure+validateErrorItem :: Value -> [ProtocolViolation]+validateErrorItem (Object obj) = concat+ [ case KM.lookup "message" obj of+ Nothing ->+ [ ProtocolViolation+ { pvMessage = "Missing 'message' field in error item"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "message string"+ , pvActual = Just "missing"+ , pvFix = Just "Add message field with error description"+ }+ ]+ Just (String msg) ->+ if T.null msg+ then+ [ ProtocolViolation+ { pvMessage = "Error message cannot be empty"+ , pvSeverity = Warning+ , pvLocation = InConnection+ , pvExpected = Just "non-empty error message"+ , pvActual = Just "empty string"+ , pvFix = Just "Provide descriptive error message"+ }+ ]+ else []+ Just other ->+ [ ProtocolViolation+ { pvMessage = "Error message must be a string"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "string"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change message to string"+ }+ ]+ , case KM.lookup "code" obj of+ Nothing -> [] -- code is optional+ Just Null -> [] -- null is allowed+ Just (String _) -> [] -- string code is valid+ Just other ->+ [ ProtocolViolation+ { pvMessage = "Error code must be a string or null"+ , pvSeverity = Error+ , pvLocation = InConnection+ , pvExpected = Just "string or null"+ , pvActual = Just $ T.pack $ show other+ , pvFix = Just "Change code to string or null"+ }+ ]+ ]+validateErrorItem _ = []++-- | Validate done item structure (only needs metadata)+validateDoneItem :: Value -> [ProtocolViolation]+validateDoneItem _item = [] -- Done items only need metadata, which is validated separately
src/Synapse/Transport.hs view
@@ -27,16 +27,20 @@ import Data.Text (Text) import qualified Data.Text as T -import Plexus.Client (SubstrateConfig(..))+import Plexus.Client (SubstrateConfig(..), cookieHeader) import qualified Plexus.Transport as ST import qualified Plexus.Types as PT import Synapse.Schema.Types import Synapse.Monad+import qualified Synapse.Log as Log -- | Fetch the root schema fetchSchema :: SynapseM PluginSchema-fetchSchema = fetchSchemaAt []+fetchSchema = do+ logger <- getLogger+ Log.logInfo logger Log.SubsystemSchema "Fetching root schema"+ fetchSchemaAt [] -- | Convert plexus-protocol TransportError to error message transportErrorToText :: PT.TransportError -> Text@@ -57,11 +61,20 @@ -- | Fetch schema at a specific path fetchSchemaAt :: Path -> SynapseM PluginSchema fetchSchemaAt path = do+ logger <- getLogger cfg <- getConfig+ Log.logDebug logger Log.SubsystemSchema $+ "Fetching schema at path: " <> T.pack (show path) result <- liftIO $ ST.fetchSchemaAt cfg path case result of- Left err -> throwNav $ FetchError (transportErrorToText err) path- Right schema -> pure schema+ Left err -> do+ Log.logInfo logger Log.SubsystemSchema $+ "Schema fetch failed: " <> T.pack (show path) <> " - " <> transportErrorToText err+ throwNav $ FetchError (transportErrorToText err) path+ Right schema -> do+ Log.logDebug logger Log.SubsystemSchema $+ "Schema fetch succeeded: " <> T.pack (show path) <> " (" <> psNamespace schema <> ")"+ pure schema -- | Fetch a specific method's schema (more efficient than full plugin schema) -- Uses the parameter-based query: plugin.schema with {"method": "name"}@@ -76,12 +89,18 @@ -- | Invoke a method and return stream items invoke :: Path -> Text -> Value -> SynapseM [HubStreamItem] invoke namespacePath method params = do+ logger <- getLogger cfg <- getConfig+ backend <- asks seBackend+ let fullPath = namespacePath ++ [method]+ Log.logInfo logger Log.SubsystemRPC $+ "Invoking RPC method: " <> T.intercalate "." fullPath <> " on " <> backend+ Log.logDebug logger Log.SubsystemRPC $+ "RPC request params: " <> T.pack (show params) result <- liftIO $ ST.invokeMethod cfg namespacePath method params case result of Left transportErr -> do -- Build context from typed transport error and environment- backend <- asks seBackend let path = namespacePath ++ [method] let ctx = TransportContext { tcMessage = transportErrorToText transportErr@@ -91,8 +110,13 @@ , tcPath = path , tcCategory = transportErrorToCategory transportErr }+ Log.logInfo logger Log.SubsystemRPC $+ "RPC invocation failed: " <> T.intercalate "." fullPath <> " - " <> transportErrorToText transportErr throwTransportWith ctx- Right items -> pure items+ Right items -> do+ Log.logDebug logger Log.SubsystemRPC $+ "RPC invocation succeeded: " <> T.intercalate "." fullPath <> " (" <> T.pack (show (length items)) <> " items)"+ pure items where getHost (PT.ConnectionRefused h _) _ = h getHost (PT.ConnectionTimeout h _) _ = h@@ -226,12 +250,14 @@ -- | Get SubstrateConfig from environment getConfig :: SynapseM SubstrateConfig getConfig = do- host <- asks seHost- port <- asks sePort+ host <- asks seHost+ port <- asks sePort backend <- asks seBackend+ mToken <- asks seToken pure $ SubstrateConfig- { substrateHost = T.unpack host- , substratePort = port- , substratePath = "/"+ { substrateHost = T.unpack host+ , substratePort = port+ , substratePath = "/" , substrateBackend = backend+ , substrateHeaders = maybe [] cookieHeader mToken }
+ test/ProtocolTestDemo.hs view
@@ -0,0 +1,51 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Demo program showing how to use the protocol test runner+--+-- Usage:+-- cabal run protocol-test-demo -- <host> <port> <backend>+--+-- Example:+-- cabal run protocol-test-demo -- localhost 3000 bash+--+module Main (main) where++import System.Environment (getArgs)+import System.Exit (exitFailure, exitSuccess)+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.IO as TIO++import Synapse.Self.Protocol.TestRunner (runProtocolTests)++main :: IO ()+main = do+ args <- getArgs+ case args of+ [hostStr, portStr, backendStr] -> do+ let host = T.pack hostStr+ let backend = T.pack backendStr+ case reads portStr of+ [(port, "")] -> runTests host port backend+ _ -> do+ TIO.putStrLn "Error: Invalid port number"+ exitFailure+ _ -> do+ TIO.putStrLn "Usage: protocol-test-demo <host> <port> <backend>"+ TIO.putStrLn "Example: protocol-test-demo localhost 3000 bash"+ exitFailure++runTests :: Text -> Int -> Text -> IO ()+runTests host port backend = do+ TIO.putStrLn $ "Running protocol tests against: " <> host <> ":" <> T.pack (show port) <> " (backend: " <> backend <> ")"+ TIO.putStrLn ""++ result <- runProtocolTests host port backend++ case result of+ Left errorReport -> do+ TIO.putStrLn errorReport+ exitFailure+ Right successReport -> do+ TIO.putStrLn successReport+ exitSuccess
+ test/ReporterDemo.hs view
@@ -0,0 +1,114 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Simple demo of the Reporter module+--+-- Usage: cabal run reporter-demo+--+-- This demonstrates all four output formats:+-- - Compiler-style (renderViolations)+-- - Summary (renderSummary)+-- - Compact (renderCompact)+-- - JSON (renderJSON)++module Main where++import qualified Data.Text.IO as TIO+import Synapse.Self.Protocol.Validator+ ( ProtocolViolation(..)+ , Severity(..)+ , Location(..)+ )+import Synapse.Self.Protocol.Reporter+ ( renderViolations+ , renderSummary+ , renderCompact+ , renderJSON+ )++main :: IO ()+main = do+ putStrLn "==================================================================="+ putStrLn "Protocol Violation Reporter Demo"+ putStrLn "==================================================================="+ putStrLn ""++ -- Create sample violations+ let violations =+ [ ProtocolViolation+ { pvMessage = "Missing StreamDone message"+ , pvSeverity = Error+ , pvLocation = InStream "echo.request" 5+ , pvExpected = Just "StreamDone message to complete stream"+ , pvActual = Just "Stream ended without StreamDone"+ , pvFix = Just "Ensure server sends StreamDone after all data"+ }+ , ProtocolViolation+ { pvMessage = "Missing 'plexus_hash' field in metadata"+ , pvSeverity = Error+ , pvLocation = InMessage (Just 42) 3 (Just "metadata")+ , pvExpected = Just "16-character lowercase hex string"+ , pvActual = Just "missing"+ , pvFix = Just "Add plexus_hash field with schema version hash"+ }+ , ProtocolViolation+ { pvMessage = "Protocol field 'contentType' uses camelCase (should be snake_case)"+ , pvSeverity = Warning+ , pvLocation = InMessage (Just 42) 2 (Just "contentType")+ , pvExpected = Just "content_type"+ , pvActual = Just "contentType"+ , pvFix = Just "Rename to: content_type"+ }+ , ProtocolViolation+ { pvMessage = "Timestamp must be an integer (not float)"+ , pvSeverity = Error+ , pvLocation = InMessage (Just 42) 1 (Just "metadata.timestamp")+ , pvExpected = Just "integer (use Math.floor(Date.now() / 1000) in JavaScript)"+ , pvActual = Just "1234567890.5"+ , pvFix = Just "Convert timestamp to integer: Math.floor(Date.now() / 1000)"+ }+ , ProtocolViolation+ { pvMessage = "Consider using progress messages for long-running operations"+ , pvSeverity = Info+ , pvLocation = InStream "process.task" 0+ , pvExpected = Nothing+ , pvActual = Nothing+ , pvFix = Just "Send periodic progress updates to improve UX"+ }+ ]++ -- 1. Compiler-style format+ putStrLn "===================================================================\n"+ putStrLn "1. COMPILER-STYLE FORMAT (renderViolations)"+ putStrLn "===================================================================\n"+ TIO.putStrLn $ renderViolations violations+ putStrLn ""++ -- 2. Summary format+ putStrLn "===================================================================\n"+ putStrLn "2. SUMMARY FORMAT (renderSummary)"+ putStrLn "===================================================================\n"+ TIO.putStrLn $ renderSummary violations+ putStrLn ""++ -- 3. Compact format+ putStrLn "===================================================================\n"+ putStrLn "3. COMPACT FORMAT (renderCompact)"+ putStrLn "===================================================================\n"+ TIO.putStrLn $ renderCompact violations+ putStrLn ""++ -- 4. JSON format+ putStrLn "===================================================================\n"+ putStrLn "4. JSON FORMAT (renderJSON)"+ putStrLn "===================================================================\n"+ TIO.putStrLn $ renderJSON violations+ putStrLn ""++ -- Test with no violations+ putStrLn "===================================================================\n"+ putStrLn "5. NO VIOLATIONS (all formats)"+ putStrLn "===================================================================\n"+ TIO.putStrLn $ renderViolations []+ TIO.putStrLn $ renderSummary []+ TIO.putStrLn $ renderCompact []+ putStrLn ""
+ test/StreamTrackerSpec.hs view
@@ -0,0 +1,101 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}++module Main where++import Test.Hspec+import Data.Aeson (Value(..), decode)+import qualified Data.ByteString.Lazy.Char8 as BL+import Data.Maybe (fromJust)++import Synapse.Self.Protocol.StreamTracker+import Synapse.Self.Protocol.Validator++main :: IO ()+main = hspec spec++spec :: Spec+spec = do+ describe "StreamTracker" $ do+ describe "trackMessage" $ do+ it "tracks subscription messages" $ do+ tracker <- newTracker+ let msg = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"data\", \"content\": \"hello\" } } }"+ violations <- trackMessage tracker msg+ violations `shouldBe` []+ count <- getStreamCount tracker+ count `shouldBe` 1++ it "detects missing StreamDone violation" $ do+ tracker <- newTracker+ let dataMsg = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"data\", \"content\": \"hello\" } } }"+ _ <- trackMessage tracker dataMsg+ violations <- checkCompletion tracker+ length violations `shouldBe` 1+ pvSeverity (head violations) `shouldBe` Error++ it "detects duplicate StreamDone violation" $ do+ tracker <- newTracker+ let doneMsg = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"done\" } } }"+ _ <- trackMessage tracker doneMsg+ violations <- trackMessage tracker doneMsg+ length violations `shouldBe` 1+ pvSeverity (head violations) `shouldBe` Error++ it "properly completes stream with StreamDone" $ do+ tracker <- newTracker+ let dataMsg = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"data\", \"content\": \"hello\" } } }"+ let doneMsg = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"done\" } } }"+ _ <- trackMessage tracker dataMsg+ _ <- trackMessage tracker doneMsg+ violations <- checkCompletion tracker+ violations `shouldBe` []++ it "tracks multiple subscriptions independently" $ do+ tracker <- newTracker+ let msg1 = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"data\", \"content\": \"stream1\" } } }"+ let msg2 = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 2, \"result\": { \"type\": \"data\", \"content\": \"stream2\" } } }"+ let done1 = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"done\" } } }"++ _ <- trackMessage tracker msg1+ _ <- trackMessage tracker msg2+ _ <- trackMessage tracker done1++ count <- getStreamCount tracker+ count `shouldBe` 2++ -- Only subscription 2 should be incomplete+ violations <- checkCompletion tracker+ length violations `shouldBe` 1++ it "detects messages after StreamDone" $ do+ tracker <- newTracker+ let doneMsg = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"done\" } } }"+ let dataMsg = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"data\", \"content\": \"after done\" } } }"++ _ <- trackMessage tracker doneMsg+ violations <- trackMessage tracker dataMsg++ length violations `shouldBe` 1+ pvSeverity (head violations) `shouldBe` Error++ describe "getAllViolations" $ do+ it "aggregates all violations from all streams" $ do+ tracker <- newTracker+ let msg1 = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"data\", \"content\": \"stream1\" } } }"+ let msg2 = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 2, \"result\": { \"type\": \"data\", \"content\": \"stream2\" } } }"+ let doneMsg1 = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"done\" } } }"+ let doneMsg1Dup = parseJson "{ \"jsonrpc\": \"2.0\", \"method\": \"subscription\", \"params\": { \"subscription\": 1, \"result\": { \"type\": \"done\" } } }"++ _ <- trackMessage tracker msg1+ _ <- trackMessage tracker msg2+ _ <- trackMessage tracker doneMsg1+ _ <- trackMessage tracker doneMsg1Dup -- Duplicate done++ allViolations <- getAllViolations tracker+ -- Should have: 1 duplicate StreamDone + 1 missing StreamDone for stream 2+ length allViolations `shouldBe` 2++-- Helper to parse JSON strings+parseJson :: String -> Value+parseJson s = fromJust $ decode (BL.pack s)
+ test/ValidationReportTest.hs view
@@ -0,0 +1,99 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Test that demonstrates Reporter working with real Validator output+--+-- Usage: cabal run validation-report-test --flag build-examples++module Main where++import Data.Aeson (Value(..), object, (.=))+import qualified Data.Aeson.KeyMap as KM+import qualified Data.Text.IO as TIO+import Synapse.Self.Protocol.Validator (validateStreamItem)+import Synapse.Self.Protocol.Reporter+ ( renderViolations+ , renderSummary+ , renderCompact+ , renderJSON+ )++main :: IO ()+main = do+ putStrLn "==================================================================="+ putStrLn "Validation + Reporter Integration Test"+ putStrLn "==================================================================="+ putStrLn ""++ -- Test 1: Invalid stream item (missing type field)+ putStrLn "Test 1: Missing 'type' field"+ putStrLn "-------------------------------------------------------------------"+ let invalidItem1 = object+ [ "metadata" .= object+ [ "provenance" .= (["test-server"] :: [String])+ , "plexus_hash" .= ("1234567890abcdef" :: String)+ , "timestamp" .= (1234567890 :: Int)+ ]+ ]+ violations1 <- validateStreamItem invalidItem1+ TIO.putStrLn $ renderCompact violations1+ putStrLn ""++ -- Test 2: Invalid metadata (bad plexus_hash)+ putStrLn "Test 2: Invalid plexus_hash (too short)"+ putStrLn "-------------------------------------------------------------------"+ let invalidItem2 = object+ [ "type" .= ("data" :: String)+ , "content_type" .= ("test.data" :: String)+ , "content" .= object ["value" .= (42 :: Int)]+ , "metadata" .= object+ [ "provenance" .= (["test-server"] :: [String])+ , "plexus_hash" .= ("abc" :: String) -- Too short!+ , "timestamp" .= (1234567890 :: Int)+ ]+ ]+ violations2 <- validateStreamItem invalidItem2+ TIO.putStrLn $ renderCompact violations2+ putStrLn ""++ -- Test 3: Valid item (no violations)+ putStrLn "Test 3: Valid stream item"+ putStrLn "-------------------------------------------------------------------"+ let validItem = object+ [ "type" .= ("data" :: String)+ , "content_type" .= ("test.data" :: String)+ , "content" .= object ["value" .= (42 :: Int)]+ , "metadata" .= object+ [ "provenance" .= (["test-server"] :: [String])+ , "plexus_hash" .= ("1234567890abcdef" :: String)+ , "timestamp" .= (1234567890 :: Int)+ ]+ ]+ violations3 <- validateStreamItem validItem+ TIO.putStrLn $ renderSummary violations3+ putStrLn ""++ -- Test 4: Multiple violations in one item+ putStrLn "Test 4: Multiple violations (missing fields + wrong types)"+ putStrLn "-------------------------------------------------------------------"+ let invalidItem4 = object+ [ "type" .= ("data" :: String)+ , "content_type" .= ("" :: String) -- Empty string+ -- Missing 'content' field+ , "metadata" .= object+ [ "provenance" .= ([] :: [String]) -- Empty array+ , "plexus_hash" .= (123 :: Int) -- Wrong type (number instead of string)+ , "timestamp" .= (1234567890.5 :: Double) -- Float instead of integer+ ]+ ]+ violations4 <- validateStreamItem invalidItem4+ TIO.putStrLn $ renderViolations violations4+ putStrLn ""++ -- Summary+ putStrLn "==================================================================="+ putStrLn "Summary Statistics"+ putStrLn "==================================================================="+ let totalViolations = length violations1 + length violations2 ++ length violations3 + length violations4+ putStrLn $ "Total violations across all tests: " ++ show totalViolations+ putStrLn ""
+ test/WebSocketRaw.hs view
@@ -0,0 +1,156 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Minimal WebSocket test - does exactly what websocat does+--+-- Usage: cabal run websocket-raw-test+--+-- This test replicates the exact behavior of:+-- echo '{"jsonrpc":"2.0","id":1,"method":"ping_pong","params":{"message":"test"}}' | websocat ws://127.0.0.1:4444++module Main (main) where++import qualified Data.Aeson as Aeson+import Data.Aeson (object, (.=))+import qualified Data.Aeson.KeyMap as KM+import qualified Data.ByteString.Lazy as LBS+import qualified Data.Text as T+import qualified Data.Text.IO as TIO+import qualified Network.WebSockets as WS+import Control.Monad (forever, void)+import Control.Concurrent (threadDelay, forkIO)+import Control.Exception (catch, SomeException)+import System.Exit (exitSuccess, exitFailure)++main :: IO ()+main = do+ TIO.putStrLn "WebSocket Raw Test - Connecting to ws://127.0.0.1:4444"++ result <- catch runTest handleError++ case result of+ True -> do+ TIO.putStrLn "\n[OK] Test PASSED"+ exitSuccess+ False -> do+ TIO.putStrLn "\n[ERR] Test FAILED"+ exitFailure++runTest :: IO Bool+runTest = WS.runClient "127.0.0.1" 4444 "/" $ \conn -> do+ TIO.putStrLn "[OK] WebSocket connection established"++ -- Send the exact JSON-RPC request that websocat sends+ let request = object+ [ "jsonrpc" .= ("2.0" :: T.Text)+ , "id" .= (1 :: Int)+ , "method" .= ("ping_pong" :: T.Text)+ , "params" .= object+ [ "message" .= ("test" :: T.Text)+ ]+ ]++ let requestBS = Aeson.encode request+ TIO.putStrLn $ "> Sending: " <> T.pack (show request)+ WS.sendTextData conn requestBS++ -- Receive messages (subscription ID, data, done)+ TIO.putStrLn "\nWaiting for responses..."++ -- Message 1: Subscription ID+ msg1 <- WS.receiveData conn+ TIO.putStrLn $ "< Received 1: " <> T.pack (show (msg1 :: LBS.ByteString))++ case Aeson.decode msg1 of+ Nothing -> do+ TIO.putStrLn "[ERR] Failed to decode message 1 as JSON"+ return False+ Just (Aeson.Object obj) -> do+ case KM.lookup "result" obj of+ Nothing -> do+ TIO.putStrLn "[ERR] Message 1 missing 'result' field (subscription ID)"+ return False+ Just (Aeson.Number subId) -> do+ TIO.putStrLn $ "[OK] Got subscription ID: " <> T.pack (show subId)++ -- Message 2: Data event+ msg2 <- WS.receiveData conn+ TIO.putStrLn $ "\n< Received 2: " <> T.pack (show (msg2 :: LBS.ByteString))++ case Aeson.decode msg2 of+ Nothing -> do+ TIO.putStrLn "[ERR] Failed to decode message 2 as JSON"+ return False+ Just (Aeson.Object obj2) -> do+ case KM.lookup "params" obj2 of+ Nothing -> do+ TIO.putStrLn "[ERR] Message 2 missing 'params' field"+ return False+ Just (Aeson.Object params) -> do+ case KM.lookup "result" params of+ Nothing -> do+ TIO.putStrLn "[ERR] Message 2 params missing 'result' field"+ return False+ Just (Aeson.Object result) -> do+ case KM.lookup "type" result of+ Just (Aeson.String "data") -> do+ TIO.putStrLn "[OK] Got data event"++ -- Message 3: Done event+ msg3 <- WS.receiveData conn+ TIO.putStrLn $ "\n< Received 3: " <> T.pack (show (msg3 :: LBS.ByteString))++ case Aeson.decode msg3 of+ Nothing -> do+ TIO.putStrLn "[ERR] Failed to decode message 3 as JSON"+ return False+ Just (Aeson.Object obj3) -> do+ case KM.lookup "params" obj3 of+ Nothing -> do+ TIO.putStrLn "[ERR] Message 3 missing 'params' field"+ return False+ Just (Aeson.Object params3) -> do+ case KM.lookup "result" params3 of+ Nothing -> do+ TIO.putStrLn "[ERR] Message 3 params missing 'result' field"+ return False+ Just (Aeson.Object result3) -> do+ case KM.lookup "type" result3 of+ Just (Aeson.String "done") -> do+ TIO.putStrLn "[OK] Got done event"+ TIO.putStrLn "\n[OK] All messages received successfully"+ return True+ _ -> do+ TIO.putStrLn "[ERR] Message 3 type is not 'done'"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 3 result is not an object"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 3 params is not an object"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 3 is not an object"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 2 type is not 'data'"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 2 result is not an object"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 2 params is not an object"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 2 is not an object"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 1 result is not a number"+ return False+ _ -> do+ TIO.putStrLn "[ERR] Message 1 is not an object"+ return False++handleError :: SomeException -> IO Bool+handleError e = do+ TIO.putStrLn $ "[ERR] Exception: " <> T.pack (show e)+ return False