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.1.0.1
+version:             0.2.0.0
 synopsis:            Distributed, stateful, homogeneous microservice framework.
 description:         Legion is a framework for writing distributed,
                      homogeneous, stateful microservices in Haskell.
@@ -31,7 +31,6 @@
     Network.Legion.Basics
     Network.Legion.ClusterState
     Network.Legion.Conduit
-    Network.Legion.ConnectionManager
     Network.Legion.Distribution
     Network.Legion.Fork
     Network.Legion.KeySet
@@ -41,6 +40,8 @@
     Network.Legion.PowerState
     Network.Legion.Propagation
     Network.Legion.Runtime
+    Network.Legion.Runtime.ConnectionManager
+    Network.Legion.Runtime.PeerMessage
     Network.Legion.Settings
     Network.Legion.StateMachine
     Network.Legion.UUID
@@ -48,12 +49,14 @@
   -- other-extensions:    
   build-depends:
     Ranged-sets >= 0.3.0 && < 0.4,
+    aeson >= 0.11.2.0 && < 0.12,
     attoparsec >= 0.13.0.1 && < 0.14,
     base >= 4.8 && < 4.9,
     binary >= 0.7.5 && < 0.9,
     binary-conduit >= 1.2.3 && < 1.3,
     bytestring >= 0.10.4.0 && < 0.11,
     canteven-http >= 0.1.1.1 && < 0.2,
+    canteven-log >= 1.0.0.0 && < 1.1,
     conduit >= 1.2.4 && < 1.3,
     conduit-extra >= 1.1.9 && < 1.2,
     containers >= 0.5.5.1 && < 0.6,
diff --git a/src/Network/Legion.hs b/src/Network/Legion.hs
--- a/src/Network/Legion.hs
+++ b/src/Network/Legion.hs
@@ -161,7 +161,10 @@
 --------------------------------------------------------------------------------
 
 -- $invocation
--- Notes on invocation.
+-- While this section is being worked on, you can check out the
+-- [legion-cache](https://github.com/taphu/legion-cache) project for a
+-- working example of how to build a basic distributed key-value store
+-- using Legion.
 
 --------------------------------------------------------------------------------
 
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
@@ -7,6 +7,7 @@
 -}
 module Network.Legion.Admin (
   runAdmin,
+  AdminMessage(..),
 ) where
 
 import Canteven.HTTP (requestLogging, logExceptionsAndContinue)
@@ -23,10 +24,11 @@
 import Network.HTTP.Types (notFound404)
 import Network.Legion.Application (LegionConstraints)
 import Network.Legion.Conduit (chanToSource)
+import Network.Legion.Distribution (Peer)
 import Network.Legion.LIO (LIO)
 import Network.Legion.PartitionKey (PartitionKey(K))
-import Network.Legion.StateMachine (AdminMessage(GetState, GetPart,
-  Eject))
+import Network.Legion.PartitionState (PartitionPowerState)
+import Network.Legion.StateMachine (NodeState)
 import Network.Wai (Middleware, modifyResponse)
 import Network.Wai.Handler.Warp (HostPreference, defaultSettings, Port,
   setHost, setPort)
@@ -120,4 +122,19 @@
     -}
     serverValue =
       encodeUtf8 (T.pack ("legion-admin/" ++ showVersion version))
+
+
+{- |
+  The type of messages sent by the admin service.
+-}
+data AdminMessage i o s
+  = GetState (NodeState i s -> LIO ())
+  | GetPart PartitionKey (Maybe (PartitionPowerState i s) -> LIO ())
+  | Eject Peer (() -> LIO ())
+
+instance Show (AdminMessage i o s) where
+  show (GetState _) = "(GetState _)"
+  show (GetPart k _) = "(GetPart " ++ show k ++ " _)"
+  show (Eject p _) = "(Eject " ++ show p ++ " _)"
+
 
diff --git a/src/Network/Legion/Application.hs b/src/Network/Legion/Application.hs
--- a/src/Network/Legion/Application.hs
+++ b/src/Network/Legion/Application.hs
@@ -1,4 +1,4 @@
-{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE ConstraintKinds #-}
 {- |
   This module contains the data types necessary for implementing the
   user application.
@@ -26,15 +26,23 @@
   >   Show o, Show s, Eq i
   > )
 -}
-class (
+type LegionConstraints i o s = (
     ApplyDelta i s, Default s, Binary i, Binary o, Binary s, Show i,
     Show o, Show s, Eq i
-  ) => LegionConstraints i o s where
+  )
 
 
 {- |
   This is the type of a user-defined Legion application. Implement this and
   allow the Legion framework to manage your cluster.
+
+  - @__i__@ is the type of request your application will handle. @__i__@ stands
+    for __"input"__.
+  - @__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"__.
 -}
 data Legionary i o s = Legionary {
     {- |
diff --git a/src/Network/Legion/BSockAddr.hs b/src/Network/Legion/BSockAddr.hs
--- a/src/Network/Legion/BSockAddr.hs
+++ b/src/Network/Legion/BSockAddr.hs
@@ -11,9 +11,7 @@
   SockAddrCan))
 
 
-{- |
-  A type useful only for creating a `Binary` instance of `SockAddr`.
--}
+{- | A type useful only for creating a `Binary` instance of `SockAddr`.  -}
 newtype BSockAddr = BSockAddr {getAddr :: SockAddr} deriving (Show, Eq)
 
 instance Binary BSockAddr where
@@ -48,6 +46,5 @@
           $ "Can't decode BSockAddr because the constructor tag "
           ++ "was not understood. Probably this data is representing "
           ++ "something else."
-
 
 
diff --git a/src/Network/Legion/ClusterState.hs b/src/Network/Legion/ClusterState.hs
--- a/src/Network/Legion/ClusterState.hs
+++ b/src/Network/Legion/ClusterState.hs
@@ -2,6 +2,7 @@
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 {- |
   This module contains the data types related to the distributed cluster state.
 -}
@@ -24,6 +25,7 @@
   heartbeat,
 ) where
 
+import Data.Aeson (ToJSON, toJSON, object, (.=))
 import Data.Binary (Binary)
 import Data.Default.Class (Default(def))
 import Data.Map (Map)
@@ -59,6 +61,14 @@
       distribution = D.empty,
              peers = Map.empty
     }
+instance ToJSON ClusterState where
+  toJSON ClusterState {distribution, peers} = object [
+      "distribution" .= distribution,
+      "peers" .= Map.fromList [
+          (show p, show a)
+          | (p, a) <- Map.toList peers
+        ]
+    ]
 
 
 {- |
@@ -75,7 +85,7 @@
 -}
 newtype ClusterPropState = ClusterPropState {
     unPropState :: PropState UUID ClusterState Peer Update
-  } deriving (Show)
+  } deriving (Show, ToJSON)
 
 
 {- |
diff --git a/src/Network/Legion/ConnectionManager.hs b/src/Network/Legion/ConnectionManager.hs
deleted file mode 100644
--- a/src/Network/Legion/ConnectionManager.hs
+++ /dev/null
@@ -1,208 +0,0 @@
-{-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE TemplateHaskell #-}
-{- |
-  This module manages connections to other nodes in the cluster.
--}
-module Network.Legion.ConnectionManager (
-  ConnectionManager,
-  newConnectionManager,
-  send,
-  newPeers,
-) where
-
-import Prelude hiding (lookup)
-
-import Control.Concurrent (Chan, writeChan, newChan, readChan)
-import Control.Exception (try, SomeException)
-import Control.Monad (void)
-import Control.Monad.Logger (logInfo, logWarn)
-import Control.Monad.Trans.Class (lift)
-import Data.Binary (Binary, encode)
-import Data.ByteString.Lazy (ByteString)
-import Data.Map (toList, insert, empty, Map, lookup)
-import Data.Text (pack)
-import Network.Legion.BSockAddr (BSockAddr(BSockAddr))
-import Network.Legion.Distribution (Peer)
-import Network.Legion.Fork (forkC)
-import Network.Legion.LIO (LIO)
-import Network.Legion.StateMachine (PeerMessage)
-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))
-import Network.Socket.ByteString.Lazy (sendAll)
-
-{- |
-  A handle on the connection manager
--}
-data ConnectionManager i o s = C (Chan (Message i o s))
-instance Show (ConnectionManager i o s) where
-  show _ = "ConnectionManager"
-
-
-{- |
-  Create a new connection manager.
--}
-newConnectionManager :: (Binary i, Binary o, Binary s)
-  => Map Peer BSockAddr
-  -> LIO (ConnectionManager i o s)
-newConnectionManager initPeers = do
-    chan <- lift newChan
-    forkC "connection manager thread" $
-      manager chan S {connections = empty}
-    let cm = C chan
-    newPeers cm initPeers
-    return cm
-  where
-    manager :: (Binary s, Binary o, Binary i)
-      => Chan (Message i o s)
-      -> State i o s
-      -> LIO ()
-    manager chan state = lift (readChan chan) >>= handle state >>= manager chan
-
-    handle :: (Binary i, Binary o, Binary s)
-      => State i o s
-      -> Message i o s
-      -> LIO (State i 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
-
-
-{- |
-  Build a new connection.
--}
-connection :: (Binary i, Binary o, Binary s)
-  => SockAddr
-  -> LIO (Chan (PeerMessage i o s))
-
-connection addr = do
-    chan <- lift newChan
-    forkC ("connection to: " ++ show addr) $
-      handle chan Nothing
-    return chan
-  where
-    handle :: (Binary i, Binary o, Binary s)
-      => Chan (PeerMessage i o s)
-      -> Maybe Socket
-      -> LIO ()
-    handle chan so =
-      lift (readChan chan) >>= sendWithRetry so . encode >>= handle chan
-
-    {- |
-      Open a socket.
-    -}
-    openSocket :: IO Socket
-    openSocket = do
-      so <- socket (fam addr) Stream defaultProtocol
-      connect so addr
-      return so
-
-    {- |
-      Try to send the payload over the socket, and if that fails, then try to
-      create a new socket and retry sending the payload. Return whatever the
-      "working" socket is.
-    -}
-    sendWithRetry :: Maybe Socket -> ByteString -> LIO (Maybe Socket)
-    sendWithRetry Nothing payload =
-      (lift . try) openSocket >>= \case
-        Left err -> do
-          $(logWarn) . pack
-            $ "Can't connect to: " ++ show addr ++ ". Dropping message on "
-            ++ "the floor: " ++ show payload ++ ". The error was: "
-            ++ show (err :: SomeException)
-          return Nothing
-        Right so -> do
-          result2 <- (lift . try) (sendAll so payload)
-          case result2 of
-            Left err -> $(logWarn) . pack
-              $ "An error happend when trying to send a payload over a socket "
-              ++ "to the address: " ++ show addr ++ ". The error was: "
-              ++ show (err :: SomeException) ++ ". This is the last straw, we "
-              ++ "are not retrying. The message is being dropped on the floor. "
-              ++ "The message was: " ++ show payload
-            Right _ -> return ()
-          return (Just so)
-    sendWithRetry (Just so) payload =
-      (lift . try) (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 ()))
-          sendWithRetry Nothing payload
-        Right _ ->
-          return (Just so)
-
-
-{- |
-  Send a message to a peer.
--}
-send
-  :: ConnectionManager i o s
-  -> Peer
-  -> PeerMessage i o s
-  -> LIO ()
-send (C chan) peer = lift . writeChan chan . Send peer
-
-
-{- |
-  Tell the connection manager about a new peer.
--}
-newPeer
-  :: ConnectionManager i o s
-  -> Peer
-  -> SockAddr
-  -> LIO ()
-newPeer (C chan) peer addr = lift $ writeChan chan (NewPeer peer addr)
-
-
-{- |
-  Tell the connection manager about all the peers known to the cluster state.
--}
-newPeers :: ConnectionManager i o s -> Map Peer BSockAddr -> LIO ()
-newPeers cm peers =
-    mapM_ oneNewPeer (toList peers)
-  where
-    oneNewPeer (peer, BSockAddr addy) = newPeer cm peer addy
-
-
-{- |
-  The internal state of the connection manager.
--}
-data State i o s = S {
-    connections :: Map Peer (Chan (PeerMessage i o s))
-  }
-
-
-{- |
-  The types of messages that the ConnectionManager understands.
--}
-data Message i o s
-  = NewPeer Peer SockAddr
-  | Send Peer (PeerMessage i 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
-
-
diff --git a/src/Network/Legion/Distribution.hs b/src/Network/Legion/Distribution.hs
--- a/src/Network/Legion/Distribution.hs
+++ b/src/Network/Legion/Distribution.hs
@@ -17,10 +17,12 @@
 
 import Prelude hiding (null)
 
+import Data.Aeson (ToJSON, toJSON, object, (.=))
 import Data.Binary (Binary)
 import Data.Function (on)
 import Data.List (sort, sortBy)
 import Data.Set (Set, toList)
+import Data.Text (pack)
 import Data.UUID (UUID)
 import GHC.Generics (Generic)
 import Network.Legion.KeySet (KeySet, member, (\\), null)
@@ -47,6 +49,11 @@
 newtype ParticipationDefaults = D {
     unD :: [(KeySet, Set Peer)]
   } deriving (Show, Binary)
+instance ToJSON ParticipationDefaults where
+  toJSON (D dist) = object [
+      pack (show ks) .= Set.map show peers
+      | (ks, peers) <- dist
+    ]
 
 
 {- |
@@ -134,38 +141,6 @@
 
     weightOf p = sum [KS.size ks | (ks, ps) <- dist, p `Set.member` ps]
 
---     -- TODO: first figure out if any replicas need re-building.
---     case sortBy (weight `on` snd) (toList dist) of
---       (p, keyspace):remaining | p == peer ->
---         case reverse remaining of
---           [] -> Nothing
---           (target, targetSpace):_ ->
---             let
---                 {- |
---                   Keys that already exist at the target are not eligible
---                   for being moved.
---                 -}
---                 eligibleSpace = keyspace \\ targetSpace
---                 migrationSize = (size keyspace - size targetSpace) `div` 2
---                 migrants = pickMigrants migrationSize eligibleSpace
---             in
---             case migrants of
---               Just keys -> Just (Move target keys)
---               Nothing -> Nothing
---       _ -> Nothing
---   where
---     weight
---       :: KeySet
---       -> KeySet
---       -> Ordering
---     weight = flip compare `on` size
--- 
---     pickMigrants :: Integer -> KeySet -> Maybe KeySet
---     pickMigrants n keyspace =
---       let migrants = take n keyspace in
---       if size migrants > 0
---         then Just migrants
---         else Nothing 
 
 
 {- |
diff --git a/src/Network/Legion/PartitionKey.hs b/src/Network/Legion/PartitionKey.hs
--- a/src/Network/Legion/PartitionKey.hs
+++ b/src/Network/Legion/PartitionKey.hs
@@ -19,9 +19,7 @@
 import Data.Word (Word64)
 
 
-{- |
-  This is how partitions are identified and referenced.
--}
+{- | This is how partitions are identified and referenced. -}
 newtype PartitionKey = K {unkey :: Word256} deriving (Eq, Ord, Show, Bounded)
 
 instance Binary PartitionKey where
@@ -35,9 +33,7 @@
   adjacentBelow (K k) = if k == minBound then Nothing else Just (K (pred k))
 
 
-{- |
-  Convert a `PartitionKey` into a hex string
--}
+{- | Convert a `PartitionKey` into a hex string. -}
 toHex :: PartitionKey -> String
 toHex (K (Word256 (Word128 a b) (Word128 c d))) =
   concatMap toHex64 [a, b, c, d]
@@ -76,9 +72,7 @@
         (True,  True,  True,  True)  -> 'f'
 
 
-{- |
-  Maybe convert a hex string into a partition key
--}
+{- | Maybe convert a hex string into a partition key -}
 fromHex :: String -> Either String PartitionKey
 fromHex str
     | length str > 64 =
diff --git a/src/Network/Legion/PartitionState.hs b/src/Network/Legion/PartitionState.hs
--- a/src/Network/Legion/PartitionState.hs
+++ b/src/Network/Legion/PartitionState.hs
@@ -19,8 +19,10 @@
   projParticipants,
   projected,
   infimum,
+  complete,
 ) where
 
+import Data.Aeson (ToJSON)
 import Data.Binary (Binary)
 import Data.Default.Class (Default)
 import Data.Set (Set)
@@ -52,7 +54,7 @@
 -}
 newtype PartitionPropState i s = PartitionPropState {
     unPropState :: PropState PartitionKey s Peer i
-  } deriving (Eq, Show)
+  } deriving (Eq, Show, ToJSON)
 
 
 -- {- |
@@ -178,5 +180,15 @@
 -}
 infimum :: PartitionPowerState i s -> s
 infimum = P.infimum . unPowerState
+
+
+{- |
+  Figure out if this propagation state has any work to do. Return 'True' if all
+  known propagation work has been completed. The implication here is that the
+  only way more work can happen is if new deltas are applied, either directly
+  or via a merge.
+-}
+complete :: PartitionPropState i s -> Bool
+complete = P.complete . unPropState
 
 
diff --git a/src/Network/Legion/PowerState.hs b/src/Network/Legion/PowerState.hs
--- a/src/Network/Legion/PowerState.hs
+++ b/src/Network/Legion/PowerState.hs
@@ -1,6 +1,7 @@
 {-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 {- |
   This module contains the fundamental distributed data object.
 -}
@@ -28,6 +29,7 @@
 
 import Prelude hiding (null)
 
+import Data.Aeson (ToJSON, toJSON, object, (.=))
 import Data.Binary (Binary(put, get))
 import Data.Default.Class (Default(def))
 import Data.DoubleWord (Word256(Word256), Word128(Word128))
@@ -50,6 +52,15 @@
      deltas :: Map (StateId p) (Delta p d, Set p)
   } deriving (Generic, Show, Eq)
 instance (Binary o, Binary s, Binary p, Binary d) => Binary (PowerState o s p d)
+instance (Show o, Show s, Show p, Show d) => ToJSON (PowerState o s p d) where
+  toJSON PowerState {origin, infimum, deltas} = object [
+      "origin" .= show origin,
+      "infimum" .= infimum,
+      "deltas" .= Map.fromList [
+          (show sid, (show d, Set.map show ps))
+          | (sid, (d, ps)) <- Map.toList deltas
+        ]
+    ]
 
 
 {- |
@@ -66,6 +77,12 @@
   Infimum s1 _ _ == Infimum s2 _ _ = s1 == s2
 instance (Ord p) => Ord (Infimum s p) where
   compare (Infimum s1 _ _) (Infimum s2 _ _) = compare s1 s2
+instance (Show s, Show p) => ToJSON (Infimum s p) where
+  toJSON Infimum {stateId, participants, stateValue} = object [
+      "stateId" .= show stateId,
+      "participants" .= Set.map show participants,
+      "stateValue" .= show stateValue
+    ]
 
 
 {- |
diff --git a/src/Network/Legion/Propagation.hs b/src/Network/Legion/Propagation.hs
--- a/src/Network/Legion/Propagation.hs
+++ b/src/Network/Legion/Propagation.hs
@@ -1,6 +1,7 @@
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 {- |
   This module defines how to propagate a PowerState amoung its participants.
 -}
@@ -26,10 +27,12 @@
   projParticipants,
   projected,
   infimum,
+  complete,
 ) where
 
 import Prelude hiding (lookup)
 
+import Data.Aeson (ToJSON, object, (.=), toJSON)
 import Data.Binary (Binary)
 import Data.Default.Class (Default)
 import Data.Map (Map, lookup)
@@ -68,6 +71,16 @@
           self :: p,
            now :: Time
   } deriving (Eq, Show)
+instance (Show o, Show s, Show p, Show d) => ToJSON (PropState o s p d) where
+  toJSON PropState {powerState, peerStates, self, now} = object [
+      "powerState" .= powerState,
+      "peerStates" .= Map.fromList [
+          (show p, show s)
+          | (p, s) <- Map.toList peerStates
+        ],
+      "self" .= show self,
+      "now" .= show now
+    ]
 
 
 {- |
@@ -364,5 +377,16 @@
 -}
 infimum :: PropPowerState o s p d -> s
 infimum = PS.infimumValue . unPowerState
+
+
+{- |
+  Figure out if this propagation state has any work to do. Return 'True' if all
+  known propagation work has been completed. The implication here is that the
+  only way more work can happen is if new deltas are applied, either directly
+  or via a merge.
+-}
+complete :: (Ord p) => PropState o s p d -> Bool
+complete PropState {powerState, peerStates} =
+  Map.null peerStates && Set.null (divergent powerState)
 
 
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,3 +1,4 @@
+{-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -17,45 +18,52 @@
 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 (void, forever, join, (>=>))
 import Control.Monad.Catch (catchAll, try, SomeException, throwM)
-import Control.Monad.IO.Class (liftIO)
+import Control.Monad.IO.Class (MonadIO, liftIO)
 import Control.Monad.Logger (logWarn, logError, logInfo, LoggingT,
-  MonadLoggerIO, runLoggingT, askLoggerIO)
+  MonadLoggerIO, runLoggingT, askLoggerIO, logDebug)
 import Control.Monad.Trans.Class (lift)
-import Data.Binary (encode)
+import Data.Binary (encode, Binary)
 import Data.Conduit (Source, ($$), (=$=), yield, await, awaitForever,
-  transPipe, ConduitM, runConduit)
+  transPipe, ConduitM, runConduit, Sink)
 import Data.Conduit.Network (sourceSocket)
 import Data.Conduit.Serialization.Binary (conduitDecode)
 import Data.Map (Map)
 import Data.Text (pack)
-import Network.Legion.Admin (runAdmin)
-import Network.Legion.Application (LegionConstraints, Legionary,
-  RequestMsg)
+import GHC.Generics (Generic)
+import Network.Legion.Admin (runAdmin, AdminMessage(GetState, GetPart,
+  Eject))
+import Network.Legion.Application (LegionConstraints,
+  Legionary(Legionary), RequestMsg, persistence, getState)
 import Network.Legion.BSockAddr (BSockAddr(BSockAddr))
 import Network.Legion.ClusterState (ClusterPowerState)
 import Network.Legion.Conduit (merge, chanToSink, chanToSource)
-import Network.Legion.ConnectionManager (newConnectionManager, send,
-  newPeers)
 import Network.Legion.Distribution (Peer, newPeer)
 import Network.Legion.Fork (forkC)
 import Network.Legion.LIO (LIO)
 import Network.Legion.PartitionKey (PartitionKey)
+import Network.Legion.Runtime.ConnectionManager (newConnectionManager,
+  send, ConnectionManager, newPeers)
+import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),
+  PeerMessagePayload(ForwardRequest, ForwardResponse, ClusterMerge,
+  PartitionMerge), MessageId, newSequence, next)
 import Network.Legion.Settings (LegionarySettings(LegionarySettings,
   adminHost, adminPort, peerBindAddr, joinBindAddr))
-import Network.Legion.StateMachine (stateMachine, LInput(J, P, R,
-  A), JoinRequest(JoinRequest), JoinResponse(JoinOk, JoinRejected),
-  LOutput(Send, NewPeers), AdminMessage, NodeState, PeerMessage,
-  newNodeState)
+import Network.Legion.StateMachine (partitionMerge, clusterMerge,
+  NodeState, newNodeState, runSM, UserResponse(Forward, Respond),
+  userRequest, heartbeat, rebalance, migrate, propagate, ClusterAction,
+  eject)
 import Network.Legion.UUID (getUUID)
 import Network.Socket (Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN),
-  SocketOption(ReuseAddr), SocketType(Stream), accept, bindSocket,
+  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 qualified Data.Conduit.List as CL
+import qualified Data.Map as Map
 import qualified Network.Legion.ClusterState as C
+import qualified Network.Legion.StateMachine as SM
 
 
 {- |
@@ -70,17 +78,17 @@
 -}
 runLegionary :: (LegionConstraints i o s)
   => Legionary i o s
-    -- ^ The user-defined legion application to run.
+    {- ^ The user-defined legion application to run.  -}
   -> LegionarySettings
-    -- ^ Settings and configuration of the legionary framework.
+    {- ^ Settings and configuration of the legionary framework.  -}
   -> StartupMode
   -> Source IO (RequestMsg i o)
-    -- ^ A source of requests, together with a way to respond to the requets.
+    {- ^ A source of requests, together with a way to respond to the requets. -}
+  -> LoggingT IO ()
     {-
-      We don't use `LIO` in the type signature here because we don't
-      export the `LIO` symbol.
+      Don't expose 'LIO' here because 'LIO' is a strictly internal
+      symbol. 'LoggingT IO' is what we expose to the world.
     -}
-  -> LoggingT IO ()
 
 runLegionary
     legionary
@@ -88,31 +96,36 @@
     startupMode
     requestSource
   = do
+    {- Start the various messages sources.  -}
     peerS <- loggingC =<< startPeerListener settings
-    (nodeState, peers) <- makeNodeState settings startupMode
-    cm <- newConnectionManager peers
-    $(logInfo) . pack
-      $ "The initial node state is: " ++ show nodeState
     adminS <- loggingC =<< runAdmin adminPort adminHost
     joinS <- loggingC (joinMsgSource settings)
+
+    (self, nodeState, peers) <- makeNodeState settings startupMode
+    cm <- newConnectionManager peers
+
+    firstMessageId <- newSequence
+    let
+      rts = RuntimeState {
+          forwarded = Map.empty,
+          nextId = firstMessageId,
+          cm,
+          self
+        }
     runConduit $
       (joinS `merge` (peerS `merge` (requestSource `merge` adminS)))
         =$= CL.map toMessage
-        =$= stateMachine legionary nodeState
-        =$= handleOutput cm
+        =$= messageSink legionary (rts, nodeState)
   where
-    handleOutput cm = awaitForever (lift . \case
-        Send peer message -> send cm peer message
-        NewPeers peers -> newPeers cm peers
-      )
-
     toMessage
       :: Either
           (JoinRequest, JoinResponse -> LIO ())
           (Either
             (PeerMessage i o s)
-            (Either (RequestMsg i o) (AdminMessage i o s)))
-      -> LInput i o s
+            (Either
+              (RequestMsg i o)
+              (AdminMessage i o s)))
+      -> RuntimeMessage i o s
     toMessage (Left m) = J m
     toMessage (Right (Left m)) = P m
     toMessage (Right (Right (Left m))) = R m
@@ -128,16 +141,228 @@
       return (transPipe (`runLoggingT` logging) c)
 
 
-{- | This defines the various ways a node can be spun up.  -}
+messageSink :: (LegionConstraints i o s)
+  => Legionary i o s
+  -> (RuntimeState i o s, NodeState i s)
+  -> Sink (RuntimeMessage i o s) LIO ()
+messageSink legionary states =
+    await >>= \case
+      Nothing -> return ()
+      Just msg -> do
+        $(logDebug) . pack
+          $ "Receieved: " ++ show msg
+        lift . handleMessage legionary msg
+          >=> lift . updatePeers legionary
+          >=> lift . clusterHousekeeping legionary
+          >=> messageSink legionary
+          $ states
+
+
+{- |
+  Make sure the connection manager knows about any new peers that have
+  joined the cluster.
+-}
+updatePeers
+  :: Legionary i o s
+  -> (RuntimeState i o s, NodeState i s)
+  -> LIO (RuntimeState i o s, NodeState i s)
+updatePeers legionary (rts, ns) = do
+  (peers, ns2) <- runSM legionary ns SM.getPeers
+  newPeers (cm rts) peers
+  return (rts, ns2)
+
+
+{- |
+  Perform any cluster management actions, and update the state
+  appropriately.
+-}
+clusterHousekeeping :: (LegionConstraints i o s)
+  => Legionary i o s
+  -> (RuntimeState i o s, NodeState i s)
+  -> LIO (RuntimeState i o s, NodeState i s)
+clusterHousekeeping legionary (rts, ns) = do
+    (actions, ns2) <- runSM legionary ns (
+        heartbeat
+        >> rebalance
+        >> migrate
+        >> propagate
+      )
+    rts2 <- foldr (>=>) return (clusterAction <$> actions) rts
+    return (rts2, ns2)
+
+
+{- |
+  Actually perform a cluster action as directed by the state
+  machine.
+-}
+clusterAction
+  :: ClusterAction i s
+  -> RuntimeState i o s
+  -> LIO (RuntimeState i o s)
+
+clusterAction
+    (SM.ClusterMerge peer ps)
+    rts@RuntimeState {self, nextId, cm}
+  = do
+    send cm peer (PeerMessage self nextId (ClusterMerge ps))
+    return rts {nextId = next nextId}
+
+clusterAction
+    (SM.PartitionMerge peer key ps)
+    rts@RuntimeState {self, nextId, cm}
+  = do
+    send cm peer (PeerMessage self nextId (PartitionMerge key ps))
+    return rts {nextId = next nextId}
+
+
+{- |
+  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 i o s)
+  => Legionary i o s
+  -> RuntimeMessage i o s
+  -> (RuntimeState i o s, NodeState i s)
+  -> LIO (RuntimeState i o s, NodeState i s)
+
+handleMessage {- Partition Merge -}
+    legionary
+    (P (PeerMessage source _ (PartitionMerge key ps)))
+    (rts, ns)
+  = do
+    ((), ns2) <- runSM legionary ns (partitionMerge source key ps)
+    return (rts, ns2)
+  
+handleMessage {- Cluster Merge -}
+    legionary
+    (P (PeerMessage source _ (ClusterMerge cs)))
+    (rts, ns)
+  = do
+    ((), ns2) <- runSM legionary ns (clusterMerge source cs)
+    return (rts, ns2)
+
+handleMessage {- Forward Request -}
+    legionary
+    (P (msg@(PeerMessage source mid (ForwardRequest key request))))
+    (rts@RuntimeState {nextId, cm, self}, ns)
+  = do
+    (output, ns2) <- runSM legionary ns (userRequest key request)
+    case output of
+      Respond response -> do
+        send cm source (
+            PeerMessage self nextId (ForwardResponse mid response)
+          )
+        return (rts {nextId = next nextId}, ns2)
+      Forward peer -> do
+        send cm peer msg
+        return (rts {nextId = next nextId}, ns2)
+    
+handleMessage {- Forward Response -}
+    _legionary
+    (msg@(P (PeerMessage _ _ (ForwardResponse mid response))))
+    (rts, ns)
+  =
+    case lookupDelete mid (forwarded rts) of
+      (Nothing, fwd) -> do
+        $(logWarn) . pack $ "Unsolicited ForwardResponse: " ++ show msg
+        return (rts {forwarded = fwd}, ns)
+      (Just respond, fwd) -> do
+        respond response
+        return (rts {forwarded = fwd}, ns)
+
+handleMessage {- User Request -}
+    legionary
+    (R ((key, request), respond))
+    (rts@RuntimeState {self, cm, nextId, forwarded}, ns)
+  = do
+    (output, ns2) <- runSM legionary ns (userRequest key request)
+    case output of
+      Respond response -> do
+        lift (respond response)
+        return (rts, ns2)
+      Forward peer -> do
+        send cm peer (
+            PeerMessage self nextId (ForwardRequest key request)
+          )
+        return (
+            rts {
+              forwarded = Map.insert nextId (lift . respond) forwarded,
+              nextId = next nextId
+            },
+            ns2
+          )
+
+handleMessage {- Join Request -}
+    legionary
+    (J (JoinRequest addy, respond))
+    (rts, ns)
+  = do
+    ((peer, cluster), ns2) <- runSM legionary ns (SM.join addy)
+    respond (JoinOk peer cluster)
+    return (rts, ns2)
+
+handleMessage {- Admin Get State -}
+    _legionary
+    (A (GetState respond))
+    (rts, ns)
+  =
+    respond ns >> return (rts, ns)
+
+handleMessage {- Admin Get Partition -}
+    Legionary {persistence}
+    (A (GetPart key respond))
+    (rts, ns)
+  = do
+    respond =<< lift (getState persistence key)
+    return (rts, ns)
+
+handleMessage {- Admin Eject Peer -}
+    legionary
+    (A (Eject peer respond))
+    (rts, ns)
+  = do
+    {-
+      TODO: we should attempt to notify the ejected peer that it has
+      been ejected instead of just cutting it off and washing our hands
+      of it. I have a vague notion that maybe ejected peers should be
+      permanently recorded in the cluster state so that if they ever
+      reconnect then we can notify them that they are no longer welcome
+      to participate.
+
+      On a related note, 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?
+    -}
+    ((), ns2) <- runSM legionary ns (eject peer)
+    respond ()
+    return (rts, ns2)
+
+
+{- | This defines the various ways a node can be spun up. -}
 data StartupMode
   = NewCluster
-    -- ^ Indicates that we should bootstrap a new cluster at startup. The
-    --   persistence layer may be safely pre-populated because the new
-    --   node will claim the entire keyspace. 
+    {- ^
+      Indicates that we should bootstrap a new cluster at startup. The
+      persistence layer may be safely pre-populated because the new node
+      will claim the entire keyspace.
+    -}
   | JoinCluster SockAddr
-    -- ^ Indicates that the node should try to join an existing cluster,
-    --   either by starting fresh, or by recovering from a shutdown
-    --   or crash.
+    {- ^
+      Indicates that the node should try to join an existing cluster,
+      either by starting fresh, or by recovering from a shutdown or crash.
+    -}
   deriving (Show, Eq)
 
 
@@ -157,7 +382,7 @@
           inputChan <- newChan
           so <- socket (fam peerBindAddr) Stream defaultProtocol
           setSocketOption so ReuseAddr 1
-          bindSocket so peerBindAddr
+          bind so peerBindAddr
           listen so 5
           return (inputChan, so)
         forkC "peer socket acceptor" $ acceptLoop so inputChan
@@ -182,7 +407,7 @@
             let runSocket =
                   sourceSocket conn
                   =$= conduitDecode
-                  $$  msgSink
+                  $$ msgSink
             void
               . lift
               . forkIO
@@ -210,18 +435,19 @@
 
 
 {- | Figure out how to construct the initial node state.  -}
-makeNodeState :: (LegionConstraints i o s)
+makeNodeState :: (Show i)
   => LegionarySettings
   -> StartupMode
-  -> LIO (NodeState i o s, Map Peer BSockAddr)
+  -> LIO (Peer, NodeState i s, Map Peer BSockAddr)
 
 makeNodeState LegionarySettings {peerBindAddr} NewCluster = do
   {- Build a brand new node state, for the first node in a cluster. -}
   self <- newPeer
   clusterId <- getUUID
-  let cluster = C.new clusterId self peerBindAddr
-  nodeState <- newNodeState self cluster
-  return (nodeState, C.getPeers cluster)
+  let
+    cluster = C.new clusterId self peerBindAddr
+    nodeState = newNodeState self cluster
+  return (self, nodeState, C.getPeers cluster)
 
 makeNodeState LegionarySettings {peerBindAddr} (JoinCluster addr) = do
     {-
@@ -230,9 +456,10 @@
     -}
     $(logInfo) "Trying to join an existing cluster."
     (self, clusterPS) <- joinCluster (JoinRequest (BSockAddr peerBindAddr))
-    let cluster = C.initProp self clusterPS
-    nodeState <- newNodeState self cluster
-    return (nodeState, C.getPeers cluster)
+    let
+      cluster = C.initProp self clusterPS
+      nodeState = newNodeState self cluster
+    return (self, nodeState, C.getPeers cluster)
   where
     joinCluster :: JoinRequest -> LIO (Peer, ClusterPowerState)
     joinCluster joinMsg = liftIO $ do
@@ -268,7 +495,7 @@
           chan <- newChan
           so <- socket (fam joinBindAddr) Stream defaultProtocol
           setSocketOption so ReuseAddr 1
-          bindSocket so joinBindAddr
+          bind so joinBindAddr
           listen so 5
           return (chan, so)
         forkC "join socket acceptor" $ acceptLoop so chan
@@ -320,7 +547,7 @@
           )
 
 
-{- | Guess the family of a `SockAddr`.  -}
+{- | Guess the family of a `SockAddr`. -}
 fam :: SockAddr -> Family
 fam SockAddrInet {} = AF_INET
 fam SockAddrInet6 {} = AF_INET6
@@ -332,13 +559,13 @@
   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.
 -}
-forkLegionary :: (LegionConstraints i o s, MonadLoggerIO io)
+forkLegionary :: (LegionConstraints i o s, MonadLoggerIO io, MonadIO io2)
   => Legionary i o s
     {- ^ The user-defined legion application to run. -}
   -> LegionarySettings
     {- ^ Settings and configuration of the legionary framework. -}
   -> StartupMode
-  -> io (PartitionKey -> i -> IO o)
+  -> io (PartitionKey -> i -> io2 o)
 
 forkLegionary legionary settings startupMode = do
   logging <- askLoggerIO
@@ -346,10 +573,51 @@
     chan <- liftIO newChan
     forkC "main legion thread" $
       runLegionary legionary settings startupMode (chanToSource chan)
-    return (\ key request -> do
+    return (\ key request -> liftIO $ do
         responseVar <- newEmptyMVar
         writeChan chan ((key, request), putMVar responseVar)
         takeMVar responseVar
       )
+
+
+{- | This is the type of message passed around in the runtime. -}
+data RuntimeMessage i o s
+  = P (PeerMessage i o s)
+  | R (RequestMsg i o)
+  | J (JoinRequest, JoinResponse -> LIO ())
+  | A (AdminMessage i o s)
+instance (Show i, Show o, Show s) => Show (RuntimeMessage i o s) where
+  show (P m) = "(P " ++ show m ++ ")"
+  show (R ((p, i), _)) = "(R ((" ++ show p ++ ", " ++ show i ++ "), _))"
+  show (J (jr, _)) = "(J (" ++ show jr ++ ", _))"
+  show (A a) = "(A (" ++ show a ++ "))"
+
+
+{- | The runtime state. -}
+data RuntimeState i o s = RuntimeState {
+         self :: Peer,
+    forwarded :: Map MessageId (o -> LIO ()),
+       nextId :: MessageId,
+           cm :: ConnectionManager i o s
+  }
+
+
+{- | This is the type of a join request message. -}
+data JoinRequest = JoinRequest BSockAddr
+  deriving (Generic, Show)
+instance Binary JoinRequest
+
+
+{- | The response to a JoinRequst message -}
+data JoinResponse
+  = JoinOk Peer ClusterPowerState
+  | JoinRejected String
+  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))
 
 
diff --git a/src/Network/Legion/Runtime/ConnectionManager.hs b/src/Network/Legion/Runtime/ConnectionManager.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Legion/Runtime/ConnectionManager.hs
@@ -0,0 +1,208 @@
+{-# LANGUAGE DeriveGeneric #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE TemplateHaskell #-}
+{- |
+  This module manages connections to other nodes in the cluster.
+-}
+module Network.Legion.Runtime.ConnectionManager (
+  ConnectionManager,
+  newConnectionManager,
+  send,
+  newPeers,
+) where
+
+import Prelude hiding (lookup)
+
+import Control.Concurrent (Chan, writeChan, newChan, readChan)
+import Control.Exception (try, SomeException)
+import Control.Monad (void)
+import Control.Monad.Logger (logInfo, logWarn)
+import Control.Monad.Trans.Class (lift)
+import Data.Binary (Binary, encode)
+import Data.ByteString.Lazy (ByteString)
+import Data.Map (toList, insert, empty, Map, lookup)
+import Data.Text (pack)
+import Network.Legion.BSockAddr (BSockAddr(BSockAddr))
+import Network.Legion.Distribution (Peer)
+import Network.Legion.Fork (forkC)
+import Network.Legion.LIO (LIO)
+import Network.Legion.Runtime.PeerMessage (PeerMessage)
+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))
+import Network.Socket.ByteString.Lazy (sendAll)
+
+{- |
+  A handle on the connection manager
+-}
+data ConnectionManager i o s = C (Chan (Message i o s))
+instance Show (ConnectionManager i o s) where
+  show _ = "ConnectionManager"
+
+
+{- |
+  Create a new connection manager.
+-}
+newConnectionManager :: (Binary i, Binary o, Binary s)
+  => Map Peer BSockAddr
+  -> LIO (ConnectionManager i o s)
+newConnectionManager initPeers = do
+    chan <- lift newChan
+    forkC "connection manager thread" $
+      manager chan S {connections = empty}
+    let cm = C chan
+    newPeers cm initPeers
+    return cm
+  where
+    manager :: (Binary s, Binary o, Binary i)
+      => Chan (Message i o s)
+      -> State i o s
+      -> LIO ()
+    manager chan state = lift (readChan chan) >>= handle state >>= manager chan
+
+    handle :: (Binary i, Binary o, Binary s)
+      => State i o s
+      -> Message i o s
+      -> LIO (State i 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
+
+
+{- |
+  Build a new connection.
+-}
+connection :: (Binary i, Binary o, Binary s)
+  => SockAddr
+  -> LIO (Chan (PeerMessage i o s))
+
+connection addr = do
+    chan <- lift newChan
+    forkC ("connection to: " ++ show addr) $
+      handle chan Nothing
+    return chan
+  where
+    handle :: (Binary i, Binary o, Binary s)
+      => Chan (PeerMessage i o s)
+      -> Maybe Socket
+      -> LIO ()
+    handle chan so =
+      lift (readChan chan) >>= sendWithRetry so . encode >>= handle chan
+
+    {- |
+      Open a socket.
+    -}
+    openSocket :: IO Socket
+    openSocket = do
+      so <- socket (fam addr) Stream defaultProtocol
+      connect so addr
+      return so
+
+    {- |
+      Try to send the payload over the socket, and if that fails, then try to
+      create a new socket and retry sending the payload. Return whatever the
+      "working" socket is.
+    -}
+    sendWithRetry :: Maybe Socket -> ByteString -> LIO (Maybe Socket)
+    sendWithRetry Nothing payload =
+      (lift . try) openSocket >>= \case
+        Left err -> do
+          $(logWarn) . pack
+            $ "Can't connect to: " ++ show addr ++ ". Dropping message on "
+            ++ "the floor: " ++ show payload ++ ". The error was: "
+            ++ show (err :: SomeException)
+          return Nothing
+        Right so -> do
+          result2 <- (lift . try) (sendAll so payload)
+          case result2 of
+            Left err -> $(logWarn) . pack
+              $ "An error happend when trying to send a payload over a socket "
+              ++ "to the address: " ++ show addr ++ ". The error was: "
+              ++ show (err :: SomeException) ++ ". This is the last straw, we "
+              ++ "are not retrying. The message is being dropped on the floor. "
+              ++ "The message was: " ++ show payload
+            Right _ -> return ()
+          return (Just so)
+    sendWithRetry (Just so) payload =
+      (lift . try) (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 ()))
+          sendWithRetry Nothing payload
+        Right _ ->
+          return (Just so)
+
+
+{- |
+  Send a message to a peer.
+-}
+send
+  :: ConnectionManager i o s
+  -> Peer
+  -> PeerMessage i o s
+  -> LIO ()
+send (C chan) peer = lift . writeChan chan . Send peer
+
+
+{- |
+  Tell the connection manager about a new peer.
+-}
+newPeer
+  :: ConnectionManager i o s
+  -> Peer
+  -> SockAddr
+  -> LIO ()
+newPeer (C chan) peer addr = lift $ writeChan chan (NewPeer peer addr)
+
+
+{- |
+  Tell the connection manager about all the peers known to the cluster state.
+-}
+newPeers :: ConnectionManager i o s -> Map Peer BSockAddr -> LIO ()
+newPeers cm peers =
+    mapM_ oneNewPeer (toList peers)
+  where
+    oneNewPeer (peer, BSockAddr addy) = newPeer cm peer addy
+
+
+{- |
+  The internal state of the connection manager.
+-}
+data State i o s = S {
+    connections :: Map Peer (Chan (PeerMessage i o s))
+  }
+
+
+{- |
+  The types of messages that the ConnectionManager understands.
+-}
+data Message i o s
+  = NewPeer Peer SockAddr
+  | Send Peer (PeerMessage i 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
+
+
diff --git a/src/Network/Legion/Runtime/PeerMessage.hs b/src/Network/Legion/Runtime/PeerMessage.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Legion/Runtime/PeerMessage.hs
@@ -0,0 +1,79 @@
+{-# LANGUAGE DeriveGeneric #-}
+{- |
+  This module contains the type of runtime messages that can be exchanged
+  between peers.
+-}
+module Network.Legion.Runtime.PeerMessage (
+  PeerMessage(..),
+  PeerMessagePayload(..),
+  MessageId,
+  newSequence,
+  next,
+) where
+
+import Control.Monad.Trans.Class (lift)
+import Data.Binary (Binary)
+import Data.UUID (UUID)
+import Data.Word (Word64)
+import GHC.Generics (Generic)
+import Network.Legion.ClusterState (ClusterPowerState)
+import Network.Legion.Distribution (Peer)
+import Network.Legion.LIO (LIO)
+import Network.Legion.PartitionKey (PartitionKey)
+import Network.Legion.PartitionState (PartitionPowerState)
+import Network.Legion.UUID (getUUID)
+
+
+{- |
+  The type of messages sent between peers.
+-}
+data PeerMessage i o s = PeerMessage {
+       source :: Peer,
+    messageId :: MessageId,
+      payload :: PeerMessagePayload i o s
+  }
+  deriving (Generic, Show)
+instance (Binary i, Binary o, Binary s) => Binary (PeerMessage i o s)
+
+
+{- |
+  The data contained within a peer message.
+
+  When we get around to implementing durability and data replication,
+  the sustained inability to confirm that a node has received one of
+  these messages should result in the ejection of that node from the
+  cluster and the blacklisting of that node so that it can never re-join.
+-}
+data PeerMessagePayload i o s
+  = PartitionMerge PartitionKey (PartitionPowerState i s)
+  | ForwardRequest PartitionKey i
+  | ForwardResponse MessageId o
+  | ClusterMerge ClusterPowerState
+  deriving (Generic, Show)
+instance (Binary i, Binary o, Binary s) => Binary (PeerMessagePayload i o s)
+
+
+data MessageId = M UUID Word64 deriving (Generic, Show, Eq, Ord)
+instance Binary MessageId
+
+
+{- |
+  Initialize a new sequence of messageIds. It would be perfectly fine to ensure
+  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
+  sid <- getUUID
+  return (M sid 0)
+
+
+{- |
+  Generate the next message id in the sequence. We would normally use
+  `succ` for this kind of thing, but making `MessageId` an instance of
+  `Enum` really isn't appropriate.
+-}
+next :: MessageId -> MessageId
+next (M sequenceId ord) = M sequenceId (ord + 1)
+
+
diff --git a/src/Network/Legion/StateMachine.hs b/src/Network/Legion/StateMachine.hs
--- a/src/Network/Legion/StateMachine.hs
+++ b/src/Network/Legion/StateMachine.hs
@@ -1,75 +1,91 @@
-{-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE TemplateHaskell #-}
 {- |
-  This module contains the state machine implementation of a legion node.
+  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.
 
-  Discussion:
+  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.)
 
-  This is a first attempt to discover a pure legion state machine and isolated
-  it from the runtime IO considerations. It is obviously not perfect, because
-  everything still lives in 'LIO', which is 'IO'-backed; but mostly this is
-  because access to the persistence layer still happens here. Once we pull that
-  out into the 'Network.Legion.Runtime' module we should be clear to remove IO
-  and make this thing look more like a pure state machine. - Rick
+  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 (
-  stateMachine,
-  LInput(..),
-  LOutput(..),
-  JoinRequest(..),
-  JoinResponse(..),
-  AdminMessage(..),
+module Network.Legion.StateMachine(
+  -- * Running the state machine.
   NodeState,
-  Forwarded(..),
-  PeerMessage(..),
-  PeerMessagePayload(..),
-  MessageId,
-  next,
   newNodeState,
-) where
+  SM,
+  runSM,
 
-import Prelude hiding (lookup)
+  -- * State machine inputs.
+  userRequest,
+  partitionMerge,
+  clusterMerge,
+  migrate,
+  propagate,
+  rebalance,
+  heartbeat,
+  eject,
+  join,
 
-import Control.Exception (throw)
+  -- * State machine outputs.
+  ClusterAction(..),
+  UserResponse(..),
+
+  -- * State inspection
+  getPeers,
+) where
+
 import Control.Monad (unless)
-import Control.Monad.Catch (try, SomeException, MonadCatch, MonadThrow)
-import Control.Monad.IO.Class (MonadIO, liftIO)
-import Control.Monad.Logger (logDebug, logWarn, logError, logInfo,
-  MonadLogger)
-import Control.Monad.Trans.Class (MonadTrans, lift)
-import Control.Monad.Trans.State (StateT, runStateT, get, put)
-import Data.Binary (Binary)
-import Data.Conduit (Source, Conduit, ($$), await, awaitForever,
-  transPipe, ConduitM, yield, ($=))
+import Control.Monad.IO.Class (MonadIO)
+import Control.Monad.Logger (MonadLogger, logWarn, logDebug, logError)
+import Control.Monad.Trans.Class (lift, MonadTrans)
+import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)
+import Control.Monad.Trans.State (StateT, runStateT, get, put, modify)
+import Data.Aeson (ToJSON, toJSON, object, (.=), encode)
+import Data.ByteString.Lazy (toStrict)
+import Data.Conduit (($=), ($$), Sink, transPipe, awaitForever)
 import Data.Default.Class (Default)
-import Data.Map (Map, insert, lookup)
+import Data.Map (Map)
 import Data.Maybe (fromMaybe)
-import Data.Set (member, minView, (\\))
-import Data.Text (pack)
+import Data.Set ((\\))
+import Data.Text (pack, unpack)
+import Data.Text.Encoding (decodeUtf8)
 import Data.Time.Clock (getCurrentTime)
-import Data.UUID (UUID)
-import Data.Word (Word64)
-import GHC.Generics (Generic)
-import Network.Legion.Application (Legionary, LegionConstraints,
-  Persistence(getState, saveState, list), Legionary(Legionary,
-  persistence, handleRequest), RequestMsg)
+import Network.Legion.Application (Legionary(Legionary), getState,
+  saveState, list, persistence, handleRequest)
 import Network.Legion.BSockAddr (BSockAddr)
-import Network.Legion.ClusterState (claimParticipation, ClusterPropState,
-  getPeers, getDistribution, ClusterPowerState)
-import Network.Legion.Distribution (rebalanceAction, RebalanceAction(
-  Invite), Peer, newPeer)
-import Network.Legion.KeySet (union, KeySet)
+import Network.Legion.ClusterState (ClusterPropState, ClusterPowerState)
+import Network.Legion.Distribution (Peer, rebalanceAction, newPeer,
+  RebalanceAction(Invite))
+import Network.Legion.KeySet (KeySet, union)
 import Network.Legion.LIO (LIO)
 import Network.Legion.PartitionKey (PartitionKey)
 import Network.Legion.PartitionState (PartitionPowerState, PartitionPropState)
 import Network.Legion.PowerState (ApplyDelta)
-import Network.Legion.UUID (getUUID)
 import qualified Data.Conduit.List as CL
 import qualified Data.Map as Map
 import qualified Data.Set as Set
@@ -79,262 +95,133 @@
 
 
 {- |
-  This conduit houses the main legionary state machine. The conduit's
-  input, internal state, and output are analogous to a "real" state
-  machine's input, state, and output. If this seems like an odd use of
-  conduit, that's ok.  Hopefully we can make this look more like a pure
-  state machine once we remove 'IO' from this module.
+  This is the portion of the local node state that is not persistence
+  related.
 -}
-stateMachine :: (LegionConstraints i o s)
-  => Legionary i o s
-  -> NodeState i o s
-  -> Conduit (LInput i o s) LIO (LOutput i o s)
-stateMachine l n = awaitForever (\msg -> do
-    newState <- runStateMT n $ do
-      handleMessage l msg
-      heartbeat
-      migrate l
-      propagate
-      rebalance l
-      logState
-    stateMachine l newState
-  )
-  where
-    logState = lift . logNodeState =<< getS
+data NodeState i s = NodeState {
+             self :: Peer,
+          cluster :: ClusterPropState,
+       partitions :: Map PartitionKey (PartitionPropState i s),
+        migration :: KeySet
+  }
+instance (Show i, Show s) => Show (NodeState i 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 i, Show s) => ToJSON (NodeState i s) where
+  toJSON (NodeState self cluster partitions migration) =
+    object [
+              "self" .= show self,
+           "cluster" .= cluster,
+        "partitions" .= Map.mapKeys show partitions,
+         "migration" .= show migration
+      ]
 
 
-{- | Handle one incomming message.  -}
-handleMessage :: (LegionConstraints i o s)
-  => Legionary i o s
-  -> LInput i o s
-  -> StateM i o s ()
+{- |
+  Make a new node state.
+-}
+newNodeState :: Peer -> ClusterPropState -> NodeState i s
+newNodeState self cluster =
+  NodeState {
+      self,
+      cluster,
+      partitions = Map.empty,
+      migration = KS.empty
+    }
 
-handleMessage l msg = do
-  NodeState {cluster} <- getS
-  let
-    {- | Return `True` if the peer is a known peer, false otherwise.  -}
-    known peer = peer `member` C.allParticipants cluster
-  $(logDebug) . pack $ "Receiving: " ++ show msg
-  case msg of
-    P peerMsg@PeerMessage {source} ->
-      if known source
-        then handlePeerMessage l peerMsg
-        else
-          $(logWarn) . pack
-            $ "Dropping message from unknown peer: " ++ show source
-    R ((key, request), respond) ->
-      case minView (C.findPartition key cluster) of
-        Nothing ->
-          $(logError) . pack
-            $ "Keyspace does not contain key: " ++ show key ++ ". This "
-            ++ "is a very bad thing and probably means there is a bug, "
-            ++ "or else this node has not joined a cluster yet."
-        Just (peer, _) ->
-          forward peer key request respond
-    J m -> handleJoinRequest m
-    A m -> handleAdminMessage l m
 
+{- |
+  This monad encapsulates the global state of the legion node (not
+  counting the runtime stuff, like open connections and what have
+  you).
 
-{- | Handles one incomming message from a peer. -}
-handlePeerMessage :: (LegionConstraints i o s)
-  => Legionary i o s
-  -> PeerMessage i o s
-  -> StateM i o s ()
+  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 i o s a = SM {
+    unSM :: ReaderT (Legionary i o s) (StateT (NodeState i s) LIO) a
+  }
+  deriving (Functor, Applicative, Monad, MonadLogger, MonadIO)
 
-handlePeerMessage -- PartitionMerge
-    Legionary {
-        persistence
-      }
-    msg@PeerMessage {
-        source,
-        payload = PartitionMerge key ps
-      }
-  = do
-    nodeState@NodeState {self, propStates, cluster} <- getS
-    propState <- lift $ maybe
-      (getStateL persistence self cluster key)
-      return
-      (lookup key propStates)
-    let
-      owners = C.findPartition key cluster
-    case P.mergeEither source ps propState of
-      Left err ->
-        $(logWarn) . pack
-          $ "Can't apply incomming partition action message "
-          ++ show msg ++ "because of: " ++ show err
-      Right newPropState -> do
-        $(logDebug) "Saving because of PartitionMerge"
-        lift $ saveStateL persistence key (
-            if P.participating newPropState
-              then Just (P.getPowerState newPropState)
-              else Nothing
-          )
-        putS nodeState {
-            propStates = if newPropState == P.new key self owners
-              then Map.delete key propStates
-              else insert key newPropState propStates
-          }
 
-handlePeerMessage -- ForwardRequest
-    Legionary {handleRequest, persistence}
-    msg@PeerMessage {
-        payload = ForwardRequest key request,
-        source,
-        messageId
-      }
-  = do
-    ns@NodeState {self, cluster, propStates} <- getS
-    let owners = C.findPartition key cluster
-    if self `member` owners
-      then do
-        let
-          respond = send source . ForwardResponse messageId
-
-        -- TODO 
-        --   - figure out some slick concurrency here, by maintaining
-        --       a map of keys that are currently being accessed or
-        --       something
-        -- 
-        either (respond . rethrow) respond =<< try (do 
-            prop <- lift $ getStateL persistence self cluster key
-            let response = handleRequest key request (P.ask prop)
-                newProp = P.delta request prop
-            $(logDebug) "Saving because of ForwardRequest"
-            lift $ saveStateL persistence key (Just (P.getPowerState newProp))
-            $(logInfo) . pack
-              $ "Handling user request: " ++ show request
-            $(logDebug) . pack
-              $ "Request details request: " ++ show prop ++ " ++ "
-              ++ show request ++ " --> " ++ show (response, newProp)
-            putS ns {propStates = insert key newProp propStates}
-            return response
-          )
-      else
-        {-
-          we don't own the key after all, someone was wrong to forward
-          us this request.
-        -}
-        case minView owners of
-          Nothing -> $(logError) . pack
-            $ "Can't find any owners for the key: " ++ show key
-          Just (peer, _) ->
-            emit (Send peer msg)
-  where
-    {- |
-      rethrow is just a reification of `throw`.
-    -}
-    rethrow :: SomeException -> a
-    rethrow = throw
-
-handlePeerMessage -- ForwardResponse
-    Legionary {}
-    msg@PeerMessage {
-        payload = ForwardResponse messageId response
-      }
-  = do
-    nodeState@NodeState {forwarded} <- getS
-    case lookup messageId (unF forwarded) of
-      Nothing -> $(logWarn) . pack
-        $  "This peer received a response for a forwarded request that it "
-        ++ "didn't send. The only time you might expect to see this is if "
-        ++ "this peer recently crashed and was immediately restarted. If "
-        ++ "you are seeing this in other circumstances then probably "
-        ++ "something is very wrong at the network level. The message was: "
-        ++ show msg
-      Just respond ->
-        lift $ respond response
-    putS nodeState {
-        forwarded = F . Map.delete messageId . unF $ forwarded
-      }
-
-handlePeerMessage -- ClusterMerge
-    Legionary {}
-    msg@PeerMessage {
-        source,
-        payload = ClusterMerge ps
-      }
-  = do
-    nodeState@NodeState {migration, cluster} <- getS
-    case C.mergeEither source ps cluster of
-      Left err ->
-        $(logWarn) . pack
-          $ "Can't apply incomming cluster action message "
-          ++ show msg ++ "because of: " ++ show err
-      Right (newCluster, newMigration) ->
-        putS nodeState {
-            migration = migration `union` newMigration,
-            cluster = newCluster
-          }
+{- |
+  Run an SM action.
+-}
+runSM
+  :: Legionary i o s
+  -> NodeState i s
+  -> SM i o s a
+  -> LIO (a, NodeState i s)
+runSM l ns action = runStateT (runReaderT (unSM action) l) ns
 
 
-{- | Handle a join request message -}
-handleJoinRequest
-  :: (JoinRequest, JoinResponse -> LIO ())
-  -> StateM i o s ()
+{- | Handle a user request. -}
+userRequest :: (ApplyDelta i s, Default s)
+  => PartitionKey
+  -> i
+  -> SM i o s (UserResponse o)
+userRequest key request = SM $ do
+  NodeState {self, cluster} <- lift get
+  Legionary {handleRequest} <- ask
+  let owners = C.findPartition key cluster
+  if self `Set.member` owners
+    then do
+      partition <- unSM $ getPartition key
+      let
+        response = handleRequest key request (P.ask partition)
+        partition2 = P.delta request partition
+      unSM $ savePartition key partition2
+      return (Respond response)
 
-handleJoinRequest (JoinRequest peerAddr, respond) = do
-  ns@NodeState {cluster} <- getS
-  peer <- lift newPeer
-  let newCluster = C.joinCluster peer peerAddr cluster
-  emit .  NewPeers . getPeers $ newCluster
-  lift $ respond (JoinOk peer (C.getPowerState newCluster))
-  putS ns {cluster = newCluster}
+    else case Set.toList owners of
+      [] -> do
+        let msg = "No owners for key: " ++ show key
+        $(logError) . pack $ msg
+        error msg
+      peer:_ -> return (Forward peer)
 
 
 {- |
-  Handle a message from the admin service.
+  Handle the state transition for a partition merge event. Returns 'Left'
+  if there is an error, and 'Right' if everything went fine.
 -}
-handleAdminMessage
-  :: Legionary i o s
-  -> AdminMessage i o s
-  -> StateM i o s ()
-handleAdminMessage _ (GetState respond) =
-  lift . respond =<< getS
-handleAdminMessage Legionary {persistence} (GetPart key respond) = lift $ do
-  partitionVal <- lift (getState persistence key)
-  respond partitionVal
-handleAdminMessage _ (Eject peer respond) = do
-    {-
-      TODO: we should attempt to notify the ejected peer that it has
-      been ejected instead of just cutting it off and washing our hands
-      of it. I have a vague notion that maybe ejected peers should be
-      permanently recorded in the cluster state so that if they ever
-      reconnect then we can notify them that they are no longer welcome
-      to participate.
-
-      On a related note, 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?
-    -}
-    modifyS eject
-    lift $ respond ()
-  where
-    eject ns@NodeState {cluster} = ns {cluster = C.eject peer cluster}
+partitionMerge :: (Show i, Show s, ApplyDelta i s, Default s)
+  => Peer
+  -> PartitionKey
+  -> PartitionPowerState i s
+  -> SM i o s ()
+partitionMerge source key foreignPartition = do
+  partition <- getPartition key
+  case P.mergeEither source foreignPartition partition of
+    Left err -> $(logWarn) . pack
+      $ "Can't apply incomming partition merge from "
+      ++ show source ++ ": " ++ show foreignPartition
+      ++ ". because of: " ++ show err
+    Right newPartition -> savePartition key newPartition
 
 
-{- | Update all of the propagation states with the current time.  -}
-heartbeat :: StateM i o s ()
-heartbeat = do
-  now <- liftIO getCurrentTime
-  ns@NodeState {cluster, propStates} <- getS
-  putS ns {
-      cluster = C.heartbeat now cluster,
-      propStates = Map.fromAscList [
-          (k, P.heartbeat now p)
-          | (k, p) <- Map.toAscList propStates
-        ]
-    }
+{- | Handle the state transition for a cluster merge event. -}
+clusterMerge
+  :: Peer
+  -> ClusterPowerState
+  -> SM i o s ()
+clusterMerge source foreignCluster = SM . lift $ do
+  nodeState@NodeState {migration, cluster} <- get
+  case C.mergeEither source foreignCluster cluster of
+    Left err -> $(logWarn) . pack
+      $ "Can't apply incomming cluster merge from "
+      ++ show source ++ ": " ++ show foreignCluster
+      ++ ". because of: " ++ show err
+    Right (newCluster, newMigration) ->
+      put nodeState {
+          migration = migration `union` newMigration,
+          cluster = newCluster
+        }
 
 
 {- |
@@ -349,351 +236,215 @@
   peer to a partition. This will cause the data to be transfered in the
   normal course of propagation.
 -}
-migrate :: (LegionConstraints i o s) => Legionary i o s -> StateM i o s ()
-migrate Legionary{persistence} = do
-    ns@NodeState {migration} <- getS
+migrate :: (ApplyDelta i s) => SM i o s ()
+migrate = do
+    NodeState {migration} <- (SM . lift) get
+    Legionary {persistence} <- SM ask
     unless (KS.null migration) $
-      putS =<< lift (
-          listL persistence
-          $= CL.filter ((`KS.member` migration) . fst)
-          $$ accum ns {migration = KS.empty}
-        )
+      transPipe (SM . lift3) (list persistence)
+      $= CL.filter ((`KS.member` migration) . fst)
+      $$ accum
+    (SM . lift) $ modify (\ns -> ns {migration = KS.empty})
   where
-    accum ns@NodeState {self, cluster, propStates} = await >>= \case
-      Nothing -> return ns
-      Just (key, ps) -> 
-        let
-          origProp = fromMaybe (P.initProp self ps) (lookup key propStates)
-          newPeers_ = C.findPartition key cluster \\ P.projParticipants origProp
-          {- This 'P.participate' is where the magic happens. -}
-          newProp = foldr P.participate origProp (Set.toList newPeers_)
-        in do
-          $(logDebug) . pack $ "Migrating: " ++ show key
-          lift (saveStateL persistence key (Just (P.getPowerState newProp)))
-          accum ns {
-              propStates = Map.insert key newProp propStates
-            }
+    accum :: (ApplyDelta i s)
+      => Sink (PartitionKey, PartitionPowerState i s) (SM i o s) ()
+    accum = awaitForever $ \ (key, ps) -> do
+      NodeState {self, cluster, partitions} <- (lift . SM . lift) get
+      let
+        partition = fromMaybe (P.initProp self ps) (Map.lookup key partitions)
+        newPeers = C.findPartition key cluster \\ P.projParticipants partition
+        newPartition = foldr P.participate partition (Set.toList newPeers)
+      $(logDebug) . pack $ "Migrating: " ++ show key
+      lift (savePartition key newPartition)
 
 
 {- |
   Handle all cluster and partition state propagation actions, and return
   an updated node state.
 -}
-propagate :: (LegionConstraints i o s) => StateM i o s ()
-propagate = do
-    ns@NodeState {cluster, propStates, self} <- getS
-    let (peers, ps, cluster2) = C.actions cluster
-    $(logDebug) . pack $ "Cluster Actions: " ++ show (peers, ps)
-    mapM_ (doClusterAction ps) (Set.toList peers)
-    propStates2 <- mapM doPartitionActions (Map.toList propStates)
-    putS ns {
-        cluster = cluster2,
-        propStates = Map.fromAscList [
-            (k, p)
-            | (k, p) <- propStates2
-            , p /= P.initProp self (P.getPowerState p)
-          ]
-      }
+propagate :: SM i o s [ClusterAction i s]
+propagate = SM $ do
+    partitionActions <- getPartitionActions
+    clusterActions <- unSM getClusterActions
+    return (clusterActions ++ partitionActions)
   where
-    doClusterAction ps peer =
-      send peer (ClusterMerge ps)
+    getPartitionActions = do
+      ns@NodeState {partitions} <- lift get
+      let
+        updates = [
+            (key, newPartition, [
+                PartitionMerge peer key ps
+                | peer <- Set.toList peers_
+              ])
+            | (key, partition) <- Map.toAscList partitions
+            , let (peers_, ps, newPartition) = P.actions partition
+          ]
+        actions = [a | (_, _, as) <- updates, a <- as]
+        newPartitions = Map.fromAscList [
+            (key, newPartition)
+            | (key, newPartition, _) <- updates
+            , not (P.complete newPartition)
+          ]
+      (lift . put) ns {
+          partitions = newPartitions
+        }
+      return actions
 
-    doPartitionActions (key, propState) = do
-        let (peers, ps, propState2) = P.actions propState
-        mapM_ (perform ps) (Set.toList peers)
-        return (key, propState2)
-      where
-        perform ps peer =
-          send peer (PartitionMerge key ps)
+    getClusterActions :: SM i o s [ClusterAction i s]
+    getClusterActions = SM $ do
+      ns@NodeState {cluster} <- lift get
+      let
+        (peers, cs, newCluster) = C.actions cluster
+        actions = [ClusterMerge peer cs | peer <- Set.toList peers]
+      (lift . put) ns {
+          cluster = newCluster
+        }
+      return actions
 
 
 {- |
   Figure out if any rebalancing actions must be taken by this node, and kick
   them off if so.
 -}
-rebalance :: (LegionConstraints i o s) => Legionary i o s -> StateM i o s ()
-rebalance _ = do
-  ns@NodeState {self, cluster} <- getS
+rebalance :: SM i o s ()
+rebalance = SM $ do
+  ns@NodeState {self, cluster} <- lift get
   let
-    allPeers = (Set.fromList . Map.keys . getPeers) cluster
-    dist = getDistribution cluster
+    allPeers = (Set.fromList . Map.keys . C.getPeers) cluster
+    dist = C.getDistribution cluster
     action = rebalanceAction self allPeers dist
   $(logDebug) . pack $ "The rebalance action is: " ++ show action
-  putS ns {
+  (lift . put) ns {
       cluster = case action of
         Nothing -> cluster
-        Just (Invite ks) -> claimParticipation self ks cluster
+        Just (Invite ks) -> C.claimParticipation self ks cluster
     }
 
 
-{- | This is the type of input accepted by the legionary state machine. -}
-data LInput i o s
-  = P (PeerMessage i o s)
-  | R (RequestMsg i o)
-  | J (JoinRequest, JoinResponse -> LIO ())
-  | A (AdminMessage i o s)
-
-instance (Show i, Show o, Show s) => Show (LInput i o s) where
-  show (P m) = "(P " ++ show m ++ ")"
-  show (R ((p, i), _)) = "(R ((" ++ show p ++ ", " ++ show i ++ "), _))"
-  show (J (jr, _)) = "(J (" ++ show jr ++ ", _))"
-  show (A a) = "(A (" ++ show a ++ "))"
-
-
-{- | This is the type of output produced by the legionary state machine. -}
-data LOutput i o s
-  = Send Peer (PeerMessage i o s)
-  | NewPeers (Map Peer BSockAddr)
-
-
-{- | A helper function to log the state of the node: -}
-logNodeState :: (LegionConstraints i o s) => NodeState i o s -> LIO ()
-logNodeState ns = $(logDebug) . pack
-    $ "The current node state is: " ++ show ns
-
-
-{- | Like `getState`, but in LIO, and provides the correct bottom value.  -}
-getStateL :: (ApplyDelta i s, Default s)
-  => Persistence i s
-  -> Peer
-  -> ClusterPropState
-  -> PartitionKey
-  -> LIO (PartitionPropState i s)
-
-getStateL p self cluster key =
-  {- dp == default participants -}
-  let dp = C.findPartition key cluster
-  in maybe
-      (P.new key self dp)
-      (P.initProp self)
-      <$> lift (getState p key)
-
-
-{- | Like `saveState`, but in LIO.  -}
-saveStateL
-  :: Persistence i s
-  -> PartitionKey
-  -> Maybe (PartitionPowerState i s)
-  -> LIO ()
-saveStateL p k = lift . saveState p k
-
-
-{- | Like `list`, but in LIO.  -}
-listL :: Persistence i s -> Source LIO (PartitionKey, PartitionPowerState i s)
-listL p = transPipe lift (list p)
-
-
-{- | This is the type of a join request message.  -}
-data JoinRequest = JoinRequest BSockAddr
-  deriving (Generic, Show)
-instance Binary JoinRequest
-
-
-{- | The response to a JoinRequst message -}
-data JoinResponse
-  = JoinOk Peer ClusterPowerState
-  | JoinRejected String
-  deriving (Generic)
-instance Binary JoinResponse
-
-
-{- |
-  The type of messages sent by the admin service.
--}
-data AdminMessage i o s
-  = GetState (NodeState i o s -> LIO ())
-  | GetPart PartitionKey (Maybe (PartitionPowerState i s) -> LIO ())
-  | Eject Peer (() -> LIO ())
-
-instance Show (AdminMessage i o s) where
-  show (GetState _) = "(GetState _)"
-  show (GetPart k _) = "(GetPart " ++ show k ++ " _)"
-  show (Eject p _) = "(Eject " ++ show p ++ " _)"
-
-
-{- | Defines the local state of a node in the cluster.  -}
-data NodeState i o s = NodeState {
-             self :: Peer,
-          cluster :: ClusterPropState,
-        forwarded :: Forwarded o,
-       propStates :: Map PartitionKey (PartitionPropState i s),
-        migration :: KeySet,
-           nextId :: MessageId
-  }
-  deriving (Show)
-
-
-{- | A set of forwarded messages.  -}
-newtype Forwarded o = F {unF :: Map MessageId (o -> LIO ())}
-instance Show (Forwarded o) where
-  show = show . Map.keys . unF
-
-
-{- |
-  The type of messages sent to us from other peers.
--}
-data PeerMessage i o s = PeerMessage {
-       source :: Peer,
-    messageId :: MessageId,
-      payload :: PeerMessagePayload i o s
-  }
-  deriving (Generic, Show)
-instance (Binary i, Binary o, Binary s) => Binary (PeerMessage i o s)
+{- | Update all of the propagation states with the current time.  -}
+heartbeat :: SM i o s ()
+heartbeat = SM $ do
+  now <- lift3 getCurrentTime
+  ns@NodeState {cluster, partitions} <- lift get
+  (lift . put) ns {
+      cluster = C.heartbeat now cluster,
+      partitions = Map.fromAscList [
+          (k, P.heartbeat now p)
+          | (k, p) <- Map.toAscList partitions
+        ]
+    }
 
 
-{- |
-  The data contained within a peer message.
-
-  When we get around to implementing durability and data replication,
-  the sustained inability to confirm that a node has received one of
-  these messages should result in the ejection of that node from the
-  cluster and the blacklisting of that node so that it can never re-join.
--}
-data PeerMessagePayload i o s
-  = PartitionMerge PartitionKey (PartitionPowerState i s)
-  | ForwardRequest PartitionKey i
-  | ForwardResponse MessageId o
-  | ClusterMerge ClusterPowerState
-  deriving (Generic, Show)
-instance (Binary i, Binary o, Binary s) => Binary (PeerMessagePayload i o s)
+{- | Eject a peer from the cluster.  -}
+eject :: Peer -> SM i o s ()
+eject peer = SM . lift $ do
+  ns@NodeState {cluster} <- get
+  put ns {cluster = C.eject peer cluster}
 
 
-data MessageId = M UUID Word64 deriving (Generic, Show, Eq, Ord)
-instance Binary MessageId
+{- | Handle a peer join request.  -}
+join :: BSockAddr -> SM i o s (Peer, ClusterPowerState)
+join peerAddr = SM $ do
+  peer <- lift2 newPeer
+  ns@NodeState {cluster} <- lift get
+  let newCluster = C.joinCluster peer peerAddr cluster
+  (lift . put) ns {cluster = newCluster}
+  return (peer, C.getPowerState newCluster)
 
 
 {- |
-  Generate the next message id in the sequence. We would normally use
-  `succ` for this kind of thing, but making `MessageId` an instance of
-  `Enum` really isn't appropriate.
+  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.
 -}
-next :: MessageId -> MessageId
-next (M sequenceId ord) = M sequenceId (ord + 1)
+data ClusterAction i s
+  = ClusterMerge Peer ClusterPowerState
+  | PartitionMerge Peer PartitionKey (PartitionPowerState i s)
 
 
 {- |
-  Initialize a new sequence of messageIds
+  The type of response to a user request, either forward to another node,
+  or respond directly.
 -}
-newSequence ::  LIO MessageId
-newSequence = lift $ do
-  sid <- getUUID
-  return (M sid 0)
+data UserResponse o
+  = Forward Peer
+  | Respond o
 
 
-{- |
-  Make a new node state.
--}
-newNodeState :: Peer -> ClusterPropState -> LIO (NodeState i o s)
-newNodeState self cluster = do
-  nextId <- newSequence
-  return NodeState {
-      self,
-      nextId,
-      cluster,
-      forwarded = F Map.empty,
-      propStates = Map.empty,
-      migration = KS.empty
-    }
+{- | Get the known peer data from the cluster. -}
+getPeers :: SM i o s (Map Peer BSockAddr)
+getPeers = SM $ C.getPeers . cluster <$> lift get
 
 
-send :: Peer -> PeerMessagePayload i o s -> StateM i o s ()
-send peer payload = do
-  ns@NodeState {self, nextId} <- getS
-  emit (Send peer PeerMessage {
-      source = self,
-      messageId = nextId,
-      payload
-    })
-  putS ns {nextId = next nextId}
+{- | Gets a partition state. -}
+getPartition :: (Default s, ApplyDelta i s)
+  => PartitionKey
+  -> SM i o s (PartitionPropState i s)
+getPartition key = SM $ do
+  Legionary {persistence} <- ask
+  NodeState {self, partitions, cluster} <- lift get
+  case Map.lookup key partitions of
+    Nothing ->
+      lift3 (getState persistence key) <&> \case
+        Nothing -> P.new key self (C.findPartition key cluster)
+        Just partition -> P.initProp self partition
+    Just partition -> return partition
 
 
-{- |
-  Forward a user request to a peer for handling, being sure to do all
-  the node state accounting.
--}
-forward
-  :: Peer
-  -> PartitionKey
-  -> i
-  -> (o -> IO ())
-  -> StateM i o s ()
-forward peer key request respond = do
-  ns@NodeState {nextId, self, forwarded} <- getS
-  emit (Send peer PeerMessage {
-      source = self,
-      messageId = nextId,
-      payload = ForwardRequest key request
-    })
-  putS ns {
-      nextId = next nextId,
-      forwarded = F . insert nextId (lift . respond) . unF $ forwarded
+{- | Saves a partition state. -}
+savePartition :: PartitionKey -> PartitionPropState i s -> SM i o s ()
+savePartition key partition = SM $ do
+  Legionary {persistence} <- ask
+  ns@NodeState {partitions} <- lift get
+  lift3 (saveState persistence key (
+      if P.participating partition
+        then Just (P.getPowerState partition)
+        else Nothing
+    ))
+  lift $ put ns {
+      partitions = if P.complete 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
     }
 
 
-{- |
-  The monad in which the internals of the state machine run. This is really
-  just a conduit, but we wrap it because we only want to allow `yield`, which
-  we have re-named `emit`.
--}
-newtype StateMT i o s m r = StateMT {
-    unStateMT ::
-      StateT
-        (NodeState i o s)
-        (ConduitM (LInput i o s) (LOutput i o s) m)
-        r
-  } deriving (
-    Functor, Applicative, Monad, MonadLogger, MonadCatch,
-    MonadThrow, MonadIO
-  )
-{-
-  We can lift things from the underlying monad straight to 'StateT',
-  bypassing the `CondutM` layer.
--}
-instance MonadTrans (StateMT i o s) where
-  lift = StateMT . lift . lift
-
-
-{- |
-  The state machine monad, in LIO.
--}
-type StateM i o s r = StateMT i o s LIO r
-
-
-{- |
-  Run the state machine monad, starting with the initial node state.
--}
-runStateMT
-  :: NodeState i o s
-  -> StateMT i o s m ()
-  -> ConduitM (LInput i o s) (LOutput i o s) m (NodeState i o s)
-runStateMT ns = fmap snd . (`runStateT` ns) . unStateMT
-
-
-{- |
-  Emit some output from the state machine.
--}
-emit :: LOutput i o s -> StateM i o s ()
-emit = StateMT . lift . yield
-
-
-{- |
-  Get the node State.
--}
-getS :: StateMT i o s m (NodeState i o s)
-getS = StateMT get
+{- | Borrowed from 'lens', like @flip fmap@. -}
+(<&>) :: (Functor f) => f a -> (a -> b) -> f b
+(<&>) = flip fmap
 
 
-{- |
-  Put the node state.
--}
-putS :: NodeState i o s -> StateMT i o s m ()
-putS = StateMT . put
+{- | Lift from two levels down in a monad transformation stack. -}
+lift2
+  :: (
+      MonadTrans a,
+      MonadTrans b,
+      Monad m,
+      Monad (b m)
+    )
+  => m r
+  -> a (b m) r
+lift2 = lift . lift
 
 
-{- |
-  Modify the node state.
--}
-modifyS :: (NodeState i o s -> NodeState i o s) -> StateMT i o s m ()
-modifyS f = putS . f =<< getS
+{- | Lift from three levels down in a monad transformation stack. -}
+lift3
+  :: (
+      MonadTrans a,
+      MonadTrans b,
+      MonadTrans c,
+      Monad m,
+      Monad (c m),
+      Monad (b (c m))
+    )
+  => m r
+  -> a (b (c m)) r
+lift3 = lift . lift . lift