[mqtt] Reject deferring actions in on_json_message (#20538)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

esphome/components/mqtt/__init__.py

@@ -224,6 +224,17 @@ def _consume_mqtt_sockets(config: ConfigType) -> ConfigType:
     return config
 
 
+# The parsed JSON document is freed once the automation returns
+JSON_MESSAGE_AUTOMATION_SCHEMA = automation.validate_synchronous_automation(
+    "on_json_message",
+    "the parsed JSON document",
+    {
+        cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(MQTTJsonMessageTrigger),
+        cv.Required(CONF_TOPIC): cv.subscribe_topic,
+        cv.Optional(CONF_QOS, default=0): cv.mqtt_qos,
+    },
+)
+
 CONFIG_SCHEMA = cv.All(
     cv.Schema(
         {
@@ -304,15 +315,7 @@ CONFIG_SCHEMA = cv.All(
                     cv.Optional(CONF_PAYLOAD): cv.string_strict,
                 }
             ),
-            cv.Optional(CONF_ON_JSON_MESSAGE): automation.validate_automation(
-                {
-                    cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
-                        MQTTJsonMessageTrigger
-                    ),
-                    cv.Required(CONF_TOPIC): cv.subscribe_topic,
-                    cv.Optional(CONF_QOS, default=0): cv.mqtt_qos,
-                }
-            ),
+            cv.Optional(CONF_ON_JSON_MESSAGE): JSON_MESSAGE_AUTOMATION_SCHEMA,
             cv.Optional(CONF_PUBLISH_NAN_AS_NONE, default=False): cv.boolean,
             cv.Optional(CONF_WAIT_FOR_CONNECTION, default=False): cv.boolean,
         }

tests/component_tests/mqtt/test_json_message_automation.py

@@ -0,0 +1,52 @@
+"""Tests that mqtt on_json_message automations reject deferring actions.
+
+The trigger passes a view into the parsed JSON document, which is freed once the automation
+returns, so the deferring-actions guard is a safety property.
+"""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.mqtt import CONFIG_SCHEMA, JSON_MESSAGE_AUTOMATION_SCHEMA
+from esphome.const import CONF_ON_JSON_MESSAGE
+from esphome.core import Lambda
+from esphome.types import ConfigType
+
+DEFERRING_ACTION = {"delay": "1s"}
+SYNCHRONOUS_ACTION = {"lambda": Lambda('ESP_LOGD("test", "ran");')}
+TRUE_CONDITION = {"lambda": Lambda("return true;")}
+
+DEFER_MESSAGE = "Deferring actions"
+
+
+def _automation(*actions: ConfigType) -> ConfigType:
+    return {"topic": "test/topic", "then": list(actions)}
+
+
+def test_schema_wired_to_on_json_message() -> None:
+    schema = CONFIG_SCHEMA.validators[0].schema
+    assert schema[cv.Optional(CONF_ON_JSON_MESSAGE)] is JSON_MESSAGE_AUTOMATION_SCHEMA
+
+
+def test_synchronous_automation_accepted() -> None:
+    JSON_MESSAGE_AUTOMATION_SCHEMA([_automation(SYNCHRONOUS_ACTION)])
+
+
+def test_deferring_action_rejected() -> None:
+    with pytest.raises(cv.Invalid, match=DEFER_MESSAGE):
+        JSON_MESSAGE_AUTOMATION_SCHEMA(
+            [_automation(SYNCHRONOUS_ACTION, DEFERRING_ACTION)]
+        )
+
+
+def test_nested_deferring_action_rejected() -> None:
+    action = {"if": {"condition": TRUE_CONDITION, "then": [DEFERRING_ACTION]}}
+    with pytest.raises(cv.Invalid, match=DEFER_MESSAGE):
+        JSON_MESSAGE_AUTOMATION_SCHEMA([_automation(action)])
+
+
+def test_deferring_action_in_second_automation_rejected() -> None:
+    with pytest.raises(cv.Invalid, match=DEFER_MESSAGE):
+        JSON_MESSAGE_AUTOMATION_SCHEMA(
+            [_automation(SYNCHRONOUS_ACTION), _automation(DEFERRING_ACTION)]
+        )