packages feed

plexus-protocol (empty) → 0.3.0.0

raw patch · 8 files changed

+2173/−0 lines, 8 filesdep +aesondep +aeson-casingdep +async

Dependencies added: aeson, aeson-casing, async, base, bytestring, containers, directory, filepath, network, stm, streaming, text, time, websockets

Files

+ plexus-protocol.cabal view
@@ -0,0 +1,43 @@+cabal-version:      3.0+name:               plexus-protocol+version:            0.3.0.0+synopsis:           Plexus RPC protocol types and client+license:            MIT+author:+maintainer:+build-type:         Simple++library+  exposed-modules:+    Plexus.Client+    Plexus.Transport+    Plexus+    Plexus.Types+    Plexus.Schema+    Plexus.Schema.Cache+    Plexus.Schema.Recursive+  build-depends:+    base ^>=4.17 || ^>=4.18 || ^>=4.19,+    aeson >= 2.0,+    aeson-casing >= 0.2,+    text >= 2.0,+    bytestring >= 0.11,+    containers >= 0.6,+    websockets >= 0.13,+    network >= 3.1,+    async >= 2.2,+    stm >= 2.5,+    streaming >= 0.2,+    directory >= 1.3,+    filepath >= 1.4,+    time >= 1.9+  hs-source-dirs:   src+  default-language: GHC2021+  default-extensions:+    OverloadedStrings+    DeriveGeneric+    DeriveAnyClass+    DerivingStrategies+    GeneralizedNewtypeDeriving+    LambdaCase+    RecordWildCards
+ src/Plexus.hs view
@@ -0,0 +1,85 @@+-- | Plexus RPC Types and Client+--+-- Re-exports from Plexus (transport and client) and Plexus (RPC types).+--+-- = Quick Start+--+-- @+-- import Plexus+-- import qualified Streaming.Prelude as S+-- import Data.Aeson (toJSON)+--+-- main :: IO ()+-- main = do+--   -- Connect to substrate+--   conn <- connect defaultConfig+--+--   -- Call bash.execute and stream results+--   S.print $ substrateRpc conn "bash_execute" (toJSON ["echo hello"])+--+--   -- Clean up+--   disconnect conn+-- @+--+-- = Stream Items+--+-- The Plexus RPC protocol returns a unified stream type with variants:+--+-- * 'StreamProgress' - Progress updates with optional percentage+-- * 'StreamData' - Actual data with content type information+-- * 'StreamError' - Error (may be recoverable)+-- * 'StreamDone' - Stream completed successfully+--+module Plexus+  ( -- * Connection (re-exported from Plexus.Client)+    SubstrateConnection+  , SubstrateConfig(..)+  , defaultConfig+  , connect+  , disconnect++    -- * Core RPC (streaming)+  , substrateRpc++    -- * High-level RPC (collected, re-exported from Plexus.Transport)+  , rpcCall+  , rpcCallWith+  , fetchSchemaAt+  , invokeMethod+  , invokeRaw++    -- * High-level RPC (streaming)+  , rpcCallStreaming+  , invokeMethodStreaming++    -- * Bidirectional Response+  , sendBidirectionalResponse++    -- * Bidirectional Types+  , StandardRequest(..)+  , StandardResponse(..)+  , SelectOption(..)++    -- * Stream Types+  , PlexusStreamItem(..)+  , Provenance(..)++    -- * Schema Types+  , PluginSchema(..)+  , MethodSchema(..)+  , ChildSummary(..)+  , PluginHash+  , SchemaResult(..)++    -- * JSON-RPC Types+  , RpcRequest(..)+  , RpcResponse(..)+  , RpcError(..)+  , RequestId(..)+  , SubscriptionId(..)+  ) where++import Plexus.Types+import Plexus.Schema.Recursive+import Plexus.Client+import Plexus.Transport
+ src/Plexus/Client.hs view
@@ -0,0 +1,265 @@+-- | Low-level WebSocket client for Substrate+--+-- The core primitive is:+--+-- @+-- substrateRpc :: SubstrateConnection -> Value -> Stream (Of PlexusStreamItem) IO ()+-- @+--+-- This establishes a subscription and yields stream items until completion.+module Plexus.Client+  ( -- * Connection+    SubstrateConnection+  , connect+  , disconnect++    -- * Core RPC primitive+  , substrateRpc++    -- * Configuration+  , SubstrateConfig(..)+  , defaultConfig+  ) where++import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.Async (Async, async, cancel, wait)+import Control.Concurrent.STM+import Control.Exception (SomeException, catch)+import Control.Monad (forever, void)+import Data.Aeson+import qualified Data.Aeson.KeyMap as KM+import Data.IORef+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Text (Text)+import qualified Data.Text as T+import Network.WebSockets (Connection)+import qualified Network.WebSockets as WS+import Streaming+import qualified Streaming.Prelude as Str++import Plexus.Types++-- | Substrate connection configuration+data SubstrateConfig = SubstrateConfig+  { substrateHost    :: String+  , substratePort    :: Int+  , substratePath    :: String+  , substrateBackend :: Text    -- ^ Backend name (e.g., "plexus")+  }+  deriving stock (Show, Eq)++-- | Default configuration for local development (requires backend)+defaultConfig :: Text -> SubstrateConfig+defaultConfig backend = SubstrateConfig+  { substrateHost = "127.0.0.1"+  , substratePort = 4444+  , substratePath = "/"+  , substrateBackend = backend+  }++-- | Pending request with its queue, waiting for subscription ID+data PendingRequest = PendingRequest+  { prQueue    :: TQueue Value        -- ^ Queue to receive notifications+  , prResponse :: TMVar RpcResponse   -- ^ Response channel+  }++-- | Active connection to substrate+data SubstrateConnection = SubstrateConnection+  { scConnection    :: Connection+  , scNextId        :: IORef Int+  , scSubscriptions :: TVar (Map SubscriptionId (TQueue Value))+  , scPendingReqs   :: TVar (Map RequestId PendingRequest)+  , scReaderThread  :: Async ()+  }++-- | Connect to substrate+-- Throws an exception if connection fails+connect :: SubstrateConfig -> IO SubstrateConnection+connect SubstrateConfig{..} = do+  -- Initialize state+  nextId <- newIORef 1+  subs   <- newTVarIO Map.empty+  pendingReqs <- newTVarIO Map.empty+  resultVar <- newEmptyTMVarIO  -- Either error message or (conn, reader)++  -- Run WebSocket client in a background thread+  -- Catch exceptions inside the thread to prevent them from being printed+  void $ forkIO $+    (WS.runClient substrateHost substratePort substratePath $ \conn -> do+      reader <- async $ readerLoop conn subs pendingReqs+      -- Signal success+      atomically $ putTMVar resultVar (Right (conn, reader))+      -- Keep alive until reader exits+      void (wait reader) `catch` \(_ :: SomeException) -> pure ()+    ) `catch` \(e :: SomeException) ->+      -- Signal failure (don't print, just capture)+      atomically $ putTMVar resultVar (Left $ show e)++  -- Wait for connection result (with timeout)+  threadDelay 200000  -- 200ms+  mResult <- atomically $ tryTakeTMVar resultVar++  case mResult of+    Nothing -> error "Connection timeout"+    Just (Left err) -> error $ "Connection failed: " <> err+    Just (Right (conn, reader)) ->+      pure SubstrateConnection+        { scConnection    = conn+        , scNextId        = nextId+        , scSubscriptions = subs+        , scPendingReqs   = pendingReqs+        , scReaderThread  = reader+        }++-- | Disconnect from substrate+disconnect :: SubstrateConnection -> IO ()+disconnect SubstrateConnection{..} = do+  cancel scReaderThread+  WS.sendClose scConnection ("bye" :: Text) `catch` \(_ :: SomeException) -> pure ()++-- | Reader loop - dispatches incoming messages to the right handler+readerLoop+  :: Connection+  -> TVar (Map SubscriptionId (TQueue Value))+  -> TVar (Map RequestId PendingRequest)+  -> IO ()+readerLoop conn subs pendingReqs = forever $ do+  msg <- WS.receiveData conn+  case eitherDecode msg of+    Left err -> putStrLn $ "Failed to decode message: " <> err+    Right val -> dispatch val+  where+    dispatch :: Value -> IO ()+    dispatch val = case val of+      Object o+        -- Check if it's a subscription notification (has "params.subscription" but no "id")+        | Just (Object params) <- KM.lookup "params" o+        , Just _ <- KM.lookup "subscription" params+        , Nothing <- KM.lookup "id" o+        -> handleNotification val++        -- Otherwise it's a response (has "id")+        | Just _ <- KM.lookup "id" o+        -> handleResponse val++      _ -> putStrLn $ "Unknown message format: " <> show val++    handleNotification :: Value -> IO ()+    handleNotification val =+      case fromJSON val of+        Success (SubscriptionNotification _ _ params) -> do+          let subId = subParamsSubscription params+              result = subParamsResult params+          mQueue <- Map.lookup subId <$> readTVarIO subs+          case mQueue of+            Just queue -> atomically $ writeTQueue queue result+            Nothing    -> putStrLn $ "Unknown subscription: " <> show subId+        Error err -> putStrLn $ "Failed to parse notification: " <> err++    handleResponse :: Value -> IO ()+    handleResponse val =+      case fromJSON val of+        Success resp -> do+          let rid = rpcRespId resp+          mPending <- Map.lookup rid <$> readTVarIO pendingReqs+          case mPending of+            Just (PendingRequest queue respVar) -> do+              -- For successful responses, register the subscription queue BEFORE+              -- signaling completion. This prevents race with notifications.+              case resp of+                RpcSuccess _ result ->+                  case fromJSON result of+                    Success subId -> atomically $ do+                      modifyTVar' subs $ Map.insert subId queue+                      putTMVar respVar resp+                    Error _ ->+                      -- Can't parse subId, just signal response+                      atomically $ putTMVar respVar resp+                RpcError{} ->+                  atomically $ putTMVar respVar resp+            Nothing -> putStrLn $ "Unknown request id: " <> show rid+        Error err -> putStrLn $ "Failed to parse response: " <> err++-- | The core RPC primitive+--+-- @substrateRpc conn method params@ sends a subscription request and returns a stream+-- of 'PlexusStreamItem' values. The stream completes when a 'StreamDone' or+-- 'StreamError' item is received.+--+-- Example:+--+-- @+-- import qualified Streaming.Prelude as S+--+-- main = do+--   conn <- connect defaultConfig+--   S.print $ substrateRpc conn "bash_execute" (toJSON ["echo hello"])+-- @+substrateRpc+  :: SubstrateConnection+  -> Text              -- ^ Method name (e.g., "bash_execute")+  -> Value             -- ^ Parameters+  -> Stream (Of PlexusStreamItem) IO ()+substrateRpc conn method params = do+  -- Get next request ID+  rid <- liftIO $ atomicModifyIORef' (scNextId conn) $ \n -> (n + 1, RequestId n)++  -- Create queue for this subscription and response channel+  queue <- liftIO newTQueueIO+  respVar <- liftIO newEmptyTMVarIO++  -- Register pending request (includes queue so reader can register subscription)+  liftIO $ atomically $ modifyTVar' (scPendingReqs conn) $+    Map.insert rid (PendingRequest queue respVar)++  -- Send subscription request+  let req = mkSubscribeRequest rid method params+  liftIO $ WS.sendTextData (scConnection conn) (encode req)++  -- Wait for subscription confirmation+  resp <- liftIO $ atomically $ takeTMVar respVar++  -- Clean up pending request+  liftIO $ atomically $ modifyTVar' (scPendingReqs conn) $ Map.delete rid++  case resp of+    RpcError _ err -> do+      -- Emit error as StreamError so caller can handle it properly+      Str.yield $ StreamError+        { itemPlexusHash = ""  -- No hash available for subscription errors+        , itemProvenance = Provenance ["substrate"]+        , itemError = T.pack $ "Subscription error: " <> show err+        , itemRecoverable = False+        }++    RpcSuccess _ result -> do+      -- The reader loop already registered the queue in scSubscriptions+      -- when it processed the response, so we just need the subId for cleanup+      case fromJSON result of+        Error err -> do+          liftIO $ putStrLn $ "Failed to parse subscription id: " <> err+          pure ()++        Success subId -> do+          -- Stream items until done, cleanup when finished+          streamItems queue subId <* liftIO (cleanup subId)+  where+    streamItems :: TQueue Value -> SubscriptionId -> Stream (Of PlexusStreamItem) IO ()+    streamItems queue subId = do+      val <- liftIO $ atomically $ readTQueue queue+      case fromJSON val of+        Error err -> do+          liftIO $ putStrLn $ "Failed to parse stream item: " <> err+          streamItems queue subId++        Success item -> do+          Str.yield item+          case item of+            StreamDone{}  -> pure ()  -- End of stream+            StreamError{} -> pure ()  -- Error terminates stream+            _             -> streamItems queue subId++    cleanup :: SubscriptionId -> IO ()+    cleanup subId = atomically $ modifyTVar' (scSubscriptions conn) $+      Map.delete subId
+ src/Plexus/Schema.hs view
@@ -0,0 +1,568 @@+-- | Schema types for dynamic CLI discovery+--+-- These types mirror the Substrate server's PlexusSchema and ActivationInfo,+-- allowing the CLI to discover available commands at runtime.+module Plexus.Schema+  ( -- * Schema Types+    PlexusSchema(..)+  , ActivationInfo(..)+  , PlexusSchemaEvent(..)+    -- * Enriched Schema Types (deprecated - use Full Schema)+  , EnrichedSchema(..)+  , SchemaProperty(..)+  , ActivationSchemaEvent(..)+    -- * Full Schema Types (from plexus_full_schema)+  , ActivationFullSchema(..)+  , MethodSchemaInfo(..)+  , FullSchemaEvent(..)+    -- * Hash Types+  , PlexusHash(..)+  , PlexusHashEvent(..)+    -- * Method Schema (parsed from enriched)+  , MethodSchema(..)+  , ParamSchema(..)+  , ParamType(..)+    -- * Parsing+  , parseMethodSchemas+  , parseMethodVariantByIndex+  , parseParamType+    -- * Stream Helpers+  , extractSchemaEvent+  , extractActivationSchemaEvent+  , extractFullSchemaEvent+  , extractHashEvent+  , extractRecursiveSchema+  ) where++import Control.Applicative ((<|>))+import Data.Aeson+import Data.Aeson.Types (Parser)+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Maybe (catMaybes, fromMaybe, mapMaybe)+import Data.Text (Text)+import qualified Data.Text as T+import GHC.Generics (Generic)++import Plexus.Types (PlexusStreamItem(..))+import qualified Plexus.Schema.Recursive as RecursiveSchema++-- ============================================================================+-- Core Schema Types (from {backend}.schema subscription)+-- ============================================================================++-- | Information about a single activation+data ActivationInfo = ActivationInfo+  { activationNamespace   :: Text   -- ^ e.g., "arbor", "cone", "health"+  , activationVersion     :: Text   -- ^ Semantic version e.g., "1.0.0"+  , activationDescription :: Text   -- ^ Human-readable description+  , activationMethods     :: [Text] -- ^ Method names e.g., ["tree_create", "tree_get"]+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON ActivationInfo where+  parseJSON = withObject "ActivationInfo" $ \o ->+    ActivationInfo+      <$> o .: "namespace"+      <*> o .: "version"+      <*> o .: "description"+      <*> o .: "methods"++instance ToJSON ActivationInfo where+  toJSON ActivationInfo{..} = object+    [ "namespace"   .= activationNamespace+    , "version"     .= activationVersion+    , "description" .= activationDescription+    , "methods"     .= activationMethods+    ]++-- | Top-level schema returned by {backend}.schema subscription+data PlexusSchema = PlexusSchema+  { schemaActivations  :: [ActivationInfo]+  , schemaTotalMethods :: Int+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON PlexusSchema where+  parseJSON = withObject "PlexusSchema" $ \o ->+    PlexusSchema+      <$> o .: "activations"+      <*> o .: "total_methods"++instance ToJSON PlexusSchema where+  toJSON PlexusSchema{..} = object+    [ "activations"   .= schemaActivations+    , "total_methods" .= schemaTotalMethods+    ]++-- | Event wrapper for plexus schema stream+data PlexusSchemaEvent+  = SchemaData PlexusSchema+  | SchemaError Text+  deriving stock (Show, Eq, Generic)++instance FromJSON PlexusSchemaEvent where+  parseJSON = withObject "PlexusSchemaEvent" $ \o -> do+    mTyp <- o .:? "type"+    case mTyp :: Maybe Text of+      Just "error" -> SchemaError <$> o .: "message"+      _ -> do+        -- Schema data comes directly without type wrapper+        mActivations <- o .:? "activations" :: Parser (Maybe Value)+        case mActivations of+          Just _ -> SchemaData <$> parseJSON (Object o)+          Nothing -> fail "PlexusSchemaEvent: expected 'activations' field"++-- ============================================================================+-- Enriched Schema Types (from plexus_activation_schema)+-- ============================================================================++-- ============================================================================+-- $ref Resolution Helpers+-- ============================================================================++-- | Resolve $ref references in a schema using $defs+resolveSchemaRefs :: EnrichedSchema -> EnrichedSchema+resolveSchemaRefs schema = case schemaRef schema of+  Just ref | Just defs <- schemaDefs schema ->+    -- Resolve the reference+    case lookupRef ref defs of+      Just resolved ->+        -- Merge resolved schema with local description+        let merged = mergeSchema schema resolved+        -- Continue resolving recursively+        in resolveSchemaRefs merged { schemaRef = Nothing, schemaDefs = schemaDefs schema }+      Nothing -> schema  -- Reference not found, keep as-is+  _ ->+    -- No ref to resolve, but recursively resolve properties and oneOf+    -- Propagate $defs to children so they can resolve their refs+    let defs = schemaDefs schema+    in schema+      { schemaProperties = resolvePropertiesRefs defs <$> schemaProperties schema+      , schemaOneOf = map (propagateDefsAndResolve defs) <$> schemaOneOf schema+      }+  where+    -- Propagate parent $defs to child schema and resolve+    propagateDefsAndResolve :: Maybe (Map Text Value) -> EnrichedSchema -> EnrichedSchema+    propagateDefsAndResolve parentDefs child =+      let childWithDefs = case (parentDefs, schemaDefs child) of+            (Just pDefs, Nothing) -> child { schemaDefs = Just pDefs }+            _ -> child  -- Child already has defs or parent has none+      in resolveSchemaRefs childWithDefs++-- | Look up a $ref path in definitions+-- e.g., "#/$defs/ConeIdentifier" -> look up "ConeIdentifier" in defs+lookupRef :: Text -> Map Text Value -> Maybe EnrichedSchema+lookupRef ref defs+  | T.isPrefixOf "#/$defs/" ref =+      let defName = T.drop 8 ref  -- Drop "#/$defs/"+      in case Map.lookup defName defs of+        Just val -> case fromJSON val of+          Success schema -> Just schema+          Error _ -> Nothing+        Nothing -> Nothing+  | otherwise = Nothing++-- | Merge a schema with a resolved reference, preserving local description+mergeSchema :: EnrichedSchema -> EnrichedSchema -> EnrichedSchema+mergeSchema local resolved = resolved+  { schemaDescription = schemaDescription local <|> schemaDescription resolved+  }++-- | Resolve $refs in schema properties+resolvePropertiesRefs :: Maybe (Map Text Value) -> Map Text SchemaProperty -> Map Text SchemaProperty+resolvePropertiesRefs defs props = Map.map (resolvePropertyRefs defs) props++-- | Resolve $ref in a single property+resolvePropertyRefs :: Maybe (Map Text Value) -> SchemaProperty -> SchemaProperty+resolvePropertyRefs maybeDefs prop = case (propRef prop, maybeDefs) of+  (Just ref, Just defs) | T.isPrefixOf "#/$defs/" ref ->+    let defName = T.drop 8 ref+    in case Map.lookup defName defs of+      Just val -> case fromJSON val of+        Success resolved ->+          -- Merge resolved with local description, then recursively resolve the result+          let merged = resolved { propDescription = propDescription prop <|> propDescription resolved }+          in resolvePropertyRefs maybeDefs merged { propRef = Nothing }+        Error _ -> prop  -- Parse error, keep original+      Nothing -> prop  -- Ref not found, keep original+  _ ->+    -- No ref to resolve, but recursively resolve nested properties+    prop+      { propProperties = resolvePropertiesRefs maybeDefs <$> propProperties prop+      , propOneOf = map (resolvePropertyRefs maybeDefs) <$> propOneOf prop+      , propItems = resolvePropertyRefs maybeDefs <$> propItems prop+      }++-- ============================================================================+-- Enriched Schema Types (from plexus_activation_schema)+-- ============================================================================++-- | Enriched JSON Schema from Substrate server+-- Mirrors the Rust Schema type from plexus/schema.rs+data EnrichedSchema = EnrichedSchema+  { schemaTitle       :: Maybe Text+  , schemaDescription :: Maybe Text+  , schemaType        :: Maybe Value         -- ^ "object", "string", etc.+  , schemaProperties  :: Maybe (Map Text SchemaProperty)+  , schemaRequired    :: Maybe [Text]+  , schemaOneOf       :: Maybe [EnrichedSchema]  -- ^ Method variants+  , schemaDefs        :: Maybe (Map Text Value)  -- ^ Schema definitions for $ref resolution+  , schemaRef         :: Maybe Text              -- ^ Reference to definition (e.g., "#/$defs/ConeIdentifier")+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON EnrichedSchema where+  parseJSON val@(Object o) = do+    -- Parse base schema+    base <- EnrichedSchema+      <$> o .:? "title"+      <*> o .:? "description"+      <*> o .:? "type"+      <*> o .:? "properties"+      <*> o .:? "required"+      <*> o .:? "oneOf"+      <*> o .:? "$defs"+      <*> o .:? "$ref"+    -- Apply $ref resolution if needed+    pure $ resolveSchemaRefs base+  parseJSON _ = fail "EnrichedSchema must be an object"++instance ToJSON EnrichedSchema where+  toJSON EnrichedSchema{..} = object $ catMaybes+    [ ("title" .=) <$> schemaTitle+    , ("description" .=) <$> schemaDescription+    , ("type" .=) <$> schemaType+    , ("properties" .=) <$> schemaProperties+    , ("required" .=) <$> schemaRequired+    , ("oneOf" .=) <$> schemaOneOf+    , ("$defs" .=) <$> schemaDefs+    , ("$ref" .=) <$> schemaRef+    ]++-- | Schema property definition+data SchemaProperty = SchemaProperty+  { propType        :: Maybe Value       -- ^ "string", "integer", ["string", "null"]+  , propDescription :: Maybe Text+  , propFormat      :: Maybe Text        -- ^ "uuid", "date-time", etc.+  , propItems       :: Maybe SchemaProperty  -- ^ For array types+  , propDefault     :: Maybe Value+  , propEnum        :: Maybe [Value]+  , propProperties  :: Maybe (Map Text SchemaProperty)  -- ^ Nested object+  , propRequired    :: Maybe [Text]      -- ^ Required fields for object types+  , propOneOf       :: Maybe [SchemaProperty]  -- ^ OneOf variants (for enums)+  , propRef         :: Maybe Text        -- ^ Reference to definition+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON SchemaProperty where+  parseJSON = withObject "SchemaProperty" $ \o ->+    SchemaProperty+      <$> o .:? "type"+      <*> o .:? "description"+      <*> o .:? "format"+      <*> o .:? "items"+      <*> o .:? "default"+      <*> o .:? "enum"+      <*> o .:? "properties"+      <*> o .:? "required"+      <*> o .:? "oneOf"+      <*> o .:? "$ref"++instance ToJSON SchemaProperty where+  toJSON SchemaProperty{..} = object $ catMaybes+    [ ("type" .=) <$> propType+    , ("description" .=) <$> propDescription+    , ("format" .=) <$> propFormat+    , ("items" .=) <$> propItems+    , ("default" .=) <$> propDefault+    , ("enum" .=) <$> propEnum+    , ("properties" .=) <$> propProperties+    , ("required" .=) <$> propRequired+    , ("oneOf" .=) <$> propOneOf+    , ("$ref" .=) <$> propRef+    ]++-- | Event wrapper for activation schema stream+data ActivationSchemaEvent+  = ActivationSchemaData EnrichedSchema+  | ActivationSchemaError Text+  deriving stock (Show, Eq, Generic)++instance FromJSON ActivationSchemaEvent where+  parseJSON val = case val of+    Object o -> do+      mErr <- o .:? "error"+      case mErr of+        Just err -> pure $ ActivationSchemaError err+        Nothing  -> ActivationSchemaData <$> parseJSON val+    _ -> fail "ActivationSchemaEvent: expected object"++-- ============================================================================+-- Full Schema Types (from plexus_full_schema)+-- ============================================================================++-- | Schema info for a single method+data MethodSchemaInfo = MethodSchemaInfo+  { methodInfoName        :: Text+  , methodInfoDescription :: Text+  , methodInfoParams      :: Maybe Value  -- ^ JSON Schema for params+  , methodInfoReturns     :: Maybe Value  -- ^ JSON Schema for return/stream events+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON MethodSchemaInfo where+  parseJSON = withObject "MethodSchemaInfo" $ \o ->+    MethodSchemaInfo+      <$> o .: "name"+      <*> o .: "description"+      <*> o .:? "params"+      <*> o .:? "returns"++instance ToJSON MethodSchemaInfo where+  toJSON MethodSchemaInfo{..} = object $ catMaybes+    [ Just ("name" .= methodInfoName)+    , Just ("description" .= methodInfoDescription)+    , ("params" .=) <$> methodInfoParams+    , ("returns" .=) <$> methodInfoReturns+    ]++-- | Full schema for an activation+data ActivationFullSchema = ActivationFullSchema+  { fullSchemaNamespace   :: Text+  , fullSchemaVersion     :: Text+  , fullSchemaDescription :: Text+  , fullSchemaMethods     :: [MethodSchemaInfo]+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON ActivationFullSchema where+  parseJSON = withObject "ActivationFullSchema" $ \o ->+    ActivationFullSchema+      <$> o .: "namespace"+      <*> o .: "version"+      <*> o .: "description"+      <*> o .: "methods"++instance ToJSON ActivationFullSchema where+  toJSON ActivationFullSchema{..} = object+    [ "namespace"   .= fullSchemaNamespace+    , "version"     .= fullSchemaVersion+    , "description" .= fullSchemaDescription+    , "methods"     .= fullSchemaMethods+    ]++-- | Event wrapper for full schema stream+data FullSchemaEvent+  = FullSchemaData ActivationFullSchema+  | FullSchemaError Text+  deriving stock (Show, Eq, Generic)++instance FromJSON FullSchemaEvent where+  parseJSON val = case val of+    Object o -> do+      mErr <- o .:? "error"+      case mErr of+        Just err -> pure $ FullSchemaError err+        Nothing  -> FullSchemaData <$> parseJSON val+    _ -> fail "FullSchemaEvent: expected object"++-- ============================================================================+-- Hash Types (from plexus_hash)+-- ============================================================================++-- | Plexus hash for cache invalidation+data PlexusHash = PlexusHash+  { plexusHash :: Text  -- ^ Hash of all activations (namespace:version:methods)+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON PlexusHash where+  parseJSON = withObject "PlexusHash" $ \o ->+    PlexusHash <$> o .: "hash"++instance ToJSON PlexusHash where+  toJSON PlexusHash{..} = object+    [ "hash" .= plexusHash ]++-- | Event wrapper for plexus hash stream+data PlexusHashEvent+  = HashData PlexusHash+  | HashError Text+  deriving stock (Show, Eq, Generic)++instance FromJSON PlexusHashEvent where+  parseJSON val = case val of+    Object o -> do+      mErr <- o .:? "error"+      case mErr of+        Just err -> pure $ HashError err+        Nothing  -> HashData <$> parseJSON val+    _ -> fail "PlexusHashEvent: expected object"++-- ============================================================================+-- Method Schema Types (parsed from EnrichedSchema)+-- ============================================================================++-- | Parameter type enumeration+data ParamType+  = ParamString+  | ParamInteger+  | ParamNumber+  | ParamBoolean+  | ParamObject+  | ParamArray ParamType+  | ParamNullable ParamType+  deriving stock (Show, Eq, Generic)++-- | Schema for a single parameter+data ParamSchema = ParamSchema+  { paramName     :: Text+  , paramType     :: ParamType+  , paramFormat   :: Maybe Text   -- ^ e.g., "uuid"+  , paramRequired :: Bool+  , paramDesc     :: Maybe Text+  , paramDefault  :: Maybe Value+  }+  deriving stock (Show, Eq, Generic)++-- | Schema for a method with full param details+data MethodSchema = MethodSchema+  { methodName        :: Text+  , methodDescription :: Maybe Text+  , methodParams      :: [ParamSchema]+  }+  deriving stock (Show, Eq, Generic)++-- | Parse ParamType from JSON Schema type field+parseParamType :: Value -> ParamType+parseParamType (String "string")  = ParamString+parseParamType (String "integer") = ParamInteger+parseParamType (String "number")  = ParamNumber+parseParamType (String "boolean") = ParamBoolean+parseParamType (String "object")  = ParamObject+parseParamType (Array types)      =+  -- Handle nullable types like ["string", "null"]+  let typeList = foldr (:) [] types+  in case filter (/= String "null") typeList of+    [t] -> ParamNullable (parseParamType t)+    _   -> ParamObject  -- fallback+parseParamType _                  = ParamObject  -- fallback++-- | Parse an EnrichedSchema into a list of MethodSchemas+-- The enriched schema has oneOf containing method variants+parseMethodSchemas :: EnrichedSchema -> [MethodSchema]+parseMethodSchemas schema = case schemaOneOf schema of+  Just variants -> mapMaybe parseMethodVariant variants+  Nothing -> []++-- | Parse a single method variant from the oneOf array+parseMethodVariant :: EnrichedSchema -> Maybe MethodSchema+parseMethodVariant variant = do+  props <- schemaProperties variant+  -- Get method name from "method" property's const/enum (may not exist)+  let methodProp = Map.lookup "method" props+      mMethodName = methodProp >>= extractMethodName+  -- Get params from "params" property+  let params = case Map.lookup "params" props of+        Just paramsProp -> extractParams paramsProp (propRequired paramsProp)+        Nothing -> []+  -- Return schema even without method name (we'll use index-based lookup)+  pure MethodSchema+    { methodName = fromMaybe "" mMethodName+    , methodDescription = schemaDescription variant+    , methodParams = params+    }++-- | Parse a method variant by index (for when method names aren't in schema)+parseMethodVariantByIndex :: EnrichedSchema -> Int -> Maybe MethodSchema+parseMethodVariantByIndex schema idx = case schemaOneOf schema of+  Just variants | idx < length variants -> parseMethodVariant (variants !! idx)+  _ -> Nothing++-- | Extract method name from the method property+extractMethodName :: SchemaProperty -> Maybe Text+extractMethodName prop = case propEnum prop of+  Just (String name : _) -> Just name+  _ -> Nothing++-- | Extract parameters from the params property+extractParams :: SchemaProperty -> Maybe [Text] -> [ParamSchema]+extractParams paramsProp mRequired =+  case propProperties paramsProp of+    Just props -> map (toParamSchema required) (Map.toList props)+    Nothing -> []+  where+    required = fromMaybe [] mRequired++-- | Convert a property to ParamSchema+toParamSchema :: [Text] -> (Text, SchemaProperty) -> ParamSchema+toParamSchema required (name, prop) = ParamSchema+  { paramName = name+  , paramType = maybe ParamObject parseParamType (propType prop)+  , paramFormat = propFormat prop+  , paramRequired = name `elem` required+  , paramDesc = propDescription prop+  , paramDefault = propDefault prop+  }++-- ============================================================================+-- Stream Helpers+-- ============================================================================++-- | Extract PlexusSchemaEvent from a stream item+-- The {backend}.schema subscription uses content_type "plexus.schema"+extractSchemaEvent :: PlexusStreamItem -> Maybe PlexusSchemaEvent+extractSchemaEvent (StreamData _ _ contentType dat)+  | contentType == "plexus.schema" =+      case fromJSON dat of+        Success evt -> Just evt+        Error _     -> Nothing+  | otherwise = Nothing+extractSchemaEvent _ = Nothing++-- | Extract ActivationSchemaEvent from a stream item+-- The plexus_activation_schema subscription uses content_type "plexus.activation_schema"+extractActivationSchemaEvent :: PlexusStreamItem -> Maybe ActivationSchemaEvent+extractActivationSchemaEvent (StreamData _ _ contentType dat)+  | contentType == "plexus.activation_schema" =+      case fromJSON dat of+        Success schema -> Just (ActivationSchemaData schema)+        Error _        -> Nothing+  | otherwise = Nothing+extractActivationSchemaEvent (StreamError _ _ err _) = Just (ActivationSchemaError err)+extractActivationSchemaEvent _ = Nothing++-- | Extract FullSchemaEvent from a stream item+-- The plexus_full_schema subscription uses content_type "plexus.full_schema"+extractFullSchemaEvent :: PlexusStreamItem -> Maybe FullSchemaEvent+extractFullSchemaEvent (StreamData _ _ contentType dat)+  | contentType == "plexus.full_schema" =+      case fromJSON dat of+        Success schema -> Just (FullSchemaData schema)+        Error _        -> Nothing+  | otherwise = Nothing+extractFullSchemaEvent (StreamError _ _ err _) = Just (FullSchemaError err)+extractFullSchemaEvent _ = Nothing++-- | Extract PlexusHashEvent from a stream item+-- The plexus_hash subscription uses content_type "plexus.hash"+extractHashEvent :: PlexusStreamItem -> Maybe PlexusHashEvent+extractHashEvent (StreamData _ _ contentType dat)+  | contentType == "plexus.hash" =+      case fromJSON dat of+        Success hash -> Just (HashData hash)+        Error _      -> Nothing+  | otherwise = Nothing+extractHashEvent (StreamError _ _ err _) = Just (HashError err)+extractHashEvent _ = Nothing++-- | Extract recursive plugin schema from stream item+-- The {backend}.schema subscription uses content_type "plexus.schema"+-- and returns the full recursive schema structure+extractRecursiveSchema :: PlexusStreamItem -> Maybe (Either Text RecursiveSchema.PluginSchema)+extractRecursiveSchema (StreamData _ _ contentType dat)+  | contentType == "plexus.schema" = Just $ RecursiveSchema.parsePluginSchema dat+  | otherwise = Nothing+extractRecursiveSchema (StreamError _ _ err _) = Just (Left err)+extractRecursiveSchema _ = Nothing
+ src/Plexus/Schema/Cache.hs view
@@ -0,0 +1,213 @@+-- | Schema caching for the dynamic CLI+--+-- Caches the PlexusSchema and enriched schemas to disk to avoid fetching on every invocation.+module Plexus.Schema.Cache+  ( -- * Types+    CachedSchema(..)+  , CacheConfig(..)+    -- * Cache Operations+  , defaultCacheConfig+  , defaultCachePath+  , loadCache+  , saveCache+  , isFresh+  , loadSchemaWithCache+    -- * Enriched Schema Lookup+  , lookupMethodSchema+  ) where++import Control.Exception (SomeException, catch)+import Data.Aeson+import qualified Data.ByteString.Lazy as LBS+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Text (Text)+import qualified Data.Text as T+import GHC.Generics (Generic)+import System.Directory (createDirectoryIfMissing, doesFileExist, getXdgDirectory, XdgDirectory(..))+import System.FilePath ((</>), takeDirectory)++import qualified Plexus.Schema+import Plexus.Schema (PlexusSchema(..), ActivationInfo(..), EnrichedSchema, ActivationFullSchema(..), MethodSchemaInfo(..), MethodSchema(..), parseMethodSchemas, PlexusHash(..), PlexusHashEvent(..), extractHashEvent)+import Plexus.Types (PlexusStreamItem)++-- ============================================================================+-- Types+-- ============================================================================++-- | Cached schema with metadata+data CachedSchema = CachedSchema+  { cachedHash        :: Text                             -- ^ Plexus hash for invalidation+  , cachedSchema      :: PlexusSchema                     -- ^ The cached plexus schema+  , cachedFullSchemas :: Map Text ActivationFullSchema    -- ^ Full schemas by namespace+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON CachedSchema where+  parseJSON = withObject "CachedSchema" $ \o ->+    CachedSchema+      <$> o .: "hash"+      <*> o .: "schema"+      <*> o .:? "full_schemas" .!= Map.empty++instance ToJSON CachedSchema where+  toJSON CachedSchema{..} = object+    [ "hash"         .= cachedHash+    , "schema"       .= cachedSchema+    , "full_schemas" .= cachedFullSchemas+    ]++-- | Cache configuration+data CacheConfig = CacheConfig+  { cachePath :: FilePath  -- ^ Path to cache file+  }+  deriving stock (Show, Eq)++-- ============================================================================+-- Configuration+-- ============================================================================++-- | Default cache configuration+-- Uses ~/.cache/symbols/schema.json on Linux/macOS+defaultCacheConfig :: IO CacheConfig+defaultCacheConfig = do+  path <- defaultCachePath+  pure CacheConfig+    { cachePath = path+    }++-- | Get the default cache file path+-- Respects XDG_CACHE_HOME on Linux/macOS+defaultCachePath :: IO FilePath+defaultCachePath = do+  cacheDir <- getXdgDirectory XdgCache "symbols"+  pure $ cacheDir </> "schema.json"++-- ============================================================================+-- Cache Operations+-- ============================================================================++-- | Load cached schema from disk+loadCache :: FilePath -> IO (Maybe CachedSchema)+loadCache path = do+  exists <- doesFileExist path+  if not exists+    then pure Nothing+    else do+      result <- (Just <$> LBS.readFile path) `catch` \(_ :: SomeException) -> pure Nothing+      case result of+        Nothing -> pure Nothing+        Just bs -> case eitherDecode bs of+          Left _err -> pure Nothing+          Right cached -> pure (Just cached)++-- | Save schema to cache+saveCache :: FilePath -> CachedSchema -> IO ()+saveCache path cached = do+  createDirectoryIfMissing True (takeDirectory path)+  LBS.writeFile path (encode cached)+    `catch` \(_ :: SomeException) -> pure ()  -- Silently fail on write errors++-- | Check if a cached schema is still fresh (hash matches current hash)+isFresh :: Text -> CachedSchema -> Bool+isFresh currentHash cached = cachedHash cached == currentHash++-- | Load schema with caching+--+-- 1. If forceRefresh is True, skip cache and fetch fresh+-- 2. Otherwise, fetch current hash and compare with cached hash+-- 3. If hashes match, return cached schema+-- 4. Otherwise, fetch fresh schema and cache it with new hash+loadSchemaWithCache+  :: Bool                                       -- ^ Force refresh (--refresh flag)+  -> CacheConfig                                -- ^ Cache configuration+  -> IO (Either Text Text)                      -- ^ Fetch plexus hash+  -> IO (Either Text PlexusSchema)              -- ^ Fetch plexus schema+  -> (Text -> IO (Maybe ActivationFullSchema))  -- ^ Fetch full schema for a namespace+  -> IO (Either Text CachedSchema)+loadSchemaWithCache forceRefresh config fetchHash fetchSchema fetchFullSchema = do+  if forceRefresh+    then fetchAndCache+    else do+      -- Fetch current hash+      hashResult <- fetchHash+      case hashResult of+        Left err -> do+          -- If we can't get hash, try to use cache anyway+          mCached <- loadCache (cachePath config)+          case mCached of+            Just cached -> pure $ Right cached+            Nothing -> pure $ Left err+        Right currentHash -> do+          -- Load cached schema+          mCached <- loadCache (cachePath config)+          case mCached of+            Nothing -> fetchAndCacheWithHash currentHash+            Just cached ->+              if isFresh currentHash cached+                then pure $ Right cached+                else fetchAndCacheWithHash currentHash+  where+    fetchAndCache = do+      hashResult <- fetchHash+      case hashResult of+        Left err -> pure $ Left err+        Right hash -> fetchAndCacheWithHash hash++    fetchAndCacheWithHash hash = do+      result <- fetchSchema+      case result of+        Left err -> do+          -- On fetch failure, try stale cache as fallback+          mCached <- loadCache (cachePath config)+          case mCached of+            Just cached -> pure $ Right cached  -- Use stale cache+            Nothing -> pure $ Left err  -- No cache available+        Right schema -> do+          -- Fetch full schemas for all activations+          fullSchemas <- fetchAllFullSchemas schema+          let cached = CachedSchema+                { cachedHash        = hash+                , cachedSchema      = schema+                , cachedFullSchemas = fullSchemas+                }+          saveCache (cachePath config) cached+          pure $ Right cached++    fetchAllFullSchemas schema = do+      let namespaces = map activationNamespace (schemaActivations schema)+      pairs <- mapM fetchPair namespaces+      pure $ Map.fromList [(ns, fs) | (ns, Just fs) <- pairs]++    fetchPair ns = do+      mSchema <- fetchFullSchema ns+      pure (ns, mSchema)++-- ============================================================================+-- Full Schema Lookup+-- ============================================================================++-- | Look up a method schema from the cached full schemas+-- Given "arbor_tree_create", looks up "arbor" full schema and finds "tree_create" method+lookupMethodSchema :: CachedSchema -> Text -> Maybe MethodSchemaInfo+lookupMethodSchema cached fullMethod = do+  -- Split "arbor_tree_create" into ("arbor", "tree_create")+  let (ns, method) = splitMethod fullMethod+  -- Look up the full schema for this namespace+  fullSchema <- Map.lookup ns (cachedFullSchemas cached)+  -- Find the matching method+  findMethod method (fullSchemaMethods fullSchema)++-- | Split "namespace_method" into (namespace, method)+-- e.g., "arbor_tree_create" -> ("arbor", "tree_create")+splitMethod :: Text -> (Text, Text)+splitMethod full =+  case T.break (== '_') full of+    (ns, rest) | not (T.null rest) -> (ns, T.drop 1 rest)+    _ -> (full, "")++-- | Find a method by name in a list of method schema infos+findMethod :: Text -> [MethodSchemaInfo] -> Maybe MethodSchemaInfo+findMethod name = find (\m -> methodInfoName m == name)+  where+    find f = foldr (\x acc -> if f x then Just x else acc) Nothing
+ src/Plexus/Schema/Recursive.hs view
@@ -0,0 +1,218 @@+-- | Shallow plugin schema for Plexus RPC+--+-- = Design+--+-- The schema is shallow: children are summaries (namespace, description, hash),+-- not full schemas. Full child schemas are fetched on demand when navigating.+--+-- This matches the coalgebraic design:+-- - Rust side: unfolds plugin structure on demand (anamorphism)+-- - Wire format: one layer at a time (shallow schema)+-- - Haskell side: folds/consumes structure (catamorphism over fetched data)+--+-- = The Functor (Conceptual)+--+-- @+-- F : Set → Set+-- F(X) = Namespace × Version × Description × Hash × [Method] × Maybe [X]+-- @+--+-- On the wire, X = ChildSummary (a reference). Resolution is lazy.+--+-- = Category Properties+--+-- The plugin system forms a free category:+-- - Objects: Schemas (identified by hash)+-- - Morphisms: Paths (sequences of child references)+-- - Identity: Empty path+-- - Composition: Path concatenation+module Plexus.Schema.Recursive+  ( -- * Core Types+    PluginSchema(..)+  , MethodSchema(..)+  , ChildSummary(..)+  , PluginHash+  , SchemaResult(..)++    -- * Queries+  , isHubActivation+  , isLeafActivation+  , pluginMethods+  , pluginChildren+  , childNamespaces++    -- * JSON Parsing+  , parsePluginSchema+  , parseSchemaResult+  ) where++import Control.Applicative ((<|>))++import Data.Aeson+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import qualified Data.Text as T+import GHC.Generics (Generic)++-- ============================================================================+-- Core Types+-- ============================================================================++-- | Content hash for cache invalidation+type PluginHash = Text++-- | Summary of a child plugin (shallow - no methods or nested children)+--+-- This is a reference to a child, not the full schema. To get the full+-- schema, fetch it via @{path}.schema@ RPC call.+data ChildSummary = ChildSummary+  { csNamespace   :: Text+  , csDescription :: Text+  , csHash        :: PluginHash+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON ChildSummary where+  parseJSON = withObject "ChildSummary" $ \o -> ChildSummary+    <$> o .: "namespace"+    <*> o .: "description"+    <*> o .: "hash"++instance ToJSON ChildSummary where+  toJSON ChildSummary{..} = object+    [ "namespace"   .= csNamespace+    , "description" .= csDescription+    , "hash"        .= csHash+    ]++-- | Schema for a single method+data MethodSchema = MethodSchema+  { methodName            :: Text+  , methodDescription     :: Text+  , methodHash            :: PluginHash+  , methodParams          :: Maybe Value  -- ^ JSON Schema for params+  , methodReturns         :: Maybe Value  -- ^ JSON Schema for return events+  , methodStreaming        :: Bool         -- ^ True if method streams multiple events+  , methodBidirectional   :: Bool         -- ^ True if method uses a bidirectional channel+  , methodRequestType     :: Maybe Value  -- ^ JSON Schema for the server→client request type (when bidirectional)+  , methodResponseType    :: Maybe Value  -- ^ JSON Schema for the client→server response type (when bidirectional)+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON MethodSchema where+  parseJSON = withObject "MethodSchema" $ \o -> MethodSchema+    <$> o .: "name"+    <*> o .: "description"+    <*> o .: "hash"+    <*> o .:? "params"+    <*> o .:? "returns"+    <*> o .:? "streaming"      .!= False+    <*> o .:? "bidirectional"  .!= False+    <*> o .:? "request_type"+    <*> o .:? "response_type"++instance ToJSON MethodSchema where+  toJSON MethodSchema{..} = object+    [ "name"           .= methodName+    , "description"    .= methodDescription+    , "hash"           .= methodHash+    , "params"         .= methodParams+    , "returns"        .= methodReturns+    , "streaming"      .= methodStreaming+    , "bidirectional"  .= methodBidirectional+    , "request_type"   .= methodRequestType+    , "response_type"  .= methodResponseType+    ]++-- | Shallow plugin schema (what we receive from {backend}.schema)+--+-- Children are summaries only - fetch full schema on-demand when navigating.+-- This is the wire format: one layer of observation at a time.+data PluginSchema = PluginSchema+  { psNamespace       :: Text+  , psVersion         :: Text+  , psDescription     :: Text+  , psLongDescription :: Maybe Text  -- ^ Extended description (no word limit)+  , psHash            :: PluginHash+  , psMethods         :: [MethodSchema]+  , psChildren        :: Maybe [ChildSummary]  -- ^ Nothing = leaf, Just = hub activation+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON PluginSchema where+  parseJSON = withObject "PluginSchema" $ \o -> PluginSchema+    <$> o .: "namespace"+    <*> o .: "version"+    <*> o .: "description"+    <*> o .:? "long_description"+    <*> o .: "hash"+    <*> o .:? "methods" .!= []+    <*> o .:? "children"++instance ToJSON PluginSchema where+  toJSON PluginSchema{..} = object+    [ "namespace"        .= psNamespace+    , "version"          .= psVersion+    , "description"      .= psDescription+    , "long_description" .= psLongDescription+    , "hash"             .= psHash+    , "methods"          .= psMethods+    , "children"         .= psChildren+    ]++-- | Result of a schema query - can be either a full plugin or just a method+data SchemaResult+  = SchemaPlugin PluginSchema+  | SchemaMethod MethodSchema+  deriving stock (Show, Eq)++instance FromJSON SchemaResult where+  parseJSON v =+    -- Try PluginSchema first (has "namespace" field)+    (SchemaPlugin <$> parseJSON v) <|>+    -- Fall back to MethodSchema (has "name" field)+    (SchemaMethod <$> parseJSON v)++instance ToJSON SchemaResult where+  toJSON (SchemaPlugin p) = toJSON p+  toJSON (SchemaMethod m) = toJSON m++-- ============================================================================+-- Basic Queries+-- ============================================================================++-- | Is this a hub activation (has children)?+isHubActivation :: PluginSchema -> Bool+isHubActivation = maybe False (not . null) . psChildren++-- | Is this a leaf activation (no children)?+isLeafActivation :: PluginSchema -> Bool+isLeafActivation = not . isHubActivation++-- | Get methods (alias for psMethods)+pluginMethods :: PluginSchema -> [MethodSchema]+pluginMethods = psMethods++-- | Get child summaries (empty list if leaf)+pluginChildren :: PluginSchema -> [ChildSummary]+pluginChildren = fromMaybe [] . psChildren++-- | Get child namespace names+childNamespaces :: PluginSchema -> [Text]+childNamespaces = map csNamespace . pluginChildren++-- ============================================================================+-- JSON Parsing Helpers+-- ============================================================================++-- | Parse a PluginSchema from the schema event content+parsePluginSchema :: Value -> Either Text PluginSchema+parsePluginSchema val = case fromJSON val of+  Success schema -> Right schema+  Error err -> Left $ T.pack err++-- | Parse a SchemaResult (plugin or method) from schema event content+parseSchemaResult :: Value -> Either Text SchemaResult+parseSchemaResult val = case fromJSON val of+  Success result -> Right result+  Error err -> Left $ T.pack err
+ src/Plexus/Transport.hs view
@@ -0,0 +1,191 @@+-- | Low-level transport for Substrate RPC calls+--+-- Pure IO functions for WebSocket communication with the Substrate backend.+-- All calls go through '<backend>.call' for routing.+module Plexus.Transport+  ( -- * RPC Calls (collected)+    rpcCall+  , rpcCallWith++    -- * RPC Calls (streaming)+  , rpcCallStreaming+  , invokeMethodStreaming++    -- * Bidirectional Response+  , sendBidirectionalResponse++    -- * Schema Fetching+  , fetchSchemaAt+  , fetchMethodSchemaAt+  , extractSchema+  , extractSchemaResult++    -- * Method Invocation (collected)+  , invokeMethod+  , invokeRaw+  ) where++import Control.Exception (SomeException, IOException, catch, fromException)+import qualified Control.Exception as E+import Data.Aeson hiding (Error)+import Data.Text (Text)+import qualified Data.Text as T+import qualified Streaming.Prelude as S+import qualified Network.Socket as NS++import Plexus.Client (SubstrateConfig(..), connect, disconnect, substrateRpc, defaultConfig)+import Plexus.Types (PlexusStreamItem(..), TransportError(..), Response(..), StandardResponse)+import Plexus.Schema.Recursive (PluginSchema, MethodSchema, SchemaResult(..), parsePluginSchema, parseSchemaResult)++-- | Low-level RPC call with default localhost config+rpcCall :: Text -> Text -> Value -> IO (Either TransportError [PlexusStreamItem])+rpcCall backend = rpcCallWith (defaultConfig backend)++-- | Low-level RPC call with custom config+rpcCallWith :: SubstrateConfig -> Text -> Value -> IO (Either TransportError [PlexusStreamItem])+rpcCallWith cfg method params = do+  result <- (Right <$> doCallInner cfg method params)+    `catch` categorizeException cfg+  pure result++-- | Categorize exceptions into typed TransportError+categorizeException :: SubstrateConfig -> SomeException -> IO (Either TransportError a)+categorizeException cfg e+  -- Check for connection refused (most common)+  | Just ioErr <- fromException e :: Maybe IOException = do+      let errMsg = show ioErr+      let host = T.pack $ substrateHost cfg+      let port = substratePort cfg+      pure $ Left $ if "refused" `isInfixOf` errMsg || "ECONNREFUSED" `isInfixOf` errMsg+        then ConnectionRefused host port+        else if "timeout" `isInfixOf` errMsg || "ETIMEDOUT" `isInfixOf` errMsg+          then ConnectionTimeout host port+          else NetworkError (T.pack errMsg)++  -- Catch protocol/parse errors+  | otherwise =+      let errMsg = show e+          host = T.pack $ substrateHost cfg+          port = substratePort cfg+      in pure $ Left $ if "protocol" `isInfixOf` errMsg || "parse" `isInfixOf` errMsg+        then ProtocolError (T.pack errMsg)+        else NetworkError (T.pack errMsg)++  where+    isInfixOf = T.isInfixOf `on` (T.toLower . T.pack)+    on f g x y = f (g x) (g y)++doCallInner :: SubstrateConfig -> Text -> Value -> IO [PlexusStreamItem]+doCallInner cfg method params = do+  conn <- connect cfg+  items <- S.toList_ $ substrateRpc conn method params+  disconnect conn+  pure items++-- | Streaming RPC call - invokes callback for each item as it arrives+rpcCallStreaming :: SubstrateConfig -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO (Either TransportError ())+rpcCallStreaming cfg method params onItem = do+  result <- (Right <$> doCallStreaming cfg method params onItem)+    `catch` categorizeException cfg+  pure result++doCallStreaming :: SubstrateConfig -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO ()+doCallStreaming cfg method params onItem = do+  conn <- connect cfg+  S.mapM_ onItem $ substrateRpc conn method params+  disconnect conn++-- | Streaming method invocation+invokeMethodStreaming :: SubstrateConfig -> [Text] -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO (Either TransportError ())+invokeMethodStreaming cfg namespacePath method params onItem = do+  let backend = substrateBackend cfg+  let fullPath = if null namespacePath then [backend] else namespacePath+  let dotPath = T.intercalate "." (fullPath ++ [method])+  let callParams = object ["method" .= dotPath, "params" .= params]+  rpcCallStreaming cfg (backend <> ".call") callParams onItem++-- | Fetch schema at a specific path+-- Empty path = root (<backend>.schema)+-- Non-empty path = child schema (e.g., ["solar", "earth"] -> solar.earth.schema)+fetchSchemaAt :: SubstrateConfig -> [Text] -> IO (Either TransportError PluginSchema)+fetchSchemaAt cfg path = do+  let backend = substrateBackend cfg+  let schemaMethod = if null path+        then backend <> ".schema"+        else T.intercalate "." path <> ".schema"+  result <- rpcCallWith cfg (backend <> ".call") (object ["method" .= schemaMethod])+  case result of+    Left transportErr -> pure $ Left transportErr+    Right items -> case extractSchema items of+      Left parseErr -> pure $ Left $ ProtocolError parseErr  -- Parse errors are protocol errors+      Right schema -> pure $ Right schema++-- | Extract PluginSchema from stream items+-- Returns Either Text for application-level errors (not transport)+extractSchema :: [PlexusStreamItem] -> Either Text PluginSchema+extractSchema items =+  case [dat | StreamData _ _ ct dat <- items, ".schema" `T.isSuffixOf` ct] of+    (dat:_) -> parsePluginSchema dat+    [] -> case [err | StreamError _ _ err _ <- items] of+      (err:_) -> Left err+      [] -> Left "No schema in response"++-- | Fetch a specific method's schema+-- Uses the parameter-based query: plugin.schema with {"method": "name"}+fetchMethodSchemaAt :: SubstrateConfig -> [Text] -> Text -> IO (Either TransportError MethodSchema)+fetchMethodSchemaAt cfg path methodName = do+  let backend = substrateBackend cfg+  let schemaMethod = if null path+        then backend <> ".schema"+        else T.intercalate "." path <> ".schema"+  result <- rpcCallWith cfg (backend <> ".call") (object+    [ "method" .= schemaMethod+    , "params" .= object ["method" .= methodName]+    ])+  case result of+    Left transportErr -> pure $ Left transportErr+    Right items -> case extractSchemaResult items of+      Left parseErr -> pure $ Left $ ProtocolError parseErr+      Right (SchemaMethod m) -> pure $ Right m+      Right (SchemaPlugin _) -> pure $ Left $ ProtocolError "Expected method schema, got plugin schema"++-- | Extract SchemaResult (plugin or method) from stream items+extractSchemaResult :: [PlexusStreamItem] -> Either Text SchemaResult+extractSchemaResult items =+  case [dat | StreamData _ _ ct dat <- items, ".schema" `T.isSuffixOf` ct] of+    (dat:_) -> parseSchemaResult dat+    [] -> case [err | StreamError _ _ err _ <- items] of+      (err:_) -> Left err+      [] -> Left "No schema in response"++-- | Invoke a method and return stream items+invokeMethod :: SubstrateConfig -> [Text] -> Text -> Value -> IO (Either TransportError [PlexusStreamItem])+invokeMethod cfg namespacePath method params = do+  let backend = substrateBackend cfg+  let fullPath = if null namespacePath then [backend] else namespacePath+  let dotPath = T.intercalate "." (fullPath ++ [method])+  let callParams = object ["method" .= dotPath, "params" .= params]+  rpcCallWith cfg (backend <> ".call") callParams++-- | Invoke with raw method path+invokeRaw :: SubstrateConfig -> Text -> Value -> IO (Either TransportError [PlexusStreamItem])+invokeRaw cfg method params = do+  let backend = substrateBackend cfg+  let callParams = object ["method" .= method, "params" .= params]+  rpcCallWith cfg (backend <> ".call") callParams++-- | Send a response back to the server for a bidirectional request+-- Uses {backend}.respond RPC method+sendBidirectionalResponse :: SubstrateConfig -> Text -> Response Value -> IO (Either TransportError ())+sendBidirectionalResponse cfg requestId response = do+  let backend = substrateBackend cfg+  let respondParams = object+        [ "request_id" .= requestId+        , "response" .= response+        ]+  result <- rpcCallWith cfg (backend <> ".respond") respondParams+  case result of+    Left transportErr -> pure $ Left transportErr+    Right items -> case [err | StreamError _ _ err _ <- items] of+      (err:_) -> pure $ Left $ ProtocolError err+      [] -> pure $ Right ()
+ src/Plexus/Types.hs view
@@ -0,0 +1,590 @@+-- | JSON-RPC types for communicating with Plexus RPC servers+module Plexus.Types+  ( -- * JSON-RPC Protocol+    RpcRequest(..)+  , RpcResponse(..)+  , RpcError(..)+  , SubscriptionNotification(..)+  , SubNotifParams(..)+  , RequestId(..)+  , SubscriptionId(..)++    -- * Plexus Stream Types+  , PlexusStreamItem(..)+  , Provenance(..)+  , StreamMetadata(..)+  , GuidanceErrorType(..)+  , GuidanceSuggestion(..)+  , TransportError(..)++    -- * Bidirectional Types+  , SelectOption(..)+  , Request(..)+  , StandardRequest+  , Response(..)+  , StandardResponse++    -- * Helpers+  , mkSubscribeRequest+  , mkUnsubscribeRequest+  , getPlexusHash+  ) where++import Data.Aeson+import Data.Aeson.Types (Parser)+import qualified Data.ByteString.Lazy as LBS+import Data.Maybe (catMaybes)+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import GHC.Generics (Generic)++-- | Request ID for JSON-RPC calls+newtype RequestId = RequestId { unRequestId :: Int }+  deriving stock (Show, Eq, Ord)+  deriving newtype (ToJSON, FromJSON)++-- | Subscription ID returned by the server (can be string or number)+newtype SubscriptionId = SubscriptionId { unSubscriptionId :: Text }+  deriving stock (Show, Eq, Ord)+  deriving newtype (ToJSON)++instance FromJSON SubscriptionId where+  parseJSON (String s) = pure $ SubscriptionId s+  parseJSON v =+    -- Store raw JSON representation as the ID+    pure $ SubscriptionId $ T.decodeUtf8 $ LBS.toStrict $ encode v++-- | JSON-RPC 2.0 request+data RpcRequest = RpcRequest+  { rpcReqJsonrpc :: Text           -- ^ Always "2.0"+  , rpcReqMethod  :: Text           -- ^ Method name (e.g., "bash_execute")+  , rpcReqParams  :: Value          -- ^ Parameters (array or object)+  , rpcReqId      :: RequestId      -- ^ Request identifier+  }+  deriving stock (Show, Eq, Generic)++instance ToJSON RpcRequest where+  toJSON RpcRequest{..} = object+    [ "jsonrpc" .= rpcReqJsonrpc+    , "method"  .= rpcReqMethod+    , "params"  .= rpcReqParams+    , "id"      .= rpcReqId+    ]++-- | JSON-RPC 2.0 response+data RpcResponse+  = RpcSuccess+      { rpcRespId     :: RequestId+      , rpcRespResult :: Value+      }+  | RpcError+      { rpcRespId    :: RequestId+      , rpcRespError :: RpcError+      }+  deriving stock (Show, Eq, Generic)++instance FromJSON RpcResponse where+  parseJSON = withObject "RpcResponse" $ \o -> do+    rid <- o .: "id"+    mResult <- o .:? "result"+    mError  <- o .:? "error"+    case (mResult, mError) of+      (Just r, Nothing) -> pure $ RpcSuccess rid r+      (Nothing, Just e) -> pure $ RpcError rid e+      _ -> fail "Expected either 'result' or 'error' field"++-- | JSON-RPC error object+data RpcError = RpcErrorObj+  { errCode    :: Int+  , errMessage :: Text+  , errData    :: Maybe Value+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON RpcError where+  parseJSON = withObject "RpcError" $ \o -> RpcErrorObj+    <$> o .: "code"+    <*> o .: "message"+    <*> o .:? "data"++-- | Subscription notification from server+-- This is what we receive for each stream item+data SubscriptionNotification = SubscriptionNotification+  { subNotifJsonrpc :: Text+  , subNotifMethod  :: Text           -- ^ Usually "subscription"+  , subNotifParams  :: SubNotifParams+  }+  deriving stock (Show, Eq, Generic)++data SubNotifParams = SubNotifParams+  { subParamsSubscription :: SubscriptionId+  , subParamsResult       :: Value+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON SubscriptionNotification where+  parseJSON = withObject "SubscriptionNotification" $ \o -> SubscriptionNotification+    <$> o .: "jsonrpc"+    <*> o .: "method"+    <*> o .: "params"++instance FromJSON SubNotifParams where+  parseJSON = withObject "SubNotifParams" $ \o -> SubNotifParams+    <$> o .: "subscription"+    <*> o .: "result"++-- | Provenance tracking nested calls through activations+-- Now just a list of namespace segments (no wrapper object)+newtype Provenance = Provenance { segments :: [Text] }+  deriving stock (Show, Eq, Generic)++instance FromJSON Provenance where+  parseJSON v = case v of+    -- New format: just an array ["plexus", "echo"]+    Array arr -> Provenance <$> mapM parseJSON (foldr (:) [] arr)+    -- Legacy format: {"segments": [...]}+    Object o -> Provenance <$> o .: "segments"+    _ -> fail "Provenance: expected array or object"++instance ToJSON Provenance where+  toJSON (Provenance segs) = toJSON segs++-- | Metadata wrapper for stream results+data StreamMetadata = StreamMetadata+  { metaProvenance :: Provenance+  , metaPlexusHash :: Text+  , metaTimestamp  :: Integer+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON StreamMetadata where+  parseJSON = withObject "StreamMetadata" $ \o -> StreamMetadata+    <$> o .: "provenance"+    <*> o .: "plexus_hash"+    <*> o .: "timestamp"++instance ToJSON StreamMetadata where+  toJSON StreamMetadata{..} = object+    [ "provenance"  .= metaProvenance+    , "plexus_hash" .= metaPlexusHash+    , "timestamp"   .= metaTimestamp+    ]++-- | Error type from guidance events+data GuidanceErrorType+  = ActivationNotFound+      { errorActivation :: Text+      }+  | MethodNotFound+      { errorActivation :: Text+      , errorMethod     :: Text+      }+  | InvalidParams+      { errorMethod :: Text+      , errorReason :: Text+      }+  | TransportErrorGuidance+      { errorTransport :: TransportError+      }+  deriving stock (Show, Eq, Generic)++instance FromJSON GuidanceErrorType where+  parseJSON = withObject "GuidanceErrorType" $ \o -> do+    kind <- o .: "error_kind" :: Parser Text+    case kind of+      "activation_not_found" -> ActivationNotFound <$> o .: "activation"+      "method_not_found" -> MethodNotFound <$> o .: "activation" <*> o .: "method"+      "invalid_params" -> InvalidParams <$> o .: "method" <*> o .: "reason"+      "transport_error" -> TransportErrorGuidance <$> parseJSON (Object o)+      _ -> fail $ "Unknown error_kind: " <> T.unpack kind++instance ToJSON GuidanceErrorType where+  toJSON (ActivationNotFound activation) = object+    [ "error_kind" .= ("activation_not_found" :: Text)+    , "activation" .= activation+    ]+  toJSON (MethodNotFound activation method) = object+    [ "error_kind" .= ("method_not_found" :: Text)+    , "activation" .= activation+    , "method" .= method+    ]+  toJSON (InvalidParams method reason) = object+    [ "error_kind" .= ("invalid_params" :: Text)+    , "method" .= method+    , "reason" .= reason+    ]+  toJSON (TransportErrorGuidance transportErr) =+    -- Merge the transport error JSON with error_kind+    case toJSON transportErr of+      Object o -> Object o+      _ -> object ["error_kind" .= ("transport_error" :: Text)]++-- | Suggestion for next action from guidance events+data GuidanceSuggestion+  = CallPlexusSchema+  | CallActivationSchema+      { suggestionNamespace :: Text+      }+  | TryMethod+      { suggestionMethod       :: Text+      , suggestionExampleParams :: Maybe Value+      }+  | CustomGuidance+      { suggestionMessage :: Text+      }+  deriving stock (Show, Eq, Generic)++instance FromJSON GuidanceSuggestion where+  parseJSON = withObject "GuidanceSuggestion" $ \o -> do+    action <- o .: "action" :: Parser Text+    case action of+      "call_plexus_schema" -> pure CallPlexusSchema  -- Legacy: maps to {backend}.schema+      "call_activation_schema" -> CallActivationSchema <$> o .: "namespace"+      "try_method" -> TryMethod <$> o .: "method" <*> o .:? "example_params"+      "custom" -> CustomGuidance <$> o .: "message"+      _ -> fail $ "Unknown action: " <> T.unpack action++instance ToJSON GuidanceSuggestion where+  toJSON CallPlexusSchema = object ["action" .= ("call_plexus_schema" :: Text)]  -- Legacy: represents {backend}.schema+  toJSON (CallActivationSchema namespace) = object+    [ "action" .= ("call_activation_schema" :: Text)+    , "namespace" .= namespace+    ]+  toJSON (TryMethod method exampleParams) = object $+    [ "action" .= ("try_method" :: Text)+    , "method" .= method+    ] <> catMaybes [("example_params" .=) <$> exampleParams]+  toJSON (CustomGuidance message) = object+    [ "action" .= ("custom" :: Text)+    , "message" .= message+    ]++-- ============================================================================+-- Transport Errors+-- ============================================================================++-- | Strongly-typed transport-level errors+-- These occur before reaching the backend (connection, network, protocol issues)+data TransportError+  = ConnectionRefused+      { transportHost :: Text+      , transportPort :: Int+      }+  | ConnectionTimeout+      { transportHost :: Text+      , transportPort :: Int+      }+  | ProtocolError+      { transportMessage :: Text+      }+  | NetworkError+      { transportMessage :: Text+      }+  deriving stock (Show, Eq, Generic)++instance ToJSON TransportError where+  toJSON (ConnectionRefused host port) = object+    [ "error_kind" .= ("connection_refused" :: Text)+    , "host" .= host+    , "port" .= port+    ]+  toJSON (ConnectionTimeout host port) = object+    [ "error_kind" .= ("connection_timeout" :: Text)+    , "host" .= host+    , "port" .= port+    ]+  toJSON (ProtocolError msg) = object+    [ "error_kind" .= ("protocol_error" :: Text)+    , "message" .= msg+    ]+  toJSON (NetworkError msg) = object+    [ "error_kind" .= ("network_error" :: Text)+    , "message" .= msg+    ]++instance FromJSON TransportError where+  parseJSON = withObject "TransportError" $ \o -> do+    kind <- o .: "error_kind" :: Parser Text+    case kind of+      "connection_refused" -> ConnectionRefused <$> o .: "host" <*> o .: "port"+      "connection_timeout" -> ConnectionTimeout <$> o .: "host" <*> o .: "port"+      "protocol_error" -> ProtocolError <$> o .: "message"+      "network_error" -> NetworkError <$> o .: "message"+      _ -> fail $ "Unknown transport error_kind: " <> T.unpack kind++-- ============================================================================+-- Bidirectional Types (Request/Response)+-- ============================================================================++-- | Option for select requests, parameterized over value type+data SelectOption t = SelectOption+  { optionValue       :: t+  , optionLabel       :: Text+  , optionDescription :: Maybe Text+  }+  deriving (Show, Eq, Generic)++instance (FromJSON t) => FromJSON (SelectOption t) where+  parseJSON = withObject "SelectOption" $ \o -> SelectOption+    <$> o .: "value"+    <*> o .: "label"+    <*> o .:? "description"++instance (ToJSON t) => ToJSON (SelectOption t) where+  toJSON SelectOption{..} = object $+    [ "value" .= optionValue, "label" .= optionLabel ]+    <> catMaybes [("description" .=) <$> optionDescription]++-- | Parameterized request type for bidirectional communication+data Request t+  = Confirm  { confirmMessage :: Text, confirmDefault :: Maybe Bool }+  | Prompt   { promptMessage :: Text, promptDefault :: Maybe t, promptPlaceholder :: Maybe Text }+  | Select   { selectMessage :: Text, selectOptions :: [SelectOption t], selectMulti :: Bool }+  | CustomRequest { customRequestData :: t }+  deriving (Show, Eq, Generic)++-- | Type alias for backwards compatibility: requests with JSON values+type StandardRequest = Request Value++instance (FromJSON t) => FromJSON (Request t) where+  parseJSON = withObject "Request" $ \o -> do+    typ <- o .: "type" :: Parser Text+    case typ of+      "confirm" -> Confirm+        <$> o .: "message"+        <*> o .:? "default"+      "prompt" -> Prompt+        <$> o .: "message"+        <*> o .:? "default"+        <*> o .:? "placeholder"+      "select" -> Select+        <$> o .: "message"+        <*> o .: "options"+        <*> o .:? "multi_select" .!= False+      "custom" -> CustomRequest <$> o .: "data"+      _ -> fail $ "Unknown request type: " <> T.unpack typ++instance (ToJSON t) => ToJSON (Request t) where+  toJSON (Confirm msg def) = object $+    [ "type" .= ("confirm" :: Text), "message" .= msg ]+    <> catMaybes [("default" .=) <$> def]+  toJSON (Prompt msg def placeholder) = object $+    [ "type" .= ("prompt" :: Text), "message" .= msg ]+    <> catMaybes [("default" .=) <$> def, ("placeholder" .=) <$> placeholder]+  toJSON (Select msg opts multi) = object+    [ "type" .= ("select" :: Text), "message" .= msg+    , "options" .= opts, "multi_select" .= multi ]+  toJSON (CustomRequest d) = object ["type" .= ("custom" :: Text), "data" .= d]++-- | Parameterized response type for bidirectional communication+--+-- Note: The 'Value' constructor corresponds to JSON @"type":"text"@ for+-- wire compatibility with the Rust implementation.+data Response t+  = Confirmed { confirmedValue :: Bool }+  | Value     { responseValue :: t }    -- ^ wire tag: "text"+  | Selected  { selectedValues :: [t] }+  | CustomResponse { customResponseData :: t }+  | Cancelled+  deriving (Show, Eq, Generic)++-- | Type alias for backwards compatibility: responses with JSON values+type StandardResponse = Response Value++instance (FromJSON t) => FromJSON (Response t) where+  parseJSON = withObject "Response" $ \o -> do+    typ <- o .: "type" :: Parser Text+    case typ of+      "confirmed" -> Confirmed <$> o .: "value"+      "text"      -> Value <$> o .: "value"+      "selected"  -> Selected <$> o .: "values"+      "custom"    -> CustomResponse <$> o .: "data"+      "cancelled" -> pure Cancelled+      _ -> fail $ "Unknown response type: " <> T.unpack typ++instance (ToJSON t) => ToJSON (Response t) where+  toJSON (Confirmed v)       = object ["type" .= ("confirmed" :: Text), "value" .= v]+  toJSON (Value v)           = object ["type" .= ("text" :: Text), "value" .= v]+  toJSON (Selected vs)       = object ["type" .= ("selected" :: Text), "values" .= vs]+  toJSON (CustomResponse d)  = object ["type" .= ("custom" :: Text), "data" .= d]+  toJSON Cancelled           = object ["type" .= ("cancelled" :: Text)]++-- ============================================================================+-- Plexus Stream Items+-- ============================================================================++-- | Unified stream item from Plexus RPC+--+-- New format uses metadata wrapper:+-- @{"type":"data","metadata":{...},"content_type":"...","content":{...}}@+data PlexusStreamItem+  = StreamProgress+      { itemPlexusHash  :: Text+      , itemProvenance  :: Provenance+      , itemMessage     :: Text+      , itemPercentage  :: Maybe Double+      }+  | StreamData+      { itemPlexusHash  :: Text+      , itemProvenance  :: Provenance+      , itemContentType :: Text+      , itemData        :: Value+      }+  | StreamGuidance+      { itemPlexusHash        :: Text+      , itemProvenance        :: Provenance+      , itemErrorType         :: GuidanceErrorType+      , itemSuggestion        :: GuidanceSuggestion+      , itemAvailableMethods  :: Maybe [Text]+      , itemMethodSchema      :: Maybe Value+      }+  | StreamError+      { itemPlexusHash  :: Text+      , itemProvenance  :: Provenance+      , itemError       :: Text+      , itemRecoverable :: Bool+      }+  | StreamDone+      { itemPlexusHash :: Text+      , itemProvenance :: Provenance+      }+  | StreamRequest+      { itemPlexusHash  :: Text+      , itemProvenance  :: Provenance+      , itemRequestId   :: Text+      , itemRequestData :: Request Value+      , itemTimeoutMs   :: Int+      }+  deriving stock (Show, Eq, Generic)++instance FromJSON PlexusStreamItem where+  parseJSON = withObject "PlexusStreamItem" $ \o -> do+    typ <- o .: "type" :: Parser Text+    -- New format: metadata is a nested object+    mMeta <- o .:? "metadata" :: Parser (Maybe StreamMetadata)+    case mMeta of+      Just meta -> parseWithMetadata typ meta o+      Nothing -> parseLegacy typ o++    where+      -- New format: metadata wrapper+      parseWithMetadata :: Text -> StreamMetadata -> Object -> Parser PlexusStreamItem+      parseWithMetadata typ meta o = do+        let hash = metaPlexusHash meta+            prov = metaProvenance meta+        case typ of+          "progress" -> StreamProgress hash prov+            <$> o .: "message"+            <*> o .:? "percentage"+          "data" -> StreamData hash prov+            <$> o .: "content_type"+            <*> o .: "content"+          "guidance" -> StreamGuidance hash prov+            <$> o .: "error_type"+            <*> o .: "suggestion"+            <*> o .:? "available_methods"+            <*> o .:? "method_schema"+          "error" -> StreamError hash prov+            <$> o .: "message"+            <*> o .:? "recoverable" .!= False+          "done" -> pure $ StreamDone hash prov+          "request" -> StreamRequest hash prov+            <$> o .: "request_id"+            <*> o .: "request"+            <*> o .: "timeout_ms"+          _ -> fail $ "Unknown event type: " <> T.unpack typ++      -- Legacy format: flat structure+      parseLegacy :: Text -> Object -> Parser PlexusStreamItem+      parseLegacy typ o = do+        hash <- o .: "plexus_hash"+        case typ of+          "progress" -> StreamProgress hash+            <$> o .: "provenance"+            <*> o .: "message"+            <*> o .:? "percentage"+          "data" -> StreamData hash+            <$> o .: "provenance"+            <*> o .: "content_type"+            <*> o .: "data"+          "guidance" -> StreamGuidance hash+            <$> o .: "provenance"+            <*> o .: "error_type"+            <*> o .: "suggestion"+            <*> o .:? "available_methods"+            <*> o .:? "method_schema"+          "error" -> StreamError hash+            <$> o .: "provenance"+            <*> o .: "error"+            <*> o .: "recoverable"+          "done" -> StreamDone hash+            <$> o .: "provenance"+          "request" -> StreamRequest hash+            <$> o .: "provenance"+            <*> o .: "request_id"+            <*> o .: "request"+            <*> o .: "timeout_ms"+          _ -> fail $ "Unknown event type: " <> T.unpack typ++instance ToJSON PlexusStreamItem where+  toJSON (StreamProgress hash prov msg pct) = object+    [ "type" .= ("progress" :: Text)+    , "metadata" .= StreamMetadata prov hash 0+    , "message" .= msg+    , "percentage" .= pct+    ]+  toJSON (StreamData hash prov ct dat) = object+    [ "type" .= ("data" :: Text)+    , "metadata" .= StreamMetadata prov hash 0+    , "content_type" .= ct+    , "content" .= dat+    ]+  toJSON (StreamGuidance hash prov errorType suggestion availMethods methodSchema) = object $+    [ "type" .= ("guidance" :: Text)+    , "metadata" .= StreamMetadata prov hash 0+    , "error_type" .= errorType+    , "suggestion" .= suggestion+    ] <> catMaybes+    [ ("available_methods" .=) <$> availMethods+    , ("method_schema" .=) <$> methodSchema+    ]+  toJSON (StreamError hash prov err rec) = object+    [ "type" .= ("error" :: Text)+    , "metadata" .= StreamMetadata prov hash 0+    , "message" .= err+    , "recoverable" .= rec+    ]+  toJSON (StreamDone hash prov) = object+    [ "type" .= ("done" :: Text)+    , "metadata" .= StreamMetadata prov hash 0+    ]+  toJSON (StreamRequest hash prov reqId reqData timeout) = object+    [ "type" .= ("request" :: Text)+    , "metadata" .= StreamMetadata prov hash 0+    , "request_id" .= reqId+    , "request" .= reqData+    , "timeout_ms" .= timeout+    ]++-- | Create a subscription request+mkSubscribeRequest :: RequestId -> Text -> Value -> RpcRequest+mkSubscribeRequest rid method params = RpcRequest+  { rpcReqJsonrpc = "2.0"+  , rpcReqMethod  = method+  , rpcReqParams  = params+  , rpcReqId      = rid+  }++-- | Create an unsubscribe request+mkUnsubscribeRequest :: RequestId -> Text -> SubscriptionId -> RpcRequest+mkUnsubscribeRequest rid unsubMethod subId = RpcRequest+  { rpcReqJsonrpc = "2.0"+  , rpcReqMethod  = unsubMethod+  , rpcReqParams  = toJSON [unSubscriptionId subId]+  , rpcReqId      = rid+  }++-- | Extract the Plexus RPC hash from a stream item+getPlexusHash :: PlexusStreamItem -> Text+getPlexusHash = itemPlexusHash