diff --git a/domain/tasks/subtasks.py b/domain/tasks/subtasks.py
index 95f9ec5bc..edb3ee508 100644
--- a/domain/tasks/subtasks.py
+++ b/domain/tasks/subtasks.py
@@ -5,6 +5,19 @@ from typing import Any, Optional
 from uuid import UUID, uuid4
 
 
+class SubTaskFailure(Exception):
+    """A *recorded* failure: the work ran, its outcome belongs on the SubTask
+    record, and the message must NOT be retried (ADR-0055 — status is a
+    record, not retry fuel). Infra crashes raise anything else and keep their
+    retry semantics. ``details`` lands on the failed SubTask's outputs."""
+
+    def __init__(
+        self, message: str, details: Optional[dict[str, Any]] = None
+    ) -> None:
+        super().__init__(message)
+        self.details = details
+
+
 class SubTaskStatus(str, Enum):
     WAITING = "waiting"
     IN_PROGRESS = "in progress"
diff --git a/tests/domain/tasks/test_subtasks.py b/tests/domain/tasks/test_subtasks.py
index 18d09a6a5..8d000ba96 100644
--- a/tests/domain/tasks/test_subtasks.py
+++ b/tests/domain/tasks/test_subtasks.py
@@ -2,7 +2,7 @@ from uuid import uuid4
 
 import pytest
 
-from domain.tasks.subtasks import SubTask, SubTaskStatus
+from domain.tasks.subtasks import SubTask, SubTaskFailure, SubTaskStatus
 
 
 def test_create_subtask_starts_waiting() -> None:
@@ -95,6 +95,31 @@ def test_complete_without_result_leaves_outputs_unset() -> None:
     assert st.outputs is None
 
 
+def test_fail_with_subtask_failure_records_structured_details() -> None:
+    # arrange
+    st = SubTask.create(task_id=uuid4())
+    st.start()
+
+    # act
+    st.fail(
+        SubTaskFailure(
+            "1 of 2 properties failed",
+            details={
+                "succeeded": 1,
+                "failed": [{"property_id": 9, "error": "degenerate prediction"}],
+            },
+        )
+    )
+
+    # assert
+    assert st.status is SubTaskStatus.FAILED
+    assert st.outputs == {
+        "error": "1 of 2 properties failed",
+        "succeeded": 1,
+        "failed": [{"property_id": 9, "error": "degenerate prediction"}],
+    }
+
+
 def test_fail_records_error_in_outputs() -> None:
     # arrange
     st = SubTask.create(task_id=uuid4())