pyln-testing: check plugin notifications against any extant notification schemas.

diff --git a/contrib/pyln-testing/pyln/testing/fixtures.py b/contrib/pyln-testing/pyln/testing/fixtures.py
index ebc34aa9..215c5402 100644
--- a/contrib/pyln-testing/pyln/testing/fixtures.py
+++ b/contrib/pyln-testing/pyln/testing/fixtures.py
@@ -3,6 +3,7 @@ from pyln.testing.db import SqliteDbProvider, PostgresDbProvider
 from pyln.testing.utils import NodeFactory, BitcoinD, ElementsD, env, LightningNode, TEST_DEBUG, TEST_NETWORK
 from pyln.client import Millisatoshi
 from typing import Dict
+from pathlib import Path
 
 import json
 import jsonschema  # type: ignore
@@ -496,6 +497,7 @@ def node_factory(request, directory, test_name, bitcoind, executor, db_provider,
     map_node_error(nf.nodes, checkBroken, "had BROKEN messages")
     map_node_error(nf.nodes, lambda n: not n.allow_warning and n.daemon.is_in_log(r' WARNING:'), "had warning messages")
     map_node_error(nf.nodes, checkReconnect, "had unexpected reconnections")
+    map_node_error(nf.nodes, checkPluginJSON, "had malformed hooks/notifications")
 
     # On any bad gossip complaints, print out every nodes' gossip_store
     if map_node_error(nf.nodes, checkBadGossip, "had bad gossip messages"):
@@ -617,6 +619,57 @@ def checkBroken(node):
     return 0
 
 
+def checkPluginJSON(node):
+    if node.bad_notifications:
+        return 0
+
+    try:
+        notificationfiles = os.listdir('doc/schemas/notification')
+    except FileNotFoundError:
+        notificationfiles = []
+
+    notifications = {}
+    for fname in notificationfiles:
+        if fname.endswith('.json'):
+            base = fname.replace('.json', '')
+            # Request is 0 and Response is 1
+            notifications[base] = _load_schema(os.path.join('doc/schemas/notification', fname))
+
+    # FIXME: add doc/schemas/hook/
+    hooks = {}
+
+    for f in (Path(node.daemon.lightning_dir) / "plugin-io").iterdir():
+        # e.g. hook_in-peer_connected-124567-358
+        io = json.loads(f.read_text())
+        parts = f.name.split('-')
+        if parts[0] == 'hook_in':
+            schema = hooks.get(parts[1])
+            req = io['result']
+            direction = 1
+        elif parts[0] == 'hook_out':
+            schema = hooks.get(parts[1])
+            req = io['params']
+            direction = 0
+        else:
+            assert parts[0] == 'notification_out'
+            schema = notifications.get(parts[1])
+            # The notification is wrapped in an object of its own name.
+            req = io['params'][parts[1]]
+            direction = 1
+
+        # Until v26.09, with channel_state_changed.null_scid, that notification will be non-schema compliant.
+        if f.name.startswith('notification_out-channel_state_changed-') and node.daemon.opts.get('allow-deprecated-apis', True) is True:
+            continue
+
+        if schema is not None:
+            try:
+                schema[direction].validate(req)
+            except jsonschema.exceptions.ValidationError as e:
+                print(f"Failed validating {f}: {e}")
+                return 1
+    return 0
+
+
 def checkBadReestablish(node):
     if node.daemon.is_in_log('Bad reestablish'):
         return 1
diff --git a/contrib/pyln-testing/pyln/testing/utils.py b/contrib/pyln-testing/pyln/testing/utils.py
index caa03513..2317c96c 100644
--- a/contrib/pyln-testing/pyln/testing/utils.py
+++ b/contrib/pyln-testing/pyln/testing/utils.py
@@ -789,11 +789,13 @@ class LightningNode(object):
                  jsonschemas={},
                  valgrind_plugins=True,
                  executable=None,
+                 bad_notifications=False,
                  **kwargs):
         self.bitcoin = bitcoind
         self.executor = executor
         self.may_fail = may_fail
         self.may_reconnect = may_reconnect
+        self.bad_notifications = bad_notifications
         self.broken_log = broken_log
         self.allow_bad_gossip = allow_bad_gossip
         self.allow_warning = allow_warning
@@ -853,6 +855,10 @@ class LightningNode(object):
         if self.cln_version >= "v24.11":
             self.daemon.opts.update({"autoconnect-seeker-peers": 0})
 
+        jsondir = Path(lightning_dir) / "plugin-io"
+        jsondir.mkdir()
+        self.daemon.opts['dev-save-plugin-io'] = jsondir
+
         if options is not None:
             self.daemon.opts.update(options)
         dsn = db.get_dsn()
@@ -1601,6 +1607,7 @@ class NodeFactory(object):
             'broken_log',
             'allow_warning',
             'may_reconnect',
+            'bad_notifications',
             'random_hsm',
             'feerates',
             'wait_for_bitcoind_sync',
diff --git a/tests/test_misc.py b/tests/test_misc.py
index 81e308fd..8a4509d2 100644
--- a/tests/test_misc.py
+++ b/tests/test_misc.py
@@ -2723,6 +2723,9 @@ def test_new_node_is_mainnet(node_factory):
     assert not os.path.isfile(os.path.join(netdir, "lightningd-bitcoin.pid"))
     assert os.path.isfile(os.path.join(basedir, "lightningd-bitcoin.pid"))
 
+    # Teardown expects this to exist...
+    os.mkdir(basedir + "/plugin-io")
+
 
 def test_unicode_rpc(node_factory, executor, bitcoind):
     node = node_factory.get_node()
@@ -3659,7 +3662,8 @@ def test_version_reexec(node_factory, bitcoind):
     # We use a file to tell our openingd wrapper where the real one is
     with open(os.path.join(l1.daemon.lightning_dir, TEST_NETWORK, "openingd-real"), 'w') as f:
         f.write(os.path.abspath('lightningd/lightning_openingd'))
-
+    # Internal restart doesn't work well with --dev-save-plugin-io
+    del l1.daemon.opts['dev-save-plugin-io']
     l1.start()
     # This is a "version" message
     verfile = os.path.join(l1.daemon.lightning_dir, TEST_NETWORK, "openingd-version")
@@ -4534,7 +4538,11 @@ def test_setconfig_changed(node_factory, bitcoind):
 
 @unittest.skipIf(os.getenv('TEST_DB_PROVIDER', 'sqlite3') != 'sqlite3', "deletes database, which is assumed sqlite3")
 def test_recover_command(node_factory, bitcoind):
-    l1, l2 = node_factory.get_nodes(2)
+    l1 = node_factory.get_node(start=False)
+    # Internal restart doesn't work well with --dev-save-plugin-io
+    del l1.daemon.opts['dev-save-plugin-io']
+    l1.start()
+    l2 = node_factory.get_node()
 
     l1oldid = l1.info['id']