Repository: AbsaOSS/pramen (language: python)

Issue:
Add more info to `TaskStatus` for notifications
## Background
There are some bits of information that is present in the pipeline state, but missing for notification targets, including:
- `val isFailure: Boolean`
- maybe more coming... 

## Feature
Add the missing information to the `TaskStatus` model objects so that notification targets other then email have access to the same info as email notification

## Example
--
## Proposed Solution
As an additional suggestion - add access to pipeline state via `PipelineState` trait if client includes `pramen-core` as a dependency, not just `pramen-api`



Discussion / hints:
I would also add:
- pipelineName
- environmentName
from PipelineNotification to WebhookNotificationTarget
Great, will add
++ There is difference in Skipped status between TaskStatus and RunStatus. Missing isWarning
Also missing notification warning/errors ... I am thinking is there a difference between classes and why not have have the same classes for both

Fix patch:
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/NotificationTarget.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/NotificationTarget.scala
index 65fb34866..b435ac67a 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/NotificationTarget.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/NotificationTarget.scala
@@ -16,7 +16,9 @@
 
 package za.co.absa.pramen.api
 
+import za.co.absa.pramen.api.status.TaskResult
+
 trait NotificationTarget extends ExternalChannel {
   /** Sends a notification after completion of a task. */
-  def sendNotification(notification: TaskNotification): Unit
+  def sendNotification(pipelineInfo: PipelineInfo, notification: TaskResult): Unit
 }
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala
new file mode 100644
index 000000000..12fae8755
--- /dev/null
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.api
+
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, RuntimeInfo}
+
+import java.time.Instant
+
+case class PipelineInfo(
+                         pipelineName: String,
+                         environment: String,
+                         runtimeInfo: RuntimeInfo,
+                         startedAt: Instant,
+                         finishedAt: Option[Instant],
+                         sparkApplicationId: Option[String],
+                         failureException: Option[Throwable],
+                         pipelineNotificationFailures: Seq[PipelineNotificationFailure]
+                       )
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineNotificationTarget.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineNotificationTarget.scala
index db46d1820..ae4e689e0 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineNotificationTarget.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineNotificationTarget.scala
@@ -16,12 +16,11 @@
 
 package za.co.absa.pramen.api
 
-import java.time.Instant
+import za.co.absa.pramen.api.status.{CustomNotification, TaskResult}
 
 trait PipelineNotificationTarget extends ExternalChannel {
   /** Sends a notification after completion of the pipeline. */
-  def sendNotification(pipelineStarted: Instant,
-                       applicationId: Option[String],
-                       appException: Option[Throwable],
-                       tasksCompleted: Seq[TaskNotification]): Unit
+  def sendNotification(pipelineInfo: PipelineInfo,
+                       tasksCompleted: Seq[TaskResult],
+                       customNotification: CustomNotification): Unit
 }
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala
index 6d6703ed0..2d416c70e 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala
@@ -19,6 +19,7 @@ package za.co.absa.pramen.api
 import com.typesafe.config.Config
 import za.co.absa.pramen.api.app.PramenFactory
 import za.co.absa.pramen.api.common.{BuildPropertiesRetriever, FactoryLoader}
+import za.co.absa.pramen.api.status.{PipelineStateSnapshot, TaskResult}
 
 /**
   * Pramen provides additional features via a client that you can access like this:
@@ -36,6 +37,12 @@ trait Pramen {
   /** This gives access to the current workflow configuration. */
   def workflowConfig: Config
 
+  /** General information about the running pipeline. */
+  def pipelineInfo: PipelineInfo
+
+  /** The current pipeline state. */
+  def pipelineState: PipelineStateSnapshot
+
   /** Gets the notification builder that you can use to add custom information to email notifications. */
   def notificationBuilder: NotificationBuilder
 
@@ -49,7 +56,7 @@ trait Pramen {
   /**
     * Returns the list of tasks completed at the current moment of time.
     */
-  def getCompletedTasks: Seq[TaskNotification]
+  def getCompletedTasks: Seq[TaskResult]
 }
 
 object Pramen {
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/TaskNotification.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/TaskNotification.scala
deleted file mode 100644
index be41bc725..000000000
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/TaskNotification.scala
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Copyright 2022 ABSA Group Limited
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package za.co.absa.pramen.api
-
-import java.time.{Instant, LocalDate}
-
-case class TaskNotification(
-                             tableName: String,
-                             tableDef: MetaTableDef,
-                             infoDate: Option[LocalDate],
-                             started: Option[Instant],
-                             finished: Option[Instant],
-                             status: TaskStatus,
-                             applicationId: String,
-                             isTransient: Boolean,
-                             isRawFilesJob: Boolean,
-                             schemaChanges: Seq[SchemaDifference],
-                             dependencyWarningTables: Seq[String],
-                             options: Map[String, String]
-                           )
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/TaskStatus.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/TaskStatus.scala
deleted file mode 100644
index 3ccd3ae1c..000000000
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/TaskStatus.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Copyright 2022 ABSA Group Limited
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package za.co.absa.pramen.api
-
-sealed trait TaskStatus
-
-object TaskStatus {
-  case class Succeeded(recordCount: Long,
-                       filesRead: Seq[String],
-                       filesWritten: Seq[String],
-                       hiveTablesUpdated: Seq[String],
-                       warnings: Seq[String]) extends TaskStatus
-
-  case class ValidationFailed(ex: Throwable) extends TaskStatus
-
-  case class Failed(ex: Throwable) extends TaskStatus
-
-  case class MissingDependencies(tables: Seq[String]) extends TaskStatus
-
-  case class FailedDependencies(tables: Seq[String]) extends TaskStatus
-
-  case object NoData extends TaskStatus
-
-  case class InsufficientData(actual: Long, expected: Long) extends TaskStatus
-
-  case class Skipped(msg: String) extends TaskStatus
-}
-
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/status/CustomNotification.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/CustomNotification.scala
new file mode 100644
index 000000000..264a73a00
--- /dev/null
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/CustomNotification.scala
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.api.status
+
+import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
+
+case class CustomNotification(
+                               customEntries: Seq[NotificationEntry],
+                               customSignature: Seq[TextElement]
+                             )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/DependencyFailure.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/DependencyFailure.scala
similarity index 92%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/DependencyFailure.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/DependencyFailure.scala
index 2e48d9dcf..c3edf36a0 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/DependencyFailure.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/DependencyFailure.scala
@@ -14,9 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.pipeline
-
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
+package za.co.absa.pramen.api.status
 
 case class DependencyFailure(
                               dep: MetastoreDependency,
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/DependencyWarning.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/DependencyWarning.scala
similarity index 94%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/DependencyWarning.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/DependencyWarning.scala
index 65190f924..a1cd03d29 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/DependencyWarning.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/DependencyWarning.scala
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.pipeline
+package za.co.absa.pramen.api.status
 
 case class DependencyWarning(
                               table: String
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/status/MetastoreDependency.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/MetastoreDependency.scala
new file mode 100644
index 000000000..7ea8c0973
--- /dev/null
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/MetastoreDependency.scala
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.api.status
+
+case class MetastoreDependency(
+                                tables: Seq[String],
+                                dateFromExpr: String,
+                                dateUntilExpr: Option[String],
+                                triggerUpdates: Boolean,
+                                isOptional: Boolean,
+                                isPassive: Boolean
+                              )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/NotificationFailure.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/NotificationFailure.scala
similarity index 95%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/NotificationFailure.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/NotificationFailure.scala
index d564044f2..8cdf7a3d9 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/NotificationFailure.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/NotificationFailure.scala
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.runner.task
+package za.co.absa.pramen.api.status
 
 import java.time.LocalDate
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/PipelineNotificationFailure.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineNotificationFailure.scala
similarity index 94%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/PipelineNotificationFailure.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineNotificationFailure.scala
index b5dee4088..59a500c1a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/PipelineNotificationFailure.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineNotificationFailure.scala
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.runner.task
+package za.co.absa.pramen.api.status
 
 case class PipelineNotificationFailure(
                                         notificationTarget: String,
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateSnapshot.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineStateSnapshot.scala
similarity index 70%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateSnapshot.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineStateSnapshot.scala
index 0984e8511..0f2941b1c 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateSnapshot.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineStateSnapshot.scala
@@ -14,15 +14,17 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.state
+package za.co.absa.pramen.api.status
 
-import za.co.absa.pramen.core.runner.task.TaskResult
+import za.co.absa.pramen.api.PipelineInfo
 
 case class PipelineStateSnapshot(
+                                  pipelineInfo: PipelineInfo,
                                   isFinished: Boolean,
                                   exitedNormally: Boolean,
                                   exitCode: Int,
                                   customShutdownHookCanRun: Boolean,
-                                  failureException: Option[Throwable] = None,
-                                  taskResults: Seq[TaskResult]
+                                  taskResults: Seq[TaskResult],
+                                  pipelineNotificationFailures: Seq[PipelineNotificationFailure],
+                                  customNotification: CustomNotification
                                 )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/RunInfo.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/RunInfo.scala
similarity index 94%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/RunInfo.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/RunInfo.scala
index fe2a4fcf6..ca9a3d749 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/RunInfo.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/RunInfo.scala
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.runner.task
+package za.co.absa.pramen.api.status
 
 import java.time.{Instant, LocalDate}
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/RunStatus.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/RunStatus.scala
similarity index 88%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/RunStatus.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/RunStatus.scala
index 2a8d7afda..15d579494 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/RunStatus.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/RunStatus.scala
@@ -14,15 +14,12 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.runner.task
-
-import za.co.absa.pramen.core.pipeline.{DependencyFailure, TaskRunReason}
-import za.co.absa.pramen.core.utils.JvmUtils.getShortExceptionDescription
+package za.co.absa.pramen.api.status
 
 sealed trait RunStatus {
   val isFailure: Boolean
 
-  def getReason(): Option[String]
+  def getReason: Option[String]
 }
 
 object RunStatus {
@@ -106,4 +103,17 @@ object RunStatus {
 
     override def getReason(): Option[String] = if (msg.isEmpty) None else Option(msg)
   }
+
+  def getShortExceptionDescription(ex: Throwable): String = {
+    if (ex.getCause == null) {
+      ex.getMessage
+    } else {
+      val cause = ex.getCause
+      if (cause.getCause == null) {
+        s"${ex.getMessage} (${ex.getCause.getMessage})"
+      } else {
+        s"${ex.getMessage} (${cause.getMessage} caused by ${cause.getCause.getMessage})"
+      }
+    }
+  }
 }
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/status/RuntimeInfo.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/RuntimeInfo.scala
new file mode 100644
index 000000000..556b956f6
--- /dev/null
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/RuntimeInfo.scala
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.api.status
+
+case class RuntimeInfo(
+                        isDryRun: Boolean = false,     // If true, the pipeline won't do any writes, just the list of jobs it would run
+                        isUndercover: Boolean = false, // If true, no bookkeeping will be done for the job
+                        minRps: Int = 0,               // Configured records per second that is considered bad if the actual rps is lower, ignored if 0
+                        goodRps: Int = 0               // Configured records per second that is considered very good if the actual rps is higher, ignored if 0
+                      )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskResult.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskResult.scala
similarity index 81%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskResult.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskResult.scala
index 6183a8fc5..d057abd16 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskResult.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskResult.scala
@@ -14,13 +14,13 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.runner.task
+package za.co.absa.pramen.api.status
 
-import za.co.absa.pramen.api.SchemaDifference
-import za.co.absa.pramen.core.pipeline.{DependencyWarning, Job}
+import za.co.absa.pramen.api.{MetaTableDef, SchemaDifference}
 
 case class TaskResult(
-                       job: Job,
+                       jobName: String,
+                       outputTable: MetaTableDef,
                        runStatus: RunStatus,
                        runInfo: Option[RunInfo],
                        applicationId: String,
@@ -28,5 +28,6 @@ case class TaskResult(
                        isRawFilesJob: Boolean,
                        schemaChanges: Seq[SchemaDifference],
                        dependencyWarnings: Seq[DependencyWarning],
-                       notificationTargetErrors: Seq[NotificationFailure]
+                       notificationTargetErrors: Seq[NotificationFailure],
+                       options: Map[String, String]
                      )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TaskRunReason.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskRunReason.scala
similarity index 96%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TaskRunReason.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskRunReason.scala
index 05599feb1..99b534f3d 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TaskRunReason.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskRunReason.scala
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.pipeline
+package za.co.absa.pramen.api.status
 
 sealed trait TaskRunReason
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/model/TaskStatus.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskStatus.scala
similarity index 98%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/model/TaskStatus.scala
rename to pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskStatus.scala
index edb0daf6a..03e446074 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/model/TaskStatus.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/TaskStatus.scala
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.model
+package za.co.absa.pramen.api.status
 
 sealed trait TaskStatus
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala
index a77fc4243..d0c04ff0e 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala
@@ -19,8 +19,9 @@ package za.co.absa.pramen.core
 import com.typesafe.config.Config
 import za.co.absa.pramen.api.app.PramenFactory
 import za.co.absa.pramen.api.common.BuildPropertiesRetriever
-import za.co.absa.pramen.api.{MetadataManager, NotificationBuilder, Pramen, TaskNotification}
-import za.co.absa.pramen.core.state.{NotificationBuilderImpl, PipelineState, PipelineStateImpl}
+import za.co.absa.pramen.api.status.{PipelineStateSnapshot, TaskResult}
+import za.co.absa.pramen.api.{MetadataManager, NotificationBuilder, PipelineInfo, Pramen}
+import za.co.absa.pramen.core.state.{NotificationBuilderImpl, PipelineState}
 import za.co.absa.pramen.core.utils.BuildPropertyUtils
 
 class PramenImpl extends Pramen {
@@ -38,23 +39,38 @@ class PramenImpl extends Pramen {
     throw new IllegalStateException("Workflow configuration is not available at the context.")
   )
 
+  override def pipelineInfo: PipelineInfo = {
+    val pipelineState = _pipelineState.getOrElse(
+      throw new IllegalStateException("Pipeline state is not available at the context.")
+    )
+
+    pipelineState.getState().pipelineInfo
+  }
+
+  override def pipelineState: PipelineStateSnapshot = {
+    val pipelineState = _pipelineState.getOrElse(
+      throw new IllegalStateException("Pipeline state is not available at the context.")
+    )
+
+    pipelineState.getState()
+  }
+
   override def notificationBuilder: NotificationBuilder = notificationBuilderImpl
 
   override def metadataManager: MetadataManager = _metadataManager.getOrElse(
     throw new IllegalStateException("Metadata manager is not available at the context.")
   )
 
-  override def getCompletedTasks: Seq[TaskNotification] = {
+  override def getCompletedTasks: Seq[TaskResult] = {
     val pipelineState = _pipelineState.getOrElse(
       throw new IllegalStateException("Pipeline state is not available at the context.")
     )
 
     val state = pipelineState.getState()
 
-    state.taskResults.flatMap(PipelineStateImpl.taskResultToTaskNotification)
+    state.taskResults
   }
 
-
   private[core] def setWorkflowConfig(config: Config): Unit = synchronized {
     _workflowConfig = Option(config)
   }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompleted.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompleted.scala
index 7cd066510..e857f71a8 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompleted.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompleted.scala
@@ -16,9 +16,9 @@
 
 package za.co.absa.pramen.core.journal.model
 
+import za.co.absa.pramen.api.status.RunStatus.Succeeded
+import za.co.absa.pramen.api.status.TaskResult
 import za.co.absa.pramen.core.pipeline.Task
-import za.co.absa.pramen.core.runner.task.RunStatus._
-import za.co.absa.pramen.core.runner.task.TaskResult
 
 import java.time.{Instant, LocalDate}
 
@@ -49,7 +49,7 @@ object TaskCompleted {
     val taskStarted = taskResult.runInfo.map(_.started.getEpochSecond).getOrElse(now)
     val taskFinished = taskResult.runInfo.map(_.finished.getEpochSecond).getOrElse(now)
     val status = taskResult.runStatus.toString
-    val failureReason = taskResult.runStatus.getReason()
+    val failureReason = taskResult.runStatus.getReason
 
     val (recordCountOld, inputRecordCount, outputRecordCount, sizeBytes) = taskResult.runStatus match {
       case s: Succeeded => (s.recordCountOld, s.recordCount, Some(s.recordCount), s.sizeBytes)
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala
index 1d528308b..0344ef4ef 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala
@@ -26,7 +26,6 @@ import za.co.absa.pramen.core.app.config.InfoDateConfig
 import za.co.absa.pramen.core.app.config.InfoDateConfig.DEFAULT_DATE_FORMAT
 import za.co.absa.pramen.core.app.config.RuntimeConfig.UNDERCOVER
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
-import za.co.absa.pramen.core.metastore.MetastoreImpl.getMetaTableDef
 import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.metastore.peristence.{MetastorePersistence, TransientJobManager}
 import za.co.absa.pramen.core.utils.ConfigUtils
@@ -205,7 +204,7 @@ class MetastoreImpl(appConfig: Config,
       override def getTableDef(tableName: String): MetaTableDef = {
         validateTable(tableName)
 
-        getMetaTableDef(metastore.getTableDef(tableName))
+        MetaTable.getMetaTableDef(metastore.getTableDef(tableName))
       }
 
       override def getTableRunInfo(tableName: String, infoDate: LocalDate): Option[MetaTableRunInfo] = {
@@ -250,20 +249,5 @@ object MetastoreImpl {
 
     new MetastoreImpl(conf, tableDefs, bookkeeper, metadataManager, isUndercover)
   }
-
-  private[core] def getMetaTableDef(table: MetaTable): MetaTableDef = {
-    MetaTableDef(
-      table.name,
-      table.description,
-      table.format,
-      table.infoDateColumn,
-      table.infoDateFormat,
-      table.hiveTable,
-      table.hivePath,
-      table.infoDateStart,
-      table.readOptions,
-      table.writeOptions
-    )
-  }
 }
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
index fedb52a45..3a3894c13 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.metastore.model
 
 import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.DataFormat
+import za.co.absa.pramen.api.{DataFormat, MetaTableDef}
 import za.co.absa.pramen.core.app.config.InfoDateConfig
 import za.co.absa.pramen.core.config.InfoDateOverride
 import za.co.absa.pramen.core.utils.{AlgorithmUtils, ConfigUtils}
@@ -157,4 +157,18 @@ object MetaTable {
       writeOptions)
   }
 
+  def getMetaTableDef(table: MetaTable): MetaTableDef = {
+    MetaTableDef(
+      table.name,
+      table.description,
+      table.format,
+      table.infoDateColumn,
+      table.infoDateFormat,
+      table.hiveTable,
+      table.hivePath,
+      table.infoDateStart,
+      table.readOptions,
+      table.writeOptions
+    )
+  }
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetastoreDependency.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetastoreDependencyFactory.scala
similarity index 83%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetastoreDependency.scala
rename to pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetastoreDependencyFactory.scala
index 7f74649a3..66f0944a2 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetastoreDependency.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetastoreDependencyFactory.scala
@@ -17,21 +17,13 @@
 package za.co.absa.pramen.core.metastore.model
 
 import com.typesafe.config.Config
+import za.co.absa.pramen.api.status.MetastoreDependency
 import za.co.absa.pramen.core.utils.ConfigUtils
 
 import scala.collection.JavaConverters._
 import scala.util.Try
 
-case class MetastoreDependency(
-                                tables: Seq[String],
-                                dateFromExpr: String,
-                                dateUntilExpr: Option[String],
-                                triggerUpdates: Boolean,
-                                isOptional: Boolean,
-                                isPassive: Boolean
-                              )
-
-object MetastoreDependency {
+object MetastoreDependencyFactory {
   val TABLES_KEY = "tables"
   val DATE_FROM_EXPR_KEY = "date.from"
   val DATE_UNTIL_EXPR_KEY = "date.to"
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/TransientJobManager.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/TransientJobManager.scala
index 0f26f275e..2a9fa089a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/TransientJobManager.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/TransientJobManager.scala
@@ -18,9 +18,10 @@ package za.co.absa.pramen.core.metastore.peristence
 
 import org.apache.spark.sql.{DataFrame, SparkSession}
 import org.slf4j.LoggerFactory
+import za.co.absa.pramen.api.status.RunStatus
 import za.co.absa.pramen.core.pipeline.Job
 import za.co.absa.pramen.core.runner.splitter.ScheduleStrategyUtils
-import za.co.absa.pramen.core.runner.task.{RunStatus, TaskRunner}
+import za.co.absa.pramen.core.runner.task.TaskRunner
 import za.co.absa.pramen.core.utils.{Emoji, TimeUtils}
 
 import java.time.{Instant, LocalDate}
@@ -226,7 +227,7 @@ object TransientJobManager {
           case _: RunStatus.Skipped           => spark.emptyDataFrame
           case RunStatus.ValidationFailed(ex) => throw new IllegalStateException(s"$jobStr validation failed.", ex)
           case RunStatus.Failed(ex)           => throw new IllegalStateException(s"$jobStr failed.", ex)
-          case runStatus                      => throw new IllegalStateException(s"$jobStr failed to run ${runStatus.getReason().map(a => s"($a)").getOrElse("")}.")
+          case runStatus                      => throw new IllegalStateException(s"$jobStr failed to run ${runStatus.getReason.map(a => s"($a)").getOrElse("")}.")
         }
       case None =>
         throw new IllegalStateException("Task runner is not set.")
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/HyperdriveNotificationTarget.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/HyperdriveNotificationTarget.scala
index 0b5010406..d7b1f1bae 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/HyperdriveNotificationTarget.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/HyperdriveNotificationTarget.scala
@@ -19,7 +19,8 @@ package za.co.absa.pramen.core.notify
 import com.typesafe.config.Config
 import org.apache.spark.sql.SparkSession
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.{ExternalChannelFactory, NotificationTarget, TaskNotification, TaskStatus}
+import za.co.absa.pramen.api.status.{RunStatus, TaskResult}
+import za.co.absa.pramen.api.{ExternalChannelFactory, NotificationTarget, PipelineInfo}
 import za.co.absa.pramen.core.notify.mq.{SingleMessageProducer, SingleMessageProducerKafka}
 import za.co.absa.pramen.core.utils.ConfigUtils
 import za.co.absa.pramen.core.utils.Emoji._
@@ -38,11 +39,11 @@ class HyperdriveNotificationTarget(conf: Config,
     // conditions are met to send a notification.
   }
 
-  override def sendNotification(notification: TaskNotification): Unit = {
+  override def sendNotification(pipelineInfo: PipelineInfo, notification: TaskResult): Unit = {
     if (notification.options.contains(TOKEN_KEY)) {
       val token = notification.options(TOKEN_KEY)
 
-      if (notification.status.isInstanceOf[TaskStatus.Succeeded]) {
+      if (notification.runStatus.isInstanceOf[RunStatus.Succeeded]) {
         log.info(s"Sending '$token' to the Hyperdrive Kafka topic: '$topic'...")
         producer.connect()
         producer.send(topic, token)
@@ -51,7 +52,7 @@ class HyperdriveNotificationTarget(conf: Config,
         log.info(s"Not sending '$token' to the Hyperdrive Kafka topic: '$topic' for the unsuccessful job...")
       }
     } else {
-      log.warn(s"$WARNING Token is not configured for ${notification.tableName}. Hyperdrive notification won't be sent. Please, set 'notification.$TOKEN_KEY' option for the job.")
+      log.warn(s"$WARNING Token is not configured for ${notification.outputTable.name}. Hyperdrive notification won't be sent. Please, set 'notification.$TOKEN_KEY' option for the job.")
     }
   }
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/NotificationTargetManager.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/NotificationTargetManager.scala
index 21af41c6d..bb4579062 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/NotificationTargetManager.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/NotificationTargetManager.scala
@@ -18,9 +18,9 @@ package za.co.absa.pramen.core.notify
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.SparkSession
-import za.co.absa.pramen.api.{NotificationTarget, TaskStatus}
+import za.co.absa.pramen.api.NotificationTarget
+import za.co.absa.pramen.api.status.TaskStatus
 import za.co.absa.pramen.core.ExternalChannelFactoryReflect
-import za.co.absa.pramen.core.runner.task.RunStatus
 
 object NotificationTargetManager {
   val NOTIFICATION_TARGETS_KEY = "pramen.notification.targets"
@@ -59,27 +59,4 @@ object NotificationTargetManager {
                (implicit spark: SparkSession): NotificationTarget = {
     ExternalChannelFactoryReflect.fromConfigByName[NotificationTarget](conf, overrideConf, NOTIFICATION_TARGETS_KEY, name, "notification target")
   }
-
-  /**
-    * Converts an internal run status to the API trait TaskStatus.
-    *
-    * Converts only statuses which corresponds to jobs actually attempted to run.
-    * Returns None otherwise.
-    *
-    * @param status The internal run status.
-    * @return The corresponding task status if applicable.
-    */
-  def runStatusToTaskStatus(status: RunStatus): Option[TaskStatus] = {
-    status match {
-      case s: RunStatus.Succeeded => Some(TaskStatus.Succeeded(s.recordCount, s.filesRead, s.filesWritten, s.hiveTablesUpdated, s.warnings))
-      case s: RunStatus.Failed => Some(TaskStatus.Failed(s.ex))
-      case s: RunStatus.ValidationFailed => Some(TaskStatus.ValidationFailed(s.ex))
-      case s: RunStatus.MissingDependencies => Some(TaskStatus.MissingDependencies(s.tables))
-      case s: RunStatus.FailedDependencies => Some(TaskStatus.FailedDependencies(s.failures.flatMap(d => d.failedTables ++ d.emptyTables).distinct.sorted))
-      case _: RunStatus.NoData => Some(TaskStatus.NoData)
-      case s: RunStatus.InsufficientData => Some(TaskStatus.InsufficientData(s.actual, s.expected))
-      case s: RunStatus.Skipped => Some(TaskStatus.Skipped(s.msg))
-      case _ => None
-    }
-  }
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala
index 3960c033b..00609b024 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala
@@ -17,7 +17,7 @@
 package za.co.absa.pramen.core.notify.pipeline
 
 import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
-import za.co.absa.pramen.core.runner.task.{PipelineNotificationFailure, TaskResult}
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, TaskResult}
 
 import java.time.Instant
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala
index a06cfb0cb..f1417c00e 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala
@@ -17,7 +17,7 @@
 package za.co.absa.pramen.core.notify.pipeline
 
 import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
-import za.co.absa.pramen.core.runner.task.{PipelineNotificationFailure, TaskResult}
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, TaskResult}
 
 import java.time.Instant
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala
index 836354784..b34fc970b 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala
@@ -18,15 +18,14 @@ package za.co.absa.pramen.core.notify.pipeline
 
 import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.{FieldChange, SchemaDifference}
 import za.co.absa.pramen.api.notification._
+import za.co.absa.pramen.api.status.RunStatus._
+import za.co.absa.pramen.api.status._
+import za.co.absa.pramen.api.{FieldChange, SchemaDifference}
 import za.co.absa.pramen.core.config.Keys.TIMEZONE
 import za.co.absa.pramen.core.exceptions.{CmdFailedException, ProcessFailedException}
 import za.co.absa.pramen.core.notify.message._
 import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilderHtml.{MIN_MEGABYTES, MIN_RPS_JOB_DURATION_SECONDS, MIN_RPS_RECORDS}
-import za.co.absa.pramen.core.pipeline.TaskRunReason
-import za.co.absa.pramen.core.runner.task.RunStatus._
-import za.co.absa.pramen.core.runner.task.{NotificationFailure, PipelineNotificationFailure, RunStatus, TaskResult}
 import za.co.absa.pramen.core.utils.JvmUtils.getShortExceptionDescription
 import za.co.absa.pramen.core.utils.StringUtils.renderThrowable
 import za.co.absa.pramen.core.utils.{BuildPropertyUtils, ConfigUtils, StringUtils, TimeUtils}
@@ -115,7 +114,6 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
 
   def renderSubject(): String = {
     val timeCreatedStr = ZonedDateTime.now(zoneId).format(timestampFmt)
-    val (someTasksSucceeded, someTasksFailed) = getSuccessFlags
 
     val dryRunStr = if (isDryRun) "(DRY RUN) " else ""
 
@@ -277,9 +275,9 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
   private[core] def renderJobException(builder: MessageBuilder, taskResult: TaskResult, ex: Throwable): MessageBuilder = {
     val paragraphBuilder = ParagraphBuilder()
       .withText("Job ", Style.Exception)
-      .withText(taskResult.job.name, Style.Error)
+      .withText(taskResult.jobName, Style.Error)
       .withText(" outputting to ", Style.Exception)
-      .withText(taskResult.job.outputTable.name, Style.Error)
+      .withText(taskResult.outputTable.name, Style.Error)
 
     taskResult.runInfo.foreach(info =>
       paragraphBuilder
@@ -351,7 +349,7 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
       t.isRawFilesJob || hasExplicitSize
     }
 
-    val haveReasonColumn = tasks.exists(t => t.runStatus.getReason().nonEmpty || t.dependencyWarnings.nonEmpty)
+    val haveReasonColumn = tasks.exists(t => t.runStatus.getReason.nonEmpty || t.dependencyWarnings.nonEmpty)
     val haveHiveColumn = tasks.exists(t => t.runStatus.isInstanceOf[Succeeded] && t.runStatus.asInstanceOf[Succeeded].hiveTablesUpdated.nonEmpty)
 
     val tableBuilder = new TableBuilderHtml
@@ -379,8 +377,8 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
     tasks.foreach(task => {
       val row = new ListBuffer[TextElement]
 
-      row.append(TextElement(task.job.name, getTransientTextStyle(task)))
-      row.append(TextElement(task.job.outputTable.name, getTransientTextStyle(task)))
+      row.append(TextElement(task.jobName, getTransientTextStyle(task)))
+      row.append(TextElement(task.outputTable.name, getTransientTextStyle(task)))
 
       if (haveHiveColumn) {
         val hiveTable = task.runStatus match {
@@ -465,7 +463,7 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
 
     val tableHeaders = new ListBuffer[TableHeader]
 
-    val taskName = s"Files sourced - ${task.job.outputTable.name} - ${task.runInfo.map(_.infoDate.toString).getOrElse(" ")}"
+    val taskName = s"Files sourced - ${task.outputTable.name} - ${task.runInfo.map(_.infoDate.toString).getOrElse(" ")}"
 
     tableHeaders.append(TableHeader(TextElement(taskName), Align.Left))
     tableBuilder.withHeaders(tableHeaders.toSeq)
@@ -596,7 +594,7 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
   }
 
   private[core] def getFailureReason(task: TaskResult): String = {
-    task.runStatus.getReason() match {
+    task.runStatus.getReason match {
       case Some(reason) => reason
       case None         =>
         if (task.dependencyWarnings.isEmpty) {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/IngestionJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/IngestionJob.scala
index f9dee5099..89108e669 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/IngestionJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/IngestionJob.scala
@@ -19,6 +19,7 @@ package za.co.absa.pramen.core.pipeline
 import com.typesafe.config.Config
 import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.{DataFrame, SparkSession}
+import za.co.absa.pramen.api.status.DependencyWarning
 import za.co.absa.pramen.api.{Query, Reason, Source, SourceResult}
 import za.co.absa.pramen.core.app.config.GeneralConfig.TEMPORARY_DIRECTORY_KEY
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala
index 90e75d3c3..cfebc5f75 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala
@@ -19,10 +19,11 @@ package za.co.absa.pramen.core.pipeline
 import com.typesafe.config.Config
 import org.apache.spark.sql.types.StructType
 import org.slf4j.{Logger, LoggerFactory}
+import za.co.absa.pramen.api.status.{DependencyFailure, DependencyWarning, MetastoreDependency}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.expr.DateExprEvaluator
 import za.co.absa.pramen.core.metastore.Metastore
-import za.co.absa.pramen.core.metastore.model.{MetaTable, MetastoreDependency}
+import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.pipeline
 import za.co.absa.pramen.core.utils.Emoji._
 
@@ -161,7 +162,7 @@ abstract class JobBase(operationDef: OperationDef,
     if (failedTables.isEmpty && emptyTables.isEmpty) {
       None
     } else {
-      Some(pipeline.DependencyFailure(dep, emptyTables, failedTables, failedDateRanges))
+      Some(DependencyFailure(dep, emptyTables, failedTables, failedDateRanges))
     }
   }
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobPreRunResult.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobPreRunResult.scala
index 231358a00..ea8222793 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobPreRunResult.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobPreRunResult.scala
@@ -16,6 +16,8 @@
 
 package za.co.absa.pramen.core.pipeline
 
+import za.co.absa.pramen.api.status.DependencyWarning
+
 case class JobPreRunResult(
                             status: JobPreRunStatus,
                             inputRecordsCount: Option[Long],
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobPreRunStatus.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobPreRunStatus.scala
index d840c9631..d13c936df 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobPreRunStatus.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobPreRunStatus.scala
@@ -16,6 +16,8 @@
 
 package za.co.absa.pramen.core.pipeline
 
+import za.co.absa.pramen.api.status.DependencyFailure
+
 trait JobPreRunStatus
 
 object JobPreRunStatus {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
index 089e80c8d..48e1a992a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
@@ -18,9 +18,10 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
+import za.co.absa.pramen.api.status.MetastoreDependency
 import za.co.absa.pramen.core.app.config.InfoDateConfig
 import za.co.absa.pramen.core.config.Keys
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
+import za.co.absa.pramen.core.metastore.model.MetastoreDependencyFactory
 import za.co.absa.pramen.core.schedule.Schedule
 import za.co.absa.pramen.core.utils.ConfigUtils
 
@@ -157,7 +158,7 @@ object OperationDef {
       val dependencyConfigs = conf.getConfigList(DEPENDENCIES_KEY)
       dependencyConfigs.asScala
         .zipWithIndex
-        .map { case (c, i) => MetastoreDependency.fromConfig(c, s"$parent[$i]") }
+        .map { case (c, i) => MetastoreDependencyFactory.fromConfig(c, s"$parent[$i]") }
         .toSeq
     } else {
       Seq.empty[MetastoreDependency]
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJob.scala
index c50cf2324..e12036534 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJob.scala
@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.{AnalysisException, DataFrame, SparkSession}
+import za.co.absa.pramen.api.status.DependencyWarning
 import za.co.absa.pramen.api.{DataFormat, Reason}
 import za.co.absa.pramen.core.app.config.GeneralConfig.TEMPORARY_DIRECTORY_KEY
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala
index bb2b17e85..3a691983c 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala
@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.{DataFrame, SparkSession}
+import za.co.absa.pramen.api.status.DependencyWarning
 import za.co.absa.pramen.api.{Reason, Sink}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.model.MetaTable
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Task.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Task.scala
index 53803992b..c48acc94d 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Task.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Task.scala
@@ -16,6 +16,8 @@
 
 package za.co.absa.pramen.core.pipeline
 
+import za.co.absa.pramen.api.status.TaskRunReason
+
 import java.time.LocalDate
 
 case class Task(
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TaskPreDef.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TaskPreDef.scala
index 7b470499e..9883a3056 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TaskPreDef.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TaskPreDef.scala
@@ -16,6 +16,8 @@
 
 package za.co.absa.pramen.core.pipeline
 
+import za.co.absa.pramen.api.status.TaskRunReason
+
 import java.time.LocalDate
 
 case class TaskPreDef(
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferJob.scala
index 0d104251e..f2c7182e4 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferJob.scala
@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.{DataFrame, SparkSession}
+import za.co.absa.pramen.api.status.DependencyWarning
 import za.co.absa.pramen.api.{Reason, Sink, Source}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.Metastore
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala
index 949e00df4..baad5c53a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala
@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.{DataFrame, SparkSession}
+import za.co.absa.pramen.api.status.DependencyWarning
 import za.co.absa.pramen.api.{Reason, Transformer}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.Metastore
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunner.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunner.scala
index 46108a93b..ca0a35413 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunner.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunner.scala
@@ -17,8 +17,8 @@
 package za.co.absa.pramen.core.runner.jobrunner
 
 import com.github.yruslan.channel.ReadChannel
+import za.co.absa.pramen.api.status.TaskResult
 import za.co.absa.pramen.core.pipeline.Job
-import za.co.absa.pramen.core.runner.task.TaskResult
 
 trait ConcurrentJobRunner {
   def startWorkerLoop(incomingJobs: ReadChannel[Job]): Unit
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala
index 2823ad794..4122bea5c 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala
@@ -19,14 +19,16 @@ package za.co.absa.pramen.core.runner.jobrunner
 import com.github.yruslan.channel.{Channel, ReadChannel}
 import org.slf4j.LoggerFactory
 import za.co.absa.pramen.api.DataFormat
+import za.co.absa.pramen.api.status.{RunStatus, TaskResult}
 import za.co.absa.pramen.core.app.config.RuntimeConfig
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.exceptions.FatalErrorWrapper
+import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.metastore.peristence.TransientJobManager
 import za.co.absa.pramen.core.pipeline.Job
 import za.co.absa.pramen.core.runner.jobrunner.ConcurrentJobRunner.JobRunResults
 import za.co.absa.pramen.core.runner.splitter.ScheduleParams
-import za.co.absa.pramen.core.runner.task.{RunStatus, TaskResult, TaskRunner}
+import za.co.absa.pramen.core.runner.task.TaskRunner
 import za.co.absa.pramen.core.utils.Emoji
 
 import java.util.concurrent.ExecutorService
@@ -104,8 +106,19 @@ class ConcurrentJobRunnerImpl(runtimeConfig: RuntimeConfig,
   }
 
   private[core] def sendFailure(ex: Throwable, job: Job, isTransient: Boolean): Unit = {
-    completedJobsChannel.send((job, TaskResult(job, RunStatus.Failed(ex), None, applicationId, isTransient,
-      job.outputTable.format.isInstanceOf[DataFormat.Raw], Nil, Nil, Nil) :: Nil, false))
+    completedJobsChannel.send((job,
+      TaskResult(job.name,
+        MetaTable.getMetaTableDef(job.outputTable),
+        RunStatus.Failed(ex),
+        None,
+        applicationId,
+        isTransient,
+        job.outputTable.format.isInstanceOf[DataFormat.Raw],
+        Nil,
+        Nil,
+        Nil,
+        job.operation.extraOptions
+      ) :: Nil, false))
   }
 
   private[core] def runJob(job: Job): Boolean = {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/orchestrator/OrchestratorImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/orchestrator/OrchestratorImpl.scala
index b7f65ba4c..cfbb4faa5 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/orchestrator/OrchestratorImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/orchestrator/OrchestratorImpl.scala
@@ -21,12 +21,13 @@ import com.typesafe.config.Config
 import org.apache.spark.sql.SparkSession
 import org.slf4j.LoggerFactory
 import za.co.absa.pramen.api.DataFormat
+import za.co.absa.pramen.api.status.{RunStatus, TaskResult}
 import za.co.absa.pramen.core.app.AppContext
 import za.co.absa.pramen.core.exceptions.{FatalErrorWrapper, ValidationException}
+import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.pipeline.{Job, JobDependency, OperationType}
 import za.co.absa.pramen.core.runner.jobrunner.ConcurrentJobRunner
 import za.co.absa.pramen.core.runner.splitter.ScheduleStrategyUtils.evaluateRunDate
-import za.co.absa.pramen.core.runner.task.{RunStatus, TaskResult}
 import za.co.absa.pramen.core.state.PipelineState
 import za.co.absa.pramen.core.utils.Emoji._
 
@@ -125,8 +126,19 @@ class OrchestratorImpl extends Orchestrator {
       val isTransient = job.outputTable.format.isTransient
       val isFailure = hasNonPassiveNonOptionalDeps(job, missingTables)
 
-      val taskResult = TaskResult(job, RunStatus.MissingDependencies(isFailure, missingTables), None, applicationId,
-        isTransient, job.outputTable.format.isInstanceOf[DataFormat.Raw], Nil, Nil, Nil)
+      val taskResult = TaskResult(
+        job.name,
+        MetaTable.getMetaTableDef(job.outputTable),
+        RunStatus.MissingDependencies(isFailure, missingTables),
+        None,
+        applicationId,
+        isTransient,
+        job.outputTable.format.isInstanceOf[DataFormat.Raw],
+        Nil,
+        Nil,
+        Nil,
+        job.operation.extraOptions
+      )
 
       state.addTaskCompletion(taskResult :: Nil)
     })
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategy.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategy.scala
index 3e9cc5b72..40baf99e6 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategy.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategy.scala
@@ -16,8 +16,8 @@
 
 package za.co.absa.pramen.core.runner.splitter
 
+import za.co.absa.pramen.api.status.MetastoreDependency
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
 import za.co.absa.pramen.core.pipeline.TaskPreDef
 import za.co.absa.pramen.core.schedule.Schedule
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategySourcing.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategySourcing.scala
index be9fcae53..29a7a9cc2 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategySourcing.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategySourcing.scala
@@ -16,10 +16,10 @@
 
 package za.co.absa.pramen.core.runner.splitter
 
+import za.co.absa.pramen.api.status.{MetastoreDependency, TaskRunReason}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
 import za.co.absa.pramen.core.pipeline
-import za.co.absa.pramen.core.pipeline.{TaskPreDef, TaskRunReason}
+import za.co.absa.pramen.core.pipeline.TaskPreDef
 import za.co.absa.pramen.core.runner.splitter.ScheduleStrategyUtils._
 import za.co.absa.pramen.core.schedule.Schedule
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategyTransformation.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategyTransformation.scala
index 1a89c7725..b67efe22a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategyTransformation.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategyTransformation.scala
@@ -16,10 +16,10 @@
 
 package za.co.absa.pramen.core.runner.splitter
 
+import za.co.absa.pramen.api.status.{MetastoreDependency, TaskRunReason}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
 import za.co.absa.pramen.core.pipeline
-import za.co.absa.pramen.core.pipeline.{TaskPreDef, TaskRunReason}
+import za.co.absa.pramen.core.pipeline.TaskPreDef
 import za.co.absa.pramen.core.runner.splitter.ScheduleStrategyUtils._
 import za.co.absa.pramen.core.schedule.Schedule
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategyUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategyUtils.scala
index 1e967384f..a62c300dd 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategyUtils.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/splitter/ScheduleStrategyUtils.scala
@@ -16,11 +16,11 @@
 
 package za.co.absa.pramen.core.runner.splitter
 
+import za.co.absa.pramen.api.status.{MetastoreDependency, TaskRunReason}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.expr.DateExprEvaluator
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
 import za.co.absa.pramen.core.pipeline
-import za.co.absa.pramen.core.pipeline.{TaskPreDef, TaskRunReason}
+import za.co.absa.pramen.core.pipeline.TaskPreDef
 import za.co.absa.pramen.core.schedule.Schedule
 
 import java.time.LocalDate
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunner.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunner.scala
index a0fc49e8b..50fe2f14b 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunner.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunner.scala
@@ -16,6 +16,7 @@
 
 package za.co.absa.pramen.core.runner.task
 
+import za.co.absa.pramen.api.status.RunStatus
 import za.co.absa.pramen.core.pipeline.{Job, TaskPreDef}
 
 import java.time.LocalDate
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala
index 8853a8216..01fece282 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala
@@ -20,16 +20,16 @@ import com.typesafe.config.Config
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.lit
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.{DataFormat, Reason, SchemaDifference, TaskNotification}
+import za.co.absa.pramen.api._
+import za.co.absa.pramen.api.status._
 import za.co.absa.pramen.core.app.config.RuntimeConfig
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.exceptions.{FatalErrorWrapper, ReasonException}
 import za.co.absa.pramen.core.journal.Journal
 import za.co.absa.pramen.core.journal.model.TaskCompleted
 import za.co.absa.pramen.core.lock.TokenLockFactory
+import za.co.absa.pramen.core.metastore.MetaTableStats
 import za.co.absa.pramen.core.metastore.model.MetaTable
-import za.co.absa.pramen.core.metastore.{MetaTableStats, MetastoreImpl}
-import za.co.absa.pramen.core.notify.NotificationTargetManager
 import za.co.absa.pramen.core.pipeline.JobPreRunStatus._
 import za.co.absa.pramen.core.pipeline._
 import za.co.absa.pramen.core.state.PipelineState
@@ -102,7 +102,7 @@ abstract class TaskRunnerBase(conf: Config,
       failedInfoDate match {
         case Some(failedDate) =>
           skipTask(task, s"Due to failure for $failedDate", isWarning = true)
-        case None             =>
+        case None =>
           val status = runTask(task)
           if (status.isFailure)
             failedInfoDate = Option(task.infoDate)
@@ -134,7 +134,18 @@ abstract class TaskRunnerBase(conf: Config,
     val runStatus = RunStatus.Skipped(reason, isWarning)
     val runInfo = RunInfo(task.infoDate, now, now)
     val isTransient = task.job.outputTable.format.isTransient
-    val taskResult = TaskResult(task.job, runStatus, Some(runInfo), applicationId, isTransient, isRawFilesJob = false, Nil, Nil, Nil)
+    val taskResult = TaskResult(
+      task.job.name,
+      MetaTable.getMetaTableDef(task.job.outputTable),
+      runStatus,
+      Option(runInfo),
+      applicationId,
+      isTransient,
+      isRawFilesJob = false,
+      Nil,
+      Nil,
+      Nil,
+      task.job.operation.extraOptions)
 
     onTaskCompletion(task, taskResult, isLazy = false)
   }
@@ -150,7 +161,10 @@ abstract class TaskRunnerBase(conf: Config,
     * @return an instance of TaskResult on the check failure or optional record count on success.
     */
   private[core] def preRunCheck(task: Task, started: Instant): Either[TaskResult, JobPreRunResult] = {
-    val outputTable = task.job.outputTable.name
+    val jobName = task.job.name
+    val outputTable = MetaTable.getMetaTableDef(task.job.outputTable)
+    val outputTableName = task.job.outputTable.name
+    val options = task.job.operation.extraOptions
     val isTransient = task.job.outputTable.format.isTransient
     val isRawFileBased = task.job.outputTable.format.isInstanceOf[DataFormat.Raw]
 
@@ -160,38 +174,38 @@ abstract class TaskRunnerBase(conf: Config,
       case Success(validationResult) =>
         val resultToReturn = validationResult.status match {
           case Ready =>
-            log.info(s"Validation of the task: $outputTable for date: ${task.infoDate} has SUCCEEDED.")
+            log.info(s"Validation of the task: $outputTableName for date: ${task.infoDate} has SUCCEEDED.")
             Right(validationResult)
           case NeedsUpdate =>
-            log.info(s"The table needs update: $outputTable for date: ${task.infoDate}.")
+            log.info(s"The table needs update: $outputTableName for date: ${task.infoDate}.")
             Right(validationResult)
           case NoData(isFailure) =>
-            log.info(s"NO DATA available for the task: $outputTable for date: ${task.infoDate}.")
-            Left(TaskResult(task.job, RunStatus.NoData(isFailure), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, validationResult.dependencyWarnings, Nil))
+            log.info(s"NO DATA available for the task: $outputTableName for date: ${task.infoDate}.")
+            Left(TaskResult(jobName, outputTable, RunStatus.NoData(isFailure), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, validationResult.dependencyWarnings, Nil, options))
           case InsufficientData(actual, expected, oldRecordCount) =>
-            log.info(s"INSUFFICIENT DATA available for the task: $outputTable for date: ${task.infoDate}. Expected = $expected, actual = $actual")
-            Left(TaskResult(task.job, RunStatus.InsufficientData(actual, expected, oldRecordCount), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, validationResult.dependencyWarnings, Nil))
+            log.info(s"INSUFFICIENT DATA available for the task: $outputTableName for date: ${task.infoDate}. Expected = $expected, actual = $actual")
+            Left(TaskResult(jobName, outputTable, RunStatus.InsufficientData(actual, expected, oldRecordCount), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, validationResult.dependencyWarnings, Nil, options))
           case AlreadyRan =>
             if (runtimeConfig.isRerun) {
-              log.info(s"RE-RUNNING the task: $outputTable for date: ${task.infoDate}.")
+              log.info(s"RE-RUNNING the task: $outputTableName for date: ${task.infoDate}.")
               Right(validationResult)
             } else {
-              log.info(s"SKIPPING already ran job: $outputTable for date: ${task.infoDate}.")
-              Left(TaskResult(task.job, RunStatus.NotRan, getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, validationResult.dependencyWarnings, Nil))
+              log.info(s"SKIPPING already ran job: $outputTableName for date: ${task.infoDate}.")
+              Left(TaskResult(jobName, outputTable, RunStatus.NotRan, getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, validationResult.dependencyWarnings, Nil, options))
             }
           case Skip(msg) =>
-            log.info(s"SKIPPING job: $outputTable for date: ${task.infoDate}. Reason: msg")
-            Left(TaskResult(task.job, RunStatus.Skipped(msg), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, validationResult.dependencyWarnings, Nil))
+            log.info(s"SKIPPING job: $outputTableName for date: ${task.infoDate}. Reason: msg")
+            Left(TaskResult(jobName, outputTable, RunStatus.Skipped(msg), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, validationResult.dependencyWarnings, Nil, options))
           case FailedDependencies(isFailure, failures) =>
-            Left(TaskResult(task.job, RunStatus.FailedDependencies(isFailure, failures), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, Nil, Nil))
+            Left(TaskResult(jobName, outputTable, RunStatus.FailedDependencies(isFailure, failures), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, Nil, Nil, options))
         }
         if (validationResult.dependencyWarnings.nonEmpty) {
-          log.warn(s"$WARNING Validation of the task: $outputTable for date: ${task.infoDate} has " +
+          log.warn(s"$WARNING Validation of the task: $outputTableName for date: ${task.infoDate} has " +
             s"optional dependency failure(s) for table(s): ${validationResult.dependencyWarnings.map(_.table).mkString(", ")} ")
         }
         resultToReturn
       case Failure(ex) =>
-        Left(TaskResult(task.job, RunStatus.ValidationFailed(ex), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, Nil, Nil))
+        Left(TaskResult(jobName, outputTable, RunStatus.ValidationFailed(ex), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, Nil, Nil, options))
     }
   }
 
@@ -206,7 +220,10 @@ abstract class TaskRunnerBase(conf: Config,
     * @return an instance of TaskResult on validation failure or optional record count on success.
     */
   private[core] def validate(task: Task, started: Instant): Either[TaskResult, JobPreRunResult] = {
-    val outputTable = task.job.outputTable.name
+    val jobName = task.job.name
+    val outputTable = MetaTable.getMetaTableDef(task.job.outputTable)
+    val outputTableName = task.job.outputTable.name
+    val options = task.job.operation.extraOptions
     val isTransient = task.job.outputTable.format.isTransient
     val isRawFileBased = task.job.outputTable.format.isInstanceOf[DataFormat.Raw]
 
@@ -220,27 +237,27 @@ abstract class TaskRunnerBase(conf: Config,
           case Success(validationResult) =>
             validationResult match {
               case Reason.Ready =>
-                log.info(s"VALIDATION is SUCCESSFUL for the task: $outputTable for date: ${task.infoDate}.")
+                log.info(s"VALIDATION is SUCCESSFUL for the task: $outputTableName for date: ${task.infoDate}.")
                 Right(status)
               case reason: Reason.Warning =>
-                log.info(s"VALIDATION is SUCCESSFUL with WARNINGS for the task: $outputTable for date: ${task.infoDate}.")
+                log.info(s"VALIDATION is SUCCESSFUL with WARNINGS for the task: $outputTableName for date: ${task.infoDate}.")
                 Right(status.copy(warnings = reason.warnings))
               case Reason.NotReady(msg) =>
-                log.info(s"NOT READY validation failure for the task: $outputTable for date: ${task.infoDate}. Reason: $msg")
-                Left(TaskResult(task.job, RunStatus.ValidationFailed(new ReasonException(Reason.NotReady(msg), msg)), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, status.dependencyWarnings, Nil))
+                log.info(s"NOT READY validation failure for the task: $outputTableName for date: ${task.infoDate}. Reason: $msg")
+                Left(TaskResult(jobName, outputTable, RunStatus.ValidationFailed(new ReasonException(Reason.NotReady(msg), msg)), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, status.dependencyWarnings, Nil, Map.empty))
               case Reason.Skip(msg) =>
-                log.info(s"SKIP validation failure for the task: $outputTable for date: ${task.infoDate}. Reason: $msg")
-                if (bookkeeper.getLatestDataChunk(outputTable, task.infoDate, task.infoDate).isEmpty) {
+                log.info(s"SKIP validation failure for the task: $outputTableName for date: ${task.infoDate}. Reason: $msg")
+                if (bookkeeper.getLatestDataChunk(outputTableName, task.infoDate, task.infoDate).isEmpty) {
                   val isTransient = task.job.outputTable.format.isTransient
-                  bookkeeper.setRecordCount(outputTable, task.infoDate, task.infoDate, task.infoDate, status.inputRecordsCount.getOrElse(0L), 0, started.getEpochSecond, Instant.now().getEpochSecond, isTransient)
+                  bookkeeper.setRecordCount(outputTableName, task.infoDate, task.infoDate, task.infoDate, status.inputRecordsCount.getOrElse(0L), 0, started.getEpochSecond, Instant.now().getEpochSecond, isTransient)
                 }
-                Left(TaskResult(task.job, RunStatus.Skipped(msg), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, status.dependencyWarnings, Nil))
+                Left(TaskResult(jobName, outputTable, RunStatus.Skipped(msg), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, status.dependencyWarnings, Nil, options))
               case Reason.SkipOnce(msg) =>
-                log.info(s"SKIP today validation failure for the task: $outputTable for date: ${task.infoDate}. Reason: $msg")
-                Left(TaskResult(task.job, RunStatus.Skipped(msg), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, status.dependencyWarnings, Nil))
+                log.info(s"SKIP today validation failure for the task: $outputTableName for date: ${task.infoDate}. Reason: $msg")
+                Left(TaskResult(jobName, outputTable, RunStatus.Skipped(msg), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, status.dependencyWarnings, Nil, options))
             }
           case Failure(ex) =>
-            Left(TaskResult(task.job, RunStatus.ValidationFailed(ex), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, status.dependencyWarnings, Nil))
+            Left(TaskResult(jobName, outputTable, RunStatus.ValidationFailed(ex), getRunInfo(task.infoDate, started), applicationId, isTransient, isRawFileBased, Nil, status.dependencyWarnings, Nil, options))
         }
     }
   }
@@ -326,7 +343,8 @@ abstract class TaskRunnerBase(conf: Config,
 
         val warnings = validationResult.warnings ++ runResult.warnings ++ saveResult.warnings ++ hiveWarnings
 
-        TaskResult(task.job,
+        TaskResult(task.job.name,
+          MetaTable.getMetaTableDef(task.job.outputTable),
           RunStatus.Succeeded(recordCountOldOpt,
             stats.recordCount,
             stats.dataSizeBytes,
@@ -341,7 +359,8 @@ abstract class TaskRunnerBase(conf: Config,
           isRawFileBased,
           schemaChangesBeforeTransform ::: schemaChangesAfterTransform,
           validationResult.dependencyWarnings,
-          Seq.empty)
+          Seq.empty,
+          task.job.operation.extraOptions)
       }
     } catch {
       case ex: Throwable => Failure(new FatalErrorWrapper("Fatal error has occurred.", ex))
@@ -355,8 +374,17 @@ abstract class TaskRunnerBase(conf: Config,
       case Success(result) =>
         result
       case Failure(ex) =>
-        TaskResult(task.job, RunStatus.Failed(ex), getRunInfo(task.infoDate, started), applicationId,
-          isTransient, isRawFileBased, Nil,validationResult.dependencyWarnings, Nil)
+        TaskResult(task.job.name,
+          MetaTable.getMetaTableDef(task.job.outputTable),
+          RunStatus.Failed(ex),
+          getRunInfo(task.infoDate, started),
+          applicationId,
+          isTransient,
+          isRawFileBased,
+          Nil,
+          validationResult.dependencyWarnings,
+          Nil,
+          task.job.operation.extraOptions)
     }
   }
 
@@ -383,41 +411,39 @@ abstract class TaskRunnerBase(conf: Config,
   }
 
   private def sendNotifications(task: Task, result: TaskResult): Seq[NotificationFailure] = {
-    task.job.notificationTargets.flatMap(notificationTarget => sendNotifications(task, result, notificationTarget))
+    val pipelineInfo = pipelineState.getState().pipelineInfo
+    task.job.notificationTargets.flatMap(notificationTarget => sendNotifications(task, result, notificationTarget, pipelineInfo))
   }
 
-  private def sendNotifications(task: Task, result: TaskResult, notificationTarget: JobNotificationTarget): Option[NotificationFailure] = {
+  private def sendNotifications(task: Task, result: TaskResult, notificationTarget: JobNotificationTarget, pipelineInfo: PipelineInfo): Option[NotificationFailure] = {
     Try {
       val target = notificationTarget.target
 
-      NotificationTargetManager.runStatusToTaskStatus(result.runStatus).foreach { taskStatus =>
-        val notification = TaskNotification(
-          task.job.outputTable.name,
-          MetastoreImpl.getMetaTableDef(task.job.outputTable),
-          Option(task.infoDate),
-          result.runInfo.map(_.started),
-          result.runInfo.map(_.finished),
-          taskStatus,
-          result.applicationId,
-          result.isTransient,
-          result.isRawFilesJob,
-          result.schemaChanges,
-          result.dependencyWarnings.map(_.table),
-          notificationTarget.options
-        )
-
-        target.connect()
-        try {
-          target.sendNotification(notification)
-        } finally {
-          target.close()
-        }
+      val notification = status.TaskResult(
+        task.job.name,
+        MetaTable.getMetaTableDef(task.job.outputTable),
+        result.runStatus,
+        result.runInfo,
+        result.applicationId,
+        result.isTransient,
+        result.isRawFilesJob,
+        result.schemaChanges,
+        result.dependencyWarnings,
+        Seq.empty,
+        notificationTarget.options
+      )
+
+      target.connect()
+      try {
+        target.sendNotification(pipelineInfo, notification)
+      } finally {
+        target.close()
       }
     } match {
       case Success(_) =>
         None
       case Failure(ex) =>
-        log.error(s"$EXCLAMATION Failed to send notifications to '${notificationTarget.name}' for task: ${result.job.outputTable.name} for '${task.infoDate}'.", ex)
+        log.error(s"$EXCLAMATION Failed to send notifications to '${notificationTarget.name}' for task: ${result.outputTable.name} for '${task.infoDate}'.", ex)
         Option(NotificationFailure(
           task.job.outputTable.name,
           notificationTarget.name,
@@ -471,23 +497,23 @@ abstract class TaskRunnerBase(conf: Config,
 
     result.runStatus match {
       case _: RunStatus.Succeeded =>
-        log.info(s"$SUCCESS $taskStr '${result.job.name}'$infoDateMsg has SUCCEEDED.")
+        log.info(s"$SUCCESS $taskStr '${result.jobName}'$infoDateMsg has SUCCEEDED.")
       case RunStatus.ValidationFailed(ex) =>
-        log.error(s"$FAILURE $taskStr '${result.job.name}'$infoDateMsg has FAILED VALIDATION", ex)
+        log.error(s"$FAILURE $taskStr '${result.jobName}'$infoDateMsg has FAILED VALIDATION", ex)
       case RunStatus.Failed(ex) =>
-        log.error(s"$FAILURE $taskStr '${result.job.name}'$infoDateMsg has FAILED", ex)
+        log.error(s"$FAILURE $taskStr '${result.jobName}'$infoDateMsg has FAILED", ex)
       case RunStatus.MissingDependencies(_, tables) =>
-        log.error(s"$emoji $taskStr '${result.job.name}'$infoDateMsg has MISSING TABLES: ${tables.mkString(", ")}")
+        log.error(s"$emoji $taskStr '${result.jobName}'$infoDateMsg has MISSING TABLES: ${tables.mkString(", ")}")
       case RunStatus.FailedDependencies(_, deps) =>
-        log.error(s"$emoji $taskStr '${result.job.name}'$infoDateMsg has FAILED DEPENDENCIES: ${deps.map(_.renderText).mkString("; ")}")
+        log.error(s"$emoji $taskStr '${result.jobName}'$infoDateMsg has FAILED DEPENDENCIES: ${deps.map(_.renderText).mkString("; ")}")
       case _: RunStatus.NoData =>
-        log.warn(s"$emoji $taskStr '${result.job.name}'$infoDateMsg has NO DATA AT SOURCE.")
+        log.warn(s"$emoji $taskStr '${result.jobName}'$infoDateMsg has NO DATA AT SOURCE.")
       case _: RunStatus.InsufficientData =>
-        log.error(s"$FAILURE $taskStr '${result.job.name}'$infoDateMsg has INSUFFICIENT DATA AT SOURCE.")
+        log.error(s"$FAILURE $taskStr '${result.jobName}'$infoDateMsg has INSUFFICIENT DATA AT SOURCE.")
       case RunStatus.Skipped(msg, _) =>
-        log.warn(s"$WARNING $taskStr '${result.job.name}'$infoDateMsg is SKIPPED: $msg.")
+        log.warn(s"$WARNING $taskStr '${result.jobName}'$infoDateMsg is SKIPPED: $msg.")
       case RunStatus.NotRan =>
-        log.info(s"$taskStr '${result.job.name}'$infoDateMsg is SKIPPED.")
+        log.info(s"$taskStr '${result.jobName}'$infoDateMsg is SKIPPED.")
     }
   }
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerMultithreaded.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerMultithreaded.scala
index ac5b4dd55..279abf770 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerMultithreaded.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerMultithreaded.scala
@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.runner.task
 
 import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
+import za.co.absa.pramen.api.status.RunStatus
 import za.co.absa.pramen.core.app.config.RuntimeConfig
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.exceptions.FatalErrorWrapper
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineState.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineState.scala
index 8047a1071..8057284ea 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineState.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineState.scala
@@ -16,7 +16,7 @@
 
 package za.co.absa.pramen.core.state
 
-import za.co.absa.pramen.core.runner.task.TaskResult
+import za.co.absa.pramen.api.status.{PipelineStateSnapshot, TaskResult}
 
 trait PipelineState extends AutoCloseable {
   def getState(): PipelineStateSnapshot
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala
index 462f88186..d85159d93 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala
@@ -19,16 +19,17 @@ package za.co.absa.pramen.core.state
 import com.typesafe.config.Config
 import org.slf4j.{Logger, LoggerFactory}
 import sun.misc.Signal
-import za.co.absa.pramen.api.{NotificationBuilder, PipelineNotificationTarget, TaskNotification}
+import za.co.absa.pramen.api.status.RunStatus.NotRan
+import za.co.absa.pramen.api.status._
+import za.co.absa.pramen.api.{NotificationBuilder, PipelineInfo, PipelineNotificationTarget}
 import za.co.absa.pramen.core.app.config.HookConfig
-import za.co.absa.pramen.core.app.config.RuntimeConfig.EMAIL_IF_NO_CHANGES
-import za.co.absa.pramen.core.metastore.MetastoreImpl
+import za.co.absa.pramen.core.app.config.RuntimeConfig.{DRY_RUN, EMAIL_IF_NO_CHANGES, UNDERCOVER}
+import za.co.absa.pramen.core.config.Keys.{GOOD_THROUGHPUT_RPS, WARN_THROUGHPUT_RPS}
 import za.co.absa.pramen.core.metastore.peristence.{TransientJobManager, TransientTableManager}
+import za.co.absa.pramen.core.notify.PipelineNotificationTargetFactory
 import za.co.absa.pramen.core.notify.pipeline.{PipelineNotification, PipelineNotificationEmail}
-import za.co.absa.pramen.core.notify.{NotificationTargetManager, PipelineNotificationTargetFactory}
 import za.co.absa.pramen.core.pipeline.PipelineDef._
-import za.co.absa.pramen.core.runner.task.RunStatus.NotRan
-import za.co.absa.pramen.core.runner.task.{PipelineNotificationFailure, TaskResult}
+import za.co.absa.pramen.core.utils.ConfigUtils
 
 import java.time.Instant
 import scala.collection.mutable.ListBuffer
@@ -49,6 +50,7 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
 
   // State
   private val startedInstant = Instant.now
+  private var finishedInstant: Option[Instant] = None
   @volatile private var exitCode = EXIT_CODE_SUCCESS
   private val taskResults = new ListBuffer[TaskResult]
   private val pipelineNotificationFailures = new ListBuffer[PipelineNotificationFailure]
@@ -75,13 +77,45 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
   }
 
   override def getState(): PipelineStateSnapshot = synchronized {
+    val appException =  if (!exitedNormally && failureException.isEmpty && signalException.isDefined) {
+      signalException
+    } else
+      failureException
+
+    val notificationBuilderImpl = notificationBuilder.asInstanceOf[NotificationBuilderImpl]
+    val customNotification = CustomNotification (
+      notificationBuilderImpl.entries,
+      notificationBuilderImpl.signature
+    )
+
+    val minRps = ConfigUtils.getOptionInt(conf, WARN_THROUGHPUT_RPS).getOrElse(0)
+    val goodRps = ConfigUtils.getOptionInt(conf, GOOD_THROUGHPUT_RPS).getOrElse(0)
+    val dryRun = ConfigUtils.getOptionBoolean(conf, DRY_RUN).getOrElse(false)
+    val undercover = ConfigUtils.getOptionBoolean(conf, UNDERCOVER).getOrElse(false)
+
     PipelineStateSnapshot(
+      PipelineInfo(
+        pipelineName,
+        environmentName,
+        RuntimeInfo(
+          dryRun,
+          undercover,
+          minRps,
+          goodRps
+        ),
+        startedInstant,
+        finishedInstant,
+        sparkAppId,
+        appException,
+        pipelineNotificationFailures.toSeq
+      ),
       isFinished,
       exitedNormally,
       exitCode,
       customShutdownHookCanRun,
-      failureException,
-      taskResults.toList
+      taskResults.toList,
+      pipelineNotificationFailures.toList,
+      customNotification
     )
   }
 
@@ -136,6 +170,7 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
       exitCode |= EXIT_CODE_SIGNAL_RECEIVED
     }
 
+    finishedInstant = Option(Instant.now())
     sendPipelineNotifications()
     runCustomShutdownHook()
     removeSignalHandlers()
@@ -183,18 +218,15 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
   }
 
   private[state] def sendPipelineNotifications(): Unit = {
-    val taskNotifications = taskResults.flatMap(taskResultToTaskNotification).toSeq
-
-    pipelineNotificationTargets.foreach(notificationTarget => sendCustomNotification(notificationTarget, taskNotifications))
+    pipelineNotificationTargets.foreach(notificationTarget => sendCustomNotification(notificationTarget, getState(), taskResults.toSeq))
   }
 
-  private[state] def sendCustomNotification(pipelineNotificationTarget: PipelineNotificationTarget, taskNotifications: Seq[TaskNotification]): Unit = {
+  private[state] def sendCustomNotification(pipelineNotificationTarget: PipelineNotificationTarget, pipelineStateSnapshot: PipelineStateSnapshot, taskResults: Seq[TaskResult]): Unit = {
     try {
       pipelineNotificationTarget.sendNotification(
-        startedInstant,
-        sparkAppId,
-        failureException,
-        taskNotifications
+        pipelineStateSnapshot.pipelineInfo,
+        taskResults.toSeq,
+        pipelineStateSnapshot.customNotification
       )
     } catch {
       case ex: Throwable =>
@@ -212,7 +244,6 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
       } else {
         taskResults.filterNot(_.isTransient)
       }
-      val finishedInstant = Instant.now
       val notificationBuilderImpl = notificationBuilder.asInstanceOf[NotificationBuilderImpl]
       val customEntries = notificationBuilderImpl.entries
       val customSignature = notificationBuilderImpl.signature
@@ -222,7 +253,7 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
         environmentName,
         sparkAppId,
         startedInstant,
-        finishedInstant,
+        finishedInstant.getOrElse(Instant.now()),
         realTaskResults.toList,
         pipelineNotificationFailures.toList,
         customEntries.toList,
@@ -279,23 +310,4 @@ object PipelineStateImpl {
   val EXIT_CODE_APP_FAILED = 1
   val EXIT_CODE_JOB_FAILED = 2
   val EXIT_CODE_SIGNAL_RECEIVED = 4
-
-  private[core] def taskResultToTaskNotification(taskResult: TaskResult): Option[TaskNotification] = {
-    NotificationTargetManager.runStatusToTaskStatus(taskResult.runStatus).map(taskStatus =>
-      TaskNotification(
-        taskResult.job.outputTable.name,
-        MetastoreImpl.getMetaTableDef(taskResult.job.outputTable),
-        taskResult.runInfo.map(_.infoDate),
-        taskResult.runInfo.map(_.started),
-        taskResult.runInfo.map(_.finished),
-        taskStatus,
-        taskResult.applicationId,
-        taskResult.isTransient,
-        taskResult.isRawFilesJob,
-        taskResult.schemaChanges,
-        taskResult.dependencyWarnings.map(_.table),
-        Map.empty
-      )
-    )
-  }
 }
diff --git a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/notification/EcsNotificationTarget.scala b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/notification/EcsNotificationTarget.scala
index 210b0c0eb..ddef5e1a9 100644
--- a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/notification/EcsNotificationTarget.scala
+++ b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/notification/EcsNotificationTarget.scala
@@ -20,7 +20,8 @@ import com.typesafe.config.Config
 import org.apache.hadoop.fs.Path
 import org.apache.http.HttpStatus
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.{DataFormat, MetaTableDef, NotificationTarget, TaskNotification}
+import za.co.absa.pramen.api.status.TaskResult
+import za.co.absa.pramen.api.{DataFormat, MetaTableDef, NotificationTarget, PipelineInfo}
 import za.co.absa.pramen.core.utils.Emoji
 import za.co.absa.pramen.extras.utils.ConfigUtils
 import za.co.absa.pramen.extras.utils.httpclient.{HttpMethod, SimpleHttpClient, SimpleHttpRequest}
@@ -36,18 +37,18 @@ class EcsNotificationTarget(conf: Config) extends NotificationTarget {
 
   override def config: Config = conf
 
-  override def sendNotification(notification: TaskNotification): Unit = {
-    if (notification.infoDate.isEmpty) {
+  override def sendNotification(pipelineInfo: PipelineInfo, notification: TaskResult): Unit = {
+    if (notification.runInfo.isEmpty) {
       log.warn(s"Information date not provided - skipping ECS cleanup.")
       return
     }
 
     val (ecsApiUrl, ecsApiKey, trustAllSslCerts) = EcsNotificationTarget.getEcsDetails(conf)
-    val tableDef = notification.tableDef
+    val tableDef = notification.outputTable
     val httpClient = getHttpClient(trustAllSslCerts)
 
     try {
-      EcsNotificationTarget.cleanUpS3VersionsForTable(tableDef, notification.infoDate.get, ecsApiUrl, ecsApiKey, httpClient)
+      EcsNotificationTarget.cleanUpS3VersionsForTable(tableDef, notification.runInfo.get.infoDate, ecsApiUrl, ecsApiKey, httpClient)
     } finally {
       httpClient.close()
     }
diff --git a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTarget.scala b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTarget.scala
index 2e427a7b0..9d4d46fd8 100644
--- a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTarget.scala
+++ b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTarget.scala
@@ -18,12 +18,11 @@ package za.co.absa.pramen.extras.notification
 
 import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.{DataFormat, PipelineNotificationTarget, TaskNotification, TaskStatus}
+import za.co.absa.pramen.api.status.{CustomNotification, RunStatus, TaskResult}
+import za.co.absa.pramen.api.{DataFormat, PipelineInfo, PipelineNotificationTarget}
 import za.co.absa.pramen.extras.utils.ConfigUtils
 import za.co.absa.pramen.extras.utils.httpclient.SimpleHttpClient
 
-import java.time.Instant
-
 /**
   * Runs the ECS cleanup API against the target partition after the job jas completed.
   *
@@ -44,10 +43,9 @@ class EcsPipelineNotificationTarget(conf: Config) extends PipelineNotificationTa
   override def config: Config = conf
 
   /** Sends a notification after completion of the pipeline. */
-  override def sendNotification(pipelineStarted: Instant,
-                                applicationId: Option[String],
-                                appException: Option[Throwable],
-                                tasksCompleted: Seq[TaskNotification]): Unit = {
+  override def sendNotification(pipelineInfo: PipelineInfo,
+                                tasksCompleted: Seq[TaskResult],
+                                customNotification: CustomNotification): Unit = {
     log.info(s"Running the ECS cleanup pipeline notification target...")
     val (ecsApiUrl, ecsApiKey, trustAllSslCerts) = EcsPipelineNotificationTarget.getEcsDetails(conf)
 
@@ -55,19 +53,19 @@ class EcsPipelineNotificationTarget(conf: Config) extends PipelineNotificationTa
 
     try {
       tasksCompleted.foreach { task =>
-        (task.infoDate, task.status) match {
-          case (Some(infoDate), _: TaskStatus.Succeeded) =>
-            if (!task.tableDef.format.isTransient &&
-              !task.tableDef.format.isInstanceOf[DataFormat.Null] &&
-              !task.tableDef.format.isInstanceOf[DataFormat.Raw]) {
-              EcsNotificationTarget.cleanUpS3VersionsForTable(task.tableDef, infoDate, ecsApiUrl, ecsApiKey, httpClient)
+        task.runInfo match {
+          case Some(runInfo) if task.runStatus.isInstanceOf[RunStatus.Succeeded] =>
+            if (!task.outputTable.format.isTransient &&
+              !task.outputTable.format.isInstanceOf[DataFormat.Null] &&
+              !task.outputTable.format.isInstanceOf[DataFormat.Raw]) {
+              EcsNotificationTarget.cleanUpS3VersionsForTable(task.outputTable, runInfo.infoDate, ecsApiUrl, ecsApiKey, httpClient)
             } else {
-              log.info(s"The task outputting to '${task.tableName}' for '$infoDate' outputs to ${task.tableDef.format.name} format - skipping ECS cleanup...")
+              log.info(s"The task outputting to '${task.outputTable.name}' for '${runInfo.infoDate}' outputs to ${task.outputTable.format.name} format - skipping ECS cleanup...")
             }
-          case (Some(infoDate), _) =>
-            log.info(s"The task outputting to '${task.tableName}' for '$infoDate' status is not a success - skipping ECS cleanup...")
-          case (None, status) =>
-            log.info(s"The task outputting to '${task.tableName}' status is not a success - skipping ECS cleanup...")
+          case Some(runInfo) =>
+            log.info(s"The task outputting to '${task.outputTable.name}' for '${runInfo.infoDate}' status is not a success - skipping ECS cleanup...")
+          case None =>
+            log.info(s"The task outputting to '${task.outputTable.name}' status is not a success - skipping ECS cleanup...")
         }
       }
     } finally {
diff --git a/pramen/version.sbt b/pramen/version.sbt
index 5bb723ae7..01c201ba2 100644
--- a/pramen/version.sbt
+++ b/pramen/version.sbt
@@ -1,1 +1,1 @@
-ThisBuild / version := "1.8.12-SNAPSHOT"
+ThisBuild / version := "1.9.0-SNAPSHOT"


Test patch:
diff --git a/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala b/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala
index 49ed2f4f6..febdaf329 100644
--- a/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala
+++ b/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala
@@ -18,16 +18,21 @@ package za.co.absa.pramen.api.mocks
 
 import com.typesafe.config.Config
 import za.co.absa.pramen.api.common.BuildPropertiesRetriever
-import za.co.absa.pramen.api.{MetadataManager, NotificationBuilder, Pramen, TaskNotification}
+import za.co.absa.pramen.api.status.{PipelineStateSnapshot, TaskResult}
+import za.co.absa.pramen.api.{MetadataManager, NotificationBuilder, PipelineInfo, Pramen}
 
 class DummyPramen extends Pramen {
   override def buildProperties: BuildPropertiesRetriever = null
 
   override def workflowConfig: Config = null
 
+  override def pipelineInfo: PipelineInfo = null
+
+  override def pipelineState: PipelineStateSnapshot = null
+
   override def notificationBuilder: NotificationBuilder = null
 
   override def metadataManager: MetadataManager = null
 
-  override def getCompletedTasks: Seq[TaskNotification] = null
+  override def getCompletedTasks: Seq[TaskResult] = null
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/OperationDefFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/OperationDefFactory.scala
index 184b8b1e8..3b5f9def2 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/OperationDefFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/OperationDefFactory.scala
@@ -17,7 +17,7 @@
 package za.co.absa.pramen.core
 
 import com.typesafe.config.{Config, ConfigFactory}
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
+import za.co.absa.pramen.api.status.MetastoreDependency
 import za.co.absa.pramen.core.pipeline.{OperationDef, OperationType, TransformExpression}
 import za.co.absa.pramen.core.schedule.Schedule
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/PramenImplSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/PramenImplSuite.scala
index aa3bcd3f3..62bc9f6e4 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/PramenImplSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/PramenImplSuite.scala
@@ -21,9 +21,9 @@ import org.mockito.Mockito.{mock, when}
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.NotificationBuilder
 import za.co.absa.pramen.api.common.BuildPropertiesRetriever
+import za.co.absa.pramen.api.status.RunStatus
 import za.co.absa.pramen.core.metadata.MetadataManagerNull
 import za.co.absa.pramen.core.mocks.{PipelineStateSnapshotFactory, TaskResultFactory}
-import za.co.absa.pramen.core.runner.task.RunStatus
 import za.co.absa.pramen.core.state.PipelineState
 
 import java.time.LocalDate
@@ -108,6 +108,40 @@ class PramenImplSuite extends AnyWordSpec {
     }
   }
 
+  "pipelineInfo" should {
+    "return pipeline info when it is available" in {
+      val pramen = PramenImpl.instance.asInstanceOf[PramenImpl]
+
+      val taskResults = Seq(
+        TaskResultFactory.getDummyTaskResult(),
+        TaskResultFactory.getDummyTaskResult(runInfo = None),
+        TaskResultFactory.getDummyTaskResult(runStatus = RunStatus.NotRan)
+      )
+
+      val pipelineState = mock(classOf[PipelineState])
+      when(pipelineState.getState()).thenReturn(PipelineStateSnapshotFactory.getDummyPipelineStateSnapshot(taskResults = taskResults))
+
+      pramen.setPipelineState(pipelineState)
+
+      val pipelineInfo = PramenImpl.instance.pipelineInfo
+
+      assert(pipelineInfo.pipelineName == "Dummy Pipeline")
+      assert(pipelineInfo.environment == "DEV")
+
+      pramen.setPipelineState(null)
+    }
+
+    "throw an exception if pipeline state is not available" in {
+      val pramen = PramenImpl.instance.asInstanceOf[PramenImpl]
+
+      pramen.setPipelineState(null)
+
+      assertThrows[IllegalStateException] {
+        PramenImpl.instance.pipelineInfo
+      }
+    }
+  }
+
   "getCompletedTasks()" should {
     "return the metadata manager if it is available" in {
       val pramen = PramenImpl.instance.asInstanceOf[PramenImpl]
@@ -125,9 +159,9 @@ class PramenImplSuite extends AnyWordSpec {
 
       val tasks = PramenImpl.instance.getCompletedTasks
 
-      assert(tasks.length == 2)
-      assert(tasks.head.infoDate.contains(LocalDate.of(2022, 2, 18)))
-      assert(tasks(1).infoDate.isEmpty)
+      assert(tasks.length == 3)
+      assert(tasks.head.runInfo.get.infoDate == LocalDate.of(2022, 2, 18))
+      assert(tasks(1).runInfo.isEmpty)
 
       pramen.setPipelineState(null)
     }
@@ -142,4 +176,37 @@ class PramenImplSuite extends AnyWordSpec {
       }
     }
   }
+
+  "pipelineState" should {
+    "return the current pipeline state if it is available" in {
+      val pramen = PramenImpl.instance.asInstanceOf[PramenImpl]
+
+      val taskResults = Seq(
+        TaskResultFactory.getDummyTaskResult(),
+        TaskResultFactory.getDummyTaskResult(runInfo = None),
+        TaskResultFactory.getDummyTaskResult(runStatus = RunStatus.NotRan)
+      )
+
+      val pipelineState = mock(classOf[PipelineState])
+      when(pipelineState.getState()).thenReturn(PipelineStateSnapshotFactory.getDummyPipelineStateSnapshot(taskResults = taskResults))
+
+      pramen.setPipelineState(pipelineState)
+
+      val state = PramenImpl.instance.pipelineState
+
+      assert(state.taskResults.length == 3)
+
+      pramen.setPipelineState(null)
+    }
+
+    "throw an exception if the pipeline state is not available" in {
+      val pramen = PramenImpl.instance.asInstanceOf[PramenImpl]
+
+      pramen.setPipelineState(null)
+
+      assertThrows[IllegalStateException] {
+        PramenImpl.instance.pipelineState
+      }
+    }
+  }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/TaskNotificationFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/TaskNotificationFactory.scala
index f6566244b..1e8920325 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/TaskNotificationFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/TaskNotificationFactory.scala
@@ -16,24 +16,37 @@
 
 package za.co.absa.pramen.core
 
-import za.co.absa.pramen.api.{MetaTableDef, SchemaDifference, TaskNotification, TaskStatus}
+import za.co.absa.pramen.api.status._
+import za.co.absa.pramen.api.{MetaTableDef, SchemaDifference}
 
 import java.time.{Instant, LocalDate}
 
 object TaskNotificationFactory {
-  def getDummyTaskNotification(tableName: String = "dummy_table",
-                               tableDef: MetaTableDef = MetaTableDefFactory.getDummyMetaTableDef(name = "dummy_table"),
-                               infoDate: Option[LocalDate] = Some(LocalDate.of(2022, 2, 18)),
-                               started: Option[Instant] = Some(Instant.ofEpochMilli(1613600000000L)),
-                               finished: Option[Instant] = Some(Instant.ofEpochMilli(1672759508000L)),
-                               status: TaskStatus = TaskStatus.Succeeded(100, Seq.empty, Seq.empty, Seq.empty, Seq.empty),
+  def getDummyTaskNotification(jobName: String = "Dummy Job",
+                               outputTable: MetaTableDef = MetaTableDefFactory.getDummyMetaTableDef(name = "dummy_table"),
+                               runInfo: Option[RunInfo] = Some(RunInfo(
+                                 LocalDate.of(2022, 2, 18),
+                                 Instant.ofEpochMilli(1613600000000L),
+                                 Instant.ofEpochMilli(1672759508000L)
+                               )),
+                               status: RunStatus = RunStatus.Succeeded(None, 100, None, TaskRunReason.New, Seq.empty, Seq.empty, Seq.empty, Seq.empty),
                                applicationId: String = "app_12345",
                                isTransient: Boolean = false,
                                isRawFilesJob: Boolean = false,
                                schemaChanges: Seq[SchemaDifference] = Seq.empty,
-                               dependencyWarningTables: Seq[String] = Seq.empty,
-                               options: Map[String, String] = Map.empty[String, String]): TaskNotification = {
-    TaskNotification(tableName, tableDef, infoDate, started, finished, status, applicationId, isTransient, isRawFilesJob, schemaChanges, dependencyWarningTables, options)
+                               dependencyWarnings: Seq[DependencyWarning] = Seq.empty,
+                               notificationTargetErrors: Seq[NotificationFailure] = Seq.empty,
+                               options: Map[String, String] = Map.empty[String, String]): TaskResult = {
+    TaskResult(jobName,
+      outputTable,
+      status,
+      runInfo,
+      applicationId,
+      isTransient,
+      isRawFilesJob,
+      schemaChanges,
+      dependencyWarnings,
+      notificationTargetErrors,
+      options)
   }
-
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/TransientJobManagerSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/TransientJobManagerSuite.scala
index 4ea958aa2..f5f130105 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/TransientJobManagerSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/TransientJobManagerSuite.scala
@@ -20,14 +20,15 @@ import org.apache.spark.sql.DataFrame
 import org.mockito.Mockito.{mock, when => whenMock}
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.{RunStatus, TaskRunReason}
 import za.co.absa.pramen.api.{CachePolicy, DataFormat}
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.fixtures.TempDirFixture
 import za.co.absa.pramen.core.metastore.peristence.TransientJobManager.{MAXIMUM_UNIONS, WARN_UNIONS}
 import za.co.absa.pramen.core.metastore.peristence.{TransientJobManager, TransientTableManager}
 import za.co.absa.pramen.core.mocks.MetaTableFactory
-import za.co.absa.pramen.core.pipeline.{Job, TaskRunReason}
-import za.co.absa.pramen.core.runner.task.{RunStatus, TaskRunner}
+import za.co.absa.pramen.core.pipeline.Job
+import za.co.absa.pramen.core.runner.task.TaskRunner
 
 import java.time.LocalDate
 import scala.concurrent.ExecutionContext.Implicits.global
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala
new file mode 100644
index 000000000..894c2273d
--- /dev/null
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.core.mocks
+
+import za.co.absa.pramen.api.PipelineInfo
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, RuntimeInfo}
+
+import java.time.Instant
+
+object PipelineInfoFactory {
+  def getDummyPipelineInfo(pipelineName: String = "Dummy Pipeline",
+                           environment: String = "DEV",
+                           runtimeInfo: RuntimeInfo = RuntimeInfo(),
+                           startedAt: Instant = Instant.ofEpochSecond(1718609409),
+                           finishedAt: Option[Instant] = None,
+                           sparkApplicationId: Option[String] = Some("testid-12345"),
+                           failureException: Option[Throwable] = None,
+                           pipelineNotificationFailures: Seq[PipelineNotificationFailure] = Seq.empty): PipelineInfo = {
+    PipelineInfo(pipelineName, environment, runtimeInfo, startedAt, finishedAt, sparkApplicationId, failureException, pipelineNotificationFailures)
+  }
+}
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineNotificationFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineNotificationFactory.scala
index 4ff6af295..26f9abab9 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineNotificationFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineNotificationFactory.scala
@@ -17,9 +17,9 @@
 package za.co.absa.pramen.core.mocks
 
 import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, TaskResult}
 import za.co.absa.pramen.core.notify.pipeline
 import za.co.absa.pramen.core.notify.pipeline.PipelineNotification
-import za.co.absa.pramen.core.runner.task.{PipelineNotificationFailure, TaskResult}
 
 import java.time.Instant
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineStateSnapshotFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineStateSnapshotFactory.scala
index acbf18c85..bdc1ad2c9 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineStateSnapshotFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineStateSnapshotFactory.scala
@@ -16,16 +16,26 @@
 
 package za.co.absa.pramen.core.mocks
 
-import za.co.absa.pramen.core.runner.task.TaskResult
-import za.co.absa.pramen.core.state.PipelineStateSnapshot
+import za.co.absa.pramen.api.PipelineInfo
+import za.co.absa.pramen.api.status.{CustomNotification, PipelineNotificationFailure, PipelineStateSnapshot, TaskResult}
 
 object PipelineStateSnapshotFactory {
-  def getDummyPipelineStateSnapshot(isFinished: Boolean = false,
+  def getDummyPipelineStateSnapshot(pipelineInfo: PipelineInfo = PipelineInfoFactory.getDummyPipelineInfo(),
+                                    isFinished: Boolean = false,
                                     exitedNormally: Boolean = false,
                                     exitCode: Int = 0,
                                     customShutdownHookCanRun: Boolean = false,
-                                    failureException: Option[Throwable] = None,
-                                    taskResults: Seq[TaskResult] = Seq.empty): PipelineStateSnapshot = {
-    PipelineStateSnapshot(isFinished, exitedNormally, exitCode, customShutdownHookCanRun, failureException, taskResults)
+                                    taskResults: Seq[TaskResult] = Seq.empty,
+                                    pipelineNotificationFailures: Seq[PipelineNotificationFailure] = Seq.empty,
+                                    customNotification: CustomNotification = CustomNotification(Seq.empty, Seq.empty)): PipelineStateSnapshot = {
+    PipelineStateSnapshot(pipelineInfo,
+      isFinished,
+      exitedNormally,
+      exitCode,
+      customShutdownHookCanRun,
+      taskResults,
+      pipelineNotificationFailures,
+      customNotification
+    )
   }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/RunStatusFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/RunStatusFactory.scala
index 610901625..784f4840c 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/RunStatusFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/RunStatusFactory.scala
@@ -16,8 +16,7 @@
 
 package za.co.absa.pramen.core.mocks
 
-import za.co.absa.pramen.core.pipeline.TaskRunReason
-import za.co.absa.pramen.core.runner.task.RunStatus
+import za.co.absa.pramen.api.status.{RunStatus, TaskRunReason}
 
 object RunStatusFactory {
   def getDummySuccess(recordCountOld: Option[Long] = None,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskCompletedFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskCompletedFactory.scala
index d0010346c..6ca935fbd 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskCompletedFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskCompletedFactory.scala
@@ -16,9 +16,9 @@
 
 package za.co.absa.pramen.core.mocks
 
+import za.co.absa.pramen.api.status.TaskStatus
 import za.co.absa.pramen.core.journal.model
 import za.co.absa.pramen.core.journal.model.TaskCompleted
-import za.co.absa.pramen.core.model.TaskStatus
 
 import java.time.LocalDate
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskResultFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskResultFactory.scala
index 37aad2e3f..c740e5d8b 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskResultFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskResultFactory.scala
@@ -16,15 +16,15 @@
 
 package za.co.absa.pramen.core.mocks
 
-import za.co.absa.pramen.api.SchemaDifference
-import za.co.absa.pramen.core.mocks.job.JobSpy
-import za.co.absa.pramen.core.pipeline.{DependencyWarning, Job, TaskRunReason}
-import za.co.absa.pramen.core.runner.task.{NotificationFailure, RunInfo, RunStatus, TaskResult}
+import za.co.absa.pramen.api.status._
+import za.co.absa.pramen.api.{MetaTableDef, SchemaDifference}
+import za.co.absa.pramen.core.metastore.model.MetaTable
 
 import java.time.{Instant, LocalDate}
 
 object TaskResultFactory {
-  def getDummyTaskResult(job: Job = new JobSpy(),
+  def getDummyTaskResult(jobName: String = "DummyJob",
+                         outputTable: MetaTableDef = MetaTable.getMetaTableDef(MetaTableFactory.getDummyMetaTable(name = "table_out")),
                          runStatus: RunStatus = RunStatus.Succeeded(Some(100), 200, Some(1000), TaskRunReason.New, Nil, Nil, Nil, Nil),
                          runInfo: Option[RunInfo] = Some(RunInfo(LocalDate.of(2022, 2, 18), Instant.ofEpochSecond(1234), Instant.ofEpochSecond(5678))),
                          applicationId: String = "app_123",
@@ -32,8 +32,10 @@ object TaskResultFactory {
                          isRawFilesJob: Boolean = false,
                          schemaDifferences: Seq[SchemaDifference] = Nil,
                          dependencyWarnings: Seq[DependencyWarning] = Nil,
-                         notificationTargetErrors: Seq[NotificationFailure] = Nil): TaskResult = {
-    TaskResult(job,
+                         notificationTargetErrors: Seq[NotificationFailure] = Nil,
+                         options: Map[String, String] = Map.empty): TaskResult = {
+    TaskResult(jobName,
+      outputTable,
       runStatus,
       runInfo,
       applicationId,
@@ -41,7 +43,8 @@ object TaskResultFactory {
       isRawFilesJob,
       schemaDifferences,
       dependencyWarnings,
-      notificationTargetErrors)
+      notificationTargetErrors,
+      options)
   }
 
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TestPrototypes.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TestPrototypes.scala
index f2b2ed94b..4764b9ff4 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TestPrototypes.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TestPrototypes.scala
@@ -16,8 +16,7 @@
 
 package za.co.absa.pramen.core.mocks
 
-import za.co.absa.pramen.core.pipeline.TaskRunReason
-import za.co.absa.pramen.core.runner.task.RunStatus
+import za.co.absa.pramen.api.status.{RunStatus, TaskRunReason}
 
 object TestPrototypes {
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobBaseDummy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobBaseDummy.scala
index a3df80773..331dedaa9 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobBaseDummy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobBaseDummy.scala
@@ -19,6 +19,7 @@ package za.co.absa.pramen.core.mocks.job
 import com.typesafe.config.Config
 import org.apache.spark.sql.DataFrame
 import za.co.absa.pramen.api.Reason
+import za.co.absa.pramen.api.status.DependencyWarning
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.Metastore
 import za.co.absa.pramen.core.metastore.model.MetaTable
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/NotificationTargetMock.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/NotificationTargetMock.scala
index b7e355e61..2db871bbc 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/NotificationTargetMock.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/NotificationTargetMock.scala
@@ -18,22 +18,23 @@ package za.co.absa.pramen.core.mocks.notify
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.SparkSession
-import za.co.absa.pramen.api.{ExternalChannelFactory, NotificationTarget, TaskNotification}
+import za.co.absa.pramen.api.status.TaskResult
+import za.co.absa.pramen.api.{ExternalChannelFactory, NotificationTarget, PipelineInfo}
 import za.co.absa.pramen.core.mocks.notify.NotificationTargetMock.TEST_NOTIFICATION_FAIL_KEY
 
 import scala.collection.mutable.ListBuffer
 
 class NotificationTargetMock(conf: Config) extends NotificationTarget {
-  val notificationsSent: ListBuffer[TaskNotification] = new ListBuffer[TaskNotification]()
+  val notificationsSent: ListBuffer[TaskResult] = new ListBuffer[TaskResult]()
 
   override def config: Config = conf
 
-  override def sendNotification(notification: TaskNotification): Unit = {
+  override def sendNotification(pipelineInfo: PipelineInfo, notification: TaskResult): Unit = {
     if (conf.hasPath(TEST_NOTIFICATION_FAIL_KEY) && conf.getBoolean(TEST_NOTIFICATION_FAIL_KEY)) {
       System.setProperty("pramen.test.notification.target.failure", "true")
       throw new RuntimeException("Notification target test exception")
     }
-    System.setProperty("pramen.test.notification.table", notification.tableName)
+    System.setProperty("pramen.test.notification.table", notification.outputTable.name)
   }
 }
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/NotificationTargetSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/NotificationTargetSpy.scala
index cab0ff738..c552f4ad8 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/NotificationTargetSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/NotificationTargetSpy.scala
@@ -18,16 +18,17 @@ package za.co.absa.pramen.core.mocks.notify
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.SparkSession
-import za.co.absa.pramen.api.{ExternalChannelFactory, NotificationTarget, TaskNotification}
+import za.co.absa.pramen.api.status.TaskResult
+import za.co.absa.pramen.api.{ExternalChannelFactory, NotificationTarget, PipelineInfo}
 
 import scala.collection.mutable.ListBuffer
 
-class NotificationTargetSpy(conf: Config, action: TaskNotification => Unit) extends NotificationTarget {
-  val notificationsSent: ListBuffer[TaskNotification] = new ListBuffer[TaskNotification]()
+class NotificationTargetSpy(conf: Config, action: TaskResult => Unit) extends NotificationTarget {
+  val notificationsSent: ListBuffer[TaskResult] = new ListBuffer[TaskResult]()
 
   override def config: Config = conf
 
-  override def sendNotification(notification: TaskNotification): Unit = {
+  override def sendNotification(pipelineInfo: PipelineInfo, notification: TaskResult): Unit = {
     notificationsSent += notification
     action(notification)
   }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala
index 10adc79e5..461f7bb31 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala
@@ -17,8 +17,8 @@
 package za.co.absa.pramen.core.mocks.notify
 
 import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, TaskResult}
 import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilder
-import za.co.absa.pramen.core.runner.task.{PipelineNotificationFailure, TaskResult}
 
 import java.time.Instant
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationTargetMock.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationTargetMock.scala
index a97e7a826..3c7cc177a 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationTargetMock.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationTargetMock.scala
@@ -17,14 +17,15 @@
 package za.co.absa.pramen.core.mocks.notify
 
 import com.typesafe.config.Config
-import za.co.absa.pramen.api.{PipelineNotificationTarget, TaskNotification}
+import za.co.absa.pramen.api.status.{CustomNotification, TaskResult}
+import za.co.absa.pramen.api.{PipelineInfo, PipelineNotificationTarget}
 import za.co.absa.pramen.core.mocks.notify.NotificationTargetMock.TEST_NOTIFICATION_FAIL_KEY
 
-import java.time.Instant
-
 class PipelineNotificationTargetMock(conf: Config) extends PipelineNotificationTarget {
 
-  override def sendNotification(pipelineStarted: Instant, applicationId: Option[String], appException: Option[Throwable], tasksCompleted: Seq[TaskNotification]): Unit = {
+  override def sendNotification(pipelineInfo: PipelineInfo,
+                                tasksCompleted: Seq[TaskResult],
+                                customNotification: CustomNotification): Unit = {
     if (conf.hasPath(TEST_NOTIFICATION_FAIL_KEY) && conf.getBoolean(TEST_NOTIFICATION_FAIL_KEY)) {
       System.setProperty("pramen.test.notification.pipeline.failure", "true")
       throw new RuntimeException("Pipeline notification target test exception")
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/runner/ConcurrentJobRunnerSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/runner/ConcurrentJobRunnerSpy.scala
index 6c46ed5f2..52464e331 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/runner/ConcurrentJobRunnerSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/runner/ConcurrentJobRunnerSpy.scala
@@ -17,10 +17,11 @@
 package za.co.absa.pramen.core.mocks.runner
 
 import com.github.yruslan.channel.{Channel, ReadChannel}
-import za.co.absa.pramen.core.pipeline.{Job, TaskRunReason}
+import za.co.absa.pramen.api.status.{RunInfo, RunStatus, TaskResult, TaskRunReason}
+import za.co.absa.pramen.core.metastore.model.MetaTable
+import za.co.absa.pramen.core.pipeline.Job
 import za.co.absa.pramen.core.runner.jobrunner.ConcurrentJobRunner
 import za.co.absa.pramen.core.runner.jobrunner.ConcurrentJobRunner.JobRunResults
-import za.co.absa.pramen.core.runner.task.{RunInfo, RunStatus, TaskResult}
 
 import java.time.{Instant, LocalDate}
 import java.util.concurrent.ExecutorService
@@ -74,7 +75,18 @@ class ConcurrentJobRunnerSpy(includeFails: Boolean = false,
         RunStatus.NoData(isNoDataFailure)
       }
 
-      val taskResult = TaskResult(job, status, Some(RunInfo(infoDate, started, finished)), "app_123", isTransient = false, isRawFilesJob = false, Nil, Nil, Nil)
+      val taskResult = TaskResult(
+        job.name,
+        MetaTable.getMetaTableDef(job.outputTable),
+        status,
+        Some(RunInfo(infoDate, started, finished)),
+        "app_123",
+        isTransient = false,
+        isRawFilesJob = false,
+        Nil,
+        Nil,
+        Nil,
+        Map.empty)
 
       completedJobsChannel.send((job, taskResult :: Nil, taskResult.runStatus.isInstanceOf[RunStatus.Succeeded] || taskResult.runStatus == RunStatus.NotRan))
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/state/PipelineStateSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/state/PipelineStateSpy.scala
index f0e82a611..5b199d3f4 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/state/PipelineStateSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/state/PipelineStateSpy.scala
@@ -16,8 +16,9 @@
 
 package za.co.absa.pramen.core.mocks.state
 
-import za.co.absa.pramen.core.runner.task.TaskResult
-import za.co.absa.pramen.core.state.{PipelineState, PipelineStateSnapshot}
+import za.co.absa.pramen.api.status.{PipelineStateSnapshot, TaskResult}
+import za.co.absa.pramen.core.mocks.{PipelineInfoFactory, PipelineStateSnapshotFactory}
+import za.co.absa.pramen.core.state.PipelineState
 
 import scala.collection.mutable.ListBuffer
 
@@ -32,13 +33,9 @@ class PipelineStateSpy extends PipelineState {
   var sparkAppId: Option[String] = None
 
   override def getState(): PipelineStateSnapshot = {
-    PipelineStateSnapshot(
-      isFinished = false,
-      exitedNormally = false,
-      0,
-      setShutdownHookCanRunCount > 0,
-      None,
-      completedStatuses.toList
+    PipelineStateSnapshotFactory.getDummyPipelineStateSnapshot(PipelineInfoFactory.getDummyPipelineInfo(sparkApplicationId = sparkAppId),
+      customShutdownHookCanRun = setShutdownHookCanRunCount > 0,
+      taskResults = completedStatuses.toList
     )
   }
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/JobBaseSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/JobBaseSuite.scala
index 31d983c91..4e6b62efa 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/JobBaseSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/JobBaseSuite.scala
@@ -19,11 +19,11 @@ package za.co.absa.pramen.core.pipeline
 import com.typesafe.config.ConfigFactory
 import org.apache.spark.sql.DataFrame
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.MetastoreDependency
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.expr.exceptions.SyntaxErrorException
 import za.co.absa.pramen.core.fixtures.TextComparisonFixture
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
 import za.co.absa.pramen.core.mocks.MetaTableFactory
 import za.co.absa.pramen.core.mocks.bookkeeper.SyncBookkeeperMock
 import za.co.absa.pramen.core.mocks.job.JobBaseDummy
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TaskRunReasonSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TaskRunReasonSuite.scala
index f32e12b8b..a26f6cd6a 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TaskRunReasonSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TaskRunReasonSuite.scala
@@ -17,6 +17,7 @@
 package za.co.absa.pramen.core.pipeline
 
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.TaskRunReason
 
 class TaskRunReasonSuite extends AnyWordSpec {
   "TaskRunReason" should {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/source/SourceValidationSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/source/SourceValidationSuite.scala
index 5b04e1b37..ca30b32ab 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/source/SourceValidationSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/source/SourceValidationSuite.scala
@@ -21,6 +21,7 @@ import org.apache.hadoop.fs.Path
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.DataFormat
+import za.co.absa.pramen.api.status.RunStatus
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.fixtures.TempDirFixture
@@ -32,7 +33,7 @@ import za.co.absa.pramen.core.mocks.state.PipelineStateSpy
 import za.co.absa.pramen.core.mocks.{MetaTableFactory, SourceTableFactory}
 import za.co.absa.pramen.core.pipeline.{IngestionJob, Job, OperationType}
 import za.co.absa.pramen.core.runner.jobrunner.ConcurrentJobRunnerImpl
-import za.co.absa.pramen.core.runner.task.{RunStatus, TaskRunnerMultithreaded}
+import za.co.absa.pramen.core.runner.task.TaskRunnerMultithreaded
 import za.co.absa.pramen.core.utils.LocalFsUtils
 import za.co.absa.pramen.core.{OperationDefFactory, RuntimeConfigFactory}
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/state/PipelineStateSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/state/PipelineStateSuite.scala
index 36cdfb050..f8d014e2a 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/state/PipelineStateSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/state/PipelineStateSuite.scala
@@ -19,11 +19,11 @@ package za.co.absa.pramen.core.state
 import com.typesafe.config.{Config, ConfigFactory, ConfigValueFactory}
 import org.scalatest.wordspec.AnyWordSpec
 import org.slf4j.{Logger, LoggerFactory}
+import za.co.absa.pramen.api.status.RunStatus
 import za.co.absa.pramen.core.PramenImpl
 import za.co.absa.pramen.core.app.config.HookConfig.SHUTDOWN_HOOK_CLASS_KEY
 import za.co.absa.pramen.core.mocks.TaskResultFactory
 import za.co.absa.pramen.core.mocks.state.{ShutdownHookFailureMock, ShutdownHookSuccessMock}
-import za.co.absa.pramen.core.runner.task.RunStatus
 
 class PipelineStateSuite extends AnyWordSpec {
   private implicit val conf: Config = ConfigFactory.parseString(
@@ -57,7 +57,7 @@ class PipelineStateSuite extends AnyWordSpec {
 
       assert(stateManager.getState().isFinished)
       assert(stateManager.getState().exitedNormally)
-      assert(stateManager.getState().failureException.isEmpty)
+      assert(stateManager.getState().pipelineInfo.failureException.isEmpty)
     }
   }
 
@@ -72,7 +72,7 @@ class PipelineStateSuite extends AnyWordSpec {
 
       assert(stateManager.getState().isFinished)
       assert(!stateManager.getState().exitedNormally)
-      assert(stateManager.getState().failureException.exists(_.isInstanceOf[RuntimeException]))
+      assert(stateManager.getState().pipelineInfo.failureException.exists(_.isInstanceOf[RuntimeException]))
     }
   }
 
@@ -127,7 +127,7 @@ class PipelineStateSuite extends AnyWordSpec {
       stateManager.runCustomShutdownHook()
 
       assert(ShutdownHookFailureMock.ranTimes > 0)
-      assert(stateManager.getState().failureException.exists(_.isInstanceOf[LinkageError]))
+      assert(stateManager.getState().pipelineInfo.failureException.exists(_.isInstanceOf[LinkageError]))
     }
 
     "handle class does not exists errors" in {
@@ -137,8 +137,8 @@ class PipelineStateSuite extends AnyWordSpec {
       stateManager.runCustomShutdownHook()
 
       assert(ShutdownHookFailureMock.ranTimes > 0)
-      assert(stateManager.getState().failureException.isDefined)
-      assert(stateManager.getState().failureException.exists(_.isInstanceOf[ClassNotFoundException]))
+      assert(stateManager.getState().pipelineInfo.failureException.isDefined)
+      assert(stateManager.getState().pipelineInfo.failureException.exists(_.isInstanceOf[ClassNotFoundException]))
 
     }
   }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TaskCompletedSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TaskCompletedSuite.scala
index f34e89677..d71193310 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TaskCompletedSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TaskCompletedSuite.scala
@@ -17,10 +17,11 @@
 package za.co.absa.pramen.core.tests.journal
 
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.{RunInfo, RunStatus, TaskResult, TaskRunReason}
 import za.co.absa.pramen.core.journal.model.TaskCompleted
+import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.mocks.job.JobSpy
-import za.co.absa.pramen.core.pipeline.{Task, TaskRunReason}
-import za.co.absa.pramen.core.runner.task.{RunInfo, RunStatus, TaskResult}
+import za.co.absa.pramen.core.pipeline.Task
 
 import java.time.{Instant, LocalDate}
 
@@ -34,7 +35,8 @@ class TaskCompletedSuite extends AnyWordSpec {
       val runReason = TaskRunReason.Update
       val task = Task(job, infoDate, runReason)
       val taskResult = TaskResult(
-        job,
+        job.name,
+        MetaTable.getMetaTableDef(job.outputTable),
         RunStatus.Succeeded(Some(1000), 2000, Some(3000), runReason, Nil, Nil, Nil, Nil),
         Some(RunInfo(infoDate, now.minusSeconds(10), now)),
         "app_123",
@@ -42,7 +44,8 @@ class TaskCompletedSuite extends AnyWordSpec {
         isRawFilesJob = false,
         Nil,
         Nil,
-        Nil)
+        Nil,
+        Map.empty)
 
       val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult)
 
@@ -68,7 +71,8 @@ class TaskCompletedSuite extends AnyWordSpec {
       val runReason = TaskRunReason.Update
       val task = Task(job, infoDate, runReason)
       val taskResult = TaskResult(
-        job,
+        job.name,
+        MetaTable.getMetaTableDef(job.outputTable),
         RunStatus.Failed(new IllegalStateException("Dummy Exception")),
         None,
         "app_123",
@@ -76,7 +80,8 @@ class TaskCompletedSuite extends AnyWordSpec {
         isRawFilesJob = false,
         Nil,
         Nil,
-        Nil)
+        Nil,
+        Map.empty)
 
       val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult)
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/HyperdriveNotificationTargetSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/HyperdriveNotificationTargetSuite.scala
index 96895c198..1dced187e 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/HyperdriveNotificationTargetSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/HyperdriveNotificationTargetSuite.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.tests.notify
 
 import com.typesafe.config.ConfigFactory
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.api.TaskStatus
+import za.co.absa.pramen.api.status.{RunStatus, TaskStatus}
 import za.co.absa.pramen.core.TaskNotificationFactory
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.mocks.notify.SingleMessageProducerSpy
@@ -68,7 +68,7 @@ class HyperdriveNotificationTargetSuite extends AnyWordSpec with SparkTestBase {
 
       val taskNotification = TaskNotificationFactory.getDummyTaskNotification(options = options)
 
-      notificationTarget.sendNotification(taskNotification)
+      notificationTarget.sendNotification(null, taskNotification)
 
       assert(producer.connectInvoked == 1)
       assert(producer.sendInvoked == 1)
@@ -81,9 +81,9 @@ class HyperdriveNotificationTargetSuite extends AnyWordSpec with SparkTestBase {
       val (notificationTarget, producer) = getUseCase
       val options = Map("hyperdrive.token" -> "ABC")
 
-      val taskNotification = TaskNotificationFactory.getDummyTaskNotification(status = TaskStatus.Skipped("dummy"),  options = options)
+      val taskNotification = TaskNotificationFactory.getDummyTaskNotification(status = RunStatus.Skipped("dummy"),  options = options)
 
-      notificationTarget.sendNotification(taskNotification)
+      notificationTarget.sendNotification(null, taskNotification)
 
       assert(producer.connectInvoked == 0)
       assert(producer.sendInvoked == 0)
@@ -94,7 +94,7 @@ class HyperdriveNotificationTargetSuite extends AnyWordSpec with SparkTestBase {
 
       val taskNotification = TaskNotificationFactory.getDummyTaskNotification()
 
-      notificationTarget.sendNotification(taskNotification)
+      notificationTarget.sendNotification(null, taskNotification)
 
       assert(producer.connectInvoked == 0)
       assert(producer.sendInvoked == 0)
@@ -108,7 +108,7 @@ class HyperdriveNotificationTargetSuite extends AnyWordSpec with SparkTestBase {
 
       val taskNotification = TaskNotificationFactory.getDummyTaskNotification(options = options)
 
-      notificationTarget.sendNotification(taskNotification)
+      notificationTarget.sendNotification(null, taskNotification)
       notificationTarget.close()
 
       assert(producer.connectInvoked == 1)
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/NotificationTargetManagerSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/NotificationTargetManagerSuite.scala
index 40bc698e2..7387a6176 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/NotificationTargetManagerSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/NotificationTargetManagerSuite.scala
@@ -18,11 +18,8 @@ package za.co.absa.pramen.core.tests.notify
 
 import com.typesafe.config.{Config, ConfigFactory}
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.api.TaskStatus
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.notify.{HyperdriveNotificationTarget, NotificationTargetManager}
-import za.co.absa.pramen.core.pipeline.DependencyFailure
-import za.co.absa.pramen.core.runner.task.RunStatus
 
 class NotificationTargetManagerSuite extends AnyWordSpec with SparkTestBase {
   private val conf: Config = ConfigFactory.parseString(
@@ -60,55 +57,4 @@ class NotificationTargetManagerSuite extends AnyWordSpec with SparkTestBase {
       }
     }
   }
-
-  "runStatusToTaskStatus" should {
-    "decode success" in {
-      assert(NotificationTargetManager.runStatusToTaskStatus(
-        RunStatus.Succeeded(None, 100, None, null, Seq("A"), Seq("B"), Seq("C"), Seq("D")))
-          .contains(TaskStatus.Succeeded(100, Seq("A"), Seq("B"), Seq("C"), Seq("D"))))
-    }
-
-    "decode failure" in {
-      val ex = new RuntimeException("dummy")
-      assert(NotificationTargetManager.runStatusToTaskStatus(RunStatus.Failed(ex)).contains(TaskStatus.Failed(ex)))
-    }
-
-    "decode validation failure" in {
-      val ex = new RuntimeException("dummy")
-      assert(NotificationTargetManager.runStatusToTaskStatus(RunStatus.ValidationFailed(ex)).contains(TaskStatus.ValidationFailed(ex)))
-    }
-
-    "decode missing dependencies" in {
-      val tables = Seq("table1", "table2")
-
-      assert(NotificationTargetManager.runStatusToTaskStatus(RunStatus.MissingDependencies(isFailure = true, tables)).contains(TaskStatus.MissingDependencies(tables)))
-    }
-
-    "decode failed dependencies" in {
-      val failedDependencies = Seq(
-        DependencyFailure(null, Seq("table0"), Seq("table1", "table2"), null),
-        DependencyFailure(null, Seq("table99"), Seq("table3", "table1"), null)
-      )
-
-      val failedTables = Seq("table0", "table1", "table2", "table3", "table99")
-
-      assert(NotificationTargetManager.runStatusToTaskStatus(RunStatus.FailedDependencies(isFailure = true, failedDependencies)).contains(TaskStatus.FailedDependencies(failedTables)))
-    }
-
-    "decode no data" in {
-      assert(NotificationTargetManager.runStatusToTaskStatus(RunStatus.NoData(false)).contains(TaskStatus.NoData))
-    }
-
-    "decode insufficient data" in {
-      assert(NotificationTargetManager.runStatusToTaskStatus(RunStatus.InsufficientData(100, 200, None)).contains(TaskStatus.InsufficientData(100, 200)))
-    }
-
-    "decode skipped" in {
-      assert(NotificationTargetManager.runStatusToTaskStatus(RunStatus.Skipped("msg", isWarning = true)).contains(TaskStatus.Skipped("msg")))
-    }
-
-    "decode not ran" in {
-      assert(NotificationTargetManager.runStatusToTaskStatus(RunStatus.NotRan).isEmpty)
-    }
-  }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationBuilderHtmlSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationBuilderHtmlSuite.scala
index 9ae64beb0..5654e351c 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationBuilderHtmlSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationBuilderHtmlSuite.scala
@@ -20,13 +20,12 @@ import com.typesafe.config.{Config, ConfigFactory}
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.notification.NotificationEntry.Paragraph
 import za.co.absa.pramen.api.notification._
+import za.co.absa.pramen.api.status.{NotificationFailure, PipelineNotificationFailure, RunStatus, TaskRunReason}
 import za.co.absa.pramen.core.exceptions.{CmdFailedException, ProcessFailedException}
 import za.co.absa.pramen.core.fixtures.TextComparisonFixture
 import za.co.absa.pramen.core.mocks.{RunStatusFactory, SchemaDifferenceFactory, TaskResultFactory, TestPrototypes}
 import za.co.absa.pramen.core.notify.message.{MessageBuilderHtml, ParagraphBuilder}
 import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilderHtml
-import za.co.absa.pramen.core.pipeline.TaskRunReason
-import za.co.absa.pramen.core.runner.task.{NotificationFailure, PipelineNotificationFailure, RunStatus}
 import za.co.absa.pramen.core.utils.ResourceUtils
 
 import java.time.{Instant, LocalDate}
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala
index ac1263853..df928ec46 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala
@@ -19,10 +19,10 @@ package za.co.absa.pramen.core.tests.notify.pipeline
 import com.typesafe.config.{Config, ConfigFactory}
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
+import za.co.absa.pramen.api.status.RunStatus
 import za.co.absa.pramen.core.mocks.notify.PipelineNotificationBuilderSpy
 import za.co.absa.pramen.core.mocks.{PipelineNotificationFactory, SchemaDifferenceFactory, TaskResultFactory}
 import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationDirector
-import za.co.absa.pramen.core.runner.task.RunStatus
 
 import java.time.Instant
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/jobrunner/TaskRunnerMultithreadedSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/jobrunner/TaskRunnerMultithreadedSuite.scala
index 0895b279d..bbc664450 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/jobrunner/TaskRunnerMultithreadedSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/jobrunner/TaskRunnerMultithreadedSuite.scala
@@ -20,6 +20,7 @@ import com.github.yruslan.channel.Channel
 import com.typesafe.config.ConfigFactory
 import org.apache.spark.sql.DataFrame
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.RunStatus.{Failed, Succeeded}
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.MetaTableStats
@@ -30,7 +31,6 @@ import za.co.absa.pramen.core.mocks.lock.TokenLockFactoryMock
 import za.co.absa.pramen.core.mocks.state.PipelineStateSpy
 import za.co.absa.pramen.core.pipeline.{Job, RunResult}
 import za.co.absa.pramen.core.runner.jobrunner.ConcurrentJobRunnerImpl
-import za.co.absa.pramen.core.runner.task.RunStatus.{Failed, Succeeded}
 import za.co.absa.pramen.core.runner.task.TaskRunnerMultithreaded
 import za.co.absa.pramen.core.{OperationDefFactory, RuntimeConfigFactory}
 
@@ -53,7 +53,8 @@ class TaskRunnerMultithreadedSuite extends AnyWordSpec with SparkTestBase {
 
       assert(results.size == 1)
       assert(results.head.runStatus.isInstanceOf[Succeeded])
-      assert(results.head.job == job)
+      assert(results.head.jobName == job.name)
+      assert(results.head.outputTable.name == job.outputTable.name)
       assert(results.head.runInfo.nonEmpty)
       assert(results.head.runInfo.get.infoDate == runDate)
       assert(results.head.schemaChanges.isEmpty)
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/orchestrator/OrchestratorSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/orchestrator/OrchestratorSuite.scala
index 4444bedc6..d25a34b9e 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/orchestrator/OrchestratorSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/orchestrator/OrchestratorSuite.scala
@@ -18,16 +18,16 @@ package za.co.absa.pramen.core.tests.runner.orchestrator
 
 import com.typesafe.config.ConfigFactory
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.MetastoreDependency
+import za.co.absa.pramen.api.status.RunStatus.{Failed, MissingDependencies, NoData, Succeeded}
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.fixtures.AppContextFixture
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
-import za.co.absa.pramen.core.pipeline.OperationDef
 import za.co.absa.pramen.core.mocks.job.JobSpy
 import za.co.absa.pramen.core.mocks.runner.ConcurrentJobRunnerSpy
 import za.co.absa.pramen.core.mocks.state.PipelineStateSpy
+import za.co.absa.pramen.core.pipeline.OperationDef
 import za.co.absa.pramen.core.runner.orchestrator.OrchestratorImpl
-import za.co.absa.pramen.core.runner.task.RunStatus.{Failed, MissingDependencies, NoData, Succeeded}
 
 class OrchestratorSuite extends AnyWordSpec with SparkTestBase with AppContextFixture {
   "runJobs" should {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/splitter/ScheduleStrategySuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/splitter/ScheduleStrategySuite.scala
index 4f64a9030..191d47dbe 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/splitter/ScheduleStrategySuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/splitter/ScheduleStrategySuite.scala
@@ -19,11 +19,10 @@ package za.co.absa.pramen.core.tests.runner.splitter
 import org.mockito.ArgumentMatchers
 import org.mockito.Mockito.{mock, when}
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.{MetastoreDependency, TaskRunReason}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
 import za.co.absa.pramen.core.mocks.DataChunkFactory.getDummyDataChunk
 import za.co.absa.pramen.core.pipeline
-import za.co.absa.pramen.core.pipeline.TaskRunReason
 import za.co.absa.pramen.core.runner.splitter.{RunMode, ScheduleParams, ScheduleStrategySourcing, ScheduleStrategyTransformation}
 import za.co.absa.pramen.core.schedule.Schedule
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/splitter/ScheduleStrategyUtilsSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/splitter/ScheduleStrategyUtilsSuite.scala
index 84c08bcc0..68a75cd36 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/splitter/ScheduleStrategyUtilsSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/splitter/ScheduleStrategyUtilsSuite.scala
@@ -18,12 +18,11 @@ package za.co.absa.pramen.core.tests.runner.splitter
 
 import org.mockito.Mockito.{mock, when}
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.{MetastoreDependency, TaskRunReason}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.expr.exceptions.SyntaxErrorException
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
 import za.co.absa.pramen.core.model.DataChunk
 import za.co.absa.pramen.core.pipeline
-import za.co.absa.pramen.core.pipeline.TaskRunReason
 import za.co.absa.pramen.core.runner.splitter.RunMode
 import za.co.absa.pramen.core.runner.splitter.ScheduleStrategyUtils._
 import za.co.absa.pramen.core.schedule.Schedule
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/RunStatusSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/RunStatusSuite.scala
index 749f5275f..30d4595d0 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/RunStatusSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/RunStatusSuite.scala
@@ -17,9 +17,7 @@
 package za.co.absa.pramen.core.tests.runner.task
 
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
-import za.co.absa.pramen.core.pipeline.{DependencyFailure, TaskRunReason}
-import za.co.absa.pramen.core.runner.task.RunStatus
+import za.co.absa.pramen.api.status.{DependencyFailure, MetastoreDependency, RunStatus, TaskRunReason}
 
 class RunStatusSuite extends AnyWordSpec {
   "toString" should {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala
index dca0781de..abaac2555 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala
@@ -20,7 +20,9 @@ import com.typesafe.config.ConfigFactory
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.lit
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.api.{DataFormat, Reason, TaskNotification, TaskStatus}
+import za.co.absa.pramen.api.status.RunStatus.{Failed, NotRan, Skipped, Succeeded}
+import za.co.absa.pramen.api.status._
+import za.co.absa.pramen.api.{DataFormat, Reason}
 import za.co.absa.pramen.core
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
@@ -28,7 +30,6 @@ import za.co.absa.pramen.core.exceptions.ReasonException
 import za.co.absa.pramen.core.fixtures.TextComparisonFixture
 import za.co.absa.pramen.core.journal.Journal
 import za.co.absa.pramen.core.metastore.MetaTableStats
-import za.co.absa.pramen.core.metastore.model.MetastoreDependency
 import za.co.absa.pramen.core.mocks.MetaTableFactory
 import za.co.absa.pramen.core.mocks.bookkeeper.SyncBookkeeperMock
 import za.co.absa.pramen.core.mocks.job.JobSpy
@@ -37,8 +38,7 @@ import za.co.absa.pramen.core.mocks.lock.TokenLockFactoryMock
 import za.co.absa.pramen.core.mocks.notify.NotificationTargetSpy
 import za.co.absa.pramen.core.mocks.state.PipelineStateSpy
 import za.co.absa.pramen.core.pipeline._
-import za.co.absa.pramen.core.runner.task.RunStatus.{Failed, NotRan, Skipped, Succeeded}
-import za.co.absa.pramen.core.runner.task.{RunStatus, TaskRunnerBase, TaskRunnerMultithreaded}
+import za.co.absa.pramen.core.runner.task.{TaskRunnerBase, TaskRunnerMultithreaded}
 import za.co.absa.pramen.core.utils.SparkUtils
 import za.co.absa.pramen.core.{OperationDefFactory, RuntimeConfigFactory}
 
@@ -56,7 +56,7 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
   "runJobTasks" should {
     "run multiple successful jobs parallel execution" in {
       val now = Instant.now()
-      val notificationTarget = new NotificationTargetSpy(ConfigFactory.empty(), (action: TaskNotification) => ())
+      val notificationTarget = new NotificationTargetSpy(ConfigFactory.empty(), (action: TaskResult) => ())
       val jobNotificationTarget = JobNotificationTarget("notification1", Map.empty[String, String], notificationTarget)
       val (runner, _, journal, state, tasks) = getUseCase(runFunction = () => RunResult(exampleDf), jobNotificationTargets = Seq(jobNotificationTarget))
 
@@ -79,7 +79,7 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
       assert(result.head.runStatus.isInstanceOf[Succeeded])
       assert(result(1).runStatus.isInstanceOf[Succeeded])
       assert(notificationTarget.notificationsSent.length == 2)
-      assert(notificationTarget.notificationsSent.head.status.isInstanceOf[TaskStatus.Succeeded])
+      assert(notificationTarget.notificationsSent.head.runStatus.isInstanceOf[RunStatus.Succeeded])
 
       val journalEntries = journal.getEntries(now, now.plusSeconds(30))
 
@@ -147,7 +147,7 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
 
     "run a single lazy job" in {
       val now = Instant.now()
-      val notificationTarget = new NotificationTargetSpy(ConfigFactory.empty(), (action: TaskNotification) => ())
+      val notificationTarget = new NotificationTargetSpy(ConfigFactory.empty(), (action: TaskResult) => ())
       val jobNotificationTarget = JobNotificationTarget("notification1", Map.empty[String, String], notificationTarget)
       val (runner, _, journal, state, tasks) = getUseCase(runFunction = () => RunResult(exampleDf), jobNotificationTargets = Seq(jobNotificationTarget))
 
@@ -162,7 +162,7 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
       assert(job.createHiveTableCount == 0)
       assert(result.isInstanceOf[Succeeded])
       assert(notificationTarget.notificationsSent.length == 1)
-      assert(notificationTarget.notificationsSent.head.status.isInstanceOf[TaskStatus.Succeeded])
+      assert(notificationTarget.notificationsSent.head.runStatus.isInstanceOf[RunStatus.Succeeded])
 
       val journalEntries = journal.getEntries(now, now.plusSeconds(30))
 
@@ -173,7 +173,7 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
 
   "run multiple failure jobs sequential execution" in {
     val now = Instant.now()
-    val notificationTarget = new NotificationTargetSpy(ConfigFactory.empty(), (action: TaskNotification) => ())
+    val notificationTarget = new NotificationTargetSpy(ConfigFactory.empty(), (action: TaskResult) => ())
     val jobNotificationTarget = JobNotificationTarget("notification1", Map.empty[String, String], notificationTarget)
     val (runner, _, journal, state, tasks) = getUseCase(allowParallel = false, runFunction = () => throw new IllegalStateException("Test exception"), jobNotificationTargets = Seq(jobNotificationTarget))
 
@@ -201,7 +201,7 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
     assert(journalEntries.length == 2)
     assert(journalEntries.head.status == "Failed")
     assert(notificationTarget.notificationsSent.length == 2)
-    assert(notificationTarget.notificationsSent.head.status.isInstanceOf[TaskStatus.Failed])
+    assert(notificationTarget.notificationsSent.head.runStatus.isInstanceOf[RunStatus.Failed])
   }
 
   "preRunCheck" when {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerMultithreadedSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerMultithreadedSuite.scala
index 78f6d0662..4d7bfb1d6 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerMultithreadedSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerMultithreadedSuite.scala
@@ -18,15 +18,16 @@ package za.co.absa.pramen.core.tests.runner.task
 
 import com.typesafe.config.ConfigFactory
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.{RunStatus, TaskRunReason}
+import za.co.absa.pramen.api.status.RunStatus.Failed
 import za.co.absa.pramen.core.RuntimeConfigFactory
 import za.co.absa.pramen.core.mocks.bookkeeper.SyncBookkeeperMock
 import za.co.absa.pramen.core.mocks.job.JobSpy
 import za.co.absa.pramen.core.mocks.journal.JournalMock
 import za.co.absa.pramen.core.mocks.lock.TokenLockFactoryMock
 import za.co.absa.pramen.core.mocks.state.PipelineStateSpy
-import za.co.absa.pramen.core.pipeline.{Task, TaskRunReason}
-import za.co.absa.pramen.core.runner.task.RunStatus.Failed
-import za.co.absa.pramen.core.runner.task.{RunStatus, TaskRunnerMultithreaded}
+import za.co.absa.pramen.core.pipeline.Task
+import za.co.absa.pramen.core.runner.task.TaskRunnerMultithreaded
 
 import java.time.LocalDate
 import java.util.concurrent.TimeUnit
diff --git a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/mocks/TestPrototypes.scala b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/mocks/TestPrototypes.scala
index aaf5d38e3..faf594ccb 100644
--- a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/mocks/TestPrototypes.scala
+++ b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/mocks/TestPrototypes.scala
@@ -16,7 +16,8 @@
 
 package za.co.absa.pramen.extras.mocks
 
-import za.co.absa.pramen.api.{DataFormat, MetaTableDef, TaskNotification, TaskStatus}
+import za.co.absa.pramen.api.status.{RunInfo, RunStatus, TaskResult, TaskRunReason}
+import za.co.absa.pramen.api.{DataFormat, MetaTableDef, status}
 import za.co.absa.pramen.extras.utils.httpclient.SimpleHttpResponse
 
 import java.time.{Instant, LocalDate}
@@ -38,19 +39,18 @@ object TestPrototypes {
     Map.empty,
     Map.empty)
 
-  val taskStatus: TaskStatus = TaskStatus.Succeeded(100, Seq.empty, Seq.empty, Seq.empty, Seq.empty)
+  val taskStatus: RunStatus = RunStatus.Succeeded(None, 100, None, TaskRunReason.New, Seq.empty, Seq.empty, Seq.empty, Seq.empty)
 
-  val taskNotification: TaskNotification = TaskNotification(
-    "table1",
+  val taskNotification: TaskResult = status.TaskResult(
+    "Dummy Job",
     metaTableDef,
-    Option(infoDate),
-    Option(Instant.ofEpochSecond(1645274606)),
-    Option(Instant.ofEpochSecond(1645278206)),
     taskStatus,
+    Option(RunInfo(infoDate, Instant.ofEpochSecond(1645274606), Instant.ofEpochSecond(1645278206))),
     "test-1234",
     isTransient = false,
     isRawFilesJob = false,
     Seq.empty,
     Seq.empty,
+    Seq.empty,
     Map.empty)
 }
diff --git a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsNotificationTargetSuite.scala b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsNotificationTargetSuite.scala
index 23cc1e5ab..ad8e76ff6 100644
--- a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsNotificationTargetSuite.scala
+++ b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsNotificationTargetSuite.scala
@@ -45,7 +45,7 @@ class EcsNotificationTargetSuite extends AnyWordSpec {
         override protected def getHttpClient(trustAllSslCerts: Boolean): SimpleHttpClient = httpClient
       }
 
-      notificationTarget.sendNotification(TestPrototypes.taskNotification.copy(tableDef = metaTableDef))
+      notificationTarget.sendNotification(null, TestPrototypes.taskNotification.copy(outputTable = metaTableDef))
 
       assert(httpClient.executeCalled == 1)
       assert(httpClient.requests.head.url == "https://dummyurl.local/kk")
diff --git a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala
index 8814c3d23..0320bc883 100644
--- a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala
+++ b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala
@@ -18,7 +18,8 @@ package za.co.absa.pramen.extras.notification
 
 import com.typesafe.config.ConfigFactory
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.api.{DataFormat, Query}
+import za.co.absa.pramen.api.status.{CustomNotification, RuntimeInfo}
+import za.co.absa.pramen.api.{DataFormat, PipelineInfo, Query}
 import za.co.absa.pramen.extras.mocks.{SimpleHttpClientSpy, TestPrototypes}
 import za.co.absa.pramen.extras.notification.EcsPipelineNotificationTarget.{ECS_API_SECRET_KEY, ECS_API_TRUST_SSL_KEY, ECS_API_URL_KEY}
 import za.co.absa.pramen.extras.utils.httpclient.SimpleHttpClient
@@ -45,15 +46,19 @@ class EcsPipelineNotificationTargetSuite extends AnyWordSpec {
         override protected def getHttpClient(trustAllSslCerts: Boolean): SimpleHttpClient = httpClient
       }
 
-      val task1 = TestPrototypes.taskNotification.copy(tableDef = metaTableDef)
+      val task1 = TestPrototypes.taskNotification.copy(outputTable = metaTableDef)
       val dataFormat2 = DataFormat.Parquet("s3a://dummy_bucket_not_exist/dummy/path2", None)
       val metaTableDef2 = TestPrototypes.metaTableDef.copy(name = "table2", format = dataFormat2)
-      val task2 = TestPrototypes.taskNotification.copy(tableName = "table3", tableDef = metaTableDef2)
+      val task2 = TestPrototypes.taskNotification.copy(jobName = "Job 3", outputTable = metaTableDef2)
       val dataFormat3 = DataFormat.Delta(Query.Table("table2"), None)
       val metaTableDef3 = TestPrototypes.metaTableDef.copy(name = "table3", format = dataFormat3)
-      val task3 = TestPrototypes.taskNotification.copy(tableName = "table3", tableDef = metaTableDef3)
+      val task3 = TestPrototypes.taskNotification.copy(jobName = "Job 3", outputTable = metaTableDef3)
 
-      notificationTarget.sendNotification(Instant.now, None, None, Seq(task1, task2, task3))
+      notificationTarget.sendNotification(
+        PipelineInfo("Dummy", "DEV", RuntimeInfo(), Instant.now, None, None, None, Seq.empty),
+        Seq(task1, task2, task3),
+        CustomNotification(Seq.empty, Seq.empty)
+      )
 
       assert(httpClient.executeCalled == 2)
       assert(httpClient.requests.head.url == "https://dummyurl.local/kk")