diff --git a/legion.cabal b/legion.cabal
--- a/legion.cabal
+++ b/legion.cabal
@@ -2,7 +2,7 @@
 -- documentation, see http://haskell.org/cabal/users-guide/
 
 name:                legion
-version:             0.9.0.2
+version:             0.10.0.0
 synopsis:            Distributed, stateful, homogeneous microservice framework.
 description:         Legion is a framework for writing distributed,
                      homogeneous, stateful microservices in Haskell.
@@ -37,6 +37,7 @@
     Network.Legion.Index
     Network.Legion.KeySet
     Network.Legion.LIO
+    Network.Legion.Launch
     Network.Legion.Lift
     Network.Legion.PartitionKey
     Network.Legion.PartitionState
@@ -45,9 +46,9 @@
     Network.Legion.Runtime
     Network.Legion.Runtime.ConnectionManager
     Network.Legion.Runtime.PeerMessage
+    Network.Legion.Runtime.State
     Network.Legion.Settings
-    Network.Legion.StateMachine
-    Network.Legion.StateMachine.Monad
+    Network.Legion.SocketUtil
     Network.Legion.UUID
     Paths_legion
   -- other-extensions:
@@ -70,6 +71,7 @@
     network            >= 2.6.2.1  && < 2.7,
     scotty             >= 0.11.0   && < 0.12,
     scotty-resource    >= 0.1      && < 0.3,
+    stm                >= 2.4.4.1  && < 2.5,
     text               >= 1.2.2.0  && < 1.3,
     time               >= 1.6.0.1  && < 1.7,
     transformers       >= 0.3.0.0  && < 0.6,
diff --git a/src/Network/Legion.hs b/src/Network/Legion.hs
--- a/src/Network/Legion.hs
+++ b/src/Network/Legion.hs
@@ -84,11 +84,13 @@
 import Network.Legion.Index (Tag(Tag, unTag), IndexRecord(IndexRecord,
   irTag, irKey), SearchTag(SearchTag, stTag, stKey),
   Indexable(indexEntries))
+import Network.Legion.Launch (forkLegionary)
 import Network.Legion.PartitionKey (PartitionKey(K, unKey))
 import Network.Legion.PartitionState (PartitionPowerState)
 import Network.Legion.PowerState (Event(apply))
-import Network.Legion.Runtime (StartupMode(NewCluster, JoinCluster, Recover),
-  forkLegionary, Runtime, makeRequest, search)
+import Network.Legion.Runtime (Runtime, makeRequest, search)
+import Network.Legion.Runtime.State (StartupMode(NewCluster, JoinCluster,
+  Recover))
 import Network.Legion.Settings (RuntimeSettings(RuntimeSettings,
   adminHost, adminPort, peerBindAddr, joinBindAddr))
 
diff --git a/src/Network/Legion/Admin.hs b/src/Network/Legion/Admin.hs
--- a/src/Network/Legion/Admin.hs
+++ b/src/Network/Legion/Admin.hs
@@ -2,38 +2,31 @@
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE TemplateHaskell #-}
-{- |
-  This module contains the admin interface code.
--}
+
+{- | This module contains the admin interface code. -}
 module Network.Legion.Admin (
   runAdmin,
-  AdminMessage(..),
+  forkAdmin,
 ) where
 
+
 import Canteven.HTTP (requestLogging, logExceptionsAndContinue)
-import Control.Concurrent (forkIO, newChan, newEmptyMVar, writeChan,
-  putMVar, takeMVar, Chan)
-import Control.Monad (void)
-import Control.Monad.Logger (askLoggerIO, runLoggingT, logDebug)
+import Control.Monad.IO.Class (liftIO)
+import Control.Monad.Logger (askLoggerIO, runLoggingT, logDebug,
+  MonadLoggerIO)
 import Control.Monad.Trans.Class (lift)
-import Data.Conduit (Source)
 import Data.Default.Class (def)
-import Data.Map (Map)
-import Data.Set (Set)
 import Data.Text.Encoding (encodeUtf8)
 import Data.Text.Lazy (Text)
-import Data.Time (UTCTime)
 import Data.Version (showVersion)
 import Network.HTTP.Types (notFound404)
 import Network.Legion.Application (LegionConstraints)
-import Network.Legion.Conduit (chanToSource)
-import Network.Legion.Distribution (Peer)
-import Network.Legion.Index (IndexRecord)
+import Network.Legion.Fork (forkC)
 import Network.Legion.LIO (LIO)
-import Network.Legion.Lift (lift2)
 import Network.Legion.PartitionKey (PartitionKey(K), unKey)
-import Network.Legion.PartitionState (PartitionPowerState)
-import Network.Legion.StateMachine.Monad (NodeState)
+import Network.Legion.Runtime (Runtime, debugRuntimeState, getDivergent,
+  debugLocalPartitions, debugPartition, eject, debugIndex)
+import Network.Legion.Settings (RuntimeSettings, adminPort, adminHost)
 import Network.Wai (Middleware, modifyResponse)
 import Network.Wai.Handler.Warp (HostPreference, defaultSettings, Port,
   setHost, setPort)
@@ -42,63 +35,64 @@
 import Paths_legion (version)
 import Text.Read (readMaybe)
 import Web.Scotty.Resource.Trans (resource, get, delete)
-import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, ActionT,
-  param, middleware, status, json)
+import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, param,
+  middleware, status, json, text)
 import qualified Data.Map as Map
 import qualified Data.Text as T
+import qualified Data.Text.Lazy as TL
 
-{- |
-  Start the admin service in a background thread.
--}
-runAdmin :: (LegionConstraints e o s)
+{- | Fork the admin website in a background thread. -}
+forkAdmin :: (LegionConstraints e o s, MonadLoggerIO m)
+  => RuntimeSettings
+    {- ^ Settings and configuration of the legion framework. -}
+  -> Runtime e o s
+  -> m ()
+forkAdmin legionSettings runtime = do
+  logging <- askLoggerIO
+  liftIO . (`runLoggingT` logging) . forkC "admin thread" $
+    runAdmin (adminPort legionSettings) (adminHost legionSettings) runtime
+
+
+{- | Run the admin service. Does not return. -}
+runAdmin :: (LegionConstraints e o s, MonadLoggerIO io)
   => Port
   -> HostPreference
-  -> LIO (Source LIO (AdminMessage e o s))
-runAdmin addr host = do
-    logging <- askLoggerIO
-    chan <- lift newChan
-    void . lift . forkIO . (`runLoggingT` logging) $
-      let
-        website :: ScottyT Text LIO ()
-        website = do
-          middleware
-            $ requestLogging logging
-            . setServer
-            . logExceptionsAndContinue logging
+  -> Runtime e o s
+  -> io ()
+runAdmin addr host runtime = do
+  logging <- askLoggerIO
+  let
+    website :: ScottyT Text LIO ()
+    website = do
+      middleware
+        $ requestLogging logging
+        . setServer
+        . logExceptionsAndContinue logging
 
-          resource "/clusterstate" $
-            get $ json =<< send chan GetState
-          resource "/index" $
-            get $ json =<< send chan GetIndex
-          resource "/divergent" $
-            get $
-              json . Map.mapKeys show =<< send chan GetDivergent
-          resource "/partitions" $
-            get $
-              json . Map.mapKeys (show . toInteger . unKey) =<< send chan GetStates
-          resource "/partitions/:key" $
-            get $ do
-              key <- K . read <$> param "key"
-              json =<< send chan (GetPart key)
-          resource "/peers/:peer" $
-            delete $
-              readMaybe <$> param "peer" >>= \case
-                Nothing -> status notFound404
-                Just peer -> do
-                  lift . $(logDebug) . T.pack $ "Ejecting peer: " ++ show peer
-                  send chan (Eject peer)
+      resource "/rts" $
+        get $ text . TL.pack . show =<< debugRuntimeState runtime
+      resource "/index" $
+        get $ json =<< debugIndex runtime
+      resource "/divergent" $
+        get $
+          json . Map.mapKeys show =<< getDivergent runtime
+      resource "/partitions" $
+        get $
+          json . Map.mapKeys (show . toInteger . unKey)
+            =<< debugLocalPartitions runtime
+      resource "/partitions/:key" $
+        get $ do
+          key <- K . read <$> param "key"
+          json =<< debugPartition runtime key
+      resource "/peers/:peer" $
+        delete $
+          readMaybe <$> param "peer" >>= \case
+            Nothing -> status notFound404
+            Just peer -> do
+              lift . $(logDebug) . T.pack $ "Ejecting peer: " ++ show peer
+              eject runtime peer
 
-      in scottyOptsT (options addr host) (`runLoggingT` logging) website
-    return (chanToSource chan)
-  where
-    send
-      :: Chan (AdminMessage e o s)
-      -> ((a -> LIO ()) -> AdminMessage e o s)
-      -> ActionT Text LIO a
-    send chan msg = lift2 $ do
-      mvar <- newEmptyMVar
-      writeChan chan (msg (lift . putMVar mvar))
-      takeMVar mvar
+  scottyOptsT (options addr host) (`runLoggingT` logging) website
 
 
 {- | Build some warp settings based on the configured socket address. -}
@@ -125,25 +119,5 @@
     {- | The value of the @Server:@ header. -}
     serverValue =
       encodeUtf8 (T.pack ("legion-admin/" ++ showVersion version))
-
-
-{- |
-  The type of messages sent by the admin service.
--}
-data AdminMessage e o s
-  = GetState (NodeState e o s -> LIO ())
-  | GetPart PartitionKey (PartitionPowerState e o s -> LIO ())
-  | Eject Peer (() -> LIO ())
-  | GetIndex (Set IndexRecord -> LIO ())
-  | GetDivergent (Map Peer (Maybe UTCTime) -> LIO ())
-  | GetStates (Map PartitionKey (PartitionPowerState e o s) -> LIO ())
-
-instance Show (AdminMessage e o s) where
-  show (GetState _) = "(GetState _)"
-  show (GetPart k _) = "(GetPart " ++ show k ++ " _)"
-  show (Eject p _) = "(Eject " ++ show p ++ " _)"
-  show (GetIndex _) = "(GetIndex _)"
-  show (GetDivergent _) = "(GetDivergent _)"
-  show (GetStates _) = "(GetStates _)"
 
 
diff --git a/src/Network/Legion/Fork.hs b/src/Network/Legion/Fork.hs
--- a/src/Network/Legion/Fork.hs
+++ b/src/Network/Legion/Fork.hs
@@ -1,16 +1,22 @@
+{-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE TemplateHaskell #-}
+
 {- |
   This module holds `forkC`, because we use it in at  least two other modules.
 -}
 module Network.Legion.Fork (
-  forkC
+  forkC,
+  forkL,
+  ForkM(..),
 ) where
 
 import Control.Concurrent (forkIO)
-import Control.Exception (SomeException, try)
+import Control.Exception (SomeException)
 import Control.Monad (void)
-import Control.Monad.Logger (logError, askLoggerIO, runLoggingT)
-import Control.Monad.Trans.Class (lift)
+import Control.Monad.Catch (try, MonadCatch)
+import Control.Monad.IO.Class (liftIO)
+import Control.Monad.Logger (logError, askLoggerIO, runLoggingT,
+  MonadLoggerIO, LoggingT, MonadLoggerIO)
 import Data.Text (pack)
 import Network.Legion.LIO (LIO)
 import System.Exit (ExitCode(ExitFailure))
@@ -23,16 +29,15 @@
   we should crash the program instead of running in some kind of zombie broken
   state.
 -}
-forkC
-  :: String
+forkC :: (ForkM m, MonadCatch m, MonadLoggerIO m)
+  => String
     -- ^ The name of the critical thread, used for logging.
-  -> LIO ()
+  -> m ()
     -- ^ The IO to execute.
-  -> LIO ()
-forkC name io = do
-  logging <- askLoggerIO
-  lift . void . forkIO $ do
-    result <- try (runLoggingT io logging)
+  -> m ()
+forkC name action =
+  forkM $ do
+    result <- try action
     case result of
       Left err -> do
         let msg =
@@ -40,10 +45,34 @@
               ++ ". We are crashing the entire program because we can't "
               ++ "continue without this thread. The error was: "
               ++ show (err :: SomeException)
-        -- write the message to every place we can think of.
-        (`runLoggingT` logging) . $(logError) . pack $ msg
-        putStrLn msg
-        hPutStrLn stderr msg
-        exitImmediately (ExitFailure 1)
+        {- write the message to every place we can think of. -}
+        $(logError) . pack $ msg
+        liftIO (putStrLn msg)
+        liftIO (hPutStrLn stderr msg)
+        liftIO (exitImmediately (ExitFailure 1))
       Right v -> return v
+
+
+{- | Fork a normal, noncritical thread in 'MonadLoggerIO'. -}
+forkL :: (MonadLoggerIO io)
+  => LIO ()
+  -> io ()
+forkL io = liftIO . void . forkIO . runLoggingT io =<< askLoggerIO
+
+
+{- |
+  Class of monads that can be forked. I'm sure there is a better solution for
+  this, maybe using MonadBaseControl or something. This needs looking into.
+-}
+class ForkM m where
+  forkM :: m () -> m ()
+
+instance ForkM IO where
+  forkM = void . forkIO
+
+instance ForkM (LoggingT IO) where
+  forkM action = do
+    logging <- askLoggerIO
+    liftIO . forkM $ runLoggingT action logging
+
 
diff --git a/src/Network/Legion/Launch.hs b/src/Network/Legion/Launch.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Legion/Launch.hs
@@ -0,0 +1,41 @@
+{- |
+  This module takes care of launching the Legion runtime system and associated
+  admin tools. It mainly exists to avoid a module dependency cycle between the
+  runtime stuff and the admin stuff.
+-}
+module Network.Legion.Launch (
+  forkLegionary
+) where
+
+import Control.Monad.Logger (MonadLoggerIO)
+import Network.Legion.Admin (forkAdmin)
+import Network.Legion.Application (LegionConstraints, Persistence)
+import Network.Legion.Runtime (forkRuntime, Runtime)
+import Network.Legion.Runtime.State (StartupMode)
+import Network.Legion.Settings (RuntimeSettings)
+
+{- |
+  Forks the legion framework in a background thread, and returns a handle
+  on the runtime environment, which can be used to make user requests.
+
+  - @__e__@ is the type of request your application will handle. @__e__@ stands
+    for __"event"__.
+  - @__o__@ is the type of response produced by your application. @__o__@ stands
+    for __"output"__
+  - @__s__@ is the type of state maintained by your application. More
+    precisely, it is the type of the individual partitions that make up
+    your global application state. @__s__@ stands for __"state"__.
+-}
+forkLegionary :: (LegionConstraints e o s, MonadLoggerIO m)
+  => Persistence e o s
+    {- ^ The persistence layer used to back the legion framework. -}
+  -> RuntimeSettings
+    {- ^ Settings and configuration of the legion framework. -}
+  -> StartupMode
+  -> m (Runtime e o s)
+forkLegionary persistence settings startupMode = do
+  runtime <- forkRuntime persistence settings startupMode
+  forkAdmin settings runtime
+  return runtime
+
+
diff --git a/src/Network/Legion/Runtime.hs b/src/Network/Legion/Runtime.hs
--- a/src/Network/Legion/Runtime.hs
+++ b/src/Network/Legion/Runtime.hs
@@ -1,895 +1,532 @@
 {-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE TemplateHaskell #-}
-{- |
-  This module is responsible for the runtime operation of the legion
-  framework. This mostly means opening sockets and piping data around to the
-  various connected pieces.
--}
-module Network.Legion.Runtime (
-  forkLegionary,
-  StartupMode(..),
-  Runtime,
-  makeRequest,
-  search,
-) where
-
-import Control.Concurrent (forkIO)
-import Control.Concurrent.Chan (writeChan, newChan, Chan)
-import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
-import Control.Monad (void, forever, join)
-import Control.Monad.Catch (catchAll, try, SomeException, throwM)
-import Control.Monad.IO.Class (MonadIO, liftIO)
-import Control.Monad.Logger (logWarn, logError, logInfo, LoggingT,
-  MonadLoggerIO, runLoggingT, askLoggerIO, logDebug)
-import Control.Monad.Trans.Class (lift)
-import Control.Monad.Trans.State (StateT, runStateT, get, put, modify)
-import Data.Binary (encode, Binary)
-import Data.Conduit (Source, ($$), (=$=), yield, await, awaitForever,
-  transPipe, ConduitM, runConduit, Sink)
-import Data.Conduit.Network (sourceSocket)
-import Data.Conduit.Serialization.Binary (conduitDecode)
-import Data.Map (Map)
-import Data.Set (Set)
-import Data.String (IsString, fromString)
-import Data.Text (pack)
-import Data.Time (UTCTime, getCurrentTime)
-import GHC.Generics (Generic)
-import Network.Legion.Admin (runAdmin, AdminMessage(GetState, GetPart,
-  Eject, GetIndex, GetDivergent, GetStates))
-import Network.Legion.Application (LegionConstraints, Persistence,
-  list, saveCluster)
-import Network.Legion.BSockAddr (BSockAddr(BSockAddr))
-import Network.Legion.ClusterState (ClusterPowerState)
-import Network.Legion.Conduit (merge, chanToSink, chanToSource)
-import Network.Legion.Distribution (Peer, newPeer)
-import Network.Legion.Fork (forkC)
-import Network.Legion.Index (IndexRecord(IndexRecord), irTag, irKey,
-  SearchTag(SearchTag), indexEntries, Indexable)
-import Network.Legion.LIO (LIO)
-import Network.Legion.Lift (lift2,  lift3)
-import Network.Legion.PartitionKey (PartitionKey)
-import Network.Legion.PartitionState (PartitionPowerState)
-import Network.Legion.PowerState (Event)
-import Network.Legion.Runtime.ConnectionManager (newConnectionManager,
-  ConnectionManager, newPeers)
-import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),
-  PeerMessagePayload(ForwardRequest, ForwardResponse, ClusterMerge,
-  PartitionMerge, Search, SearchResponse, JoinNext, JoinNextResponse),
-  MessageId, newSequence, nextMessageId, JoinNextResponse(Joined,
-  JoinFinished))
-import Network.Legion.Settings (RuntimeSettings(RuntimeSettings,
-  adminHost, adminPort, peerBindAddr, joinBindAddr))
-import Network.Legion.StateMachine (partitionMerge, clusterMerge,
-  newNodeState, UserResponse(Forward, Respond), userRequest, eject,
-  minimumCompleteServiceSet, joinNext, joinNextResponse)
-import Network.Legion.StateMachine.Monad (NodeState, runSM, ClusterAction,
-  SM, popActions, nsIndex)
-import Network.Legion.UUID (getUUID)
-import Network.Socket (Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN),
-  SocketOption(ReuseAddr), SocketType(Stream), accept, bind,
-  defaultProtocol, listen, setSocketOption, socket, SockAddr(SockAddrInet,
-  SockAddrInet6, SockAddrUnix, SockAddrCan), connect, getPeerName, Socket)
-import Network.Socket.ByteString.Lazy (sendAll)
-import System.IO (stderr, hPutStrLn)
-import qualified Data.Conduit.List as CL
-import qualified Data.Map as Map
-import qualified Data.Set as Set
-import qualified Network.Legion.ClusterState as C
-import qualified Network.Legion.PowerState as PS
-import qualified Network.Legion.Runtime.ConnectionManager as CM
-import qualified Network.Legion.StateMachine as SM
-import qualified Network.Legion.StateMachine.Monad as SMM
-
-
-{- |
-  Run the legion node framework program, with the given user definitions,
-  framework settings, and request source. This function never returns
-  (except maybe with an exception if something goes horribly wrong).
--}
-runLegionary :: (LegionConstraints e o s)
-  => Persistence e o s
-    {- ^ The persistence layer used to back the legion framework. -}
-  -> RuntimeSettings
-    {- ^ Settings and configuration of the legionframework.  -}
-  -> StartupMode
-  -> Source IO (RequestMsg e o)
-    {- ^ A source of requests, together with a way to respond to the requets. -}
-  -> LoggingT IO ()
-    {-
-      Don't expose 'LIO' here because 'LIO' is a strictly internal
-      symbol. 'LoggingT IO' is what we expose to the world.
-    -}
-
-runLegionary
-    persistence
-    settings@RuntimeSettings {adminHost, adminPort}
-    startupMode
-    requestSource
-  = do
-    {- Start the various messages sources. -}
-    peerS <- loggingC =<< startPeerListener settings
-    adminS <- loggingC =<< runAdmin adminPort adminHost
-    joinS <- loggingC (joinMsgSource settings)
-
-    (self, nodeState, peers) <- makeNodeState persistence settings startupMode
-    rts <- newRuntimeState self peers
-    let
-      messageSource = transPipe lift (
-          (joinS =$= CL.map J) `merge`
-          (peerS =$= CL.map P) `merge`
-          (requestSource =$= CL.map R) `merge`
-          (adminS =$= CL.map A)
-        )
-    void . runRTS persistence nodeState rts . runConduit $
-      messageSource
-      =$= messageSink
-  where
-    newRuntimeState :: (Binary e, Binary o, Binary s)
-      => Peer
-      -> Map Peer BSockAddr
-      -> LoggingT IO (RuntimeState e o s)
-    newRuntimeState self peers = do
-      cm <- newConnectionManager peers
-      firstMessageId <- newSequence
-      return RuntimeState {
-          forwarded = Map.empty,
-          nextId = firstMessageId,
-          cm,
-          self,
-          commClock = Map.empty,
-          searches = Map.empty
-        }
-
-    {- |
-      Turn an LIO-based conduit into an IO-based conduit, so that it
-      will work with `merge`.
-    -}
-    loggingC :: ConduitM e o LIO r -> LIO (ConduitM e o IO r)
-    loggingC c = do
-      logging <- askLoggerIO
-      return (transPipe (`runLoggingT` logging) c)
-
-
-{- |
-  This is how requests are packaged when they are sent to the legion framework
-  for handling. It includes the request information itself, a partition key to
-  which the request is directed, and a way for the framework to deliver the
-  response to some interested party.
--}
-data RequestMsg e o
-  = Request PartitionKey e (o -> IO ())
-  | SearchDispatch SearchTag (Maybe IndexRecord -> IO ())
-instance (Show e) => Show (RequestMsg e o) where
-  show (Request k e _) = "(Request " ++ show k ++ " " ++ show e ++ " _)"
-  show (SearchDispatch s _) = "(SearchDispatch " ++ show s ++ " _)"
-
-
-messageSink :: (LegionConstraints e o s)
-  => Sink (RuntimeMessage e o s) (RTS e o s) ()
-messageSink = awaitForever (\msg -> do
-    $(logDebug) . pack $ "Receieved: " ++ show msg
-    lift $ do
-      case msg of
-        P (PeerMessage source _ _) ->
-          updateRecvClock source
-        _ -> return ()
-      handleMessage msg
-      updatePeers
-      clusterActions
-  )
-
-
-{- | Make progress on outstanding cluster actions. -}
-clusterActions :: RTS e o s ()
-clusterActions =
-    mapM_ clusterAction =<< popActions
-  where
-    {- |
-      Actually perform a cluster action as directed by the state
-      machine.
-    -}
-    clusterAction
-      :: ClusterAction e o s
-      -> RTS e o s ()
-
-    clusterAction (SMM.ClusterMerge peer ps) =
-      void $ send peer (ClusterMerge ps)
-
-    clusterAction (SMM.PartitionMerge peer key ps) =
-      void $ send peer (PartitionMerge key ps)
-
-    clusterAction (SMM.PartitionJoin peer keys) =
-      void $ send peer (JoinNext keys)
-
-
-{- |
-  Make sure the connection manager knows about any new peers that have
-  joined the cluster.
--}
-updatePeers :: RTS e o s ()
-updatePeers = do
-  peers <- SM.getPeers
-  RuntimeState {cm} <- lift get
-  lift2 $ newPeers cm peers
-
-
-{- |
-  Handle an individual runtime message, accepting an initial runtime
-  state and an initial node state, and producing an updated runtime
-  state and node state.
--}
-handleMessage :: (LegionConstraints e o s)
-  => RuntimeMessage e o s
-  -> RTS e o s ()
-
-handleMessage {- Join Next Response -}
-    (P (PeerMessage source _ (JoinNextResponse _messageId response)))
-  =
-    joinNextResponse source (toMaybe response)
-  where
-    toMaybe
-      :: JoinNextResponse e o s
-      -> Maybe (PartitionKey, PartitionPowerState e o s)
-    toMaybe (Joined key partition) = Just (key, partition)
-    toMaybe JoinFinished = Nothing
-
-handleMessage {- Join Next -}
-    (P (PeerMessage source messageId (JoinNext askKeys)))
-  =
-    joinNext source askKeys >>= \case
-      Nothing -> void $
-        send source (JoinNextResponse messageId JoinFinished)
-      Just (gotKey, partition) -> void $
-        send source (JoinNextResponse messageId (Joined gotKey partition))
-
-handleMessage {- Partition Merge -}
-    (P (PeerMessage _ _ (PartitionMerge key ps)))
-  =
-    partitionMerge key ps
-
-handleMessage {- Cluster Merge -}
-    (P (PeerMessage _ _ (ClusterMerge cs)))
-  =
-    clusterMerge cs
-
-handleMessage {- Forward Request -}
-    (P (msg@(PeerMessage source mid (ForwardRequest key request))))
-  = do
-    output <- userRequest key request
-    case output of
-      Respond response -> void $ send source (ForwardResponse mid response)
-      Forward peer -> forward peer msg
-
-handleMessage {- Forward Response -}
-    (msg@(P (PeerMessage _ _ (ForwardResponse mid response))))
-  = do
-    rts <- lift get
-    case lookupDelete mid (forwarded rts) of
-      (Nothing, fwd) -> do
-        $(logWarn) . pack $ "Unsolicited ForwardResponse: " ++ show msg
-        (lift . put) rts {forwarded = fwd}
-      (Just respond, fwd) -> do
-        lift2 $ respond response
-        (lift . put) rts {forwarded = fwd}
-
-handleMessage {- User Request -}
-    (R (Request key request respond))
-  = do
-    output <- userRequest key request
-    case output of
-      Respond response -> lift3 (respond response)
-      Forward peer -> do
-        messageId <- send peer (ForwardRequest key request)
-        (lift . modify) $ \rts@RuntimeState {forwarded} -> rts {
-            forwarded = Map.insert messageId (lift . respond) forwarded
-          }
-
-handleMessage {- Search Dispatch -}
-    {-
-      This is where we send out search request to all the appropriate
-      nodes in the cluster.
-    -}
-    (R (SearchDispatch searchTag respond))
-  =
-    Map.lookup searchTag . searches <$> lift get >>= \case
-      Nothing -> do
-        {-
-          No identical search is currently being executed, kick off a
-          new one.
-        -}
-        mcss <- minimumCompleteServiceSet
-        mapM_ sendOne (Set.toList mcss)
-        rts@RuntimeState {searches} <- lift get
-        (lift . put) rts {
-            searches = Map.insert
-              searchTag
-              (mcss, Nothing, [lift . respond])
-              searches
-          }
-      Just (peers, best, responders) -> do
-        {-
-          A search for this tag is already in progress, just add the
-          responder to the responder list.
-        -}
-        rts@RuntimeState {searches} <- lift get
-        (lift . put) rts {
-            searches = Map.insert
-              searchTag
-              (peers, best, (lift . respond):responders)
-              searches
-          }
-  where
-    sendOne :: Peer -> RTS e o s ()
-    sendOne peer =
-      void $ send peer (Search searchTag)
-
-handleMessage {- Search Execution -}
-    {- This is where we handle local search execution. -}
-    (P (PeerMessage source _ (Search searchTag)))
-  = do
-    output <- SM.search searchTag 
-    void $ send source (SearchResponse searchTag output)
-
-handleMessage {- Search Response -}
-    {-
-      This is where we gather all the responses from the various peers
-      to which we dispatched search requests.
-    -}
-    (msg@(P (PeerMessage source _ (SearchResponse searchTag response))))
-  =
-    {- TODO: see if this function can't be made more elegant. -}
-    Map.lookup searchTag . searches <$> lift get >>= \case
-      Nothing ->
-        {- There is no search happening. -}
-        $(logWarn) . pack $ "Unsolicited SearchResponse: " ++ show msg
-      Just (peers, best, responders) ->
-        if source `Set.member` peers
-          then
-            let peers2 = Set.delete source peers
-            in if null peers2
-              then do
-                {-
-                  All peers have responded, go ahead and respond to
-                  the client.
-                -}
-                lift2 $ mapM_ ($ bestOf best response) responders
-                rts@RuntimeState {searches} <- lift get
-                (lift . put) rts {searches = Map.delete searchTag searches}
-              else do
-                {- We are still waiting on some outstanding requests. -}
-                rts@RuntimeState {searches} <- lift get
-                (lift . put) rts {
-                    searches = Map.insert
-                      searchTag
-                      (peers2, bestOf best response, responders)
-                      searches
-                  }
-          else
-            {-
-              There is a search happening, but the peer that responded
-              is not part of it.
-            -}
-            $(logWarn) . pack $ "Unsolicited SearchResponse: " ++ show msg
-  where
-    {- |
-      Figure out which index record returned to us by the various peers
-      is the most appropriate to return to the user. This is mostly like
-      'min' but we can't use 'min' (or fancy applicative formulations)
-      because we want to favor 'Just' instead of 'Nothing'.
-    -}
-    bestOf :: Maybe IndexRecord -> Maybe IndexRecord -> Maybe IndexRecord
-    bestOf (Just a) (Just b) = Just (min a b)
-    bestOf Nothing b = b
-    bestOf a Nothing = a
-
-handleMessage {- Join Request -}
-    (J (JoinRequest addy, respond))
-  = do
-    (peer, cluster) <- SM.join addy
-    lift2 $ respond (JoinOk peer cluster)
-
-handleMessage {- Admin Get State -}
-    (A (GetState respond))
-  = 
-    lift2 . respond =<< SMM.getNodeState
-
-handleMessage {- Admin Get Partition -}
-    (A (GetPart key respond))
-  =
-    lift2 . respond =<< SM.getPartition key
-
-handleMessage {- Admin Eject Peer -}
-    (A (Eject peer respond))
-  = do
-    eject peer
-    lift2 $ respond ()
-
-handleMessage {- Admin Get Index -}
-    (A (GetIndex respond))
-  =
-    lift2 . respond =<< SMM.nsIndex <$> SMM.getNodeState
-
-handleMessage {- Admin Get Divergent -}
-    (A (GetDivergent respond))
-  = do
-    RuntimeState {commClock} <- lift get
-    diverging <- divergentPeers . SMM.partitions <$> SMM.getNodeState
-    lift2 . respond $ Map.fromAscList [
-        (peer, r)
-        | (peer, (_, r)) <- Map.toAscList commClock
-        , peer `Set.member` diverging
-      ]
-  where
-    divergentPeers :: Map PartitionKey (PartitionPowerState e o s) -> Set Peer
-    divergentPeers =
-      foldr Set.union Set.empty . fmap (PS.divergent . snd) . Map.toList
-
-handleMessage {- Admin Get States -}
-    (A (GetStates respond))
-  = do
-    persistence <- SMM.getPersistence
-    lift2 . respond . Map.fromList =<< runConduit (
-        transPipe liftIO (list persistence)
-        =$= CL.consume
-      )
-
-
-{- | This defines the various ways a node can be spun up. -}
-data StartupMode
-  = NewCluster
-    {- ^ Indicates that we should bootstrap a new cluster at startup. -}
-  | JoinCluster SockAddr
-    {- ^ Indicates that the node should try to join an existing cluster. -}
-  | Recover Peer ClusterPowerState
-    {- ^
-      Recover from a crash as the given peer, using the given cluster
-      state.
-    -}
-  deriving (Show, Eq)
-
-
-{- |
-  Construct a source of incoming peer messages.  We have to start the
-  peer listener first before we spin up the cluster management, which
-  is why this is an @LIO (Source LIO PeerMessage)@ instead of a
-  @Source LIO PeerMessage@.
--}
-startPeerListener :: (LegionConstraints e o s)
-  => RuntimeSettings
-  -> LIO (Source LIO (PeerMessage e o s))
-
-startPeerListener RuntimeSettings {peerBindAddr} =
-    catchAll (do
-        (inputChan, so) <- lift $ do
-          inputChan <- newChan
-          so <- socket (fam peerBindAddr) Stream defaultProtocol
-          setSocketOption so ReuseAddr 1
-          bind so peerBindAddr
-          listen so 5
-          return (inputChan, so)
-        forkC "peer socket acceptor" $ acceptLoop so inputChan
-        return (chanToSource inputChan)
-      ) (\err -> do
-        $(logError) . pack
-          $ "Couldn't start incomming peer message service, because of: "
-          ++ show (err :: SomeException)
-        throwM err
-      )
-  where
-    acceptLoop :: (LegionConstraints e o s)
-      => Socket
-      -> Chan (PeerMessage e o s)
-      -> LIO ()
-    acceptLoop so inputChan =
-        catchAll (
-          forever $ do
-            (conn, _) <- lift $ accept so
-            remoteAddr <- lift $ getPeerName conn
-            logging <- askLoggerIO
-            let runSocket =
-                  sourceSocket conn
-                  =$= conduitDecode
-                  $$ msgSink
-            void
-              . lift
-              . forkIO
-              . (`runLoggingT` logging)
-              . logErrors remoteAddr
-              $ runSocket
-        ) (\err -> do
-          $(logError) . pack
-            $ "error in peer message accept loop: "
-            ++ show (err :: SomeException)
-          throwM err
-        )
-      where
-        msgSink = chanToSink inputChan
-
-        logErrors :: SockAddr -> LIO () -> LIO ()
-        logErrors remoteAddr io = do
-          result <- try io
-          case result of
-            Left err ->
-              $(logWarn) . pack
-                $ "Incomming peer connection (" ++ show remoteAddr
-                ++ ") crashed because of: " ++ show (err :: SomeException)
-            Right v -> return v
-
-
-{- | Figure out how to construct the initial node state.  -}
-makeNodeState :: (Event e o s, Indexable s)
-  => Persistence e o s
-  -> RuntimeSettings
-  -> StartupMode
-  -> LIO (Peer, NodeState e o s, Map Peer BSockAddr)
-
-makeNodeState
-    persistence
-    settings@RuntimeSettings {peerBindAddr}
-    NewCluster
-  = do
-    {- Build a brand new node state, for the first node in a cluster. -}
-    verifyClearPersistence persistence
-    self <- newPeer
-    clusterId <- getUUID
-    let
-      cluster = C.new clusterId self peerBindAddr
-    makeNodeState persistence settings (Recover self cluster)
-
-makeNodeState
-    persistence
-    settings@RuntimeSettings {peerBindAddr}
-    (JoinCluster addr)
-  = do
-    {-
-      Join a cluster by either starting fresh, or recovering from a
-      shutdown or crash.
-    -}
-    verifyClearPersistence persistence
-    $(logInfo) "Trying to join an existing cluster."
-    (self, cluster) <- joinCluster (JoinRequest (BSockAddr peerBindAddr))
-    makeNodeState persistence settings (Recover self cluster)
-  where
-    joinCluster :: JoinRequest -> LIO (Peer, ClusterPowerState)
-    joinCluster joinMsg = liftIO $ do
-      so <- socket (fam addr) Stream defaultProtocol
-      connect so addr
-      sendAll so (encode joinMsg)
-      {-
-        using sourceSocket and conduitDecode is easier than building
-        a recive/decode state loop, even though we only read a single
-        response.
-      -}
-      sourceSocket so =$= conduitDecode $$ do
-        response <- await
-        case response of
-          Nothing -> fail
-            $ "Couldn't join a cluster because there was no response "
-            ++ "to our join request!"
-          Just (JoinOk self cps) ->
-            return (self, cps)
-
-makeNodeState persistence _ (Recover self cluster) = do
-    {- Make sure to rebuild the index in the case of recovery. -}
-    index <- runConduit . transPipe liftIO $
-      list persistence
-      =$= CL.fold addIndexRecords Set.empty
-    let
-      nodeState = (newNodeState self cluster) {nsIndex = index}
-    liftIO $ saveCluster persistence self cluster
-    return (self, nodeState, C.getPeers cluster)
-  where
-    addIndexRecords :: (Indexable s, Event e o s)
-      => Set IndexRecord
-      -> (PartitionKey, PartitionPowerState e o s)
-      -> Set IndexRecord
-    addIndexRecords index (key, partition) =
-      let
-        newRecords =
-          Set.map
-            (`IndexRecord` key)
-            (indexEntries (PS.projectedValue partition))
-      in Set.union index newRecords
-
-
-{- |
-  Helper for 'makeNodeState'. Verify that there is nothing in the
-  persistence layer.
--}
-verifyClearPersistence :: (MonadLoggerIO io) => Persistence e o s -> io ()
-verifyClearPersistence persistence = 
-  liftIO (runConduit (list persistence =$= CL.head)) >>= \case
-    Just _ -> do
-      let
-        msg :: (IsString a) => a
-        msg = fromString
-          $ "We are trying to start up a new peer, but the persistence "
-          ++ "layer already has data in it. This is an invalid state. "
-          ++ "New nodes must be started from a totally clean, empty state."
-      $(logError) msg
-      liftIO $ do
-        hPutStrLn stderr msg
-        putStrLn msg
-        error msg
-    Nothing ->
-      return ()
-
-
-{- | A source of cluster join request messages.  -}
-joinMsgSource
-  :: RuntimeSettings
-  -> Source LIO (JoinRequest, JoinResponse -> LIO ())
-
-joinMsgSource RuntimeSettings {joinBindAddr} = join . lift $
-    catchAll (do
-        (chan, so) <- lift $ do
-          chan <- newChan
-          so <- socket (fam joinBindAddr) Stream defaultProtocol
-          setSocketOption so ReuseAddr 1
-          bind so joinBindAddr
-          listen so 5
-          return (chan, so)
-        forkC "join socket acceptor" $ acceptLoop so chan
-        return (chanToSource chan)
-      ) (\err -> do
-        $(logError) . pack
-          $ "Couldn't start join request service, because of: "
-          ++ show (err :: SomeException)
-        throwM err
-      )
-  where
-    acceptLoop :: Socket -> Chan (JoinRequest, JoinResponse -> LIO ()) -> LIO ()
-    acceptLoop so chan =
-        catchAll (
-          forever $ do
-            (conn, _) <- lift $ accept so
-            logging <- askLoggerIO
-            (void . lift . forkIO . (`runLoggingT` logging) . logErrors) (
-                sourceSocket conn
-                =$= conduitDecode
-                =$= attachResponder conn
-                $$  chanToSink chan
-              )
-        ) (\err -> do
-          $(logError) . pack
-            $ "error in join request accept loop: "
-            ++ show (err :: SomeException)
-          throwM err
-        )
-      where
-        logErrors :: LIO () -> LIO ()
-        logErrors io = do
-          result <- try io
-          case result of
-            Left err ->
-              $(logWarn) . pack
-                $ "Incomming join connection crashed because of: "
-                ++ show (err :: SomeException)
-            Right v -> return v
-
-        attachResponder
-          :: Socket
-          -> ConduitM JoinRequest (JoinRequest, JoinResponse -> LIO ()) LIO ()
-        attachResponder conn = awaitForever (\msg -> do
-            mvar <- liftIO newEmptyMVar
-            yield (msg, lift . putMVar mvar)
-            response <- liftIO $ takeMVar mvar
-            liftIO $ sendAll conn (encode response)
-          )
-
-
-{- | Guess the family of a `SockAddr`. -}
-fam :: SockAddr -> Family
-fam SockAddrInet {} = AF_INET
-fam SockAddrInet6 {} = AF_INET6
-fam SockAddrUnix {} = AF_UNIX
-fam SockAddrCan {} = AF_CAN
-
-
-{- |
-  Forks the legion framework in a background thread, and returns a way to
-  send user requests to it and retrieve the responses to those requests.
-
-  - @__e__@ is the type of request your application will handle. @__e__@ stands
-    for __"event"__.
-  - @__o__@ is the type of response produced by your application. @__o__@ stands
-    for __"output"__
-  - @__s__@ is the type of state maintained by your application. More
-    precisely, it is the type of the individual partitions that make up
-    your global application state. @__s__@ stands for __"state"__.
--}
-forkLegionary :: (LegionConstraints e o s, MonadLoggerIO io)
-  => Persistence e o s
-    {- ^ The persistence layer used to back the legion framework. -}
-  -> RuntimeSettings
-    {- ^ Settings and configuration of the legion framework. -}
-  -> StartupMode
-  -> io (Runtime e o)
-
-forkLegionary persistence settings startupMode = do
-  logging <- askLoggerIO
-  liftIO . (`runLoggingT` logging) $ do
-    chan <- liftIO newChan
-    forkC "main legion thread" $
-      runLegionary persistence settings startupMode (chanToSource chan)
-    return Runtime {
-        rtMakeRequest = \key request -> liftIO $ do
-          responseVar <- newEmptyMVar
-          writeChan chan (Request key request (putMVar responseVar))
-          takeMVar responseVar,
-        rtSearch =
-          let
-            findNext :: SearchTag -> IO (Maybe IndexRecord)
-            findNext searchTag = do
-              responseVar <- newEmptyMVar
-              writeChan chan (SearchDispatch searchTag (putMVar responseVar))
-              takeMVar responseVar
-          in findNext
-
-      }
-
-
-{- |
-  This type represents a handle to the runtime environment of your
-  Legion application. This allows you to make requests and access the
-  partition index.
-
-  'Runtime' is an opaque structure. Use 'makeRequest' and 'search' to
-  access it.
--}
-data Runtime e o = Runtime {
-    {- |
-      Send an application request to the legion runtime, and get back
-      a response.
-    -}
-    rtMakeRequest :: PartitionKey -> e -> IO o,
-
-    {- | Query the index to find a set of partition keys.  -}
-    rtSearch :: SearchTag -> IO (Maybe IndexRecord)
-  }
-
-
-{- | Send a user request to the legion runtime. -}
-makeRequest :: (MonadIO io) => Runtime e o -> PartitionKey -> e -> io o
-makeRequest rt key = liftIO . rtMakeRequest rt key
-
-
-{- |
-  Send a search request to the legion runtime. Returns results that are
-  __strictly greater than__ the provided 'SearchTag'.
--}
-search :: (MonadIO io) => Runtime e o -> SearchTag -> Source io IndexRecord
-search rt tag =
-  liftIO (rtSearch rt tag) >>= \case
-    Nothing -> return ()
-    Just record@IndexRecord {irTag, irKey} -> do
-      yield record
-      search rt (SearchTag irTag (Just irKey))
-
-
-{- | This is the type of message passed around in the runtime. -}
-data RuntimeMessage e o s
-  = P (PeerMessage e o s)
-  | R (RequestMsg e o)
-  | J (JoinRequest, JoinResponse -> LIO ())
-  | A (AdminMessage e o s)
-instance (Show e, Show o, Show s) => Show (RuntimeMessage e o s) where
-  show (P m) = "(P " ++ show m ++ ")"
-  show (R m) = "(R " ++ show m ++ ")"
-  show (J (jr, _)) = "(J (" ++ show jr ++ ", _))"
-  show (A a) = "(A (" ++ show a ++ "))"
-
-
-{- |
-  The runtime state.
-
-  The 'searches' field is a little weird.
-
-  It turns out that searches are deterministic over the parameters of
-  'SearchTag' and cluster state. This should make sense, because everything in
-  Haskell is deterministic given __all__ the parameters. Since the cluster
-  state only changes over time, searches that happen "at the same time" and
-  for the same 'SearchTag' can be considered identical. I don't think it is too
-  much of a stretch to say that searches that have overlapping execution times
-  can be considered to be happening "at the same time", therefore the
-  search tag becomes determining factor in the result of the search.
-
-  This is a long-winded way of justifying the fact that, if we are currently
-  executing a search and an identical search requests arrives, then the second
-  identical search is just piggy-backed on the results of the currently
-  executing search. Whether this counts as a premature optimization hack or a
-  beautifully elegant expression of platonic reality is left as an exercise for
-  the reader. It does help simplify the code a little bit because we don't have
-  to specify some kind of UUID to differentiate otherwise identical searches.
--}
-data RuntimeState e o s = RuntimeState {
-         self :: Peer,
-    forwarded :: Map MessageId (o -> LIO ()),
-       nextId :: MessageId,
-           cm :: ConnectionManager e o s,
-    commClock :: Map Peer (Maybe UTCTime, Maybe UTCTime),
-                 {- ^ When did we last communicate with a peer. (sent, recv). -}
-     searches :: Map
-                   SearchTag
-                   (Set Peer, Maybe IndexRecord, [Maybe IndexRecord -> LIO ()])
-  }
-
-
-{- | This is the type of a join request message. -}
-newtype JoinRequest = JoinRequest BSockAddr
-  deriving (Generic, Show)
-instance Binary JoinRequest
-
-
-{- | The response to a JoinRequst message -}
-data JoinResponse
-  = JoinOk Peer ClusterPowerState
-  deriving (Generic)
-instance Binary JoinResponse
-
-
-{- | Lookup a key from a map, and also delete the key if it exists. -}
-lookupDelete :: (Ord k) => k -> Map k v -> (Maybe v, Map k v)
-lookupDelete = Map.updateLookupWithKey (const (const Nothing))
-
-
-{- | The runtime monad.  -}
-type RTS e o s =
-  SM e o s (
-  StateT (RuntimeState e o s)
-  LIO)
-
-
-{- | Shorthand for running the RTS monad. -}
-runRTS
-  :: Persistence e o s
-  -> NodeState e o s
-  -> RuntimeState e o s
-  -> RTS e o s a
-  -> LIO (a, NodeState e o s, [ClusterAction e o s], RuntimeState e o s)
-runRTS persistence ns rts =
-    fmap flatten
-    . (`runStateT` rts)
-    . runSM persistence ns
-  where
-    flatten ((a, b, c), d) = (a, b, c, d)
-
-
-{- |
-  Send a peer message in the RTS monad, automatically taking care of
-  necessary state updates.
--}
-send :: Peer -> PeerMessagePayload e o s -> RTS e o s MessageId
-send target payload = do
-  rts@RuntimeState {cm, self, nextId} <- lift get
-  (lift . put) rts {nextId = nextMessageId nextId}
-  lift2 $ CM.send cm target (PeerMessage self nextId payload)
-  return nextId
-
-
-{- | Forward an existing message to another peer. -}
-forward :: Peer -> PeerMessage e o s -> RTS e o s ()
-forward target message = do
-  RuntimeState {cm} <- lift get
-  lift2 $ CM.send cm target message
-
-
-{- | Update the time when we last received a message from a peer. -}
-updateRecvClock :: Peer -> RTS e o s ()
-updateRecvClock peer = do
-  now <- liftIO getCurrentTime
-  (lift . modify) (\rts@RuntimeState {commClock} ->
-      let
-        newCommClock = case Map.lookup peer commClock of
-          Nothing -> Map.insert peer (Nothing, Just now) commClock
-          Just (s, _) -> Map.insert peer (s, Just now) commClock
-      in newCommClock `seq` rts {
-          commClock = newCommClock
-        }
-    )
+{-# LANGUAGE ExplicitForAll #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TemplateHaskell #-}
+
+module Network.Legion.Runtime (
+  -- * User-facing interface.
+  forkRuntime,
+  makeRequest,
+  search,
+  Runtime,
+
+  -- * Internal interface.
+  eject,
+  getDivergent,
+
+  -- * Debugging interface.
+  debugLocalPartitions,
+  debugRuntimeState,
+  debugPartition,
+  debugIndex,
+
+) where
+
+
+import Control.Concurrent (writeChan, newChan, Chan)
+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
+import Control.Monad (void, forever)
+import Control.Monad.Catch (catchAll, try, SomeException, throwM,
+  MonadCatch)
+import Control.Monad.IO.Class (liftIO, MonadIO)
+import Control.Monad.Logger (MonadLoggerIO, logError, logWarn, logDebug,
+  askLoggerIO, runLoggingT)
+import Control.Monad.Trans.Class (lift)
+import Data.Aeson (Value)
+import Data.Binary (encode)
+import Data.Conduit ((.|), runConduit, awaitForever, Source, yield)
+import Data.Conduit.Network (sourceSocket)
+import Data.Conduit.Serialization.Binary (conduitDecode)
+import Data.Map (Map)
+import Data.Set (Set)
+import Data.Time (UTCTime)
+import Network.Legion.Application (LegionConstraints, Persistence)
+import Network.Legion.Conduit (chanToSource)
+import Network.Legion.Distribution (Peer)
+import Network.Legion.Fork (forkC, forkL, ForkM)
+import Network.Legion.Index (IndexRecord(IndexRecord),
+  SearchTag(SearchTag), irTag, irKey)
+import Network.Legion.LIO (LIO)
+import Network.Legion.PartitionKey (PartitionKey)
+import Network.Legion.PartitionState (PartitionPowerState)
+import Network.Legion.Runtime.ConnectionManager (send)
+import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),
+  PeerMessagePayload(PartitionMerge, ForwardRequest, ForwardResponse,
+  ClusterMerge, Search, SearchResponse, JoinNext, JoinNextResponse),
+  payload, source, messageId, JoinNextResponse(JoinFinished, Joined))
+import Network.Legion.Runtime.State (makeRuntimeState, StartupMode,
+  RuntimeT, JoinRequest(JoinRequest), JoinResponse(JoinOk), runRuntimeT,
+  updateRecvClock, userRequest, forwardResponse, clusterMerge,
+  getCM, searchResponse, joinNext, partitionMerge, joinNextResponse,
+  forwardedRequest, searchDispatch)
+import Network.Legion.Settings (RuntimeSettings(RuntimeSettings),
+  peerBindAddr, joinBindAddr)
+import Network.Legion.SocketUtil (fam)
+import Network.Socket (SocketOption(ReuseAddr), SocketType(Stream),
+  accept, bind, defaultProtocol, listen, setSocketOption, socket,
+  SockAddr, getPeerName, Socket)
+import Network.Socket.ByteString.Lazy (sendAll)
+import qualified Data.Text as T
+import qualified Network.Legion.Runtime.State as S
+
+
+{- |
+  This type represents a handle to the runtime environment of your
+  Legion application. This allows you to make requests and access the
+  partition index.
+
+  'Runtime' is an opaque structure. Use 'makeRequest' and 'search' to
+  access it.
+-}
+newtype Runtime e o s = Runtime {
+    unRuntime :: Chan (RuntimeMessage e o s)
+  }
+
+
+{- | Fork the runtime in a background thread. -}
+forkRuntime :: (LegionConstraints e o s, MonadLoggerIO m)
+  => Persistence e o s
+    {- ^ The persistence layer used to back the legion framework. -}
+  -> RuntimeSettings
+    {- ^ Settings and configuration of the legion framework. -}
+  -> StartupMode
+  -> m (Runtime e o s)
+forkRuntime persistence settings startupMode = do
+  runtime <- Runtime <$> liftIO newChan
+  logging <- askLoggerIO
+  liftIO . (`runLoggingT` logging) . forkC "main legion thread" $
+    executeRuntime persistence settings startupMode runtime
+  return runtime
+
+
+{- | Send a user request to the legion runtime. -}
+makeRequest :: (MonadIO m) => Runtime e o s -> PartitionKey -> e -> m o
+makeRequest runtime key e = call runtime (RMUserRequest key e)
+
+
+{- |
+  Send a search request to the legion runtime. Returns results that are
+  __strictly greater than__ the provided 'SearchTag'.
+-}
+search :: (MonadIO m) => Runtime e o s -> SearchTag -> Source m IndexRecord
+search runtime tag =
+  call runtime (RMUserSearch tag) >>= \case
+    Nothing -> return ()
+    Just record@IndexRecord {irTag, irKey} -> do
+      yield record
+      search runtime (SearchTag irTag (Just irKey))
+
+
+{- | Get the runtime state for debugging. -}
+debugRuntimeState :: (MonadIO m)
+  => Runtime e o s
+  -> m Value
+debugRuntimeState runtime = call runtime RMDebugRuntimeState
+
+
+{- | Get a partition for debugging. -}
+debugPartition :: (MonadIO m)
+  => Runtime e o s
+  -> PartitionKey
+  -> m (Maybe (PartitionPowerState e o s))
+debugPartition runtime = call runtime . RMDebugPartition
+
+
+{- | Eject a peer. -}
+eject :: (MonadIO m)
+  => Runtime e o s
+  -> Peer
+  -> m ()
+eject runtime = call runtime . RMEject
+
+
+{- | Get the index for debugging. -}
+debugIndex :: (MonadIO m)
+  => Runtime e o s
+  -> m (Set IndexRecord)
+debugIndex runtime = call runtime RMDebugIndex
+
+
+{- | Get the divergent peers. -}
+getDivergent :: (MonadIO m)
+  => Runtime e o s
+  -> m (Map Peer (Maybe UTCTime))
+getDivergent runtime = call runtime RMGetDivergent
+
+
+{- | Dump all of the locally managed partitions, for debugging. -}
+debugLocalPartitions :: (MonadIO m)
+  => Runtime e o s
+  -> m (Map PartitionKey (PartitionPowerState e o s))
+debugLocalPartitions runtime = call runtime RMDebugLocalPartitions
+
+
+{- |
+  Execute the Legion runtime, with the given user definitions, and
+  framework settings. This function never returns (except maybe with an
+  exception if something goes horribly wrong).
+-}
+executeRuntime :: (
+      ForkM m,
+      LegionConstraints e o s,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => Persistence e o s
+    {- ^ The persistence layer used to back the legion framework. -}
+  -> RuntimeSettings
+    {- ^ Settings and configuration of the legionframework.  -}
+  -> StartupMode
+  -> Runtime e o s
+    {- ^ A source of requests, together with a way to respond to the requets. -}
+  -> m ()
+executeRuntime
+    persistence
+    settings
+    startupMode
+    runtime
+  = do
+    {- Start the various messages sources. -}
+    startPeerListener settings runtime
+    startJoinListener settings runtime
+
+    rts <- makeRuntimeState persistence settings startupMode
+    void . runRuntimeT persistence rts . runConduit $
+      chanToSource (unRuntime runtime)
+      .| awaitForever (\msg -> do
+          $(logDebug) . T.pack $ "Receieved: " ++ show msg
+          lift (handleRuntimeMessage runtime msg)
+        )
+
+
+{- | Handle runtime message. -}
+handleRuntimeMessage :: (
+      ForkM m,
+      LegionConstraints e o s,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => Runtime e o s
+     {- ^
+       A handle on our own runtime, used to send messages back to
+       ourselves.
+     -}
+  -> RuntimeMessage e o s
+  -> RuntimeT e o s m ()
+
+handleRuntimeMessage
+    runtime
+    (RMPeerMessage msg@(PeerMessage source _ _))
+  = do
+    updateRecvClock source
+    handlePeerMessage runtime msg
+
+handleRuntimeMessage _ (RMJoinRequest (JoinRequest addr) responder) = do
+  (peer, cluster) <- S.joinCluster addr
+  respond responder (JoinOk peer cluster)
+
+handleRuntimeMessage _ (RMDebugRuntimeState responder) =
+  respond responder =<< S.debugRuntimeState
+
+handleRuntimeMessage _ (RMDebugPartition key responder) =
+  respond responder =<< S.debugPartition key
+
+handleRuntimeMessage _ (RMEject peer responder) =
+  respond responder =<< S.eject peer
+
+handleRuntimeMessage _ (RMDebugIndex responder) =
+  respond responder =<< S.debugIndex
+
+handleRuntimeMessage _ (RMGetDivergent responder) =
+  respond responder =<< S.getDivergent
+
+handleRuntimeMessage _ (RMDebugLocalPartitions responder) =
+  respond responder =<< S.debugLocalPartitions
+
+handleRuntimeMessage _ (RMUserRequest key request responder) =
+  userRequest key request (respond responder)
+
+handleRuntimeMessage _ (RMUserSearch tag responder) =
+  searchDispatch tag (respond responder)
+
+
+{- | Handle a peer message. -}
+handlePeerMessage :: (
+      ForkM m,
+      LegionConstraints e o s,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => Runtime e o s
+     {- ^
+       A handle on our own runtime, used to send messages back to
+       ourselves.
+     -}
+  -> PeerMessage e o s
+  -> RuntimeT e o s m ()
+
+handlePeerMessage {- PartitionMerge -}
+    _runtime
+    PeerMessage {
+        payload = (PartitionMerge key partition)
+      }
+  =
+    partitionMerge key partition
+
+handlePeerMessage {- ForwardRequest -}
+    _runtime
+    PeerMessage {
+        source,
+        messageId,
+        payload = ForwardRequest key event
+      }
+  =
+    forwardedRequest source messageId key event
+
+handlePeerMessage {- ForwardResponse -}
+    _runtime
+    PeerMessage {
+        payload = ForwardResponse forMessageId output
+      }
+  =
+    forwardResponse forMessageId output
+
+handlePeerMessage {- ClusterMerge -}
+    _runtime
+    PeerMessage {
+        payload = ClusterMerge cluster
+      }
+  =
+    clusterMerge cluster
+
+handlePeerMessage {- Search -}
+    {- This is where we handle local search execution. -}
+    _runtime
+    PeerMessage {
+        source,
+        payload = Search searchTag
+      }
+  = do
+    searchResult <- S.search searchTag
+    cm <- getCM
+    void $ send cm source (SearchResponse searchTag searchResult)
+
+handlePeerMessage {- SearchResponse -}
+    {-
+      This is where we gather all the responses from the various peers
+      to which we dispatched search requests.
+    -}
+    _runtime
+    PeerMessage {
+        source,
+        payload = SearchResponse searchTag record
+      }
+  =
+    searchResponse source searchTag record
+
+handlePeerMessage {- JoinNext -}
+    _runtime
+    PeerMessage {
+        source,
+        messageId,
+        payload = JoinNext askKeys
+      }
+  = do
+    cm <- getCM
+    joinNext source askKeys (\case
+        Nothing -> void $
+          send cm source (JoinNextResponse messageId JoinFinished)
+        Just (gotKey, partition) -> void $
+          send cm source (JoinNextResponse messageId (Joined gotKey partition))
+      )
+
+handlePeerMessage {- JoinNextResponse -}
+    _runtime
+    PeerMessage {
+        source,
+        payload = JoinNextResponse _toMessageId response
+      }
+  =
+    joinNextResponse source (toMaybe response)
+  where
+    toMaybe
+      :: JoinNextResponse e o s
+      -> Maybe (PartitionKey, PartitionPowerState e o s)
+    toMaybe (Joined key partition) = Just (key, partition)
+    toMaybe JoinFinished = Nothing
+
+
+{- | A way for the runtime to respond to a message. -}
+newtype Responder a = Responder {
+    unResponder :: a -> IO ()
+  }
+instance Show (Responder a) where
+  show _ = "Responder"
+
+
+{- | Respond to a messag, using the given responder, in 'MonadIO'. -}
+respond :: (MonadIO m) => Responder a -> a -> m ()
+respond responder = liftIO . unResponder responder
+
+
+{- | Send a message to the runtime that blocks on a response. -}
+call :: (MonadIO m)
+  => Runtime e o s
+  -> (Responder a -> RuntimeMessage e o s)
+  -> m a
+call runtime withResonder = liftIO $ do
+  mvar <- newEmptyMVar
+  cast runtime (withResonder (Responder (putMVar mvar)))
+  takeMVar mvar
+
+
+{- | Send a message to the runtime. Do not wait for a result. -}
+cast :: Runtime e o s -> RuntimeMessage e o s -> IO ()
+cast runtime = writeChan (unRuntime runtime)
+
+
+data RuntimeMessage e o s
+  = RMPeerMessage (PeerMessage e o s)
+  | RMJoinRequest JoinRequest (Responder JoinResponse)
+  | RMDebugRuntimeState (Responder Value)
+  | RMDebugPartition
+      PartitionKey
+      (Responder (Maybe (PartitionPowerState e o s)))
+  | RMEject Peer (Responder ())
+  | RMDebugIndex (Responder (Set IndexRecord))
+  | RMGetDivergent (Responder (Map Peer (Maybe UTCTime)))
+  | RMDebugLocalPartitions
+      (Responder (Map PartitionKey (PartitionPowerState e o s)))
+  | RMUserRequest PartitionKey e (Responder o)
+  | RMUserSearch SearchTag (Responder (Maybe IndexRecord))
+  deriving (Show)
+
+
+{- |
+  Start the peer listener, which accepts peer messages from the network
+  and sends them to the runtime.
+-}
+startPeerListener :: (
+      ForkM m,
+      LegionConstraints e o s,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => RuntimeSettings
+  -> Runtime e o s
+  -> m ()
+
+startPeerListener RuntimeSettings {peerBindAddr} runtime =
+    catchAll (do
+        so <- liftIO $ do
+          so <- socket (fam peerBindAddr) Stream defaultProtocol
+          setSocketOption so ReuseAddr 1
+          bind so peerBindAddr
+          listen so 5
+          return so
+        forkC "peer socket acceptor" $ acceptLoop so runtime
+      ) (\err -> do
+        $(logError) . T.pack
+          $ "Couldn't start incomming peer message service, because of: "
+          ++ show (err :: SomeException)
+        throwM err
+      )
+  where
+    acceptLoop :: (MonadLoggerIO m, LegionConstraints e o s, MonadCatch m)
+      => Socket
+      -> Runtime e o s
+      -> m ()
+    acceptLoop so runtime_ =
+        catchAll (
+          forever $ do
+            (conn, _) <- liftIO $ accept so
+            remoteAddr <- liftIO $ getPeerName conn
+            void
+              . forkL
+              . logErrors remoteAddr
+              $ runConduit (
+                  sourceSocket conn
+                  .| conduitDecode
+                  .| awaitForever (liftIO . cast runtime_ . RMPeerMessage)
+                )
+        ) (\err -> do
+          $(logError) . T.pack
+            $ "error in peer message accept loop: "
+            ++ show (err :: SomeException)
+          throwM err
+        )
+      where
+        logErrors :: SockAddr -> LIO () -> LIO ()
+        logErrors remoteAddr io = do
+          result <- try io
+          case result of
+            Left err ->
+              $(logWarn) . T.pack
+                $ "Incomming peer connection (" ++ show remoteAddr
+                ++ ") crashed because of: " ++ show (err :: SomeException)
+            Right v -> return v
+
+
+{- |
+  Starts the join listener, which accepts cluster join requests from
+  the network and sends them to the runtime.
+-}
+startJoinListener :: (MonadCatch m, MonadLoggerIO m, ForkM m)
+  => RuntimeSettings
+  -> Runtime e o s
+  -> m ()
+
+startJoinListener RuntimeSettings {joinBindAddr} runtime =
+    catchAll (do
+        so <- liftIO $ do
+          so <- socket (fam joinBindAddr) Stream defaultProtocol
+          setSocketOption so ReuseAddr 1
+          bind so joinBindAddr
+          listen so 5
+          return so
+        forkC "join socket acceptor" $ acceptLoop so
+      ) (\err -> do
+        $(logError) . T.pack
+          $ "Couldn't start join request service, because of: "
+          ++ show (err :: SomeException)
+        throwM err
+      )
+  where
+    acceptLoop :: (MonadCatch m, MonadLoggerIO m) => Socket -> m ()
+    acceptLoop so =
+        catchAll (
+          forever $ do
+            (conn, _) <- liftIO (accept so)
+            void
+              . forkL
+              . logErrors
+              . liftIO
+              $ runConduit (
+                  sourceSocket conn
+                  .| conduitDecode
+                  .| awaitForever (\req -> liftIO $
+                      sendAll conn . encode
+                        =<< call runtime (RMJoinRequest req)
+                    )
+                )
+        ) (\err -> do
+          $(logError) . T.pack
+            $ "error in join request accept loop: "
+            ++ show (err :: SomeException)
+          throwM err
+        )
+      where
+        logErrors :: (MonadCatch m, MonadLoggerIO m) => m () -> m ()
+        logErrors m = do
+          result <- try m
+          case result of
+            Left err ->
+              $(logWarn) . T.pack
+                $ "Incomming join connection crashed because of: "
+                ++ show (err :: SomeException)
+            Right v -> return v
 
 
diff --git a/src/Network/Legion/Runtime/ConnectionManager.hs b/src/Network/Legion/Runtime/ConnectionManager.hs
--- a/src/Network/Legion/Runtime/ConnectionManager.hs
+++ b/src/Network/Legion/Runtime/ConnectionManager.hs
@@ -9,28 +9,35 @@
   ConnectionManager,
   newConnectionManager,
   send,
+  forward,
   newPeers,
 ) where
 
 import Prelude hiding (lookup)
 
-import Control.Concurrent (Chan, writeChan, newChan, readChan)
-import Control.Exception (try, SomeException, bracketOnError)
+import Control.Concurrent (Chan, writeChan, newChan, readChan,
+  newEmptyMVar, putMVar, takeMVar)
+import Control.Exception (SomeException, bracketOnError)
 import Control.Monad (void)
-import Control.Monad.Logger (logInfo, logWarn)
+import Control.Monad.Catch (MonadCatch, try)
+import Control.Monad.IO.Class (MonadIO, liftIO)
+import Control.Monad.Logger (logInfo, logWarn, MonadLoggerIO)
 import Control.Monad.Trans.Class (lift)
 import Data.Binary (Binary, encode)
 import Data.ByteString.Lazy (ByteString)
+import Data.Conduit (Sink, runConduit, (.|), await)
 import Data.Map (toList, insert, empty, Map, lookup)
 import Data.Text (pack)
 import Network.Legion.BSockAddr (BSockAddr(BSockAddr))
+import Network.Legion.Conduit (chanToSource)
 import Network.Legion.Distribution (Peer)
-import Network.Legion.Fork (forkC)
-import Network.Legion.LIO (LIO)
-import Network.Legion.Runtime.PeerMessage (PeerMessage)
+import Network.Legion.Fork (forkC, ForkM)
+import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),
+  MessageId, PeerMessagePayload, source, messageId, payload,
+  nextMessageId, newSequence)
+import Network.Legion.SocketUtil (fam)
 import Network.Socket (SockAddr, Socket, socket, SocketType(Stream),
-  defaultProtocol, connect, close, SockAddr(SockAddrInet, SockAddrInet6,
-  SockAddrUnix, SockAddrCan), Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN))
+  defaultProtocol, connect, close, SockAddr)
 import Network.Socket.ByteString.Lazy (sendAll)
 
 {- |
@@ -44,63 +51,113 @@
 {- |
   Create a new connection manager.
 -}
-newConnectionManager :: (Binary e, Binary o, Binary s)
-  => Map Peer BSockAddr
-  -> LIO (ConnectionManager e o s)
-newConnectionManager initPeers = do
-    chan <- lift newChan
+newConnectionManager :: (
+      Binary e,
+      Binary o,
+      Binary s,
+      ForkM m,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => Peer
+  -> Map Peer BSockAddr
+  -> m (ConnectionManager e o s)
+newConnectionManager self initPeers = do
+    chan <- liftIO newChan
+    nextId <- newSequence
     forkC "connection manager thread" $
-      manager chan S {connections = empty}
+      manager chan S {
+          nextId,
+          connections = empty
+        }
     let cm = C chan
     newPeers cm initPeers
     return cm
   where
-    manager :: (Binary s, Binary o, Binary e)
+    manager :: (
+          Binary e,
+          Binary o,
+          Binary s,
+          ForkM m,
+          MonadCatch m,
+          MonadLoggerIO m
+        )
       => Chan (Message e o s)
       -> State e o s
-      -> LIO ()
-    manager chan state = lift (readChan chan) >>= handle state >>= manager chan
+      -> m ()
+    manager chan state =
+      runConduit (chanToSource chan .| handle state)
 
-    handle :: (Binary e, Binary o, Binary s)
+    handle :: (
+          Binary e,
+          Binary o,
+          Binary s,
+          ForkM m,
+          MonadCatch m,
+          MonadLoggerIO m
+        )
       => State e o s
-      -> Message e o s
-      -> LIO (State e o s)
-    handle s@S {connections} (NewPeer peer addr) =
-      case lookup peer connections of
-        Nothing -> do
-          conn <- connection addr
-          return s {
-              connections = insert peer conn connections
-            }
-        Just _ ->
-          return s
-
-    handle s@S {connections} (Send peer msg) = do
-      case lookup peer connections of
-        Nothing -> $(logWarn) . pack $ "unknown peer: " ++ show peer
-        Just conn -> lift $ writeChan conn msg
-      return s
+      -> Sink (Message e o s) m ()
+    handle s@S {connections, nextId} =
+      await >>= \case
+        Nothing -> return ()
+        Just (NewPeer peer addr) ->
+          handle =<< case lookup peer connections of
+            Nothing -> do
+              conn <- lift (connection addr)
+              return s {
+                  connections = insert peer conn connections
+                }
+            Just _ ->
+              return s
+        Just (Send peer payload respond) -> do
+          case lookup peer connections of
+            Nothing -> $(logWarn) . pack $ "unknown peer: " ++ show peer
+            Just conn -> liftIO $
+              writeChan conn PeerMessage {
+                source = self,
+                messageId = nextId,
+                payload
+              }
+          liftIO (respond nextId)
+          handle s {nextId = nextMessageId nextId}
+        Just (Forward peer msg) ->
+          case lookup peer connections of
+            Nothing -> $(logWarn) . pack $ "unknown peer: " ++ show peer
+            Just conn -> liftIO $ writeChan conn msg
 
 
-{- |
-  Build a new connection.
--}
-connection :: (Binary e, Binary o, Binary s)
+{- | Build a new connection. -}
+connection :: (
+      Binary e,
+      Binary o,
+      Binary s,
+      ForkM m,
+      MonadCatch m,
+      MonadIO m,
+      MonadLoggerIO m
+    )
   => SockAddr
-  -> LIO (Chan (PeerMessage e o s))
+  -> m (Chan (PeerMessage e o s))
 
 connection addr = do
-    chan <- lift newChan
+    chan <- liftIO newChan
     forkC ("connection to: " ++ show addr) $
       handle chan Nothing
     return chan
   where
-    handle :: (Binary e, Binary o, Binary s)
+    handle :: (
+          Binary e,
+          Binary o,
+          Binary s,
+          MonadCatch m,
+          MonadLoggerIO m
+        )
       => Chan (PeerMessage e o s)
       -> Maybe Socket
-      -> LIO ()
+      -> m ()
     handle chan so =
-      lift (readChan chan) >>= sendWithRetry so . encode >>= handle chan
+      liftIO (readChan chan) >>= sendWithRetry so . encode >>= handle chan
 
     {- | Open a socket. -}
     openSocket :: IO Socket
@@ -122,9 +179,12 @@
       create a new socket and retry sending the payload. Return whatever the
       "working" socket is.
     -}
-    sendWithRetry :: Maybe Socket -> ByteString -> LIO (Maybe Socket)
+    sendWithRetry :: (MonadCatch m, MonadLoggerIO m)
+      => Maybe Socket
+      -> ByteString
+      -> m (Maybe Socket)
     sendWithRetry Nothing payload =
-      (lift . try) openSocket >>= \case
+      try (liftIO openSocket) >>= \case
         Left err -> do
           $(logWarn) . pack
             $ "Can't connect to: " ++ show addr ++ ". Dropping message on "
@@ -132,7 +192,7 @@
             ++ show (err :: SomeException)
           return Nothing
         Right so -> do
-          result2 <- (lift . try) (sendAll so payload)
+          result2 <- try (liftIO (sendAll so payload))
           case result2 of
             Left err -> $(logWarn) . pack
               $ "An error happend when trying to send a payload over a socket "
@@ -142,43 +202,57 @@
             Right _ -> return ()
           return (Just so)
     sendWithRetry (Just so) payload =
-      (lift . try) (sendAll so payload) >>= \case
+      try (liftIO (sendAll so payload)) >>= \case
         Left err -> do
           $(logInfo) . pack
             $ "Socket to " ++ show addr ++ " died. Retrying on a new "
             ++ "socket. The error was: " ++ show (err :: SomeException)
-          (lift . void) (try (close so) :: IO (Either SomeException ()))
+          (liftIO . void) (try (close so) :: IO (Either SomeException ()))
           sendWithRetry Nothing payload
         Right _ ->
           return (Just so)
 
 
-{- |
-  Send a message to a peer.
--}
-send
-  :: ConnectionManager e o s
+{- | Send a message to a peer. -}
+send :: (MonadIO m)
+  => ConnectionManager e o s
   -> Peer
+  -> PeerMessagePayload e o s
+  -> m MessageId
+send (C chan) peer payload = do
+  mvar <- liftIO newEmptyMVar
+  liftIO . writeChan chan $ Send peer payload (putMVar mvar)
+  liftIO (takeMVar mvar)
+
+
+{- | Forward a message. -}
+forward :: (MonadIO m)
+  => ConnectionManager e o s
+  -> Peer
   -> PeerMessage e o s
-  -> LIO ()
-send (C chan) peer = lift . writeChan chan . Send peer
+  -> m ()
+forward (C chan) peer =
+  liftIO . writeChan chan . Forward peer
 
 
 {- |
   Tell the connection manager about a new peer.
 -}
-newPeer
-  :: ConnectionManager e o s
+newPeer :: (MonadIO io)
+  => ConnectionManager e o s
   -> Peer
   -> SockAddr
-  -> LIO ()
-newPeer (C chan) peer addr = lift $ writeChan chan (NewPeer peer addr)
+  -> io ()
+newPeer (C chan) peer addr = liftIO $ writeChan chan (NewPeer peer addr)
 
 
 {- |
   Tell the connection manager about all the peers known to the cluster state.
 -}
-newPeers :: ConnectionManager e o s -> Map Peer BSockAddr -> LIO ()
+newPeers :: (MonadIO io)
+  => ConnectionManager e o s
+  -> Map Peer BSockAddr
+  -> io ()
 newPeers cm peers =
     mapM_ oneNewPeer (toList peers)
   where
@@ -188,7 +262,8 @@
 {- |
   The internal state of the connection manager.
 -}
-newtype State e o s = S {
+data State e o s = S {
+         nextId :: MessageId,
     connections :: Map Peer (Chan (PeerMessage e o s))
   }
 
@@ -198,16 +273,7 @@
 -}
 data Message e o s
   = NewPeer Peer SockAddr
-  | Send Peer (PeerMessage e o s)
-
-
-{- |
-  Guess the family of a `SockAddr`.
--}
-fam :: SockAddr -> Family
-fam SockAddrInet {} = AF_INET
-fam SockAddrInet6 {} = AF_INET6
-fam SockAddrUnix {} = AF_UNIX
-fam SockAddrCan {} = AF_CAN
+  | Forward Peer (PeerMessage e o s)
+  | Send Peer (PeerMessagePayload e o s) (MessageId -> IO ())
 
 
diff --git a/src/Network/Legion/Runtime/PeerMessage.hs b/src/Network/Legion/Runtime/PeerMessage.hs
--- a/src/Network/Legion/Runtime/PeerMessage.hs
+++ b/src/Network/Legion/Runtime/PeerMessage.hs
@@ -12,7 +12,7 @@
   nextMessageId,
 ) where
 
-import Control.Monad.Trans.Class (lift)
+import Control.Monad.IO.Class (MonadIO)
 import Data.Binary (Binary)
 import Data.UUID (UUID)
 import Data.Word (Word64)
@@ -21,7 +21,6 @@
 import Network.Legion.Distribution (Peer)
 import Network.Legion.Index (SearchTag, IndexRecord)
 import Network.Legion.KeySet (KeySet)
-import Network.Legion.LIO (LIO)
 import Network.Legion.PartitionKey (PartitionKey)
 import Network.Legion.PartitionState (PartitionPowerState)
 import Network.Legion.UUID (getUUID)
@@ -77,8 +76,8 @@
   unique message ids by generating a unique UUID for each one, but generating
   UUIDs is not free, and we are probably going to be generating a lot of these.
 -}
-newSequence ::  LIO MessageId
-newSequence = lift $ do
+newSequence :: (MonadIO io) => io MessageId
+newSequence = do
   sid <- getUUID
   return (M sid 0)
 
diff --git a/src/Network/Legion/Runtime/State.hs b/src/Network/Legion/Runtime/State.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Legion/Runtime/State.hs
@@ -0,0 +1,1035 @@
+{-# LANGUAGE DeriveGeneric #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TemplateHaskell #-}
+
+{- | This module contains low level state types. -}
+module Network.Legion.Runtime.State (
+  -- * Runtime State
+  RuntimeState,
+  makeRuntimeState,
+
+  -- * Runtime Monads
+  RuntimeT,
+  runRuntimeT,
+  runConcurrentT,
+
+  -- * Runtime Monad Operations
+  updateRecvClock,
+  joinCluster,
+  eject,
+  getDivergent,
+  userRequest,
+  forwardedRequest,
+  forwardResponse,
+  clusterMerge,
+  getCM,
+  searchDispatch,
+  search,
+  searchResponse,
+  joinNext,
+  joinNextResponse,
+  partitionMerge,
+  
+
+  -- * Debug Monad Operations
+  debugIndex,
+  debugRuntimeState,
+  debugLocalPartitions,
+  debugPartition,
+
+  -- * Other Types
+  StartupMode(..),
+  JoinRequest(..),
+  JoinResponse(..),
+  UserResponse(..),
+) where
+
+
+import Control.Concurrent.STM (TVar, atomically, readTVar, newTVar,
+  STM, writeTVar, modifyTVar)
+import Control.Monad (unless, void)
+import Control.Monad.Catch (throwM, MonadThrow, MonadCatch)
+import Control.Monad.IO.Class (MonadIO, liftIO)
+import Control.Monad.Logger (MonadLoggerIO, MonadLogger, logInfo,
+  logError, logWarn, logDebug)
+import Control.Monad.Trans.Class (lift, MonadTrans)
+import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)
+import Control.Monad.Trans.State (StateT, runStateT, modify, get, put)
+import Data.Aeson (Value, toJSON, ToJSON)
+import Data.Binary (encode, Binary)
+import Data.Bool (bool)
+import Data.Conduit ((.|), await, transPipe, runConduit)
+import Data.Conduit.Network (sourceSocket)
+import Data.Conduit.Serialization.Binary (conduitDecode)
+import Data.Default.Class (Default)
+import Data.Map (Map)
+import Data.Maybe (catMaybes, fromMaybe)
+import Data.Set (Set, (\\))
+import Data.String (IsString, fromString)
+import Data.Text (Text)
+import Data.Time (UTCTime, getCurrentTime)
+import GHC.Generics (Generic)
+import Network.Legion.Application (Persistence, list, saveCluster,
+  getState, saveState)
+import Network.Legion.BSockAddr (BSockAddr(BSockAddr))
+import Network.Legion.ClusterState (ClusterPowerState, RebalanceOrd,
+  ClusterPowerStateT)
+import Network.Legion.Distribution (Peer, newPeer)
+import Network.Legion.Fork (ForkM, forkM)
+import Network.Legion.Index (IndexRecord(IndexRecord),
+  SearchTag(SearchTag), Indexable, indexEntries, stTag, stKey, irTag,
+  irKey)
+import Network.Legion.KeySet (KeySet)
+import Network.Legion.PartitionKey (PartitionKey)
+import Network.Legion.PartitionState (PartitionPowerState,
+  PartitionPowerStateT)
+import Network.Legion.PowerState (Event)
+import Network.Legion.PowerState.Monad (PropAction(Send, DoNothing))
+import Network.Legion.Runtime.ConnectionManager (ConnectionManager,
+  newConnectionManager)
+import Network.Legion.Runtime.PeerMessage (MessageId, newSequence,
+  PeerMessagePayload(ClusterMerge, PartitionMerge, JoinNext,
+  ForwardRequest, ForwardResponse, Search), PeerMessage(PeerMessage),
+  source, messageId, payload)
+import Network.Legion.Settings (RuntimeSettings(RuntimeSettings,
+  peerBindAddr))
+import Network.Legion.SocketUtil (fam)
+import Network.Legion.UUID (getUUID)
+import Network.Socket (SocketType(Stream), defaultProtocol, socket,
+  SockAddr, connect)
+import Network.Socket.ByteString.Lazy (sendAll)
+import System.IO (stderr, hPutStrLn)
+import qualified Data.Aeson as A
+import qualified Data.ByteString.Lazy.Char8 as BSL8
+import qualified Data.Conduit.List as CL
+import qualified Data.Map as Map
+import qualified Data.Set as Set
+import qualified Data.Text as T
+import qualified Network.Legion.ClusterState as C
+import qualified Network.Legion.Distribution as D
+import qualified Network.Legion.KeySet as KS
+import qualified Network.Legion.PowerState as PS
+import qualified Network.Legion.PowerState.Monad as PM
+import qualified Network.Legion.Runtime.ConnectionManager as CM
+
+
+{- | The state of the runtime system. -}
+data RuntimeState e o s m = RuntimeState {
+             self :: Peer,
+          cluster :: ClusterPowerState,
+       partitions :: Map PartitionKey (TVar (PartitionWorkerState e o s m)),
+         rtsIndex :: Set IndexRecord,
+            joins :: Map Peer KeySet,
+                     {- ^ Outstanding joins. -}
+    lastRebalance :: RebalanceOrd,
+        forwarded :: Map MessageId (o -> m ()),
+           nextId :: MessageId,
+               cm :: ConnectionManager e o s,
+        recvClock :: Map Peer (Maybe UTCTime),
+                     {- ^ When did we last receive a message from a peer. -}
+         searches :: Map
+                       SearchTag
+                       (
+                         Set Peer,
+                         Maybe IndexRecord,
+                         [Maybe IndexRecord -> m ()]
+                       )
+                     {- ^
+                       A map of currently dispatched searches. The values
+                       are the peers from which we are still expecting
+                       a result, best result so far, and the responders
+                       to which to send the eventual best result.
+
+                       The 'searches' field is a little weird.
+
+                       It turns out that searches are deterministic
+                       over the parameters of 'SearchTag' and cluster
+                       state. This should make sense, because everything
+                       in Haskell is deterministic given __all__
+                       the parameters. Since the cluster state only
+                       changes over time, searches that happen "at the
+                       same time" and for the same 'SearchTag' can be
+                       considered identical. I don't think it is too
+                       much of a stretch to say that searches that have
+                       overlapping execution times can be considered
+                       to be happening "at the same time", therefore
+                       the search tag becomes determining factor in the
+                       result of the search.
+
+                       This is a long-winded way of justifying the fact
+                       that, if we are currently executing a search and
+                       an identical search requests arrives, then the
+                       second identical search is just piggy-backed on the
+                       results of the currently executing search. Whether
+                       this counts as a premature optimization hack
+                       or a beautifully elegant expression of platonic
+                       reality is left as an exercise for the reader. It
+                       does help simplify the code a little bit because
+                       we don't have to specify some kind of UUID to
+                       differentiate otherwise identical searches.
+                     -}
+  }
+instance Show (RuntimeState e o s m) where
+  show = BSL8.unpack . A.encode
+instance ToJSON (RuntimeState e o s m) where
+  toJSON _ = toJSON ("RuntimeState" :: Text)
+
+
+{- | The state of an individual asynchronous partition worker. -}
+data PartitionWorkerState e o s m = PWS {
+             pwsCm :: ConnectionManager e o s,
+            pwsKey :: PartitionKey,
+           pwsSelf :: Peer,
+    pwsPersistence :: Persistence e o s,
+       pwsCacheVal :: Maybe (PartitionPowerState e o s),
+       pwsJobQueue :: [(ClusterPowerState, PartitionPowerStateT e o s m ())]
+  }
+
+
+{- | The Runtime Monad Transformer. -}
+newtype RuntimeT e o s m a = RuntimeT {
+    unRuntimeT :: StateT (RuntimeState e o s m) (ReaderT (Persistence e o s) m) a
+  }
+  deriving (Functor, Applicative, Monad, MonadIO, MonadLogger, MonadThrow)
+instance MonadTrans (RuntimeT e o s) where
+  lift = RuntimeT . lift . lift
+
+
+{- | Execute a 'RuntimeT'.  -}
+runRuntimeT
+  :: Persistence e o s
+  -> RuntimeState e o s m
+  -> RuntimeT e o s m a
+  -> m (a, RuntimeState e o s m)
+runRuntimeT persistence rts =
+  (`runReaderT` persistence) . (`runStateT` rts) . unRuntimeT
+
+
+{- | Initialize the runtime state. -}
+makeRuntimeState :: (
+      Binary e,
+      Binary o,
+      Binary s,
+      Event e o s,
+      ForkM m,
+      Indexable s,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => Persistence e o s
+  -> RuntimeSettings
+  -> StartupMode
+  -> m (RuntimeState e o s m)
+
+makeRuntimeState
+    persistence
+    settings@RuntimeSettings {peerBindAddr}
+    NewCluster
+  = do
+    {- Build a brand new node state, for the first node in a cluster. -}
+    verifyClearPersistence persistence
+    self <- newPeer
+    clusterId <- getUUID
+    let
+      cluster = C.new clusterId self peerBindAddr
+    makeRuntimeState persistence settings (Recover self cluster)
+
+makeRuntimeState
+    persistence
+    settings@RuntimeSettings {peerBindAddr}
+    (JoinCluster addr)
+  = do
+    {-
+      Join a cluster by either starting fresh, or recovering from a
+      shutdown or crash.
+    -}
+    verifyClearPersistence persistence
+    $(logInfo) "Trying to join an existing cluster."
+    (self, cluster) <- requestJoin (JoinRequest (BSockAddr peerBindAddr))
+    makeRuntimeState persistence settings (Recover self cluster)
+  where
+    requestJoin :: (MonadLoggerIO io)
+      => JoinRequest
+      -> io (Peer, ClusterPowerState)
+    requestJoin joinMsg = liftIO $ do
+      so <- socket (fam addr) Stream defaultProtocol
+      connect so addr
+      sendAll so (encode joinMsg)
+      {-
+        using sourceSocket and conduitDecode is easier than building
+        a recive/decode state loop, even though we only read a single
+        response.
+      -}
+      runConduit $ sourceSocket so .| conduitDecode .| do
+        response <- await
+        case response of
+          Nothing -> fail
+            $ "Couldn't join a cluster because there was no response "
+            ++ "to our join request!"
+          Just (JoinOk self cps) ->
+            return (self, cps)
+
+makeRuntimeState persistence _ (Recover self cluster) = do
+    {- Make sure to rebuild the index in the case of recovery. -}
+    rtsIndex <- runConduit . transPipe liftIO $
+      list persistence
+      .| CL.fold addIndexRecords Set.empty
+    firstMessageId <- newSequence
+    cm <- newConnectionManager self (C.getPeers cluster)
+    liftIO $ saveCluster persistence self cluster
+    return RuntimeState {
+        self,
+        cluster,
+        partitions = Map.empty,
+        rtsIndex,
+        joins = Map.empty,
+        lastRebalance = minBound,
+        forwarded = Map.empty,
+        nextId = firstMessageId,
+        cm,
+        recvClock = Map.empty,
+        searches = Map.empty
+      }
+  where
+    addIndexRecords :: (Indexable s, Event e o s)
+      => Set IndexRecord
+      -> (PartitionKey, PartitionPowerState e o s)
+      -> Set IndexRecord
+    addIndexRecords index (key, partition) =
+      let
+        newRecords =
+          Set.map
+            (`IndexRecord` key)
+            (indexEntries (PS.projectedValue partition))
+      in Set.union index newRecords
+
+
+{- | This defines the various ways a node can be spun up. -}
+data StartupMode
+  = NewCluster
+    {- ^ Indicates that we should bootstrap a new cluster at startup. -}
+  | JoinCluster SockAddr
+    {- ^ Indicates that the node should try to join an existing cluster. -}
+  | Recover Peer ClusterPowerState
+    {- ^
+      Recover from a crash as the given peer, using the given cluster
+      state.
+    -}
+  deriving (Show, Eq)
+
+
+{- | This is the type of a join request message. -}
+newtype JoinRequest = JoinRequest BSockAddr
+  deriving (Generic, Show)
+instance Binary JoinRequest
+
+
+{- | The response to a JoinRequest message -}
+data JoinResponse
+  = JoinOk Peer ClusterPowerState
+  deriving (Generic)
+instance Binary JoinResponse
+
+
+{- |
+  Helper for 'makeRuntimeState'. Verify that there is nothing in the
+  persistence layer.
+-}
+verifyClearPersistence :: (MonadLoggerIO io) => Persistence e o s -> io ()
+verifyClearPersistence persistence = 
+  liftIO (runConduit (list persistence .| CL.head)) >>= \case
+    Just _ -> do
+      let
+        msg :: (IsString a) => a
+        msg = fromString
+          $ "We are trying to start up a new peer, but the persistence "
+          ++ "layer already has data in it. This is an invalid state. "
+          ++ "New nodes must be started from a totally clean, empty state."
+      $(logError) msg
+      liftIO $ do
+        hPutStrLn stderr msg
+        putStrLn msg
+        error msg
+    Nothing ->
+      return ()
+
+
+{- | Update the time when we last received a message from a peer. -}
+updateRecvClock :: (MonadIO m) => Peer -> RuntimeT e o s m ()
+updateRecvClock peer = RuntimeT $ do
+  now <- liftIO getCurrentTime
+  modify (\rts@RuntimeState {recvClock} ->
+      let
+        newRecvClock = Map.insert peer (Just now) recvClock
+      in newRecvClock `seq` rts {
+          recvClock = newRecvClock
+        }
+    )
+
+
+{- | Return the current index for debugging. -}
+debugIndex :: (Monad m) => RuntimeT e o s m (Set IndexRecord)
+debugIndex = RuntimeT $ rtsIndex <$> get
+
+
+{- | Return the runtime state for debugging. -}
+debugRuntimeState :: (Monad m) => RuntimeT e o s m Value
+debugRuntimeState = toJSON <$> RuntimeT get
+
+
+{- | Return all of the local partitions for debugging. -}
+debugLocalPartitions :: (MonadIO m)
+  => RuntimeT e o s m (Map PartitionKey (PartitionPowerState e o s))
+debugLocalPartitions = do
+  persistence <- RuntimeT (lift ask)
+  Map.fromList <$> runConduit (
+      transPipe liftIO (list persistence)
+      .| CL.consume
+    )
+  
+
+
+{- | Return a specific partition, for debugging. -}
+debugPartition :: (MonadIO m)
+  => PartitionKey
+  -> RuntimeT e o s m (Maybe (PartitionPowerState e o s))
+debugPartition key = RuntimeT $ do
+  persistence <- lift ask
+  liftIO (getState persistence key)
+
+
+{- | Let a new peer join the cluster. -}
+joinCluster :: (MonadIO m, MonadThrow m)
+  => BSockAddr
+  -> RuntimeT e o s m (Peer, ClusterPowerState)
+joinCluster addr = do
+  peer <- newPeer
+  runClusterPowerStateT (C.joinCluster peer addr)
+  cluster <- getCluster
+  return (peer, cluster)
+
+
+{- | Eject a peer from the cluster. -}
+eject :: (MonadIO m, MonadThrow m) => Peer -> RuntimeT e o s m ()
+eject peer = do
+  {-
+    We need to think very hard about the split brain problem. A random
+    thought about that is that we should consider the extreme case where
+    the network just fails completely and every node believes that every
+    other node should be or has been ejected. This would obviously be
+    catastrophic in terms of data durability unless we have some way to
+    reintegrate an ejected node. So, either we have to guarantee that
+    such a situation can never happen, or else implement a reintegration
+    strategy.  It might be acceptable for the reintegration strategy to
+    be very costly if it is characterized as an extreme recovery scenario.
+
+    Question: would a reintegration strategy become less costly if the
+    "next state id" for a peer were global across all power states
+    instead of local to each power state?
+  -}
+  runClusterPowerStateT (C.eject peer)
+  {-
+    'runClusterPowerStateT (C.eject peer)' will cause us to attempt to
+    notify the peer that they have been ejected, but that notification
+    is almost certainly going to go unacknowledged because the peer
+    is probably down.
+    
+    This call to 'eject' was presumably invoked as a result of user
+    action, and we must therefore trust the user to know that the peer
+    is really down and not coming back. This "guarantee" allows us to
+    acknowledge the ejection on the peer's behalf.
+
+    This call will acknowledge the drop on behalf of the peer, and also
+    remove that peer from the keyspace distribution map.
+  -}
+  runClusterPowerStateTAs peer (return ())
+
+
+{- |
+  Gets the peers that the local node thinks are diverging, and the time
+  we last received a message from those peers.
+-}
+getDivergent :: (MonadIO m) => RuntimeT e o s m (Map Peer (Maybe UTCTime))
+getDivergent = RuntimeT $ do
+    RuntimeState {recvClock, partitions} <- get
+    diverging <- lift . lift $ divergentPeers partitions
+    return $ Map.fromAscList [
+        (peer, r)
+        | (peer, r) <- Map.toAscList recvClock
+        , peer `Set.member` diverging
+      ]
+  where
+    divergentPeers :: (MonadIO m)
+      => Map PartitionKey (TVar (PartitionWorkerState e o s m))
+      -> m (Set Peer)
+    divergentPeers partitions = liftIO $
+      foldr Set.union Set.empty . catMaybes <$> sequence [
+          fmap PS.divergent . pwsCacheVal <$> atomically (readTVar tvar)
+          | (_key, tvar) <- Map.toList partitions
+        ]
+
+
+{- | Handle a user request, and pass the response to the continuation. -}
+userRequest :: (
+      Default s,
+      Eq e,
+      Event e o s,
+      ForkM m,
+      Indexable s,
+      MonadCatch m,
+      MonadLoggerIO m,
+      Show e,
+      Show s
+    )
+  => PartitionKey
+  -> e
+  -> (o -> m ())
+  -> RuntimeT e o s m ()
+userRequest key request k = do
+  RuntimeState {self, cm} <- RuntimeT get
+  route key >>= \case
+    p | p == self -> 
+      runConcurrentT key (
+          lift . k =<< PM.event request
+        )
+    p -> do
+      messageId <- CM.send cm p (ForwardRequest key request)
+      (RuntimeT . modify) (\rts@RuntimeState {forwarded} -> rts {
+          forwarded = Map.insert messageId k forwarded
+        })
+
+
+{- | Handle a forwarded request. -}
+forwardedRequest :: (
+      Default s,
+      Eq e,
+      Event e o s,
+      ForkM m,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => Peer
+  -> MessageId
+  -> PartitionKey
+  -> e
+  -> RuntimeT e o s m ()
+forwardedRequest source messageId key event = do
+  RuntimeState {self, cm} <- RuntimeT get
+  route key >>= \case
+    p | p == self ->
+      runConcurrentT key (do
+          o <- PM.event event
+          (void . lift) (CM.send cm source (ForwardResponse messageId o))
+        )
+    p ->
+      {-
+        No need to keep track of the forwarded message, we just need to
+        reconstruct the original message and send it on its way.
+
+        TODO think about implementing cycle detection. Cycles should
+        not exists, so if we detect one then that classifies as
+        a bug. So really this is more of an opportunity for bug
+        detection.
+      -}
+      CM.forward cm p PeerMessage {
+          source,
+          messageId,
+          payload = ForwardRequest key event
+        }
+
+
+{- | Find the route for a user request. -}
+route :: (MonadLogger m)
+  => PartitionKey
+  -> RuntimeT e o s m Peer
+route key = RuntimeT $ do
+  RuntimeState {self, cluster} <- get
+  let routes = C.findRoute key cluster
+  if self `Set.member` routes
+    then return self
+    else case Set.toList routes of
+      [] -> do
+        let msg = "No routes for key: " ++ show key
+        $(logError) . T.pack $ msg
+        error msg
+      peer:_ -> return peer
+  
+
+
+
+{- | Receive a response to a forwarded user request. -}
+forwardResponse :: (MonadLoggerIO m, Show o)
+  => MessageId
+  -> o
+  -> RuntimeT e o s m ()
+forwardResponse forMessageId output = do
+  rts@RuntimeState{forwarded} <- RuntimeT get
+  let (r, fwd) = lookupAndDelete forMessageId forwarded
+  RuntimeT $ put rts {forwarded = fwd}
+  case r of
+    Nothing ->
+      $(logWarn) . T.pack
+        $ "Received unexpected forward response: "
+        ++ show (forMessageId, output)
+    Just respond ->
+      lift (respond output)
+
+
+{- | Merge a forenig cluster power state. -}
+clusterMerge :: (MonadIO m, MonadThrow m)
+  => ClusterPowerState
+  -> RuntimeT e o s m ()
+clusterMerge cluster =
+  runClusterPowerStateT (PM.merge cluster)
+
+
+{- | Return the handle to the connection manager. -}
+getCM :: (Monad m) => RuntimeT e o s m (ConnectionManager e o s)
+getCM = RuntimeT $ cm <$> get
+
+
+{- | Dispatch a distributed search request. -}
+searchDispatch :: (MonadIO m)
+  => SearchTag
+  -> (Maybe IndexRecord -> m ())
+  -> RuntimeT e o s m ()
+searchDispatch searchTag k =
+    Map.lookup searchTag . searches <$> RuntimeT get >>= \case
+      Nothing -> do
+        {-
+          No identical search is currently being executed, kick off a
+          new one.
+        -}
+        mcss <- minimumCompleteServiceSet
+        mapM_ sendSearch (Set.toList mcss)
+        (RuntimeT . modify) (\rts@RuntimeState {searches} -> rts {
+            searches = Map.insert
+              searchTag
+              (mcss, Nothing, [k])
+              searches
+          })
+      Just (peers, best, responders) ->
+        {-
+          A search for this tag is already in progress, just add the
+          responder to the responder list.
+        -}
+        (RuntimeT . modify) (\rts@RuntimeState {searches} -> rts {
+            searches = Map.insert
+              searchTag
+              (peers, best, k:responders)
+              searches
+          })
+  where
+    sendSearch :: (MonadIO m)
+      => Peer
+      -> RuntimeT e o s m ()
+    sendSearch peer = do
+      cm <- getCM
+      void $ CM.send cm peer (Search searchTag)
+
+
+{- |
+  Search the index, and return the first record that is __strictly
+  greater than__ the provided search tag, if such a record exists.
+-}
+search :: (Monad m)
+  => SearchTag
+  -> RuntimeT e o s m (Maybe IndexRecord)
+
+search SearchTag {stTag, stKey = Nothing} = RuntimeT $ do
+  index <- rtsIndex <$> get
+  return (Set.lookupGE IndexRecord {irTag = stTag, irKey = minBound} index)
+
+search SearchTag {stTag, stKey = Just key} = RuntimeT $ do
+  index <- rtsIndex <$> get
+  return (Set.lookupGT IndexRecord {irTag = stTag, irKey = key} index)
+
+
+{- | Handle an incomming search response. -}
+searchResponse :: (MonadLogger m)
+  => Peer
+  -> SearchTag
+  -> Maybe IndexRecord
+  -> RuntimeT e o s m ()
+
+searchResponse source searchTag response =
+    {- TODO: see if this function can't be made more elegant. -}
+    Map.lookup searchTag . searches <$> RuntimeT get >>= \case
+      Nothing ->
+        {- There is no search happening. -}
+        $(logWarn) . T.pack
+          $ "Unsolicited SearchResponse: "
+          ++ show (source, searchTag, response)
+      Just (peers, best, responders) ->
+        if source `Set.member` peers
+          then
+            let peers2 = Set.delete source peers
+            in if null peers2
+              then do
+                {-
+                  All peers have responded, go ahead and respond to
+                  the client.
+                -}
+                lift $ mapM_ ($ bestOf best response) responders
+                rts@RuntimeState {searches} <- RuntimeT get
+                (RuntimeT . put) rts {searches = Map.delete searchTag searches}
+              else do
+                {- We are still waiting on some outstanding requests. -}
+                rts@RuntimeState {searches} <- RuntimeT get
+                (RuntimeT . put) rts {
+                    searches = Map.insert
+                      searchTag
+                      (peers2, bestOf best response, responders)
+                      searches
+                  }
+          else
+            {-
+              There is a search happening, but the peer that responded
+              is not part of it.
+            -}
+            $(logWarn) . T.pack
+              $ "Unsolicited SearchResponse: "
+              ++ show (source, searchTag, response)
+  where
+    {- |
+      Figure out which index record returned to us by the various peers
+      is the most appropriate to return to the user. This is mostly like
+      'min' but we can't use 'min' (or fancy applicative formulations)
+      because we want to favor 'Just' instead of 'Nothing'.
+    -}
+    bestOf :: Maybe IndexRecord -> Maybe IndexRecord -> Maybe IndexRecord
+    bestOf (Just a) (Just b) = Just (min a b)
+    bestOf Nothing b = b
+    bestOf a Nothing = a
+
+
+{- |
+  Allow a peer to participate in the replication of the partition with the
+  __minimum__ key that is within the indicated partition key set. Calls
+  the continuation with @Nothing@ if there is no such partition, or @Just
+  (key, partition)@ where @key@ is the partition key that was joined
+  and @partition@ is the resulting partition power state.
+-}
+joinNext :: (
+      Default s,
+      Eq e,
+      Event e o s,
+      ForkM m,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => Peer
+  -> KeySet
+  -> (Maybe (PartitionKey, PartitionPowerState e o s) -> m ())
+  -> RuntimeT e o s m ()
+joinNext peer askKeys k = do
+  persistence <- RuntimeT (lift ask)
+  (lift . runConduit) (
+      transPipe liftIO (list persistence)
+      .| CL.filter ((`KS.member` askKeys) . fst)
+      .| CL.head
+    ) >>= \case
+      Nothing -> lift (k Nothing)
+      Just (gotKey, _) ->
+        runConcurrentT gotKey (do
+            PM.participate peer
+            PM.acknowledge
+            partition <- PM.getPowerState
+            lift (k (Just (gotKey, partition)))
+          )
+
+
+{- | Receive the result of a JoinNext request. -}
+joinNextResponse :: (
+      Default s,
+      Eq e,
+      Event e o s,
+      ForkM m,
+      MonadLoggerIO m,
+      MonadCatch m,
+      Show e,
+      Show s
+    )
+  => Peer
+  -> Maybe (PartitionKey, PartitionPowerState e o s)
+  -> RuntimeT e o s m ()
+joinNextResponse peer response = do
+  RuntimeState {cluster, lastRebalance} <- RuntimeT get
+  if lastRebalance > fst (C.nextAction cluster)
+    then
+      {- We are receiving messages from an old rebalance. Log and ignore. -}
+      $(logWarn) . T.pack
+        $ "Received an old join response: "
+        ++ show (peer, response, cluster, lastRebalance)
+    else do
+      case response of
+        Just (key, partition) -> do
+          partitionMerge key partition
+          RuntimeState {joins, cm} <- RuntimeT get
+          case (KS.\\ KS.fromRange minBound key) <$> Map.lookup peer joins of
+            Nothing ->
+              {- An unexpected peer sent us this message, Ignore. -}
+              $(logWarn) . T.pack
+                $ "Unexpected join next: " ++ show (peer, response)
+            Just needsJoinSet -> do
+              unless (KS.null needsJoinSet) (
+                  void $ CM.send cm peer (JoinNext needsJoinSet)
+                )
+              (RuntimeT . modify) (\rts -> rts {
+                  joins = Map.filter
+                    (not . KS.null)
+                    (Map.insert peer needsJoinSet joins)
+                })
+        Nothing ->
+          (RuntimeT . modify) (\rts@RuntimeState {joins} -> rts {
+              joins = Map.delete peer joins
+            })
+      Map.null . joins <$> RuntimeT get >>= bool
+        (return ())
+        (runClusterPowerStateT C.finishRebalance)
+
+
+{- | Merge a foreign partition replica with the local partion replica. -}
+partitionMerge :: (
+      Default s,
+      Eq e,
+      Event e o s,
+      ForkM m,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => PartitionKey
+  -> PartitionPowerState e o s
+  -> RuntimeT e o s m ()
+partitionMerge key foreignPartition =
+  runConcurrentT key (PM.merge foreignPartition)
+
+
+{- | Get the current cluster state. -}
+getCluster :: (Monad m) => RuntimeT e o s m ClusterPowerState
+getCluster = RuntimeT $ cluster <$> get
+
+
+{- |
+  Run a 'ClusterPowerStateT', and perform any resulting propagation
+  actions.
+-}
+runClusterPowerStateT :: (MonadIO m, MonadThrow m)
+  => ClusterPowerStateT m a
+  -> RuntimeT e o s m a
+runClusterPowerStateT m = do
+  as <- RuntimeT $ self <$> get
+  runClusterPowerStateTAs as m
+
+
+{- |
+  Run a clusterstate-flavored 'PowerStateT' in the 'RuntimeT' monad,
+  automatically acknowledging the resulting power state.
+
+  Generalized to run as any peer, in order to support exceptional cases
+  like 'eject'.
+-}
+runClusterPowerStateTAs :: (MonadIO m, MonadThrow m)
+  => Peer {- ^ The peer to run as. -}
+  -> ClusterPowerStateT m a
+  -> RuntimeT e o s m a
+runClusterPowerStateTAs as m = do
+  RuntimeState {cluster, self} <- RuntimeT get
+  persistence <- RuntimeT (lift ask)
+  lift (PM.runPowerStateT as cluster (m <* PM.acknowledge)) >>= \case
+    Left err -> throwM err
+    Right (a, action, cluster2, _outputs) -> do
+      RuntimeT (modify (\rts -> rts {cluster = cluster2}))
+      liftIO (saveCluster persistence self cluster2)
+      case action of
+        Send -> sequence_ [
+            getCM >>= (\cm -> CM.send cm p (ClusterMerge cluster2))
+            | p <- Set.toList (PS.allParticipants cluster2)
+            , p /= self
+          ]
+        DoNothing -> return ()
+      return a
+
+
+{- |
+  The type of response to a user request, either forward to another node,
+  or respond directly.
+-}
+data UserResponse o
+  = Forward Peer
+  | Respond o
+
+
+{- | The action is executed in a background thread. -}
+runConcurrentT :: (
+      Default s,
+      Eq e,
+      Event e o s,
+      ForkM m,
+      MonadCatch m,
+      MonadLoggerIO m
+    )
+  => PartitionKey
+  -> PartitionPowerStateT e o s m ()
+  -> RuntimeT e o s m ()
+runConcurrentT key action_ = do
+    rts@RuntimeState {partitions} <- RuntimeT get
+    persistence <- RuntimeT (lift ask)
+    let job = (cluster rts, action_)
+    case Map.lookup key partitions of
+      Nothing -> do
+        tvar <- liftIO (atomically (newTVar PWS {
+            pwsCm = cm rts,
+            pwsKey = key,
+            pwsSelf = self rts,
+            pwsPersistence = persistence,
+            pwsCacheVal = Nothing,
+            pwsJobQueue = []
+          }))
+        RuntimeT $ put rts {partitions = Map.insert key tvar partitions}
+        lift =<< liftIO (atomically (queueAction tvar job))
+      Just tvar ->
+        lift =<< liftIO (atomically (queueAction tvar job))
+  where
+    {- |
+      Put the partition action on the execution queue, and maybe also
+      start the execution thread if there isn't already one running.
+    -}
+    queueAction :: (
+          Default s,
+          Eq e,
+          Event e o s,
+          ForkM m,
+          MonadLoggerIO m
+        )
+      => TVar (PartitionWorkerState e o s m)
+      -> (ClusterPowerState, PartitionPowerStateT e o s m ())
+      -> STM (m ())
+    queueAction tvar job = do
+      pws@PWS {pwsJobQueue} <- readTVar tvar
+      let
+        forkJobWorker = return (forkM (jobWorker tvar job))
+      writeTVar tvar pws {pwsJobQueue = pwsJobQueue ++ [job]}
+      if null pwsJobQueue
+        then forkJobWorker
+        else return (return ())
+
+    jobWorker :: (MonadLoggerIO m, Default s, Event e o s, Eq e)
+      => TVar (PartitionWorkerState e o s m)
+      -> (ClusterPowerState, PartitionPowerStateT e o s m ())
+      -> m ()
+    jobWorker tvar job =
+        doJob tvar job >> nextJob >>= maybe shutdown (jobWorker tvar)
+      where
+        nextJob = liftIO . atomically $ do
+          pws@PWS {pwsCacheVal, pwsJobQueue} <- readTVar tvar
+          case pwsJobQueue of
+            _:next:more -> do
+              {- Pop the last job off the queue, and promote the next job. -}
+              writeTVar tvar pws {pwsJobQueue = next:more}
+              return (Just next)
+            [_] -> do
+              {-
+                All jobs complete. Pop the last job off the stack and
+                clean up the cache if necessary.
+              -}
+              writeTVar tvar $
+                case Set.null . PS.divergent <$> pwsCacheVal of
+                  Just False -> pws {pwsJobQueue = []}
+                  _ -> pws {pwsCacheVal = Nothing, pwsJobQueue = []}
+              return Nothing
+            [] ->
+              {- This shouldn't happen. See about using non-empty lists. -}
+              return Nothing
+        shutdown = return ()
+
+    doJob :: (MonadLoggerIO m, Default s, Event e o s, Eq e)
+      => TVar (PartitionWorkerState e o s m)
+      -> (ClusterPowerState, PartitionPowerStateT e o s m ())
+      -> m ()
+    doJob tvar (cluster, action) = do
+        $(logDebug) . T.pack $ "Starting job on " ++ show key
+        PWS {
+            pwsCm,
+            pwsSelf,
+            pwsPersistence,
+            pwsCacheVal
+          } <- liftIO (atomically (readTVar tvar))
+        partition <- case pwsCacheVal of
+          Nothing ->
+            fromMaybe (PS.new key (C.findOwners key cluster))
+              <$> liftIO (getState pwsPersistence key)
+          Just partition -> return partition
+        PM.runPowerStateT pwsSelf partition (
+            action <* (removeObsolete >> PM.acknowledge)
+          ) >>= \case
+            Left err ->
+              $(logError) . T.pack
+                $ "Partition error: " ++ show (err, key)
+            Right ((), propAction, newPartition, _outputs) -> do
+              liftIO . atomically . modifyTVar tvar $ (\pws ->
+                  pws {pwsCacheVal = Just newPartition}
+                )
+              liftIO (saveState pwsPersistence key (Just newPartition))
+              case propAction of
+                Send -> sequence_ [
+                    CM.send pwsCm p (PartitionMerge key newPartition)
+                    | p <- Set.toList (PS.allParticipants newPartition)
+                    , p /= pwsSelf
+                  ]
+                DoNothing -> return ()
+        $(logDebug) . T.pack $ "Finished job on " ++ show key
+      where
+        {- |
+          Remove obsolete peers. Obsolete peers are peers that are no longer
+          participating in the replication of this partition, due to a
+          rebalance. Such peers are removed lazily here at read time.
+        -}
+        removeObsolete :: (Monad m, Event e o s, Eq e)
+          => PartitionPowerStateT e o s m ()
+        removeObsolete = do
+          let owners = C.findOwners key cluster
+          peers <- PS.projParticipants <$> PM.getPowerState
+          let obsolete = peers \\ owners
+          mapM_
+            (\peer -> PM.disassociate peer >> PM.acknowledgeAs peer)
+            (Set.toList obsolete)
+
+
+{- | Lookup a key from a map, and also delete the key if it exists. -}
+lookupAndDelete :: (Ord k) => k -> Map k v -> (Maybe v, Map k v)
+lookupAndDelete = Map.updateLookupWithKey (const (const Nothing))
+
+
+{- |
+  Figure out the set of nodes to which search requests should be
+  dispatched. "Minimum complete service set" means the minimum set
+  of peers that, together, service the whole partition key space;
+  thereby guaranteeing that if any particular partition is indexed,
+  the corresponding index record will exist on one of these peers.
+
+  Implementation considerations:
+
+  There will usually be more than one solution for the MCSS. For now,
+  we just compute a deterministic solution, but we should implement
+  a random (or pseudo-random) solution in order to maximally balance
+  cluster resources.
+
+  Also, it is not clear that the minimum complete service set is even
+  what we really want. MCSS will reduce overall network utilization,
+  but it may actually increase latency. If we were to dispatch redundant
+  requests to multiple nodes, we could continue with whichever request
+  returns first, and ignore the slow responses. This is probably the
+  best solution. We will call this "fastest competitive search".
+
+  TODO: implement fastest competitive search.
+-}
+minimumCompleteServiceSet :: (Monad m) => RuntimeT e o s m (Set Peer)
+minimumCompleteServiceSet = do
+  RuntimeState {cluster} <- RuntimeT get
+  return (D.minimumCompleteServiceSet (C.getDistribution cluster))
+
+
diff --git a/src/Network/Legion/SocketUtil.hs b/src/Network/Legion/SocketUtil.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Legion/SocketUtil.hs
@@ -0,0 +1,19 @@
+
+{- | Socket utilities. -}
+module Network.Legion.SocketUtil (
+  fam,
+) where
+
+
+import Network.Socket (SockAddr, SockAddr(SockAddrInet, SockAddrInet6,
+  SockAddrUnix, SockAddrCan), Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN))
+
+
+{- | Guess the family of a `SockAddr`. -}
+fam :: SockAddr -> Family
+fam SockAddrInet {} = AF_INET
+fam SockAddrInet6 {} = AF_INET6
+fam SockAddrUnix {} = AF_UNIX
+fam SockAddrCan {} = AF_CAN
+
+
diff --git a/src/Network/Legion/StateMachine.hs b/src/Network/Legion/StateMachine.hs
deleted file mode 100644
--- a/src/Network/Legion/StateMachine.hs
+++ /dev/null
@@ -1,597 +0,0 @@
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE TemplateHaskell #-}
-{- |
-  This module contains the "pure-ish" state machine that defines what
-  it means to be a legion node. As described on 'SM', the state machine
-  is modeled in monadic fashion, where the state machine sate is modeled
-  as monadic context, state machine input is modeled as various monadic
-  functions, and state machine output is modeled as the result of those
-  monadic functions.
-
-  The reason the state lives behind a monad is because part of the
-  node state (i.e. the persistence layer) really does live behind IO,
-  and cannot be accessed purely. Therefore, the state is divided into a
-  pure part, modeled by 'NodeState'; and an impure part, modeled by the
-  persistence layer interface. We wrap these two components inside
-  of a new, opaque, monad called 'SM' by using a monad transformation
-  stack, where 'StateT' wraps the pure part of the state, and IO wraps
-  the impure part of the state. (This is a simplified description. The
-  actual monad transformation stack is more complicated, because it
-  incorporates logging and access to the user-defined request handler.)
-
-  The overall purpose of all of this is to separate as much as
-  possible the abstract idea of what a legion node is with its runtime
-  considerations. The state machine contained in this module defines how a
-  legion node should behave when faced with various inputs, and it would
-  be completely pure but for the persistence layer interface. The runtime
-  system 'Network.Legion.Runtime' implements the mechanisms by which
-  such input is collected and any behavior associated with the output
-  (e.g. managing network connections, sending data across the wire,
-  reading data from the wire, transforming those data into inputs to
-  the state machine, etc.).
--}
-module Network.Legion.StateMachine(
-  -- * Running the state machine.
-  newNodeState,
-
-  -- * State machine inputs.
-  userRequest,
-  partitionMerge,
-  clusterMerge,
-  eject,
-  join,
-  minimumCompleteServiceSet,
-  search,
-
-  joinNext,
-  joinNextResponse,
-
-  -- * State machine outputs.
-  UserResponse(..),
-
-  -- * State inspection
-  getPeers,
-  getPartition,
-) where
-
-import Control.Monad (void, unless)
-import Control.Monad.Catch (throwM, MonadThrow)
-import Control.Monad.IO.Class (MonadIO, liftIO)
-import Control.Monad.Logger (logDebug, logError, MonadLoggerIO, logWarn)
-import Control.Monad.Trans.Class (lift)
-import Data.Bool (bool)
-import Data.Conduit ((=$=), runConduit, transPipe, awaitForever)
-import Data.Default.Class (Default)
-import Data.Map (Map)
-import Data.Maybe (fromMaybe)
-import Data.Set (Set, (\\), member)
-import Data.Text (pack)
-import Network.Legion.Application (getState, saveState, list, saveCluster)
-import Network.Legion.BSockAddr (BSockAddr)
-import Network.Legion.ClusterState (ClusterPowerState, ClusterPowerStateT)
-import Network.Legion.Distribution (Peer, newPeer, RebalanceAction(Invite,
-  Drop))
-import Network.Legion.Index (IndexRecord(IndexRecord), stTag, stKey,
-  irTag, irKey, SearchTag(SearchTag), indexEntries, Indexable)
-import Network.Legion.KeySet (KeySet)
-import Network.Legion.PartitionKey (PartitionKey)
-import Network.Legion.PartitionState (PartitionPowerState, PartitionPowerStateT)
-import Network.Legion.PowerState (Event)
-import Network.Legion.PowerState.Monad (PropAction(Send, DoNothing))
-import Network.Legion.StateMachine.Monad (SM, NodeState(NodeState),
-  ClusterAction(PartitionMerge, ClusterMerge, PartitionJoin),
-  self, cluster, partitions, nsIndex, getPersistence, getNodeState,
-  modifyNodeState, pushActions, joins, lastRebalance)
-import qualified Data.Conduit.List as CL
-import qualified Data.Map as Map
-import qualified Data.Set as Set
-import qualified Network.Legion.ClusterState as C
-import qualified Network.Legion.Distribution as D
-import qualified Network.Legion.KeySet as KS
-import qualified Network.Legion.PowerState as PS
-import qualified Network.Legion.PowerState.Monad as PM
-
-
-{- | Make a new node state. -}
-newNodeState :: Peer -> ClusterPowerState -> NodeState e o s
-newNodeState self cluster =
-  NodeState {
-      self,
-      cluster,
-      partitions = Map.empty,
-      nsIndex = Set.empty,
-      joins = Map.empty,
-      lastRebalance = minBound
-    }
-
-
-{- | Handle a user request. -}
-userRequest :: (
-      Default s,
-      Eq e,
-      Event e o s,
-      Indexable s,
-      MonadLoggerIO m,
-      MonadThrow m,
-      Show e,
-      Show s
-    )
-  => PartitionKey
-  -> e
-  -> SM e o s m (UserResponse o)
-userRequest key request = do
-  NodeState {self, cluster} <- getNodeState
-  let routes = C.findRoute key cluster
-  if self `Set.member` routes
-    then do
-      (response, _) <- runPartitionPowerStateT key (
-          PM.event request
-        )
-      return (Respond response)
-
-    else case Set.toList routes of
-      [] -> do
-        let msg = "No routes for key: " ++ show key
-        $(logError) . pack $ msg
-        error msg
-      peer:_ -> return (Forward peer)
-
-
-{- |
-  Handle the state transition for a partition merge event. Returns 'Left'
-  if there is an error, and 'Right' if everything went fine.
--}
-partitionMerge :: (
-      Default s,
-      Eq e,
-      Event e o s,
-      Indexable s,
-      MonadLoggerIO m,
-      MonadThrow m,
-      Show e,
-      Show s
-    )
-  => PartitionKey
-  -> PartitionPowerState e o s
-  -> SM e o s m ()
-partitionMerge key foreignPartition =
-  void $ runPartitionPowerStateT key (PM.merge foreignPartition)
-
-
-{- | Handle the state transition for a cluster merge event. -}
-clusterMerge :: (
-      Default s,
-      Eq e,
-      Event e o s,
-      Indexable s,
-      MonadLoggerIO m,
-      MonadThrow m,
-      Show e,
-      Show s
-    )
-  => ClusterPowerState
-  -> SM e o s m ()
-clusterMerge foreignCluster = do
-  runClusterPowerStateT (PM.merge foreignCluster)
-  nodeState@NodeState {lastRebalance, cluster, self} <- getNodeState
-  $(logDebug) . pack
-    $ "Next Rebalance: "
-    ++ show (lastRebalance, C.nextAction cluster, nodeState)
-  case C.nextAction cluster of
-    (ord, Invite peer keys) | ord > lastRebalance && peer == self -> do
-      {-
-        The current action is an Invite, and this peer is the target.
-
-        Send the join request message to every peer, update lastRebalance
-        so we don't repeat this on every trivial cluster merge, update
-        the expected joins so we can keep track of progress, then sit
-        back and wait.
-      -}
-      let
-        askPeers =
-          Set.toList . Set.delete self . Map.keysSet . C.getPeers $ cluster
-      pushActions [
-          PartitionJoin p keys
-          | p <- askPeers
-        ]
-      modifyNodeState (\ns -> ns {
-          joins = Map.fromList [
-              (p, keys)
-              | p <- askPeers
-            ],
-          lastRebalance = ord
-        })
-    (ord, Drop peer keys) | ord > lastRebalance && peer == self -> do
-      persistence <- getPersistence
-      runConduit (
-          transPipe liftIO (list persistence)
-          =$= CL.map fst
-          =$= CL.filter (`KS.member` keys)
-          =$= awaitForever (\key ->
-              lift $ runPartitionPowerStateT key (
-                  PM.disassociate self
-                )
-            )
-        )
-      modifyNodeState (\ns -> ns {
-          lastRebalance = ord
-        })
-      runClusterPowerStateT C.finishRebalance
-    _ -> return ()
-
-
-{- | Eject a peer from the cluster.  -}
-eject :: (MonadLoggerIO m, MonadThrow m) => Peer -> SM e o s m ()
-eject peer = do
-  {-
-    We need to think very hard about the split brain problem. A random
-    thought about that is that we should consider the extreme case where
-    the network just fails completely and every node believes that every
-    other node should be or has been ejected. This would obviously be
-    catastrophic in terms of data durability unless we have some way to
-    reintegrate an ejected node. So, either we have to guarantee that
-    such a situation can never happen, or else implement a reintegration
-    strategy.  It might be acceptable for the reintegration strategy to
-    be very costly if it is characterized as an extreme recovery scenario.
-
-    Question: would a reintegration strategy become less costly if the
-    "next state id" for a peer were global across all power states
-    instead of local to each power state?
-  -}
-  runClusterPowerStateT (C.eject peer)
-  {-
-    'runClusterPowerStateT (C.eject peer)' will cause us to attempt to
-    notify the peer that they have been ejected, but that notification
-    is almost certainly going to go unacknowledged because the peer
-    is probably down.
-    
-    This call to 'eject' was presumably invoked as a result of user
-    action, and we must therefore trust the user to know that the peer
-    is really down and not coming back. This "guarantee" allows us to
-    acknowledge the ejection on the peer's behalf.
-
-    This call will acknowledge the drop on behalf of the peer, and also
-    remove that peer from the keyspace distribution map.
-  -}
-  runClusterPowerStateTAs peer (return ())
-
-
-{- | Handle a peer join request.  -}
-join :: (MonadIO m, MonadThrow m)
-  => BSockAddr
-  -> SM e o s m (Peer, ClusterPowerState)
-join peerAddr = do
-  peer <- newPeer
-  void $ runClusterPowerStateT (C.joinCluster peer peerAddr)
-  NodeState {cluster} <- getNodeState
-  return (peer, cluster)
-
-
-{- |
-  Figure out the set of nodes to which search requests should be
-  dispatched. "Minimum complete service set" means the minimum set
-  of peers that, together, service the whole partition key space;
-  thereby guaranteeing that if any particular partition is indexed,
-  the corresponding index record will exist on one of these peers.
-
-  Implementation considerations:
-
-  There will usually be more than one solution for the MCSS. For now,
-  we just compute a deterministic solution, but we should implement
-  a random (or pseudo-random) solution in order to maximally balance
-  cluster resources.
-
-  Also, it is not clear that the minimum complete service set is even
-  what we really want. MCSS will reduce overall network utilization,
-  but it may actually increase latency. If we were to dispatch redundant
-  requests to multiple nodes, we could continue with whichever request
-  returns first, and ignore the slow responses. This is probably the
-  best solution. We will call this "fastest competitive search".
-
-  TODO: implement fastest competitive search.
--}
-minimumCompleteServiceSet :: (Monad m) => SM e o s m (Set Peer)
-minimumCompleteServiceSet = do
-  NodeState {cluster} <- getNodeState
-  return (D.minimumCompleteServiceSet (C.getDistribution cluster))
-
-
-{- |
-  Search the index, and return the first record that is __strictly
-  greater than__ the provided search tag, if such a record exists.
--}
-search :: (Monad m) => SearchTag -> SM e o s m (Maybe IndexRecord)
-search SearchTag {stTag, stKey = Nothing} = do
-  NodeState {nsIndex} <- getNodeState
-  return (Set.lookupGE IndexRecord {irTag = stTag, irKey = minBound} nsIndex)
-search SearchTag {stTag, stKey = Just key} = do
-  NodeState {nsIndex} <- getNodeState
-  return (Set.lookupGT IndexRecord {irTag = stTag, irKey = key} nsIndex)
-
-
-{- |
-  Allow a peer to participate in the replication of the partition that is
-  __greater than or equal to__ the indicated partition key. Returns @Nothing@
-  if there is no such partition, or @Just (key, partition)@ where @key@ is the
-  partition key that was joined and @partition@ is the resulting partition
-  power state.
--}
-joinNext :: (
-      Default s,
-      Eq e,
-      Event e o s,
-      Indexable s,
-      MonadLoggerIO m,
-      MonadThrow m
-    )
-  => Peer
-  -> KeySet
-  -> SM e o s m (Maybe (PartitionKey, PartitionPowerState e o s))
-joinNext peer askKeys = do
-  persistence <- getPersistence
-  runConduit (
-      transPipe liftIO (list persistence)
-      =$= CL.filter ((`KS.member` askKeys) . fst)
-      =$= CL.head
-    ) >>= \case
-      Nothing -> return Nothing
-      Just (gotKey, partition) -> do
-        {-
-          This is very similar to the 'runPartitionPowerStateT' code,
-          but there are some important differences. First, 'list' has
-          already done to the trouble of fetching the partition value,
-          so we don't want to have 'runPartitionPowerStateT' do it
-          again. Second, and more importantly, 'runPartitionPowerStateT'
-          will cause a 'PartitionMerge' message to be sent to @peer@, but
-          that message would be redundant, because it contains a subset
-          of the information contained within the 'JoinNextResponse'
-          message that this function produces.
-        -}
-        NodeState {self} <- getNodeState
-        PM.runPowerStateT self partition (do
-            PM.participate peer
-            PM.acknowledge
-          ) >>= \case
-            Left err -> throwM err
-            Right ((), action, partition2, _infOutputs) -> do
-              case action of
-                Send -> pushActions [
-                    PartitionMerge p gotKey partition2
-                    | p <- Set.toList (PS.allParticipants partition2)
-                      {-
-                        Don't send a 'PartitionMerge' to @peer@. We
-                        are already going to send it a more informative
-                        'JoinNextResponse'
-                      -}
-                    , p /= peer
-                    , p /= self
-                  ]
-                DoNothing -> return ()
-              savePartition gotKey partition2
-              return (Just (gotKey, partition2))
-
-
-{- | Receive the result of a JoinNext request. -}
-joinNextResponse :: (
-      Default s,
-      Eq e,
-      Event e o s,
-      Indexable s,
-      MonadLoggerIO m,
-      MonadThrow m,
-      Show e,
-      Show s
-    )
-  => Peer
-  -> Maybe (PartitionKey, PartitionPowerState e o s)
-  -> SM e o s m ()
-joinNextResponse peer response = do
-  NodeState {cluster, lastRebalance} <- getNodeState
-  if lastRebalance > fst (C.nextAction cluster)
-    then
-      {- We are receiving messages from an old rebalance. Log and ignore. -}
-      $(logWarn) . pack
-        $ "Received an old join response: "
-        ++ show (peer, response, cluster, lastRebalance)
-    else do
-      case response of
-        Just (key, partition) -> do
-          partitionMerge key partition
-          NodeState {joins} <- getNodeState
-          case (KS.\\ KS.fromRange minBound key) <$> Map.lookup peer joins of
-            Nothing ->
-              {- An unexpected peer sent us this message, Ignore. TODO log. -}
-              return ()
-            Just needsJoinSet -> do
-              unless (KS.null needsJoinSet)
-                (pushActions [PartitionJoin peer needsJoinSet])
-              modifyNodeState (\ns -> ns {
-                  joins = Map.filter
-                    (not . KS.null)
-                    (Map.insert peer needsJoinSet joins)
-                })
-        Nothing ->
-          modifyNodeState (\ns@NodeState {joins} -> ns {
-              joins = Map.delete peer joins
-            })
-      Map.null . joins <$> getNodeState >>= bool
-        (return ())
-        (runClusterPowerStateT C.finishRebalance)
-
-
-{- |
-  The type of response to a user request, either forward to another node,
-  or respond directly.
--}
-data UserResponse o
-  = Forward Peer
-  | Respond o
-
-
-{- | Get the known peer data from the cluster. -}
-getPeers :: (Monad m) => SM e o s m (Map Peer BSockAddr)
-getPeers = C.getPeers . cluster <$> getNodeState
-
-
-{- | Gets a partition state. -}
-getPartition :: (Default s, MonadIO m)
-  => PartitionKey
-  -> SM e o s m (PartitionPowerState e o s)
-getPartition key = do
-  persistence <- getPersistence
-  NodeState {partitions, cluster} <- getNodeState
-  case Map.lookup key partitions of
-    Nothing ->
-      fromMaybe (PS.new key (C.findOwners key cluster)) <$>
-        liftIO (getState persistence key)
-    Just partition -> return partition
-
-
-{- |
-  Saves a partition state. This function automatically handles the cache
-  for active propagations, as well as reindexing of partitions.
--}
-savePartition :: (Default s, Event e o s, Indexable s, MonadLoggerIO m)
-  => PartitionKey
-  -> PartitionPowerState e o s
-  -> SM e o s m ()
-savePartition key partition = do
-  persistence <- getPersistence
-  oldTags <- indexEntries . PS.projectedValue <$> getPartition key
-  let
-    currentTags = indexEntries (PS.projectedValue partition)
-    {- TODO: maybe use Set.mapMonotonic for performance?  -}
-    obsoleteRecords = Set.map (flip IndexRecord key) (oldTags \\ currentTags)
-    newRecords = Set.map (flip IndexRecord key) currentTags
-
-  $(logDebug) . pack
-    $ "Tagging " ++ show key ++ " with: "
-    ++ show (currentTags, obsoleteRecords, newRecords)
-
-  NodeState {self} <- getNodeState
-  liftIO (saveState persistence key (
-      if self `member` PS.allParticipants partition
-        then Just partition
-        else Nothing
-    ))
-  modifyNodeState (\ns@NodeState {partitions, nsIndex} ->
-      nsIndex `seq`
-      ns {
-          partitions = if Set.null (PS.divergent partition)
-            then
-              {-
-                Remove the partition from the working cache because there
-                is no remaining work that needs to be done to propagage
-                its changes.
-              -}
-              Map.delete key partitions
-            else
-              Map.insert key partition partitions,
-          nsIndex = (nsIndex \\ obsoleteRecords) `Set.union` newRecords
-        }
-    )
-
-
--- {- |
---   Create the log message for origin conflict errors.  The reason this
---   function only creates the log message, instead of doing the logging
---   as well, is because doing the logging here would screw up the source
---   location that the template-haskell logging functions generate for us.
--- -}
--- originError :: (Show o) => DifferentOrigins o -> Text
--- originError (DifferentOrigins a b) = pack
---   $ "Tried to merge powerstates with different origins: "
---   ++ show (a, b)
-
-
-{- | Run a partition-flavored 'PowerStateT' in the 'SM' monad. -}
-runPartitionPowerStateT :: (
-      Default s,
-      Eq e,
-      Event e o s,
-      Indexable s,
-      MonadLoggerIO m,
-      MonadThrow m,
-      Show e,
-      Show s
-    )
-  => PartitionKey
-  -> PartitionPowerStateT e o s (SM e o s m) a
-  -> SM e o s m (a, PartitionPowerState e o s)
-runPartitionPowerStateT key m = do
-    NodeState {self} <- getNodeState
-    partition <- getPartition key
-    PM.runPowerStateT self partition (
-        m <* (removeObsolete >> PM.acknowledge)
-      ) >>= \case
-        Left err -> throwM err
-        Right (a, action, partition2, _infOutputs) -> do
-          case action of
-            Send -> pushActions [
-                PartitionMerge p key partition2
-                | p <- Set.toList (PS.allParticipants partition2)
-                , p /= self
-              ]
-            DoNothing -> return ()
-          $(logDebug) . pack
-            $ "Partition update: " ++ show partition
-            ++ " --> " ++ show partition2 ++ " :: " ++ show action
-          savePartition key partition2
-          return (a, partition2)
-  where
-    {- |
-      Remove obsolete peers. Obsolete peers are peers that are no longer
-      participating in the replication of this partition, due to a
-      rebalance. Such peers are removed lazily here at read time.
-    -}
-    removeObsolete :: (Eq e, Event e o s, Monad m)
-      => PartitionPowerStateT e o s (SM e o s m) ()
-    removeObsolete = do
-      owners <- C.findOwners key . cluster <$> lift getNodeState
-      peers <- PS.projParticipants <$> PM.getPowerState
-      let obsolete = peers \\ owners
-      mapM_
-        (\peer -> PM.disassociate peer >> PM.acknowledgeAs peer)
-        (Set.toList obsolete)
-
-
-{- | Like 'runClusterPowerStateTAs', but run as the local peer. -}
-runClusterPowerStateT :: (MonadThrow m, MonadIO m)
-  => ClusterPowerStateT (SM e o s m) a
-  -> SM e o s m a
-runClusterPowerStateT m = do
-  NodeState {self} <- getNodeState
-  runClusterPowerStateTAs self m
-
-
-{- |
-  Run a clusterstate-flavored 'PowerStateT' in the 'SM' monad,
-  automatically acknowledging the resulting power state.
-
-  Generalized to run as any peer, in order to support exceptional cases
-  like 'eject'.
--}
-runClusterPowerStateTAs :: (MonadThrow m, MonadIO m)
-  => Peer {- ^ The peer to run as. -}
-  -> ClusterPowerStateT (SM e o s m) a
-  -> SM e o s m a
-runClusterPowerStateTAs as m = do
-  NodeState {cluster, self} <- getNodeState
-  PM.runPowerStateT as cluster (m <* PM.acknowledge) >>= \case
-    Left err -> throwM err
-    Right (a, action, cluster2, _outputs) -> do
-      getPersistence >>= \p -> liftIO (saveCluster p self cluster2)
-      case action of
-        Send -> pushActions [
-            ClusterMerge p cluster2
-            | p <- Set.toList (PS.allParticipants cluster2)
-            , p /= self
-          ]
-        DoNothing -> return ()
-      modifyNodeState (\ns -> ns {cluster = cluster2})
-      return a
-
-
diff --git a/src/Network/Legion/StateMachine/Monad.hs b/src/Network/Legion/StateMachine/Monad.hs
deleted file mode 100644
--- a/src/Network/Legion/StateMachine/Monad.hs
+++ /dev/null
@@ -1,165 +0,0 @@
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE OverloadedStrings #-}
-{- |
-  This module contains the legion state machine monad and some
-  primitives for manipulating the state. It is the foundation upon wish
-  the 'Network.Legion.StateMachine' module is built. It is separate from
-  that module because some of the primitives we export here go some small
-  way to avoiding bugs that might arise if that module had direct access
-  to the internals of this monad.
--}
-module Network.Legion.StateMachine.Monad (
-  -- * Run the monad
-  runSM,
-
-  -- * State Inspection
-  getPersistence,
-  getNodeState,
-
-  -- * State Modification
-  modifyNodeState,
-  pushActions,
-  popActions,
-
-  -- * Other symbols
-  SM,
-  NodeState(..),
-  ClusterAction(..),
-) where
-
-import Control.Monad.Catch (MonadThrow)
-import Control.Monad.IO.Class (MonadIO)
-import Control.Monad.Logger (MonadLogger)
-import Control.Monad.Trans.Class (lift, MonadTrans)
-import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)
-import Control.Monad.Trans.State (StateT, runStateT, get, modify, put)
-import Data.Aeson (ToJSON, toJSON, object, (.=), encode)
-import Data.ByteString.Lazy (toStrict)
-import Data.Map (Map)
-import Data.Set (Set)
-import Data.Text (unpack)
-import Data.Text.Encoding (decodeUtf8)
-import Network.Legion.Application (Persistence)
-import Network.Legion.ClusterState (ClusterPowerState, RebalanceOrd)
-import Network.Legion.Distribution (Peer)
-import Network.Legion.Index (IndexRecord)
-import Network.Legion.KeySet (KeySet)
-import Network.Legion.Lift (lift2, lift3)
-import Network.Legion.PartitionKey (PartitionKey)
-import Network.Legion.PartitionState (PartitionPowerState)
-import qualified Data.Map as Map
-
-
-{- |
-  Run an SM action.
--}
-runSM :: (Functor m)
-  => Persistence e o s
-  -> NodeState e o s
-  -> SM e o s m a
-  -> m (a, NodeState e o s, [ClusterAction e o s])
-runSM p ns =
-    fmap flatten
-    . (`runStateT` [])
-    . (`runStateT` ns)
-    . (`runReaderT` p)
-    . unSM
-  where
-    flatten :: ((a, b), c) -> (a, b, c)
-    flatten ((a, b), c) = (a, b, c)
-
-
-{- | Get the handle to the persistence layer. -}
-getPersistence :: (Monad m) => SM e o s m (Persistence e o s)
-getPersistence = SM ask
-
-
-{- | Get the current node state. -}
-getNodeState :: (Monad m) => SM e o s m (NodeState e o s)
-getNodeState = (SM . lift) get
-
-
-{- | Update current node state. -}
-modifyNodeState :: (Monad m)
-  => (NodeState e o s -> NodeState e o s)
-  -> SM e o s m ()
-modifyNodeState = SM . lift . modify
-
-
-{- | Accumulate some cluster propagation actions. -}
-pushActions :: (Monad m) => [ClusterAction e o s] -> SM e o s m ()
-pushActions = SM . lift2 . modify . flip (++)
-
-
-{- | Return and reset the accumulated cluster actions. -}
-popActions :: (Monad m) => SM e o s m [ClusterAction e o s]
-popActions = SM . lift2 $ do
-  actions <- get
-  put []
-  return actions
-
-
-{- |
-  This monad encapsulates the global state of the legion node (not
-  counting the runtime stuff, like open connections and what have
-  you).
-
-  The main reason that the state is hidden behind a monad is because part
-  of the sate (i.e. the partition data) lives behind 'IO'.  Therefore,
-  if we want to model the global state of the node as a single unit,
-  we have to do so using a monad.
--}
-newtype SM e o s m a = SM {
-    unSM ::
-      ReaderT (Persistence e o s) (
-      StateT (NodeState e o s) (
-      StateT [ClusterAction e o s]
-      m)) a
-  }
-  deriving (Functor, Applicative, Monad, MonadLogger, MonadIO, MonadThrow)
-instance MonadTrans (SM e o s) where
-  lift = SM . lift3
-
-
-{- |
-  This is the portion of the local node state that is not persistence
-  related.
--}
-data NodeState e o s = NodeState {
-             self :: Peer,
-          cluster :: ClusterPowerState,
-       partitions :: Map PartitionKey (PartitionPowerState e o s),
-          nsIndex :: Set IndexRecord,
-            joins :: Map Peer KeySet,
-    lastRebalance :: RebalanceOrd
-  }
-instance (Show e, Show s) => Show (NodeState e o s) where
-  show = unpack . decodeUtf8 . toStrict . encode
-{-
-  The ToJSON instance is mainly for debugging. The Haskell-generated 'Show'
-  instance is very hard to read.
--}
-instance (Show e, Show s) => ToJSON (NodeState e o s) where
-  toJSON (NodeState self_ cluster_ partitions_ nsIndex_ joins_ lastUpdate_) =
-    object [
-                 "self" .= show self_,
-              "cluster" .= cluster_,
-           "partitions" .= Map.map show (Map.mapKeys show partitions_),
-              "nsIndex" .= show nsIndex_,
-                "joins" .= Map.map show (Map.mapKeys show joins_),
-        "lastRebalance" .= show lastUpdate_
-      ]
-
-
-{- |
-  These are the actions that a node can take which allow it to coordinate
-  with other nodes. It is up to the runtime system to implement the
-  actions.
--}
-data ClusterAction e o s
-  = PartitionMerge Peer PartitionKey (PartitionPowerState e o s)
-  | ClusterMerge Peer ClusterPowerState
-  | PartitionJoin Peer KeySet
-  deriving (Show)
-
-