diff --git a/database/sql/src/main/scala/cromwell/database/slick/MetadataSlickDatabase.scala b/database/sql/src/main/scala/cromwell/database/slick/MetadataSlickDatabase.scala
index 1405f7d101f..0a33e4b6b72 100644
--- a/database/sql/src/main/scala/cromwell/database/slick/MetadataSlickDatabase.scala
+++ b/database/sql/src/main/scala/cromwell/database/slick/MetadataSlickDatabase.scala
@@ -515,4 +515,9 @@ class MetadataSlickDatabase(originalDatabaseConfig: Config)
   override def getMetadataTableSizeInformation()(implicit ec: ExecutionContext): Future[Option[InformationSchemaEntry]] = {
     runAction(dataAccess.metadataTableSizeInformation())
   }
+
+  override def getFailedJobsMetadataWithWorkflowId(rootWorkflowId: String)(implicit ec: ExecutionContext): Future[Vector[MetadataEntry]] = {
+    val isPostgres = databaseConfig.getValue("db.driver").toString.toLowerCase().contains("postgres")
+    runLobAction(dataAccess.failedJobsMetadataWithWorkflowId(rootWorkflowId, isPostgres))
+  }
 }
diff --git a/database/sql/src/main/scala/cromwell/database/slick/tables/MetadataEntryComponent.scala b/database/sql/src/main/scala/cromwell/database/slick/tables/MetadataEntryComponent.scala
index 32256c448fb..f4b2724b6a7 100644
--- a/database/sql/src/main/scala/cromwell/database/slick/tables/MetadataEntryComponent.scala
+++ b/database/sql/src/main/scala/cromwell/database/slick/tables/MetadataEntryComponent.scala
@@ -310,6 +310,84 @@ trait MetadataEntryComponent {
     }).headOption
   }
 
+  def failedJobsMetadataWithWorkflowId(rootWorkflowId: String, isPostgres: Boolean) = {
+    val getMetadataEntryResult = GetResult(r => {
+      MetadataEntry(r.<<, r.<<, r.<<, r.<<, r.<<, r.nextClobOption().map(clob => new SerialClob(clob)), r.<<, r.<<, r.<<)
+    })
+
+    def dbIdentifierWrapper(identifier: String, isPostgres: Boolean) = {
+      if(isPostgres) s"${'"'}$identifier${'"'}" else identifier
+    }
+
+    def dbMetadataValueColCheckName(isPostgres: Boolean): String = {
+      if(isPostgres) "obj.data" else "METADATA_VALUE"
+    }
+
+    def targetCallsSelectStatement(callFqn: String, scatterIndex: String, retryAttempt: String): String = {
+      s"SELECT ${callFqn}, MAX(COALESCE(${scatterIndex}, 0)) as maxScatter, MAX(COALESCE(${retryAttempt}, 0)) AS maxRetry"
+    }
+
+    def pgObjectInnerJoinStatement(isPostgres: Boolean, metadataValColName: String): String = {
+      if(isPostgres) s"INNER JOIN pg_largeobject obj ON me.${metadataValColName} = cast(obj.loid as text)" else ""
+    }
+
+    val workflowUuid = dbIdentifierWrapper("WORKFLOW_EXECUTION_UUID", isPostgres)
+    val callFqn = dbIdentifierWrapper("CALL_FQN", isPostgres)
+    val scatterIndex = dbIdentifierWrapper("JOB_SCATTER_INDEX", isPostgres)
+    val retryAttempt = dbIdentifierWrapper("JOB_RETRY_ATTEMPT", isPostgres)
+    val metadataKey = dbIdentifierWrapper("METADATA_KEY", isPostgres)
+    val metadataValueType = dbIdentifierWrapper("METADATA_VALUE_TYPE", isPostgres)
+    val metadataTimestamp = dbIdentifierWrapper("METADATA_TIMESTAMP", isPostgres)
+    val metadataJournalId = dbIdentifierWrapper("METADATA_JOURNAL_ID", isPostgres)
+    val rootUuid = dbIdentifierWrapper("ROOT_WORKFLOW_EXECUTION_UUID", isPostgres)
+    val metadataValue = dbIdentifierWrapper("METADATA_VALUE", isPostgres)
+    val metadataEntry = dbIdentifierWrapper("METADATA_ENTRY", isPostgres)
+    val wmse = dbIdentifierWrapper("WORKFLOW_METADATA_SUMMARY_ENTRY", isPostgres)
+    val resultSetColumnNames = s"me.${workflowUuid}, me.${callFqn}, me.${scatterIndex}, me.${retryAttempt}, me.${metadataKey}, me.${metadataValue}, me.${metadataValueType}, me.${metadataTimestamp}, me.${metadataJournalId}"
+
+    val query = sql"""
+      SELECT #${resultSetColumnNames}
+      FROM #${metadataEntry} me
+      INNER JOIN (
+        #${targetCallsSelectStatement(callFqn, scatterIndex, retryAttempt)}
+        FROM #${metadataEntry} me
+        INNER JOIN #${wmse} wmse
+        ON wmse.#${workflowUuid} = me.#${workflowUuid}
+        WHERE (wmse.#${rootUuid} = $rootWorkflowId OR wmse.#${workflowUuid} = $rootWorkflowId)
+        AND #${callFqn} IS NOT NULL
+        GROUP BY #${callFqn}
+      ) AS targetCalls
+      ON me.#${callFqn} = targetCalls.#${callFqn}
+      LEFT JOIN (
+        SELECT DISTINCT #${callFqn}
+        FROM #${metadataEntry} me
+        INNER JOIN #${wmse} wmse
+        ON wmse.#${workflowUuid} = me.#${workflowUuid}
+        WHERE (wmse.#${rootUuid} = $rootWorkflowId OR wmse.#${workflowUuid} = $rootWorkflowId)
+        AND me.#${metadataKey} = 'subWorkflowId'
+        GROUP BY #${callFqn}
+      ) AS avoidedCalls
+      ON me.#${callFqn} = avoidedCalls.#${callFqn}
+      INNER JOIN #${wmse} wmse
+      ON wmse.#${workflowUuid} = me.#${workflowUuid}
+      #${pgObjectInnerJoinStatement(isPostgres, metadataValue)}
+      WHERE avoidedCalls.#${callFqn} IS NULL
+      AND (me.#${metadataKey} in ('executionStatus', 'backendStatus') AND #${dbMetadataValueColCheckName(isPostgres)} = 'Failed')
+      AND (
+        (COALESCE(me.#${retryAttempt}, 0) = targetCalls.maxRetry AND me.#${scatterIndex} IS NULL)
+        OR (COALESCE(me.#${retryAttempt}, 0) = targetCalls.maxRetry AND me.#${scatterIndex} = targetCalls.maxScatter)
+       )
+      GROUP BY #${resultSetColumnNames}
+      HAVING me.#${workflowUuid} IN (
+        SELECT DISTINCT wmse.#${workflowUuid}
+        FROM #${wmse} wmse
+        WHERE wmse.#${rootUuid} = $rootWorkflowId OR wmse.#${workflowUuid} = $rootWorkflowId
+      )
+    """
+
+    query.as(getMetadataEntryResult)
+  }
+
   private[this] def metadataEntryHasMetadataKeysLike(metadataEntry: MetadataEntries,
                                                      metadataKeysToFilterFor: List[String],
                                                      metadataKeysToFilterOut: List[String]): Rep[Boolean] = {
diff --git a/database/sql/src/main/scala/cromwell/database/sql/MetadataSqlDatabase.scala b/database/sql/src/main/scala/cromwell/database/sql/MetadataSqlDatabase.scala
index 30534818f85..9139c819999 100644
--- a/database/sql/src/main/scala/cromwell/database/sql/MetadataSqlDatabase.scala
+++ b/database/sql/src/main/scala/cromwell/database/sql/MetadataSqlDatabase.scala
@@ -197,4 +197,6 @@ trait MetadataSqlDatabase extends SqlDatabase {
   def countWorkflowsLeftToDeleteThatEndedOnOrBeforeThresholdTimestamp(workflowEndTimestampThreshold: Timestamp)(implicit ec: ExecutionContext): Future[Int]
 
   def getMetadataTableSizeInformation()(implicit ec: ExecutionContext): Future[Option[InformationSchemaEntry]]
+
+  def getFailedJobsMetadataWithWorkflowId(rootWorkflowId: String)(implicit ec: ExecutionContext): Future[Vector[MetadataEntry]]
 }
diff --git a/docs/api/RESTAPI.md b/docs/api/RESTAPI.md
index 7a0585ca08e..64ffe19a3fd 100644
--- a/docs/api/RESTAPI.md
+++ b/docs/api/RESTAPI.md
@@ -1,5 +1,5 @@
 <!--
-This file was generated by `sbt generateRestApiDocs` on Mon, 27 Feb 2023 15:52:59 -0500
+This file was generated by `sbt generateRestApiDocs` on Wed, 12 Jul 2023 15:55:37 -0400
 
 !!! DO NOT CHANGE THIS FILE DIRECTLY !!!
 
@@ -699,6 +699,36 @@ GET /api/workflows/{version}/{id}/metadata
 * Workflows
 
 
+<a name="failed-jobs"></a>
+## Get call-level metadata of failed tasks for a specified root workflow
+```
+GET /api/workflows/{version}/{id}/metadata/failed-jobs
+```
+
+
+#### Parameters
+
+|Type|Name|Description|Schema|Default|
+|---|---|---|---|---|
+|**Path**|**id**  <br>*required*|A workflow ID|string||
+|**Path**|**version**  <br>*required*|Cromwell API Version|string|`"v1"`|
+
+
+#### Responses
+
+|HTTP Code|Description|Schema|
+|---|---|---|
+|**200**|Successful request|[WorkflowMetadataResponse](#workflowmetadataresponse)|
+|**400**|Malformed Workflow ID|No Content|
+|**404**|Workflow ID Not Found|No Content|
+|**500**|Internal Error|No Content|
+
+
+#### Tags
+
+* Workflows
+
+
 <a name="outputs"></a>
 ## Get the outputs for a workflow
 ```
diff --git a/engine/src/main/resources/swagger/cromwell.yaml b/engine/src/main/resources/swagger/cromwell.yaml
index bc34cb6476b..dca96b20ceb 100644
--- a/engine/src/main/resources/swagger/cromwell.yaml
+++ b/engine/src/main/resources/swagger/cromwell.yaml
@@ -510,6 +510,26 @@ paths:
           $ref: '#/responses/NotFound'
         '500':
           $ref: '#/responses/ServerError'
+  '/api/workflows/{version}/{id}/metadata/failed-jobs':
+    get:
+      operationId: failed-jobs
+      summary: Get call-level metadata of failed tasks for a specified root workflow
+      parameters:
+        - $ref: '#/parameters/versionParam'
+        - $ref: '#/parameters/singleId'
+      tags:
+        - Workflows
+      responses:
+        '200':
+          description: Successful request
+          schema:
+            $ref: '#/definitions/WorkflowMetadataResponse'
+        '400':
+          $ref: '#/responses/BadRequest'
+        '404':
+          $ref: '#/responses/NotFound'
+        '500':
+          $ref: '#/responses/ServerError'
   '/api/workflows/{version}/{id}/metadata':
     get:
       operationId: metadata
diff --git a/engine/src/main/scala/cromwell/webservice/routes/MetadataRouteSupport.scala b/engine/src/main/scala/cromwell/webservice/routes/MetadataRouteSupport.scala
index 237e21a2edd..b84ff59d5a3 100644
--- a/engine/src/main/scala/cromwell/webservice/routes/MetadataRouteSupport.scala
+++ b/engine/src/main/scala/cromwell/webservice/routes/MetadataRouteSupport.scala
@@ -69,6 +69,17 @@ trait MetadataRouteSupport extends HttpInstrumentation {
         }
       }
     },
+    encodeResponse {
+      path("workflows" / Segment / Segment / "metadata" / "failed-jobs") { (_, possibleWorkflowId) =>
+        instrumentRequest {
+          metadataLookup(
+            possibleWorkflowId,
+            (w: WorkflowId) => FetchFailedJobsMetadataWithWorkflowId(w),
+            serviceRegistryActor
+          )
+        }
+      }
+    },
     encodeResponse {
       path("workflows" / Segment / Segment / "metadata") { (_, possibleWorkflowId) =>
         instrumentRequest {
@@ -212,8 +223,7 @@ object MetadataRouteSupport {
   }
 
   def completeMetadataBuilderResponse(response: Future[MetadataJsonResponse]): Route = {
-    onComplete(response) {
-      case Success(r: SuccessfulMetadataJsonResponse) => complete(r.responseJson)
+    onComplete(response) { case Success(r: SuccessfulMetadataJsonResponse) => complete(r.responseJson)
       case Success(r: FailedMetadataJsonResponse) => r.reason.errorRequest(StatusCodes.InternalServerError)
       case Failure(_: AskTimeoutException) if CromwellShutdown.shutdownInProgress() => serviceShuttingDownResponse
       case Failure(e: UnrecognizedWorkflowException) => e.failRequest(StatusCodes.NotFound)
diff --git a/engine/src/test/scala/cromwell/webservice/MetadataBuilderActorSpec.scala b/engine/src/test/scala/cromwell/webservice/MetadataBuilderActorSpec.scala
index a520faee1f2..700c0274026 100644
--- a/engine/src/test/scala/cromwell/webservice/MetadataBuilderActorSpec.scala
+++ b/engine/src/test/scala/cromwell/webservice/MetadataBuilderActorSpec.scala
@@ -36,6 +36,7 @@ class MetadataBuilderActorSpec extends TestKitSuite with AsyncFlatSpecLike with
                              events: Seq[MetadataEvent],
                              expectedRes: String,
                              metadataBuilderActorName: String,
+                             failedTasks: Boolean = false
                             ): Future[Assertion] = {
     val mockReadMetadataWorkerActor = TestProbe("mockReadMetadataWorkerActor")
     def readMetadataWorkerMaker = () => mockReadMetadataWorkerActor.props
@@ -45,13 +46,17 @@ class MetadataBuilderActorSpec extends TestKitSuite with AsyncFlatSpecLike with
       props = MetadataBuilderActor.props(readMetadataWorkerMaker, 1000000),
       name = metadataBuilderActorName,
     )
+
     val response = mba.ask(action).mapTo[MetadataJsonResponse]
     mockReadMetadataWorkerActor.expectMsg(defaultTimeout, action)
-    mockReadMetadataWorkerActor.reply(MetadataLookupResponse(queryReply, events))
+    mockReadMetadataWorkerActor.reply(
+      if(failedTasks) FetchFailedJobsMetadataLookupResponse(events) else MetadataLookupResponse(queryReply, events)
+    )
     response map { r => r shouldBe a [SuccessfulMetadataJsonResponse] }
     response.mapTo[SuccessfulMetadataJsonResponse] map { b => b.responseJson shouldBe expectedRes.parseJson}
   }
 
+
   def assertMetadataFailureResponse(action: MetadataServiceAction,
                                     metadataServiceResponse: MetadataServiceResponse,
                                     expectedException: Exception,
@@ -95,37 +100,37 @@ class MetadataBuilderActorSpec extends TestKitSuite with AsyncFlatSpecLike with
     // We'll use a Query instead of a SingleWorkflowMetadataGet, so we expect the WorkflowID this time:
     val expectedRes =
       s"""{
-        |  "calls": {
-        |    "callB": [{
-        |      "attempt": 1,
-        |      "NOT_CHECKED": "NOT_CHECKED",
-        |      "shardIndex": -1
-        |    }, {
-        |      "attempt": 1,
-        |      "NOT_CHECKED": "NOT_CHECKED",
-        |      "shardIndex": 0
-        |    }, {
-        |      "attempt": 1,
-        |      "NOT_CHECKED": "NOT_CHECKED",
-        |      "shardIndex": 1
-        |    }, {
-        |      "attempt": 2,
-        |      "NOT_CHECKED": "NOT_CHECKED",
-        |      "shardIndex": 1
-        |    }, {
-        |      "attempt": 3,
-        |      "NOT_CHECKED": "NOT_CHECKED",
-        |      "shardIndex": 1
-        |    }],
-        |    "callA": [{
-        |      "attempt": 1,
-        |      "NOT_CHECKED": "NOT_CHECKED",
-        |      "shardIndex": -1
-        |    }]
-        |  },
-        |  "NOT_CHECKED": "NOT_CHECKED",
-        |  "id": "$workflowA"
-      |}""".stripMargin
+         |  "calls": {
+         |    "callB": [{
+         |      "attempt": 1,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": -1
+         |    }, {
+         |      "attempt": 1,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": 0
+         |    }, {
+         |      "attempt": 1,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": 1
+         |    }, {
+         |      "attempt": 2,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": 1
+         |    }, {
+         |      "attempt": 3,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": 1
+         |    }],
+         |    "callA": [{
+         |      "attempt": 1,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": -1
+         |    }]
+         |  },
+         |  "NOT_CHECKED": "NOT_CHECKED",
+         |  "id": "$workflowA"
+         |}""".stripMargin
 
     val mdQuery = MetadataQuery(workflowA, None, None, None, None, expandSubWorkflows = false)
     val queryAction = GetMetadataAction(mdQuery)
@@ -134,7 +139,7 @@ class MetadataBuilderActorSpec extends TestKitSuite with AsyncFlatSpecLike with
       queryReply = mdQuery,
       events = workflowAEvents,
       expectedRes = expectedRes,
-      metadataBuilderActorName = "mba-scope-tree",
+      metadataBuilderActorName = "mba-scope-tree"
     )
   }
 
@@ -162,6 +167,7 @@ class MetadataBuilderActorSpec extends TestKitSuite with AsyncFlatSpecLike with
                                  eventMaker: WorkflowId => (String, MetadataValue, OffsetDateTime) => MetadataEvent =
                                  makeEvent,
                                  metadataBuilderActorName: String,
+                                 isFailedTaskFetch: Boolean = false
                                 ): Future[Assertion] = {
 
     val events = eventList map { e => (e._1, MetadataValue(e._2), e._3) } map Function.tupled(eventMaker(workflow))
@@ -172,6 +178,74 @@ class MetadataBuilderActorSpec extends TestKitSuite with AsyncFlatSpecLike with
     assertMetadataResponse(queryAction, mdQuery, events, expectedRes, metadataBuilderActorName)
   }
 
+  it should "build the call list for failed tasks when prompted" in {
+
+    def makeEvent(workflow: WorkflowId, key: Option[MetadataJobKey]) = {
+      MetadataEvent(MetadataKey(workflow, key, "NOT_CHECKED"), MetadataValue("NOT_CHECKED"))
+    }
+
+    val workflowA = WorkflowId.randomId()
+
+    val workflowACalls = List(
+      Option(MetadataJobKey("callB", Option(1), 3)),
+      Option(MetadataJobKey("callB", None, 1)),
+      Option(MetadataJobKey("callB", Option(1), 2)),
+      Option(MetadataJobKey("callA", None, 1)),
+      Option(MetadataJobKey("callB", Option(1), 1)),
+      Option(MetadataJobKey("callB", Option(0), 1)),
+      None
+    )
+    val workflowAEvents = workflowACalls map {
+      makeEvent(workflowA, _)
+    }
+
+    val expectedRes =
+      s"""{
+         |"${workflowA}": {
+         |  "calls": {
+         |    "callB": [{
+         |      "attempt": 1,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": -1
+         |    }, {
+         |      "attempt": 1,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": 0
+         |    }, {
+         |      "attempt": 1,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": 1
+         |    }, {
+         |      "attempt": 2,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": 1
+         |    }, {
+         |      "attempt": 3,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": 1
+         |    }],
+         |    "callA": [{
+         |      "attempt": 1,
+         |      "NOT_CHECKED": "NOT_CHECKED",
+         |      "shardIndex": -1
+         |    }]
+         |  },
+         |  "NOT_CHECKED": "NOT_CHECKED"
+         | }
+         |}""".stripMargin
+
+    val mdQuery = MetadataQuery(workflowA, None, None, None, None, expandSubWorkflows = false)
+    val queryAction = GetMetadataAction(mdQuery)
+    assertMetadataResponse(
+      action = queryAction,
+      queryReply = mdQuery,
+      events = workflowAEvents,
+      expectedRes = expectedRes,
+      failedTasks = true,
+      metadataBuilderActorName = "mba-failed-tasks",
+    )
+  }
+
   it should "assume the event list is ordered and keep last event if 2 events have same key" in {
     val eventBuilderList = List(
       ("a", "aLater", OffsetDateTime.parse("2000-01-02T12:00:00Z")),
diff --git a/engine/src/test/scala/cromwell/webservice/routes/CromwellApiServiceSpec.scala b/engine/src/test/scala/cromwell/webservice/routes/CromwellApiServiceSpec.scala
index c40c84397f7..4e7caf932ef 100644
--- a/engine/src/test/scala/cromwell/webservice/routes/CromwellApiServiceSpec.scala
+++ b/engine/src/test/scala/cromwell/webservice/routes/CromwellApiServiceSpec.scala
@@ -568,6 +568,7 @@ object CromwellApiServiceSpec {
         MetadataEvent(MetadataKey(workflowId, None, "testKey2a"), MetadataValue("myValue2a", MetadataString)),
       )
     }
+
     private def wesFullMetadataResponse(workflowId: WorkflowId) = {
       List(
         MetadataEvent(MetadataKey(workflowId, None, "status"), MetadataValue("Running", MetadataString)),
@@ -576,6 +577,7 @@ object CromwellApiServiceSpec {
       )
     }
 
+
     def responseMetadataValues(workflowId: WorkflowId, withKeys: List[String], withoutKeys: List[String]): JsObject = {
       def keyFilter(keys: List[String])(m: MetadataEvent) = keys.exists(k => m.key.key.startsWith(k))
       val metadataEvents = if (workflowId == wesWorkflowId) wesFullMetadataResponse(workflowId) else fullMetadataResponse(workflowId)
@@ -645,6 +647,8 @@ object CromwellApiServiceSpec {
         sender() ! SuccessfulMetadataJsonResponse(request, MetadataBuilderActor.processOutputsResponse(id, event))
       case request @ GetLogs(id) =>
         sender() ! SuccessfulMetadataJsonResponse(request, MetadataBuilderActor.workflowMetadataResponse(id, logsEvents(id), includeCallsIfEmpty = false, Map.empty))
+      case request @ FetchFailedJobsMetadataWithWorkflowId(id) =>
+        sender() ! SuccessfulMetadataJsonResponse(request, responseMetadataValues(id, List.empty, List.empty))
       case request @ GetMetadataAction(MetadataQuery(id, _, _, withKeys, withoutKeys, _), _) =>
         val withKeysList = withKeys.map(_.toList).getOrElse(List.empty)
         val withoutKeysList = withoutKeys.map(_.toList).getOrElse(List.empty)
diff --git a/engine/src/test/scala/cromwell/webservice/routes/MetadataRouteSupportSpec.scala b/engine/src/test/scala/cromwell/webservice/routes/MetadataRouteSupportSpec.scala
index 07b780a3aaf..73cc257e828 100644
--- a/engine/src/test/scala/cromwell/webservice/routes/MetadataRouteSupportSpec.scala
+++ b/engine/src/test/scala/cromwell/webservice/routes/MetadataRouteSupportSpec.scala
@@ -313,6 +313,20 @@ class MetadataRouteSupportSpec extends AsyncFlatSpec with ScalatestRouteTest wit
       }
   }
 
+  it should "return 200 with metadata for failed tasks on a workflow" in {
+    Get(s"/workflows/$version/${CromwellApiServiceSpec.ExistingWorkflowId}/metadata/failed-jobs") ~>
+      akkaHttpService.metadataRoutes ~>
+      check {
+        status should be(StatusCodes.OK)
+        val result = responseAs[JsObject]
+        result.fields.keys should contain allOf("testKey1a", "testKey1b", "testKey2a")
+        result.fields.keys shouldNot contain("testKey3")
+        result.fields("testKey1a") should be(JsString("myValue1a"))
+        result.fields("testKey1b") should be(JsString("myValue1b"))
+        result.fields("testKey2a") should be(JsString("myValue2a"))
+      }
+  }
+
   behavior of "REST API /query GET endpoint"
   it should "return good results for a good query" in {
     Get(s"/workflows/$version/query?status=Succeeded&id=${CromwellApiServiceSpec.ExistingWorkflowId}") ~>
diff --git a/services/src/main/scala/cromwell/services/metadata/MetadataQuery.scala b/services/src/main/scala/cromwell/services/metadata/MetadataQuery.scala
index d3264b839c3..1b38bd2f479 100644
--- a/services/src/main/scala/cromwell/services/metadata/MetadataQuery.scala
+++ b/services/src/main/scala/cromwell/services/metadata/MetadataQuery.scala
@@ -26,7 +26,9 @@ object MetadataKey {
 
 object MetadataEvent {
   def apply(key: MetadataKey, value: MetadataValue) = new MetadataEvent(key, Option(value), OffsetDateTime.now)
+
   def apply(key: MetadataKey, optionalValue: Option[MetadataValue]) = new MetadataEvent(key, optionalValue, OffsetDateTime.now)
+
   def empty(key: MetadataKey) = new MetadataEvent(key, None, OffsetDateTime.now)
 
   def labelsToMetadataEvents(labels: Labels, workflowId: WorkflowId): Iterable[MetadataEvent] = {
diff --git a/services/src/main/scala/cromwell/services/metadata/MetadataService.scala b/services/src/main/scala/cromwell/services/metadata/MetadataService.scala
index 61acd8052c8..dd45cf5535d 100644
--- a/services/src/main/scala/cromwell/services/metadata/MetadataService.scala
+++ b/services/src/main/scala/cromwell/services/metadata/MetadataService.scala
@@ -127,6 +127,7 @@ object MetadataService {
   final case class ValidateWorkflowIdInMetadata(possibleWorkflowId: WorkflowId) extends MetadataServiceAction
   final case class ValidateWorkflowIdInMetadataSummaries(possibleWorkflowId: WorkflowId) extends MetadataServiceAction
   final case class FetchWorkflowMetadataArchiveStatusAndEndTime(workflowId: WorkflowId) extends MetadataServiceAction
+  final case class FetchFailedJobsMetadataWithWorkflowId(workflowId: WorkflowId) extends BuildWorkflowMetadataJsonAction
 
   /**
     * Responses
@@ -140,8 +141,10 @@ object MetadataService {
   final case class MetadataLookupStreamFailed(id: WorkflowId, reason: Throwable) extends MetadataServiceResponse
   final case class MetadataLookupFailedTooLargeResponse(query: MetadataQuery, metadataSizeRows: Int) extends MetadataServiceResponse
   final case class MetadataLookupFailedTimeoutResponse(query: MetadataQuery) extends MetadataServiceResponse
-
+  final case class FetchFailedTasksTimeoutResponse(workflowId: WorkflowId) extends MetadataServiceResponse
   final case class MetadataLookupResponse(query: MetadataQuery, eventList: Seq[MetadataEvent]) extends MetadataServiceResponse
+  final case class FetchFailedJobsMetadataLookupResponse(events: Seq[MetadataEvent]) extends MetadataServiceResponse
+  final case class FetchFailedJobsMetadataLookupFailed(workflowId: WorkflowId, reason: Throwable) extends MetadataServiceFailure
   final case class MetadataServiceKeyLookupFailed(query: MetadataQuery, reason: Throwable) extends MetadataServiceFailure
 
   final case class StatusLookupResponse(workflowId: WorkflowId, status: WorkflowState) extends MetadataServiceResponse
diff --git a/services/src/main/scala/cromwell/services/metadata/impl/MetadataDatabaseAccess.scala b/services/src/main/scala/cromwell/services/metadata/impl/MetadataDatabaseAccess.scala
index 271251bcdd7..0069ffe8821 100644
--- a/services/src/main/scala/cromwell/services/metadata/impl/MetadataDatabaseAccess.scala
+++ b/services/src/main/scala/cromwell/services/metadata/impl/MetadataDatabaseAccess.scala
@@ -114,7 +114,7 @@ trait MetadataDatabaseAccess {
       labelMetadataKey = WorkflowMetadataKeys.Labels)
   }
 
-  private def metadataToMetadataEvents(workflowId: WorkflowId)(metadata: Seq[MetadataEntry]): Seq[MetadataEvent] = {
+  def metadataToMetadataEvents(workflowId: WorkflowId)(metadata: Seq[MetadataEntry]): Seq[MetadataEvent] = {
     metadata map { m =>
       // If callFullyQualifiedName is non-null then attempt will also be non-null and there is a MetadataJobKey.
       val metadataJobKey: Option[MetadataJobKey] = for {
@@ -382,4 +382,7 @@ trait MetadataDatabaseAccess {
     metadataDatabaseInterface.countWorkflowsLeftToDeleteThatEndedOnOrBeforeThresholdTimestamp(workflowEndTimestampThreshold.toSystemTimestamp)
 
   def getMetadataTableSizeInformation()(implicit ec: ExecutionContext): Future[Option[InformationSchemaEntry]] = metadataDatabaseInterface.getMetadataTableSizeInformation()
+
+  def getFailedJobsMetadataWithWorkflowId(rootWorkflowId: WorkflowId)(implicit ec: ExecutionContext): Future[Vector[MetadataEntry]] =
+    metadataDatabaseInterface.getFailedJobsMetadataWithWorkflowId(rootWorkflowId.toString)
 }
diff --git a/services/src/main/scala/cromwell/services/metadata/impl/ReadDatabaseMetadataWorkerActor.scala b/services/src/main/scala/cromwell/services/metadata/impl/ReadDatabaseMetadataWorkerActor.scala
index 3a67ad504ec..165c1bc2f14 100644
--- a/services/src/main/scala/cromwell/services/metadata/impl/ReadDatabaseMetadataWorkerActor.scala
+++ b/services/src/main/scala/cromwell/services/metadata/impl/ReadDatabaseMetadataWorkerActor.scala
@@ -1,13 +1,12 @@
 package cromwell.services.metadata.impl
 
 import java.sql.SQLTimeoutException
-
 import akka.actor.{Actor, ActorLogging, ActorRef, PoisonPill, Props}
 import cromwell.core.Dispatcher.ServiceDispatcher
 import cromwell.core.{WorkflowId, WorkflowSubmitted}
 import cromwell.services.MetadataServicesStore
 import cromwell.services.metadata.MetadataService._
-import cromwell.services.metadata.{MetadataQuery, WorkflowQueryParameters}
+import cromwell.services.metadata.{MetadataEvent, MetadataQuery, WorkflowQueryParameters}
 
 import scala.concurrent.Future
 import scala.concurrent.duration.Duration
@@ -27,6 +26,9 @@ class ReadDatabaseMetadataWorkerActor(metadataReadTimeout: Duration, metadataRea
   implicit val ec = context.dispatcher
 
   def receive = {
+    case FetchFailedJobsMetadataWithWorkflowId(w: WorkflowId) => {
+      evaluateRespondAndStop(sender(), getFailedJobs(w))
+    }
     case GetMetadataAction(query: MetadataQuery, checkTotalMetadataRowNumberBeforeQuerying: Boolean) =>
       evaluateRespondAndStop(sender(), getMetadata(query, checkTotalMetadataRowNumberBeforeQuerying))
     case GetMetadataStreamAction(workflowId) =>
@@ -84,6 +86,16 @@ class ReadDatabaseMetadataWorkerActor(metadataReadTimeout: Duration, metadataRea
       case t => MetadataServiceKeyLookupFailed(query, t)
     }
 
+  private def getFailedJobs(workflowId: WorkflowId): Future[MetadataServiceResponse] = {
+   val futureEvents: Future[Seq[MetadataEvent]] = getFailedJobsMetadataWithWorkflowId(workflowId) map metadataToMetadataEvents(workflowId)
+    futureEvents map {
+      m => FetchFailedJobsMetadataLookupResponse(m)
+    } recover {
+      case _: SQLTimeoutException => FetchFailedTasksTimeoutResponse(workflowId)
+      case t => FetchFailedJobsMetadataLookupFailed(workflowId, t)
+    }
+  }
+
   private def getStatus(id: WorkflowId): Future[MetadataServiceResponse] = {
 
     getWorkflowStatus(id) map {
diff --git a/services/src/main/scala/cromwell/services/metadata/impl/ReadMetadataRegulatorActor.scala b/services/src/main/scala/cromwell/services/metadata/impl/ReadMetadataRegulatorActor.scala
index 1c50cb00404..869ae79e10f 100644
--- a/services/src/main/scala/cromwell/services/metadata/impl/ReadMetadataRegulatorActor.scala
+++ b/services/src/main/scala/cromwell/services/metadata/impl/ReadMetadataRegulatorActor.scala
@@ -1,12 +1,11 @@
 package cromwell.services.metadata.impl
 
 import java.util.UUID
-
 import akka.actor.{Actor, ActorLogging, ActorRef, Props}
 import cromwell.core.Dispatcher.ApiDispatcher
 import cromwell.services.MetadataJsonResponse
 import cromwell.services.metadata.MetadataService
-import cromwell.services.metadata.MetadataService.{BuildMetadataJsonAction, BuildWorkflowMetadataJsonAction, GetMetadataStreamAction, MetadataLookupStreamSuccess, MetadataQueryResponse, MetadataServiceAction, MetadataServiceResponse, RootAndSubworkflowLabelsLookupResponse}
+import cromwell.services.metadata.MetadataService.{BuildMetadataJsonAction, BuildWorkflowMetadataJsonAction, FetchFailedJobsMetadataWithWorkflowId, GetMetadataStreamAction, MetadataLookupStreamSuccess, MetadataQueryResponse, MetadataServiceAction, MetadataServiceResponse, RootAndSubworkflowLabelsLookupResponse}
 import cromwell.services.metadata.impl.ReadMetadataRegulatorActor.PropsMaker
 import cromwell.services.metadata.impl.builder.MetadataBuilderActor
 
@@ -28,11 +27,18 @@ class ReadMetadataRegulatorActor(metadataBuilderActorProps: PropsMaker, readMeta
     // This indirection via 'MetadataReadAction' lets the compiler make sure we cover all cases in the sealed trait:
     case action: BuildMetadataJsonAction =>
       action match {
+        case fetchFailedJobsMetadataAction: FetchFailedJobsMetadataWithWorkflowId =>
+          val currentRequesters = apiRequests.getOrElse(fetchFailedJobsMetadataAction, Set.empty)
+          apiRequests.put(fetchFailedJobsMetadataAction, currentRequesters + sender())
+          if (currentRequesters.isEmpty) {
+            val builderActor = context.actorOf(metadataBuilderActorProps().withDispatcher(ApiDispatcher), MetadataBuilderActor.uniqueActorName(fetchFailedJobsMetadataAction.workflowId.toString))
+            builderRequests.put(builderActor, fetchFailedJobsMetadataAction)
+            builderActor ! fetchFailedJobsMetadataAction
+          }
         case singleWorkflowAction: BuildWorkflowMetadataJsonAction =>
           val currentRequesters = apiRequests.getOrElse(singleWorkflowAction, Set.empty)
           apiRequests.put(singleWorkflowAction, currentRequesters + sender())
           if (currentRequesters.isEmpty) {
-
             val builderActor = context.actorOf(metadataBuilderActorProps().withDispatcher(ApiDispatcher), MetadataBuilderActor.uniqueActorName(singleWorkflowAction.workflowId.toString))
             builderRequests.put(builderActor, singleWorkflowAction)
             builderActor ! singleWorkflowAction
diff --git a/services/src/main/scala/cromwell/services/metadata/impl/builder/MetadataBuilderActor.scala b/services/src/main/scala/cromwell/services/metadata/impl/builder/MetadataBuilderActor.scala
index cc77522811f..05e494680a3 100644
--- a/services/src/main/scala/cromwell/services/metadata/impl/builder/MetadataBuilderActor.scala
+++ b/services/src/main/scala/cromwell/services/metadata/impl/builder/MetadataBuilderActor.scala
@@ -278,6 +278,8 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props, metadataReadRow
       allDone()
     case Event(MetadataLookupResponse(query, metadata), HasWorkData(target, originalRequest)) =>
       processMetadataResponse(query, metadata, target, originalRequest)
+    case Event(FetchFailedJobsMetadataLookupResponse(metadata), HasWorkData(target, originalRequest)) =>
+      processFailedJobsMetadataResponse(metadata, target, originalRequest)
     case Event(MetadataLookupFailedTooLargeResponse(query, metadataSizeRows), HasWorkData(target, originalRequest)) =>
       val metadataTooLargeNumberOfRowsException = new MetadataTooLargeNumberOfRowsException(query.workflowId, metadataSizeRows, metadataReadRowNumberSafetyThreshold)
       target ! FailedMetadataJsonResponse(originalRequest, metadataTooLargeNumberOfRowsException)
@@ -286,6 +288,10 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props, metadataReadRow
       val metadataTooLargeTimeoutException = new MetadataTooLargeTimeoutException(query.workflowId)
       target ! FailedMetadataJsonResponse(originalRequest, metadataTooLargeTimeoutException)
       allDone()
+    case Event(FetchFailedTasksTimeoutResponse(workflowId), HasWorkData(target, originalRequest)) =>
+      val metadataTooLargeTimeoutException = new MetadataTooLargeTimeoutException(workflowId)
+      target ! FailedMetadataJsonResponse(originalRequest, metadataTooLargeTimeoutException)
+      allDone()
     case Event(failure: MetadataServiceFailure, HasWorkData(target, originalRequest)) =>
       target ! FailedMetadataJsonResponse(originalRequest, failure.reason)
       allDone()
@@ -371,4 +377,11 @@ class MetadataBuilderActor(readMetadataWorkerMaker: () => Props, metadataReadRow
       buildAndStop(query, eventsList, Map.empty, target, originalRequest)
     }
   }
+
+  def processFailedJobsMetadataResponse(eventsList: Seq[MetadataEvent], target: ActorRef, originalRequest: BuildMetadataJsonAction) = {
+    val groupedEvents = groupEvents(eventsList)
+    val res = MetadataBuilderActor.parse(groupedEvents, Map.empty)
+    target ! SuccessfulMetadataJsonResponse(originalRequest, res)
+    allDone()
+  }
 }
diff --git a/services/src/test/scala/cromwell/services/database/MetadataSlickDatabaseSpec.scala b/services/src/test/scala/cromwell/services/database/MetadataSlickDatabaseSpec.scala
index 3a8a320e1d2..86167977050 100644
--- a/services/src/test/scala/cromwell/services/database/MetadataSlickDatabaseSpec.scala
+++ b/services/src/test/scala/cromwell/services/database/MetadataSlickDatabaseSpec.scala
@@ -1,7 +1,6 @@
 package cromwell.services.database
 
 import java.time.OffsetDateTime
-
 import com.dimafeng.testcontainers.Container
 import common.assertion.CromwellTimeoutSpec
 import cromwell.core.Tags.DbmsTest
@@ -17,8 +16,9 @@ import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import javax.sql.rowset.serial.SerialClob
 import scala.concurrent.duration._
-import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
 import scala.language.postfixOps
 
 class MetadataSlickDatabaseSpec extends AnyFlatSpec with CromwellTimeoutSpec with Matchers with ScalaFutures {
@@ -46,6 +46,15 @@ class MetadataSlickDatabaseSpec extends AnyFlatSpec with CromwellTimeoutSpec wit
 
     val subSubWorkflowCountableId = "subsubworkflow id: countable stuff"
 
+    val failedParentWorkflowId = "bbf4c25b-282b-4a18-a914-441f9684b69e"
+    val successfulParentWorkflowId = "4c1cf43d-1fbd-47af-944c-c63216f293ae"
+
+    val failedChildWorkflowId = "9ff3d855-0585-48e4-b3a1-189101f611e5"
+    val successfulChildWorkflowId = "73886096-2e06-48f6-ba42-f365dbf23de5"
+
+    val failedStatusMetadataValue = Option(new SerialClob("Failed".toCharArray()))
+    val doneStatusMetadataValue = Option(new SerialClob("Done".toCharArray()))
+
     it should "set up the test data" taggedAs DbmsTest in {
       database.runTestTransaction(
         database.dataAccess.metadataEntries ++= Seq(
@@ -116,7 +125,7 @@ class MetadataSlickDatabaseSpec extends AnyFlatSpec with CromwellTimeoutSpec wit
 
           WorkflowMetadataSummaryEntry(rootCountableId, Option("workflow name"), Option("Succeeded"), Option(now), Option(now), Option(now), None, None, None, None),
           WorkflowMetadataSummaryEntry(subWorkflowCountableId, Option("subworkflow name"), Option("Succeeded"), Option(now), Option(now), Option(now), Option(rootCountableId), Option(rootCountableId), None, None),
-          WorkflowMetadataSummaryEntry(subSubWorkflowCountableId, Option("subsubworkflow name"), Option("Succeeded"), Option(now), Option(now), Option(now), Option(subWorkflowCountableId), Option(rootCountableId), None, None)
+          WorkflowMetadataSummaryEntry(subSubWorkflowCountableId, Option("subsubworkflow name"), Option("Succeeded"), Option(now), Option(now), Option(now), Option(subWorkflowCountableId), Option(rootCountableId), None, None),
         )
       ).futureValue(Timeout(10.seconds))
     }
@@ -198,6 +207,116 @@ class MetadataSlickDatabaseSpec extends AnyFlatSpec with CromwellTimeoutSpec wit
       }
     }
 
+    it should "fetch failed tasks from a failed workflow" taggedAs DbmsTest in {
+      database.runTestTransaction(
+        database.dataAccess.metadataEntries ++= Seq(
+          //Failed parent workflow calls (successful calls and older attempts and runs mixed in for negative checks)
+          MetadataEntry(failedParentWorkflowId, Option("failedWorkflowCall"), Option(0), Option(0), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedParentWorkflowId, Option("failedWorkflowCall"), Option(0), Option(1), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedParentWorkflowId, Option("failedWorkflowCall"), Option(1), Option(0), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedParentWorkflowId, Option("failedWorkflowCall"), Option(1), Option(1), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedParentWorkflowId, Option("successfulWorkflowCall"), Option(0), Option(0), "executionStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedParentWorkflowId, Option("successfulSubWorkflowCall"), Option(0), Option(0), "subWorkflowId", Option(new SerialClob(failedChildWorkflowId.toCharArray)), Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+
+          //child workflow calls (successful calls and previous failed attempts/shards are mixed in for negative checks)
+          //notFailedSubWorkflowCall should not be returned since it succeeded on the last attempt and has no scatters
+          MetadataEntry(failedChildWorkflowId, Option("notActuallyFailedSubWorkflowCall"), None, Option(1), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("notActuallyFailedSubWorkflowCall"), None, Option(2), "backendStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("successfulSubWorkflowCall"), Option(0), Option(0), "executionStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+
+          //Failed child workflow calls (successful calls and previous failed attempts/shards are mixed in for negative checks)
+          //failedSubWorkflowCall2 should be returned since it never succeeded
+          MetadataEntry(failedChildWorkflowId, Option("failedSubWorkflowCall"), None, Option(1), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("failedSubWorkflowCall"), None, Option(2), "backendStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("successfulSubWorkflowCall2"), Option(0), Option(0), "executionStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+
+          //Third set of child workflow calls, similar to above however this set consists of retries and scatters
+          //It's safe to assume that if one scatter fails then they all fail, so pull the last scatter on the last attempt
+          //failedSubWorkflowCall2 should return since the scatters failed on the last attempt
+          MetadataEntry(failedChildWorkflowId, Option("failedSubWorkflowCall2"), Option(1), Option(1), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("failedSubWorkflowCall2"), Option(2), Option(1), "backendStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("failedSubWorkflowCall2"), Option(1), Option(2), "backendStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("failedSubWorkflowCall2"), Option(2), Option(2), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("successfulSubWorkflowCall3"), Option(0), Option(0), "executionStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+
+          //Fourth set of child workflow calls
+          //This set should not return anything since the scatters succeeded on the last attempt
+          MetadataEntry(failedChildWorkflowId, Option("notActuallySubWorkflowCall2"), Option(1), Option(1), "executionStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("notActuallySubWorkflowCall2"), Option(2), Option(1), "backendStatus", failedStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("notActuallySubWorkflowCall2"), Option(1), Option(2), "executionStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("notActuallySubWorkflowCall2"), Option(2), Option(2), "backendStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(failedChildWorkflowId, Option("successfulSubWorkflowCall4"), Option(0), Option(0), "executionStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+
+          //Successful parent workflow call (negative check)
+          MetadataEntry(successfulParentWorkflowId, Option("successfulWorkflowCall"), Option(0), Option(0), "executionStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+
+          //Successful child workflow calls (negative check)
+          MetadataEntry(successfulChildWorkflowId, Option("successfulSubWorkflowCall"), Option(0), Option(0), "executionStatus", doneStatusMetadataValue, Option("String"), OffsetDateTime.now().toSystemTimestamp, None),
+          MetadataEntry(successfulChildWorkflowId, Option("successfulSubWorkflowCall"), Option(0), Option(0), "subWorkflowId", Option(new SerialClob(successfulParentWorkflowId.toCharArray)), Option("String"), OffsetDateTime.now().toSystemTimestamp, None)
+        )
+      ).futureValue(Timeout(10.seconds))
+
+      database.runTestTransaction(
+        database.dataAccess.workflowMetadataSummaryEntries ++= Seq(
+          //Failed WorkflowMetadataSummaryEntry setup
+          WorkflowMetadataSummaryEntry(failedParentWorkflowId, Option("failedParentWorkflow"), Option("Failed"), Option(now), Option(now), Option(now), None, None, None, None),
+          WorkflowMetadataSummaryEntry(failedChildWorkflowId, Option("failedChildWorkflow"), Option("Failed"), Option(now), Option(now), Option(now), Option(failedParentWorkflowId), Option(failedParentWorkflowId), None, None),
+          WorkflowMetadataSummaryEntry(successfulParentWorkflowId, Option("successfulParentWorkflow"), Option("Succeeded"), Option(now), Option(now), Option(now), None, None, None, None),
+          WorkflowMetadataSummaryEntry(successfulChildWorkflowId, Option("successfulChildWorkflow"), Option("Succeeded"), Option(now), Option(now), Option(now), Option(successfulParentWorkflowId), Option(successfulParentWorkflowId), None, None)
+        )
+      ).futureValue(Timeout(10.seconds))
+
+      val futureEntries: Future[Seq[MetadataEntry]] = database.getFailedJobsMetadataWithWorkflowId(failedParentWorkflowId)
+      var entriesFound = false
+
+      val recordCount = Map(
+        failedParentWorkflowId -> scala.collection.mutable.Map(
+          "expected" -> 1,
+          "actual" -> 0
+        ),
+        failedChildWorkflowId -> scala.collection.mutable.Map(
+          "expected" -> 2,
+          "actual" -> 0
+        )
+      )
+
+      whenReady(futureEntries, timeout(scaled(5 seconds))) {
+        entries =>
+          entries.foreach(entry => {
+            entriesFound = true
+            val workflowId = entry.workflowExecutionUuid
+            recordCount.getOrElse(workflowId, None) should not be(None)
+            recordCount(workflowId)("actual") += 1
+            val metadataValueClob = entry.metadataValue.get
+            val metadataValueString = metadataValueClob.getSubString(1, metadataValueClob.length().toInt)
+            metadataValueString should be("Failed")
+            entry.metadataKey should not be("subWorkflowId")
+            entry.callFullyQualifiedName.getOrElse("") match {
+              case "failedWorkflowCall" => {
+                entry.jobIndex.get should be(1)
+                entry.jobAttempt.get should be(1)
+                entry.metadataKey should be("executionStatus")
+              }
+              case "failedSubWorkflowCall" => {
+                entry.jobIndex should be(None)
+                entry.jobAttempt.get should be(2)
+                entry.metadataKey should be("backendStatus")
+              }
+              case "failedSubWorkflowCall2" => {
+                entry.jobIndex.get should be(2)
+                entry.jobAttempt.get should be(2)
+                entry.metadataKey should be("executionStatus")
+              }
+              case _ => fail(s"Entry ${entry.callFullyQualifiedName.getOrElse("N/A")} | Index: ${entry.jobIndex.get} | Attempt: ${entry.jobAttempt.get} should not be in result set")
+            }
+          })
+          entriesFound should be(true)
+          recordCount.foreach(record => {
+            record._2("actual") should be(record._2("expected"))
+          })
+      }
+    }
+
     it should "clean up & close the database" taggedAs DbmsTest in {
       // Not relevant in Travis where all state gets nuked but useful for testing locally
       database.runTestTransaction(database.dataAccess.metadataEntries.delete).futureValue(Timeout(10.seconds))
diff --git a/services/src/test/scala/cromwell/services/metadata/impl/WriteMetadataActorSpec.scala b/services/src/test/scala/cromwell/services/metadata/impl/WriteMetadataActorSpec.scala
index 0df5efab18b..6ed7fa6c402 100644
--- a/services/src/test/scala/cromwell/services/metadata/impl/WriteMetadataActorSpec.scala
+++ b/services/src/test/scala/cromwell/services/metadata/impl/WriteMetadataActorSpec.scala
@@ -308,6 +308,8 @@ class WriteMetadataActorSpec extends TestKitSuite with AnyFlatSpecLike with Matc
     override def countWorkflowsLeftToDeleteThatEndedOnOrBeforeThresholdTimestamp(workflowEndTimestampThreshold: Timestamp)(implicit ec: ExecutionContext): Future[Int] = notImplemented()
 
     override def getMetadataTableSizeInformation()(implicit ec: ExecutionContext): Future[Option[InformationSchemaEntry]] = notImplemented()
+
+    override def getFailedJobsMetadataWithWorkflowId(rootWorkflowId: String)(implicit ec: ExecutionContext): Future[Vector[MetadataEntry]] = notImplemented()
   }
 }