diff --git a/dataflower.cabal b/dataflower.cabal
--- a/dataflower.cabal
+++ b/dataflower.cabal
@@ -4,10 +4,10 @@
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: b4d9c5533fc659e6801a98a43183cb91426ce81a603cc0267578d5a1cba664ad
+-- hash: f8c905e3886d672d9325bc3d2737600644d15f243a4c93fdfb69cde4c8de0b6c
 
 name:           dataflower
-version:        0.1.4.0
+version:        0.2.0.0
 synopsis:       A Pure-Haskell Timely Dataflow System
 description:    See README
 homepage:       https://github.com/doublecrowngaming/dataflower#readme
diff --git a/src/Dataflow.hs b/src/Dataflow.hs
--- a/src/Dataflow.hs
+++ b/src/Dataflow.hs
@@ -32,7 +32,7 @@
 
 import           Control.Monad              (void)
 import           Control.Monad.IO.Class     (MonadIO (..))
-import           Control.Monad.State.Strict (execStateT, runStateT)
+import           Control.Monad.State.Strict (evalStateT, execStateT, runStateT)
 import           Data.Traversable           (Traversable)
 import           Dataflow.Primitives
 import           Dataflow.Vertices
@@ -56,8 +56,10 @@
 -- have the same 'Timestamp' associated with them.
 --
 -- @since 0.1.0.0
-execute :: (MonadIO io, Traversable t) => t i -> Program i -> io ()
-execute corpus Program{..} = liftIO $ execDataflow feedInput
-  where
-    feedInput           = input corpus programInput
-    execDataflow action = void $ execStateT (runDataflow action) programState
+execute :: (MonadIO io, Traversable t) => t i -> Program i -> io (Program i)
+execute corpus Program{..} = liftIO $ do
+  newProgramState <- evalStateT (runDataflow duplicateDataflowState) programState
+
+  void $ execStateT (runDataflow $ input corpus programInput) newProgramState
+
+  return $ Program programInput newProgramState
diff --git a/src/Dataflow/Primitives.hs b/src/Dataflow/Primitives.hs
--- a/src/Dataflow/Primitives.hs
+++ b/src/Dataflow/Primitives.hs
@@ -1,12 +1,14 @@
 {-# LANGUAGE ExistentialQuantification  #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE MultiParamTypeClasses      #-}
+{-# LANGUAGE RecordWildCards            #-}
 
 module Dataflow.Primitives (
   Dataflow(..),
   DataflowState,
   Vertex(..),
   initDataflowState,
+  duplicateDataflowState,
   StateRef,
   newState,
   readState,
@@ -23,7 +25,9 @@
 ) where
 
 import           Control.Arrow              ((>>>))
-import           Control.Monad.State.Strict (StateT, gets, modify)
+import           Control.Monad              (forM, (>=>))
+import           Control.Monad.IO.Class     (liftIO)
+import           Control.Monad.State.Strict (StateT, get, gets, modify)
 import           Control.Monad.Trans        (lift)
 import           Data.Hashable              (Hashable (..))
 import           Data.IORef                 (IORef, atomicModifyIORef',
@@ -36,6 +40,7 @@
 
 
 newtype VertexID    = VertexID        Int deriving (Eq, Ord, Show)
+newtype StateID     = StateID         Int deriving (Eq, Ord, Show)
 newtype Epoch       = Epoch       Natural deriving (Eq, Ord, Hashable, Show)
 
 -- | 'Timestamp's represent instants in the causal timeline.
@@ -56,6 +61,9 @@
 instance Incrementable VertexID where
   inc (VertexID n) = VertexID (n + 1)
 
+instance Incrementable StateID where
+  inc (StateID n) = StateID (n + 1)
+
 instance Incrementable Epoch where
   inc (Epoch n) = Epoch (n + 1)
 
@@ -69,8 +77,10 @@
 
 data DataflowState = DataflowState {
   dfsVertices       :: Vector ErasedType,
+  dfsStates         :: Vector (IORef ErasedType),
   dfsFinalizers     :: [Timestamp -> Dataflow ()],
   dfsLastVertexID   :: VertexID,
+  dfsLastStateID    :: StateID,
   dfsLastInputEpoch :: Epoch
 }
 
@@ -83,11 +93,24 @@
 initDataflowState :: DataflowState
 initDataflowState = DataflowState {
   dfsVertices       = empty,
+  dfsStates         = empty,
   dfsFinalizers     = [],
   dfsLastVertexID   = VertexID (-1),
+  dfsLastStateID    = StateID (-1),
   dfsLastInputEpoch = Epoch 0
 }
 
+duplicateDataflowState :: Dataflow (DataflowState)
+duplicateDataflowState = Dataflow $ do
+  DataflowState{..} <- get
+
+  newStates <- liftIO $ forM dfsStates dupIORef
+
+  return $ DataflowState { dfsStates = newStates, .. }
+
+  where
+    dupIORef = readIORef >=> newIORef
+
 -- | Get the next input Epoch.
 incrementEpoch :: Dataflow Epoch
 incrementEpoch =
@@ -138,31 +161,57 @@
 -- | Mutable state that holds an `a`.
 --
 -- @since 0.1.0.0
-newtype StateRef a = StateRef (IORef a)
+newtype StateRef a = StateRef StateID
 
 -- | Create a `StateRef` initialized to the provided `a`.
 --
 -- @since 0.1.0.0
 newState :: a -> Dataflow (StateRef a)
-newState a = StateRef <$> (Dataflow $ lift $ newIORef a)
+newState a =
+  Dataflow $ do
+    sid   <- gets (dfsLastStateID >>> inc)
+    ioref <- lift $ newIORef (EraseType a)
 
+    modify $ addState ioref sid
+
+    return (StateRef sid)
+
+  where
+    addState ref sid s = s {
+      dfsStates      = dfsStates s `snoc` ref,
+      dfsLastStateID = sid
+    }
+
+lookupStateRef :: StateRef s -> Dataflow (IORef ErasedType)
+lookupStateRef (StateRef (StateID sindex)) =
+  Dataflow $ do
+    states <- gets dfsStates
+
+    return (states ! sindex)
+
 -- | Read the value stored in the `StateRef`.
 --
 -- @since 0.1.0.0
 readState :: StateRef a -> Dataflow a
-readState (StateRef ref) = Dataflow $ lift (readIORef ref)
+readState sref = do
+  ioref <- lookupStateRef sref
+  Dataflow $ lift $ (unEraseType <$> readIORef ioref)
 
 -- | Overwrite the value stored in the `StateRef`.
 --
 -- @since 0.1.0.0
 writeState :: StateRef a -> a -> Dataflow ()
-writeState (StateRef ref) x = Dataflow $ lift $ atomicWriteIORef ref x
+writeState sref x = do
+  ioref <- lookupStateRef sref
+  Dataflow $ lift $ atomicWriteIORef ioref (EraseType x)
 
 -- | Update the value stored in `StateRef`.
 --
 -- @since 0.1.0.0
 modifyState :: StateRef a -> (a -> a) -> Dataflow ()
-modifyState (StateRef ref) op = Dataflow $ lift $ atomicModifyIORef' ref (\x -> (op x, ()))
+modifyState sref op = do
+  ioref <- lookupStateRef sref
+  Dataflow $ lift $ atomicModifyIORef' ioref (\x -> (EraseType $ op (unEraseType x), ()))
 
 {-# INLINEABLE input #-}
 input :: Traversable t => t i -> Edge i -> Dataflow ()
diff --git a/src/Test/Dataflow.hs b/src/Test/Dataflow.hs
--- a/src/Test/Dataflow.hs
+++ b/src/Test/Dataflow.hs
@@ -3,6 +3,7 @@
 ) where
 
 import           Control.Concurrent.STM.TVar (newTVarIO, readTVarIO)
+import           Control.Monad               (void)
 import           Control.Monad.IO.Class      (MonadIO (..))
 import           Dataflow                    (Dataflow, Edge, compile, execute,
                                               outputTVar)
@@ -17,6 +18,6 @@
     out     <- newTVarIO []
     program <- compile (dataflow =<< outputTVar (:) out)
 
-    execute inputs program
+    void $ execute inputs program
 
     reverse <$> readTVarIO out
diff --git a/test/DataflowSpec.hs b/test/DataflowSpec.hs
--- a/test/DataflowSpec.hs
+++ b/test/DataflowSpec.hs
@@ -2,13 +2,15 @@
 
 module DataflowSpec (spec) where
 
-import           Control.Monad   ((>=>))
+import           Control.Concurrent.STM.TVar (newTVarIO, readTVarIO)
+import           Control.Monad               (void, (>=>))
 import           Dataflow
-import           Prelude         hiding (map)
+import           Prelude
 
-import           Test.Dataflow   (runDataflow)
+import           Test.Dataflow               (runDataflow)
 import           Test.Hspec
-import           Test.QuickCheck hiding (discard)
+import           Test.QuickCheck             hiding (discard)
+import           Test.QuickCheck.Modifiers   (NonEmptyList (..))
 
 
 spec :: Spec
@@ -18,6 +20,24 @@
 
     \(numbers :: [Integer]) -> runDataflow passthrough numbers `shouldReturn` numbers
 
+  describe "execute" $ do
+    it "isolates the state of runs from each other" $ property $ \(NonEmpty numbers) -> do
+      out     <- newTVarIO []
+      program <- compile (integrate =<< outputTVar (:) out)
+
+      void $ execute numbers program
+      void $ execute numbers program
+
+      (reverse <$> readTVarIO out) `shouldReturn` (scanl1 (+) numbers ++ scanl1 (+) numbers)
+
+    it "bundles all the execution state into a Program" $ property $ \(NonEmpty numbers) -> do
+      out     <- newTVarIO 0
+      program <- compile (integrate =<< outputTVar const out)
+
+      void $ execute numbers program >>= execute numbers >>= execute numbers
+
+      readTVarIO out `shouldReturn` (3 * sum numbers)
+
   describe "finalize" $ do
     it "finalizes vertices" $ property $
       \(numbers :: [Int]) -> runDataflow storeAndForward numbers `shouldReturn` numbers
@@ -37,3 +57,13 @@
     forward sref t = do
       mapM_ (send next t) =<< reverse <$> readState sref
       writeState sref []
+
+integrate :: Edge Int -> Dataflow (Edge Int)
+integrate next = statefulVertex 0 recv finalize
+  where
+    recv s t i   = do
+      modifyState s (+ i)
+
+      send next t =<< readState s
+
+    finalize _ _ = return ()