diff --git a/detection_rules/rule.py b/detection_rules/rule.py
index b341dff9f..571844ac6 100644
--- a/detection_rules/rule.py
+++ b/detection_rules/rule.py
@@ -682,6 +682,8 @@ class QueryValidator:
 
         required: list[dict[str, Any]] = []
         unique_fields: list[str] = self.unique_fields or []
+        if isinstance(self, ESQLValidator):
+            unique_fields = [f for f in unique_fields if not f.startswith(definitions.ESQL_DYNAMIC_FIELD_PREFIXES)]
 
         for fld in unique_fields:
             field_type = ecs_schema.get(fld, {}).get("type")
diff --git a/detection_rules/rule_validators.py b/detection_rules/rule_validators.py
index 8ebe4017e..9b3f412c8 100644
--- a/detection_rules/rule_validators.py
+++ b/detection_rules/rule_validators.py
@@ -47,7 +47,7 @@ from .integrations import (
 )
 from .rule import EQLRuleData, QueryRuleData, QueryValidator, RuleMeta, TOMLRuleContents, set_eql_config
 from .schemas import get_latest_stack_version, get_stack_schemas, get_stack_versions
-from .schemas.definitions import FROM_SOURCES_REGEX
+from .schemas.definitions import ESQL_DYNAMIC_FIELD_PREFIXES, FROM_SOURCES_REGEX
 
 EQL_ERROR_TYPES = (
     eql.EqlCompileError
@@ -792,7 +792,7 @@ class ESQLValidator(QueryValidator):
         for column in query_columns:
             column_name = column["name"]
             # Skip Dynamic fields
-            if column_name.startswith(("Esql.", "Esql_priv.")):
+            if column_name.startswith(ESQL_DYNAMIC_FIELD_PREFIXES):
                 continue
             # Skip internal fields
             if column_name in ("_id", "_version", "_index"):
diff --git a/detection_rules/schemas/definitions.py b/detection_rules/schemas/definitions.py
index 7d0132793..b7644728e 100644
--- a/detection_rules/schemas/definitions.py
+++ b/detection_rules/schemas/definitions.py
@@ -79,6 +79,7 @@ MINOR_SEMVER = re.compile(r"^\d+\.\d+$")
 FROM_SOURCES_REGEX = re.compile(
     r"^\s*FROM\s+(?P<sources>(?:.+?(?:,\s*)?\n?)+?)\s*(?:\||\bmetadata\b|//|$)", re.IGNORECASE | re.MULTILINE
 )
+ESQL_DYNAMIC_FIELD_PREFIXES = ("Esql.", "Esql_priv.")
 BRANCH_PATTERN = f"{VERSION_PATTERN}|^master$"
 ELASTICSEARCH_EQL_FEATURES = {
     "allow_negation": (Version.parse("8.9.0"), None),
diff --git a/pyproject.toml b/pyproject.toml
index 8923f91a0..6f15cb4ac 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "detection_rules"
-version = "1.6.29"
+version = "1.6.30"
 description = "Detection Rules is the home for rules used by Elastic Security. This repository is used for the development, maintenance, testing, validation, and release of rules for Elastic Security’s Detection Engine."
 readme = "README.md"
 requires-python = ">=3.12"
diff --git a/tests/test_rules_remote.py b/tests/test_rules_remote.py
index 3e4cd2fa4..3c1f6c712 100644
--- a/tests/test_rules_remote.py
+++ b/tests/test_rules_remote.py
@@ -19,7 +19,9 @@ from detection_rules.misc import (
     get_default_config,
     getdefault,
 )
+from detection_rules.rule import ESQLRuleData
 from detection_rules.rule_loader import RuleCollection
+from detection_rules.schemas.definitions import ESQL_DYNAMIC_FIELD_PREFIXES
 from detection_rules.utils import get_path, load_rule_contents
 
 from .base import BaseRuleTest
@@ -244,6 +246,20 @@ class TestRemoteRules(BaseRuleTest):
         """
         _ = RuleCollection().load_dict(production_rule)
 
+    def test_esql_required_fields_omit_engine_columns(self):
+        """ESQL required_fields must not list Esql.* / Esql_priv.* (not index mappings)."""
+        for rule in self.all_rules:
+            data = rule.contents.data
+            if not isinstance(data, ESQLRuleData):
+                continue
+            index = data.get("index") or []
+            for rf in data.get_required_fields(index) or []:
+                name = rf["name"]
+                assert not name.startswith(ESQL_DYNAMIC_FIELD_PREFIXES), (
+                    f"{rule.id} - {rule.name}: required_fields must not include ES|QL engine columns "
+                    f"(not index mappings): {name!r}"
+                )
+
     def test_esql_endpoint_unknown_index(self):
         """Test an ESQL rule's index validation. This is expected to error on an unknown index."""
         file_path = get_path(["tests", "data", "command_control_dummy_production_rule.toml"])