diff --git a/plexus-protocol.cabal b/plexus-protocol.cabal
new file mode 100644
--- /dev/null
+++ b/plexus-protocol.cabal
@@ -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
diff --git a/src/Plexus.hs b/src/Plexus.hs
new file mode 100644
--- /dev/null
+++ b/src/Plexus.hs
@@ -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
diff --git a/src/Plexus/Client.hs b/src/Plexus/Client.hs
new file mode 100644
--- /dev/null
+++ b/src/Plexus/Client.hs
@@ -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
diff --git a/src/Plexus/Schema.hs b/src/Plexus/Schema.hs
new file mode 100644
--- /dev/null
+++ b/src/Plexus/Schema.hs
@@ -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
diff --git a/src/Plexus/Schema/Cache.hs b/src/Plexus/Schema/Cache.hs
new file mode 100644
--- /dev/null
+++ b/src/Plexus/Schema/Cache.hs
@@ -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
diff --git a/src/Plexus/Schema/Recursive.hs b/src/Plexus/Schema/Recursive.hs
new file mode 100644
--- /dev/null
+++ b/src/Plexus/Schema/Recursive.hs
@@ -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
diff --git a/src/Plexus/Transport.hs b/src/Plexus/Transport.hs
new file mode 100644
--- /dev/null
+++ b/src/Plexus/Transport.hs
@@ -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 ()
diff --git a/src/Plexus/Types.hs b/src/Plexus/Types.hs
new file mode 100644
--- /dev/null
+++ b/src/Plexus/Types.hs
@@ -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