diff --git a/detection_rules/rule.py b/detection_rules/rule.py
index 432ec4cf7..f9d2a48b6 100644
--- a/detection_rules/rule.py
+++ b/detection_rules/rule.py
@@ -988,14 +988,28 @@ class ESQLRuleData(QueryRuleData):
                 f" Add 'metadata _id, _version, _index' to the from command or add an aggregate function."
             )
 
-        # Enforce KEEP command for ESQL rules
+        # Enforce KEEP command for ESQL rules and that METADATA fields are present in non-aggregate queries
         # Match | followed by optional whitespace/newlines and then 'keep'
-        keep_pattern = re.compile(r"\|\s*keep\b", re.IGNORECASE | re.DOTALL)
-        if not keep_pattern.search(query_lower):
+        keep_pattern = re.compile(r"\|\s*keep\b\s+([^\|]+)", re.IGNORECASE | re.DOTALL)
+        keep_match = keep_pattern.search(query_lower)
+        if not keep_match:
             raise EsqlSemanticError(
                 f"Rule: {data['name']} does not contain a 'keep' command -> Add a 'keep' command to the query."
             )
 
+        # Ensure that keep clause includes metadata fields on non-aggregate queries
+        aggregate_pattern = re.compile(r"\|\s*stats\b(?:\s+([^\|]+?))?(?:\s+by\s+([^\|]+))?", re.IGNORECASE | re.DOTALL)
+        if not aggregate_pattern.search(query_lower):
+            keep_fields = [field.strip() for field in keep_match.group(1).split(",")]
+            if "*" not in keep_fields:
+                required_metadata = {"_id", "_version", "_index"}
+                if not required_metadata.issubset(set(map(str.strip, keep_fields))):
+                    raise EsqlSemanticError(
+                        f"Rule: {data['name']} contains a keep clause without"
+                        f" metadata fields '_id', '_version', and '_index' ->"
+                        f" Add '_id', '_version', '_index' to the keep command."
+                    )
+
 
 @dataclass(frozen=True, kw_only=True)
 class ThreatMatchRuleData(QueryRuleData):
diff --git a/pyproject.toml b/pyproject.toml
index 55deeadbe..14ff5d8c2 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "detection_rules"
-version = "1.5.28"
+version = "1.5.29"
 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 507654411..be6b027f6 100644
--- a/tests/test_rules_remote.py
+++ b/tests/test_rules_remote.py
@@ -39,7 +39,7 @@ class TestRemoteRules(BaseRuleTest):
         and aws.cloudtrail.user_identity.arn is not null
         and aws.cloudtrail.user_identity.type == "IAMUser"
         | keep
-        aws.cloudtrail.user_identity.type
+        aws.cloudtrail.user_identity.type, _id, _version, _index
         """
         rule = RuleCollection().load_dict(production_rule)
         related_integrations = rule.contents.to_api_format()["related_integrations"]
@@ -61,7 +61,7 @@ class TestRemoteRules(BaseRuleTest):
         user.id,
         gen_ai.request.model.id,
         cloud.account.id,
-        gen_ai.response.error_code
+        gen_ai.response.error_code, _id, _version, _index
         """
         rule = RuleCollection().load_dict(production_rule)
         related_integrations = rule.contents.to_api_format()["related_integrations"]
@@ -81,7 +81,7 @@ class TestRemoteRules(BaseRuleTest):
         and event.dataset in ("aws.billing")
         and aws.cloudtrail.user_identity.type == "IAMUser"
         | keep
-        aws.cloudtrail.user_identity.type
+        aws.cloudtrail.user_identity.type, _id, _version, _index
         """
         with pytest.raises(EsqlSchemaError):
             _ = RuleCollection().load_dict(production_rule)
@@ -99,7 +99,7 @@ class TestRemoteRules(BaseRuleTest):
         and event.dataset in ("aws.cloudtrail", "aws.billing")
         and aws.cloudtrail.user_identity.type == 5
         | keep
-        aws.cloudtrail.user_identity.type
+        aws.cloudtrail.user_identity.type, _id, _version, _index
         """
         with pytest.raises(EsqlTypeMismatchError):
             _ = RuleCollection().load_dict(production_rule)
@@ -117,7 +117,7 @@ class TestRemoteRules(BaseRuleTest):
         and event.dataset in ("aws.cloudtrail", "aws.billing")
         and aws.cloudtrail.user_identity.type = "IAMUser"
         | keep
-        aws.cloudtrail.user_identity.type
+        aws.cloudtrail.user_identity.type, _id, _version, _index
         """
         with pytest.raises(EsqlSyntaxError):
             _ = RuleCollection().load_dict(production_rule)
@@ -134,7 +134,7 @@ class TestRemoteRules(BaseRuleTest):
         | where @timestamp > now() - 30 minutes
         and aws.cloudtrail.user_identity.type == "IAMUser"
         | keep
-        aws.*
+        aws.*, _id, _version, _index
         """
         _ = RuleCollection().load_dict(production_rule)
 
@@ -150,7 +150,7 @@ class TestRemoteRules(BaseRuleTest):
         | where @timestamp > now() - 30 minutes
         and aws.cloudtrail.user_identity.type == "IAMUser"
         | keep
-        aws.cloudtrail.user_identity.type
+        aws.cloudtrail.user_identity.type, _id, _version, _index
         """
         with pytest.raises(EsqlSchemaError):
             _ = RuleCollection().load_dict(production_rule)
@@ -167,7 +167,7 @@ class TestRemoteRules(BaseRuleTest):
         | where @timestamp > now() - 30 minutes
         and aws.cloudtrail.user_identity.type == "IAMUser"
         | keep
-        aws.*
+        aws.*, _id, _version, _index
         """
         _ = RuleCollection().load_dict(production_rule)
 
@@ -179,7 +179,7 @@ class TestRemoteRules(BaseRuleTest):
         production_rule["rule"]["query"] = """
         from logs-endpoint.alerts-*
         | where event.code in ("malicious_file", "memory_signature", "shellcode_thread") and rule.name is not null
-        | keep host.id, rule.name, event.code
+        | keep host.id, rule.name, event.code, _id, _version, _index
         | stats Esql.host_id_count_distinct = count_distinct(host.id) by rule.name, event.code
         | where Esql.host_id_count_distinct >= 3
         """
@@ -193,7 +193,7 @@ class TestRemoteRules(BaseRuleTest):
         production_rule["rule"]["query"] = """
         from logs-endpoint.fake-*
         | where event.code in ("malicious_file", "memory_signature", "shellcode_thread") and rule.name is not null
-        | keep host.id, rule.name, event.code
+        | keep host.id, rule.name, event.code, _id, _version, _index
         | stats Esql.host_id_count_distinct = count_distinct(host.id) by rule.name, event.code
         | where Esql.host_id_count_distinct >= 3
         """
@@ -209,7 +209,7 @@ class TestRemoteRules(BaseRuleTest):
         production_rule["rule"]["query"] = """
         from logs-endpoint.alerts-*
         | where event.code in ("malicious_file", "memory_signature", "shellcode_thread") and rule.name is not null and file.Ext.entry_modified > 0
-        | keep host.id, rule.name, event.code, file.Ext.entry_modified
+        | keep host.id, rule.name, event.code, file.Ext.entry_modified, _id, _version, _index
         | stats Esql.host_id_count_distinct = count_distinct(host.id) by rule.name, event.code, file.Ext.entry_modified
         | where Esql.host_id_count_distinct >= 3
         """
@@ -228,7 +228,7 @@ class TestRemoteRules(BaseRuleTest):
         production_rule["rule"]["query"] = """
         from logs-aws.billing* metadata _id, _version, _index
         | where @timestamp > now() - 30 minutes and aws.cloudtrail.user_identity.type == "IAMUser"
-        | keep host.id, rule.name, event.code
+        | keep host.id, rule.name, event.code, _id, _version, _index
         | stats Esql.host_id_count_distinct = count_distinct(host.id) by rule.name, event.code
         | where Esql.host_id_count_distinct >= 3
         """
@@ -248,6 +248,6 @@ class TestRemoteRules(BaseRuleTest):
         and event.outcome == "success"
         and azure.signinlogs.properties.user_id is not null
         | keep
-        event.outcome
+        event.outcome, _id, _version, _index
         """
         _ = RuleCollection().load_dict(production_rule)
diff --git a/tests/test_schemas.py b/tests/test_schemas.py
index 1215e4b0f..5f5e609eb 100644
--- a/tests/test_schemas.py
+++ b/tests/test_schemas.py
@@ -324,7 +324,7 @@ class TestESQLValidation(unittest.TestCase):
         query = """
             FROM logs-windows.powershell_operational* METADATA _id, _version, _index
             | WHERE event.code == "4104"
-            | KEEP event.code
+            | KEEP event.code, _id, _version, _index
         """
         rule_dict["rule"]["query"] = query
         _ = RuleCollection().load_dict(rule_dict, path=rule_path)
@@ -334,7 +334,7 @@ class TestESQLValidation(unittest.TestCase):
         query = """
             FROM logs-windows.powershell_operational* METADATA _id, _index, _version
             | WHERE event.code == "4104"
-            | KEEP event.code
+            | KEEP event.code, _id, _version, _index
         """
         rule_dict["rule"]["query"] = query
         _ = RuleCollection().load_dict(rule_dict, path=rule_path)
@@ -344,7 +344,7 @@ class TestESQLValidation(unittest.TestCase):
             query = """
                 FROM logs-windows.powershell_operational* METADATA _foo, _index
                 | WHERE event.code == "4104"
-                | KEEP event.code
+                | KEEP event.code, _id, _version, _index
             """
             rule_dict["rule"]["query"] = query
             _ = RuleCollection().load_dict(rule_dict, path=rule_path)