diff --git a/dns_sync/cli.py b/dns_sync/cli.py
index 96395a9..170c2b6 100644
--- a/dns_sync/cli.py
+++ b/dns_sync/cli.py
@@ -17,6 +17,20 @@ from .registry import load_adapter, load_provider, load_source
 from .reporting import print_endpoint_plans, print_plan
 
 
+def _nonnegative_int(value: str) -> int:
+    try:
+        parsed = int(value)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError(
+            "must be a non-negative integer"
+        ) from exc
+    if parsed < 0:
+        raise argparse.ArgumentTypeError(
+            "must be a non-negative integer"
+        )
+    return parsed
+
+
 def build_parser():
     parser = argparse.ArgumentParser(prog="dns-sync", description="Synchronize reverse-proxy hostnames to a pluggable DNS provider")
     sub = parser.add_subparsers(dest="command", required=True)
@@ -27,7 +41,7 @@ def build_parser():
 
     reconcile = sub.add_parser(
         "reconcile",
-        help="Run the persistent generic engine without applying changes",
+        help="Run the persistent generic reconciliation engine",
     )
     reconcile.add_argument(
         "--state-db",
@@ -47,6 +61,12 @@ def build_parser():
         action="store_true",
         help="Apply and verify planned changes; default is read-only",
     )
+    reconcile.add_argument(
+        "--max-changes",
+        type=_nonnegative_int,
+        default=None,
+        help="Maximum CREATE/UPDATE actions allowed with --apply",
+    )
 
     sub.add_parser("plugins", help="Show the configured source/provider plugins")
     return parser
@@ -175,6 +195,7 @@ def run_reconcile(args):
         for plan in result.plans
         for action in plan.actions
     )
+    change_count = create_count + update_count
     noop_count = sum(
         action.operation.value == "noop"
         for plan in result.plans
@@ -198,6 +219,18 @@ def run_reconcile(args):
             "Refusing to apply while discovery failures are present"
         )
 
+    if (
+        args.max_changes is not None
+        and change_count > args.max_changes
+    ):
+        raise RuntimeError(
+            f"Refusing to apply {change_count} change(s); "
+            f"--max-changes allows {args.max_changes}"
+        )
+
+    if args.max_changes is not None:
+        print(f"Apply mutation budget: {args.max_changes}")
+
     with SQLiteStateStore(args.state_db) as store:
         execution = execute_endpoint_plans(
             adapters,
diff --git a/tests/test_cli_reconcile.py b/tests/test_cli_reconcile.py
index fb07d00..864dd46 100644
--- a/tests/test_cli_reconcile.py
+++ b/tests/test_cli_reconcile.py
@@ -64,11 +64,24 @@ class ReconcileCliTests(unittest.TestCase):
         )
         self.assertTrue(args.include_disabled)
         self.assertFalse(args.apply)
+        self.assertIsNone(args.max_changes)
 
         apply_args = build_parser().parse_args(
-            ["reconcile", "--apply"]
+            [
+                "reconcile",
+                "--apply",
+                "--max-changes",
+                "0",
+            ]
         )
         self.assertTrue(apply_args.apply)
+        self.assertEqual(apply_args.max_changes, 0)
+
+        with redirect_stderr(io.StringIO()):
+            with self.assertRaises(SystemExit):
+                build_parser().parse_args(
+                    ["reconcile", "--max-changes", "-1"]
+                )
 
     def test_reconcile_builds_plan_without_applying_it(self):
         npm_endpoint = Endpoint("npm", "192.168.1.3:81")
@@ -162,6 +175,7 @@ class ReconcileCliTests(unittest.TestCase):
                         state_db=str(state_path),
                         include_disabled=False,
                         apply=False,
+                        max_changes=None,
                     )
                 )
 
@@ -230,6 +244,7 @@ class ReconcileCliTests(unittest.TestCase):
                         state_db=str(state_path),
                         include_disabled=False,
                         apply=True,
+                        max_changes=None,
                     )
                 )
 
@@ -244,6 +259,60 @@ class ReconcileCliTests(unittest.TestCase):
             stdout.getvalue(),
         )
 
+    def test_apply_budget_blocks_executor_before_mutation(self):
+        config = SimpleNamespace(
+            target_ip="192.168.1.101",
+            include_disabled=False,
+            domain_matching=False,
+            domain=None,
+        )
+        adapter = object()
+        action = SimpleNamespace(
+            operation=SimpleNamespace(value="create"),
+        )
+        plan = SimpleNamespace(actions=(action,))
+        reconciliation = SimpleNamespace(
+            observations=(),
+            desired=(),
+            plans=(plan,),
+            failures=(),
+        )
+
+        with tempfile.TemporaryDirectory() as directory:
+            state_path = Path(directory) / "state.sqlite3"
+
+            with patch(
+                "dns_sync.cli.AppConfig.from_env",
+                return_value=config,
+            ), patch(
+                "dns_sync.cli._configured_adapters",
+                return_value=[adapter],
+            ), patch(
+                "dns_sync.cli.discover_reconcile_plan",
+                return_value=reconciliation,
+            ), patch(
+                "dns_sync.cli.print_endpoint_plans",
+            ), patch(
+                "dns_sync.cli.execute_endpoint_plans",
+            ) as execute, redirect_stdout(io.StringIO()):
+                with self.assertRaisesRegex(
+                    RuntimeError,
+                    (
+                        r"Refusing to apply 1 change\(s\); "
+                        r"--max-changes allows 0"
+                    ),
+                ):
+                    run_reconcile(
+                        SimpleNamespace(
+                            state_db=str(state_path),
+                            include_disabled=False,
+                            apply=True,
+                            max_changes=0,
+                        )
+                    )
+
+        execute.assert_not_called()
+
 
 if __name__ == "__main__":
     unittest.main()