diff --git a/CHANGELOG.md b/CHANGELOG.md
index e8f5c3c..3e959ce 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,7 +1,9 @@
 ## v1.?.? - 2024-??-?? - ???
 
 * Improved handling of present, but empty/None config file values.
-
+* Add PlanJson plan_output support
+* Include `record_type` in Change data
+ 
 ## v1.8.0 - 2024-06-10 - Set the records straight
 
 * Add support for SVCB and HTTPS records
diff --git a/octodns/provider/plan.py b/octodns/provider/plan.py
index 825204c..18939c8 100644
--- a/octodns/provider/plan.py
+++ b/octodns/provider/plan.py
@@ -2,7 +2,9 @@
 #
 #
 
+from collections import defaultdict
 from io import StringIO
+from json import dumps
 from logging import DEBUG, ERROR, INFO, WARN, getLogger
 from sys import stdout
 
@@ -220,6 +222,21 @@ def _value_stringifier(record, sep):
     return sep.join(values)
 
 
+class PlanJson(_PlanOutput):
+    def __init__(self, name, indent=None, sort_keys=True):
+        super().__init__(name)
+        self.indent = indent
+        self.sort_keys = sort_keys
+
+    def run(self, plans, fh=stdout, *args, **kwargs):
+        data = defaultdict(dict)
+        for target, plan in plans:
+            data[target.id][plan.desired.name] = plan.data
+
+        fh.write(dumps(data, indent=self.indent, sort_keys=self.sort_keys))
+        fh.write('\n')
+
+
 class PlanMarkdown(_PlanOutput):
     def run(self, plans, fh=stdout, *args, **kwargs):
         if plans:
diff --git a/octodns/record/change.py b/octodns/record/change.py
index ec9889c..83393e9 100644
--- a/octodns/record/change.py
+++ b/octodns/record/change.py
@@ -27,7 +27,11 @@ class Create(Change):
 
     @property
     def data(self):
-        return {'type': 'create', 'new': self.new.data}
+        return {
+            'type': 'create',
+            'new': self.new.data,
+            'record_type': self.new._type,
+        }
 
     def __repr__(self, leader=''):
         source = self.new.source.id if self.new.source else ''
@@ -43,6 +47,7 @@ class Update(Change):
             'type': 'update',
             'existing': self.existing.data,
             'new': self.new.data,
+            'record_type': self.new._type,
         }
 
     # Leader is just to allow us to work around heven eating leading whitespace
@@ -65,7 +70,11 @@ class Delete(Change):
 
     @property
     def data(self):
-        return {'type': 'delete', 'existing': self.existing.data}
+        return {
+            'type': 'delete',
+            'existing': self.existing.data,
+            'record_type': self.existing._type,
+        }
 
     def __repr__(self, leader=''):
         return f'Delete {self.existing}'
diff --git a/tests/test_octodns_plan.py b/tests/test_octodns_plan.py
index 2fda278..943d4af 100644
--- a/tests/test_octodns_plan.py
+++ b/tests/test_octodns_plan.py
@@ -3,6 +3,7 @@
 #
 
 from io import StringIO
+from json import loads
 from logging import getLogger
 from unittest import TestCase
 
@@ -11,6 +12,7 @@ from helpers import SimpleProvider
 from octodns.provider.plan import (
     Plan,
     PlanHtml,
+    PlanJson,
     PlanLogger,
     PlanMarkdown,
     RootNsChange,
@@ -126,6 +128,17 @@ class TestPlanHtml(TestCase):
         )
 
 
+class TestPlanJson(TestCase):
+    def test_basics(self):
+        out = StringIO()
+        PlanJson('json').run(plans, fh=out)
+        data = loads(out.getvalue())
+        for key in ('test', 'unit.tests.', 'changes'):
+            self.assertTrue(key in data)
+            data = data[key]
+        self.assertEqual(4, len(data))
+
+
 class TestPlanMarkdown(TestCase):
     log = getLogger('TestPlanMarkdown')
 
@@ -394,18 +407,25 @@ class TestPlanSafety(TestCase):
         # 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(
+            ['existing', 'record_type', 'type'], sorted(delete_data.keys())
+        )
         self.assertEqual('delete', delete_data['type'])
+        self.assertEqual('A', delete_data['record_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(
+            ['new', 'record_type', 'type'], sorted(create_data.keys())
+        )
         self.assertEqual('create', create_data['type'])
+        self.assertEqual('CNAME', create_data['record_type'])
         self.assertEqual(create.new.data, create_data['new'])
 
         update_data = data['changes'][3]  # update
         self.assertEqual(
-            ['existing', 'new', 'type'], sorted(update_data.keys())
+            ['existing', 'new', 'record_type', 'type'],
+            sorted(update_data.keys()),
         )
         self.assertEqual('update', update_data['type'])
         self.assertEqual(update.existing.data, update_data['existing'])