Expose Datagen history reads and retain folded codec metadata

crates/lance-context-api/src/lib.rs

@@ -347,8 +347,16 @@ pub trait DatagenStoreApi {
         item_id: &str,
     ) -> impl Future<Output = ContextResult<Vec<DatagenFailureDto>>> + Send;
 
-    /// Raw-dump every event for a root and its projected descendants, oldest
-    /// first. The transport-thin read the inspection tree is folded from
+    /// Read one item's raw history in `(item_seq, event_id)` order, without blob bytes.
+    /// Missing items return an empty list; use `get_blob(event_id)` for payloads.
+    fn events_for_item(
+        &self,
+        item_id: &str,
+    ) -> impl Future<Output = ContextResult<Vec<DatagenEventDto>>> + Send;
+
+    /// Raw-dump every event for a root and its projected descendants, grouped by
+    /// item and ordered by `(item_seq, event_id)` within each item, without blob bytes.
+    /// The transport-thin read the inspection tree is folded from
     /// client-side, so the same `DatagenItemTree` assembly runs for embedded and
     /// remote without duplicating fold logic on the server.
     fn events_for_root(
@@ -1149,6 +1157,12 @@ pub struct AddDatagenEventsResponse {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DatagenFieldStateDto {
     pub mode: String,
+    /// Absent only when reading a response from an older server. This is unknown
+    /// metadata, not a default codec; callers must not infer it from a declaration.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub field_type: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub codec_version: Option<i32>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub value: Option<DatagenValueDto>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
@@ -1544,6 +1558,18 @@ pub struct TaskListResponse {
 mod tests {
     use super::*;
 
+    #[test]
+    fn datagen_legacy_folded_field_has_unknown_codec_not_a_default() {
+        let field: DatagenFieldStateDto = serde_json::from_value(serde_json::json!({
+            "mode": "set",
+            "value": {"kind": "str", "value": "2021-02-03"},
+        }))
+        .unwrap();
+        assert_eq!(field.field_type, None);
+        assert_eq!(field.codec_version, None);
+        assert!(field.value.is_some());
+    }
+
     #[test]
     fn search_request_legacy_payload_defaults_filters_and_lifecycle() {
         // Clients written against the pre-#89 shape send only query/limit.

crates/lance-context-client/src/lib.rs

@@ -587,6 +587,15 @@ impl DatagenStoreApi for RemoteDatagenStore {
         Ok(resp.failures)
     }
 
+    async fn events_for_item(&self, item_id: &str) -> ContextResult<Vec<DatagenEventDto>> {
+        let resp = self
+            .client
+            .datagen_events_for_item(&self.store_name, item_id)
+            .await
+            .map_err(to_ctx_err)?;
+        Ok(resp.events)
+    }
+
     async fn events_for_root(&self, root_item_id: &str) -> ContextResult<Vec<DatagenEventDto>> {
         let resp = self
             .client
@@ -1365,19 +1374,41 @@ impl ContextClient {
         Self::handle_response(resp).await
     }
 
-    /// Fetch every raw event whose root item is `root_item_id`. The client
+    /// Fetch one item's raw history in sequence order, without blob bytes.
+    pub async fn datagen_events_for_item(
+        &self,
+        name: &str,
+        item_id: &str,
+    ) -> Result<ListDatagenEventsResponse, ClientError> {
+        self.datagen_events(name, "items", item_id).await
+    }
+
+    /// Fetch every raw event whose root item is `root_item_id`, grouped by item
+    /// and ordered by sequence within each item, without blob bytes. The client
     /// folds these into a tree via `DatagenItemTree::build`; the server does no
     /// fold/tree work.
     pub async fn datagen_events_for_root(
         &self,
         name: &str,
         root_item_id: &str,
     ) -> Result<ListDatagenEventsResponse, ClientError> {
-        let resp = self
-            .http
-            .get(self.url(&format!("/datagen/{}/roots/{}/events", name, root_item_id)))
-            .send()
-            .await?;
+        self.datagen_events(name, "roots", root_item_id).await
+    }
+
+    async fn datagen_events(
+        &self,
+        name: &str,
+        scope: &str,
+        id: &str,
+    ) -> Result<ListDatagenEventsResponse, ClientError> {
+        let mut request = self.http.get(self.url("/datagen")).build()?;
+        // IDs can contain fan-out slashes and URL delimiters; each is one path segment.
+        request
+            .url_mut()
+            .path_segments_mut()
+            .expect("HTTP URL has path segments")
+            .extend([name, scope, id, "events"]);
+        let resp = self.http.execute(request).await?;
         Self::handle_response(resp).await
     }
 

crates/lance-context-core/src/api_impl.rs

@@ -815,6 +815,13 @@ impl DatagenStoreApi for DatagenStore {
         Ok(failures.iter().map(failure_to_dto).collect())
     }
 
+    async fn events_for_item(&self, item_id: &str) -> ContextResult<Vec<DatagenEventDto>> {
+        let events = DatagenStore::events_for_item(self, item_id)
+            .await
+            .map_err(to_ctx_err)?;
+        Ok(events.iter().map(datagen_event_to_dto).collect())
+    }
+
     async fn events_for_root(&self, root_item_id: &str) -> ContextResult<Vec<DatagenEventDto>> {
         let events = DatagenStore::events_for_root(self, root_item_id)
             .await
@@ -1007,13 +1014,25 @@ pub fn folded_item_to_dto(item: &FoldedDatagenItem) -> FoldedDatagenItemDto {
 
 fn field_state_to_dto(state: &DatagenFieldState) -> DatagenFieldStateDto {
     match state {
-        DatagenFieldState::Set(value) => DatagenFieldStateDto {
+        DatagenFieldState::Set {
+            value,
+            field_type,
+            codec_version,
+        } => DatagenFieldStateDto {
             mode: "set".to_string(),
+            field_type: Some(field_type.clone()),
+            codec_version: Some(*codec_version),
             value: Some(datagen_value_to_dto(value)),
             values: Vec::new(),
         },
-        DatagenFieldState::Appended(values) => DatagenFieldStateDto {
+        DatagenFieldState::Appended {
+            values,
+            field_type,
+            codec_version,
+        } => DatagenFieldStateDto {
             mode: "append".to_string(),
+            field_type: Some(field_type.clone()),
+            codec_version: Some(*codec_version),
             value: None,
             values: values.iter().map(datagen_value_to_dto).collect(),
         },