remote client: platform jobs API (describe/resolve/cancel)

Bind the client to the server's platform jobs endpoints:

- describe_platform_job -> POST /v1/jobs/describe: registry lifecycle
  state (IN_PROGRESS/CANCELLED/FAILED/DONE) plus the owner-written
  status payload (units/rows/error); 404 -> None.
- resolve_platform_job_id -> POST /v1/jobs/list with the manifest-id
  filter: one-call resolution from the submission id to the platform
  id; None until the job registers (dispatch is async).
- cancel_platform_job -> POST /v1/jobs/cancel: idempotent on terminal
  jobs.

Database trait defaults to NotSupported so non-server backends are
unaffected; Connection passes through.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

rust/lancedb/src/connection.rs

@@ -25,8 +25,8 @@ use crate::database::listing::ListingDatabase;
 use crate::database::{
     CloneTableRequest, CreateFunctionRequest, CreateMaterializedViewRequest, Database,
     DatabaseOptions, FunctionInfo, JobErrorInfo, JobHistoryInfo, JobInfo, MaterializedViewInfo,
-    MvRefreshPlan, OpenTableRequest, ReadConsistency, RefreshMaterializedViewRequest,
-    TableLineageRequest, TableNamesRequest,
+    MvRefreshPlan, OpenTableRequest, PlatformJobDescription, ReadConsistency,
+    RefreshMaterializedViewRequest, TableLineageRequest, TableNamesRequest,
 };
 use crate::embeddings::{EmbeddingRegistry, MemoryRegistry};
 use crate::error::{Error, Result};
@@ -573,6 +573,33 @@ impl Connection {
         self.internal.cancel_job(job_id).await
     }
 
+    /// Describe a platform job (`POST /v1/jobs/describe`): registry-backed
+    /// lifecycle state plus the owner-written status payload. `None` when the
+    /// registry has no such job.
+    pub async fn describe_platform_job(
+        &self,
+        platform_job_id: &str,
+    ) -> Result<Option<PlatformJobDescription>> {
+        self.internal.describe_platform_job(platform_job_id).await
+    }
+
+    /// Resolve a submission (manifest) job id to its platform job id. `None`
+    /// until the job has registered (dispatch is async).
+    pub async fn resolve_platform_job_id(
+        &self,
+        manifest_job_id: &str,
+        table_hint: Option<&str>,
+    ) -> Result<Option<String>> {
+        self.internal
+            .resolve_platform_job_id(manifest_job_id, table_hint)
+            .await
+    }
+
+    /// Cancel a platform job. Idempotent on already-terminal jobs.
+    pub async fn cancel_platform_job(&self, platform_job_id: &str) -> Result<()> {
+        self.internal.cancel_platform_job(platform_job_id).await
+    }
+
     /// Look up a single server-side job by id -- the `wait()`/status poll path.
     /// `table_hint` (the job's table) enables an O(1) server-side lookup; `None`
     /// scans the database's active jobs. A `None` result means unknown / not

rust/lancedb/src/database.rs

@@ -321,6 +321,23 @@ pub struct MaterializedViewInfo {
     pub auto_refresh: bool,
 }
 
+/// A described platform job (`POST /v1/jobs/describe`): the job registry's
+/// lifecycle state plus the owner-written status payload.
+#[derive(Debug, Clone)]
+pub struct PlatformJobDescription {
+    /// The platform (registry) job id -- what describe/cancel accept.
+    pub job_id: String,
+    pub job_type: String,
+    pub job_subtype: String,
+    /// "IN_PROGRESS" | "CANCELLED" | "FAILED" | "DONE".
+    pub job_state: String,
+    pub creation_ms: i64,
+    /// The owner-written status payload -- `units_done` / `units_total` /
+    /// `rows_committed` / `error` when present. Records whose owner has not
+    /// written a payload yet carry the raw status-store URI string instead.
+    pub status: serde_json::Value,
+}
+
 /// A row from `list_jobs`: one inflight server-side job (index build,
 /// compaction, column refresh, view refresh, ...).
 #[derive(Debug, Clone)]
@@ -510,6 +527,33 @@ pub trait Database:
     async fn list_jobs(&self) -> Result<Vec<JobInfo>> {
         not_supported("list_jobs")
     }
+
+    /// Describe a platform job (`POST /v1/jobs/describe`): registry-backed
+    /// lifecycle state plus the owner-written status payload. `None` when the
+    /// registry has no such job.
+    async fn describe_platform_job(
+        &self,
+        _platform_job_id: &str,
+    ) -> Result<Option<PlatformJobDescription>> {
+        not_supported("describe_platform_job")
+    }
+
+    /// Resolve a submission (manifest) job id to its platform job id via the
+    /// registry's manifest-id filter (`POST /v1/jobs/list`). `None` until the
+    /// job has registered (dispatch is async).
+    async fn resolve_platform_job_id(
+        &self,
+        _manifest_job_id: &str,
+        _table_hint: Option<&str>,
+    ) -> Result<Option<String>> {
+        not_supported("resolve_platform_job_id")
+    }
+
+    /// Cancel a platform job (`POST /v1/jobs/cancel`). Idempotent: cancelling
+    /// an already-terminal job is a no-op success.
+    async fn cancel_platform_job(&self, _platform_job_id: &str) -> Result<()> {
+        not_supported("cancel_platform_job")
+    }
     /// Cancel an inflight server-side job by id. Returns true if a
     /// matching inflight job was found and flagged for cancellation,
     /// false if none was inflight (best-effort, like SQL `CANCEL JOB`).

rust/lancedb/src/remote/db.rs

@@ -158,6 +158,30 @@ struct RemoteJobEntry {
     error: Option<String>,
 }
 
+#[derive(Deserialize)]
+struct RemoteDescribePlatformJobResponse {
+    job_id: String,
+    job_type: String,
+    #[serde(default)]
+    job_subtype: String,
+    job_state: String,
+    #[serde(default)]
+    creation_ms: i64,
+    #[serde(default)]
+    status: serde_json::Value,
+}
+
+#[derive(Deserialize)]
+struct RemoteListPlatformJobsResponse {
+    #[serde(default)]
+    jobs: Vec<RemotePlatformJobRow>,
+}
+
+#[derive(Deserialize)]
+struct RemotePlatformJobRow {
+    job_id: String,
+}
+
 #[derive(serde::Deserialize)]
 struct RemoteListJobsResponse {
     jobs: Vec<RemoteJobEntry>,
@@ -1074,6 +1098,56 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         Ok(body.job.map(JobInfo::from))
     }
 
+    async fn describe_platform_job(
+        &self,
+        platform_job_id: &str,
+    ) -> Result<Option<PlatformJobDescription>> {
+        let req = self
+            .client
+            .post("/v1/jobs/describe")
+            .json(&serde_json::json!({ "job_id": platform_job_id }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        if rsp.status().as_u16() == 404 {
+            return Ok(None);
+        }
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteDescribePlatformJobResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(Some(PlatformJobDescription {
+            job_id: body.job_id,
+            job_type: body.job_type,
+            job_subtype: body.job_subtype,
+            job_state: body.job_state,
+            creation_ms: body.creation_ms,
+            status: body.status,
+        }))
+    }
+
+    async fn resolve_platform_job_id(
+        &self,
+        manifest_job_id: &str,
+        table_hint: Option<&str>,
+    ) -> Result<Option<String>> {
+        let req = self.client.post("/v1/jobs/list").json(&serde_json::json!({
+            "manifest_job_id": manifest_job_id,
+            "table_name": table_hint,
+            "job_type": "indexer",
+        }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteListPlatformJobsResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(body.jobs.into_iter().next().map(|row| row.job_id))
+    }
+
+    async fn cancel_platform_job(&self, platform_job_id: &str) -> Result<()> {
+        let req = self
+            .client
+            .post("/v1/jobs/cancel")
+            .json(&serde_json::json!({ "job_id": platform_job_id }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        self.client.check_response(&request_id, rsp).await?;
+        Ok(())
+    }
+
     async fn cancel_job(&self, job_id: &str) -> Result<bool> {
         let req = self.client.post(&format!("/v1/job/{}/cancel", job_id));
         let (request_id, rsp) = self.client.send(req).await?;