diff --git a/dns_sync/cli.py b/dns_sync/cli.py
index f72c963..96395a9 100644
--- a/dns_sync/cli.py
+++ b/dns_sync/cli.py
@@ -10,6 +10,7 @@ from .engine import (
     summarize_plan,
     validate_desired,
 )
+from .executor import execute_endpoint_plans
 from .orchestrator import discover_reconcile_plan
 from .persistence import SQLiteStateStore
 from .registry import load_adapter, load_provider, load_source
@@ -41,6 +42,11 @@ def build_parser():
         action="store_true",
         help="Include disabled proxy hosts in desired DNS state",
     )
+    reconcile.add_argument(
+        "--apply",
+        action="store_true",
+        help="Apply and verify planned changes; default is read-only",
+    )
 
     sub.add_parser("plugins", help="Show the configured source/provider plugins")
     return parser
@@ -148,7 +154,8 @@ def run_reconcile(args):
             file=sys.stderr,
         )
 
-    print("Generic reconciliation mode: READ ONLY")
+    mode = "APPLY" if args.apply else "READ ONLY"
+    print(f"Generic reconciliation mode: {mode}")
     print(f"State database: {args.state_db}")
     print(f"Configured endpoints: {len(adapters)}")
     print(f"Cached/live observations: {len(result.observations)}")
@@ -181,7 +188,28 @@ def run_reconcile(args):
         f"{update_count} update, "
         f"{noop_count} unchanged"
     )
-    print("READ ONLY: no endpoint changes were applied.")
+
+    if not args.apply:
+        print("READ ONLY: no endpoint changes were applied.")
+        return 0
+
+    if result.failures:
+        raise RuntimeError(
+            "Refusing to apply while discovery failures are present"
+        )
+
+    with SQLiteStateStore(args.state_db) as store:
+        execution = execute_endpoint_plans(
+            adapters,
+            result.plans,
+            store,
+        )
+
+    print(
+        "APPLY COMPLETE: "
+        f"{len(execution.applied)} action(s) applied and verified; "
+        f"{len(execution.skipped)} unchanged/skipped."
+    )
     return 0
 
 
diff --git a/tests/test_cli_reconcile.py b/tests/test_cli_reconcile.py
index 833cbe3..fb07d00 100644
--- a/tests/test_cli_reconcile.py
+++ b/tests/test_cli_reconcile.py
@@ -63,6 +63,12 @@ class ReconcileCliTests(unittest.TestCase):
             "/tmp/test-state.sqlite3",
         )
         self.assertTrue(args.include_disabled)
+        self.assertFalse(args.apply)
+
+        apply_args = build_parser().parse_args(
+            ["reconcile", "--apply"]
+        )
+        self.assertTrue(apply_args.apply)
 
     def test_reconcile_builds_plan_without_applying_it(self):
         npm_endpoint = Endpoint("npm", "192.168.1.3:81")
@@ -155,6 +161,7 @@ class ReconcileCliTests(unittest.TestCase):
                     SimpleNamespace(
                         state_db=str(state_path),
                         include_disabled=False,
+                        apply=False,
                     )
                 )
 
@@ -182,5 +189,61 @@ class ReconcileCliTests(unittest.TestCase):
         self.assertEqual(verizon.verify_count, 0)
 
 
+    def test_apply_flag_invokes_verified_executor(self):
+        config = SimpleNamespace(
+            target_ip="192.168.1.101",
+            include_disabled=False,
+            domain_matching=False,
+            domain=None,
+        )
+        adapter = object()
+        reconciliation = SimpleNamespace(
+            observations=(),
+            desired=(),
+            plans=(),
+            failures=(),
+        )
+        execution = SimpleNamespace(
+            applied=(),
+            skipped=(),
+        )
+
+        with tempfile.TemporaryDirectory() as directory:
+            state_path = Path(directory) / "state.sqlite3"
+            stdout = io.StringIO()
+
+            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.execute_endpoint_plans",
+                return_value=execution,
+            ) as execute, redirect_stdout(stdout):
+                result = run_reconcile(
+                    SimpleNamespace(
+                        state_db=str(state_path),
+                        include_disabled=False,
+                        apply=True,
+                    )
+                )
+
+        self.assertEqual(result, 0)
+        execute.assert_called_once()
+        self.assertIn(
+            "Generic reconciliation mode: APPLY",
+            stdout.getvalue(),
+        )
+        self.assertIn(
+            "APPLY COMPLETE: 0 action(s) applied and verified",
+            stdout.getvalue(),
+        )
+
+
 if __name__ == "__main__":
     unittest.main()