diff --git a/aws_lambda_powertools/utilities/batch/base.py b/aws_lambda_powertools/utilities/batch/base.py
index 149f94f7ff8..d5d6c64381b 100644
--- a/aws_lambda_powertools/utilities/batch/base.py
+++ b/aws_lambda_powertools/utilities/batch/base.py
@@ -575,7 +575,8 @@ def _process_record(self, record: dict) -> SuccessResponse | FailureResponse:
             model = getattr(exc, "model", None) or getattr(exc, "title", None)
             model_name = getattr(self.model, "__name__", None)
 
-            if model in (self.model, model_name):
+            # `data is None` means converting the record failed, e.g. a validator raised an error Pydantic doesn't wrap
+            if model in (self.model, model_name) or data is None:
                 return self._register_model_validation_error_record(record)
 
             return self.failure_handler(record=data, exception=sys.exc_info())
@@ -724,7 +725,8 @@ async def _async_process_record(self, record: dict) -> SuccessResponse | Failure
             model = getattr(exc, "model", None) or getattr(exc, "title", None)
             model_name = getattr(self.model, "__name__", None)
 
-            if model in (self.model, model_name):
+            # `data is None` means converting the record failed, e.g. a validator raised an error Pydantic doesn't wrap
+            if model in (self.model, model_name) or data is None:
                 return self._register_model_validation_error_record(record)
 
             return self.failure_handler(record=data, exception=sys.exc_info())
diff --git a/tests/functional/batch/_pydantic/sample_models.py b/tests/functional/batch/_pydantic/sample_models.py
index 4ba2197be7e..56c67befc60 100644
--- a/tests/functional/batch/_pydantic/sample_models.py
+++ b/tests/functional/batch/_pydantic/sample_models.py
@@ -49,3 +49,11 @@ class OrderDynamoDBChangeRecord(DynamoDBStreamChangedRecordModel):
 
 class OrderDynamoDBRecord(DynamoDBStreamRecordModel):
     dynamodb: OrderDynamoDBChangeRecord
+
+
+class OrderIdSqs(SqsRecordModel):
+    @field_validator("body")
+    @classmethod
+    def parse_order_id(cls, value: str):
+        # Pydantic only wraps ValueError/AssertionError, so a KeyError escapes model_validate as-is
+        return json.loads(value)["order_id"]
diff --git a/tests/functional/batch/_pydantic/test_utilities_batch_pydantic.py b/tests/functional/batch/_pydantic/test_utilities_batch_pydantic.py
index 15a7c681fbf..34ed70371fb 100644
--- a/tests/functional/batch/_pydantic/test_utilities_batch_pydantic.py
+++ b/tests/functional/batch/_pydantic/test_utilities_batch_pydantic.py
@@ -29,6 +29,7 @@
 from aws_lambda_powertools.utilities.parser.types import Literal
 from tests.functional.batch._pydantic.sample_models import (
     OrderDynamoDBRecord,
+    OrderIdSqs,
     OrderKinesisRecord,
     OrderSqs,
 )
@@ -512,6 +513,51 @@ def test_batch_processor_model_with_partial_validation_error(
     }
 
 
+def test_batch_processor_model_with_non_validation_error_during_conversion(sqs_event_factory):
+    # GIVEN a model whose validator raises an exception Pydantic doesn't wrap in a ValidationError
+    first_record = sqs_event_factory(json.dumps({"order_id": 1}))
+    malformed_record = sqs_event_factory(json.dumps({"unexpected": 1}))
+    second_record = sqs_event_factory(json.dumps({"order_id": 2}))
+    records = [first_record, malformed_record, second_record]
+
+    # WHEN
+    processor = BatchProcessor(event_type=EventType.SQS, model=OrderIdSqs)
+    with processor(records, lambda record: record.body) as batch:
+        batch.process()
+
+    # THEN only the malformed record is reported, with the original exception
+    assert batch.response() == {
+        "batchItemFailures": [
+            {"itemIdentifier": malformed_record["messageId"]},
+        ],
+    }
+    assert batch.exceptions[0][0] is KeyError
+
+
+def test_async_batch_processor_model_with_non_validation_error_during_conversion(sqs_event_factory):
+    # GIVEN a model whose validator raises an exception Pydantic doesn't wrap in a ValidationError
+    first_record = sqs_event_factory(json.dumps({"order_id": 1}))
+    malformed_record = sqs_event_factory(json.dumps({"unexpected": 1}))
+    second_record = sqs_event_factory(json.dumps({"order_id": 2}))
+    records = [first_record, malformed_record, second_record]
+
+    async def async_record_handler(record):
+        return record.body
+
+    # WHEN
+    processor = AsyncBatchProcessor(event_type=EventType.SQS, model=OrderIdSqs)
+    with processor(records, async_record_handler) as batch:
+        batch.async_process()
+
+    # THEN only the malformed record is reported, with the original exception
+    assert batch.response() == {
+        "batchItemFailures": [
+            {"itemIdentifier": malformed_record["messageId"]},
+        ],
+    }
+    assert batch.exceptions[0][0] is KeyError
+
+
 def test_batch_processor_dynamodb_context_model_with_partial_validation_error(
     dynamodb_record_handler_model: Callable,
     dynamodb_event_factory,