feat(ingestion-progress): total assets, run-start emit, agent on discovery

ingestion/src/metadata/workflow/base.py

@@ -32,6 +32,9 @@
     IngestionPipeline,
     PipelineState,
 )
+from metadata.generated.schema.entity.services.ingestionPipelines.progressUpdate import (
+    ProgressUpdateType,
+)
 from metadata.generated.schema.entity.services.ingestionPipelines.status import (
     StackTraceError,
 )
@@ -83,7 +86,7 @@
     Base workflow implementation
     """
 
     config: Union[Any, Dict]  # noqa: UP006, UP007
     _run_id: Optional[str] = None  # noqa: UP045
     metadata: OpenMetadata
     metadata_config: OpenMetadataConnection
@@ -91,7 +94,7 @@
 
     def __init__(
         self,
         config: Union[Any, Dict],  # noqa: UP006, UP007
         workflow_config: WorkflowConfig,
         service_type: ServiceType,
         output_handler: WorkflowOutputHandler = WorkflowOutputHandler(),
@@ -289,6 +292,13 @@
         pipeline_state = PipelineState.success
         self.timer.trigger()
         diagnostics.install(self)
+        # Emit a "run started" update immediately. The reporting timer's first
+        # tick is a full REPORTS_INTERVAL_SECONDS away, so without this a run
+        # that finishes inside that window would only ever emit its terminal
+        # event — and any live viewer would see nothing while it ran. This
+        # registers the run with the server up front so it is visible the moment
+        # it starts, regardless of duration.
+        self.send_progress_update(ProgressUpdateType.DISCOVERY)
         # `self.config` is typed Union[Any, Dict]; getattr keeps the static
         # checker happy without changing behavior (the Dict branch never
         # carries this attribute at runtime).

ingestion/src/metadata/workflow/progress_render.py

@@ -100,6 +100,12 @@
         reported even when the active tree is momentarily empty."""
         return self._registry.global_counters()
 
+    def assets_ingested(self) -> int:
+        """Monotonic run-total of leaf assets ingested (tables + stored
+        procedures + other leaf entities). Independent of the tree, so it stays
+        correct after scope pruning and is authoritative on the terminal event."""
+        return self._registry.assets_ingested()
+
     def _driver_counter(
         self,
         counters: "Optional[List[Tuple[str, int, Optional[int]]]]" = None,  # noqa: UP006,UP045
@@ -136,14 +142,14 @@
         result = None
         if snapshot is not None:
             tree = snapshot_to_progress_payload(snapshot)
             if tree is not None:
                 tree["processed"] = self._registry.assets_ingested()
                 tree["expected"] = None
                 result = tree
         return result
 
 
 def _render_joined(node: ProgressNodeSnapshot, ancestors: "List[str]", lines: "List[str]") -> None:  # noqa: UP006
     """Render active nodes with ancestor labels joined by '.' so sibling-level
     hierarchy collapses into a single readable label, e.g. ``sales.public  Table 45/310``."""
     path = [*ancestors, node.label] if node.label else list(ancestors)

ingestion/src/metadata/workflow/workflow_status_mixin.py

@@ -203,6 +203,7 @@ def send_progress_update(self, update_type: ProgressUpdateType = ProgressUpdateT
                 progress_data = reporter.payload() if reporter is not None else None
                 counters = reporter.global_counters() if reporter is not None else []
                 eta_seconds = reporter.eta_seconds() if reporter is not None else None
+                total_assets = reporter.assets_ingested() if reporter is not None else None
 
                 progress_update = ProgressUpdate(
                     runId=self.run_id,
@@ -213,6 +214,7 @@ def send_progress_update(self, update_type: ProgressUpdateType = ProgressUpdateT
                         {"entityType": type_, "done": done, "total": total} for type_, done, total in counters
                     ],
                     estimatedSecondsRemaining=eta_seconds,
+                    totalAssetsIngested=total_assets,
                     stepName=None,
                     currentEntity=None,
                     message=None,

ingestion/tests/unit/workflow/test_base_workflow.py

@@ -249,10 +249,13 @@ def test_close_steps_runs_before_status_publishing_and_stop(self):
             workflow.execute()
 
         ordered_names = [mock_call[0] for mock_call in manager.mock_calls]
-        # `close_steps` is recorded twice: once from execute() before
-        # print_status, and once from inside stop() to keep the public
-        # cleanup contract. The second call is a no-op via _steps_closed.
+        # An initial `send_progress_update` fires at the very start so short
+        # runs (shorter than the reporting interval) still emit a "run started"
+        # event. `close_steps` is recorded twice: once from execute() before
+        # print_status, and once from inside stop() to keep the public cleanup
+        # contract. The second call is a no-op via _steps_closed.
         assert ordered_names == [
+            "send_progress_update",
             "close_steps",
             "build_ingestion_status",
             "set_ingestion_pipeline_status",
@@ -261,7 +264,8 @@ def test_close_steps_runs_before_status_publishing_and_stop(self):
             "stop",
             "close_steps",
         ]
-        mock_send_progress_update.assert_called_once_with(ProgressUpdateType.ERROR)
+        progress_types = [call.args[0] for call in mock_send_progress_update.call_args_list]
+        assert progress_types == [ProgressUpdateType.DISCOVERY, ProgressUpdateType.ERROR]
 
     def test_success_states_map_to_pipeline_complete_progress(self):
         workflow = SimpleWorkflow(config=config)
@@ -281,7 +285,40 @@ def test_failed_workflow_sends_terminal_error_progress(self):
         ) as mock_send_progress_update:
             workflow.execute()
 
-        mock_send_progress_update.assert_called_once_with(ProgressUpdateType.ERROR)
+        progress_types = [call.args[0] for call in mock_send_progress_update.call_args_list]
+        assert progress_types == [ProgressUpdateType.DISCOVERY, ProgressUpdateType.ERROR]
+
+    def test_execute_emits_initial_progress_before_work(self):
+        """A run shorter than the reporting interval still emits events: an
+        initial DISCOVERY at start (so the run is registered/visible live) and
+        a terminal update at the end — never zero events."""
+        workflow = SimpleWorkflow(config=config)
+
+        emitted = []
+        original = workflow.send_progress_update
+
+        def record(update_type=ProgressUpdateType.PROCESSING):
+            emitted.append((update_type, workflow.execute_internal_started))
+            return original(update_type)
+
+        workflow.execute_internal_started = False
+        real_execute_internal = workflow.execute_internal
+
+        def flag_execute_internal():
+            workflow.execute_internal_started = True
+            return real_execute_internal()
+
+        with (
+            patch.object(workflow, "send_progress_update", side_effect=record),
+            patch.object(workflow, "execute_internal", side_effect=flag_execute_internal),
+        ):
+            workflow.execute()
+
+        # First emit is DISCOVERY and happens before execute_internal runs.
+        assert emitted[0][0] is ProgressUpdateType.DISCOVERY
+        assert emitted[0][1] is False
+        # At least the initial + a terminal event, so never zero.
+        assert len(emitted) >= 2
 
     def test_stop_still_runs_when_print_status_raises(self):
         workflow = SimpleWorkflow(config=config)

ingestion/tests/unit/workflow/test_progress_rendering.py

@@ -35,6 +35,30 @@ def _tree_snapshot():
     return reg.snapshot()
 
 
+class TestReporterAssetsIngested:
+    def test_counts_all_leaf_types(self):
+        reg = ProgressRegistry()
+        reg.open([], "Database", None)
+        reg.open(["db"], "DatabaseSchema", 1)
+        reg.open(["db", "sch"], "Table", 3)
+        reg.open(["db", "sch"], "StoredProcedure", 2)
+        for _ in range(3):
+            reg.advance(["db", "sch"], "Table")
+        for _ in range(2):
+            reg.advance(["db", "sch"], "StoredProcedure")
+        assert ProgressReporter(reg).assets_ingested() == 5
+
+    def test_survives_scope_pruning(self):
+        reg = ProgressRegistry()
+        reg.open([], "Database", None)
+        reg.open(["db", "sch"], "Table", 2)
+        reg.advance(["db", "sch"], "Table")
+        reg.advance(["db", "sch"], "Table")
+        reg.close(["db", "sch"])
+        reg.close(["db"])
+        assert ProgressReporter(reg).assets_ingested() == 2
+
+
 class TestProgressRendering:
     def test_cli_renders_database_line(self):
         out = render_progress_tree(_tree_snapshot())

ingestion/tests/unit/workflow/test_status_mixin_progress.py

@@ -13,6 +13,9 @@
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+from metadata.generated.schema.entity.services.ingestionPipelines.progressUpdate import (
+    ProgressUpdateType,
+)
 from metadata.utils.progress_registry import ProgressRegistry
 from metadata.workflow.workflow_status_mixin import WorkflowStatusMixin
 
@@ -60,3 +63,39 @@ def test_null_when_not_computable(self):
         reg.set_total("DatabaseSchema", 10)
         _Harness(reg, metadata).send_progress_update()
         assert _sent_update(metadata).estimatedSecondsRemaining is None
+
+
+class TestTotalAssetsIngested:
+    def test_processing_update_carries_leaf_total(self):
+        reg = ProgressRegistry()
+        metadata = MagicMock()
+        reg.open([], "Database", None)
+        reg.open(["db", "sch"], "Table", 3)
+        reg.open(["db", "sch"], "StoredProcedure", 2)
+        for _ in range(3):
+            reg.advance(["db", "sch"], "Table")
+        for _ in range(2):
+            reg.advance(["db", "sch"], "StoredProcedure")
+        _Harness(reg, metadata).send_progress_update()
+        assert _sent_update(metadata).totalAssetsIngested == 5
+
+    def test_terminal_update_carries_final_total_after_pruning(self):
+        reg = ProgressRegistry()
+        metadata = MagicMock()
+        reg.open([], "Database", None)
+        reg.open(["db", "sch"], "Table", 2)
+        reg.advance(["db", "sch"], "Table")
+        reg.advance(["db", "sch"], "Table")
+        reg.close(["db", "sch"])
+        reg.close(["db"])
+        _Harness(reg, metadata).send_progress_update(ProgressUpdateType.PIPELINE_COMPLETE)
+        update = _sent_update(metadata)
+        assert update.updateType is ProgressUpdateType.PIPELINE_COMPLETE
+        assert update.totalAssetsIngested == 2
+
+    def test_null_when_no_registry(self):
+        metadata = MagicMock()
+        harness = _Harness(ProgressRegistry(), metadata)
+        harness._steps = []
+        harness.send_progress_update()
+        assert _sent_update(metadata).totalAssetsIngested is None

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java

@@ -1313,10 +1313,29 @@ public RestUtil.PutResponse<?> updateProgress(
       return new RestUtil.PutResponse<>(Response.Status.OK, progressUpdate, ENTITY_FIELDS_CHANGED);
     }
 
-    progressTracker.updateProgress(fqn, runId, progressUpdate);
+    progressTracker.updateProgress(
+        fqn, runId, progressUpdate, discoveredPipeline(fqn, progressUpdate));
     return new RestUtil.PutResponse<>(Response.Status.OK, progressUpdate, ENTITY_FIELDS_CHANGED);
   }
 
+  /**
+   * The pipeline entity to attach to the run's opening event, so the UI can discover a newly
+   * created agent from the stream itself. Loaded only for the DISCOVERY update (once per run) and
+   * never fails the progress update — a lookup error just omits the entity.
+   */
+  private IngestionPipeline discoveredPipeline(String fqn, ProgressUpdate update) {
+    IngestionPipeline result = null;
+    if (update.getUpdateType() == ProgressUpdateType.DISCOVERY) {
+      try {
+        result = getByName(null, fqn, getFields(""));
+      } catch (Exception e) {
+        LOG.debug(
+            "Could not load ingestion pipeline {} for progress discovery: {}", fqn, e.getMessage());
+      }
+    }
+    return result;
+  }
+
   public RestUtil.PutResponse<?> addOperationMetrics(
       String fqn, UUID runId, OperationMetricsBatch batch) {
     if (progressTracker == null) {

openmetadata-service/src/main/java/org/openmetadata/service/monitoring/IngestionProgressTracker.java

@@ -31,6 +31,7 @@
 import javax.inject.Inject;
 import javax.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
+import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline;
 import org.openmetadata.schema.entity.services.ingestionPipelines.OperationMetricsBatch;
 import org.openmetadata.schema.entity.services.ingestionPipelines.ProgressUpdate;
 import org.openmetadata.schema.entity.services.ingestionPipelines.ProgressUpdateType;
@@ -87,6 +88,11 @@ public IngestionProgressTracker(MeterRegistry meterRegistry) {
   }
 
   public void updateProgress(String pipelineFqn, UUID runId, ProgressUpdate update) {
+    updateProgress(pipelineFqn, runId, update, null);
+  }
+
+  public void updateProgress(
+      String pipelineFqn, UUID runId, ProgressUpdate update, IngestionPipeline ingestionPipeline) {
     String key = buildKey(pipelineFqn, runId);
     progressUpdatesReceived.increment();
 
@@ -98,7 +104,7 @@ public void updateProgress(String pipelineFqn, UUID runId, ProgressUpdate update
     }
 
     notifyListeners(key, update);
-    routeToService(pipelineFqn, runId, update);
+    routeToService(pipelineFqn, runId, update, ingestionPipeline);
   }
 
   public void addMetricsBatch(String pipelineFqn, UUID runId, OperationMetricsBatch batch) {
@@ -169,15 +175,17 @@ private String buildKey(String pipelineFqn, UUID runId) {
     return pipelineFqn + "/" + runId.toString();
   }
 
-  private void routeToService(String pipelineFqn, UUID runId, ProgressUpdate update) {
+  private void routeToService(
+      String pipelineFqn, UUID runId, ProgressUpdate update, IngestionPipeline ingestionPipeline) {
     String serviceFqn = FullyQualifiedName.getParentFQN(pipelineFqn);
     if (!nullOrEmpty(serviceFqn)) {
       String runKey = buildKey(pipelineFqn, runId);
       ServiceProgressEvent event =
           new ServiceProgressEvent()
               .withPipelineFqn(pipelineFqn)
               .withRunId(runId.toString())
-              .withEvent(update);
+              .withEvent(update)
+              .withIngestionPipeline(ingestionPipeline);
       recordActiveRun(serviceFqn, runKey, event);
       notifyServiceListeners(serviceFqn, event);
       dropRunIfTerminal(serviceFqn, runKey, update);

openmetadata-service/src/test/java/org/openmetadata/service/monitoring/IngestionProgressTrackerTest.java

@@ -23,6 +23,7 @@
 import java.util.function.Consumer;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline;
 import org.openmetadata.schema.entity.services.ingestionPipelines.OperationMetric;
 import org.openmetadata.schema.entity.services.ingestionPipelines.OperationMetricsBatch;
 import org.openmetadata.schema.entity.services.ingestionPipelines.ProgressNode;
@@ -333,6 +334,40 @@ void testUnregisterServiceListenerStopsDelivery() {
     assertEquals(1, received.size());
   }
 
+  @Test
+  void serviceEventCarriesIngestionPipelineWhenProvided() {
+    UUID run = UUID.randomUUID();
+    List<ServiceProgressEvent> received = new ArrayList<>();
+    tracker.registerServiceListener("svc", received::add);
+
+    IngestionPipeline pipeline =
+        new IngestionPipeline().withName("metadata").withFullyQualifiedName("svc.metadata");
+    tracker.updateProgress("svc.metadata", run, discovery(run), pipeline);
+
+    assertEquals(1, received.size());
+    assertNotNull(received.get(0).getIngestionPipeline());
+    assertEquals("svc.metadata", received.get(0).getIngestionPipeline().getFullyQualifiedName());
+  }
+
+  @Test
+  void serviceEventHasNoIngestionPipelineByDefault() {
+    UUID run = UUID.randomUUID();
+    List<ServiceProgressEvent> received = new ArrayList<>();
+    tracker.registerServiceListener("svc", received::add);
+
+    tracker.updateProgress("svc.metadata", run, processing(run, "p"));
+
+    assertEquals(1, received.size());
+    assertNull(received.get(0).getIngestionPipeline());
+  }
+
+  private static ProgressUpdate discovery(UUID runId) {
+    return new ProgressUpdate()
+        .withRunId(runId.toString())
+        .withTimestamp(System.currentTimeMillis())
+        .withUpdateType(ProgressUpdateType.DISCOVERY);
+  }
+
   private static ProgressUpdate processing(UUID runId, String msg) {
     return new ProgressUpdate()
         .withRunId(runId.toString())

openmetadata-spec/src/main/resources/json/schema/entity/services/ingestionPipelines/progressUpdate.json

@@ -91,6 +91,10 @@
       "description": "Estimated seconds until the run completes; null when not yet computable",
       "type": ["integer", "null"]
     },
+    "totalAssetsIngested": {
+      "description": "Monotonic count of leaf assets ingested so far (tables + stored procedures + other leaf entities). Survives scope pruning and is present on the terminal PIPELINE_COMPLETE event, so it carries the run's final asset total.",
+      "type": ["integer", "null"]
+    },
     "currentEntity": {
       "description": "FQN of the entity currently being processed",
       "type": "string"

openmetadata-spec/src/main/resources/json/schema/entity/services/ingestionPipelines/serviceProgressEvent.json

@@ -17,6 +17,10 @@
     "event": {
       "description": "The per-run progress update",
       "$ref": "progressUpdate.json"
+    },
+    "ingestionPipeline": {
+      "description": "The ingestion pipeline (agent) this run belongs to. Populated on the DISCOVERY event that opens a run, so the UI can discover a newly created agent without a separate lookup.",
+      "$ref": "ingestionPipeline.json"
     }
   },
   "required": ["pipelineFqn", "runId", "event"],

openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/progressUpdate.ts

@@ -47,6 +47,12 @@ export interface ProgressUpdate {
      * When this update was created
      */
     timestamp: number;
+    /**
+     * Monotonic count of leaf assets ingested so far (tables + stored procedures + other leaf
+     * entities). Survives scope pruning and is present on the terminal PIPELINE_COMPLETE event,
+     * so it carries the run's final asset total.
+     */
+    totalAssetsIngested?: number | null;
     /**
      * Type of progress update
      */

openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/serviceProgressEvent.ts

@@ -67,6 +67,12 @@ export interface ProgressUpdate {
      * When this update was created
      */
     timestamp: number;
+    /**
+     * Monotonic count of leaf assets ingested so far (tables + stored procedures + other leaf
+     * entities). Survives scope pruning and is present on the terminal PIPELINE_COMPLETE event,
+     * so it carries the run's final asset total.
+     */
+    totalAssetsIngested?: number | null;
     /**
      * Type of progress update
      */