diff --git a/tests/utilities/aws_lambda/test_task_handler.py b/tests/utilities/aws_lambda/test_task_handler.py
index b986c1c5e..2bacf38e0 100644
--- a/tests/utilities/aws_lambda/test_task_handler.py
+++ b/tests/utilities/aws_lambda/test_task_handler.py
@@ -8,6 +8,7 @@ import pytest
 from sqlalchemy import Engine, text
 from sqlmodel import Session
 
+from domain.tasks.subtasks import SubTaskFailure
 from domain.tasks.tasks import Source
 from orchestration.task_orchestrator import TaskOrchestrator
 from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
@@ -153,6 +154,50 @@ def test_task_handler_attaches_to_a_supplied_task_and_subtask(
         assert conn.execute(text("SELECT count(*) FROM sub_task")).scalar_one() == 1
 
 
+def test_recorded_failure_fails_the_subtask_without_an_sqs_retry(
+    harness: Harness,
+) -> None:
+    """A SubTaskFailure is a database record, not retry fuel (ADR-0055): the
+    sub_task fails with the structured details, but the message is NOT
+    reported to SQS as a batch item failure."""
+    # arrange
+    task, subtask = harness.orchestrator.create_task_with_subtask(
+        task_source="app:modelling_run", inputs={"property_ids": [1, 2]}
+    )
+
+    @task_handler(
+        task_source="modelling_e2e",
+        source=Source.PROPERTY,
+        orchestrator_cm=harness.factory,
+    )
+    def handler(body: dict[str, Any], context: Any) -> None:
+        raise SubTaskFailure(
+            "1 of 2 properties failed",
+            details={"succeeded": 1, "failed": [{"property_id": 2, "error": "x"}]},
+        )
+
+    event = {
+        "Records": [
+            {
+                "messageId": "m-1",
+                "body": (
+                    f'{{"task_id": "{task.id}", "subtask_id": "{subtask.id}", '
+                    f'"property_ids": [1, 2]}}'
+                ),
+            }
+        ]
+    }
+
+    # act
+    result = handler(event, context=None)
+
+    # assert — recorded, not retried
+    assert result["batchItemFailures"] == []
+    failed = harness.subtasks.get(subtask.id)
+    assert failed.status.value == "failed"
+    assert failed.outputs is not None and failed.outputs["succeeded"] == 1
+
+
 def test_task_handler_leaves_cloudwatch_url_unset_outside_lambda(
     harness: Harness, monkeypatch: pytest.MonkeyPatch
 ) -> None: