diff --git a/tests/tools/etm-powersync-schema-check-test.py b/tests/tools/etm-powersync-schema-check-test.py
new file mode 100644
index 0000000..7e661b4
--- /dev/null
+++ b/tests/tools/etm-powersync-schema-check-test.py
@@ -0,0 +1,240 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+# Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync
+"""Test de tools/etm-powersync-schema-check et du contrat OpenRPC (lot « contrat OpenRPC »).
+
+    python3 -I tests/tools/etm-powersync-schema-check-test.py
+
+Trois choses :
+1. ce que le MOTEUR émet est conforme : les fichiers de référence du sérialiseur `GetPlan`
+   (tests/auto/loadmodel/data/optimizer/, eux-mêmes épinglés au sérialiseur par la suite
+   `loadmodel`) et les traces RÉELLES du moteur (tests/tools/data/) ;
+2. l'OPTIMISEUR DE RÉFÉRENCE, mode par mode, sur une vraie session socket capturée : les modes
+   « bons » sont valides, chaque mode « mauvais » est rejeté par le schéma OU par une règle
+   sémantique nommée — et laquelle (`TABLE_MODES`, recopiée au rapport) ;
+3. le validateur lui-même ne peut pas se taire : un mot-clé non géré, une règle orpheline, un
+   champ ajouté, un mauvais type le font échouer.
+"""
+
+import copy
+import glob
+import json
+import os
+import select
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+
+ICI = os.path.dirname(os.path.abspath(__file__))
+RACINE = os.path.join(ICI, '..', '..')
+OUTIL = os.path.join(RACINE, 'tools', 'etm-powersync-schema-check')
+REF = os.path.join(RACINE, 'tools', 'etm-powersync-ref-optimizer', 'etm-powersync-ref-optimizer')
+SCHEMA = os.path.join(RACINE, 'docs', 'etm-powersync-optimizer-protocol.openrpc.json')
+GOLDEN = sorted(glob.glob(os.path.join(RACINE, 'tests', 'auto', 'loadmodel', 'data', 'optimizer', '*.json')))
+TRACES = sorted(glob.glob(os.path.join(ICI, 'data', '*.ndjson')))
+REQUETE = os.path.join(RACINE, 'tests', 'auto', 'loadmodel', 'data', 'optimizer', 'site-pv-ecs-borne-sgready.json')
+TOKEN = 'jeton-de-test-0123456789'
+
+# mode → (qui rejette, verdict sémantique attendu). « schéma:<mot-clé> » : écart de forme ;
+# « règle » : la forme passe, une règle x-etm-semantic-rules refuse ; « — » : rien ne rejette.
+# Le verdict est le code que le MOTEUR publie (fallbackCode / lastErrorCode, suite refoptimizer).
+TABLE_MODES = {
+    'normal':         ('—', 'valide'),
+    'abstain':        ('—', 'abstain'),
+    'slow':           ('—', 'valide'),
+    'protocol-1.0':   ('—', 'valide'),
+    'deviate':        ('—', 'valide'),
+    'echo-token':     ('—', 'valide'),
+    'wrong-type':     ('schéma:type', 'valide'),
+    'invalid':        ('schéma:type', 'malformed-plan'),
+    'expired':        ('règle', 'expired'),
+    'wrong-major':    ('règle', 'major-mismatch'),
+    'drop':           ('transport', None),
+    'unknown-load':   ('règle', 'unknown-load'),
+    'forbidden-kind': ('règle', 'kind-mismatch'),
+    'missing-load':   ('règle', 'incomplete-slot'),
+}
+
+
+def verifier(*args):
+    p = subprocess.run([sys.executable, '-I', OUTIL, *args], capture_output=True, text=True, timeout=60)
+    resume = None
+    for ligne in p.stdout.splitlines():
+        if ligne.startswith('RÉSUMÉ '):
+            resume = json.loads(ligne[len('RÉSUMÉ '):])
+    return p.returncode, resume, p.stdout + p.stderr
+
+
+def lire_ligne(sock, tampon, delai=5.0):
+    fin = time.monotonic() + delai
+    while b'\n' not in tampon[0]:
+        reste = fin - time.monotonic()
+        if reste <= 0 or not select.select([sock], [], [], reste)[0]:
+            return None
+        bloc = sock.recv(65536)
+        if not bloc:
+            return None
+        tampon[0] += bloc
+    ligne, tampon[0] = tampon[0].split(b'\n', 1)
+    return json.loads(ligne)
+
+
+def session(mode, dossier):
+    """Une vraie session socket avec l'optimiseur de référence ; rend le fichier capturé."""
+    sock_path = os.path.join(dossier, 'opt.sock')
+    capture = os.path.join(dossier, f'{mode}.ndjson')
+    jeton = os.path.join(dossier, 'jeton')
+    with open(jeton, 'w') as f:
+        f.write(TOKEN)
+    proc = subprocess.Popen([sys.executable, '-I', '-B', REF, '--listen', f'unix://{sock_path}', '--mode', mode,
+                             '--capture', capture, '--slow-s', '0.2', '--token-file', jeton],
+                            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
+    try:
+        assert proc.stdout.readline().startswith('ÉCOUTE'), 'optimiseur de référence muet'
+        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        s.connect(sock_path)
+        tampon = [b'']
+        with open(REQUETE, encoding='utf-8') as f:
+            params = json.load(f)
+        s.sendall(json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'Handshake',
+                              'params': {'protocolVersion': '1.1', 'engine': 'etm-powersync-energy-plugin-etm/test',
+                                         'token': TOKEN}}).encode() + b'\n')
+        hs = lire_ligne(s, tampon)
+        if hs and str(hs.get('result', {}).get('protocolVersion', '')).startswith('1'):
+            s.sendall(json.dumps({'jsonrpc': '2.0', 'id': 2, 'method': 'GetPlan', 'params': params}).encode() + b'\n')
+            lire_ligne(s, tampon)
+        s.close()
+    finally:
+        proc.terminate()
+        proc.wait(5)
+        proc.stdout.close()
+    return capture
+
+
+class TestCeQueLeMoteurEmet(unittest.TestCase):
+    def test_les_references_du_serialiseur_sont_conformes(self):
+        self.assertGreaterEqual(len(GOLDEN), 4, 'fichiers de référence du sérialiseur introuvables')
+        for f in GOLDEN:
+            code, r, sortie = verifier('--method', 'GetPlan', '--params', f)
+            self.assertEqual(code, 0, f'{os.path.basename(f)} :\n{sortie}')
+            self.assertEqual(r['comptes'], {'GetPlan.params': 1})
+
+    def test_les_traces_reelles_du_moteur_sont_conformes(self):
+        self.assertTrue(TRACES, 'aucune trace réelle sous tests/tools/data')
+        requetes = reponses = 0
+        for f in TRACES:
+            code, r, sortie = verifier('--trace', f, '--semantic')
+            self.assertEqual(code, 0, f'{os.path.basename(f)} :\n{sortie}')
+            requetes += r['comptes'].get('GetPlan.params', 0)
+            reponses += r['comptes'].get('GetPlan.result', 0)
+        # Le contrôle ne doit pas pouvoir se vider : des requêtes ET des réponses d'optimiseur jugées.
+        self.assertGreater(requetes, 0)
+        self.assertGreater(reponses, 0)
+
+
+class TestOptimiseurDeReference(unittest.TestCase):
+    def test_chaque_mode_et_qui_le_rejette(self):
+        with tempfile.TemporaryDirectory() as d:
+            obtenu = {}
+            for mode in TABLE_MODES:
+                capture = session(mode, d)
+                code, r, sortie = verifier('--ndjson', capture, '--semantic')
+                with open(capture, encoding='utf-8') as f:
+                    self.assertNotIn(TOKEN, f.read(), 'le token est écrit en clair dans la capture')
+                motifs = sorted({l.rsplit('[', 1)[1].rstrip(']') for l in sortie.splitlines()
+                                 if l.startswith('ligne ') and l.endswith(']')})
+                verdicts = {k: v for k, v in r['verdicts'].items() if k != 'valide'} if r else {}
+                if not r['comptes'].get('GetPlan.result') and not verdicts and mode != 'wrong-major':
+                    qui, verdict = 'transport', None
+                else:
+                    verdict = next(iter(verdicts), 'valide')
+                    # `expired` ne compte pas comme un échec (§7), mais le plan n'est pas appliqué :
+                    # c'est bien une règle qui l'écarte. `abstain` est une réponse permise.
+                    qui = ('schéma:' + ','.join(motifs)) if motifs else \
+                        ('règle' if verdict not in ('valide', 'abstain') else '—')
+                obtenu[mode] = (qui, verdict)
+                print(f'  {mode:15s} → {qui:12s} {verdict}', file=sys.stderr)
+            self.assertEqual(obtenu, TABLE_MODES)
+
+
+class TestLeValidateurNeSeTaitPas(unittest.TestCase):
+    def setUp(self):
+        self.d = tempfile.TemporaryDirectory()
+        with open(SCHEMA, encoding='utf-8') as f:
+            self.schema = json.load(f)
+        with open(REQUETE, encoding='utf-8') as f:
+            self.requete = json.load(f)
+
+    def tearDown(self):
+        self.d.cleanup()
+
+    def ecrire(self, nom, contenu):
+        chemin = os.path.join(self.d.name, nom)
+        with open(chemin, 'w', encoding='utf-8') as f:
+            json.dump(contenu, f, ensure_ascii=False)
+        return chemin
+
+    def test_un_mot_cle_non_gere_est_une_erreur_de_schema(self):
+        s = copy.deepcopy(self.schema)
+        s['components']['schemas']['Meter']['uniqueItems'] = True
+        code, _, sortie = verifier('--schema', self.ecrire('s.json', s), '--method', 'GetPlan',
+                                   '--params', REQUETE)
+        self.assertEqual(code, 2, sortie)
+        self.assertIn('« uniqueItems » non géré', sortie)
+
+    def test_une_regle_semantique_sans_implantation_est_une_erreur(self):
+        s = copy.deepcopy(self.schema)
+        s['x-etm-semantic-rules']['rules'].append(dict(s['x-etm-semantic-rules']['rules'][0], name='nouvelle-regle'))
+        code, _, sortie = verifier('--schema', self.ecrire('s.json', s), '--method', 'GetPlan', '--params', REQUETE)
+        self.assertEqual(code, 2, sortie)
+        self.assertIn('nouvelle-regle', sortie)
+        s = copy.deepcopy(self.schema)
+        s['x-etm-semantic-rules']['rules'] = [r for r in s['x-etm-semantic-rules']['rules'] if r['name'] != 'unknown-load']
+        code, _, sortie = verifier('--schema', self.ecrire('s.json', s), '--method', 'GetPlan', '--params', REQUETE)
+        self.assertEqual(code, 2, sortie)
+
+    def test_un_champ_ajoute_par_le_moteur_est_refuse(self):
+        q = copy.deepcopy(self.requete)
+        q['loads'][2]['telemetry']['champNouveau'] = 1
+        code, r, sortie = verifier('--method', 'GetPlan', '--params', self.ecrire('q.json', q))
+        self.assertEqual(code, 1, sortie)
+        self.assertIn('/loads/2/telemetry/champNouveau', sortie)
+        q = copy.deepcopy(self.requete)
+        q['nouveauParam'] = True
+        code, _, sortie = verifier('--method', 'GetPlan', '--params', self.ecrire('q.json', q))
+        self.assertEqual(code, 1, sortie)
+
+    def test_un_mauvais_type_est_refuse(self):
+        q = copy.deepcopy(self.requete)
+        q['loads'][0]['priority'] = 1.5
+        code, _, sortie = verifier('--method', 'GetPlan', '--params', self.ecrire('q.json', q))
+        self.assertEqual(code, 1, sortie)
+        self.assertIn('[type]', sortie)
+
+    def test_une_cle_de_motif_non_documentee_est_refusee(self):
+        plan = {'slots': [{'from': '2026-10-12T13:42:00+02:00', 'to': '2026-10-12T13:47:00+02:00', 'actions': [
+            {'loadId': lc['id'], 'kind': 'state' if lc['adapter'] == 'sg-ready' else 'setpoint',
+             **({'state': 2} if lc['adapter'] == 'sg-ready' else {'powerW': 0}),
+             'reason': 'test', 'reasonCode': 'SURPLUS_SETPOINT', 'reasonParams': {'budgetW': 1}}
+            for lc in self.requete['loads']]}]}
+        code, r, sortie = verifier('--method', 'GetPlan', '--result', self.ecrire('p.json', plan),
+                                   '--request', REQUETE, '--semantic')
+        self.assertEqual(code, 0, sortie)
+        self.assertEqual(r['verdicts'], {'valide': 1})
+        plan['slots'][0]['actions'][0]['reasonParams']['texteLibre'] = 'x'
+        code, _, sortie = verifier('--method', 'GetPlan', '--result', self.ecrire('p.json', plan))
+        self.assertEqual(code, 1, sortie)
+        self.assertIn('[x-etm-reason-params]', sortie)
+
+    def test_le_sous_ensemble_se_dit(self):
+        code, _, sortie = verifier('--subset')
+        self.assertEqual(code, 0)
+        self.assertIn('oneOf', sortie)
+        self.assertNotIn('uniqueItems', sortie)
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)
diff --git a/tools/etm-powersync-schema-check b/tools/etm-powersync-schema-check
new file mode 100755
index 0000000..4869bad
--- /dev/null
+++ b/tools/etm-powersync-schema-check
@@ -0,0 +1,672 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+# Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync
+"""Valide des messages du protocole moteur ↔ optimiseur contre son contrat OpenRPC.
+
+    etm-powersync-schema-check [--schema FICHIER] --method M --params FICHIER.json
+    etm-powersync-schema-check [--schema FICHIER] --method M --result FICHIER.json
+                               [--request PARAMS.json] [--handshake RESULT.json] [--semantic]
+    etm-powersync-schema-check [--schema FICHIER] --ndjson SESSION.ndjson [--semantic] [--requests-only]
+    etm-powersync-schema-check [--schema FICHIER] --trace TRACE.ndjson [--semantic]
+    etm-powersync-schema-check --subset
+
+Le schéma par défaut est `docs/etm-powersync-optimizer-protocol.openrpc.json` du dépôt. Aucun
+accès réseau ; bibliothèque standard seule.
+
+SOUS-ENSEMBLE DE JSON SCHEMA géré (le schéma n'en utilise pas d'autre) : `$ref` (local, `#/…`),
+`type` (dont `integer` : 3.0 en est un), `enum`, `const`, `properties`, `required`,
+`additionalProperties` (booléen ou schéma), `propertyNames`, `items` (un schéma), `minItems`,
+`maxItems`, `minimum`, `maximum`, `exclusiveMinimum`, `minLength`, `maxLength` (en caractères),
+`pattern` (recherche, non ancrée), `oneOf`, `anyOf`, `allOf`. Annotations lues sans effet :
+`title`, `description`, `examples`, `default`, `deprecated`, `$comment`, et toute clé `x-…`.
+**Tout autre mot-clé est une ERREUR de schéma (code 2), jamais ignoré en silence** : un mot-clé
+qu'on croirait appliqué et qui ne l'est pas ferait passer pour valide ce qu'il devait refuser.
+
+Ce que l'outil lit en plus du JSON Schema :
+- les `params` d'une méthode `by-name` forment un objet FERMÉ aux noms déclarés (le moteur les
+  émet : un nom de plus est un champ ajouté sans le contrat), sauf `x-etm-params-open: true` ;
+- `x-etm-reason-params` : clés de `reasonParams` admises par `reasonCode` (contrôle de forme) ;
+- `x-etm-semantic-rules` (avec `--semantic`) : ce que le validateur du moteur refuse EN PLUS de
+  la forme, dans le même ordre, sous le même nom. Une règle du schéma sans implantation ici, ou
+  l'inverse, est une erreur de schéma (code 2) : le contrôle ne disparaît pas en silence.
+
+Modes :
+- `--ndjson` : une session, un message JSON-RPC par ligne, dans les deux sens. Une réponse est
+  rapprochée de sa requête par `id` ; un `GetPlan` est jugé avec SA requête et le dernier
+  `Handshake` accepté. `--requests-only` ne juge que ce qu'émet le moteur (requêtes).
+- `--trace` : traces de décision du moteur (annexe D) — `request` contre les params de
+  `GetPlan` ; `response.result` contre son résultat quand la ligne vient d'un optimiseur
+  (`latencyMs` présent) — le plan du rule-based tracé n'est pas une réponse d'optimiseur.
+
+Sortie : une ligne par écart, puis `RÉSUMÉ {json}` (comptes par méthode et partie, écarts,
+verdicts sémantiques). Code : 0 conforme ; 1 au moins un écart de forme, ou (avec `--semantic`) un
+verdict qui REFUSE (`invalid`, `connection-refused`) ; 2 erreur d'usage ou de schéma.
+Les verdicts `abstention` et `expired` sont dits, sans échec : ce sont des réponses permises.
+"""
+
+import argparse
+import json
+import math
+import os
+import re
+import sys
+import unicodedata
+from datetime import datetime
+
+ICI = os.path.dirname(os.path.abspath(__file__))
+SCHEMA_DEFAUT = os.path.join(ICI, '..', 'docs', 'etm-powersync-optimizer-protocol.openrpc.json')
+
+GERES = {'$ref', 'type', 'enum', 'const', 'properties', 'required', 'additionalProperties',
+         'propertyNames', 'items', 'minItems', 'maxItems', 'minimum', 'maximum',
+         'exclusiveMinimum', 'minLength', 'maxLength', 'pattern', 'oneOf', 'anyOf', 'allOf'}
+ANNOTATIONS = {'title', 'description', 'examples', 'default', 'deprecated', '$comment'}
+TYPES = {'null', 'boolean', 'object', 'array', 'number', 'integer', 'string'}
+
+
+class ErreurSchema(Exception):
+    """Le schéma lui-même est inutilisable (mot-clé non géré, $ref cassé, règle orpheline)."""
+
+
+# --- Validation JSON Schema (sous-ensemble) -------------------------------------------------------
+
+def est_nombre(v):
+    return isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v)
+
+
+def a_le_type(v, t):
+    if t == 'null':
+        return v is None
+    if t == 'boolean':
+        return isinstance(v, bool)
+    if t == 'object':
+        return isinstance(v, dict)
+    if t == 'array':
+        return isinstance(v, list)
+    if t == 'string':
+        return isinstance(v, str)
+    if t == 'number':
+        return est_nombre(v)
+    if t == 'integer':
+        return est_nombre(v) and float(v).is_integer()
+    return False
+
+
+def egal_json(a, b):
+    """Égalité JSON : 1 == 1.0, mais true n'est pas 1."""
+    if isinstance(a, bool) or isinstance(b, bool):
+        return isinstance(a, bool) and isinstance(b, bool) and a == b
+    if est_nombre(a) and est_nombre(b):
+        return a == b
+    if isinstance(a, list) and isinstance(b, list):
+        return len(a) == len(b) and all(egal_json(x, y) for x, y in zip(a, b))
+    if isinstance(a, dict) and isinstance(b, dict):
+        return a.keys() == b.keys() and all(egal_json(a[k], b[k]) for k in a)
+    return type(a) is type(b) and a == b
+
+
+def court(v):
+    t = json.dumps(v, ensure_ascii=False)
+    return t if len(t) <= 60 else t[:57] + '…'
+
+
+class Validateur:
+    def __init__(self, doc):
+        self.doc = doc
+        self.regex = {}
+        self.auditer()
+
+    # -- audit : chaque mot-clé de chaque sous-schéma est géré, chaque $ref se résout
+    def auditer(self):
+        racines = []
+        for nom, s in (self.doc.get('components', {}).get('schemas', {}) or {}).items():
+            racines.append((f'#/components/schemas/{nom}', s))
+        for m in self.doc.get('methods', []):
+            for p in m.get('params', []):
+                racines.append((f'méthode {m.get("name")}, param {p.get("name")}', p.get('schema')))
+            if 'result' in m:
+                racines.append((f'méthode {m.get("name")}, result', m['result'].get('schema')))
+        for ou, s in racines:
+            self.auditer_schema(s, ou)
+
+    def auditer_schema(self, s, ou):
+        if isinstance(s, bool):
+            return
+        if not isinstance(s, dict):
+            raise ErreurSchema(f'{ou} : un schéma doit être un objet ou un booléen')
+        for cle, val in s.items():
+            if cle in ANNOTATIONS or cle.startswith('x-'):
+                continue
+            if cle not in GERES:
+                raise ErreurSchema(f'{ou} : mot-clé « {cle} » non géré par cet outil — refusé plutôt '
+                                   f'qu\'ignoré (sous-ensemble : --subset)')
+            if cle == '$ref':
+                self.resoudre(val, ou)
+            elif cle == 'type':
+                for t in [val] if isinstance(val, str) else val:
+                    if t not in TYPES:
+                        raise ErreurSchema(f'{ou} : type « {t} » inconnu')
+            elif cle == 'properties':
+                for k, sous in val.items():
+                    self.auditer_schema(sous, f'{ou}/properties/{k}')
+            elif cle in ('items', 'propertyNames') or (cle == 'additionalProperties' and isinstance(val, dict)):
+                self.auditer_schema(val, f'{ou}/{cle}')
+            elif cle in ('oneOf', 'anyOf', 'allOf'):
+                for i, sous in enumerate(val):
+                    self.auditer_schema(sous, f'{ou}/{cle}/{i}')
+            elif cle == 'pattern':
+                self.motif(val)
+
+    def resoudre(self, ref, ou='?'):
+        if not isinstance(ref, str) or not ref.startswith('#/'):
+            raise ErreurSchema(f'{ou} : $ref « {ref} » non local')
+        noeud = self.doc
+        for morceau in ref[2:].split('/'):
+            morceau = morceau.replace('~1', '/').replace('~0', '~')
+            if not isinstance(noeud, dict) or morceau not in noeud:
+                raise ErreurSchema(f'{ou} : $ref « {ref} » ne se résout pas')
+            noeud = noeud[morceau]
+        return noeud
+
+    def motif(self, p):
+        if p not in self.regex:
+            try:
+                self.regex[p] = re.compile(p)
+            except re.error as e:
+                raise ErreurSchema(f'pattern « {p} » illisible : {e}')
+        return self.regex[p]
+
+    def discriminant_rate(self, v, s):
+        """Vrai si une propriété à const/enum de la branche contredit l'instance."""
+        while isinstance(s, dict) and '$ref' in s:
+            s = self.resoudre(s['$ref'])
+        if not isinstance(v, dict) or not isinstance(s, dict):
+            return False
+        for k, p in (s.get('properties') or {}).items():
+            if not isinstance(p, dict) or not ('const' in p or 'enum' in p):
+                continue
+            if k not in v:
+                if k in s.get('required', []):
+                    return True
+            elif self.valider(v[k], p):
+                return True
+        return False
+
+    # -- validation : liste d'écarts (chemin, mot-clé, message)
+    def valider(self, v, s, chemin=''):
+        if s is True or s == {}:
+            return []
+        if s is False:
+            return [(chemin or '/', 'false', 'aucune valeur permise')]
+        e = []
+        if '$ref' in s:
+            e += self.valider(v, self.resoudre(s['$ref']), chemin)
+        if 'type' in s:
+            types = [s['type']] if isinstance(s['type'], str) else s['type']
+            if not any(a_le_type(v, t) for t in types):
+                return e + [(chemin or '/', 'type', f'{court(v)} n\'est pas {"|".join(types)}')]
+        if 'const' in s and not egal_json(v, s['const']):
+            e.append((chemin or '/', 'const', f'{court(v)} ≠ {court(s["const"])}'))
+        if 'enum' in s and not any(egal_json(v, x) for x in s['enum']):
+            e.append((chemin or '/', 'enum', f'{court(v)} hors de {court(s["enum"])}'))
+        if isinstance(v, dict):
+            props = s.get('properties', {})
+            for k in s.get('required', []):
+                if k not in v:
+                    e.append((chemin or '/', 'required', f'« {k} » manquant'))
+            for k, x in v.items():
+                if k in props:
+                    e += self.valider(x, props[k], f'{chemin}/{k}')
+                elif 'additionalProperties' in s:
+                    ap = s['additionalProperties']
+                    if ap is False:
+                        e.append((f'{chemin}/{k}', 'additionalProperties', f'champ « {k} » non déclaré'))
+                    elif isinstance(ap, dict):
+                        e += self.valider(x, ap, f'{chemin}/{k}')
+                if 'propertyNames' in s:
+                    e += [(f'{chemin}/{k}', 'propertyNames', m) for _, _, m in self.valider(k, s['propertyNames'], '')]
+        if isinstance(v, list):
+            if 'minItems' in s and len(v) < s['minItems']:
+                e.append((chemin or '/', 'minItems', f'{len(v)} élément(s) < {s["minItems"]}'))
+            if 'maxItems' in s and len(v) > s['maxItems']:
+                e.append((chemin or '/', 'maxItems', f'{len(v)} élément(s) > {s["maxItems"]}'))
+            if 'items' in s:
+                for i, x in enumerate(v):
+                    e += self.valider(x, s['items'], f'{chemin}/{i}')
+        if est_nombre(v):
+            if 'minimum' in s and v < s['minimum']:
+                e.append((chemin or '/', 'minimum', f'{v} < {s["minimum"]}'))
+            if 'maximum' in s and v > s['maximum']:
+                e.append((chemin or '/', 'maximum', f'{v} > {s["maximum"]}'))
+            if 'exclusiveMinimum' in s and v <= s['exclusiveMinimum']:
+                e.append((chemin or '/', 'exclusiveMinimum', f'{v} ≤ {s["exclusiveMinimum"]}'))
+        if isinstance(v, str):
+            if 'minLength' in s and len(v) < s['minLength']:
+                e.append((chemin or '/', 'minLength', f'{len(v)} caractère(s) < {s["minLength"]}'))
+            if 'maxLength' in s and len(v) > s['maxLength']:
+                e.append((chemin or '/', 'maxLength', f'{len(v)} caractère(s) > {s["maxLength"]}'))
+            if 'pattern' in s and not self.motif(s['pattern']).search(v):
+                e.append((chemin or '/', 'pattern', f'{court(v)} ne suit pas {s["pattern"]}'))
+        for sous in s.get('allOf', []):
+            e += self.valider(v, sous, chemin)
+        if 'anyOf' in s:
+            essais = [self.valider(v, sous, chemin) for sous in s['anyOf']]
+            if all(essais):
+                e += [(c, k, m + ' (aucune branche de anyOf)') for c, k, m in min(essais, key=len)]
+        if 'oneOf' in s:
+            essais = [self.valider(v, sous, chemin) for sous in s['oneOf']]
+            ok = sum(1 for x in essais if not x)
+            if ok == 0:
+                # La branche dont le DISCRIMINANT (propriété à const/enum) correspond : c'est elle
+                # que l'émetteur visait, ses écarts sont ceux qui l'informent.
+                rang = [(self.discriminant_rate(v, sous), len(x)) for sous, x in zip(s['oneOf'], essais)]
+                meilleur = essais[rang.index(min(rang))]
+                e += [(c, k, m + ' (aucune branche de oneOf)') for c, k, m in meilleur]
+            elif ok > 1:
+                e.append((chemin or '/', 'oneOf', f'{ok} branches de oneOf valides, une seule attendue'))
+        return e
+
+
+# --- Le contrat : méthodes, contrôles propres au protocole ----------------------------------------
+
+class Contrat:
+    def __init__(self, doc):
+        self.doc = doc
+        self.v = Validateur(doc)
+        self.methodes = {m['name']: m for m in doc.get('methods', [])}
+        regles = (doc.get('x-etm-semantic-rules') or {}).get('rules', [])
+        noms = {r['name'] for r in regles}
+        manquantes = noms - set(REGLES_IMPLANTEES)
+        orphelines = set(REGLES_IMPLANTEES) - noms
+        if manquantes or orphelines:
+            raise ErreurSchema('x-etm-semantic-rules et cet outil divergent — '
+                               f'sans implantation : {sorted(manquantes)} ; absentes du schéma : {sorted(orphelines)}')
+        self.regles = {r['name']: r for r in regles}
+        self.kinds = self.v.resoudre('#/components/schemas/ActionKind').get('enum', [])
+        self.version = str(self.v.resoudre('#/components/schemas/ProtocolVersion').get('const', ''))
+        self.cles_motif = doc.get('x-etm-reason-params') or {}
+
+    def methode(self, nom):
+        if nom not in self.methodes:
+            raise ErreurSchema(f'méthode « {nom} » absente du schéma')
+        return self.methodes[nom]
+
+    def schema_params(self, nom):
+        m = self.methode(nom)
+        if m.get('paramStructure', 'either') not in ('by-name', 'either'):
+            raise ErreurSchema(f'{nom} : seuls les params par nom sont gérés')
+        return {'type': 'object',
+                'properties': {p['name']: p['schema'] for p in m.get('params', [])},
+                'required': [p['name'] for p in m.get('params', []) if p.get('required')],
+                'additionalProperties': bool(m.get('x-etm-params-open', False))}
+
+    def valider_params(self, nom, params):
+        return self.v.valider(params, self.schema_params(nom))
+
+    def valider_result(self, nom, result):
+        m = self.methode(nom)
+        if 'result' not in m:
+            return [('/', 'result', f'{nom} est une notification : pas de résultat')]
+        e = self.v.valider(result, m['result']['schema'])
+        if nom == 'GetPlan' and isinstance(result, dict) and isinstance(result.get('slots'), list):
+            e += self.controler_motifs(result['slots'])
+        return e
+
+    def controler_motifs(self, slots):
+        """x-etm-reason-params : seules les clés documentées du code."""
+        e = []
+        for i, s in enumerate(slots):
+            if not isinstance(s, dict) or not isinstance(s.get('actions'), list):
+                continue
+            for j, a in enumerate(s['actions']):
+                if not isinstance(a, dict) or not isinstance(a.get('reasonParams'), dict):
+                    continue
+                admises = self.cles_motif.get(a.get('reasonCode'))
+                if admises is None:
+                    continue    # code hors liste : déjà un écart d'enum
+                for k in a['reasonParams']:
+                    if k not in admises:
+                        e.append((f'/slots/{i}/actions/{j}/reasonParams/{k}', 'x-etm-reason-params',
+                                  f'clé non admise pour {a.get("reasonCode")} (admises : {admises})'))
+        return e
+
+
+# --- Règles sémantiques : le validateur du moteur, transcrit dans son ordre -----------------------
+
+REGLES_IMPLANTEES = ['abstain', 'malformed-plan', 'empty-plan', 'bad-time', 'slots-unordered',
+                     'malformed-action', 'unknown-load', 'unknown-kind', 'kind-mismatch',
+                     'kind-not-announced', 'missing-reason', 'bad-value', 'duplicate-load', 'expired',
+                     'no-current-slot', 'incomplete-slot',
+                     'handshake-refused', 'invalid-handshake', 'major-mismatch']
+
+INSTANT = re.compile(r'^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}(:[0-9]{2}(\.[0-9]+)?)?(Z|[+-][0-9]{2}:[0-9]{2})$')
+
+
+def instant(v):
+    """lireInstant() du moteur : chaîne ISO 8601 AVEC décalage, sinon None."""
+    if not isinstance(v, str) or not INSTANT.match(v):
+        return None
+    try:
+        return datetime.fromisoformat(v.replace('Z', '+00:00'))
+    except ValueError:
+        return None
+
+
+def vrai_qvariant(v):
+    """QVariant::toBool() — ce que lit le moteur pour `abstain`."""
+    if isinstance(v, bool):
+        return v
+    if est_nombre(v):
+        return v != 0
+    if isinstance(v, str):
+        return v.strip().lower() not in ('', '0', 'false')
+    return False
+
+
+def assaini(v):
+    """assainirTexte() : contrôles, séparateurs de ligne et formats → blanc, puis resserré."""
+    if not isinstance(v, str):
+        return ''
+    return ''.join(' ' if unicodedata.category(c) in ('Cc', 'Zl', 'Zp', 'Cf') else c for c in v).strip()
+
+
+def kind_du_mecanisme(adapter):
+    if adapter in ('evcharger', 'relay-router', 'etmvariableload'):
+        return 'setpoint'
+    if adapter == 'sg-ready':
+        return 'state'
+    return ''
+
+
+def nombre_moteur(v):
+    """lireNombre() : un nombre JSON fini ; ni booléen, ni chaîne."""
+    return v if est_nombre(v) else None
+
+
+def juger_action(a, charges, kinds, annonces):
+    if not isinstance(a, dict):
+        return 'malformed-action'
+    ident = a.get('loadId')
+    if not isinstance(ident, str) or not ident:
+        return 'malformed-action'
+    lc = charges.get(ident)
+    if lc is None:
+        return 'unknown-load'
+    kind = a.get('kind')
+    if not isinstance(kind, str) or kind not in kinds:
+        return 'unknown-kind'
+    attendu = kind_du_mecanisme(lc.get('adapter'))
+    if kind != attendu:
+        return 'kind-mismatch'
+    if annonces is not None and kind not in annonces:
+        return 'kind-not-announced'
+    if not assaini(a.get('reason')):
+        return 'missing-reason'
+    if attendu == 'state':
+        s = nombre_moteur(a.get('state'))
+        if s is None or not float(s).is_integer() or not 1 <= s <= 4:
+            return 'bad-value'
+    elif lc.get('adapter') == 'evcharger':
+        a_courant, a_puissance = 'currentA' in a, 'powerW' in a
+        c, p = nombre_moteur(a.get('currentA')), nombre_moteur(a.get('powerW'))
+        ph = nombre_moteur(a.get('phases'))
+        ok = ((a_courant or a_puissance)
+              and (not a_courant or (c is not None and c >= 0))
+              and (not a_puissance or (p is not None and p >= 0))
+              and ('phases' not in a or ph in (1, 3)))
+        if not ok:
+            return 'bad-value'
+    else:
+        p = nombre_moteur(a.get('powerW'))
+        if p is None or p < 0:
+            return 'bad-value'
+    return None
+
+
+def juger_plan(result, request, handshake, kinds):
+    """OptimizerPlan::valider() : le premier code qui refuse, ou None si le plan est applicable."""
+    if vrai_qvariant(result.get('abstain')):
+        return 'abstain'
+    slots = result.get('slots')
+    if not isinstance(slots, list):
+        return 'malformed-plan'
+    if not slots:
+        return 'empty-plan'
+    charges = {}
+    for lc in (request or {}).get('loads') or []:
+        if isinstance(lc, dict) and isinstance(lc.get('id'), str):
+            charges[lc['id']] = lc
+    annonces = None
+    caps = (handshake or {}).get('capabilities')
+    if isinstance(caps, dict) and 'kinds' in caps:
+        annonces = [str(k) for k in caps['kinds']] if isinstance(caps['kinds'], list) else []
+    creneaux = []
+    for s in slots:
+        if not isinstance(s, dict):
+            return 'malformed-plan'
+        debut, fin = instant(s.get('from')), instant(s.get('to'))
+        if debut is None or fin is None or debut >= fin:
+            return 'bad-time'
+        if creneaux and debut < creneaux[-1][1]:
+            return 'slots-unordered'
+        actions = s.get('actions')
+        if actions is not None and not isinstance(actions, list):
+            return 'malformed-plan'
+        vues = set()
+        for a in actions or []:
+            code = juger_action(a, charges, kinds, annonces)
+            if code:
+                return code
+            if a['loadId'] in vues:
+                return 'duplicate-load'
+            vues.add(a['loadId'])
+        creneaux.append((debut, fin, vues))
+    maintenant = instant((request or {}).get('timestamp'))
+    if 'validUntil' in result and result['validUntil'] is not None:
+        vu = instant(result['validUntil'])
+        if vu is None:
+            return 'bad-time'
+        if maintenant is not None and vu <= maintenant:
+            return 'expired'
+    courant = None
+    for c in creneaux:
+        if maintenant is not None and c[0] <= maintenant < c[1]:
+            courant = c
+    if courant is None:
+        return 'no-current-slot'
+    for ident, lc in charges.items():
+        if (lc.get('telemetry') or {}).get('available', True) is not False and ident not in courant[2]:
+            return 'incomplete-slot'
+    return None
+
+
+def juger_handshake(message, version_moteur):
+    if 'error' in message:
+        return 'handshake-refused'
+    r = message.get('result')
+    if not isinstance(r, dict) or not isinstance(r.get('protocolVersion'), str) or not r['protocolVersion']:
+        return 'invalid-handshake'
+    if 'capabilities' in r and not isinstance(r['capabilities'], dict):
+        return 'invalid-handshake'
+    if r['protocolVersion'].split('.')[0] != version_moteur.split('.')[0]:
+        return 'major-mismatch'
+    return None
+
+
+# --- Lecture, comptes, sortie ---------------------------------------------------------------------
+
+class Bilan:
+    def __init__(self, contrat, semantique):
+        self.c = contrat
+        self.semantique = semantique
+        self.comptes = {}
+        self.ecarts = []
+        self.verdicts = {}
+        self.refus = 0
+
+    def compter(self, cle):
+        self.comptes[cle] = self.comptes.get(cle, 0) + 1
+
+    def noter(self, ou, partie, ecarts):
+        self.compter(partie)
+        for chemin, mot, msg in ecarts:
+            self.ecarts.append({'ou': ou, 'partie': partie, 'chemin': chemin, 'mot': mot, 'message': msg})
+            print(f'{ou} {partie} {chemin} : {msg} [{mot}]')
+
+    def verdict(self, ou, code):
+        if code is None:
+            code = 'valide'
+        self.verdicts[code] = self.verdicts.get(code, 0) + 1
+        regle = self.c.regles.get(code)
+        if regle and regle['issue'] in ('invalid', 'connection-refused'):
+            self.refus += 1
+            print(f'{ou} règle sémantique « {code} » ({regle["issue"]}, fallbackCode {regle["fallbackCode"]}) : '
+                  f'{regle["description"]}')
+        elif regle:
+            print(f'{ou} verdict « {code} » ({regle["issue"]}) — réponse permise')
+
+    def code(self):
+        return 1 if self.ecarts or (self.semantique and self.refus) else 0
+
+    def resume(self):
+        print('RÉSUMÉ ' + json.dumps({'comptes': self.comptes, 'ecarts': len(self.ecarts),
+                                     'verdicts': self.verdicts, 'refus': self.refus},
+                                    ensure_ascii=False, sort_keys=True))
+
+
+def lire_json(chemin):
+    with open(chemin, encoding='utf-8') as f:
+        return json.load(f)
+
+
+def lignes(chemin):
+    with open(chemin, encoding='utf-8') as f:
+        for n, ligne in enumerate(f, 1):
+            if ligne.strip():
+                try:
+                    yield n, json.loads(ligne)
+                except ValueError as e:
+                    yield n, ErreurLigne(str(e))
+
+
+class ErreurLigne:
+    def __init__(self, msg):
+        self.msg = msg
+
+
+def session(b, chemin, requetes_seules):
+    """Un message JSON-RPC par ligne ; réponses rapprochées par id."""
+    attente = {}        # id → (méthode, params)
+    handshake = None    # dernier result de Handshake accepté
+    for n, m in lignes(chemin):
+        ou = f'ligne {n}'
+        if isinstance(m, ErreurLigne) or not isinstance(m, dict):
+            b.noter(ou, 'ligne', [('/', 'json', 'ligne illisible ou non objet')])
+            continue
+        if m.get('jsonrpc') != '2.0':
+            b.noter(ou, 'enveloppe', [('/jsonrpc', 'const', 'jsonrpc doit valoir "2.0"')])
+        if 'method' in m:
+            nom = m['method']
+            if nom not in b.c.methodes:
+                b.noter(ou, 'enveloppe', [('/method', 'enum', f'méthode « {nom} » absente du contrat')])
+                continue
+            b.noter(ou, f'{nom}.params', b.c.valider_params(nom, m.get('params', {})))
+            if m.get('id') is not None:
+                attente[m['id']] = (nom, m.get('params'))
+            continue
+        if requetes_seules:
+            continue
+        if m.get('id') not in attente:
+            b.noter(ou, 'enveloppe', [('/id', 'id', 'réponse sans requête correspondante')])
+            continue
+        nom, params = attente.pop(m['id'])
+        if nom == 'Handshake':
+            if 'result' in m:
+                b.noter(ou, 'Handshake.result', b.c.valider_result('Handshake', m['result']))
+            code = juger_handshake(m, b.c.version)
+            if b.semantique:
+                b.verdict(ou, code)
+            if code is None:
+                handshake = m['result']
+        elif 'error' in m:
+            b.compter(f'{nom}.error')
+            print(f'{ou} {nom} : réponse error (compte comme un échec de la requête, rpc-error)')
+            if b.semantique:
+                b.refus += 1
+        else:
+            b.noter(ou, f'{nom}.result', b.c.valider_result(nom, m.get('result')))
+            if b.semantique and nom == 'GetPlan' and isinstance(m.get('result'), dict):
+                b.verdict(ou, juger_plan(m['result'], params, handshake, b.c.kinds))
+
+
+def trace(b, chemin):
+    """Traces de décision (annexe D)."""
+    for n, t in lignes(chemin):
+        ou = f'ligne {n}'
+        if isinstance(t, ErreurLigne) or not isinstance(t, dict):
+            b.noter(ou, 'ligne', [('/', 'json', 'ligne illisible ou non objet')])
+            continue
+        if 'request' not in t:
+            b.compter('sans-requête')      # L2 : aucune requête, rien à juger
+            continue
+        b.noter(ou, 'GetPlan.params', b.c.valider_params('GetPlan', t['request']))
+        reponse = t.get('response') or {}
+        if 'latencyMs' in t and isinstance(reponse, dict) and 'result' in reponse:
+            b.noter(ou, 'GetPlan.result', b.c.valider_result('GetPlan', reponse['result']))
+            if b.semantique and isinstance(reponse['result'], dict):
+                b.verdict(ou, juger_plan(reponse['result'], t['request'], None, b.c.kinds))
+
+
+def main(argv=None):
+    p = argparse.ArgumentParser(prog='etm-powersync-schema-check',
+                                description='Valide des messages du protocole optimiseur contre son contrat OpenRPC.')
+    p.add_argument('--schema', default=SCHEMA_DEFAUT, help='document OpenRPC (défaut : celui du dépôt)')
+    p.add_argument('--method', help='méthode du message (--params / --result)')
+    p.add_argument('--params', help='fichier JSON : les params d\'une requête')
+    p.add_argument('--result', help='fichier JSON : le result d\'une réponse')
+    p.add_argument('--request', help='fichier JSON : les params de la requête (règles sémantiques de GetPlan)')
+    p.add_argument('--handshake', help='fichier JSON : le result du Handshake (kinds annoncés)')
+    p.add_argument('--ndjson', help='session JSON-RPC, un message par ligne')
+    p.add_argument('--trace', help='traces de décision du moteur (annexe D)')
+    p.add_argument('--semantic', action='store_true', help='applique x-etm-semantic-rules')
+    p.add_argument('--requests-only', action='store_true', help='--ndjson : ne juge que les requêtes')
+    p.add_argument('--subset', action='store_true', help='écrit le sous-ensemble géré et sort')
+    a = p.parse_args(argv)
+    if a.subset:
+        print('mots-clés : ' + ' '.join(sorted(GERES)))
+        print('annotations : ' + ' '.join(sorted(ANNOTATIONS)) + ' x-…')
+        return 0
+    try:
+        contrat = Contrat(lire_json(a.schema))
+        b = Bilan(contrat, a.semantic)
+        if a.ndjson:
+            session(b, a.ndjson, a.requests_only)
+        elif a.trace:
+            trace(b, a.trace)
+        elif a.method and (a.params or a.result):
+            if a.params:
+                b.noter(a.params, f'{a.method}.params', contrat.valider_params(a.method, lire_json(a.params)))
+            if a.result:
+                r = lire_json(a.result)
+                b.noter(a.result, f'{a.method}.result', contrat.valider_result(a.method, r))
+                if a.semantic and a.method == 'GetPlan' and isinstance(r, dict):
+                    req = lire_json(a.request) if a.request else None
+                    hs = lire_json(a.handshake) if a.handshake else None
+                    if req is None:
+                        raise ErreurSchema('--semantic sur GetPlan exige --request (charges, instant)')
+                    b.verdict(a.result, juger_plan(r, req, hs, contrat.kinds))
+                elif a.semantic and a.method == 'Handshake':
+                    b.verdict(a.result, juger_handshake({'result': r}, contrat.version))
+        else:
+            p.error('rien à valider : --ndjson, --trace, ou --method avec --params / --result')
+    except ErreurSchema as e:
+        print(f'ERREUR DE SCHÉMA : {e}', file=sys.stderr)
+        return 2
+    except (OSError, ValueError) as e:
+        print(f'ERREUR : {e}', file=sys.stderr)
+        return 2
+    b.resume()
+    return b.code()
+
+
+if __name__ == '__main__':
+    sys.exit(main())