diff --git a/crates/lance-context-api/src/lib.rs b/crates/lance-context-api/src/lib.rs
index 368f029..4fd716f 100644
--- a/crates/lance-context-api/src/lib.rs
+++ b/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.
diff --git a/crates/lance-context-client/src/lib.rs b/crates/lance-context-client/src/lib.rs
index 36db341..d77a508 100644
--- a/crates/lance-context-client/src/lib.rs
+++ b/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,7 +1374,17 @@ 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(
@@ -1373,11 +1392,23 @@ impl ContextClient {
         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
     }
 
diff --git a/crates/lance-context-core/src/api_impl.rs b/crates/lance-context-core/src/api_impl.rs
index c570383..e9d0d19 100644
--- a/crates/lance-context-core/src/api_impl.rs
+++ b/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(),
         },
diff --git a/crates/lance-context-core/src/datagen.rs b/crates/lance-context-core/src/datagen.rs
index 9211d49..841cb88 100644
--- a/crates/lance-context-core/src/datagen.rs
+++ b/crates/lance-context-core/src/datagen.rs
@@ -313,10 +313,19 @@ pub struct DatagenStreamPosition {
 }
 
 /// How a field folds: FIELD_SET replaces (last-writer-wins); FIELD_APPEND accumulates in order.
+/// Codec metadata describes the retained value, or every element of an appended list.
 #[derive(Debug, Clone, PartialEq)]
 pub enum DatagenFieldState {
-    Set(DatagenValue),
-    Appended(Vec<DatagenValue>),
+    Set {
+        value: DatagenValue,
+        field_type: String,
+        codec_version: i32,
+    },
+    Appended {
+        values: Vec<DatagenValue>,
+        field_type: String,
+        codec_version: i32,
+    },
 }
 
 /// A pointer to one completed step position (a single STEP_COMPLETED).
@@ -1101,8 +1110,8 @@ fn drop_blob_bytes(item: &mut FoldedDatagenItem) {
     }
     for state in item.fields.values_mut() {
         match state {
-            DatagenFieldState::Set(value) => strip(value),
-            DatagenFieldState::Appended(values) => values.iter_mut().for_each(strip),
+            DatagenFieldState::Set { value, .. } => strip(value),
+            DatagenFieldState::Appended { values, .. } => values.iter_mut().for_each(strip),
         }
     }
 }
@@ -1218,7 +1227,10 @@ fn apply_event(item: &mut FoldedDatagenItem, event: &DatagenEvent) -> Result<(),
             let value = event.value.clone().unwrap();
             // Group D: a field's value kind is fixed by its first write. A later SET that
             // drifts to another kind means the pipeline wrote the field inconsistently.
-            if let Some(DatagenFieldState::Set(existing)) = item.fields.get(&field_name) {
+            if let Some(DatagenFieldState::Set {
+                value: existing, ..
+            }) = item.fields.get(&field_name)
+            {
                 if existing.kind() != value.kind() {
                     return Err(format!(
                         "field '{}' changes value kind from {} to {}",
@@ -1229,19 +1241,35 @@ fn apply_event(item: &mut FoldedDatagenItem, event: &DatagenEvent) -> Result<(),
                 }
             }
             record_blob_event_id(item, &field_name, &value, &event.event_id);
-            item.fields
-                .insert(field_name, DatagenFieldState::Set(value));
+            item.fields.insert(
+                field_name,
+                DatagenFieldState::Set {
+                    value,
+                    field_type: event.field_type.clone().unwrap(),
+                    codec_version: event.codec_version.unwrap(),
+                },
+            );
         }
         DatagenEventType::FieldAppend => {
             let field_name = event.field_name.clone().unwrap();
             let value = event.value.clone().unwrap();
+            let field_type = event.field_type.as_ref().unwrap();
+            let codec_version = event.codec_version.unwrap();
             record_blob_event_id(item, &field_name, &value, &event.event_id);
             match item.fields.entry(field_name.clone()) {
                 std::collections::btree_map::Entry::Vacant(entry) => {
-                    entry.insert(DatagenFieldState::Appended(vec![value]));
+                    entry.insert(DatagenFieldState::Appended {
+                        values: vec![value],
+                        field_type: field_type.clone(),
+                        codec_version,
+                    });
                 }
                 std::collections::btree_map::Entry::Occupied(mut entry) => match entry.get_mut() {
-                    DatagenFieldState::Appended(values) => {
+                    DatagenFieldState::Appended {
+                        values,
+                        field_type: existing_type,
+                        codec_version: existing_version,
+                    } => {
                         // Group D: the same kind rule holds within one appended list.
                         if let Some(existing) = values.first() {
                             if existing.kind() != value.kind() {
@@ -1253,9 +1281,19 @@ fn apply_event(item: &mut FoldedDatagenItem, event: &DatagenEvent) -> Result<(),
                                 ));
                             }
                         }
+                        if existing_type != field_type || *existing_version != codec_version {
+                            return Err(format!(
+                                "field '{}' changes append codec from {}@{} to {}@{}",
+                                field_name,
+                                existing_type,
+                                existing_version,
+                                field_type,
+                                codec_version
+                            ));
+                        }
                         values.push(value);
                     }
-                    DatagenFieldState::Set(_) => {
+                    DatagenFieldState::Set { .. } => {
                         return Err(format!(
                             "field '{}' mixes FIELD_SET and FIELD_APPEND",
                             entry.key()
@@ -1467,14 +1505,22 @@ mod tests {
             .unwrap();
         assert_eq!(
             folded.fields.get("draft"),
-            Some(&DatagenFieldState::Set(DatagenValue::Str("v2".to_string())))
+            Some(&DatagenFieldState::Set {
+                value: DatagenValue::Str("v2".to_string()),
+                field_type: "str".to_string(),
+                codec_version: 1,
+            })
         );
         assert_eq!(
             folded.fields.get("revisions"),
-            Some(&DatagenFieldState::Appended(vec![
-                DatagenValue::Json(json!({"n": "a"})),
-                DatagenValue::Json(json!({"n": "b"})),
-            ]))
+            Some(&DatagenFieldState::Appended {
+                values: vec![
+                    DatagenValue::Json(json!({"n": "a"})),
+                    DatagenValue::Json(json!({"n": "b"})),
+                ],
+                field_type: "json".to_string(),
+                codec_version: 1,
+            })
         );
     }
 
@@ -1516,6 +1562,65 @@ mod tests {
         assert!(err.contains("changes value kind"), "{err}");
     }
 
+    #[test]
+    fn set_metadata_tracks_the_winning_value_including_across_attempts() {
+        let mut first = leaf_completed(1, "gen", 0, None);
+        first.event_type = DatagenEventType::FieldSet;
+        first.field_name = Some("date".into());
+        first.field_type = Some("date".into());
+        first.codec_version = Some(1);
+        first.value = Some(DatagenValue::Str("2021-02-03".into()));
+        let mut last = first.clone();
+        last.item_seq = 2;
+        last.checkpoint_id = "next-attempt".into();
+        last.event_id = datagen_event_id("5", "next-attempt", 0);
+        last.attempt = 1;
+        last.field_type = Some("path".into());
+        last.codec_version = Some(2);
+
+        let prefix = fold_datagen_events(&[created(0), first.clone()])
+            .unwrap()
+            .unwrap();
+        let folded = fold_datagen_events(&[last, first, created(0)])
+            .unwrap()
+            .unwrap();
+        for (item, expected_type, expected_version) in [(&prefix, "date", 1), (&folded, "path", 2)]
+        {
+            let dto = crate::api_impl::folded_item_to_dto(item);
+            let field = &dto.fields["date"];
+            assert_eq!(field.field_type.as_deref(), Some(expected_type));
+            assert_eq!(field.codec_version, Some(expected_version));
+            assert_eq!(field.mode, "set");
+        }
+    }
+
+    #[test]
+    fn append_rejects_field_type_or_codec_version_drift_with_the_same_value_kind() {
+        let mut first = leaf_completed(1, "gen", 0, None);
+        first.event_type = DatagenEventType::FieldAppend;
+        first.field_name = Some("dates".into());
+        first.field_type = Some("date".into());
+        first.codec_version = Some(1);
+        first.value = Some(DatagenValue::Str("2021-02-03".into()));
+
+        for (field_type, codec_version) in [("path", 1), ("date", 2)] {
+            let mut next = first.clone();
+            next.item_seq = 2;
+            next.checkpoint_id = "next-attempt".into();
+            next.event_id = datagen_event_id("5", "next-attempt", 0);
+            next.attempt = 1;
+            next.field_type = Some(field_type.into());
+            next.codec_version = Some(codec_version);
+            let err = fold_datagen_events(&[created(0), first.clone(), next]).unwrap_err();
+            assert!(
+                err.contains(&format!(
+                    "field 'dates' changes append codec from date@1 to {field_type}@{codec_version}"
+                )),
+                "{err}"
+            );
+        }
+    }
+
     #[test]
     fn terminal_sets_lifecycle_status() {
         let mut terminal = event(2, DatagenEventType::Terminal);
@@ -1581,7 +1686,11 @@ mod tests {
         assert_eq!(folded.last_item_seq, 4);
         assert_eq!(
             folded.fields.get("draft"),
-            Some(&DatagenFieldState::Set(DatagenValue::Str("v2".to_string())))
+            Some(&DatagenFieldState::Set {
+                value: DatagenValue::Str("v2".to_string()),
+                field_type: "str".to_string(),
+                codec_version: 1,
+            })
         );
 
         // The failure lens still surfaces the attempt-0 failure, tagged with its attempt.
@@ -1652,7 +1761,11 @@ mod tests {
         assert_eq!(folded.status, DatagenItemStatus::Completed);
         assert_eq!(
             folded.fields.get("draft"),
-            Some(&DatagenFieldState::Set(DatagenValue::Str("v1".into())))
+            Some(&DatagenFieldState::Set {
+                value: DatagenValue::Str("v1".into()),
+                field_type: "str".into(),
+                codec_version: 1,
+            })
         );
         assert_eq!(folded.query_tags, Some(json!({"lang": "en"})));
     }
@@ -1704,7 +1817,11 @@ mod tests {
         assert_eq!(folded.last_attempt, 1);
         assert_eq!(
             folded.fields.get("draft"),
-            Some(&DatagenFieldState::Set(DatagenValue::Str("v2".into())))
+            Some(&DatagenFieldState::Set {
+                value: DatagenValue::Str("v2".into()),
+                field_type: "str".into(),
+                codec_version: 1,
+            })
         );
     }
 
@@ -1777,9 +1894,11 @@ mod tests {
             .unwrap();
         assert_eq!(
             folded.fields.get("messages"),
-            Some(&DatagenFieldState::Appended(vec![DatagenValue::Json(
-                json!({"role": "assistant"})
-            )]))
+            Some(&DatagenFieldState::Appended {
+                values: vec![DatagenValue::Json(json!({"role": "assistant"}))],
+                field_type: "json".into(),
+                codec_version: 1,
+            })
         );
     }
 
@@ -1925,25 +2044,35 @@ mod tests {
         let lazy = fold_datagen_events_with(&events, DatagenBlobProjection::Lazy)
             .unwrap()
             .unwrap();
-        let DatagenFieldState::Set(DatagenValue::Blob(lazy_blob)) =
-            lazy.fields.get("image").unwrap()
+        let DatagenFieldState::Set {
+            value: DatagenValue::Blob(lazy_blob),
+            field_type,
+            codec_version,
+        } = lazy.fields.get("image").unwrap()
         else {
             panic!("expected a blob field");
         };
         assert_eq!(lazy_blob.bytes, None);
         assert_eq!(lazy_blob.size, 9);
+        assert_eq!(field_type, "blob");
+        assert_eq!(*codec_version, 1);
         // The default fold is the lazy one.
         assert_eq!(fold_datagen_events(&events).unwrap().unwrap(), lazy);
 
         let eager = fold_datagen_events_with(&events, DatagenBlobProjection::Eager)
             .unwrap()
             .unwrap();
-        let DatagenFieldState::Set(DatagenValue::Blob(eager_blob)) =
-            eager.fields.get("image").unwrap()
+        let DatagenFieldState::Set {
+            value: DatagenValue::Blob(eager_blob),
+            field_type,
+            codec_version,
+        } = eager.fields.get("image").unwrap()
         else {
             panic!("expected a blob field");
         };
         assert_eq!(eager_blob.bytes.as_deref(), Some(&b"png-bytes"[..]));
+        assert_eq!(field_type, "blob");
+        assert_eq!(*codec_version, 1);
     }
 
     #[test]
diff --git a/crates/lance-context-core/src/datagen_store.rs b/crates/lance-context-core/src/datagen_store.rs
index cf78c7a..718fa71 100644
--- a/crates/lance-context-core/src/datagen_store.rs
+++ b/crates/lance-context-core/src/datagen_store.rs
@@ -1166,6 +1166,16 @@ mod tests {
 
             let events = store.events_for_item("item-1").await.unwrap();
             assert_eq!(events.len(), 6);
+            assert!(store.pending_wal_generations().await.unwrap() > 0);
+            assert_eq!(
+                Dataset::open(&uri)
+                    .await
+                    .unwrap()
+                    .count_rows(None)
+                    .await
+                    .unwrap(),
+                0
+            );
             let blob_event = events
                 .iter()
                 .find(|event| event.field_name.as_deref() == Some("screenshot"))
@@ -1185,7 +1195,11 @@ mod tests {
             assert_eq!(folded.status, DatagenItemStatus::Completed);
             assert_eq!(
                 folded.fields.get("score"),
-                Some(&DatagenFieldState::Set(DatagenValue::Int(i64::MAX)))
+                Some(&DatagenFieldState::Set {
+                    value: DatagenValue::Int(i64::MAX),
+                    field_type: "int".into(),
+                    codec_version: 1,
+                })
             );
             assert_eq!(folded.trajectory.ordered.len(), 1);
             assert_eq!(folded.query_tags, Some(json!({"domain": "math"})));
@@ -1243,8 +1257,10 @@ mod tests {
 
             // Lazy (the default) leaves the bytes behind; eager projects them into the fold.
             let lazy = store.fold_item("item-1").await.unwrap();
-            let DatagenFieldState::Set(DatagenValue::Blob(lazy_blob)) =
-                lazy.folded().unwrap().fields.get("screenshot").unwrap()
+            let DatagenFieldState::Set {
+                value: DatagenValue::Blob(lazy_blob),
+                ..
+            } = lazy.folded().unwrap().fields.get("screenshot").unwrap()
             else {
                 panic!("screenshot should be a blob");
             };
@@ -1254,8 +1270,10 @@ mod tests {
                 .fold_item_with("item-1", DatagenBlobProjection::Eager)
                 .await
                 .unwrap();
-            let DatagenFieldState::Set(DatagenValue::Blob(eager_blob)) =
-                eager.folded().unwrap().fields.get("screenshot").unwrap()
+            let DatagenFieldState::Set {
+                value: DatagenValue::Blob(eager_blob),
+                ..
+            } = eager.folded().unwrap().fields.get("screenshot").unwrap()
             else {
                 panic!("screenshot should be a blob");
             };
@@ -1650,7 +1668,11 @@ mod tests {
             assert_eq!(root_node.item.query_tags, Some(json!({"lang": "en"})));
             assert_eq!(
                 root_node.item.fields.get("draft"),
-                Some(&DatagenFieldState::Set(DatagenValue::Str("v1".into())))
+                Some(&DatagenFieldState::Set {
+                    value: DatagenValue::Str("v1".into()),
+                    field_type: "str".into(),
+                    codec_version: 1,
+                })
             );
 
             let child_node = tree.node(&child_id).unwrap();
diff --git a/crates/lance-context-server/src/routes/datagen.rs b/crates/lance-context-server/src/routes/datagen.rs
index f8bb19e..b1c90dc 100644
--- a/crates/lance-context-server/src/routes/datagen.rs
+++ b/crates/lance-context-server/src/routes/datagen.rs
@@ -207,27 +207,41 @@ async fn datagen_failures_refreshing_on_empty(
         .map_err(AppError::from_context)
 }
 
-async fn datagen_events_for_root_refreshing_on_empty(
+async fn datagen_events_at_latest(
     store_lock: &RwLock<DatagenStore>,
-    root_item_id: &str,
+    id: &str,
+    scope: EventScope,
 ) -> Result<Vec<lance_context_api::DatagenEventDto>, AppError> {
     {
-        let store = store_lock.read().await;
-        let events = DatagenStoreApi::events_for_root(&*store, root_item_id)
-            .await
-            .map_err(AppError::from_context)?;
-        if !events.is_empty() || store.is_version_pinned() {
-            return Ok(events);
+        // A merge can leave a nonempty cached base missing newer events whose WAL
+        // generations have been removed. Refresh even when the old prefix is nonempty.
+        let mut store = store_lock.write().await;
+        if !store.is_version_pinned() {
+            store.refresh_latest().await.map_err(AppError::from_lance)?;
         }
     }
+    let store = store_lock.read().await;
+    scope.read(&store, id).await
+}
 
-    let mut store = store_lock.write().await;
-    if !store.is_version_pinned() {
-        store.refresh_latest().await.map_err(AppError::from_lance)?;
-    }
-    DatagenStoreApi::events_for_root(&*store, root_item_id)
-        .await
+#[derive(Clone, Copy)]
+enum EventScope {
+    Item,
+    Root,
+}
+
+impl EventScope {
+    async fn read(
+        self,
+        store: &DatagenStore,
+        id: &str,
+    ) -> Result<Vec<lance_context_api::DatagenEventDto>, AppError> {
+        match self {
+            Self::Item => DatagenStoreApi::events_for_item(store, id).await,
+            Self::Root => DatagenStoreApi::events_for_root(store, id).await,
+        }
         .map_err(AppError::from_context)
+    }
 }
 
 async fn datagen_root_statuses_refreshing_on_missing(
@@ -310,6 +324,16 @@ pub async fn datagen_item_failures(
     Ok(Json(ListDatagenFailuresResponse { failures }))
 }
 
+/// Read one item's raw history in sequence order, without materializing blobs.
+pub async fn datagen_events_for_item(
+    State(state): State<Arc<AppState>>,
+    Path((name, item_id)): Path<(String, String)>,
+) -> Result<Json<lance_context_api::ListDatagenEventsResponse>, AppError> {
+    let store_lock = state.get_or_open_datagen_store(&name).await?;
+    let events = datagen_events_at_latest(&store_lock, &item_id, EventScope::Item).await?;
+    Ok(Json(ListDatagenEventsResponse { events }))
+}
+
 /// Dump every raw event whose root item is `root_item_id`. The server does no
 /// fold/tree assembly; the client builds the item tree from these events.
 pub async fn datagen_events_for_root(
@@ -317,7 +341,7 @@ pub async fn datagen_events_for_root(
     Path((name, root_item_id)): Path<(String, String)>,
 ) -> Result<Json<lance_context_api::ListDatagenEventsResponse>, AppError> {
     let store_lock = state.get_or_open_datagen_store(&name).await?;
-    let events = datagen_events_for_root_refreshing_on_empty(&store_lock, &root_item_id).await?;
+    let events = datagen_events_at_latest(&store_lock, &root_item_id, EventScope::Root).await?;
     Ok(Json(ListDatagenEventsResponse { events }))
 }
 
@@ -422,6 +446,72 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn raw_event_reads_refresh_after_an_external_merge() {
+        let (state, _dir) = test_state().await;
+        for (scope, name, seed_cached_base) in [
+            (EventScope::Item, "item-history-empty", false),
+            (EventScope::Root, "root-history-empty", false),
+            (EventScope::Item, "item-history-nonempty", true),
+            (EventScope::Root, "root-history-nonempty", true),
+        ] {
+            let _ = create_datagen_store(
+                State(state.clone()),
+                Json(CreateDatagenStoreRequest {
+                    name: name.into(),
+                    storage_options: None,
+                }),
+            )
+            .await
+            .unwrap();
+            let cached = state.get_or_open_datagen_store(name).await.unwrap();
+            let mut external = DatagenStore::open_existing_with_options(
+                &state.datagen_uri(name),
+                DatagenStoreOptions {
+                    shard_id: Some("external".into()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+            if seed_cached_base {
+                external
+                    .append(&[event("root", 0, "seed", 0, DatagenEventType::ItemCreated)])
+                    .await
+                    .unwrap();
+                assert_eq!(external.cleanup_own_shard().await.unwrap(), 1);
+                // Load the seeded base so the next merge leaves a nonempty stale prefix.
+                let seeded = datagen_events_at_latest(&cached, "root", scope)
+                    .await
+                    .unwrap();
+                assert_eq!(seeded.len(), 1);
+            }
+            external
+                .append(&[event(
+                    "root",
+                    i64::from(seed_cached_base),
+                    "created",
+                    0,
+                    DatagenEventType::ItemCreated,
+                )])
+                .await
+                .unwrap();
+            assert_eq!(external.cleanup_own_shard().await.unwrap(), 1);
+            assert_eq!(external.pending_wal_generations().await.unwrap(), 0);
+            assert!(external.version() > cached.read().await.version());
+            let events = datagen_events_at_latest(&cached, "root", scope)
+                .await
+                .unwrap();
+            assert_eq!(
+                events.len(),
+                1 + usize::from(seed_cached_base),
+                "a full-history read must not return a stale prefix"
+            );
+            assert_eq!(events[0].item_id, "root");
+            assert_eq!(cached.read().await.version(), external.version());
+        }
+    }
+
     #[tokio::test]
     async fn point_reads_refresh_a_base_advanced_by_an_external_writer() {
         let (state, _dir) = test_state().await;
diff --git a/crates/lance-context-server/src/routes/mod.rs b/crates/lance-context-server/src/routes/mod.rs
index 32de356..7f5b2ee 100644
--- a/crates/lance-context-server/src/routes/mod.rs
+++ b/crates/lance-context-server/src/routes/mod.rs
@@ -139,6 +139,10 @@ pub fn router() -> Router<Arc<AppState>> {
             "/api/v1/datagen/{name}/items/{item_id}/failures",
             get(datagen::datagen_item_failures),
         )
+        .route(
+            "/api/v1/datagen/{name}/items/{item_id}/events",
+            get(datagen::datagen_events_for_item),
+        )
         .route(
             "/api/v1/datagen/{name}/overview",
             get(datagen::datagen_overview),
diff --git a/crates/lance-context/src/unified_datagen.rs b/crates/lance-context/src/unified_datagen.rs
index daf6a21..bf1615a 100644
--- a/crates/lance-context/src/unified_datagen.rs
+++ b/crates/lance-context/src/unified_datagen.rs
@@ -188,6 +188,10 @@ impl DatagenStoreApi for DatagenStore {
         dispatch_ref!(self, item_failures, item_id)
     }
 
+    async fn events_for_item(&self, item_id: &str) -> ContextResult<Vec<DatagenEventDto>> {
+        dispatch_ref!(self, events_for_item, item_id)
+    }
+
     async fn events_for_root(&self, root_item_id: &str) -> ContextResult<Vec<DatagenEventDto>> {
         dispatch_ref!(self, events_for_root, root_item_id)
     }
diff --git a/docs/src/design/using-datagen-store.md b/docs/src/design/using-datagen-store.md
index d54558c..dff8ab8 100644
--- a/docs/src/design/using-datagen-store.md
+++ b/docs/src/design/using-datagen-store.md
@@ -177,6 +177,29 @@ match store.fold_item("5").await? {
 `Found` item's `status` is always a *lifecycle* value (`running` / `completed`
 / `filtered`) — `failed` never appears here.
 
+Each folded field retains the `field_type` and `codec_version` of its stored
+value, alongside the existing `mode` and `value` / `values`. In Python:
+
+```python
+{
+    "mode": "set",
+    "field_type": "date",
+    "codec_version": 1,
+    "value": {"kind": "str", "value": "2021-02-03"},
+}
+```
+
+`FIELD_SET` retains the last value and that event's codec metadata, preserving
+the existing value-kind check. `FIELD_APPEND` retains one codec pair for the
+entire list: every append must agree on both `field_type` and `codec_version`,
+as well as the value kind. A mismatch is a fold error, not a silent choice of
+the first element's codec. Raw events remain readable for diagnosis.
+
+Before decoding on resume, the application must compare the stored codec pair
+with its expected codec. For example, `date` and `Path` may both encode as
+`kind="str"` but must not silently decode as each other. Returning metadata
+enables this check; the store does not perform application-specific decoding.
+
 ### Resume: the open frame
 
 The trajectory records both started and completed positions. `started \ completed`
@@ -197,6 +220,75 @@ let open: Vec<_> = item.trajectory.started
 let events = store.events_for_root("5").await?;
 ```
 
+### Raw history from Python (embedded or remote)
+
+`DatagenStore.item_events(item_id)` and `root_events(root_item_id)` return
+raw event dicts, including overwritten field values, provenance, attempts and
+codec metadata. These reads scan the base table **and flushed MemWAL**;
+`lance.dataset(uri)` alone can report zero rows while the events are still in
+the WAL. As with existing reads, remote writes become visible after the
+server's flush, not necessarily immediately after append returns.
+Before each remote raw-history scan, the server refreshes an unpinned handle's
+base version. This adds a manifest read but avoids returning an old, nonempty
+prefix after an external writer merges its WAL into the base. The potentially
+large scan runs under a read lock, not the exclusive refresh lock.
+An embedded handle has no such automatic refresh: it scans the base version
+it last loaded, so a long-lived reader can miss events another process has
+already merged. Reopen the store, or use the server, when reading history
+written elsewhere.
+
+```python
+from lance_context import DatagenStore
+
+store = DatagenStore.open("./experiment.lance")
+# Or: store = DatagenStore.connect("http://localhost:3000", "experiment")
+events = store.item_events("5")
+root_events = store.root_events("5")
+
+# Reconstruct state at a completed-step boundary in the application:
+completed = [event for event in events if event["event_type"] == "STEP_COMPLETED"]
+if completed:
+    cutoff = completed[0]["item_seq"]
+    prefix = [event for event in events if event["item_seq"] <= cutoff]
+    # Apply the application's historical fold to this prefix.
+```
+
+Missing items or roots return `[]`. A missing remote *store* is still an error.
+Item events are ordered by `(item_seq, event_id)`; root events are ordered by
+`(item_id, item_seq, event_id)`. Sequence counters are per item, so a root's
+result is **not** a single cross-item timeline. Group by item before folding
+prefixes, and use `parent_item_id` to link nodes in the inspection tree.
+
+Raw reads omit blob bytes. For an overwritten blob, use the historical row's
+`event_id` with `store.get_blob(event_id)`, rather than the latest folded
+field's blob pointer. This avoids eagerly downloading every historical blob.
+
+The HTTP item-history route is
+`GET /api/v1/datagen/{name}/items/{item_id}/events`; the existing root-history
+route is `GET /api/v1/datagen/{name}/roots/{root_item_id}/events`. IDs must be
+encoded as single URL path segments, including any fan-out slashes. The
+official client handles this encoding.
+
+### Compatibility of history and codec reads
+
+- No event-schema change or data rewrite is needed: existing field events
+  already contain the codec columns.
+- Existing Python `mode`, `value` and `values` keys remain. Code that compares
+  whole field dicts must account for the two added metadata keys.
+- Older servers omit codec metadata; the new Python client reports `None` for
+  it. A codec-safe reader must explicitly reject unknown metadata or use a
+  deliberate application migration policy, never infer it from the current
+  declaration. Upgrade the server before using the new item-history route.
+- Previously accepted histories that mix append codecs now fail to fold,
+  including folds used by resume, trees and overviews. This is an intentional
+  correctness change; raw history can still be inspected.
+- Rust source users must update `DatagenFieldState` tuple variants to named
+  fields (`Set { value, field_type, codec_version }` and
+  `Appended { values, field_type, codec_version }`), DTO struct literals, and
+  custom `DatagenStoreApi` implementations for `events_for_item`.
+- These changes are specific to Datagen. Context, Rollout and Generic store
+  contracts and the shared MemWAL implementation are unchanged.
+
 ### Bulk startup classification
 
 ```rust
diff --git a/docs/src/specs/datagen-checkpoint-schema.md b/docs/src/specs/datagen-checkpoint-schema.md
index f045f8f..c531011 100644
--- a/docs/src/specs/datagen-checkpoint-schema.md
+++ b/docs/src/specs/datagen-checkpoint-schema.md
@@ -122,6 +122,11 @@ for the exact `_rowid`.
    (started minus completed) is the driver frame open at crash time.
 7. `FAILED` and `TERMINAL` are ordinary log events, not separate datasets.
 8. Blob bytes are loaded only when the corresponding lazy reference is used.
+9. Folded fields retain `field_type` and `codec_version`. A `FIELD_SET` keeps
+   the winning event's metadata; all values in a `FIELD_APPEND` list must have
+   the same codec pair. Mixed append codecs are rejected during folding,
+   including across attempts. Consumers compare the retained pair with the
+   expected codec before decoding on resume.
 
 ## Maintenance
 
diff --git a/python/python/lance_context/api.py b/python/python/lance_context/api.py
index 744431a..7fcffad 100644
--- a/python/python/lance_context/api.py
+++ b/python/python/lance_context/api.py
@@ -2875,6 +2875,11 @@ def fold_item(
 
         ``load_blobs`` materializes blob-field bytes inline; the default leaves them
         lazy, to be resolved through :meth:`get_blob` / :meth:`load_blob`.
+
+        Each field includes ``field_type`` and ``codec_version`` from the stored
+        events. Compare them with the expected codec before decoding on resume.
+        Older servers return ``None`` for unavailable metadata, not a default codec.
+        Appended fields with mixed codec metadata are rejected during folding.
         """
         return self._sync.fold_item(item_id, load_blobs)
 
@@ -2900,6 +2905,25 @@ def item_failures(self, item_id: str) -> list[dict[str, Any]]:
         """All failure records for an item (the failure lens), oldest first."""
         return self._sync.item_failures(item_id)
 
+    def item_events(self, item_id: str) -> list[dict[str, Any]]:
+        """Every event for one item in ``(item_seq, event_id)`` order.
+
+        Reads the base table and flushed WAL, for embedded and remote stores.
+        Missing items return an empty list. Blob bytes are omitted; resolve a
+        historical blob using that row's ``event_id`` with :meth:`get_blob`.
+        """
+        return self._sync.item_events(item_id)
+
+    def root_events(self, root_item_id: str) -> list[dict[str, Any]]:
+        """Every event for a root and its descendants, grouped by item.
+
+        Ordered by ``(item_id, item_seq, event_id)``, not a cross-item timeline.
+        Reads the base table and flushed WAL; missing roots return an empty list.
+        Blob bytes are omitted; resolve them with :meth:`get_blob(event_id)`.
+        Works for embedded and remote stores.
+        """
+        return self._sync.root_events(root_item_id)
+
     def get_blob(self, event_id: str) -> bytes | None:
         """Materialize one ``FIELD_*`` event's blob bytes by event id, or ``None``."""
         return self._sync.get_blob(event_id)
diff --git a/python/src/lib.rs b/python/src/lib.rs
index 63504d1..8f1b8f7 100644
--- a/python/src/lib.rs
+++ b/python/src/lib.rs
@@ -2795,6 +2795,33 @@ impl DatagenStore {
         Ok(list.into_pyobject(py)?.unbind().into())
     }
 
+    /// One item's raw history in sequence order, without blob bytes.
+    fn item_events(&self, py: Python<'_>, item_id: &str) -> PyResult<PyObject> {
+        let events = py
+            .allow_threads(|| self.runtime.block_on(self.store.events_for_item(item_id)))
+            .map_err(ctx_to_py_err)?;
+        let list = PyList::empty(py);
+        for event in &events {
+            list.append(event_dto_to_py(py, event)?)?;
+        }
+        Ok(list.into_pyobject(py)?.unbind().into())
+    }
+
+    /// A root's raw history, grouped by item and ordered within each item, without blob bytes.
+    fn root_events(&self, py: Python<'_>, root_item_id: &str) -> PyResult<PyObject> {
+        let events = py
+            .allow_threads(|| {
+                self.runtime
+                    .block_on(self.store.events_for_root(root_item_id))
+            })
+            .map_err(ctx_to_py_err)?;
+        let list = PyList::empty(py);
+        for event in &events {
+            list.append(event_dto_to_py(py, event)?)?;
+        }
+        Ok(list.into_pyobject(py)?.unbind().into())
+    }
+
     /// Materialize one FIELD_* event's blob bytes by event id, or `None`.
     fn get_blob(&self, py: Python<'_>, event_id: &str) -> PyResult<Option<Py<PyBytes>>> {
         let bytes = py
@@ -3370,6 +3397,8 @@ fn folded_item_to_py(py: Python<'_>, item: &FoldedDatagenItemDto) -> PyResult<Py
 fn field_state_to_py(py: Python<'_>, state: &DatagenFieldStateDto) -> PyResult<PyObject> {
     let dict = PyDict::new(py);
     dict.set_item("mode", &state.mode)?;
+    dict.set_item("field_type", state.field_type.as_deref())?;
+    dict.set_item("codec_version", state.codec_version)?;
     if let Some(value) = &state.value {
         dict.set_item("value", value_to_py(py, value)?)?;
     }
diff --git a/python/tests/conftest.py b/python/tests/conftest.py
new file mode 100644
index 0000000..99d0550
--- /dev/null
+++ b/python/tests/conftest.py
@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import os
+import socket
+import subprocess
+import tempfile
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import pytest
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+_SERVER_BIN = Path(__file__).resolve().parents[2] / "target/debug/lance-context-server"
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def _wait_for_health(base_url: str, timeout: float = 30.0) -> None:
+    deadline = time.time() + timeout
+    url = f"{base_url}/api/v1/health"
+    while time.time() < deadline:
+        try:
+            with urllib.request.urlopen(url, timeout=1) as resp:  # noqa: S310
+                if resp.status == 200:
+                    return
+        except (urllib.error.URLError, ConnectionError, OSError):
+            time.sleep(0.2)
+    raise RuntimeError(f"server did not become healthy at {url}")
+
+
+@pytest.fixture()
+def server() -> Iterator[str]:
+    """Run a real HTTP server with an isolated dataset directory."""
+    if not _SERVER_BIN.exists():
+        # Locally a missing binary is a fair reason to skip. In CI it is not:
+        # these suites cover the remote/HTTP client path, so a silent skip would
+        # hide a regression. Fail loudly instead.
+        msg = f"server binary not built at {_SERVER_BIN}"
+        if os.environ.get("CI"):
+            pytest.fail(f"{msg} (run `cargo build -p lance-context-server`)")
+        pytest.skip(msg)
+    port = _free_port()
+    with tempfile.TemporaryDirectory() as data_dir:
+        # Rows are durable on `add` but only become visible when the server's
+        # sweeper seals the memtable. The 30s production default would make
+        # every write-then-assert below hang; 1s keeps the tests honest about
+        # the async-visibility contract without waiting on it.
+        env = {**os.environ, "ROLLOUT_FLUSH_INTERVAL_SECS": "1"}
+        with tempfile.TemporaryFile() as log:
+            proc = subprocess.Popen(
+                [
+                    str(_SERVER_BIN),
+                    "--host",
+                    "127.0.0.1",
+                    "--port",
+                    str(port),
+                    "--data-dir",
+                    data_dir,
+                ],
+                env=env,
+                stdout=log,
+                stderr=log,
+            )
+            base_url = f"http://127.0.0.1:{port}"
+            try:
+                _wait_for_health(base_url)
+                yield base_url
+            finally:
+                proc.terminate()
+                try:
+                    proc.wait(timeout=10)
+                except subprocess.TimeoutExpired:
+                    proc.kill()
+                    proc.wait(timeout=10)
+                log.seek(0)
+                print(log.read().decode(errors="replace"))
diff --git a/python/tests/test_datagen.py b/python/tests/test_datagen.py
index 8cd3b6a..26d1cae 100644
--- a/python/tests/test_datagen.py
+++ b/python/tests/test_datagen.py
@@ -1,12 +1,17 @@
 from __future__ import annotations
 
 import sys
+import time
 from pathlib import Path
+from typing import Any
+
+import pytest
 
 PACKAGE_ROOT = Path(__file__).resolve().parents[2] / "python" / "python"
 if str(PACKAGE_ROOT) not in sys.path:
     sys.path.insert(0, str(PACKAGE_ROOT))
 
+from lance_context import InvalidRequestError  # noqa: E402
 from lance_context.api import DatagenStore, DatagenStreamWriter  # noqa: E402
 
 _CONTEXT = {"run_id": "run-1", "writer_epoch": "writer-1"}
@@ -53,6 +58,8 @@ def test_open_stream_writer_appends_and_folds(tmp_path: Path) -> None:
     assert folded["status"] == "completed"
     assert folded["fields"]["draft"] == {
         "mode": "set",
+        "field_type": "str",
+        "codec_version": 1,
         "value": {"kind": "str", "value": "v1"},
     }
     assert folded["query_tags"] == {"lang": "en"}
@@ -79,6 +86,8 @@ def test_resume_stream_bumps_attempt(tmp_path: Path) -> None:
     assert folded["last_attempt"] == 1
     assert folded["fields"]["draft"] == {
         "mode": "set",
+        "field_type": "str",
+        "codec_version": 1,
         "value": {"kind": "str", "value": "v2"},
     }
 
@@ -146,3 +155,174 @@ def test_item_failed_is_failure_lens_not_terminal(tmp_path: Path) -> None:
     folded = store.fold_item("5")
     assert folded is not None
     assert folded["status"] == "running"
+
+
+@pytest.fixture(params=["embedded", "remote"])
+def history_store(request: pytest.FixtureRequest, tmp_path: Path) -> DatagenStore:
+    """Exercise the same public read contract through both backends."""
+    if request.param == "embedded":
+        return DatagenStore.open(str(tmp_path / "log"))
+    return DatagenStore.connect_or_create(
+        request.getfixturevalue("server"), "datagen-history"
+    )
+
+
+def _wait_for_events(
+    store: DatagenStore, item_id: str, count: int
+) -> list[dict[str, Any]]:
+    deadline = time.monotonic() + 15
+    while True:
+        events = store.item_events(item_id)
+        if len(events) == count:
+            return events
+        if time.monotonic() >= deadline:
+            raise AssertionError(f"expected {count} events, got {events!r}")
+        time.sleep(0.1)
+
+
+def test_raw_history_retains_overwritten_values_and_groups_descendants(
+    history_store: DatagenStore,
+) -> None:
+    store = history_store
+    # Exercise fan-out slashes and URL delimiters without changing their identity.
+    root_id = "root ?#%'"
+    child_id = f"{root_id}/expand:0"
+    root = store.open_stream(root_id, **_CONTEXT)
+    first = root.step_completed(_leaf_position("gen", 0), [_set_field("draft", "v1")])
+    store.append_checkpoint(first)
+    child = store.open_stream(child_id, parent_item_id=root_id, **_CONTEXT)
+    child_checkpoint = child.step_completed(
+        _leaf_position("child", 0), [_set_field("draft", "child")]
+    )
+    store.append_checkpoint(child_checkpoint)
+    last = root.step_completed(_leaf_position("edit", 1), [_set_field("draft", "v2")])
+    store.append_checkpoint(last)
+    store.append_checkpoint(
+        last
+    )  # An ambiguous-response retry must not duplicate rows.
+    store.open_stream("unrelated", **_CONTEXT)
+
+    root_events = _wait_for_events(store, root_id, 5)
+    child_events = _wait_for_events(store, child_id, 3)
+    assert [row["event_id"] for row in root_events[1:]] == [
+        row["event_id"] for row in [*first, *last]
+    ]
+    assert [row["value"]["value"] for row in root_events if row["field_name"]] == [
+        "v1",
+        "v2",
+    ]
+    assert [row["event_id"] for row in child_events[1:]] == [
+        row["event_id"] for row in child_checkpoint
+    ]
+    assert all(row["item_id"] == child_id for row in child_events)
+    assert all(row["parent_item_id"] == root_id for row in child_events)
+    assert store.root_events(root_id) == root_events + child_events
+    assert store.item_events("missing") == []
+    assert store.root_events("missing") == []
+    for row in root_events[1:]:
+        if row["field_name"]:
+            assert row["field_type"] == "str"
+            assert row["codec_version"] == 1
+
+
+def test_fold_preserves_winning_codec_and_append_metadata(
+    history_store: DatagenStore,
+) -> None:
+    store = history_store
+    writer = store.open_stream("root", **_CONTEXT)
+    first = {
+        **_set_field("when", "2021-02-03"),
+        "field_type": "date",
+        "codec_version": 3,
+    }
+    appended = {**first, "name": "dates", "op": "append"}
+    store.append_checkpoint(
+        writer.step_completed(_leaf_position("first", 0), [first, appended])
+    )
+    _wait_for_events(store, "root", 4)
+    assert store.fold_item("root")["fields"]["when"]["field_type"] == "date"
+    last = {**first, "field_type": "path", "codec_version": 4, "value": first["value"]}
+    store.append_checkpoint(
+        writer.step_completed(_leaf_position("last", 1), [last, appended])
+    )
+    _wait_for_events(store, "root", 7)
+    folded = store.fold_item("root")
+    assert folded is not None
+    assert folded["fields"]["when"] == {
+        "mode": "set",
+        "field_type": "path",
+        "codec_version": 4,
+        "value": first["value"],
+    }
+    assert folded["fields"]["dates"] == {
+        "mode": "append",
+        "field_type": "date",
+        "codec_version": 3,
+        "values": [first["value"], first["value"]],
+    }
+    tree = store.item_tree("root")
+    assert tree["nodes"]["root"]["item"]["fields"] == folded["fields"]
+    resumed = store.resume_stream("root", run_id="run-2", writer_epoch="writer-2")
+    assert resumed is not None
+    assert resumed.attempt == 1
+
+
+@pytest.mark.parametrize(("next_type", "next_version"), [("path", 1), ("date", 2)])
+def test_fold_rejects_mixed_append_codecs_but_raw_history_remains_readable(
+    history_store: DatagenStore,
+    next_type: str,
+    next_version: int,
+) -> None:
+    store = history_store
+    writer = store.open_stream("root", **_CONTEXT)
+    first = {
+        **_set_field("dates", "2021-02-03"),
+        "op": "append",
+        "field_type": "date",
+    }
+    store.append_checkpoint(writer.step_completed(_leaf_position("first", 0), [first]))
+    store.append_checkpoint(
+        writer.step_completed(
+            _leaf_position("next", 1),
+            [{**first, "field_type": next_type, "codec_version": next_version}],
+        )
+    )
+    events = _wait_for_events(store, "root", 5)
+    assert len(store.root_events("root")) == len(events)
+    with pytest.raises(InvalidRequestError, match="changes append codec"):
+        store.fold_item("root")
+    with pytest.raises(InvalidRequestError, match="changes append codec"):
+        store.resume_stream("root", run_id="run-2", writer_epoch="writer-2")
+
+
+def test_historical_blobs_are_lazy_and_resolved_by_event_id(
+    history_store: DatagenStore,
+) -> None:
+    store = history_store
+    writer = store.open_stream("root", **_CONTEXT)
+    for index, payload in enumerate([b"old-image", b"new-image"]):
+        field = {
+            "name": "image",
+            "op": "set",
+            "field_type": "image",
+            "codec_version": 7,
+            "value": {"kind": "blob", "bytes": payload, "size": len(payload)},
+        }
+        store.append_checkpoint(
+            writer.step_completed(_leaf_position("capture", index), [field])
+        )
+    events = _wait_for_events(store, "root", 5)
+    blobs = [row for row in events if row["field_name"] == "image"]
+    assert len(blobs) == 2
+    assert all(row["value"]["bytes"] is None for row in blobs)
+    assert [store.get_blob(row["event_id"]) for row in blobs] == [
+        b"old-image",
+        b"new-image",
+    ]
+    for eager in [False, True]:
+        folded = store.fold_item("root", load_blobs=eager)
+        assert folded is not None
+        field = folded["fields"]["image"]
+        assert field["field_type"] == "image"
+        assert field["codec_version"] == 7
+        assert field["value"]["bytes"] == (b"new-image" if eager else None)
diff --git a/python/tests/test_rollout_remote.py b/python/tests/test_rollout_remote.py
index 4c1cdf2..333c6c1 100644
--- a/python/tests/test_rollout_remote.py
+++ b/python/tests/test_rollout_remote.py
@@ -10,40 +10,10 @@
 from __future__ import annotations
 
 import asyncio
-import os
-import socket
-import subprocess
-import tempfile
 import time
-import urllib.error
-import urllib.request
-from pathlib import Path
 
-import pytest
 from lance_context import AsyncRolloutStore
 
-_REPO_ROOT = Path(__file__).resolve().parents[2]
-_SERVER_BIN = _REPO_ROOT / "target" / "debug" / "lance-context-server"
-
-
-def _free_port() -> int:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
-
-
-def _wait_for_health(base_url: str, timeout: float = 30.0) -> None:
-    deadline = time.time() + timeout
-    url = f"{base_url}/api/v1/health"
-    while time.time() < deadline:
-        try:
-            with urllib.request.urlopen(url, timeout=1) as resp:  # noqa: S310
-                if resp.status == 200:
-                    return
-        except (urllib.error.URLError, ConnectionError, OSError):
-            time.sleep(0.2)
-    raise RuntimeError(f"server did not become healthy at {url}")
-
 
 async def _eventually(fn, predicate, timeout: float = 15.0):
     """Poll `fn` until `predicate` holds, or fail after `timeout`.
@@ -63,49 +33,6 @@ async def _eventually(fn, predicate, timeout: float = 15.0):
     raise AssertionError(f"condition not met within {timeout}s; last value: {last!r}")
 
 
-@pytest.fixture()
-def server():
-    if not _SERVER_BIN.exists():
-        # Locally a missing binary is a fair reason to skip. In CI it is not:
-        # this file is the only coverage of the remote/HTTP client path, so a
-        # silent skip would hide a regression. Fail loudly instead.
-        msg = f"server binary not built at {_SERVER_BIN}"
-        if os.environ.get("CI"):
-            pytest.fail(f"{msg} (run `cargo build -p lance-context-server`)")
-        pytest.skip(msg)
-    port = _free_port()
-    with tempfile.TemporaryDirectory() as data_dir:
-        # Rows are durable on `add` but only become visible when the server's
-        # sweeper seals the memtable. The 30s production default would make
-        # every write-then-assert below hang; 1s keeps the tests honest about
-        # the async-visibility contract without waiting on it.
-        env = {**os.environ, "ROLLOUT_FLUSH_INTERVAL_SECS": "1"}
-        proc = subprocess.Popen(
-            [
-                str(_SERVER_BIN),
-                "--host",
-                "127.0.0.1",
-                "--port",
-                str(port),
-                "--data-dir",
-                data_dir,
-            ],
-            env=env,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-        )
-        base_url = f"http://127.0.0.1:{port}"
-        try:
-            _wait_for_health(base_url)
-            yield base_url
-        finally:
-            proc.terminate()
-            try:
-                proc.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                proc.kill()
-
-
 def test_remote_roundtrip(server):
     async def run():
         store = await AsyncRolloutStore.connect_or_create(server, "rl-run-1")