feat!: replace get_job/job_history with describe_job/query_job_events (#4130)

A 1M-row column refresh over 200 fragments produced no visible result,
and the client could only ever say `"running"`. Everything needed to
diagnose it already existed server-side — the job registry records a
`claim`/`claim_complete` pair per fragment carrying `rows_processed` —
but none of it was reachable.

## Before

Four ways to ask about a job, none of which told you much.

```python
job = table.refresh_column_async("embedding")
job.status()               # "running". That was the entire debug surface.
db.get_job(job_id)         # state, and a spec. No result, no progress.
db.job_history(job_id)     # raw record batches, no limit, no filter
db.job(job_id)             # a handle that knew nothing
```

## After

Open a job the way you open a table; the handle answers everything.

```python
job = db.open_job(job_id)      # raises JobNotFoundError if there is no such job
```

```python
>>> print(job)
Job(
    id='job-1',
    state='failed',
    job_type='refresh_column',
    creation_ms=1757000000000,
    spec={
        "column": "embedding",
        "num_workers": 4
    },
    failure=JobFailureInfo(phase='execute', message='worker died', retryable=True),
)
```

Individual fields are there too — `job.state`, `job.job_type`,
`job.creation_ms`, `job.spec`, `job.result`, `job.failure` — and
`job.result` carries `rows_assigned` / `rows_failed` as soon as the job
succeeds, with no `wait()` required.

Per-fragment progress *while it is still running*:

```python
done = job.events(filter="state = 'claim_complete'", limit=10_000)
done.column("rows_processed").to_pylist()      # [5000, 5000, ...]
```

The handle an async action returns is the same object, one `refresh()`
away:

```python
job = table.refresh_column_async("embedding")
job.refresh()
job.state, job.result
```

TypeScript is the same experience, down to `console.log`:

```ts
const job = await db.openJob(jobId);   // rejects if there is no such job
console.log(job);                      // same multi-line layout
job.state; job.jobType; job.spec; job.result; job.failure;
const done = await job.events({ filter: "state = 'claim_complete'", limit: 10_000 });
```

## Why each piece matters

- **A result without waiting.** `rows_assigned` / `rows_failed` used to
live only on the terminal result, so a job that never terminated
reported nothing at all.
- **`limit`.** The server caps event rows at 1000 and truncates without
saying so, which silently hid most of a 200-fragment job's history.
- **`filter`.** `claim_complete` rows carry per-claim `rows_processed` —
the only progress signal that exists mid-flight.
- **Events outlive the worker.** They live in the job registry, not in
pod logs that vanish with the pod.
- **One place to ask.** `open_job` replaces `describe_job`,
`query_job_events` and `job`, so a question about a job has one answer
instead of one per calling location.
- **A missing job is an error, not a `None`.** The common case is a job
id copied out of a log, where absence is the surprise worth raising —
and it matches `open_table`.
- **Printing is the debug surface.** Every field on its own line, JSON
payloads keeping their structure. An unrefreshed handle stays on one
line, because there is nothing to lay out.
- **In-process jobs say so.** A local refresh reports `state` and leaves
the rest null rather than inventing fields it has no record for.

`list_jobs` and `cancel_job` stay as they were: one lists, the other is
a one-shot action that should not need a describe first.

## Breaking

All shipped in 0.38.0. No deprecated aliases.

| Was | Now |
| --- | --- |
| `Connection.get_job` → `describe_job` | `Connection.open_job` returns
a populated `Job`, or raises |
| `Connection.job_history` → `query_job_events` | `job.events(...)` |
| `Connection.job` | `Connection.open_job` |
| Python events → `List[pa.RecordBatch]` | `pa.Table` |
| `JobDescription.spec_json` / `.result_json` | internal; use `job.spec`
/ `job.result` |

Node's `Job` is now a TypeScript class wrapping the native handle, so it
returns an Arrow table and parsed values like Python does. New
`Error::JobNotFound` / `JobNotFoundError`; the three job exceptions are
now in the Python API reference.

python/python/lancedb/_lancedb.pyi

@@ -148,17 +148,13 @@ class Connection(object):
         start_after: Optional[str],
         limit: Optional[int],
     ) -> list[str]: ...  # Deprecated: Use list_tables instead
-    def job(self, job_id: str) -> Job: ...
+    async def open_job(self, job_id: str) -> Job: ...
     async def create_function_async(self, request_json: str) -> Job: ...
     async def get_function(self, name: str, version: str) -> str: ...
     async def list_functions(self) -> List[str]: ...
     async def drop_function(self, name: str, version: str) -> bool: ...
     async def list_jobs(self) -> List[JobInfo]: ...
-    async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
     async def cancel_job(self, job_id: str) -> bool: ...
-    async def job_history(
-        self, job_id: Optional[str] = None
-    ) -> List[pa.RecordBatch]: ...
     async def execute_query_async(
         self,
         query: str,
@@ -244,9 +240,20 @@ class BlobFile:
 class Job:
     @property
     def id(self) -> Optional[str]: ...
+    @property
+    def _state(self) -> Optional[str]: ...
+    @property
+    def _description(self) -> Optional[JobDescription]: ...
     async def status(self) -> str: ...
     async def wait(self) -> Optional[str]: ...
     async def cancel(self) -> None: ...
+    async def refresh(self) -> None: ...
+    async def events(
+        self,
+        *,
+        limit: Optional[int] = None,
+        filter: Optional[str] = None,
+    ) -> pa.Table: ...
 
 class JobInfo:
     @property
@@ -278,7 +285,13 @@ class JobDescription:
     @property
     def creation_ms(self) -> int: ...
     @property
-    def spec_json(self) -> Optional[str]: ...
+    def _spec_json(self) -> Optional[str]: ...
+    @property
+    def _result_json(self) -> Optional[str]: ...
+    @property
+    def spec(self) -> Optional[Any]: ...
+    @property
+    def result(self) -> Optional[Any]: ...
     @property
     def failure(self) -> Optional[JobFailureInfo]: ...
 

python/python/lancedb/db.py

@@ -76,7 +76,7 @@ if TYPE_CHECKING:
     from .pydantic import LanceModel
 
     from ._lancedb import Connection as LanceDbConnection
-    from ._lancedb import JobDescription, JobInfo
+    from ._lancedb import JobInfo
     from .common import DATA, URI
     from .embeddings import EmbeddingFunctionConfig
     from ._lancedb import Session
@@ -745,26 +745,23 @@ class DBConnection(EnforceOverrides):
             "Function catalog operations are not supported for this connection type"
         )
 
-    def job(self, job_id: str) -> Job:
-        """A [Job][lancedb.job.Job] handle for a server-side job by id.
+    def open_job(self, job_id: str) -> Job:
+        """Open a server-side job by id, returning a handle with its record
+        already populated.
 
-        The handle is constructed without a server round trip; an unknown id
-        surfaces when the handle is used. Dropping the handle has no effect
-        on the job itself.
+        The returned [Job][lancedb.job.Job] answers for its own state,
+        specification, result, failure and event history, so there is no
+        separate connection-level call for any of them.
+
+        Raises `JobNotFoundError` when the server has no such job, the way
+        `open_table` does for a missing table.
         """
-        raise NotImplementedError("job is not supported for this connection type")
+        raise NotImplementedError("open_job is not supported for this connection type")
 
     def list_jobs(self) -> List[JobInfo]:
         """List server-side jobs across the database's tables."""
         raise NotImplementedError("list_jobs is not supported for this connection type")
 
-    def get_job(self, job_id: str) -> Optional[JobDescription]:
-        """Describe a single server-side job by id.
-
-        Returns None when the server has no such job.
-        """
-        raise NotImplementedError("get_job is not supported for this connection type")
-
     def cancel_job(self, job_id: str) -> bool:
         """Request cancellation of a server-side job by id.
 
@@ -776,15 +773,6 @@ class DBConnection(EnforceOverrides):
             "cancel_job is not supported for this connection type"
         )
 
-    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
-        """The lifecycle event history of a server-side job, as Arrow batches.
-
-        Lists history across all jobs when `job_id` is None.
-        """
-        raise NotImplementedError(
-            "job_history is not supported for this connection type"
-        )
-
     def execute_query(
         self,
         query: str,
@@ -1462,14 +1450,11 @@ class LanceDBConnection(DBConnection):
         )
 
     @override
-    def job(self, job_id: str) -> Job:
-        """A [Job][lancedb.job.Job] handle for a server-side job by id.
-
-        The handle is constructed without a server round trip; an unknown id
-        surfaces when the handle is used. Dropping the handle has no effect
-        on the job itself.
+    def open_job(self, job_id: str) -> Job:
+        """Open a server-side job by id. See
+        [DBConnection.open_job][lancedb.db.DBConnection.open_job].
         """
-        return Job(self._conn.job(job_id))
+        return Job(LOOP.run(self._conn.open_job(job_id)))
 
     @override
     def create_function_async(self, definition: UdfDefinition) -> Job[FunctionVersion]:
@@ -1493,14 +1478,6 @@ class LanceDBConnection(DBConnection):
         """List server-side jobs across the database's tables."""
         return LOOP.run(self._conn.list_jobs())
 
-    @override
-    def get_job(self, job_id: str) -> Optional[JobDescription]:
-        """Describe a single server-side job by id.
-
-        Returns None when the server has no such job.
-        """
-        return LOOP.run(self._conn.get_job(job_id))
-
     @override
     def cancel_job(self, job_id: str) -> bool:
         """Request cancellation of a server-side job by id.
@@ -1511,14 +1488,6 @@ class LanceDBConnection(DBConnection):
         """
         return LOOP.run(self._conn.cancel_job(job_id))
 
-    @override
-    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
-        """The lifecycle event history of a server-side job, as Arrow batches.
-
-        Lists history across all jobs when `job_id` is None.
-        """
-        return LOOP.run(self._conn.job_history(job_id))
-
     @override
     def namespace_client(self) -> LanceNamespace:
         """Get the equivalent namespace client for this connection.
@@ -2289,15 +2258,11 @@ class AsyncConnection(object):
             namespace_path = []
         await self._inner.drop_all_tables(namespace_path=namespace_path)
 
-    def job(self, job_id: str) -> AsyncJob:
-        """An [AsyncJob][lancedb.job.AsyncJob] handle for a server-side job
-        by id.
-
-        The handle is constructed without a server round trip; an unknown id
-        surfaces when the handle is used. Dropping the handle has no effect
-        on the job itself.
+    async def open_job(self, job_id: str) -> AsyncJob:
+        """Open a server-side job by id. See
+        [DBConnection.open_job][lancedb.db.DBConnection.open_job].
         """
-        return AsyncJob(self._inner.job(job_id))
+        return AsyncJob(await self._inner.open_job(job_id))
 
     async def create_function_async(
         self, definition: UdfDefinition
@@ -2337,13 +2302,6 @@ class AsyncConnection(object):
         """List server-side jobs across the database's tables."""
         return await self._inner.list_jobs()
 
-    async def get_job(self, job_id: str) -> Optional[JobDescription]:
-        """Describe a single server-side job by id.
-
-        Returns None when the server has no such job.
-        """
-        return await self._inner.get_job(job_id)
-
     async def cancel_job(self, job_id: str) -> bool:
         """Request cancellation of a server-side job by id.
 
@@ -2353,13 +2311,6 @@ class AsyncConnection(object):
         """
         return await self._inner.cancel_job(job_id)
 
-    async def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
-        """The lifecycle event history of a server-side job, as Arrow batches.
-
-        Lists history across all jobs when `job_id` is None.
-        """
-        return await self._inner.job_history(job_id)
-
     async def execute_query(
         self,
         query: str,

python/python/lancedb/exceptions.py

@@ -35,3 +35,9 @@ class JobCancelledError(RuntimeError):
     """Exception raised when an asynchronous job was cancelled."""
 
     pass
+
+
+class JobNotFoundError(ValueError):
+    """Exception raised when opening a job the server does not have."""
+
+    pass

python/python/lancedb/job.py

@@ -4,15 +4,27 @@
 """Handles to operations a server may run asynchronously."""
 
 import asyncio
+import json
 from datetime import timedelta
 from typing import Any, Callable, Generic, Optional, TypeVar, cast
 
+import pyarrow as pa
+
 from lancedb.background_loop import LOOP
 
 from . import _lancedb
+from ._lancedb import JobDescription, JobFailureInfo, JobInfo
 
 T = TypeVar("T")
 
+__all__ = [
+    "AsyncJob",
+    "Job",
+    "JobDescription",
+    "JobFailureInfo",
+    "JobInfo",
+]
+
 
 class AsyncJob(Generic[T]):
     """A handle to an operation that may still be running.
@@ -78,6 +90,149 @@ class AsyncJob(Generic[T]):
             return
         await self._inner.cancel()
 
+    async def refresh(self) -> None:
+        """Ask the backend for this job's current state, and for a server-side
+        job its full record, then cache it for the properties below.
+
+        The properties are all `None` until this runs, because submitting an
+        operation returns only a job id. `status` fetches the whole record too;
+        `wait` records only the terminal state it establishes.
+        """
+        if self._inner is None:
+            return
+        await self._inner.refresh()
+
+    @property
+    def state(self) -> Optional[str]:
+        """The last observed lifecycle state, without contacting the backend.
+
+        `None` until the handle has talked to it. See :meth:`AsyncJob.refresh`.
+        """
+        if self._inner is None:
+            return "finished"
+        return self._inner._state
+
+    @property
+    def job_type(self) -> Optional[str]:
+        """The job's type, as the server names it.
+
+        `None` for an in-process job, which has no server-side record.
+        """
+        return self._field("job_type")
+
+    @property
+    def creation_ms(self) -> Optional[int]:
+        """When the job was created, in milliseconds since the epoch."""
+        return self._field("creation_ms")
+
+    @property
+    def spec(self) -> Optional[Any]:
+        """The job-type-specific specification it was submitted with."""
+        return self._field("spec")
+
+    @property
+    def result(self) -> Optional[Any]:
+        """The job-type-specific terminal result, as reported data rather than
+        the typed model :meth:`AsyncJob.wait` returns.
+
+        `None` until the job succeeds, so a job that never terminates reports
+        its progress through :meth:`AsyncJob.events` instead.
+        """
+        return self._field("result")
+
+    @property
+    def failure(self) -> Optional[JobFailureInfo]:
+        """Why the job failed, when it failed and the server reports a reason."""
+        return self._field("failure")
+
+    @property
+    def _spec_json(self) -> Optional[str]:
+        return self._field("_spec_json")
+
+    @property
+    def _result_json(self) -> Optional[str]:
+        return self._field("_result_json")
+
+    def _field(self, name: str) -> Optional[Any]:
+        description = self._inner._description if self._inner is not None else None
+        return getattr(description, name) if description is not None else None
+
+    async def events(
+        self,
+        *,
+        limit: Optional[int] = None,
+        filter: Optional[str] = None,
+    ) -> "pa.Table":
+        """This job's recorded lifecycle events.
+
+        Where the properties above report a terminal result only once the job
+        reaches one, events are written as the job runs and outlive the workers
+        that produced them. A distributed job records a `claim`/`claim_complete`
+        pair per unit of work, each carrying `rows_processed`, so a job that
+        never finishes still accounts for what it did.
+
+        Parameters
+        ----------
+        limit: int, optional
+            Maximum event rows to return. The server caps results at 1000 by
+            default and 10,000 at most, and truncates without saying so, so
+            pass this for a job that emits an event per fragment.
+        filter: str, optional
+            SQL-like expression over the `state`, `updated_by`, `emitted_from`,
+            `emitted_by`, and `claim_entity` columns, such as
+            ``state = 'claim_complete'``.
+        """
+        if self._inner is None:
+            raise NotImplementedError(
+                "job event history is only available for server-side jobs"
+            )
+        return await self._inner.events(limit=limit, filter=filter)
+
+    def __repr__(self) -> str:
+        return _job_repr("AsyncJob", self)
+
+
+_REPR_INDENT = " " * 4
+
+
+def _repr_payload(value: Any) -> str:
+    """Render a job payload as indented JSON, aligned under its field."""
+    try:
+        rendered = json.dumps(value, indent=4)
+    except TypeError:
+        return repr(value)
+    return rendered.replace("\n", "\n" + _REPR_INDENT)
+
+
+def _job_repr(kind: str, job: Any) -> str:
+    """Render every field the handle currently knows, omitting the rest.
+
+    One field per line, with the JSON payloads indented, because a refresh
+    job's spec and result are the point of printing it.
+    """
+    state = job.state
+    if state is None:
+        # Nothing has been fetched yet, so there is nothing to lay out.
+        known = f"id={job.id!r}, " if job.id is not None else ""
+        return f"{kind}({known}not refreshed)"
+
+    fields = []
+    if job.id is not None:
+        fields.append(f"id={job.id!r}")
+    fields.append(f"state={state!r}")
+    for name in ("job_type", "creation_ms"):
+        value = getattr(job, name)
+        if value is not None:
+            fields.append(f"{name}={value!r}")
+    for name in ("spec", "result"):
+        value = getattr(job, name)
+        if value is not None:
+            fields.append(f"{name}={_repr_payload(value)}")
+    if job.failure is not None:
+        fields.append(f"failure={job.failure!r}")
+    body = "".join(f"\n{_REPR_INDENT}{field}," for field in fields)
+    return f"{kind}({body}\n)"
+
 
 class Job(Generic[T]):
     """Synchronous counterpart of `AsyncJob` with the same result type."""
@@ -122,6 +277,75 @@ class Job(Generic[T]):
             return
         LOOP.run(self._inner.cancel())
 
+    def refresh(self) -> None:
+        """Ask the backend for this job's current state and record.
+
+        See :meth:`AsyncJob.refresh`.
+        """
+        if self._inner is None:
+            return
+        LOOP.run(self._inner.refresh())
+
+    @property
+    def state(self) -> Optional[str]:
+        """The last observed lifecycle state. See :attr:`AsyncJob.state`."""
+        return self._inner.state if self._inner is not None else "finished"
+
+    @property
+    def job_type(self) -> Optional[str]:
+        """The job's type. See :attr:`AsyncJob.job_type`."""
+        return self._field("job_type")
+
+    @property
+    def creation_ms(self) -> Optional[int]:
+        """When the job was created. See :attr:`AsyncJob.creation_ms`."""
+        return self._field("creation_ms")
+
+    @property
+    def spec(self) -> Optional[Any]:
+        """The job's specification. See :attr:`AsyncJob.spec`."""
+        return self._field("spec")
+
+    @property
+    def result(self) -> Optional[Any]:
+        """The job's terminal result. See :attr:`AsyncJob.result`."""
+        return self._field("result")
+
+    @property
+    def failure(self) -> Optional[JobFailureInfo]:
+        """Why the job failed. See :attr:`AsyncJob.failure`."""
+        return self._field("failure")
+
+    @property
+    def _spec_json(self) -> Optional[str]:
+        return self._field("_spec_json")
+
+    @property
+    def _result_json(self) -> Optional[str]:
+        return self._field("_result_json")
+
+    def _field(self, name: str) -> Optional[Any]:
+        return getattr(self._inner, name) if self._inner is not None else None
+
+    def events(
+        self,
+        *,
+        limit: Optional[int] = None,
+        filter: Optional[str] = None,
+    ) -> "pa.Table":
+        """This job's recorded lifecycle events.
+
+        See :meth:`AsyncJob.events`.
+        """
+        if self._inner is None:
+            raise NotImplementedError(
+                "job event history is only available for server-side jobs"
+            )
+        return LOOP.run(self._inner.events(limit=limit, filter=filter))
+
+    def __repr__(self) -> str:
+        return _job_repr("Job", self)
+
 
 def _typed_job(
     inner: "_lancedb.Job", result_decoder: Callable[[str], T]

python/python/lancedb/remote/db.py

@@ -31,7 +31,7 @@ from ..sql import QueryDescription
 from ..materialized_view import MaterializedView, SelectArg
 
 if TYPE_CHECKING:
-    from .._lancedb import JobDescription, JobInfo
+    from .._lancedb import JobInfo
 from ..embeddings import EmbeddingFunctionConfig
 from lance_namespace import (
     LanceNamespace,
@@ -739,14 +739,11 @@ class RemoteDBConnection(DBConnection):
         )
 
     @override
-    def job(self, job_id: str) -> Job:
-        """A [Job][lancedb.job.Job] handle for a server-side job by id.
-
-        The handle is constructed without a server round trip; an unknown id
-        surfaces when the handle is used. Dropping the handle has no effect
-        on the job itself.
+    def open_job(self, job_id: str) -> Job:
+        """Open a server-side job by id. See
+        [DBConnection.open_job][lancedb.db.DBConnection.open_job].
         """
-        return Job(self._conn.job(job_id))
+        return Job(LOOP.run(self._conn.open_job(job_id)))
 
     @override
     def create_function_async(self, definition: UdfDefinition) -> Job[FunctionVersion]:
@@ -769,14 +766,6 @@ class RemoteDBConnection(DBConnection):
         """List server-side jobs across the database's tables."""
         return LOOP.run(self._conn.list_jobs())
 
-    @override
-    def get_job(self, job_id: str) -> Optional["JobDescription"]:
-        """Describe a single server-side job by id.
-
-        Returns None when the server has no such job.
-        """
-        return LOOP.run(self._conn.get_job(job_id))
-
     @override
     def cancel_job(self, job_id: str) -> bool:
         """Request cancellation of a server-side job by id.
@@ -787,14 +776,6 @@ class RemoteDBConnection(DBConnection):
         """
         return LOOP.run(self._conn.cancel_job(job_id))
 
-    @override
-    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
-        """The lifecycle event history of a server-side job, as Arrow batches.
-
-        Lists history across all jobs when `job_id` is None.
-        """
-        return LOOP.run(self._conn.job_history(job_id))
-
     @override
     def execute_query_async(
         self,

python/python/tests/test_remote_db.py

@@ -2467,7 +2467,7 @@ def test_remote_blob_byte_apis_not_supported_on_old_server():
 
 
 def test_remote_connection_jobs_surface():
-    from lancedb.exceptions import JobFailedError
+    from lancedb.exceptions import JobFailedError, JobNotFoundError
 
     schema = pa.schema([("state", pa.string())])
     batch = pa.record_batch([pa.array(["created", "done"])], schema=schema)
@@ -2475,6 +2475,7 @@ def test_remote_connection_jobs_surface():
     with pa.ipc.new_stream(sink, schema) as writer:
         writer.write_batch(batch)
     events_body = sink.getvalue().to_pybytes()
+    query_events_payloads = []
 
     def handler(request):
         content_len = int(request.headers.get("Content-Length", 0))
@@ -2512,6 +2513,22 @@ def test_remote_connection_jobs_surface():
             request.end_headers()
             request.wfile.write(json.dumps(rsp).encode())
         elif request.path == "/v1/jobs/describe":
+            if payload["job_id"] == "job-2":
+                request.send_response(200)
+                request.send_header("Content-Type", "application/json")
+                request.end_headers()
+                request.wfile.write(
+                    json.dumps(
+                        dict(
+                            job_id="job-2",
+                            job_type="refresh_column",
+                            job_state="DONE",
+                            creation_ms=2000,
+                            result=dict(rows_assigned=1000000, rows_failed=0),
+                        )
+                    ).encode()
+                )
+                return
             if payload["job_id"] != "job-1":
                 request.send_response(404)
                 request.end_headers()
@@ -2543,7 +2560,7 @@ def test_remote_connection_jobs_surface():
             request.end_headers()
             request.wfile.write(b'{"job_id": "job-1"}')
         elif request.path == "/v1/jobs/query_events":
-            assert payload["job_id"] == "job-1"
+            query_events_payloads.append(payload)
             request.send_response(200)
             request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
             request.end_headers()
@@ -2559,24 +2576,109 @@ def test_remote_connection_jobs_surface():
         assert jobs[0].table == "t1"
         assert jobs[1].state == "finished"
 
-        description = db.get_job("job-1")
-        assert description.job_type == "create_index"
-        assert description.state == "failed"
-        assert json.loads(description.spec_json) == {"column": "vec"}
-        assert description.failure.message == "worker died"
-        assert description.failure.retryable is True
-        assert db.get_job("missing") is None
-
         assert db.cancel_job("job-1") is True
         assert db.cancel_job("missing") is False
 
-        batches = db.job_history("job-1")
-        assert len(batches) == 1
-        assert batches[0].num_rows == 2
-        assert batches[0].column("state").to_pylist() == ["created", "done"]
+        # Opening a job hands back a populated handle; a missing one fails.
+        with pytest.raises(JobNotFoundError, match="missing"):
+            db.open_job("missing")
+        finished = db.open_job("job-2")
+        assert finished.state == "finished"
+        assert finished.result == {"rows_assigned": 1000000, "rows_failed": 0}
 
-        job = db.job("job-1")
+        job = db.open_job("job-1")
         assert job.id == "job-1"
+        # Opening already populated the handle.
+        assert job.state == "failed"
+        assert job.spec == {"column": "vec"}
+        assert job.failure.message == "worker died"
         assert job.status() == "failed"
         with pytest.raises(JobFailedError, match="worker died"):
             job.wait(timeout=timedelta(seconds=5))
+
+
+def test_remote_job_handle_reports_its_own_detail():
+    schema = pa.schema([("state", pa.string())])
+    batch = pa.record_batch([pa.array(["claim_complete"])], schema=schema)
+    sink = pa.BufferOutputStream()
+    with pa.ipc.new_stream(sink, schema) as writer:
+        writer.write_batch(batch)
+    events_body = sink.getvalue().to_pybytes()
+    event_payloads = []
+
+    def handler(request):
+        content_len = int(request.headers.get("Content-Length", 0))
+        body = request.rfile.read(content_len) if content_len > 0 else b""
+        payload = json.loads(body) if body else {}
+        if request.path == "/v1/jobs/describe":
+            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_type="refresh_column",
+                        job_state="DONE",
+                        creation_ms=2000,
+                        spec=dict(column="vec"),
+                        result=dict(rows_assigned=1000000),
+                    )
+                ).encode()
+            )
+        elif request.path == "/v1/jobs/query_events":
+            event_payloads.append(payload)
+            request.send_response(200)
+            request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
+            request.end_headers()
+            request.wfile.write(events_body)
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        job = db.open_job("job-1")
+
+        # Opening populates the handle in the same round trip.
+        assert job.state == "finished"
+        job.refresh()
+        assert job.job_type == "refresh_column"
+        assert job.creation_ms == 2000
+        assert job.spec == {"column": "vec"}
+        assert job.result == {"rows_assigned": 1000000}
+        assert job.failure is None
+        # The JSON payloads stay reachable, but as internal APIs.
+        assert json.loads(job._spec_json) == {"column": "vec"}
+        assert json.loads(job._result_json) == {"rows_assigned": 1000000}
+
+        # print() shows everything the handle knows and nothing it does not.
+        # print() lays every known field out on its own line, with the JSON
+        # payloads indented rather than crammed onto one line.
+        assert repr(job) == "\n".join(
+            [
+                "Job(",
+                "    id='job-1',",
+                "    state='finished',",
+                "    job_type='refresh_column',",
+                "    creation_ms=2000,",
+                "    spec={",
+                '        "column": "vec"',
+                "    },",
+                "    result={",
+                '        "rows_assigned": 1000000',
+                "    },",
+                ")",
+            ]
+        )
+        # Nothing it does not know shows up.
+        assert "failure" not in repr(job)
+
+        events = job.events(filter="state = 'claim_complete'", limit=500)
+        assert isinstance(events, pa.Table)
+        assert events.column("state").to_pylist() == ["claim_complete"]
+        # The handle supplies job_id; the caller only narrows the query.
+        assert event_payloads[-1] == {
+            "job_id": "job-1",
+            "limit": 500,
+            "filter": "state = 'claim_complete'",
+        }

python/src/connection.rs

@@ -13,11 +13,7 @@ use crate::{
     runtime::future_into_py,
     table::Table,
 };
-use arrow::{
-    datatypes::Schema,
-    ffi_stream::ArrowArrayStreamReader,
-    pyarrow::{FromPyArrow, ToPyArrow},
-};
+use arrow::{datatypes::Schema, ffi_stream::ArrowArrayStreamReader, pyarrow::FromPyArrow};
 use lancedb::{
     connection::Connection as LanceConnection,
     connection::NamespaceClientPushdownOperation,
@@ -28,7 +24,7 @@ use pyo3::{
     Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
     exceptions::{PyRuntimeError, PyValueError},
     pyclass, pyfunction, pymethods,
-    types::{PyAnyMethods, PyDict, PyDictMethods, PyList, PyListMethods},
+    types::{PyAnyMethods, PyDict, PyDictMethods, PyList},
 };
 
 #[pyclass]
@@ -644,9 +640,12 @@ impl Connection {
         })
     }
 
-    pub fn job(&self, job_id: String) -> PyResult<crate::job::Job> {
-        let inner = self.get_inner()?.clone();
-        Ok(crate::job::Job::new(inner.job(job_id).infer_error()?))
+    pub fn open_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let job = inner.open_job(&job_id).await.infer_error()?;
+            Ok(crate::job::Job::new(job))
+        })
     }
 
     pub fn create_function_async(
@@ -716,38 +715,12 @@ impl Connection {
         })
     }
 
-    pub fn get_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
-        let inner = self_.get_inner()?.clone();
-        future_into_py(self_.py(), async move {
-            let description = inner.get_job(&job_id).await.infer_error()?;
-            Ok(description.map(crate::job::JobDescription::from))
-        })
-    }
-
     pub fn cancel_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.get_inner()?.clone();
         future_into_py(self_.py(), async move {
             inner.cancel_job(&job_id).await.infer_error()
         })
     }
-
-    #[pyo3(signature = (job_id=None))]
-    pub fn job_history(
-        self_: PyRef<'_, Self>,
-        job_id: Option<String>,
-    ) -> PyResult<Bound<'_, PyAny>> {
-        let inner = self_.get_inner()?.clone();
-        future_into_py(self_.py(), async move {
-            let batches = inner.job_history(job_id.as_deref()).await.infer_error()?;
-            Python::attach(|py| {
-                let list = PyList::empty(py);
-                for batch in batches {
-                    list.append(batch.to_pyarrow(py)?)?;
-                }
-                Ok(list.unbind())
-            })
-        })
-    }
 }
 
 #[pyfunction]

python/src/error.rs

@@ -114,6 +114,12 @@ impl<T> PythonErrorExt<T> for std::result::Result<T, LanceError> {
                         .getattr(intern!(py, "JobCancelledError"))?;
                     Err(PyErr::from_value(cls.call1((err.to_string(),))?))
                 }),
+                LanceError::JobNotFound { .. } => Python::attach(|py| {
+                    let cls = py
+                        .import(intern!(py, "lancedb.exceptions"))?
+                        .getattr(intern!(py, "JobNotFoundError"))?;
+                    Err(PyErr::from_value(cls.call1((err.to_string(),))?))
+                }),
                 _ => self.runtime_error(),
             },
         }

python/src/job.rs

@@ -4,11 +4,50 @@
 use std::sync::Arc;
 
 use crate::runtime::future_into_py;
-use pyo3::{Bound, PyAny, PyRef, PyResult, pyclass, pymethods};
+use arrow::{
+    datatypes::Schema,
+    pyarrow::{IntoPyArrow, Table as PyArrowTable},
+};
+use lancedb::job::JobEventsRequest;
+use pyo3::{
+    Bound, PyAny, PyRef, PyResult, Python,
+    exceptions::PyValueError,
+    pyclass, pymethods,
+    types::{PyAnyMethods, PyDict, PyDictMethods},
+};
 use serde::Serialize;
 
 use crate::error::PythonErrorExt;
 
+const REPR_INDENT: &str = "    ";
+
+/// Parse a stored JSON payload into Python data. The bindings carry these as
+/// strings because that is what crosses the boundary cheaply; the public
+/// Python surface is the parsed form.
+fn parse_json_payload<'py>(
+    py: Python<'py>,
+    raw: Option<&str>,
+) -> PyResult<Option<Bound<'py, PyAny>>> {
+    match raw {
+        None => Ok(None),
+        Some(raw) => Ok(Some(py.import("json")?.call_method1("loads", (raw,))?)),
+    }
+}
+
+/// A payload rendered as indented JSON, aligned under the field that holds it.
+fn pretty_json_payload(py: Python<'_>, raw: Option<&str>) -> PyResult<Option<String>> {
+    let Some(parsed) = parse_json_payload(py, raw)? else {
+        return Ok(None);
+    };
+    let kwargs = PyDict::new(py);
+    kwargs.set_item("indent", 4)?;
+    let rendered: String = py
+        .import("json")?
+        .call_method("dumps", (parsed,), Some(&kwargs))?
+        .extract()?;
+    Ok(Some(rendered.replace('\n', &format!("\n{REPR_INDENT}"))))
+}
+
 #[pyclass]
 pub struct Job {
     inner: Arc<lancedb::Job<std::result::Result<Option<String>, String>>>,
@@ -67,6 +106,48 @@ impl Job {
             Ok(())
         })
     }
+
+    pub fn refresh(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            inner.refresh().await.infer_error()?;
+            Ok(())
+        })
+    }
+
+    /// The last observed lifecycle state, without contacting the backend.
+    #[getter]
+    pub fn _state(&self) -> Option<String> {
+        self.inner.state()
+    }
+
+    /// The last observed server-side record. `None` for an in-process job.
+    #[getter]
+    pub fn _description(&self) -> Option<JobDescription> {
+        self.inner.description().map(JobDescription::from)
+    }
+
+    #[pyo3(signature = (*, limit=None, filter=None))]
+    pub fn events(
+        self_: PyRef<'_, Self>,
+        limit: Option<u32>,
+        filter: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        let request = JobEventsRequest { limit, filter };
+        future_into_py(self_.py(), async move {
+            let batches = inner.events(request).await.infer_error()?;
+            Python::attach(|py| {
+                let schema = batches
+                    .first()
+                    .map(|batch| batch.schema())
+                    .unwrap_or_else(|| Arc::new(Schema::empty()));
+                let table = PyArrowTable::try_new(batches, schema)
+                    .map_err(|err| PyValueError::new_err(err.to_string()))?;
+                table.into_pyarrow(py).map(|table| table.unbind())
+            })
+        })
+    }
 }
 
 /// A row from `Connection.list_jobs`: one server-side job.
@@ -121,7 +202,7 @@ impl JobFailureInfo {
     }
 }
 
-/// A described job from `Connection.get_job`.
+/// The server-side record behind a `Job` handle.
 #[pyclass(get_all, skip_from_py_object)]
 #[derive(Clone)]
 pub struct JobDescription {
@@ -129,17 +210,49 @@ pub struct JobDescription {
     job_type: String,
     state: String,
     creation_ms: i64,
-    spec_json: Option<String>,
+    /// Internal: the wire form behind the `spec` property.
+    _spec_json: Option<String>,
+    /// Internal: the wire form behind the `result` property.
+    _result_json: Option<String>,
     failure: Option<JobFailureInfo>,
 }
 
 #[pymethods]
 impl JobDescription {
-    fn __repr__(&self) -> String {
-        format!(
-            "JobDescription(job_id={:?}, job_type={:?}, state={:?}, creation_ms={})",
-            self.job_id, self.job_type, self.state, self.creation_ms
-        )
+    /// The job-type-specific specification it was submitted with.
+    #[getter]
+    fn spec<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
+        parse_json_payload(py, self._spec_json.as_deref())
+    }
+
+    /// The job-type-specific terminal result. `None` until the job succeeds.
+    #[getter]
+    fn result<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
+        parse_json_payload(py, self._result_json.as_deref())
+    }
+
+    fn __repr__(&self, py: Python<'_>) -> PyResult<String> {
+        let mut fields = vec![
+            format!("job_id={:?}", self.job_id),
+            format!("job_type={:?}", self.job_type),
+            format!("state={:?}", self.state),
+            format!("creation_ms={}", self.creation_ms),
+        ];
+        // Lay the payloads out as indented JSON, the same way the `Job` repr
+        // does, so the two agree on how the same data looks.
+        for (name, payload) in [("spec", &self._spec_json), ("result", &self._result_json)] {
+            if let Some(rendered) = pretty_json_payload(py, payload.as_deref())? {
+                fields.push(format!("{name}={rendered}"));
+            }
+        }
+        if let Some(failure) = &self.failure {
+            fields.push(format!("failure={}", failure.__repr__()));
+        }
+        let body = fields
+            .iter()
+            .map(|field| format!("\n{REPR_INDENT}{field},"))
+            .collect::<String>();
+        Ok(format!("JobDescription({body}\n)"))
     }
 }
 
@@ -150,7 +263,11 @@ impl From<lancedb::database::JobDescription> for JobDescription {
             job_type: description.job_type,
             state: description.state,
             creation_ms: description.creation_ms,
-            spec_json: (!description.spec.is_null()).then(|| description.spec.to_string()),
+            _spec_json: (!description.spec.is_null()).then(|| description.spec.to_string()),
+            _result_json: description
+                .result
+                .filter(|result| !result.is_null())
+                .map(|result| result.to_string()),
             failure: description.failure.map(|failure| JobFailureInfo {
                 phase: failure.phase,
                 message: failure.message,