diff --git a/python/feldera/pipeline.py b/python/feldera/pipeline.py
index ce135e29a16..4869dde141d 100644
--- a/python/feldera/pipeline.py
+++ b/python/feldera/pipeline.py
@@ -387,11 +387,19 @@ def wait_for_completion(
             raise RuntimeError("Pipeline must be running to wait for completion")
 
         start_time = time.monotonic()
-        long_op = LongOperationWarning(
+        # Reused by the warning message below; is_complete() would fetch and
+        # discard the same stats, hiding whether the wait is stalled or just
+        # slow (e.g. mid-transaction-commit).
+        latest_stats: Optional[PipelineStatistics] = None
+        wait_for_completion = LongOperationWarning(
             logger,
-            lambda elapsed: f"still waiting for pipeline {self.name} to complete, "
-            f"waited {elapsed:.1f} seconds",
-            lambda elapsed: f"pipeline {self.name} completed after {elapsed:.1f} seconds",
+            lambda elapsed: (
+                f"still waiting for pipeline {self.name} to complete, "
+                f"waited {elapsed:.1f} seconds ({latest_stats.global_metrics.progress_summary()})"
+            ),
+            lambda elapsed: (
+                f"pipeline {self.name} completed after {elapsed:.1f} seconds"
+            ),
         )
 
         while True:
@@ -403,16 +411,19 @@ def wait_for_completion(
                         f" pipeline '{self.name}' to complete"
                     )
 
-            pipeline_complete: bool = self.is_complete()
+            latest_stats = self.stats()
+            pipeline_complete: Optional[bool] = (
+                latest_stats.global_metrics.pipeline_complete
+            )
             if pipeline_complete is None:
                 raise RuntimeError(
                     "received unknown metrics from the pipeline, pipeline_complete is None"
                 )
             elif pipeline_complete:
-                long_op.done()
+                wait_for_completion.done()
                 break
 
-            long_op.check()
+            wait_for_completion.check()
             time.sleep(1)
 
         if force_stop:
diff --git a/python/feldera/rest/feldera_client.py b/python/feldera/rest/feldera_client.py
index 7f35d1e7aab..07328bf32a3 100644
--- a/python/feldera/rest/feldera_client.py
+++ b/python/feldera/rest/feldera_client.py
@@ -176,7 +176,7 @@ def _wait_for_compilation(
         """Wait for pipeline compilation -- internal use only."""
         wait = ["Pending", "CompilingSql", "SqlCompiled", "CompilingRust"]
         start_time = time.monotonic()
-        long_op = LongOperationWarning(
+        wait_for_compilation = LongOperationWarning(
             logger,
             lambda elapsed: f"still compiling {name}, waited {elapsed:.1f} seconds",
             lambda elapsed: f"{name} finished compiling after {elapsed:.1f} seconds",
@@ -193,7 +193,7 @@ def _wait_for_compilation(
             status = p.program_status
 
             if status == "Success":
-                long_op.done()
+                wait_for_compilation.done()
                 if expected_program_version is None:
                     return self.get_pipeline(name, PipelineFieldSelector.ALL)
 
@@ -231,7 +231,7 @@ def _wait_for_compilation(
 
                 raise RuntimeError(error_message)
 
-            long_op.check()
+            wait_for_compilation.check()
             time.sleep(poll_interval_s)
 
     def __wait_for_pipeline_state(
@@ -243,7 +243,7 @@ def __wait_for_pipeline_state(
         poll_interval_s: float = 0.5,
     ):
         start_time = time.monotonic()
-        long_op = LongOperationWarning(
+        wait_for_state = LongOperationWarning(
             logger,
             lambda elapsed: f"still waiting for {pipeline_name} to transition to "
             f"'{state}', waited {elapsed:.1f} seconds",
@@ -264,7 +264,7 @@ def __wait_for_pipeline_state(
             status = resp.deployment_status
 
             if status.lower() == state.lower():
-                long_op.done()
+                wait_for_state.done()
                 break
             elif (
                 status == "Stopped"
@@ -278,7 +278,7 @@ def __wait_for_pipeline_state(
 {resp.deployment_error.get("message", "")}"""
                 )
 
-            long_op.check()
+            wait_for_state.check()
             time.sleep(poll_interval_s)
 
     def __wait_for_pipeline_state_one_of(
@@ -291,7 +291,7 @@ def __wait_for_pipeline_state_one_of(
     ) -> PipelineStatus:
         start_time = time.monotonic()
         states = [state.lower() for state in states]
-        long_op = LongOperationWarning(
+        wait_for_states = LongOperationWarning(
             logger,
             lambda elapsed: f"still waiting for {pipeline_name} to transition to "
             f"one of {states}, waited {elapsed:.1f} seconds",
@@ -311,7 +311,7 @@ def __wait_for_pipeline_state_one_of(
             status = resp.deployment_status
 
             if status.lower() in states:
-                long_op.done()
+                wait_for_states.done()
                 return PipelineStatus.from_str(status)
             elif (
                 status == "Stopped"
@@ -324,7 +324,7 @@ def __wait_for_pipeline_state_one_of(
 Reason: The pipeline is in a STOPPED state due to the following error:
 {resp.deployment_error.get("message", "")}"""
                 )
-            long_op.check()
+            wait_for_states.check()
             time.sleep(poll_interval_s)
 
     def create_pipeline(self, pipeline: Pipeline, wait: bool = True) -> Pipeline:
@@ -762,7 +762,7 @@ def stop_pipeline(
             return
 
         start = time.monotonic()
-        long_op = LongOperationWarning(
+        wait_for_stop = LongOperationWarning(
             logger,
             lambda elapsed: f"still stopping {pipeline_name}, waited {elapsed:.1f} seconds",
             lambda elapsed: f"{pipeline_name} stopped after {elapsed:.1f} seconds",
@@ -779,10 +779,10 @@ def stop_pipeline(
             ).deployment_status
 
             if status == "Stopped":
-                long_op.done()
+                wait_for_stop.done()
                 return
 
-            long_op.check()
+            wait_for_stop.check()
             time.sleep(0.1)
 
     def dismiss_error_pipeline(
@@ -823,7 +823,7 @@ def clear_storage(
             return
 
         start = time.monotonic()
-        long_op = LongOperationWarning(
+        wait_for_clear = LongOperationWarning(
             logger,
             lambda elapsed: f"still clearing {pipeline_name}, waited {elapsed:.1f} seconds",
             lambda elapsed: f"{pipeline_name} storage cleared after {elapsed:.1f} seconds",
@@ -838,10 +838,10 @@ def clear_storage(
             ).storage_status
 
             if status == "Cleared":
-                long_op.done()
+                wait_for_clear.done()
                 return
 
-            long_op.check()
+            wait_for_clear.check()
             time.sleep(poll_interval_s)
 
     def start_transaction(self, pipeline_name: str) -> int:
@@ -1037,7 +1037,7 @@ def commit_transaction(
         if not wait:
             return
 
-        long_op = LongOperationWarning(
+        wait_for_commit = LongOperationWarning(
             logger,
             lambda elapsed: f"transaction {transaction_id} on {pipeline_name} "
             f"hasn't committed, waited {elapsed:.1f} seconds",
@@ -1052,10 +1052,10 @@ def commit_transaction(
 
             stats = self.get_pipeline_stats(pipeline_name)
             if stats["global_metrics"]["transaction_id"] != transaction_id:
-                long_op.done()
+                wait_for_commit.done()
                 return
 
-            long_op.check()
+            wait_for_commit.check()
             time.sleep(poll_interval_s)
 
     def checkpoint_pipeline(self, pipeline_name: str) -> int:
@@ -1268,7 +1268,7 @@ def wait_for_token(
         max_backoff = 5
         exponent = 1.2
         retries = 0
-        long_op = LongOperationWarning(
+        wait_for_token_processed = LongOperationWarning(
             logger,
             lambda elapsed: f"still waiting for inputs represented by {token} "
             f"to be processed, waited {elapsed:.1f} seconds",
@@ -1286,10 +1286,10 @@ def wait_for_token(
                     )
 
             if self.completion_token_processed(pipeline_name, token):
-                long_op.done()
+                wait_for_token_processed.done()
                 break
 
-            long_op.check()
+            wait_for_token_processed.check()
 
             retries += 1
             backoff = min(max_backoff, initial_backoff * (exponent**retries))
diff --git a/python/feldera/stats.py b/python/feldera/stats.py
index 4f28cca4396..f5d067e9538 100644
--- a/python/feldera/stats.py
+++ b/python/feldera/stats.py
@@ -86,8 +86,22 @@ def from_dict(cls, d: Mapping[str, Any]):
         metrics.start_time = datetime.fromtimestamp(d["start_time"])
         metrics.initial_start_time = datetime.fromtimestamp(d["start_time"])
         metrics.transaction_status = TransactionStatus.from_str(d["transaction_status"])
+        commit_progress = d.get("commit_progress")
+        metrics.commit_progress = (
+            CommitProgressSummary.from_dict(commit_progress)
+            if commit_progress is not None
+            else None
+        )
         return metrics
 
+    def progress_summary(self) -> str:
+        """Human-readable progress indicator for long-running wait loops:
+        commit progress while a transaction is committing, otherwise the
+        overall record count processed so far."""
+        if self.commit_progress is not None:
+            return str(self.commit_progress)
+        return f"{self.total_processed_records}/{self.total_input_records} records processed"
+
 
 class ConnectorError:
     """Represents a connector error item reported by connector status endpoints."""
@@ -334,6 +348,16 @@ def from_dict(cls, d: Mapping[str, Any]):
         status.__dict__.update(d)
         return status
 
+    def __str__(self) -> str:
+        # Mirrors CommitProgressSummary's Display impl in
+        # crates/feldera-types/src/transaction.rs.
+        return (
+            f"completed: {self.completed} operators, "
+            f"evaluating: {self.in_progress} operators "
+            f"[{self.in_progress_processed_records}/{self.in_progress_total_records} "
+            f"changes processed], remaining: {self.remaining} operators"
+        )
+
 
 class TransactionInitiators:
     """Initiators for an ongoing transaction."""
diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py
index 6f877b14792..d3091a46ecf 100644
--- a/python/feldera/testutils.py
+++ b/python/feldera/testutils.py
@@ -1,5 +1,6 @@
 "Utility functions for writing tests against a Feldera instance."
 
+import logging
 import math
 import os
 import platform
@@ -10,6 +11,7 @@
 from typing import List, Optional, cast
 from datetime import datetime
 
+from feldera._long_operation_warning import LongOperationWarning
 from feldera.enums import CompilationProfile
 from feldera.pipeline import Pipeline
 from feldera.pipeline_builder import PipelineBuilder
@@ -17,6 +19,8 @@
 from feldera.rest import FelderaClient
 from feldera.rest._helpers import requests_verify_from_env
 
+logger = logging.getLogger(__name__)
+
 API_KEY = os.environ.get("FELDERA_API_KEY")
 
 
@@ -359,9 +363,34 @@ def check_end_of_input(pipeline: Pipeline) -> bool:
 
 def wait_end_of_input(pipeline: Pipeline, timeout_s: Optional[int] = None):
     start_time = time.monotonic()
-    while not check_end_of_input(pipeline):
+    # Reused by the warning message below so a stalled ingest can be told
+    # apart from a slow one; check_end_of_input() would fetch and discard it.
+    latest_stats = None
+    wait_for_input_end = LongOperationWarning(
+        logger,
+        lambda elapsed: (
+            f"still waiting for end of input on pipeline {pipeline.name}, "
+            f"waited {elapsed:.1f} seconds ({latest_stats.global_metrics.progress_summary()})"
+        ),
+        lambda elapsed: (
+            f"end of input reached on pipeline {pipeline.name} "
+            f"after {elapsed:.1f} seconds"
+        ),
+    )
+
+    while True:
+        latest_stats = pipeline.stats()
+        if all(
+            input_endpoint.metrics.end_of_input
+            for input_endpoint in latest_stats.inputs
+        ):
+            wait_for_input_end.done()
+            return
+
         if timeout_s is not None and time.monotonic() - start_time > timeout_s:
             raise TimeoutError("Timeout waiting for end of input")
+
+        wait_for_input_end.check()
         time.sleep(3)
 
 
diff --git a/python/tests/unit/test_stats.py b/python/tests/unit/test_stats.py
new file mode 100644
index 00000000000..b792f9d2bb3
--- /dev/null
+++ b/python/tests/unit/test_stats.py
@@ -0,0 +1,74 @@
+"""Tests for GlobalPipelineMetrics.progress_summary(), which lets wait loops
+(Pipeline.wait_for_completion, testutils.wait_end_of_input) report whether a
+long wait is stalled or just slow."""
+
+from feldera.stats import CommitProgressSummary, GlobalPipelineMetrics
+
+BASE_METRICS_DICT = {
+    "state": "running",
+    "incarnation_uuid": "00000000-0000-0000-0000-000000000000",
+    "start_time": 0,
+    "transaction_status": "NoTransaction",
+    "total_processed_records": 42,
+    "total_input_records": 100,
+}
+
+
+def test_commit_progress_absent_when_no_transaction():
+    metrics = GlobalPipelineMetrics.from_dict(BASE_METRICS_DICT)
+    assert metrics.commit_progress is None
+
+
+def test_commit_progress_parsed_into_object():
+    d = {
+        **BASE_METRICS_DICT,
+        "commit_progress": {
+            "completed": 3,
+            "in_progress": 2,
+            "remaining": 1,
+            "in_progress_processed_records": 10,
+            "in_progress_total_records": 50,
+        },
+    }
+    metrics = GlobalPipelineMetrics.from_dict(d)
+    assert isinstance(metrics.commit_progress, CommitProgressSummary)
+    assert metrics.commit_progress.completed == 3
+    assert metrics.commit_progress.in_progress_total_records == 50
+
+
+def test_commit_progress_str_matches_rust_display_format():
+    progress = CommitProgressSummary.from_dict(
+        {
+            "completed": 3,
+            "in_progress": 2,
+            "remaining": 1,
+            "in_progress_processed_records": 10,
+            "in_progress_total_records": 50,
+        }
+    )
+    assert str(progress) == (
+        "completed: 3 operators, evaluating: 2 operators "
+        "[10/50 changes processed], remaining: 1 operators"
+    )
+
+
+def test_progress_summary_falls_back_to_record_counts_outside_transaction():
+    metrics = GlobalPipelineMetrics.from_dict(BASE_METRICS_DICT)
+    assert metrics.progress_summary() == "42/100 records processed"
+
+
+def test_progress_summary_prefers_commit_progress_during_transaction():
+    d = {
+        **BASE_METRICS_DICT,
+        "commit_progress": {
+            "completed": 0,
+            "in_progress": 1,
+            "remaining": 7,
+            "in_progress_processed_records": 0,
+            "in_progress_total_records": 169,
+        },
+    }
+    metrics = GlobalPipelineMetrics.from_dict(d)
+    summary = metrics.progress_summary()
+    assert "0/169 changes processed" in summary
+    assert "records processed" not in summary