diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3a7df78..b8633bf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,8 @@
 ## v1.?.0 - 2023-??-?? -
 
+* Beta support for Manager.enable_checksum and octodns-sync --checksum Allows a
+  safer plan & apply workflow where the apply only moves forward if the apply
+  phase plan exactly matches the previous round's planning.
 * Fix for bug in MetaProcessor _up_to_date check that was failing when there was
   a plan with a single change type with a single value, e.g. CNAME.
 * Support added for config env variable expansion on nested levels, not just
diff --git a/octodns/cmds/sync.py b/octodns/cmds/sync.py
index 6efa4f5..d98862b 100755
--- a/octodns/cmds/sync.py
+++ b/octodns/cmds/sync.py
@@ -19,7 +19,7 @@ def main():
         '--doit',
         action='store_true',
         default=False,
-        help='Whether to take action or just show what would change',
+        help='Whether to take action or just show what would change, ignored when Manager.enable_checksum is used',
     )
     parser.add_argument(
         '--force',
@@ -28,6 +28,11 @@ def main():
         help='Acknowledge that significant changes are being '
         'made and do them',
     )
+    parser.add_argument(
+        '--checksum',
+        default=None,
+        help="Provide the expected checksum, apply will only continue if it matches the plan's computed checksum",
+    )
 
     parser.add_argument(
         'zone',
@@ -60,6 +65,7 @@ def main():
         eligible_targets=args.target,
         dry_run=not args.doit,
         force=args.force,
+        checksum=args.checksum,
     )
 
 
diff --git a/octodns/manager.py b/octodns/manager.py
index 791b3ec..7529a45 100644
--- a/octodns/manager.py
+++ b/octodns/manager.py
@@ -4,9 +4,11 @@
 
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
+from hashlib import sha256
 from importlib import import_module
 from importlib.metadata import PackageNotFoundError
 from importlib.metadata import version as module_version
+from json import dumps
 from logging import getLogger
 from os import environ
 from sys import stdout
@@ -87,7 +89,12 @@ class Manager(object):
         return len(plan.changes[0].record.zone.name) if plan.changes else 0
 
     def __init__(
-        self, config_file, max_workers=None, include_meta=False, auto_arpa=False
+        self,
+        config_file,
+        max_workers=None,
+        include_meta=False,
+        auto_arpa=False,
+        enable_checksum=False,
     ):
         version = self._try_version('octodns', version=__version__)
         self.log.info(
@@ -108,6 +115,9 @@ class Manager(object):
         self.include_meta = self._config_include_meta(
             manager_config, include_meta
         )
+        self.enable_checksum = self._config_enable_checksum(
+            manager_config, enable_checksum
+        )
 
         self.auto_arpa = self._config_auto_arpa(manager_config, auto_arpa)
 
@@ -195,6 +205,15 @@ class Manager(object):
         self.log.info('_config_include_meta: include_meta=%s', include_meta)
         return include_meta
 
+    def _config_enable_checksum(self, manager_config, enable_checksum=False):
+        enable_checksum = enable_checksum or manager_config.get(
+            'enable_checksum', False
+        )
+        self.log.info(
+            '_config_enable_checksum: enable_checksum=%s', enable_checksum
+        )
+        return enable_checksum
+
     def _config_auto_arpa(self, manager_config, auto_arpa=False):
         auto_arpa = auto_arpa or manager_config.get('auto_arpa', False)
         self.log.info('_config_auto_arpa: auto_arpa=%s', auto_arpa)
@@ -561,15 +580,16 @@ class Manager(object):
         dry_run=True,
         force=False,
         plan_output_fh=stdout,
+        checksum=None,
     ):
         self.log.info(
-            'sync: eligible_zones=%s, eligible_targets=%s, dry_run=%s, '
-            'force=%s, plan_output_fh=%s',
+            'sync: eligible_zones=%s, eligible_targets=%s, dry_run=%s, force=%s, plan_output_fh=%s, checksum=%s',
             eligible_zones,
             eligible_targets,
             dry_run,
             force,
             getattr(plan_output_fh, 'name', plan_output_fh.__class__.__name__),
+            checksum,
         )
 
         zones = self.config['zones']
@@ -764,13 +784,26 @@ class Manager(object):
         for output in self.plan_outputs.values():
             output.run(plans=plans, log=self.plan_log, fh=plan_output_fh)
 
+        computed_checksum = None
+        if plans and self.enable_checksum:
+            data = [p[1].data for p in plans]
+            data = dumps(data)
+            csum = sha256()
+            csum.update(data.encode('utf-8'))
+            computed_checksum = csum.hexdigest()
+            self.log.info('sync: checksum=%s', computed_checksum)
+
         if not force:
             self.log.debug('sync:   checking safety')
             for target, plan in plans:
                 plan.raise_if_unsafe()
 
-        if dry_run:
+        if dry_run and not checksum:
             return 0
+        elif computed_checksum and computed_checksum != checksum:
+            raise ManagerException(
+                f'checksum={checksum} does not match computed={computed_checksum}'
+            )
 
         total_changes = 0
         self.log.debug('sync:   applying')
diff --git a/octodns/provider/plan.py b/octodns/provider/plan.py
index e0f235d..49b7156 100644
--- a/octodns/provider/plan.py
+++ b/octodns/provider/plan.py
@@ -78,6 +78,10 @@ class Plan(object):
             existing_n,
         )
 
+    @property
+    def data(self):
+        return {'changes': [c.data for c in self.changes]}
+
     def raise_if_unsafe(self):
         if (
             self.existing
diff --git a/octodns/record/change.py b/octodns/record/change.py
index 59b4810..ec9889c 100644
--- a/octodns/record/change.py
+++ b/octodns/record/change.py
@@ -25,6 +25,10 @@ class Create(Change):
     def __init__(self, new):
         super().__init__(None, new)
 
+    @property
+    def data(self):
+        return {'type': 'create', 'new': self.new.data}
+
     def __repr__(self, leader=''):
         source = self.new.source.id if self.new.source else ''
         return f'Create {self.new} ({source})'
@@ -33,6 +37,14 @@ class Create(Change):
 class Update(Change):
     CLASS_ORDERING = 2
 
+    @property
+    def data(self):
+        return {
+            'type': 'update',
+            'existing': self.existing.data,
+            'new': self.new.data,
+        }
+
     # Leader is just to allow us to work around heven eating leading whitespace
     # in our output. When we call this from the Manager.sync plan summary
     # section we'll pass in a leader, otherwise we'll just let it default and
@@ -51,5 +63,9 @@ class Delete(Change):
     def __init__(self, existing):
         super().__init__(existing, None)
 
+    @property
+    def data(self):
+        return {'type': 'delete', 'existing': self.existing.data}
+
     def __repr__(self, leader=''):
         return f'Delete {self.existing}'
diff --git a/tests/test_octodns_manager.py b/tests/test_octodns_manager.py
index b93c9b8..a369b1b 100644
--- a/tests/test_octodns_manager.py
+++ b/tests/test_octodns_manager.py
@@ -184,6 +184,36 @@ class TestManager(TestCase):
             ).sync(dry_run=False, force=True)
             self.assertEqual(33, tc)
 
+    def test_enable_checksum(self):
+        with TemporaryDirectory() as tmpdir:
+            environ['YAML_TMP_DIR'] = tmpdir.dirname
+            environ['YAML_TMP_DIR2'] = tmpdir.dirname
+            manager = Manager(
+                get_config_filename('simple.yaml'), enable_checksum=True
+            )
+
+            # initial/dry run is fine w/o checksum
+            tc = manager.sync(dry_run=True)
+            self.assertEqual(0, tc)
+
+            # trying to apply it fails w/o required checksum
+            with self.assertRaises(ManagerException) as ctx:
+                manager.sync(dry_run=False)
+            msg, checksum = str(ctx.exception).rsplit('=', 1)
+            self.assertEqual('checksum=None does not match computed', msg)
+            self.assertTrue(checksum)
+
+            # wrong checksum fails
+            with self.assertRaises(ManagerException) as ctx:
+                manager.sync(checksum='xyz')
+            msg, checksum = str(ctx.exception).rsplit('=', 1)
+            self.assertEqual('checksum=xyz does not match computed', msg)
+            self.assertTrue(checksum)
+
+            # correct checksum applies (w/o dry_run=False)
+            tc = manager.sync(checksum=checksum)
+            self.assertEqual(28, tc)
+
     def test_idna_eligible_zones(self):
         # loading w/simple, but we'll be blowing it away and doing some manual
         # stuff
diff --git a/tests/test_octodns_plan.py b/tests/test_octodns_plan.py
index b2a19b0..1942666 100644
--- a/tests/test_octodns_plan.py
+++ b/tests/test_octodns_plan.py
@@ -295,3 +295,32 @@ class TestPlanSafety(TestCase):
         with self.assertRaises(RootNsChange) as ctx:
             plan.raise_if_unsafe()
         self.assertTrue('Root Ns record change', str(ctx.exception))
+
+    def test_data(self):
+        data = plans[0][1].data
+        # plans should have a single key, changes
+        self.assertEqual(('changes',), tuple(data.keys()))
+        # it should be a list
+        self.assertIsInstance(data['changes'], list)
+        # w/4 elements
+        self.assertEqual(4, len(data['changes']))
+
+        # we'll test the change .data's here while we're at it since they don't
+        # have a dedicated test (file)
+        delete_data = data['changes'][0]  # delete
+        self.assertEqual(['existing', 'type'], sorted(delete_data.keys()))
+        self.assertEqual('delete', delete_data['type'])
+        self.assertEqual(delete.existing.data, delete_data['existing'])
+
+        create_data = data['changes'][1]  # create
+        self.assertEqual(['new', 'type'], sorted(create_data.keys()))
+        self.assertEqual('create', create_data['type'])
+        self.assertEqual(create.new.data, create_data['new'])
+
+        update_data = data['changes'][3]  # update
+        self.assertEqual(
+            ['existing', 'new', 'type'], sorted(update_data.keys())
+        )
+        self.assertEqual('update', update_data['type'])
+        self.assertEqual(update.existing.data, update_data['existing'])
+        self.assertEqual(update.new.data, update_data['new'])