diff --git a/states/statemgr/doc.go b/states/statemgr/doc.go
new file mode 100644
index 000000000..7b7bc0f6f
--- /dev/null
+++ b/states/statemgr/doc.go
@@ -0,0 +1,21 @@
+// Package statemgr defines the interfaces and some supporting functionality
+// for "stage managers", which are components responsible for writing state
+// to some persistent storage and then later retrieving it.
+//
+// State managers will usually (but not necessarily) use the state file formats
+// implemented in the sibling directory "statefile" to serialize the persistent
+// parts of state for storage.
+//
+// State managers are responsible for ensuring that stored state can be updated
+// safely across multiple, possibly-concurrent Terraform runs (with reasonable
+// constraints and limitations). The rest of Terraform considers state to be
+// a mutable data structure, with state managers preserving that illusion
+// by creating snapshots of the state and updating them over time.
+//
+// From the perspective of callers of the general state manager API, a state
+// manager is able to return the latest snapshot and to replace that snapshot
+// with a new one. Some state managers may also preserve historical snapshots
+// using facilities offered by their storage backend, but this is always an
+// implementation detail: the historical versions are not visible to a user
+// of these interfaces.
+package statemgr
diff --git a/states/statemgr/filesystem.go b/states/statemgr/filesystem.go
new file mode 100644
index 000000000..6cfdb2a30
--- /dev/null
+++ b/states/statemgr/filesystem.go
@@ -0,0 +1,352 @@
+package statemgr
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	multierror "github.com/hashicorp/go-multierror"
+
+	"github.com/hashicorp/terraform/states"
+	"github.com/hashicorp/terraform/states/statefile"
+)
+
+// Filesystem is a full state manager that uses a file in the local filesystem
+// for persistent storage.
+//
+// The transient storage for Filesystem is always in-memory.
+type Filesystem struct {
+	mu sync.Mutex
+
+	// path is the location where a file will be created or replaced for
+	// each persistent snapshot.
+	path string
+
+	// readPath is read by RefreshState instead of "path" until the first
+	// call to PersistState, after which it is ignored.
+	//
+	// The file at readPath must never be written to by this manager.
+	readPath string
+
+	// the file handle corresponding to PathOut
+	stateFileOut *os.File
+
+	// While the stateFileOut will correspond to the lock directly,
+	// store and check the lock ID to maintain a strict state.Locker
+	// implementation.
+	lockID string
+
+	// created is set to true if stateFileOut didn't exist before we created it.
+	// This is mostly so we can clean up emtpy files during tests, but doesn't
+	// hurt to remove file we never wrote to.
+	created bool
+
+	file     *statefile.File
+	readFile *statefile.File
+	written  bool
+}
+
+var (
+	_ Full           = (*Filesystem)(nil)
+	_ PersistentMeta = (*Filesystem)(nil)
+)
+
+// NewFilesystem creates a filesystem-based state manager that reads and writes
+// state snapshots at the given filesystem path.
+//
+// This is equivalent to calling NewFileSystemBetweenPaths with statePath as
+// both of the path arguments.
+func NewFilesystem(statePath string) *Filesystem {
+	return &Filesystem{
+		path:     statePath,
+		readPath: statePath,
+	}
+}
+
+// NewFilesystemBetweenPaths creates a filesystem-based state manager that
+// reads an initial snapshot from readPath and then writes all new snapshots to
+// writePath.
+func NewFilesystemBetweenPaths(readPath, writePath string) *Filesystem {
+	return &Filesystem{
+		path:     writePath,
+		readPath: readPath,
+	}
+}
+
+// State is an implementation of Reader.
+func (s *Filesystem) State() *states.State {
+	defer s.mutex()()
+	if s.file == nil {
+		return nil
+	}
+	return s.file.DeepCopy().State
+}
+
+// WriteState is an incorrect implementation of Writer that actually also
+// persists.
+// WriteState for LocalState always persists the state as well.
+//
+// StateWriter impl.
+func (s *Filesystem) WriteState(state *states.State) error {
+	// TODO: this should use a more robust method of writing state, by first
+	// writing to a temp file on the same filesystem, and renaming the file over
+	// the original.
+
+	defer s.mutex()()
+
+	if s.stateFileOut == nil {
+		if err := s.createStateFiles(); err != nil {
+			return nil
+		}
+	}
+	defer s.stateFileOut.Sync()
+
+	s.file = s.file.DeepCopy()
+	s.file.State = state.DeepCopy()
+
+	if _, err := s.stateFileOut.Seek(0, os.SEEK_SET); err != nil {
+		return err
+	}
+	if err := s.stateFileOut.Truncate(0); err != nil {
+		return err
+	}
+
+	if state == nil {
+		// if we have no state, don't write anything else.
+		return nil
+	}
+
+	if !statefile.StatesMarshalEqual(s.file.State, s.readFile.State) {
+		s.file.Serial++
+	}
+
+	if err := statefile.Write(s.file, s.stateFileOut); err != nil {
+		return err
+	}
+
+	s.written = true
+	return nil
+}
+
+// PersistState is an implementation of Persister that does nothing because
+// this type's Writer implementation does its own persistence.
+func (s *Filesystem) PersistState() error {
+	return nil
+}
+
+// RefreshState is an implementation of Refresher.
+func (s *Filesystem) RefreshState() error {
+	defer s.mutex()()
+
+	var reader io.Reader
+
+	// The s.readPath file is only OK to read if we have not written any state out
+	// (in which case the same state needs to be read in), and no state output file
+	// has been opened (possibly via a lock) or the input path is different
+	// than the output path.
+	// This is important for Windows, as if the input file is the same as the
+	// output file, and the output file has been locked already, we can't open
+	// the file again.
+	if !s.written && (s.stateFileOut == nil || s.readPath != s.path) {
+		// we haven't written a state file yet, so load from readPath
+		f, err := os.Open(s.readPath)
+		if err != nil {
+			// It is okay if the file doesn't exist; we'll treat that as a nil state.
+			if !os.IsNotExist(err) {
+				return err
+			}
+
+			// we need a non-nil reader for ReadState and an empty buffer works
+			// to return EOF immediately
+			reader = bytes.NewBuffer(nil)
+
+		} else {
+			defer f.Close()
+			reader = f
+		}
+	} else {
+		// no state to refresh
+		if s.stateFileOut == nil {
+			return nil
+		}
+
+		// we have a state file, make sure we're at the start
+		s.stateFileOut.Seek(0, os.SEEK_SET)
+		reader = s.stateFileOut
+	}
+
+	f, err := statefile.Read(reader)
+	// if there's no state we just assign the nil return value
+	if err != nil && err != statefile.ErrNoState {
+		return err
+	}
+
+	s.file = f
+	s.readFile = s.file.DeepCopy()
+	return nil
+}
+
+// Lock implements Locker using filesystem discretionary locks.
+func (s *Filesystem) Lock(info *LockInfo) (string, error) {
+	defer s.mutex()()
+
+	if s.stateFileOut == nil {
+		if err := s.createStateFiles(); err != nil {
+			return "", err
+		}
+	}
+
+	if s.lockID != "" {
+		return "", fmt.Errorf("state %q already locked", s.stateFileOut.Name())
+	}
+
+	if err := s.lock(); err != nil {
+		info, infoErr := s.lockInfo()
+		if infoErr != nil {
+			err = multierror.Append(err, infoErr)
+		}
+
+		lockErr := &LockError{
+			Info: info,
+			Err:  err,
+		}
+
+		return "", lockErr
+	}
+
+	s.lockID = info.ID
+	return s.lockID, s.writeLockInfo(info)
+}
+
+// Unlock is the companion to Lock, completing the implemention of Locker.
+func (s *Filesystem) Unlock(id string) error {
+	defer s.mutex()()
+
+	if s.lockID == "" {
+		return fmt.Errorf("LocalState not locked")
+	}
+
+	if id != s.lockID {
+		idErr := fmt.Errorf("invalid lock id: %q. current id: %q", id, s.lockID)
+		info, err := s.lockInfo()
+		if err != nil {
+			err = multierror.Append(idErr, err)
+		}
+
+		return &LockError{
+			Err:  idErr,
+			Info: info,
+		}
+	}
+
+	os.Remove(s.lockInfoPath())
+
+	fileName := s.stateFileOut.Name()
+
+	unlockErr := s.unlock()
+
+	s.stateFileOut.Close()
+	s.stateFileOut = nil
+	s.lockID = ""
+
+	// clean up the state file if we created it an never wrote to it
+	stat, err := os.Stat(fileName)
+	if err == nil && stat.Size() == 0 && s.created {
+		os.Remove(fileName)
+	}
+
+	return unlockErr
+}
+
+// StateSnapshotMeta returns the metadata from the most recently persisted
+// or refreshed persistent state snapshot.
+//
+// This is an implementation of PersistentMeta.
+func (s *Filesystem) StateSnapshotMeta() SnapshotMeta {
+	if s.file == nil {
+		return SnapshotMeta{} // placeholder
+	}
+
+	return SnapshotMeta{
+		Lineage: s.file.Lineage,
+		Serial:  s.file.Serial,
+
+		TerraformVersion: s.file.TerraformVersion,
+	}
+}
+
+// Open the state file, creating the directories and file as needed.
+func (s *Filesystem) createStateFiles() error {
+
+	// This could race, but we only use it to clean up empty files
+	if _, err := os.Stat(s.path); os.IsNotExist(err) {
+		s.created = true
+	}
+
+	// Create all the directories
+	if err := os.MkdirAll(filepath.Dir(s.path), 0755); err != nil {
+		return err
+	}
+
+	f, err := os.OpenFile(s.path, os.O_RDWR|os.O_CREATE, 0666)
+	if err != nil {
+		return err
+	}
+
+	s.stateFileOut = f
+	return nil
+}
+
+// return the path for the lockInfo metadata.
+func (s *Filesystem) lockInfoPath() string {
+	stateDir, stateName := filepath.Split(s.readPath)
+	if stateName == "" {
+		panic("empty state file path")
+	}
+
+	if stateName[0] == '.' {
+		stateName = stateName[1:]
+	}
+
+	return filepath.Join(stateDir, fmt.Sprintf(".%s.lock.info", stateName))
+}
+
+// lockInfo returns the data in a lock info file
+func (s *Filesystem) lockInfo() (*LockInfo, error) {
+	path := s.lockInfoPath()
+	infoData, err := ioutil.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+
+	info := LockInfo{}
+	err = json.Unmarshal(infoData, &info)
+	if err != nil {
+		return nil, fmt.Errorf("state file %q locked, but could not unmarshal lock info: %s", s.readPath, err)
+	}
+	return &info, nil
+}
+
+// write a new lock info file
+func (s *Filesystem) writeLockInfo(info *LockInfo) error {
+	path := s.lockInfoPath()
+	info.Path = s.readPath
+	info.Created = time.Now().UTC()
+
+	err := ioutil.WriteFile(path, info.Marshal(), 0600)
+	if err != nil {
+		return fmt.Errorf("could not write lock info for %q: %s", s.readPath, err)
+	}
+	return nil
+}
+
+func (s *Filesystem) mutex() func() {
+	s.mu.Lock()
+	return s.mu.Unlock
+}
diff --git a/states/statemgr/filesystem_lock_unix.go b/states/statemgr/filesystem_lock_unix.go
new file mode 100644
index 000000000..6b2672e7f
--- /dev/null
+++ b/states/statemgr/filesystem_lock_unix.go
@@ -0,0 +1,34 @@
+// +build !windows
+
+package statemgr
+
+import (
+	"os"
+	"syscall"
+)
+
+// use fcntl POSIX locks for the most consistent behavior across platforms, and
+// hopefully some campatibility over NFS and CIFS.
+func (s *Filesystem) lock() error {
+	flock := &syscall.Flock_t{
+		Type:   syscall.F_RDLCK | syscall.F_WRLCK,
+		Whence: int16(os.SEEK_SET),
+		Start:  0,
+		Len:    0,
+	}
+
+	fd := s.stateFileOut.Fd()
+	return syscall.FcntlFlock(fd, syscall.F_SETLK, flock)
+}
+
+func (s *Filesystem) unlock() error {
+	flock := &syscall.Flock_t{
+		Type:   syscall.F_UNLCK,
+		Whence: int16(os.SEEK_SET),
+		Start:  0,
+		Len:    0,
+	}
+
+	fd := s.stateFileOut.Fd()
+	return syscall.FcntlFlock(fd, syscall.F_SETLK, flock)
+}
diff --git a/states/statemgr/filesystem_lock_windows.go b/states/statemgr/filesystem_lock_windows.go
new file mode 100644
index 000000000..1e2f49fab
--- /dev/null
+++ b/states/statemgr/filesystem_lock_windows.go
@@ -0,0 +1,108 @@
+// +build windows
+
+package statemgr
+
+import (
+	"math"
+	"syscall"
+	"unsafe"
+)
+
+var (
+	modkernel32      = syscall.NewLazyDLL("kernel32.dll")
+	procLockFileEx   = modkernel32.NewProc("LockFileEx")
+	procCreateEventW = modkernel32.NewProc("CreateEventW")
+)
+
+const (
+	// dwFlags defined for LockFileEx
+	// https://msdn.microsoft.com/en-us/library/windows/desktop/aa365203(v=vs.85).aspx
+	_LOCKFILE_FAIL_IMMEDIATELY = 1
+	_LOCKFILE_EXCLUSIVE_LOCK   = 2
+)
+
+func (s *Filesystem) lock() error {
+	// even though we're failing immediately, an overlapped event structure is
+	// required
+	ol, err := newOverlapped()
+	if err != nil {
+		return err
+	}
+	defer syscall.CloseHandle(ol.HEvent)
+
+	return lockFileEx(
+		syscall.Handle(s.stateFileOut.Fd()),
+		_LOCKFILE_EXCLUSIVE_LOCK|_LOCKFILE_FAIL_IMMEDIATELY,
+		0,              // reserved
+		0,              // bytes low
+		math.MaxUint32, // bytes high
+		ol,
+	)
+}
+
+func (s *Filesystem) unlock() error {
+	// the file is closed in Unlock
+	return nil
+}
+
+func lockFileEx(h syscall.Handle, flags, reserved, locklow, lockhigh uint32, ol *syscall.Overlapped) (err error) {
+	r1, _, e1 := syscall.Syscall6(
+		procLockFileEx.Addr(),
+		6,
+		uintptr(h),
+		uintptr(flags),
+		uintptr(reserved),
+		uintptr(locklow),
+		uintptr(lockhigh),
+		uintptr(unsafe.Pointer(ol)),
+	)
+	if r1 == 0 {
+		if e1 != 0 {
+			err = error(e1)
+		} else {
+			err = syscall.EINVAL
+		}
+	}
+	return
+}
+
+// newOverlapped creates a structure used to track asynchronous
+// I/O requests that have been issued.
+func newOverlapped() (*syscall.Overlapped, error) {
+	event, err := createEvent(nil, true, false, nil)
+	if err != nil {
+		return nil, err
+	}
+	return &syscall.Overlapped{HEvent: event}, nil
+}
+
+func createEvent(sa *syscall.SecurityAttributes, manualReset bool, initialState bool, name *uint16) (handle syscall.Handle, err error) {
+	var _p0 uint32
+	if manualReset {
+		_p0 = 1
+	}
+	var _p1 uint32
+	if initialState {
+		_p1 = 1
+	}
+
+	r0, _, e1 := syscall.Syscall6(
+		procCreateEventW.Addr(),
+		4,
+		uintptr(unsafe.Pointer(sa)),
+		uintptr(_p0),
+		uintptr(_p1),
+		uintptr(unsafe.Pointer(name)),
+		0,
+		0,
+	)
+	handle = syscall.Handle(r0)
+	if handle == syscall.InvalidHandle {
+		if e1 != 0 {
+			err = error(e1)
+		} else {
+			err = syscall.EINVAL
+		}
+	}
+	return
+}
diff --git a/states/statemgr/filesystem_test.go b/states/statemgr/filesystem_test.go
new file mode 100644
index 000000000..ae7868db3
--- /dev/null
+++ b/states/statemgr/filesystem_test.go
@@ -0,0 +1,221 @@
+package statemgr
+
+import (
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"sync"
+	"testing"
+
+	version "github.com/hashicorp/go-version"
+
+	"github.com/hashicorp/terraform/states/statefile"
+)
+
+func TestFilesystem(t *testing.T) {
+	ls := testFilesystem(t)
+	defer os.Remove(ls.readPath)
+	TestFull(t, ls)
+}
+
+func TestFilesystemRace(t *testing.T) {
+	ls := testFilesystem(t)
+	defer os.Remove(ls.readPath)
+
+	current := TestFullInitialState()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ls.WriteState(current)
+		}()
+	}
+}
+
+func TestFilesystemLocks(t *testing.T) {
+	s := testFilesystem(t)
+	defer os.Remove(s.readPath)
+
+	// lock first
+	info := NewLockInfo()
+	info.Operation = "test"
+	lockID, err := s.Lock(info)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	out, err := exec.Command("go", "run", "testdata/lockstate.go", s.path).CombinedOutput()
+	if err != nil {
+		t.Fatal("unexpected lock failure", err, string(out))
+	}
+
+	if string(out) != "lock failed" {
+		t.Fatal("expected 'locked failed', got", string(out))
+	}
+
+	// check our lock info
+	lockInfo, err := s.lockInfo()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if lockInfo.Operation != "test" {
+		t.Fatalf("invalid lock info %#v\n", lockInfo)
+	}
+
+	// a noop, since we unlock on exit
+	if err := s.Unlock(lockID); err != nil {
+		t.Fatal(err)
+	}
+
+	// local locks can re-lock
+	lockID, err = s.Lock(info)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.Unlock(lockID); err != nil {
+		t.Fatal(err)
+	}
+
+	// we should not be able to unlock the same lock twice
+	if err := s.Unlock(lockID); err == nil {
+		t.Fatal("unlocking an unlocked state should fail")
+	}
+
+	// make sure lock info is gone
+	lockInfoPath := s.lockInfoPath()
+	if _, err := os.Stat(lockInfoPath); !os.IsNotExist(err) {
+		t.Fatal("lock info not removed")
+	}
+}
+
+// Verify that we can write to the state file, as Windows' mandatory locking
+// will prevent writing to a handle different than the one that hold the lock.
+func TestFilesystem_writeWhileLocked(t *testing.T) {
+	s := testFilesystem(t)
+	defer os.Remove(s.readPath)
+
+	// lock first
+	info := NewLockInfo()
+	info.Operation = "test"
+	lockID, err := s.Lock(info)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := s.Unlock(lockID); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	if err := s.WriteState(TestFullInitialState()); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestFilesystem_pathOut(t *testing.T) {
+	f, err := ioutil.TempFile("", "tf")
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	f.Close()
+	defer os.Remove(f.Name())
+
+	ls := testFilesystem(t)
+	ls.path = f.Name()
+	defer os.Remove(ls.path)
+
+	TestFull(t, ls)
+}
+
+func TestFilesystem_nonExist(t *testing.T) {
+	ls := NewFilesystem("ishouldntexist")
+	if err := ls.RefreshState(); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if state := ls.State(); state != nil {
+		t.Fatalf("bad: %#v", state)
+	}
+}
+
+func TestFilesystem_impl(t *testing.T) {
+	var _ Reader = new(Filesystem)
+	var _ Writer = new(Filesystem)
+	var _ Persister = new(Filesystem)
+	var _ Refresher = new(Filesystem)
+	var _ Locker = new(Filesystem)
+}
+
+func testFilesystem(t *testing.T) *Filesystem {
+	f, err := ioutil.TempFile("", "tf")
+	if err != nil {
+		t.Fatalf("failed to create temporary file %s", err)
+	}
+	t.Logf("temporary state file at %s", f.Name())
+
+	err = statefile.Write(&statefile.File{
+		Lineage:          "test-lineage",
+		Serial:           0,
+		TerraformVersion: version.Must(version.NewVersion("1.2.3")),
+		State:            TestFullInitialState(),
+	}, f)
+	if err != nil {
+		t.Fatalf("failed to write initial state to %s: %s", f.Name(), err)
+	}
+	f.Close()
+
+	ls := NewFilesystem(f.Name())
+	if err := ls.RefreshState(); err != nil {
+		t.Fatalf("initial refresh failed: %s", err)
+	}
+
+	return ls
+}
+
+// Make sure we can refresh while the state is locked
+func TestFilesystem_refreshWhileLocked(t *testing.T) {
+	f, err := ioutil.TempFile("", "tf")
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	err = statefile.Write(&statefile.File{
+		Lineage:          "test-lineage",
+		Serial:           0,
+		TerraformVersion: version.Must(version.NewVersion("1.2.3")),
+		State:            TestFullInitialState(),
+	}, f)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	f.Close()
+
+	s := NewFilesystem(f.Name())
+	defer os.Remove(s.path)
+
+	// lock first
+	info := NewLockInfo()
+	info.Operation = "test"
+	lockID, err := s.Lock(info)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := s.Unlock(lockID); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	if err := s.RefreshState(); err != nil {
+		t.Fatal(err)
+	}
+
+	readState := s.State()
+	if readState == nil {
+		t.Fatal("missing state")
+	}
+}
diff --git a/states/statemgr/interfaces.go b/states/statemgr/interfaces.go
new file mode 100644
index 000000000..37f621d2d
--- /dev/null
+++ b/states/statemgr/interfaces.go
@@ -0,0 +1 @@
+package statemgr
diff --git a/states/statemgr/locker.go b/states/statemgr/locker.go
new file mode 100644
index 000000000..25f3f5587
--- /dev/null
+++ b/states/statemgr/locker.go
@@ -0,0 +1,224 @@
+package statemgr
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"html/template"
+	"math/rand"
+	"os"
+	"os/user"
+	"strings"
+	"time"
+
+	uuid "github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/terraform/version"
+)
+
+var rngSource = rand.New(rand.NewSource(time.Now().UnixNano()))
+
+// Locker is the interface for state managers that are able to manage
+// mutual-exclusion locks for state.
+//
+// Implementing Locker alongside Persistent relaxes some of the usual
+// implemention constraints for implementations of Refresher and Persister,
+// under the assumption that the locking mechanism effectively prevents
+// multiple Terraform processes from reading and writing state concurrently.
+// In particular, a type that implements both Locker and Persistent is only
+// required to that the Persistent implementation is concurrency-safe within
+// a single Terraform process.
+//
+// A Locker implementation must ensure that another processes with a
+// similarly-configured state manager cannot successfully obtain a lock while
+// the current process is holding it, or vice-versa, assuming that both
+// processes agree on the locking mechanism.
+//
+// A Locker is not required to prevent non-cooperating processes from
+// concurrently modifying the state, but is free to do so as an extra
+// protection. If a mandatory locking mechanism of this sort is implemented,
+// the state manager must ensure that RefreshState and PersistState calls
+// can succeed if made through the same manager instance that is holding the
+// lock, such has by retaining some sort of lock token that the Persistent
+// methods can then use.
+type Locker interface {
+	// Lock attempts to obtain a lock, using the given lock information.
+	//
+	// The result is an opaque id that can be passed to Unlock to release
+	// the lock, or an error if the lock cannot be acquired. Lock returns
+	// an instance of LockError immediately if the lock is already held,
+	// and the helper function LockWithContext uses this to automatically
+	// retry lock acquisition periodically until a timeout is reached.
+	Lock(info *LockInfo) (string, error)
+
+	// Unlock releases a lock previously acquired by Lock.
+	//
+	// If the lock cannot be released -- for example, if it was stolen by
+	// another user with some sort of administrative override privilege --
+	// then an error is returned explaining the situation in a way that
+	// is suitable for returning to an end-user.
+	Unlock(id string) error
+}
+
+// test hook to verify that LockWithContext has attempted a lock
+var postLockHook func()
+
+// LockWithContext locks the given state manager using the provided context
+// for both timeout and cancellation.
+//
+// This method has a built-in retry/backoff behavior up to the context's
+// timeout.
+func LockWithContext(ctx context.Context, s Locker, info *LockInfo) (string, error) {
+	delay := time.Second
+	maxDelay := 16 * time.Second
+	for {
+		id, err := s.Lock(info)
+		if err == nil {
+			return id, nil
+		}
+
+		le, ok := err.(*LockError)
+		if !ok {
+			// not a lock error, so we can't retry
+			return "", err
+		}
+
+		if le == nil || le.Info == nil || le.Info.ID == "" {
+			// If we don't have a complete LockError then there's something
+			// wrong with the lock.
+			return "", err
+		}
+
+		if postLockHook != nil {
+			postLockHook()
+		}
+
+		// there's an existing lock, wait and try again
+		select {
+		case <-ctx.Done():
+			// return the last lock error with the info
+			return "", err
+		case <-time.After(delay):
+			if delay < maxDelay {
+				delay *= 2
+			}
+		}
+	}
+}
+
+// LockInfo stores lock metadata.
+//
+// Only Operation and Info are required to be set by the caller of Lock.
+// Most callers should use NewLockInfo to create a LockInfo value with many
+// of the fields populated with suitable default values.
+type LockInfo struct {
+	// Unique ID for the lock. NewLockInfo provides a random ID, but this may
+	// be overridden by the lock implementation. The final value if ID will be
+	// returned by the call to Lock.
+	ID string
+
+	// Terraform operation, provided by the caller.
+	Operation string
+
+	// Extra information to store with the lock, provided by the caller.
+	Info string
+
+	// user@hostname when available
+	Who string
+
+	// Terraform version
+	Version string
+
+	// Time that the lock was taken.
+	Created time.Time
+
+	// Path to the state file when applicable. Set by the Lock implementation.
+	Path string
+}
+
+// NewLockInfo creates a LockInfo object and populates many of its fields
+// with suitable default values.
+func NewLockInfo() *LockInfo {
+	// this doesn't need to be cryptographically secure, just unique.
+	// Using math/rand alleviates the need to check handle the read error.
+	// Use a uuid format to match other IDs used throughout Terraform.
+	buf := make([]byte, 16)
+	rngSource.Read(buf)
+
+	id, err := uuid.FormatUUID(buf)
+	if err != nil {
+		// this of course shouldn't happen
+		panic(err)
+	}
+
+	// don't error out on user and hostname, as we don't require them
+	userName := ""
+	if userInfo, err := user.Current(); err == nil {
+		userName = userInfo.Username
+	}
+	host, _ := os.Hostname()
+
+	info := &LockInfo{
+		ID:      id,
+		Who:     fmt.Sprintf("%s@%s", userName, host),
+		Version: version.Version,
+		Created: time.Now().UTC(),
+	}
+	return info
+}
+
+// Err returns the lock info formatted in an error
+func (l *LockInfo) Err() error {
+	return errors.New(l.String())
+}
+
+// Marshal returns a string json representation of the LockInfo
+func (l *LockInfo) Marshal() []byte {
+	js, err := json.Marshal(l)
+	if err != nil {
+		panic(err)
+	}
+	return js
+}
+
+// String return a multi-line string representation of LockInfo
+func (l *LockInfo) String() string {
+	tmpl := `Lock Info:
+  ID:        {{.ID}}
+  Path:      {{.Path}}
+  Operation: {{.Operation}}
+  Who:       {{.Who}}
+  Version:   {{.Version}}
+  Created:   {{.Created}}
+  Info:      {{.Info}}
+`
+
+	t := template.Must(template.New("LockInfo").Parse(tmpl))
+	var out bytes.Buffer
+	if err := t.Execute(&out, l); err != nil {
+		panic(err)
+	}
+	return out.String()
+}
+
+// LockError is a specialization of type error that is returned by Locker.Lock
+// to indicate that the lock is already held by another process and that
+// retrying may be productive to take the lock once the other process releases
+// it.
+type LockError struct {
+	Info *LockInfo
+	Err  error
+}
+
+func (e *LockError) Error() string {
+	var out []string
+	if e.Err != nil {
+		out = append(out, e.Err.Error())
+	}
+
+	if e.Info != nil {
+		out = append(out, e.Info.String())
+	}
+	return strings.Join(out, "\n")
+}
diff --git a/states/statemgr/persistent.go b/states/statemgr/persistent.go
new file mode 100644
index 000000000..c15e84af2
--- /dev/null
+++ b/states/statemgr/persistent.go
@@ -0,0 +1,104 @@
+package statemgr
+
+import (
+	version "github.com/hashicorp/go-version"
+)
+
+// Persistent is a union of the Refresher and Persistent interfaces, for types
+// that deal with persistent snapshots.
+//
+// Persistent snapshots are ones that are retained in storage that will
+// outlive a particular Terraform process, and are shared with other Terraform
+// processes that have a similarly-configured state manager.
+//
+// A manager may also choose to retain historical persistent snapshots, but
+// that is an implementation detail and not visible via this API.
+type Persistent interface {
+	Refresher
+	Persister
+}
+
+// Refresher is the interface for managers that can read snapshots from
+// persistent storage.
+//
+// Refresher is usually implemented in conjunction with Reader, with
+// RefreshState copying the latest persistent snapshot into the latest
+// transient snapshot.
+//
+// For a type that implements both Refresher and Persister, RefreshState must
+// return the result of the most recently completed successful call to
+// PersistState, unless another concurrently-running process has persisted
+// another snapshot in the mean time.
+//
+// The Refresher implementation must guarantee that the snapshot is read
+// from persistent storage in a way that is safe under concurrent calls to
+// PersistState that may be happening in other processes.
+type Refresher interface {
+	// RefreshState retrieves a snapshot of state from persistent storage,
+	// returning an error if this is not possible.
+	//
+	// Types that implement RefreshState generally also implement a State
+	// method that returns the result of the latest successful refresh.
+	//
+	// Since only a subset of the data in a state is included when persisting,
+	// a round-trip through PersistState and then RefreshState will often
+	// return only a subset of what was written. Callers must assume that
+	// ephemeral portions of the state may be unpopulated after calling
+	// RefreshState.
+	RefreshState() error
+}
+
+// Persister is the interface for managers that can write snapshots to
+// persistent storage.
+//
+// Persister is usually implemented in conjunction with Writer, with
+// PersistState copying the latest transient snapshot to be the new latest
+// persistent snapshot.
+//
+// A Persister implementation must detect updates made by other processes
+// that may be running concurrently and avoid destroying those changes. This
+// is most commonly achieved by making use of atomic write capabilities on
+// the remote storage backend in conjunction with book-keeping with the
+// Serial and Lineage fields in the standard state file formats.
+type Persister interface {
+	PersistState() error
+}
+
+// PersistentMeta is an optional extension to Persistent that allows inspecting
+// the metadata associated with the snapshot that was most recently either
+// read by RefreshState or written by PersistState.
+type PersistentMeta interface {
+	// StateSnapshotMeta returns metadata about the state snapshot most
+	// recently created either by a call to PersistState or read by a call
+	// to RefreshState.
+	//
+	// If no persistent snapshot is yet available in the manager then
+	// the return value is meaningless. This method is primarily available
+	// for testing and logging purposes, and is of little use otherwise.
+	StateSnapshotMeta() SnapshotMeta
+}
+
+// SnapshotMeta contains metadata about a persisted state snapshot.
+//
+// This metadata is usually (but not necessarily) included as part of the
+// "header" of a state file, which is then written to a raw blob storage medium
+// by a persistent state manager.
+//
+// Not all state managers will have useful values for all fields in this
+// struct, so SnapshotMeta values are of little use beyond testing and logging
+// use-cases.
+type SnapshotMeta struct {
+	// Lineage and Serial can be used to understand the relationships between
+	// snapshots.
+	//
+	// If two snapshots both have an identical, non-empty Lineage
+	// then the one with the higher Serial is newer than the other.
+	// If the Lineage values are different or empty then the two snapshots
+	// are unrelated and cannot be compared for relative age.
+	Lineage string
+	Serial  uint64
+
+	// TerraformVersion is the number of the version of Terraform that created
+	// the snapshot.
+	TerraformVersion *version.Version
+}
diff --git a/states/statemgr/statemgr.go b/states/statemgr/statemgr.go
new file mode 100644
index 000000000..355eef055
--- /dev/null
+++ b/states/statemgr/statemgr.go
@@ -0,0 +1,16 @@
+package statemgr
+
+// Full is the union of all of the more-specific state interfaces.
+//
+// This interface may grow over time, so state implementations aiming to
+// implement it may need to be modified for future changes. To ensure that
+// this need can be detected, always include a statement nearby the declaration
+// of the implementing type that will fail at compile time if the interface
+// isn't satisfied, such as:
+//
+//     var _ statemgr.Full = (*ImplementingType)(nil)
+type Full interface {
+	Transient
+	Persistent
+	Locker
+}
diff --git a/states/statemgr/statemgr_fake.go b/states/statemgr/statemgr_fake.go
new file mode 100644
index 000000000..42d2b9bb3
--- /dev/null
+++ b/states/statemgr/statemgr_fake.go
@@ -0,0 +1,96 @@
+package statemgr
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/hashicorp/terraform/states"
+)
+
+// NewFullFake returns a full state manager that really only supports transient
+// snapshots. This is primarily intended for testing and is not suitable for
+// general use.
+//
+// The persistent part of the interface is stubbed out as an in-memory store,
+// and so its snapshots are effectively also transient.
+//
+// The given Transient implementation is used to implement the transient
+// portion of the interface. If nil is given, NewTransientInMemory is
+// automatically called to create an in-memory transient manager with no
+// initial transient snapshot.
+//
+// If the given initial state is non-nil then a copy of it will be used as
+// the initial persistent snapshot.
+//
+// The Locker portion of the returned manager uses a local mutex to simulate
+// mutually-exclusive access to the fake persistent portion of the object.
+func NewFullFake(t Transient, initial *states.State) Full {
+	if t == nil {
+		t = NewTransientInMemory(nil)
+	}
+
+	// The "persistent" part of our manager is actually just another in-memory
+	// transient used to fake a secondary storage layer.
+	fakeP := NewTransientInMemory(initial.DeepCopy())
+
+	return &fakeFull{
+		t:     t,
+		fakeP: fakeP,
+	}
+}
+
+type fakeFull struct {
+	t     Transient
+	fakeP Transient
+
+	lockLock sync.Mutex
+	locked   bool
+}
+
+var _ Full = (*fakeFull)(nil)
+
+func (m *fakeFull) State() *states.State {
+	return m.t.State()
+}
+
+func (m *fakeFull) WriteState(s *states.State) error {
+	return m.t.WriteState(s)
+}
+
+func (m *fakeFull) RefreshState() error {
+	return m.t.WriteState(m.fakeP.State())
+}
+
+func (m *fakeFull) PersistState() error {
+	return m.fakeP.WriteState(m.t.State())
+}
+
+func (m *fakeFull) Lock(info *LockInfo) (string, error) {
+	m.lockLock.Lock()
+	defer m.lockLock.Unlock()
+
+	if m.locked {
+		return "", &LockError{
+			Err:  errors.New("fake state manager is locked"),
+			Info: info,
+		}
+	}
+
+	m.locked = true
+	return "placeholder", nil
+}
+
+func (m *fakeFull) Unlock(id string) error {
+	m.lockLock.Lock()
+	defer m.lockLock.Unlock()
+
+	if !m.locked {
+		return errors.New("fake state manager is not locked")
+	}
+	if id != "placeholder" {
+		return errors.New("wrong lock id for fake state manager")
+	}
+
+	m.locked = false
+	return nil
+}
diff --git a/states/statemgr/statemgr_test.go b/states/statemgr/statemgr_test.go
new file mode 100644
index 000000000..62625780f
--- /dev/null
+++ b/states/statemgr/statemgr_test.go
@@ -0,0 +1,102 @@
+package statemgr
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"io/ioutil"
+	"log"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/terraform/helper/logging"
+)
+
+func TestNewLockInfo(t *testing.T) {
+	info1 := NewLockInfo()
+	info2 := NewLockInfo()
+
+	if info1.ID == "" {
+		t.Fatal("LockInfo missing ID")
+	}
+
+	if info1.Version == "" {
+		t.Fatal("LockInfo missing version")
+	}
+
+	if info1.Created.IsZero() {
+		t.Fatal("LockInfo missing Created")
+	}
+
+	if info1.ID == info2.ID {
+		t.Fatal("multiple LockInfo with identical IDs")
+	}
+
+	// test the JSON output is valid
+	newInfo := &LockInfo{}
+	err := json.Unmarshal(info1.Marshal(), newInfo)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestLockWithContext(t *testing.T) {
+	s := NewFullFake(nil, TestFullInitialState())
+
+	id, err := s.Lock(NewLockInfo())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// use a cancelled context for an immediate timeout
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	info := NewLockInfo()
+	info.Info = "lock with context"
+	_, err = LockWithContext(ctx, s, info)
+	if err == nil {
+		t.Fatal("lock should have failed immediately")
+	}
+
+	// block until LockwithContext has made a first attempt
+	attempted := make(chan struct{})
+	postLockHook = func() {
+		close(attempted)
+		postLockHook = nil
+	}
+
+	// unlock the state during LockWithContext
+	unlocked := make(chan struct{})
+	go func() {
+		defer close(unlocked)
+		<-attempted
+		if err := s.Unlock(id); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	ctx, cancel = context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	id, err = LockWithContext(ctx, s, info)
+	if err != nil {
+		t.Fatal("lock should have completed within 2s:", err)
+	}
+
+	// ensure the goruotine completes
+	<-unlocked
+}
+
+func TestMain(m *testing.M) {
+	flag.Parse()
+	if testing.Verbose() {
+		// if we're verbose, use the logging requested by TF_LOG
+		logging.SetOutput()
+	} else {
+		// otherwise silence all logs
+		log.SetOutput(ioutil.Discard)
+	}
+	os.Exit(m.Run())
+}
diff --git a/states/statemgr/testdata/lockstate.go b/states/statemgr/testdata/lockstate.go
new file mode 100644
index 000000000..f0b336068
--- /dev/null
+++ b/states/statemgr/testdata/lockstate.go
@@ -0,0 +1,28 @@
+package main
+
+import (
+	"io"
+	"log"
+	"os"
+
+	"github.com/hashicorp/terraform/states/statemgr"
+)
+
+// Attempt to open and lock a terraform state file.
+// Lock failure exits with 0 and writes "lock failed" to stderr.
+func main() {
+	if len(os.Args) != 2 {
+		log.Fatal(os.Args[0], "statefile")
+	}
+
+	s := statemgr.NewFilesystem(os.Args[1])
+
+	info := statemgr.NewLockInfo()
+	info.Operation = "test"
+	info.Info = "state locker"
+
+	_, err := s.Lock(info)
+	if err != nil {
+		io.WriteString(os.Stderr, "lock failed")
+	}
+}
diff --git a/states/statemgr/testing.go b/states/statemgr/testing.go
new file mode 100644
index 000000000..8b2f2cbb3
--- /dev/null
+++ b/states/statemgr/testing.go
@@ -0,0 +1,157 @@
+package statemgr
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/davecgh/go-spew/spew"
+
+	"github.com/hashicorp/terraform/states/statefile"
+
+	"github.com/hashicorp/terraform/addrs"
+	"github.com/zclconf/go-cty/cty"
+
+	"github.com/hashicorp/terraform/states"
+)
+
+// TestFull is a helper for testing full state manager implementations. It
+// expects that the given implementation is pre-loaded with a snapshot of the
+// result from TestFullInitialState.
+//
+// If the given state manager also implements PersistentMeta, this function
+// will test that the snapshot metadata changes as expected between calls
+// to the methods of Persistent.
+func TestFull(t *testing.T, s Full) {
+	t.Helper()
+
+	if err := s.RefreshState(); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	// Check that the initial state is correct.
+	// These do have different Lineages, but we will replace current below.
+	initial := TestFullInitialState()
+	if state := s.State(); !state.Equal(initial) {
+		t.Fatalf("state does not match expected initial state\n\ngot:\n%s\nwant:\n%s", spew.Sdump(state), spew.Sdump(initial))
+	}
+
+	var initialMeta SnapshotMeta
+	if sm, ok := s.(PersistentMeta); ok {
+		initialMeta = sm.StateSnapshotMeta()
+	}
+
+	// Now we've proven that the state we're starting with is an initial
+	// state, we'll complete our work here with that state, since otherwise
+	// further writes would violate the invariant that we only try to write
+	// states that share the same lineage as what was initially written.
+	current := s.State()
+
+	// Write a new state and verify that we have it
+	current.RootModule().SetOutputValue("bar", cty.StringVal("baz"), false)
+
+	if err := s.WriteState(current); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if actual := s.State(); !actual.Equal(current) {
+		t.Fatalf("bad:\n%#v\n\n%#v", actual, current)
+	}
+
+	// Test persistence
+	if err := s.PersistState(); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	// Refresh if we got it
+	if err := s.RefreshState(); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	var newMeta SnapshotMeta
+	if sm, ok := s.(PersistentMeta); ok {
+		newMeta = sm.StateSnapshotMeta()
+		if got, want := newMeta.Lineage, initialMeta.Lineage; got != want {
+			t.Errorf("Lineage changed from %q to %q", want, got)
+		}
+		if after, before := newMeta.Serial, initialMeta.Serial; after == before {
+			t.Errorf("Serial didn't change from %d after new module added", before)
+		}
+	}
+
+	// Same serial
+	serial := newMeta.Serial
+	if err := s.WriteState(current); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	if err := s.PersistState(); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if sm, ok := s.(PersistentMeta); ok {
+		newMeta = sm.StateSnapshotMeta()
+		if newMeta.Serial != serial {
+			t.Fatalf("serial changed after persisting with no changes: got %d, want %d", newMeta.Serial, serial)
+		}
+	}
+
+	if sm, ok := s.(PersistentMeta); ok {
+		newMeta = sm.StateSnapshotMeta()
+	}
+
+	// Change the serial
+	current = current.DeepCopy()
+	current.EnsureModule(addrs.RootModuleInstance).SetOutputValue(
+		"serialCheck", cty.StringVal("true"), false,
+	)
+	if err := s.WriteState(current); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	if err := s.PersistState(); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if sm, ok := s.(PersistentMeta); ok {
+		oldMeta := newMeta
+		newMeta = sm.StateSnapshotMeta()
+
+		if newMeta.Serial <= serial {
+			t.Fatalf("serial incorrect after persisting with changes: got %d, want > %d", newMeta.Serial, serial)
+		}
+
+		if newMeta.TerraformVersion != oldMeta.TerraformVersion {
+			t.Fatalf("TFVersion changed from %s to %s", oldMeta.TerraformVersion, newMeta.TerraformVersion)
+		}
+
+		// verify that Lineage doesn't change along with Serial, or during copying.
+		if newMeta.Lineage != oldMeta.Lineage {
+			t.Fatalf("Lineage changed from %q to %q", oldMeta.Lineage, newMeta.Lineage)
+		}
+	}
+
+	// Check that State() returns a copy by modifying the copy and comparing
+	// to the current state.
+	stateCopy := s.State()
+	stateCopy.EnsureModule(addrs.RootModuleInstance.Child("another", addrs.NoKey))
+	if reflect.DeepEqual(stateCopy, s.State()) {
+		t.Fatal("State() should return a copy")
+	}
+
+	// our current expected state should also marshal identically to the persisted state
+	if !statefile.StatesMarshalEqual(current, s.State()) {
+		t.Fatalf("Persisted state altered unexpectedly.\n\ngot:\n%s\nwant:\n%s", spew.Sdump(s.State()), spew.Sdump(current))
+	}
+}
+
+// TestFullInitialState is a state that should be snapshotted into a
+// full state manager before passing it into TestFull.
+func TestFullInitialState() *states.State {
+	state := states.NewState()
+	childMod := state.EnsureModule(addrs.RootModuleInstance.Child("child", addrs.NoKey))
+	rAddr := addrs.Resource{
+		Mode: addrs.ManagedResourceMode,
+		Type: "null_resource",
+		Name: "foo",
+	}
+	childMod.SetResourceMeta(rAddr, states.EachList, rAddr.DefaultProviderConfig().Absolute(addrs.RootModuleInstance))
+	return state
+}
diff --git a/states/statemgr/transient.go b/states/statemgr/transient.go
new file mode 100644
index 000000000..087fd2785
--- /dev/null
+++ b/states/statemgr/transient.go
@@ -0,0 +1,66 @@
+package statemgr
+
+import "github.com/hashicorp/terraform/states"
+
+// Transient is a union of the Reader and Writer interfaces, for types that
+// deal with transient snapshots.
+//
+// Transient snapshots are ones that are generally retained only locally and
+// to not create any historical version record when updated. Transient
+// snapshots are not expected to outlive a particular Terraform process,
+// and are not shared with any other process.
+//
+// A state manager type that is primarily concerned with persistent storage
+// may embed type Transient and then call State from its PersistState and
+// WriteState from its RefreshState in order to build on any existing
+// Transient implementation, such as the one returned by NewTransientInMemory.
+type Transient interface {
+	Reader
+	Writer
+}
+
+// Reader is the interface for managers that can return transient snapshots
+// of state.
+//
+// Retrieving the snapshot must not fail, so retrieving a snapshot from remote
+// storage (for example) should be dealt with elsewhere, often in an
+// implementation of Refresher. For a type that implements both Reader
+// and Refresher, it is okay for State to return nil if called before
+// a RefreshState call has completed.
+//
+// For a type that implements both Reader and Writer, State must return the
+// result of the most recently completed call to WriteState, and the state
+// manager must accept concurrent calls to both State and WriteState.
+//
+// Each caller of this function must get a distinct copy of the state, and
+// it must also be distinct from any instance cached inside the reader, to
+// ensure that mutations of the returned state will not affect the values
+// returned to other callers.
+type Reader interface {
+	// State returns the latest state.
+	//
+	// Each call to State returns an entirely-distinct copy of the state, with
+	// no storage shared with any other call, so the caller may freely mutate
+	// the returned object via the state APIs.
+	State() *states.State
+}
+
+// Writer is the interface for managers that can create transient snapshots
+// from state.
+//
+// Writer is the opposite of Reader, and so it must update whatever the State
+// method reads from. It does not write the state to any persistent
+// storage, and (for managers that support historical versions) must not
+// be recorded as a persistent new version of state.
+//
+// Implementations that cache the state in memory must take a deep copy of it,
+// since the caller may continue to modify the given state object after
+// WriteState returns.
+type Writer interface {
+	// Write state saves a transient snapshot of the given state.
+	//
+	// The caller must ensure that the given state object is not concurrently
+	// modified while a WriteState call is in progress. WriteState itself
+	// will never modify the given state.
+	WriteState(*states.State) error
+}
diff --git a/states/statemgr/transient_inmem.go b/states/statemgr/transient_inmem.go
new file mode 100644
index 000000000..07fd3726f
--- /dev/null
+++ b/states/statemgr/transient_inmem.go
@@ -0,0 +1,41 @@
+package statemgr
+
+import (
+	"sync"
+
+	"github.com/hashicorp/terraform/states"
+)
+
+// NewTransientInMemory returns a Transient implementation that retains
+// transient snapshots only in memory, as part of the object.
+//
+// The given initial state, if any, must not be modified concurrently while
+// this function is running, but may be freely modified once this function
+// returns without affecting the stored transient snapshot.
+func NewTransientInMemory(initial *states.State) Transient {
+	return &transientInMemory{
+		current: initial.DeepCopy(),
+	}
+}
+
+type transientInMemory struct {
+	lock    sync.RWMutex
+	current *states.State
+}
+
+var _ Transient = (*transientInMemory)(nil)
+
+func (m *transientInMemory) State() *states.State {
+	m.lock.RLock()
+	defer m.lock.RUnlock()
+
+	return m.current.DeepCopy()
+}
+
+func (m *transientInMemory) WriteState(new *states.State) error {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	m.current = new.DeepCopy()
+	return nil
+}