diff --git a/nodejs/src/job.rs b/nodejs/src/job.rs
index 6aaeee174..14013fd27 100644
--- a/nodejs/src/job.rs
+++ b/nodejs/src/job.rs
@@ -14,9 +14,12 @@ pub struct Job {
 }
 
 impl Job {
-    pub(crate) fn new(inner: lancedb::Job) -> Self {
+    pub(crate) fn new<T>(inner: lancedb::Job<T>) -> Self
+    where
+        T: Clone + Send + Sync + 'static,
+    {
         Self {
-            inner: Arc::new(inner),
+            inner: Arc::new(inner.map(|_| ())),
         }
     }
 }
diff --git a/python/python/lancedb/__init__.py b/python/python/lancedb/__init__.py
index aa473c5f8..df8950686 100644
--- a/python/python/lancedb/__init__.py
+++ b/python/python/lancedb/__init__.py
@@ -29,6 +29,7 @@ from .functions import (
     FunctionRegistrationRequest as FunctionRegistrationRequest,
     FunctionVersion as FunctionVersion,
     PythonRuntimeSpec as PythonRuntimeSpec,
+    RefreshColumnResult as RefreshColumnResult,
     UdfDefinition as UdfDefinition,
     udf as udf,
 )
diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi
index 648469579..1e314ede8 100644
--- a/python/python/lancedb/_lancedb.pyi
+++ b/python/python/lancedb/_lancedb.pyi
@@ -147,7 +147,7 @@ class Connection(object):
         limit: Optional[int],
     ) -> list[str]: ...  # Deprecated: Use list_tables instead
     def job(self, job_id: str) -> Job: ...
-    async def create_function_async(self, request_json: str) -> FunctionJob: ...
+    async def create_function_async(self, request_json: str) -> Job: ...
     async def get_function(self, name: str, version: str) -> str: ...
     async def list_jobs(self) -> List[JobInfo]: ...
     async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
@@ -234,14 +234,7 @@ class Job:
     @property
     def id(self) -> Optional[str]: ...
     async def status(self) -> str: ...
-    async def wait(self) -> None: ...
-    async def cancel(self) -> None: ...
-
-class FunctionJob:
-    @property
-    def id(self) -> Optional[str]: ...
-    async def status(self) -> str: ...
-    async def wait(self) -> str: ...
+    async def wait(self) -> Optional[str]: ...
     async def cancel(self) -> None: ...
 
 class JobInfo:
diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py
index 7ad749920..51b8d9993 100644
--- a/python/python/lancedb/db.py
+++ b/python/python/lancedb/db.py
@@ -46,7 +46,7 @@ from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
 from . import __version__
 from ._lancedb import connect as lancedb_connect  # type: ignore
 from .functions import FunctionVersion, UdfDefinition
-from .job import AsyncJob, Job, _function_job
+from .job import AsyncJob, Job, _typed_job
 from .materialized_view import (
     AsyncMaterializedView,
     MaterializedView,
@@ -2237,7 +2237,7 @@ class AsyncConnection(object):
         inner = await self._inner.create_function_async(
             definition.registration_request.to_canonical_json()
         )
-        return _function_job(inner)
+        return _typed_job(inner, FunctionVersion.from_json)
 
     async def get_function(self, name: str, *, version: str) -> FunctionVersion:
         """Open one exact immutable Function version from the remote catalog."""
diff --git a/python/python/lancedb/functions.py b/python/python/lancedb/functions.py
index 9532ab8b9..c4ff9a9b3 100644
--- a/python/python/lancedb/functions.py
+++ b/python/python/lancedb/functions.py
@@ -1,10 +1,12 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
-"""Canonical values exchanged with LanceDB Enterprise Function services.
+"""Canonical Function values exchanged with LanceDB Enterprise services.
 
 These immutable models contain client/wire state only. Catalog persistence,
 environment bake, secret resolution, and execution are owned by Sophon.
+``RefreshColumnResult`` is also the backend-neutral result of a local
+expression-backed refresh job.
 """
 
 from __future__ import annotations
@@ -460,7 +462,11 @@ class FunctionBinding(_RemoteValue):
 
 
 class RefreshColumnResult(_RemoteValue):
-    """Terminal result of a remote Function-column refresh Job."""
+    """Terminal result of an expression-backed or Function-backed refresh Job.
+
+    Local jobs produce this value in process. LanceDB Cloud and Enterprise
+    decode the same value from the durable server-job terminal payload.
+    """
 
     rows_assigned: _UInt64
     rows_failed: _UInt64
diff --git a/python/python/lancedb/job.py b/python/python/lancedb/job.py
index 7bd600a74..f688768cb 100644
--- a/python/python/lancedb/job.py
+++ b/python/python/lancedb/job.py
@@ -5,12 +5,11 @@
 
 import asyncio
 from datetime import timedelta
-from typing import Any, Generic, Optional, TypeVar, cast
+from typing import Any, Callable, Generic, Optional, TypeVar, cast
 
 from lancedb.background_loop import LOOP
 
 from . import _lancedb
-from .functions import FunctionVersion
 
 T = TypeVar("T")
 
@@ -18,11 +17,18 @@ T = TypeVar("T")
 class AsyncJob(Generic[T]):
     """A handle to an operation that may still be running.
 
-    The operation may already be complete when the handle is created.
+    The operation may already be complete when the handle is created. ``T``
+    is the endpoint's terminal result type; unit-result jobs resolve to
+    ``None``.
     """
 
-    def __init__(self, inner: Optional[Any]):
+    def __init__(
+        self,
+        inner: Optional[Any],
+        result_decoder: Optional[Callable[[Any], T]] = None,
+    ):
         self._inner = inner
+        self._result_decoder = result_decoder
 
     @property
     def id(self) -> Optional[str]:
@@ -50,17 +56,21 @@ class AsyncJob(Generic[T]):
     async def wait(self, timeout: Optional[timedelta] = None) -> T:
         """Wait until the operation reaches a terminal state.
 
+        Returns the endpoint's typed result, or ``None`` for a unit-result
+        job.
+
         Raises `JobFailedError` if the operation failed, `JobCancelledError`
         if it was cancelled, and `TimeoutError` if `timeout` elapses first.
         """
         if self._inner is None:
             return cast(T, None)
         if timeout is None:
-            return cast(T, await self._inner.wait())
-        return cast(
-            T,
-            await asyncio.wait_for(self._inner.wait(), timeout.total_seconds()),
-        )
+            result = await self._inner.wait()
+        else:
+            result = await asyncio.wait_for(self._inner.wait(), timeout.total_seconds())
+        if self._result_decoder is not None:
+            return self._result_decoder(result)
+        return cast(T, result)
 
     async def cancel(self):
         """Request cancellation. Cancelling a finished operation is a no-op."""
@@ -70,7 +80,7 @@ class AsyncJob(Generic[T]):
 
 
 class Job(Generic[T]):
-    """Synchronous counterpart of `AsyncJob`."""
+    """Synchronous counterpart of `AsyncJob` with the same result type."""
 
     def __init__(self, inner: Optional[AsyncJob[T]]):
         self._inner = inner
@@ -96,6 +106,9 @@ class Job(Generic[T]):
     def wait(self, timeout: Optional[timedelta] = None) -> T:
         """Block until the operation reaches a terminal state.
 
+        Returns the endpoint's typed result, or ``None`` for a unit-result
+        job.
+
         Raises `JobFailedError` if the operation failed, `JobCancelledError`
         if it was cancelled, and `TimeoutError` if `timeout` elapses first.
         """
@@ -110,23 +123,8 @@ class Job(Generic[T]):
         LOOP.run(self._inner.cancel())
 
 
-class _FunctionJobAdapter:
-    def __init__(self, inner: "_lancedb.FunctionJob"):
-        self._inner = inner
-
-    @property
-    def id(self) -> Optional[str]:
-        return self._inner.id
-
-    async def status(self) -> str:
-        return await self._inner.status()
-
-    async def wait(self) -> FunctionVersion:
-        return FunctionVersion.from_json(await self._inner.wait())
-
-    async def cancel(self):
-        await self._inner.cancel()
-
-
-def _function_job(inner: "_lancedb.FunctionJob") -> AsyncJob[FunctionVersion]:
-    return AsyncJob(_FunctionJobAdapter(inner))
+def _typed_job(
+    inner: "_lancedb.Job", result_decoder: Callable[[str], T]
+) -> AsyncJob[T]:
+    """Bind an internal JSON-producing job to its public result model."""
+    return AsyncJob(inner, result_decoder)
diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py
index b97f8f194..41394ed71 100644
--- a/python/python/lancedb/remote/table.py
+++ b/python/python/lancedb/remote/table.py
@@ -49,7 +49,7 @@ from lancedb.index import (
     LabelList,
 )
 from lancedb.job import Job
-from lancedb.functions import FunctionApplication
+from lancedb.functions import FunctionApplication, RefreshColumnResult
 from lancedb.remote.db import LOOP
 from lancedb.table import IndexConfigType, KNOWN_METRICS
 import pyarrow as pa
@@ -972,7 +972,7 @@ class RemoteTable(Table):
     def refresh_column(self, column: str):
         return LOOP.run(self._table.refresh_column(column))
 
-    def refresh_column_async(self, column: str) -> Job:
+    def refresh_column_async(self, column: str) -> Job[RefreshColumnResult]:
         return Job(LOOP.run(self._table.refresh_column_async(column)))
 
     def alter_columns(
diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py
index 0c84b4036..75b7b1db8 100644
--- a/python/python/lancedb/table.py
+++ b/python/python/lancedb/table.py
@@ -40,7 +40,7 @@ from ._blob import (
 from .types import BlobMode
 from lancedb.arrow import peek_reader
 from lancedb.background_loop import LOOP, embedding_executor
-from lancedb.job import AsyncJob, Job
+from lancedb.job import AsyncJob, Job, _typed_job
 from .dependencies import (
     _check_for_hugging_face,
     _check_for_lance,
@@ -72,7 +72,10 @@ from .index import (
     FTS,
 )
 from .expr import Expr
-from .functions import FunctionApplication
+from .functions import (
+    FunctionApplication,
+    RefreshColumnResult as RefreshColumnJobResult,
+)
 from .merge import LanceMergeInsertBuilder
 from .pydantic import LanceModel, model_to_dict
 from .query import (
@@ -2039,7 +2042,7 @@ class Table(ABC):
         """
 
     @abstractmethod
-    def refresh_column_async(self, column: str) -> Job:
+    def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
         """
         Like :meth:`refresh_column`, but returns a handle to the refresh job
         instead of blocking until it completes.
@@ -2050,6 +2053,12 @@ class Table(ABC):
         than failing the job. On local tables the job runs in-process; on
         LanceDB Cloud and Enterprise it is the server's backfill job.
 
+        Returns
+        -------
+        Job[RefreshColumnResult]
+            A job whose successful ``wait`` returns row counts plus the source
+            and published table versions.
+
         Examples
         --------
         >>> import lancedb
@@ -2058,7 +2067,9 @@ class Table(ABC):
         >>> table.add_columns(computed={"doubled": "x * 2"})
         AddColumnsResult(version=2)
         >>> job = table.refresh_column_async("doubled")
-        >>> job.wait()
+        >>> result = job.wait()
+        >>> result.rows_assigned
+        2
         >>> job.status()
         'finished'
         """
@@ -4082,7 +4093,7 @@ class LanceTable(Table):
         [`AsyncTable.refresh_column`][lancedb.AsyncTable.refresh_column]."""
         return LOOP.run(self._table.refresh_column(column))
 
-    def refresh_column_async(self, column: str) -> Job:
+    def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
         """Fill a computed column's unfilled rows, returning a handle to the
         refresh job. See
         [`Table.refresh_column_async`][lancedb.table.Table.refresh_column_async].
@@ -6122,7 +6133,9 @@ class AsyncTable:
         """
         return await self._inner.refresh_column(column)
 
-    async def refresh_column_async(self, column: str) -> AsyncJob:
+    async def refresh_column_async(
+        self, column: str
+    ) -> AsyncJob[RefreshColumnJobResult]:
         """
         Like :meth:`refresh_column`, but returns a handle to the refresh job
         instead of blocking until it completes.
@@ -6134,6 +6147,12 @@ class AsyncTable:
         in-process; on LanceDB Cloud and Enterprise it is the server's
         backfill job.
 
+        Returns
+        -------
+        AsyncJob[RefreshColumnResult]
+            A job whose successful ``wait`` returns row counts plus the source
+            and published table versions.
+
         Examples
         --------
         >>> import asyncio
@@ -6143,12 +6162,16 @@ class AsyncTable:
         ...     table = await db.create_table("computed_job_async_demo", [{"x": 1}])
         ...     await table.add_columns(computed={"doubled": "x * 2"})
         ...     job = await table.refresh_column_async("doubled")
-        ...     await job.wait()
+        ...     result = await job.wait()
+        ...     assert result.rows_assigned == 1
         ...     return await job.status()
         >>> asyncio.run(refresh_in_background())
         'finished'
         """
-        return AsyncJob(await self._inner.refresh_column_async(column))
+        return _typed_job(
+            await self._inner.refresh_column_async(column),
+            RefreshColumnJobResult.from_json,
+        )
 
     async def alter_columns(
         self, *alterations: Iterable[dict[str, Any]]
diff --git a/python/python/tests/test_db.py b/python/python/tests/test_db.py
index 38bbb53fb..aeb9feeb8 100644
--- a/python/python/tests/test_db.py
+++ b/python/python/tests/test_db.py
@@ -774,7 +774,7 @@ def test_drop_table_async(tmp_db: lancedb.DBConnection):
     job = tmp_db.drop_table_async("test")
     assert job.id is None
     assert job.status() == "finished"
-    job.wait()
+    assert job.wait() is None
     assert tmp_db.table_names() == []
 
     tmp_db.create_table("test", data=data)
@@ -790,7 +790,7 @@ async def test_drop_table_async_connection(tmp_db_async: lancedb.AsyncConnection
     job = await tmp_db_async.drop_table_async("test")
     assert job.id is None
     assert await job.status() == "finished"
-    await job.wait()
+    assert await job.wait() is None
     assert await tmp_db_async.table_names() == []
 
 
diff --git a/python/python/tests/test_first_class_function_slice2.py b/python/python/tests/test_first_class_function_slice2.py
index 99c68876c..a20347771 100644
--- a/python/python/tests/test_first_class_function_slice2.py
+++ b/python/python/tests/test_first_class_function_slice2.py
@@ -187,7 +187,7 @@ def _mock_remote_function_catalog():
                     "job_state": "DONE",
                     "result": state["version"],
                 }
-            elif self.path == "/v1/functions/get":
+            elif self.path == "/v1/functions/describe":
                 assert body == {
                     "name": "normalize_score",
                     "version": "fv_exact",
diff --git a/python/python/tests/test_index.py b/python/python/tests/test_index.py
index 94268a53e..fe6ebe87a 100644
--- a/python/python/tests/test_index.py
+++ b/python/python/tests/test_index.py
@@ -88,7 +88,7 @@ async def binary_table(db_async):
 async def test_create_index_async_returns_done_job(some_table: AsyncTable):
     job = await some_table.create_index_async("id", config=BTree())
     assert job.id is None
-    await job.wait()
+    assert await job.wait() is None
     assert len(await some_table.list_indices()) == 1
     await job.cancel()
 
diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py
index 08f5550eb..2952f00a4 100644
--- a/python/python/tests/test_remote_db.py
+++ b/python/python/tests/test_remote_db.py
@@ -875,11 +875,85 @@ 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"
-        job.wait(timeout=timedelta(seconds=30))
+        assert job.wait(timeout=timedelta(seconds=30)) is None
         assert len(describe_calls) == 2
         job.cancel()
 
 
+def test_remote_refresh_async_returns_typed_terminal_result():
+    terminal_result = {
+        "rows_assigned": 12,
+        "rows_failed": 0,
+        "rows_remaining": 0,
+        "source_version": 7,
+        "published_version": 8,
+    }
+
+    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-1"}')
+        elif request.path == "/v1/jobs/describe":
+            assert json.loads(body)["job_id"] == "refresh-1"
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps(
+                    {
+                        "job_id": "refresh-1",
+                        "job_type": "function_refresh",
+                        "job_state": "DONE",
+                        "result": terminal_result,
+                    }
+                ).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")
+        assert job.id == "refresh-1"
+        result = job.wait(timeout=timedelta(seconds=30))
+
+    assert isinstance(result, lancedb.RefreshColumnResult)
+    assert result.model_dump() == terminal_result
+    assert result.rows_filled == 12
+    assert result.version == 8
+
+
 def test_remote_job_wait_raises_on_failure():
     from lancedb.exceptions import JobFailedError
     from lancedb.index import BTree
diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py
index 0f98219bf..56e0eacfd 100644
--- a/python/python/tests/test_table.py
+++ b/python/python/tests/test_table.py
@@ -1467,7 +1467,7 @@ def test_create_index_async_returns_done_job(mem_db: DBConnection):
     table = mem_db.create_table("job_test", [{"id": i} for i in range(10)])
     job = table.create_index_async("id", config=BTree())
     assert job.id is None
-    job.wait()
+    assert job.wait() is None
     assert len(table.list_indices()) == 1
     job.cancel()
 
@@ -3947,10 +3947,21 @@ def test_refresh_column_async_returns_job(tmp_path):
 
     job = table.refresh_column_async("doubled")
     assert job.id is None  # in-process jobs have no server id
-    assert job.wait() is None
+    result = job.wait()
+    assert isinstance(result, lancedb.RefreshColumnResult)
+    assert result.rows_assigned == 2
+    assert result.rows_failed == 0
+    assert result.rows_remaining == 0
+    assert result.source_version == 2
+    assert result.published_version == 3
     assert job.status() == "finished"
     assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]
 
+    no_op = table.refresh_column_async("doubled").wait()
+    assert no_op.rows_assigned == 0
+    assert no_op.source_version == 3
+    assert no_op.published_version is None
+
     # Bad input raises at the call, not through the job.
     with pytest.raises(Exception, match="not a computed column"):
         table.refresh_column_async("x")
@@ -3963,6 +3974,10 @@ async def test_refresh_column_async_job_async_table(tmp_path):
     await table.add_columns(computed={"tripled": "x * 3"})
 
     job = await table.refresh_column_async("tripled")
-    assert await job.wait() is None
+    result = await job.wait()
+    assert isinstance(result, lancedb.RefreshColumnResult)
+    assert result.rows_assigned == 1
+    assert result.source_version == 2
+    assert result.published_version == 3
     assert await job.status() == "finished"
     assert (await table.to_arrow())["tripled"].to_pylist() == [9]
diff --git a/python/src/connection.rs b/python/src/connection.rs
index 4143ac9c9..902489f4f 100644
--- a/python/src/connection.rs
+++ b/python/src/connection.rs
@@ -609,7 +609,7 @@ impl Connection {
                 .create_function_async(request)
                 .await
                 .infer_error()
-                .map(crate::job::FunctionJob::new)
+                .map(crate::job::Job::new_typed)
         })
     }
 
diff --git a/python/src/job.rs b/python/src/job.rs
index a08b958a6..688cba7f9 100644
--- a/python/src/job.rs
+++ b/python/src/job.rs
@@ -5,72 +5,33 @@ use std::sync::Arc;
 
 use crate::runtime::future_into_py;
 use pyo3::{Bound, PyAny, PyRef, PyResult, pyclass, pymethods};
+use serde::Serialize;
 
 use crate::error::PythonErrorExt;
 
 #[pyclass]
 pub struct Job {
-    inner: Arc<lancedb::Job>,
-}
-
-/// Python bridge for a typed remote Function registration job.
-///
-/// The public Python layer decodes the canonical JSON returned by `wait`
-/// into its immutable `FunctionVersion` model.
-#[pyclass]
-pub struct FunctionJob {
-    inner: Arc<lancedb::Job<lancedb::function::FunctionVersion>>,
-}
-
-impl FunctionJob {
-    pub(crate) fn new(inner: lancedb::Job<lancedb::function::FunctionVersion>) -> Self {
-        Self {
-            inner: Arc::new(inner),
-        }
-    }
+    inner: Arc<lancedb::Job<std::result::Result<Option<String>, String>>>,
 }
 
 impl Job {
     pub(crate) fn new(inner: lancedb::Job) -> Self {
         Self {
-            inner: Arc::new(inner),
+            inner: Arc::new(inner.map(|()| Ok(None))),
         }
     }
-}
 
-#[pymethods]
-impl FunctionJob {
-    #[getter]
-    pub fn id(&self) -> Option<String> {
-        self.inner.id().map(str::to_string)
-    }
-
-    pub fn status(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
-        let inner = self_.inner.clone();
-        future_into_py(
-            self_.py(),
-            async move { inner.status().await.infer_error() },
-        )
-    }
-
-    pub fn wait(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
-        let inner = self_.inner.clone();
-        future_into_py(self_.py(), async move {
-            inner
-                .wait()
-                .await
-                .infer_error()?
-                .to_canonical_json()
-                .infer_error()
-        })
-    }
-
-    pub fn cancel(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
-        let inner = self_.inner.clone();
-        future_into_py(self_.py(), async move {
-            inner.cancel().await.infer_error()?;
-            Ok(())
-        })
+    pub(crate) fn new_typed<T>(inner: lancedb::Job<T>) -> Self
+    where
+        T: Clone + Serialize + Send + Sync + 'static,
+    {
+        Self {
+            inner: Arc::new(inner.map(|result| {
+                serde_json::to_string(&result)
+                    .map(Some)
+                    .map_err(|error| format!("failed to serialize typed job result: {error}"))
+            })),
+        }
     }
 }
 
@@ -92,8 +53,10 @@ impl Job {
     pub fn wait(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner.clone();
         future_into_py(self_.py(), async move {
-            inner.wait().await.infer_error()?;
-            Ok(None::<()>)
+            let result = inner.wait().await.infer_error()?;
+            result
+                .map_err(|message| lancedb::Error::Runtime { message })
+                .infer_error()
         })
     }
 
diff --git a/python/src/lib.rs b/python/src/lib.rs
index c1dfbc02b..8d3eab787 100644
--- a/python/src/lib.rs
+++ b/python/src/lib.rs
@@ -47,7 +47,6 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<Session>()?;
     m.add_class::<Table>()?;
     m.add_class::<crate::job::Job>()?;
-    m.add_class::<crate::job::FunctionJob>()?;
     m.add_class::<crate::job::JobInfo>()?;
     m.add_class::<crate::job::JobDescription>()?;
     m.add_class::<crate::job::JobFailureInfo>()?;
diff --git a/python/src/table.rs b/python/src/table.rs
index 5cdcc3653..b225b191f 100644
--- a/python/src/table.rs
+++ b/python/src/table.rs
@@ -1619,7 +1619,7 @@ impl Table {
         let inner = self_.inner_ref()?.clone();
         future_into_py(self_.py(), async move {
             let job = inner.refresh_column_async(column).await.infer_error()?;
-            Ok(crate::job::Job::new(job))
+            Ok(crate::job::Job::new_typed(job))
         })
     }
 
diff --git a/rust/lancedb/src/function.rs b/rust/lancedb/src/function.rs
index 835fca9a2..f02158871 100644
--- a/rust/lancedb/src/function.rs
+++ b/rust/lancedb/src/function.rs
@@ -1,7 +1,8 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
-//! Canonical values exchanged with the Enterprise Function service.
+//! Canonical Function values exchanged with the Enterprise service, plus the
+//! backend-neutral terminal result of a computed-column refresh.
 //!
 //! This module contains client/wire values only. Catalog persistence,
 //! environment bake, secret resolution, and execution are owned by Sophon.
@@ -580,13 +581,22 @@ impl FunctionBinding {
 
 impl_json!(FunctionBinding);
 
-/// Stable terminal result of a remote Function-column refresh Job.
+/// Stable terminal result of an expression-backed or Function-backed column
+/// refresh [`crate::Job`].
+///
+/// Local refresh jobs produce this value in process. LanceDB Cloud and
+/// Enterprise decode the same value from the durable job's terminal payload.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RefreshColumnResult {
+    /// Rows assigned a value by this refresh.
     pub rows_assigned: u64,
+    /// Rows whose computation failed.
     pub rows_failed: u64,
+    /// Rows that still need a value when the job completes.
     pub rows_remaining: u64,
+    /// Exact table version the refresh read.
     pub source_version: u64,
+    /// Table version made visible by the refresh, when one was published.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub published_version: Option<u64>,
 }
diff --git a/rust/lancedb/src/job.rs b/rust/lancedb/src/job.rs
index 1a76c7683..94d1ba2b6 100644
--- a/rust/lancedb/src/job.rs
+++ b/rust/lancedb/src/job.rs
@@ -6,7 +6,7 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use serde::de::DeserializeOwned;
+use serde::{Serialize, de::DeserializeOwned};
 use serde_json::Value;
 use tokio::sync::watch;
 use tokio::task::{AbortHandle, JoinHandle};
@@ -26,20 +26,16 @@ pub(crate) trait JobHandle: Send + Sync {
 }
 
 /// A backend-neutral successful terminal result.
-///
-/// Local operations do not carry a value. Remote operations may carry JSON
-/// that the public [`Job`] decodes according to its result type.
+#[derive(Clone)]
 pub(crate) struct TerminalResult {
-    #[allow(dead_code)] // Typed remote submit endpoints consume this after Slice 1.
     value: Option<Value>,
-    #[allow(dead_code)] // Preserved so typed decode errors retain request correlation.
     request_id: Option<String>,
 }
 
 impl TerminalResult {
-    pub(crate) fn local() -> Self {
+    fn local(value: Value) -> Self {
         Self {
-            value: None,
+            value: Some(value),
             request_id: None,
         }
     }
@@ -51,23 +47,31 @@ impl TerminalResult {
         }
     }
 
-    #[allow(dead_code)] // Exercised by the remote typed-result fixtures in Slice 1.
     fn decode<T: DeserializeOwned>(self) -> Result<T> {
-        let request_id = self.request_id.unwrap_or_default();
-        let value = self.value.ok_or_else(|| Error::Http {
-            source: "successful typed job response did not contain a result".into(),
-            request_id: request_id.clone(),
-            status_code: None,
+        let value = self.value.ok_or_else(|| match &self.request_id {
+            Some(request_id) => Error::Http {
+                source: "successful typed job response did not contain a result".into(),
+                request_id: request_id.clone(),
+                status_code: None,
+            },
+            None => Error::Runtime {
+                message: "successful typed job did not contain a result".to_string(),
+            },
         })?;
-        serde_json::from_value(value).map_err(|error| Error::Http {
-            source: format!("failed to parse typed job result: {error}").into(),
-            request_id,
-            status_code: None,
+        serde_json::from_value(value).map_err(|error| match self.request_id {
+            Some(request_id) => Error::Http {
+                source: format!("failed to parse typed job result: {error}").into(),
+                request_id,
+                status_code: None,
+            },
+            None => Error::Runtime {
+                message: format!("failed to parse typed job result: {error}"),
+            },
         })
     }
 }
 
-type ResultDecoder<T> = fn(TerminalResult) -> Result<T>;
+type ResultDecoder<T> = Arc<dyn Fn(TerminalResult) -> Result<T> + Send + Sync>;
 
 enum JobInner<T> {
     Handle {
@@ -79,7 +83,9 @@ enum JobInner<T> {
 
 /// A handle to an operation that may still be running.
 ///
-/// The operation may already be complete when the handle is created.
+/// The operation may already be complete when the handle is created. `T` is
+/// the endpoint's successful terminal result; unit-result operations use the
+/// default `Job<()>`.
 pub struct Job<T = ()>
 where
     T: Clone + Send + Sync + 'static,
@@ -111,15 +117,10 @@ impl Job<()> {
         Self {
             inner: JobInner::Handle {
                 handle,
-                decode: |_| Ok(()),
+                decode: Arc::new(|_| Ok(())),
             },
         }
     }
-
-    /// A unit-result job running as a task in this process.
-    pub(crate) fn spawned(task: JoinHandle<Result<()>>) -> Self {
-        Self::new(Box::new(SpawnedJob::new(task)))
-    }
 }
 
 impl<T> Job<T>
@@ -131,12 +132,22 @@ where
         Self {
             inner: JobInner::Handle {
                 handle,
-                decode: TerminalResult::decode::<T>,
+                decode: Arc::new(TerminalResult::decode::<T>),
             },
         }
     }
 }
 
+impl<T> Job<T>
+where
+    T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static,
+{
+    /// A typed job running as a task in this process.
+    pub(crate) fn spawned(task: JoinHandle<Result<T>>) -> Self {
+        Self::new_typed(Box::new(SpawnedJob::new(task)))
+    }
+}
+
 impl<T> Job<T>
 where
     T: Clone + Send + Sync + 'static,
@@ -169,11 +180,13 @@ where
 
     /// Waits until the operation reaches a terminal state.
     ///
+    /// Returns the endpoint's typed result. Unit-result jobs return `()`.
+    ///
     /// Returns [`crate::Error::JobFailed`] if the operation failed and
     /// [`crate::Error::JobCancelled`] if it was cancelled.
     pub async fn wait(&self) -> Result<T> {
         match &self.inner {
-            JobInner::Handle { handle, decode } => decode(handle.wait().await?),
+            JobInner::Handle { handle, decode } => (decode)(handle.wait().await?),
             JobInner::Completed(result) => Ok(result.clone()),
         }
     }
@@ -187,21 +200,53 @@ where
             JobInner::Completed(_) => Ok(()),
         }
     }
+
+    /// Maps a successful terminal result without changing the job lifecycle.
+    /// The mapping may run once for each call to [`Job::wait`], so it should
+    /// be deterministic and free of externally visible side effects.
+    ///
+    /// ```
+    /// use lancedb::{Job, function::RefreshColumnResult};
+    ///
+    /// # async fn rows_assigned(
+    /// #     job: Job<RefreshColumnResult>,
+    /// # ) -> lancedb::Result<u64> {
+    /// let job = job.map(|result| result.rows_assigned);
+    /// job.wait().await
+    /// # }
+    /// ```
+    pub fn map<U, F>(self, map: F) -> Job<U>
+    where
+        U: Clone + Send + Sync + 'static,
+        F: Fn(T) -> U + Send + Sync + 'static,
+    {
+        match self.inner {
+            JobInner::Handle { handle, decode } => Job {
+                inner: JobInner::Handle {
+                    handle,
+                    decode: Arc::new(move |result| Ok(map((decode)(result)?))),
+                },
+            },
+            JobInner::Completed(result) => Job {
+                inner: JobInner::Completed(map(result)),
+            },
+        }
+    }
 }
 
 /// How an in-process operation ended. Cloneable so every waiter can be given
 /// the outcome; [`Error`] is not, so failures share one behind an [`Arc`].
 #[derive(Clone)]
 enum Outcome {
-    Succeeded,
+    Succeeded(TerminalResult),
     Failed(Arc<Error>),
     Cancelled,
 }
 
 impl Outcome {
-    fn into_result(self) -> Result<()> {
+    fn into_result(self) -> Result<TerminalResult> {
         match self {
-            Self::Succeeded => Ok(()),
+            Self::Succeeded(result) => Ok(result),
             Self::Failed(source) => Err(Error::JobFailed {
                 job_id: None,
                 failure: JobFailure::from_source(source),
@@ -220,12 +265,20 @@ struct SpawnedJob {
 }
 
 impl SpawnedJob {
-    fn new(task: JoinHandle<Result<()>>) -> Self {
+    fn new<T>(task: JoinHandle<Result<T>>) -> Self
+    where
+        T: Serialize + Send + 'static,
+    {
         let abort = task.abort_handle();
         let (tx, outcome) = watch::channel(None);
         tokio::spawn(async move {
             let outcome = match task.await {
-                Ok(Ok(())) => Outcome::Succeeded,
+                Ok(Ok(result)) => match serde_json::to_value(result) {
+                    Ok(value) => Outcome::Succeeded(TerminalResult::local(value)),
+                    Err(err) => Outcome::Failed(Arc::new(Error::Runtime {
+                        message: format!("failed to serialize job result: {err}"),
+                    })),
+                },
                 Ok(Err(err)) => Outcome::Failed(Arc::new(err)),
                 Err(err) if err.is_cancelled() => Outcome::Cancelled,
                 Err(err) => Outcome::Failed(Arc::new(Error::Runtime {
@@ -243,7 +296,7 @@ impl JobHandle for SpawnedJob {
     async fn status(&self) -> Result<String> {
         let label = match &*self.outcome.borrow() {
             None => "running",
-            Some(Outcome::Succeeded) => "finished",
+            Some(Outcome::Succeeded(_)) => "finished",
             Some(Outcome::Failed(_)) => "failed",
             Some(Outcome::Cancelled) => "cancelled",
         };
@@ -256,12 +309,11 @@ impl JobHandle for SpawnedJob {
             .wait_for(|outcome| outcome.is_some())
             .await
             .map_err(|_| Error::Runtime {
-                message: "index job outcome was dropped before it completed".to_string(),
+                message: "job outcome was dropped before it completed".to_string(),
             })?
             .clone()
             .expect("wait_for returns once an outcome is set");
-        settled.into_result()?;
-        Ok(TerminalResult::local())
+        settled.into_result()
     }
 
     async fn cancel(&self) -> Result<()> {
@@ -269,3 +321,29 @@ impl JobHandle for SpawnedJob {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::future::pending;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn mapped_spawned_job_reuses_outcome() {
+        let job = Job::spawned(tokio::spawn(async { Ok(41_u64) })).map(|value| value + 1);
+
+        assert_eq!(job.wait().await.unwrap(), 42);
+        assert_eq!(job.wait().await.unwrap(), 42);
+        assert_eq!(job.status().await.unwrap(), "finished");
+    }
+
+    #[tokio::test]
+    async fn mapped_spawned_job_preserves_cancellation() {
+        let job = Job::spawned(tokio::spawn(async { pending::<Result<u64>>().await }))
+            .map(|value| value.to_string());
+
+        job.cancel().await.unwrap();
+        assert!(matches!(job.wait().await, Err(Error::JobCancelled { .. })));
+        assert_eq!(job.status().await.unwrap(), "cancelled");
+    }
+}
diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs
index 169d0fda5..8265d22c0 100644
--- a/rust/lancedb/src/remote/db.rs
+++ b/rust/lancedb/src/remote/db.rs
@@ -513,7 +513,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     async fn get_function(&self, name: &str, version: &str) -> Result<FunctionVersion> {
         let req = self
             .client
-            .post("/v1/functions/get")
+            .post("/v1/functions/describe")
             .json(&serde_json::json!({
                 "name": name,
                 "version": version,
@@ -2520,7 +2520,7 @@ mod tests {
         );
         let conn = Connection::new_with_handler(|request| {
             assert_eq!(request.method(), &reqwest::Method::POST);
-            assert_eq!(request.url().path(), "/v1/functions/get");
+            assert_eq!(request.url().path(), "/v1/functions/describe");
             let body: serde_json::Value =
                 serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
             assert_eq!(
diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs
index fb7278db9..c6b078282 100644
--- a/rust/lancedb/src/remote/table.rs
+++ b/rust/lancedb/src/remote/table.rs
@@ -2864,7 +2864,10 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         })
     }
 
-    async fn refresh_column_async(&self, column: &str) -> Result<Job> {
+    async fn refresh_column_async(
+        &self,
+        column: &str,
+    ) -> Result<Job<crate::function::RefreshColumnResult>> {
         self.check_mutable().await?;
         let mut body = serde_json::json!({ "column": column });
         self.apply_branch_body(&mut body);
@@ -2885,7 +2888,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
             status_code: None,
         })?;
 
-        Ok(Job::new(Box::new(FreshnessJob {
+        Ok(Job::new_typed(Box::new(FreshnessJob {
             inner: RemoteJob::new(self.client.clone(), response.job_id),
             freshness: self.freshness.clone(),
             version: self.version.clone(),
@@ -3280,6 +3283,21 @@ mod tests {
         },
     };
 
+    fn refresh_done(job_id: &str) -> String {
+        json!({
+            "job_id": job_id,
+            "job_state": "DONE",
+            "result": {
+                "rows_assigned": 12,
+                "rows_failed": 0,
+                "rows_remaining": 0,
+                "source_version": 7,
+                "published_version": 8,
+            }
+        })
+        .to_string()
+    }
+
     #[tokio::test]
     async fn test_not_found() {
         let table = Table::new_with_handler("my_table", |_| {
@@ -6892,7 +6910,7 @@ mod tests {
                     .unwrap(),
                 "/v1/jobs/describe" => http::Response::builder()
                     .status(200)
-                    .body(r#"{"job_id": "j-7", "job_state": "DONE"}"#.to_string())
+                    .body(refresh_done("j-7"))
                     .unwrap(),
                 "/v1/table/my_table/count_rows/" => {
                     saw.store(
@@ -6908,7 +6926,9 @@ mod tests {
             });
 
         let job = table.refresh_column_async("doubled").await.unwrap();
-        job.wait().await.unwrap();
+        let result = job.wait().await.unwrap();
+        assert_eq!(result.rows_assigned, 12);
+        assert_eq!(result.published_version, Some(8));
         table.count_rows(None).await.unwrap();
         assert!(
             saw_min_timestamp.load(std::sync::atomic::Ordering::SeqCst),
@@ -6930,7 +6950,7 @@ mod tests {
                     .unwrap(),
                 "/v1/jobs/describe" => http::Response::builder()
                     .status(200)
-                    .body(r#"{"job_id": "j-8", "job_state": "DONE"}"#.to_string())
+                    .body(refresh_done("j-8"))
                     .unwrap(),
                 "/v1/table/my_table/describe/" => {
                     let schema = Schema::new(vec![Field::new("x", DataType::Int32, true)]);
@@ -6976,7 +6996,7 @@ mod tests {
                     .unwrap(),
                 "/v1/jobs/describe" => http::Response::builder()
                     .status(200)
-                    .body(r#"{"job_id": "j-9", "job_state": "DONE"}"#.to_string())
+                    .body(refresh_done("j-9"))
                     .unwrap(),
                 "/v1/table/my_table/tags/version/" => http::Response::builder()
                     .status(200)
@@ -7040,7 +7060,7 @@ mod tests {
                 }
                 "/v1/jobs/describe" => http::Response::builder()
                     .status(200)
-                    .body(r#"{"job_id": "j-10", "job_state": "DONE"}"#.to_string())
+                    .body(refresh_done("j-10"))
                     .unwrap(),
                 "/v1/table/my_table/describe/" => {
                     let schema = Schema::new(vec![Field::new("x", DataType::Int32, true)]);
@@ -7113,7 +7133,7 @@ mod tests {
                 }
                 "/v1/jobs/describe" => http::Response::builder()
                     .status(200)
-                    .body(r#"{"job_id": "j-11", "job_state": "DONE"}"#.to_string())
+                    .body(refresh_done("j-11"))
                     .unwrap(),
                 "/v1/table/my_table/count_rows/" => {
                     *saw.lock().unwrap() = request
diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs
index c2c12fff5..41551bd37 100644
--- a/rust/lancedb/src/table.rs
+++ b/rust/lancedb/src/table.rs
@@ -771,7 +771,10 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
     }
     /// Fill a computed column's unfilled rows, returning a [`Job`] tracking
     /// the operation.
-    async fn refresh_column_async(&self, _column: &str) -> Result<Job> {
+    async fn refresh_column_async(
+        &self,
+        _column: &str,
+    ) -> Result<Job<crate::function::RefreshColumnResult>> {
         Err(Error::NotSupported {
             message: "computed columns are supported only on local tables".into(),
         })
@@ -1708,7 +1711,9 @@ impl Table {
     /// operation instead of blocking until it completes.
     ///
     /// The job may already be complete when returned, and callers must not
-    /// assume the column is filled until [`Job::wait`] returns. Invalid input
+    /// assume the column is filled until [`Job::wait`] returns. A successful
+    /// wait returns the durable [`crate::function::RefreshColumnResult`] for
+    /// both expression-backed and Function-backed columns. Invalid input
     /// -- an unknown column, or one that is not computed -- is reported by
     /// this call rather than by the job. On local tables the job runs as an
     /// in-process task; on LanceDB Cloud and Enterprise it is the server's
@@ -1719,11 +1724,15 @@ impl Table {
     /// # async fn refresh_in_background(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
     /// let job = table.refresh_column_async("doubled").await?;
     /// println!("refresh running: {:?}", job.status().await?);
-    /// job.wait().await?;
+    /// let result = job.wait().await?;
+    /// println!("assigned {} rows", result.rows_assigned);
     /// # Ok(())
     /// # }
     /// ```
-    pub async fn refresh_column_async(&self, column: impl AsRef<str>) -> Result<Job> {
+    pub async fn refresh_column_async(
+        &self,
+        column: impl AsRef<str>,
+    ) -> Result<Job<crate::function::RefreshColumnResult>> {
         self.inner.refresh_column_async(column.as_ref()).await
     }
 
@@ -3425,7 +3434,10 @@ impl BaseTable for NativeTable {
         Ok(result)
     }
 
-    async fn refresh_column_async(&self, column: &str) -> Result<Job> {
+    async fn refresh_column_async(
+        &self,
+        column: &str,
+    ) -> Result<Job<crate::function::RefreshColumnResult>> {
         refresh::execute_refresh_column_async(self, column).await
     }
 
diff --git a/rust/lancedb/src/table/refresh.rs b/rust/lancedb/src/table/refresh.rs
index 7d74bbae7..35f883411 100644
--- a/rust/lancedb/src/table/refresh.rs
+++ b/rust/lancedb/src/table/refresh.rs
@@ -49,11 +49,25 @@ pub struct RefreshColumnResult {
     pub version: u64,
 }
 
+struct RefreshExecution {
+    result: RefreshColumnResult,
+    source_version: u64,
+}
+
 /// Internal implementation of the refresh logic.
 pub(crate) async fn execute_refresh_column(
     table: &NativeTable,
     column: &str,
 ) -> Result<RefreshColumnResult> {
+    Ok(execute_refresh_column_with_source(table, column)
+        .await?
+        .result)
+}
+
+async fn execute_refresh_column_with_source(
+    table: &NativeTable,
+    column: &str,
+) -> Result<RefreshExecution> {
     table.dataset.ensure_mutable()?;
     ensure_no_lsm_write_spec(table).await?;
     let dataset = table.dataset.get().await?;
@@ -87,9 +101,13 @@ pub(crate) async fn execute_refresh_column(
     }
 
     if replacements.is_empty() {
-        return Ok(RefreshColumnResult {
-            rows_filled: 0,
-            version: dataset.version().version,
+        let source_version = dataset.version().version;
+        return Ok(RefreshExecution {
+            result: RefreshColumnResult {
+                rows_filled: 0,
+                version: source_version,
+            },
+            source_version,
         });
     }
 
@@ -110,14 +128,20 @@ pub(crate) async fn execute_refresh_column(
 
     let version = new_dataset.version().version;
     table.dataset.update(new_dataset);
-    Ok(RefreshColumnResult {
-        rows_filled,
-        version,
+    Ok(RefreshExecution {
+        result: RefreshColumnResult {
+            rows_filled,
+            version,
+        },
+        source_version: read_version,
     })
 }
 
 /// Run the refresh as a [`Job`] in this process.
-pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &str) -> Result<Job> {
+pub(crate) async fn execute_refresh_column_async(
+    table: &NativeTable,
+    column: &str,
+) -> Result<Job<crate::function::RefreshColumnResult>> {
     // Validate before spawning so bad input is reported by this call rather
     // than only by the job.
     table.dataset.ensure_mutable()?;
@@ -129,9 +153,16 @@ pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &s
     let table = table.clone();
     let column = column.to_string();
     Ok(Job::spawned(tokio::spawn(async move {
-        execute_refresh_column(&table, &column).await?;
+        let execution = execute_refresh_column_with_source(&table, &column).await?;
         table.bump_freshness();
-        Ok(())
+        Ok(crate::function::RefreshColumnResult {
+            rows_assigned: execution.result.rows_filled,
+            rows_failed: 0,
+            rows_remaining: 0,
+            source_version: execution.source_version,
+            published_version: (execution.result.rows_filled > 0)
+                .then_some(execution.result.version),
+        })
     })))
 }
 
@@ -396,6 +427,17 @@ mod tests {
             read(&table, "doubled").await,
             vec![Some(2), Some(4), Some(6)]
         );
+
+        let no_op = table
+            .refresh_column_async("doubled")
+            .await
+            .unwrap()
+            .wait()
+            .await
+            .unwrap();
+        assert_eq!(no_op.rows_assigned, 0);
+        assert_eq!(no_op.source_version, 3);
+        assert_eq!(no_op.published_version, None);
     }
 
     /// Values written after the last refresh must be reachable by another one.
@@ -646,7 +688,12 @@ mod tests {
 
         let job = table.refresh_column_async("doubled").await.unwrap();
         assert!(job.id().is_none(), "in-process jobs have no server id");
-        job.wait().await.unwrap();
+        let result = job.wait().await.unwrap();
+        assert_eq!(result.rows_assigned, 3);
+        assert_eq!(result.rows_failed, 0);
+        assert_eq!(result.rows_remaining, 0);
+        assert_eq!(result.source_version, 2);
+        assert_eq!(result.published_version, Some(3));
         assert_eq!(job.status().await.unwrap(), "finished");
         assert_eq!(
             read(&table, "doubled").await,
@@ -672,9 +719,9 @@ mod tests {
         declare_doubled(&table).await.unwrap();
 
         let job = table.refresh_column_async("doubled").await.unwrap();
-        job.wait().await.unwrap();
+        let first = job.wait().await.unwrap();
         // A second wait after completion observes the same outcome.
-        job.wait().await.unwrap();
+        assert_eq!(job.wait().await.unwrap(), first);
         assert_eq!(job.status().await.unwrap(), "finished");
     }