feat: expose structured job progress to SDK callers (#4808)

Job callers currently lose the server's structured progress when
describing, listing, or waiting for a job. Expose Rust and Python
progress snapshots and optional wait observers using the existing
backend poller. Python supports callbacks, a caller-owned tqdm-like bar,
or `progress=True`; `status()` still returns a string and the default
wait behavior is unchanged.

Preserve unknown progress fields and phases from newer servers, and
report omitted counters as unknown. Each wait calibrates a caller-owned
bar from its first complete snapshot, then uses a high-water mark within
the same execution epoch; a new epoch or smaller plan resets the bar.
ETA is best-effort and omitted for missing, stale, paused, or
incompatible samples.

Newly submitted index and refresh handles forward observers through
their freshness wrapper while retaining successful-wait read
consistency. The Jobs guide includes draft Python/Rust snapshot and
observer examples for release; TypeScript progress support remains
tracked under ENT-3500.

Validation on the development EC2 host: the initial 11 Python progress
tests, REST fixture caller, and two Rust progress regressions passed.
After fixing freshness-wrapper forwarding, rebuilt the Python extension
and ran two submitted-handle Python tests plus the new Rust refresh and
index regressions (one test each); all passed. Strict Clippy, rustfmt,
and targeted Ruff passed. The final commit only wraps the new guide
section in an MDX draft and clarifies language support. Repository-wide
Ruff reports 48 existing findings in six unchanged docs/web-tests files.

This publishes the SDK portion of ENT-3500. Server-side live-split and
terminal-progress changes and the combined server/SDK end-to-end
scenario remain in progress. Refs ENT-3500.

BREAKING CHANGE: Rust struct literals for `JobInfo` must add `progress:
None`, `phase: None`, and `paused: false`. Literals for `JobDescription`
must add `progress: None`, `phase: None`, `queue_reason: None`,
`waiting_for_job_id: None`, and `paused_at_millis: None` when that
metadata is unavailable. Existing Python calls remain compatible.

python/python/lancedb/_lancedb.pyi

@@ -1,6 +1,17 @@
 from datetime import date, datetime, timedelta
 from decimal import Decimal
-from typing import Dict, List, Optional, Sequence, Tuple, Any, TypedDict, Union, Literal
+from typing import (
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Sequence,
+    Tuple,
+    Any,
+    TypedDict,
+    Union,
+    Literal,
+)
 from uuid import UUID
 
 import pyarrow as pa
@@ -388,7 +399,10 @@ class Job:
     @property
     def _description(self) -> Optional[JobDescription]: ...
     async def status(self) -> str: ...
-    async def wait(self) -> Optional[str]: ...
+    async def wait(
+        self,
+        progress: Optional[Callable[[dict[str, Any]], None]] = None,
+    ) -> Optional[str]: ...
     async def cancel(self) -> None: ...
     async def refresh(self) -> None: ...
     async def events(
@@ -409,6 +423,10 @@ class JobInfo:
     def state(self) -> str: ...
     @property
     def created_at_millis(self) -> int: ...
+    @property
+    def progress(self) -> Optional[dict[str, Any]]: ...
+    @property
+    def paused(self) -> bool: ...
 
 class SessionStatus:
     @property
@@ -467,6 +485,8 @@ class JobDescription:
     @property
     def result(self) -> Optional[Any]: ...
     @property
+    def progress(self) -> Optional[dict[str, Any]]: ...
+    @property
     def failure(self) -> Optional[JobFailureInfo]: ...
 
 class SqlQuery:

python/python/lancedb/job.py

@@ -5,6 +5,7 @@
 
 import asyncio
 import json
+import time
 from datetime import timedelta
 from typing import Any, Callable, Generic, Optional, TypeVar, cast
 
@@ -49,6 +50,140 @@ def _timeout_message(
     )
 
 
+def _normalize_job_progress(progress):
+    """Normalize ``wait(progress=...)`` using the same convention as writes."""
+    if progress is True:
+        from tqdm.auto import tqdm
+
+        return tqdm(unit=" rows"), True
+    if progress is False or progress is None:
+        return None, False
+    return progress, False
+
+
+_ETA_MAX_SAMPLE_INTERVAL_MILLIS = 10 * 60 * 1_000
+
+
+def _progress_eta(previous, current):
+    """Best-effort remaining time from two same-execution observations."""
+    if not previous or current.get("state") not in ("running", "finished"):
+        return None
+    remaining = current.get("rows_remaining")
+    if current.get("state") == "finished":
+        return 0 if remaining == 0 else None
+    if current.get("paused") or current.get("phase") in (
+        "queued",
+        "starting",
+        "paused",
+    ):
+        return None
+    epoch = current.get("execution_epoch")
+    if epoch is None or previous.get("execution_epoch") != epoch:
+        return None
+    sample_at = current.get("sample_at_millis")
+    previous_at = previous.get("sample_at_millis")
+    if not isinstance(sample_at, int) or not isinstance(previous_at, int):
+        return None
+    elapsed = sample_at - previous_at
+    if not 0 < elapsed <= _ETA_MAX_SAMPLE_INTERVAL_MILLIS:
+        return None
+    if time.time() * 1_000 - sample_at > _ETA_MAX_SAMPLE_INTERVAL_MILLIS:
+        return None
+    planned = current.get("rows_planned")
+    if not isinstance(planned, int) or not isinstance(remaining, int):
+        return None
+    if remaining == 0:
+        return 0
+    if planned != previous.get("rows_planned"):
+        return None
+    if current.get("fragments_planned") != previous.get("fragments_planned"):
+        return None
+    accounted = current.get("rows_computed", 0) + current.get("rows_skipped", 0)
+    previous_accounted = previous.get("rows_computed", 0) + previous.get(
+        "rows_skipped", 0
+    )
+    advanced = accounted - previous_accounted
+    previous_remaining = previous.get("rows_remaining")
+    if advanced <= 0 or not isinstance(previous_remaining, int):
+        return None
+    if remaining > previous_remaining:
+        return None
+    return remaining * elapsed // advanced
+
+
+def _with_progress_eta(previous, current):
+    snapshot = dict(current)
+    estimate = _progress_eta(previous, snapshot)
+    if estimate is not None:
+        snapshot["estimated_remaining_time_millis"] = estimate
+    return snapshot
+
+
+def _job_progress_callback(progress, enrich):
+    """Adapt a callback or tqdm-like object to structured job snapshots."""
+    if progress is None:
+        return None
+    if callable(progress):
+        return lambda snapshot: progress(enrich(snapshot))
+
+    # A caller-owned bar can outlive one wait call. Resume from its displayed
+    # position instead of adding the same completed rows on the next call.
+    last_rows = getattr(progress, "n", 0) or 0
+    last_planned = getattr(progress, "total", None)
+    last_epoch = None
+    first_snapshot = True
+
+    def update(snapshot):
+        nonlocal first_snapshot, last_epoch, last_planned, last_rows
+        snapshot = enrich(snapshot)
+        planned = snapshot.get("rows_planned")
+        remaining = snapshot.get("rows_remaining")
+        computed = snapshot.get("rows_computed")
+        if isinstance(planned, int):
+            epoch = snapshot.get("execution_epoch")
+            reset = (isinstance(last_planned, int) and planned < last_planned) or (
+                isinstance(last_epoch, int)
+                and isinstance(epoch, int)
+                and epoch != last_epoch
+            )
+            if reset:
+                last_rows = 0
+                if hasattr(progress, "reset"):
+                    progress.reset(total=planned)
+                elif hasattr(progress, "n"):
+                    progress.n = 0
+            progress.total = planned
+            last_planned = planned
+            if isinstance(epoch, int):
+                last_epoch = epoch
+        completed = None
+        if isinstance(planned, int) and isinstance(remaining, int):
+            completed = max(planned - remaining, 0)
+        elif computed is not None:
+            completed = computed
+        if completed is not None:
+            if isinstance(planned, int):
+                completed = min(completed, planned)
+            if first_snapshot and completed < last_rows:
+                # The server may have started a new execution while a
+                # caller-owned bar retained its old position.
+                last_rows = completed
+                if hasattr(progress, "n"):
+                    progress.n = completed
+                if hasattr(progress, "refresh"):
+                    progress.refresh()
+            if completed > last_rows:
+                delta = completed - last_rows
+                last_rows = completed
+                progress.update(delta)
+            first_snapshot = False
+        phase = snapshot.get("phase") or snapshot.get("state")
+        if phase is not None and hasattr(progress, "set_postfix_str"):
+            progress.set_postfix_str(phase)
+
+    return update
+
+
 class AsyncJob(Generic[T]):
     """A handle to an operation that may still be running.
 
@@ -64,6 +199,12 @@ class AsyncJob(Generic[T]):
     ):
         self._inner = inner
         self._result_decoder = result_decoder
+        self._last_progress_snapshot = None
+
+    def _enrich_progress(self, snapshot):
+        enriched = _with_progress_eta(self._last_progress_snapshot, snapshot)
+        self._last_progress_snapshot = enriched
+        return enriched
 
     @property
     def id(self) -> Optional[str]:
@@ -88,12 +229,44 @@ class AsyncJob(Generic[T]):
             return "finished"
         return await self._inner.status()
 
-    async def wait(self, timeout: Optional[timedelta] = None) -> T:
+    async def progress(self) -> Optional[dict[str, Any]]:
+        """Fetch a structured progress snapshot.
+
+        Registry jobs always report lifecycle fields. Queue metadata and
+        counters are included when supplied by the server. Older servers may
+        omit counters, and in-process operations return ``None``. A missing
+        counter is unknown rather than zero. This is additive to :meth:`status`,
+        whose string contract is unchanged.
+        """
+        if self._inner is None:
+            return None
+        await self._inner.status()
+        description = getattr(self._inner, "_description", None)
+        if description is None or description.progress is None:
+            return None
+        return self._enrich_progress(description.progress)
+
+    async def wait(
+        self,
+        timeout: Optional[timedelta] = None,
+        progress: Any = False,
+    ) -> T:
         """Wait until the operation reaches a terminal state.
 
         Returns the endpoint's typed result, or ``None`` for a unit-result
         job.
 
+        ``progress=True`` renders a progress bar, ``progress=callable``
+        receives each structured snapshot, and a tqdm-like object is updated in
+        place. Observations use the backend's existing wait poller and never
+        start a second job. The bar advances by remaining rows when available,
+        because completed computation can include repeated work after a plan
+        change. On each wait call, the first complete observation synchronizes
+        a caller-owned bar with the server's current completed rows and total;
+        later observations in a stable execution epoch use a high-water mark. A
+        new execution epoch or smaller plan resets the bar to that plan.
+        ``False`` preserves the original wait behavior.
+
         Raises `JobFailedError` if the operation failed, `JobCancelledError`
         if it was cancelled, and `TimeoutError` if `timeout` elapses first.
         Transient connection and read errors are retried with capped exponential
@@ -109,6 +282,8 @@ class AsyncJob(Generic[T]):
         """
         if self._inner is None:
             return cast(T, None)
+        progress_obj, owns = _normalize_job_progress(progress)
+        progress_callback = _job_progress_callback(progress_obj, self._enrich_progress)
         last_error = None
 
         async def wait_with_retries():
@@ -122,7 +297,9 @@ class AsyncJob(Generic[T]):
                         await self._inner.status()
                         last_error = None
                         delay = 0.25
-                    return await self._inner.wait()
+                    if progress_callback is None:
+                        return await self._inner.wait()
+                    return await self._inner.wait(progress=progress_callback)
                 except (HttpError, RetryError) as error:
                     cause = error.__cause__ if isinstance(error, RetryError) else error
                     if not isinstance(cause, HttpError) or not cause.is_transport_error:
@@ -132,32 +309,38 @@ class AsyncJob(Generic[T]):
                 delay = min(delay * 2, 30.0)
 
         try:
-            if timeout is None:
-                result = await wait_with_retries()
-            else:
-                result = await asyncio.wait_for(
-                    wait_with_retries(), timeout.total_seconds()
-                )
-        except asyncio.TimeoutError as error:
-            if last_error is None:
+            try:
                 if timeout is None:
-                    raise
-                raise TimeoutError(
-                    _timeout_message(self.id, timeout, self.job_type)
-                ) from error
-            identity = f"job {self.id!r}" if self.id is not None else "the operation"
-            if self.id is not None and self.job_type != "query":
-                reconnect = (
-                    f"Use db.open_job({self.id!r}) to reconnect, then call wait() "
-                    "on the returned job."
+                    result = await wait_with_retries()
+                else:
+                    result = await asyncio.wait_for(
+                        wait_with_retries(), timeout.total_seconds()
+                    )
+            except asyncio.TimeoutError as error:
+                if last_error is None:
+                    if timeout is None:
+                        raise
+                    raise TimeoutError(
+                        _timeout_message(self.id, timeout, self.job_type)
+                    ) from error
+                identity = (
+                    f"job {self.id!r}" if self.id is not None else "the operation"
                 )
-            else:
-                reconnect = "Call wait() on this job handle to reconnect."
-            raise ConnectionError(
-                f"Lost connection while waiting and the timeout expired; "
-                f"{identity} may still be running. "
-                f"{reconnect}"
-            ) from last_error
+                if self.id is not None and self.job_type != "query":
+                    reconnect = (
+                        f"Use db.open_job({self.id!r}) to reconnect, then call wait() "
+                        "on the returned job."
+                    )
+                else:
+                    reconnect = "Call wait() on this job handle to reconnect."
+                raise ConnectionError(
+                    f"Lost connection while waiting and the timeout expired; "
+                    f"{identity} may still be running. "
+                    f"{reconnect}"
+                ) from last_error
+        finally:
+            if owns:
+                progress_obj.close()
         if self._result_decoder is not None:
             return self._result_decoder(result)
         return cast(T, result)
@@ -343,7 +526,11 @@ class Job(Generic[T]):
             return "finished"
         return LOOP.run(self._inner.status())
 
-    def wait(self, timeout: Optional[timedelta] = None) -> T:
+    def wait(
+        self,
+        timeout: Optional[timedelta] = None,
+        progress: Any = False,
+    ) -> T:
         """Block until the operation reaches a terminal state.
 
         Returns the endpoint's typed result, or ``None`` for a unit-result
@@ -364,7 +551,13 @@ class Job(Generic[T]):
         """
         if self._inner is None:
             return cast(T, None)
-        return LOOP.run(self._inner.wait(timeout))
+        return LOOP.run(self._inner.wait(timeout, progress=progress))
+
+    def progress(self) -> Optional[dict[str, Any]]:
+        """Fetch a structured progress snapshot. See :meth:`AsyncJob.progress`."""
+        if self._inner is None:
+            return None
+        return LOOP.run(self._inner.progress())
 
     def cancel(self):
         """Request cancellation. Cancelling a finished operation is a no-op."""

python/python/tests/test_job.py

@@ -2,13 +2,14 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
 import asyncio
+import time
 from datetime import timedelta
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
 from lancedb.exceptions import JobCancelledError, JobFailedError
-from lancedb.job import AsyncJob, Job
+from lancedb.job import AsyncJob, Job, _progress_eta
 from lancedb.remote.errors import HttpError, RetryError
 
 
@@ -238,6 +239,20 @@ class _PendingQuery:
         await asyncio.sleep(60)
 
 
+class _ProgressJob:
+    id = "job-progress"
+    job_type = "refresh_column"
+
+    async def status(self):
+        return "finished"
+
+    async def wait(self, progress=None):
+        if progress is not None:
+            progress({"state": "running", "rows_planned": 4, "rows_computed": 2})
+            progress({"state": "finished", "rows_planned": 4, "rows_computed": 4})
+        return None
+
+
 def _assert_timeout(error: TimeoutError) -> None:
     message = str(error)
     assert "job 'job-timeout' still running after 0.01 seconds" in message
@@ -264,6 +279,421 @@ def test_job_wait_timeout_explains_reattach():
     _assert_timeout(exc_info.value)
 
 
+@pytest.mark.asyncio
+async def test_progress_is_additive_to_string_status_and_returns_none_locally():
+    inner = MagicMock(id="job-progress", job_type="refresh_column")
+    inner.status = AsyncMock(return_value="running")
+    inner._description = type(
+        "Description", (), {"progress": {"state": "running", "rows_computed": 2}}
+    )()
+    job = AsyncJob(inner)
+
+    assert await job.status() == "running"
+    assert await job.progress() == {"state": "running", "rows_computed": 2}
+    assert await AsyncJob(None).progress() is None
+    assert await AsyncJob(_ProgressJob()).progress() is None
+    assert Job(AsyncJob(None)).progress() is None
+
+
+def test_progress_eta_uses_two_stable_advancing_same_epoch_samples():
+    now = int(time.time() * 1_000)
+    previous = {
+        "state": "running",
+        "execution_epoch": 7,
+        "sample_at_millis": now - 1_000,
+        "rows_planned": 100,
+        "rows_computed": 20,
+        "rows_skipped": 0,
+        "rows_remaining": 80,
+        "fragments_planned": 4,
+    }
+    current = {
+        **previous,
+        "sample_at_millis": now,
+        "rows_computed": 40,
+        "rows_remaining": 60,
+    }
+    assert _progress_eta(previous, current) == 3_000
+    assert _progress_eta(previous, {**current, "rows_planned": 120}) is None
+    assert _progress_eta(previous, {**current, "execution_epoch": 8}) is None
+    assert _progress_eta(previous, {**current, "rows_computed": 20}) is None
+    stale = {
+        **current,
+        "sample_at_millis": now - 11 * 60 * 1_000,
+    }
+    assert (
+        _progress_eta({**previous, "sample_at_millis": now - 12 * 60 * 1_000}, stale)
+        is None
+    )
+    assert _progress_eta(previous, {**current, "state": "finished"}) is None
+    assert (
+        _progress_eta(previous, {**current, "state": "finished", "rows_remaining": 0})
+        == 0
+    )
+
+
+@pytest.mark.asyncio
+async def test_progress_retains_eta_from_same_handle_observations():
+    now = int(time.time() * 1_000)
+    first = {
+        "state": "running",
+        "execution_epoch": 7,
+        "sample_at_millis": now - 1_000,
+        "rows_planned": 100,
+        "rows_computed": 20,
+        "rows_skipped": 0,
+        "rows_remaining": 80,
+        "fragments_planned": 4,
+    }
+    second = {
+        **first,
+        "sample_at_millis": now,
+        "rows_computed": 40,
+        "rows_remaining": 60,
+    }
+    inner = MagicMock(id="job-progress", job_type="refresh_column")
+    inner.status = AsyncMock(return_value="running")
+    description = type("Description", (), {"progress": first})()
+    inner._description = description
+    job = AsyncJob(inner)
+
+    assert await job.progress() == first
+    description.progress = second
+    assert (await job.progress())["estimated_remaining_time_millis"] == 3_000
+
+
+@pytest.mark.asyncio
+async def test_wait_progress_callback_observes_backend_poll_updates():
+    updates = []
+
+    result = await AsyncJob(_ProgressJob()).wait(progress=updates.append)
+
+    assert result is None
+    assert updates == [
+        {"state": "running", "rows_planned": 4, "rows_computed": 2},
+        {"state": "finished", "rows_planned": 4, "rows_computed": 4},
+    ]
+
+
+@pytest.mark.asyncio
+async def test_wait_progress_callback_uses_the_public_eta_enrichment(monkeypatch):
+    now = 10_000_000
+    monkeypatch.setattr("lancedb.job.time.time", lambda: now / 1_000)
+
+    class EtaJob:
+        id = "job-eta"
+        job_type = "refresh_column"
+
+        async def wait(self, progress):
+            first = {
+                "state": "running",
+                "execution_epoch": 7,
+                "sample_at_millis": now - 1_000,
+                "rows_planned": 100,
+                "rows_computed": 20,
+                "rows_skipped": 0,
+                "rows_remaining": 80,
+                "fragments_planned": 4,
+            }
+            progress(first)
+            progress(
+                {
+                    **first,
+                    "sample_at_millis": now,
+                    "rows_computed": 40,
+                    "rows_remaining": 60,
+                }
+            )
+
+    updates = []
+    assert await AsyncJob(EtaJob()).wait(progress=updates.append) is None
+
+    assert "estimated_remaining_time_millis" not in updates[0]
+    assert updates[1]["estimated_remaining_time_millis"] == 3_000
+
+
+@pytest.mark.asyncio
+async def test_wait_progress_true_updates_and_closes_only_the_owned_bar(monkeypatch):
+    tqdm_module = pytest.importorskip("tqdm.auto")
+
+    class Bar:
+        def __init__(self):
+            self.total = None
+            self.updates = []
+            self.phases = []
+            self.closed = False
+
+        def update(self, delta):
+            self.updates.append(delta)
+
+        def set_postfix_str(self, phase):
+            self.phases.append(phase)
+
+        def close(self):
+            self.closed = True
+
+    bar = Bar()
+    monkeypatch.setattr(tqdm_module, "tqdm", lambda **kwargs: bar)
+
+    assert await AsyncJob(_ProgressJob()).wait(progress=True) is None
+    assert bar.total == 4
+    assert bar.updates == [2, 2]
+    assert bar.phases == ["running", "finished"]
+    assert bar.closed
+
+
+@pytest.mark.asyncio
+async def test_wait_progress_bar_resumes_caller_owned_bar_after_timeout():
+    class ResumableJob:
+        id = "job-resume"
+        job_type = "refresh_column"
+
+        def __init__(self, bar):
+            self.calls = 0
+            self.bar = bar
+
+        async def wait(self, progress):
+            self.calls += 1
+            if self.calls == 1:
+                progress(
+                    {
+                        "state": "running",
+                        "execution_epoch": 1,
+                        "rows_planned": 4,
+                        "rows_remaining": 2,
+                    }
+                )
+                await asyncio.sleep(0.1)
+            else:
+                progress(
+                    {
+                        "state": "running",
+                        "execution_epoch": 2,
+                        "rows_planned": 4,
+                        "rows_remaining": 3,
+                    }
+                )
+                assert self.bar.n == 1
+            progress(
+                {
+                    "state": "finished",
+                    "execution_epoch": 2,
+                    "rows_planned": 4,
+                    "rows_remaining": 0,
+                }
+            )
+
+    class Bar:
+        def __init__(self):
+            self.total = None
+            self.n = 0
+            self.updates = []
+
+        def update(self, delta):
+            self.updates.append(delta)
+            self.n += delta
+
+        def set_postfix_str(self, _phase):
+            pass
+
+    bar = Bar()
+    job = AsyncJob(ResumableJob(bar))
+
+    with pytest.raises(TimeoutError):
+        await job.wait(timeout=timedelta(seconds=0.01), progress=bar)
+
+    assert await job.wait(progress=bar) is None
+    assert bar.total == 4
+    assert bar.n == 4
+    assert bar.updates == [2, 3]
+
+
+@pytest.mark.asyncio
+async def test_wait_progress_bar_resets_display_for_new_plan_or_epoch():
+    class ReplannedJob:
+        id = "job-replanned"
+        job_type = "refresh_column"
+
+        async def wait(self, progress):
+            progress(
+                {
+                    "state": "running",
+                    "execution_epoch": 1,
+                    "rows_planned": 6,
+                    "rows_remaining": 2,
+                }
+            )
+            progress(
+                {
+                    "state": "running",
+                    "execution_epoch": 1,
+                    "rows_planned": 4,
+                    "rows_remaining": 3,
+                }
+            )
+            progress(
+                {
+                    "state": "running",
+                    "execution_epoch": 2,
+                    "rows_planned": 4,
+                    "rows_remaining": 3,
+                }
+            )
+            progress(
+                {
+                    "state": "finished",
+                    "execution_epoch": 2,
+                    "rows_planned": 4,
+                    "rows_remaining": 0,
+                }
+            )
+
+    class Bar:
+        def __init__(self):
+            self.total = None
+            self.n = 0
+            self.updates = []
+            self.resets = []
+
+        def update(self, delta):
+            self.updates.append(delta)
+            self.n += delta
+
+        def reset(self, total=None):
+            self.resets.append(total)
+            self.total = total
+            self.n = 0
+
+        def set_postfix_str(self, _phase):
+            pass
+
+    bar = Bar()
+
+    assert await AsyncJob(ReplannedJob()).wait(progress=bar) is None
+    assert bar.total == 4
+    assert bar.n == 4
+    assert bar.n <= bar.total
+    assert bar.resets == [4, 4]
+    assert bar.updates == [4, 1, 1, 3]
+
+
+@pytest.mark.asyncio
+async def test_wait_progress_bar_uses_remaining_rows_when_computation_is_repeated(
+    monkeypatch,
+):
+    tqdm_module = pytest.importorskip("tqdm.auto")
+
+    class Bar:
+        def __init__(self):
+            self.total = None
+            self.updates = []
+            self.closed = False
+
+        def update(self, delta):
+            self.updates.append(delta)
+
+        def set_postfix_str(self, _phase):
+            pass
+
+        def close(self):
+            self.closed = True
+
+    class RepeatedComputationJob:
+        id = "job-recomputed"
+        job_type = "refresh_column"
+
+        async def wait(self, progress):
+            progress(
+                {
+                    "state": "running",
+                    "rows_planned": 4,
+                    "rows_computed": 6,
+                    "rows_remaining": 2,
+                }
+            )
+            progress(
+                {
+                    "state": "finished",
+                    "rows_planned": 4,
+                    "rows_computed": 8,
+                    "rows_remaining": 0,
+                }
+            )
+
+    bar = Bar()
+    monkeypatch.setattr(tqdm_module, "tqdm", lambda **kwargs: bar)
+
+    assert await AsyncJob(RepeatedComputationJob()).wait(progress=True) is None
+    assert bar.total == 4
+    assert bar.updates == [2, 2]
+    assert bar.closed
+
+
+@pytest.mark.asyncio
+async def test_wait_progress_bar_tracks_late_planning_without_regressing(
+    monkeypatch,
+):
+    tqdm_module = pytest.importorskip("tqdm.auto")
+
+    class Bar:
+        def __init__(self):
+            self.total = None
+            self.updates = []
+            self.closed = False
+
+        def update(self, delta):
+            self.updates.append(delta)
+
+        def set_postfix_str(self, _phase):
+            pass
+
+        def close(self):
+            self.closed = True
+
+    class RegressingPlanJob:
+        id = "job-regressing-plan"
+        job_type = "refresh_column"
+
+        async def wait(self, progress):
+            samples = [
+                (0, 0),
+                (4, 2),
+                (4, 3),
+                (4, 2),
+                (6, 1),
+                (6, 0),
+            ]
+            for planned, remaining in samples:
+                progress(
+                    {
+                        "state": "running",
+                        "rows_planned": planned,
+                        "rows_remaining": remaining,
+                    }
+                )
+
+    bar = Bar()
+    monkeypatch.setattr(tqdm_module, "tqdm", lambda **kwargs: bar)
+
+    assert await AsyncJob(RegressingPlanJob()).wait(progress=True) is None
+    assert bar.total == 6
+    # The remaining-row dip at (4, 3) must not rewind the high-water mark.
+    assert bar.updates == [2, 3, 1]
+    assert bar.closed
+
+
+@pytest.mark.asyncio
+async def test_wait_progress_false_keeps_legacy_inner_wait_call():
+    class NoProgressArgument:
+        id = "job-legacy"
+        job_type = None
+
+        async def wait(self):
+            return None
+
+    assert await AsyncJob(NoProgressArgument()).wait(progress=False) is None
+
+
 @pytest.mark.asyncio
 async def test_query_wait_timeout_requires_keeping_original_handle():
     job = AsyncJob(_PendingQuery())

python/python/tests/test_remote_db.py

@@ -1738,12 +1738,16 @@ def test_remote_create_index_async_returns_job():
             assert json.loads(body)["job_id"] == "job-1"
             describe_calls.append(1)
             state = "IN_PROGRESS" if len(describe_calls) == 1 else "DONE"
+            response = dict(job_id="job-1", job_state=state)
+            if len(describe_calls) == 1:
+                response.update(
+                    progress=dict(rows_planned=10, rows_computed=3),
+                    phase="building",
+                )
             request.send_response(200)
             request.send_header("Content-Type", "application/json")
             request.end_headers()
-            request.wfile.write(
-                json.dumps(dict(job_id="job-1", job_state=state)).encode()
-            )
+            request.wfile.write(json.dumps(response).encode())
         elif request.path == "/v1/jobs/cancel":
             assert json.loads(body)["job_id"] == "job-1"
             request.send_response(200)
@@ -1779,8 +1783,13 @@ def test_remote_create_index_async_returns_job():
         table = db.create_table("test", [{"id": 1}])
         job = table.create_index_async("id", config=BTree())
         assert job.id == "job-1"
-        assert job.wait(timeout=timedelta(seconds=30)) is None
+        observed = []
+        assert job.wait(timeout=timedelta(seconds=30), progress=observed.append) is None
         assert len(describe_calls) == 2
+        assert observed[0]["state"] == "running"
+        assert observed[0]["rows_planned"] == 10
+        assert observed[0]["rows_computed"] == 3
+        assert observed[-1]["state"] == "finished"
         job.cancel()
 
 
@@ -1886,6 +1895,91 @@ async def test_remote_refresh_async_returns_typed_terminal_result(
     assert result.version == 8
 
 
+def test_remote_refresh_async_forwards_progress_from_submitted_handle():
+    terminal_result = {
+        "rows_assigned": 12,
+        "rows_failed": 0,
+        "rows_remaining": 0,
+        "source_version": 7,
+        "published_version": 8,
+    }
+    describe_calls = []
+
+    def handler(request):
+        content_len = int(request.headers.get("Content-Length", 0))
+        body = request.rfile.read(content_len) if content_len > 0 else b""
+        if request.path == "/v1/table/test/backfill_column":
+            assert json.loads(body) == {"column": "derived"}
+            request.send_response(202)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b'{"job_id": "refresh-progress"}')
+        elif request.path == "/v1/jobs/describe":
+            assert json.loads(body)["job_id"] == "refresh-progress"
+            describe_calls.append(1)
+            if len(describe_calls) == 1:
+                response = {
+                    "job_id": "refresh-progress",
+                    "job_type": "function_refresh",
+                    "job_state": "IN_PROGRESS",
+                    "progress": {"rows_planned": 12, "rows_computed": 4},
+                    "phase": "running",
+                }
+            else:
+                response = {
+                    "job_id": "refresh-progress",
+                    "job_type": "function_refresh",
+                    "job_state": "DONE",
+                    "result": terminal_result,
+                }
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(json.dumps(response).encode())
+        elif request.path == "/v1/table/test/create/?mode=create":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b"{}")
+        elif request.path == "/v1/table/test/describe/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps(
+                    {
+                        "version": 1,
+                        "schema": {
+                            "fields": [
+                                {
+                                    "name": "id",
+                                    "type": {"type": "int64"},
+                                    "nullable": False,
+                                }
+                            ]
+                        },
+                    }
+                ).encode()
+            )
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        table = db.create_table("test", [{"id": 1}])
+        job = table.refresh_column_async("derived")
+        observed = []
+        result = job.wait(timeout=timedelta(seconds=30), progress=observed.append)
+
+    assert result.rows_filled == 12
+    assert result.version == 8
+    assert len(describe_calls) == 2
+    assert observed[0]["state"] == "running"
+    assert observed[0]["rows_planned"] == 12
+    assert observed[0]["rows_computed"] == 4
+    assert observed[-1]["state"] == "finished"
+
+
 @pytest.mark.parametrize(
     "refresh_kwargs,message",
     [
@@ -4961,6 +5055,9 @@ def test_remote_connection_jobs_surface():
                             job_type="create_index",
                             state="in_progress",
                             created_at_millis=1000,
+                            progress=dict(rows_planned=10, rows_computed=4),
+                            phase="running",
+                            paused=True,
                         )
                     ],
                     page_token="next",
@@ -5014,6 +5111,11 @@ def test_remote_connection_jobs_surface():
                         job_state="FAILED",
                         creation_ms=1000,
                         spec=dict(column="vec"),
+                        progress=dict(rows_planned=10, rows_computed=4),
+                        phase="initializing",
+                        queue_reason="waiting for capacity",
+                        waiting_for_job_id="job-maintenance",
+                        paused_at_millis=1200,
                         failure=dict(
                             phase="execute", message="worker died", retryable=True
                         ),
@@ -5064,6 +5166,16 @@ def test_remote_connection_jobs_surface():
         assert [job.job_id for job in jobs] == ["job-1", "job-2"]
         assert jobs[0].state == "running"
         assert jobs[0].table == "t1"
+        assert jobs[0].progress == {
+            "job_type": "create_index",
+            "state": "running",
+            "creation_ms": 1000,
+            "paused": True,
+            "phase": "running",
+            "rows_planned": 10,
+            "rows_computed": 4,
+        }
+        assert jobs[0].paused is True
         assert jobs[1].state == "finished"
 
         assert db.cancel_job("job-1") is True
@@ -5084,13 +5196,25 @@ def test_remote_connection_jobs_surface():
         # Opening already populated the handle.
         assert job.state == "failed"
         assert job.spec == {"column": "vec"}
+        progress = job.progress()
+        assert progress["state"] == "failed"
+        assert progress["rows_computed"] == 4
+        assert progress["paused"] is True
+        assert progress["phase"] == "initializing"
+        assert progress["queue_reason"] == "waiting for capacity"
+        assert progress["waiting_for_job_id"] == "job-maintenance"
+        assert progress["paused_at_millis"] == 1200
         assert job.failure.message == "worker died"
         assert repr(job.failure) == (
             'JobFailureInfo(phase="execute", message="worker died", retryable=True)'
         )
         assert job.status() == "failed"
+        observed = []
         with pytest.raises(JobFailedError, match="worker died"):
-            job.wait(timeout=timedelta(seconds=5))
+            job.wait(timeout=timedelta(seconds=5), progress=observed.append)
+        assert observed[0]["state"] == "failed"
+        assert observed[0]["rows_computed"] == 4
+        assert finished.progress()["paused"] is False
 
 
 def test_remote_job_handle_reports_its_own_detail():

python/src/job.rs

@@ -10,7 +10,7 @@ use arrow::{
 };
 use lancedb::job::JobEventsRequest;
 use pyo3::{
-    Bound, PyAny, PyRef, PyResult, Python,
+    Bound, Py, PyAny, PyRef, PyResult, Python,
     exceptions::PyValueError,
     pyclass, pymethods,
     types::{PyAnyMethods, PyDict, PyDictMethods},
@@ -94,10 +94,30 @@ impl Job {
         )
     }
 
-    pub fn wait(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+    #[pyo3(signature = (progress=None))]
+    pub fn wait(self_: PyRef<'_, Self>, progress: Option<Py<PyAny>>) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner.clone();
         future_into_py(self_.py(), async move {
-            let result = inner.wait().await.infer_error()?;
+            let result = if let Some(progress) = progress {
+                let observer = move |description: &lancedb::database::JobDescription| {
+                    Python::attach(|py| {
+                        let snapshot = description.progress_snapshot();
+                        let callback_result: PyResult<()> = (|| {
+                            let raw = snapshot.to_string();
+                            let parsed = parse_json_payload(py, Some(&raw))?
+                                .ok_or_else(|| PyValueError::new_err("missing progress"))?;
+                            progress.bind(py).call1((parsed,))?;
+                            Ok(())
+                        })();
+                        if let Err(error) = callback_result {
+                            log::warn!("job progress callback error: {error}");
+                        }
+                    });
+                };
+                inner.wait_with_progress(observer).await.infer_error()?
+            } else {
+                inner.wait().await.infer_error()?
+            };
             result
                 .map_err(|message| lancedb::Error::Runtime { message })
                 .infer_error()
@@ -164,6 +184,8 @@ pub struct JobInfo {
     job_type: String,
     state: String,
     created_at_millis: i64,
+    _progress_json: Option<String>,
+    paused: bool,
 }
 
 #[pymethods]
@@ -174,16 +196,32 @@ impl JobInfo {
             self.job_id, self.table, self.job_type, self.state, self.created_at_millis
         )
     }
+
+    /// Structured progress for job types that report it. Missing fields are
+    /// unknown, not zero.
+    #[getter]
+    fn progress<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
+        parse_json_payload(py, self._progress_json.as_deref())
+    }
+
+    /// Whether an operator has parked this job.
+    #[getter]
+    fn paused(&self) -> bool {
+        self.paused
+    }
 }
 
 impl From<lancedb::database::JobInfo> for JobInfo {
     fn from(info: lancedb::database::JobInfo) -> Self {
+        let progress = info.progress_snapshot().to_string();
         Self {
             job_id: info.job_id,
             table: info.table,
             job_type: info.job_type,
             state: info.state,
             created_at_millis: info.created_at_millis,
+            _progress_json: Some(progress),
+            paused: info.paused,
         }
     }
 }
@@ -229,6 +267,7 @@ pub struct JobDescription {
     _spec_json: Option<String>,
     /// Internal: the wire form behind the `result` property.
     _result_json: Option<String>,
+    _progress_json: Option<String>,
     failure: Option<JobFailureInfo>,
 }
 
@@ -246,6 +285,13 @@ impl JobDescription {
         parse_json_payload(py, self._result_json.as_deref())
     }
 
+    /// Structured progress snapshot. Lifecycle and queue fields are present
+    /// even when a job type or older server supplies no counters.
+    #[getter]
+    fn progress<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
+        parse_json_payload(py, self._progress_json.as_deref())
+    }
+
     fn __repr__(&self, py: Python<'_>) -> PyResult<String> {
         let mut fields = vec![
             format!("job_id={:?}", self.job_id),
@@ -273,6 +319,7 @@ impl JobDescription {
 
 impl From<lancedb::database::JobDescription> for JobDescription {
     fn from(description: lancedb::database::JobDescription) -> Self {
+        let progress_json = description.progress_snapshot().to_string();
         Self {
             job_id: description.job_id,
             job_type: description.job_type,
@@ -283,6 +330,7 @@ impl From<lancedb::database::JobDescription> for JobDescription {
                 .result
                 .filter(|result| !result.is_null())
                 .map(|result| result.to_string()),
+            _progress_json: Some(progress_json),
             failure: description.failure.map(|failure| JobFailureInfo {
                 phase: failure.phase,
                 message: failure.message,