From 4e54b4a0a7dafc3a04c473b597c3bc6746fcbd5a Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Sun, 4 Oct 2026 18:26:55 +0000 Subject: [PATCH 1/4] feat: add typed trace records with batch ID lookup --- README.md | 5 + crates/lance-context-api/src/lib.rs | 2 + crates/lance-context-api/src/trace.rs | 251 ++++++++++++++++++ crates/lance-context-core/src/lib.rs | 4 +- .../lance-context-core/tests/trace_store.rs | 181 +++++++++++++ .../src/routes/generic.rs | 54 ++++ crates/lance-context/src/lib.rs | 12 +- docs/trace-records.md | 118 ++++++++ 8 files changed, 620 insertions(+), 7 deletions(-) create mode 100644 crates/lance-context-api/src/trace.rs create mode 100644 crates/lance-context-core/tests/trace_store.rs create mode 100644 docs/trace-records.md diff --git a/README.md b/README.md index ce587165..1918324d 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,11 @@ one). Full documentation: **** +For deduplicated conversation turns, see [Trace records](docs/trace-records.md): +`TraceRecord` supplies `session_id`, ordered `turn_id`, role and content in one +dataset, with typed local/remote access and ID-only batch lookup through the +existing GenericStore API. + ## Install ```bash diff --git a/crates/lance-context-api/src/lib.rs b/crates/lance-context-api/src/lib.rs index 8ca1a8ca..3393e2c2 100644 --- a/crates/lance-context-api/src/lib.rs +++ b/crates/lance-context-api/src/lib.rs @@ -1,5 +1,7 @@ pub mod schema_spec; pub use schema_spec::{ColumnSpec, ColumnType, SchemaSpec, ID_COLUMN}; +mod trace; +pub use trace::{trace_schema, TraceRecord, TraceStore}; use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; use chrono::{DateTime, Utc}; diff --git a/crates/lance-context-api/src/trace.rs b/crates/lance-context-api/src/trace.rs new file mode 100644 index 00000000..c521010d --- /dev/null +++ b/crates/lance-context-api/src/trace.rs @@ -0,0 +1,251 @@ +//! A typed conversation-turn schema over the shared generic storage API. + +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; + +use crate::{ + AddRowsResponse, ColumnSpec, ColumnType, ContextError, ContextResult, GenericStoreApi, + SchemaSpec, +}; + +/// One observed turn, not a complete model call or a copy of its input history. +/// +/// `turn_id` is a zero-based position in the observed session, independent of +/// arrival order. It is not the storage key: different content at the same +/// `(session_id, turn_id)` must use different `id`s. Replays must reuse their ID. +/// A caller can derive `id` from session, turn, role, content type and exact +/// content (including null versus empty); the store does not generate it. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TraceRecord { + pub id: String, + pub session_id: String, + pub turn_id: i64, + /// Open vocabulary: user, assistant, system, tool, or a source-specific role. + pub role: String, + /// Exact turn content. Null and an empty string remain distinct. + pub content: Option, + #[serde(default = "text_plain")] + pub content_type: String, + /// Source dataset/system, not an individual occurrence of this shared turn. + #[serde(default)] + pub source: Option, + /// Optional source-specific annotations, physically encoded as JSON text. + /// This does not implicitly accumulate provenance when an ID is replayed. + #[serde(default)] + pub metadata: Option, +} + +fn text_plain() -> String { + "text/plain".into() +} + +impl TraceRecord { + pub fn validate(&self) -> ContextResult<()> { + for (name, value) in [ + ("id", &self.id), + ("session_id", &self.session_id), + ("role", &self.role), + ("content_type", &self.content_type), + ] { + if value.trim().is_empty() { + return Err(ContextError::InvalidRequest(format!( + "trace {name} must not be empty" + ))); + } + } + if self.turn_id < 0 { + return Err(ContextError::InvalidRequest( + "trace turn_id must be nonnegative".into(), + )); + } + Ok(()) + } + + /// Encode for the existing generic batch/HTTP API, validating before I/O. + pub fn to_row(&self) -> ContextResult> { + self.validate()?; + let Value::Object(mut row) = serde_json::to_value(self).map_err(codec_error)? else { + unreachable!("TraceRecord serializes as an object") + }; + row.insert( + "metadata".into(), + match &self.metadata { + Some(value) => Value::String(serde_json::to_string(value).map_err(codec_error)?), + None => Value::Null, + }, + ); + Ok(row) + } + + /// Decode a complete generic row. Projected rows are not complete records. + pub fn from_row(mut row: Map) -> ContextResult { + // Decode separately to preserve Some(JSON null) versus absent metadata. + let metadata = match row.remove("metadata") { + None | Some(Value::Null) => None, + Some(Value::String(text)) => Some(serde_json::from_str(&text).map_err(codec_error)?), + Some(_) => { + return Err(ContextError::InvalidRequest( + "trace metadata column must contain JSON text".into(), + )) + } + }; + let mut record: Self = serde_json::from_value(Value::Object(row)).map_err(codec_error)?; + record.metadata = metadata; + record.validate()?; + Ok(record) + } +} + +fn codec_error(error: serde_json::Error) -> ContextError { + ContextError::InvalidRequest(format!("invalid trace record: {error}")) +} + +/// Built-in trace schema, persisted by GenericStore like every other SchemaSpec. +/// `id` is the indexed storage key; `(session_id, turn_id)` is deliberately not +/// unique so branches and conflicting observations can coexist. +pub fn trace_schema() -> SchemaSpec { + let string = || ColumnType::String { large: false }; + SchemaSpec::new(vec![ + ("id".into(), ColumnSpec::required(string())), + ("session_id".into(), ColumnSpec::required(string())), + ("turn_id".into(), ColumnSpec::required(ColumnType::Int64)), + ("role".into(), ColumnSpec::required(string())), + ( + "content".into(), + ColumnSpec::new(ColumnType::String { large: true }), + ), + ("content_type".into(), ColumnSpec::required(string())), + ("source".into(), ColumnSpec::new(string())), + ( + "metadata".into(), + ColumnSpec::new(ColumnType::String { large: true }), + ), + ]) +} + +/// Typed trace access over an embedded or remote GenericStore, with the same +/// WAL visibility, indexed batch lookup, and newest-ID resolution. +/// +/// This is one dataset, not an additional store or index. `existing_ids` is a +/// read, not an atomic insert-if-absent operation: the caller must coordinate +/// competing writers and flush before relying on a previous append's visibility. +pub struct TraceStore { + store: S, +} + +impl TraceStore { + /// Reject a mismatched schema before writing to an existing store. + pub fn new(store: S) -> ContextResult { + if store.spec() != &trace_schema() { + return Err(ContextError::InvalidRequest( + "store does not have the trace schema".into(), + )); + } + Ok(Self { store }) + } + + /// Access projections, filtering, and the backend's maintenance interfaces. + pub fn as_generic(&self) -> &S { + &self.store + } + + /// Recover the owned backend for shutdown or exclusive index/WAL maintenance. + pub fn into_generic(self) -> S { + self.store + } + + /// Validate the entire batch before appending once. Repeated IDs have the + /// generic store's newest-write-wins semantics, not immutable conflict checks. + /// Distinct content variants must have distinct IDs. Visibility on return + /// requires seal-on-add, or a subsequent `flush`. + pub async fn add(&self, records: &[TraceRecord]) -> ContextResult { + let rows = records + .iter() + .map(TraceRecord::to_row) + .collect::>>()?; + self.store.add(&rows).await + } + + /// Fetch up to MAX_BATCH_GET_IDS records, including pending flushed WAL. + /// Missing IDs are omitted; duplicates appear once in first-requested order. + pub async fn get_many(&self, ids: &[String]) -> ContextResult> { + self.store + .get_many(ids, None) + .await? + .into_iter() + .map(TraceRecord::from_row) + .collect() + } + + /// Batch deduplication probe using an ID-only projection, without reading + /// content or metadata. Shares get_many's bounds, order and WAL semantics. + pub async fn existing_ids(&self, ids: &[String]) -> ContextResult> { + self.store + .get_many(ids, Some(&["id".into()])) + .await? + .into_iter() + .map(|mut row| match row.remove("id") { + Some(Value::String(id)) => Ok(id), + _ => Err(ContextError::Internal("trace lookup returned no ID".into())), + }) + .collect() + } + + /// Publish the active WAL generation for subsequent deduplication probes. + pub async fn flush(&self) -> ContextResult<()> { + self.store.flush().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn record() -> TraceRecord { + serde_json::from_value(serde_json::json!({ + "id":"variant", "session_id":"s", "turn_id":0, + "role":"user", "content":"你好" + })) + .unwrap() + } + + #[test] + fn trace_codec_preserves_content_and_json_nulls() { + trace_schema().validate().unwrap(); + for content in [None, Some(String::new()), Some("原样\n雪".into())] { + for metadata in [ + None, + Some(Value::Null), + Some(serde_json::json!({"x":[1,true]})), + ] { + let mut record = record(); + record.content = content.clone(); + record.metadata = metadata; + assert_eq!( + TraceRecord::from_row(record.to_row().unwrap()).unwrap(), + record + ); + } + } + } + + #[test] + fn invalid_identity_and_unknown_columns_are_rejected() { + let original = record(); + let mut invalid = original.clone(); + invalid.turn_id = -1; + assert!(invalid.to_row().is_err()); + for field in ["id", "session_id", "role", "content_type"] { + let mut row = original.to_row().unwrap(); + row.insert(field.into(), Value::String(" ".into())); + assert!(TraceRecord::from_row(row).is_err()); + } + let mut row = original.to_row().unwrap(); + row.insert("position".into(), Value::from(0)); + assert!(TraceRecord::from_row(row).is_err()); + let mut row = original.to_row().unwrap(); + row.insert("metadata".into(), Value::String("invalid JSON".into())); + assert!(TraceRecord::from_row(row).is_err()); + } +} diff --git a/crates/lance-context-core/src/lib.rs b/crates/lance-context-core/src/lib.rs index 6fce40f2..7ccfaa2e 100644 --- a/crates/lance-context-core/src/lib.rs +++ b/crates/lance-context-core/src/lib.rs @@ -81,7 +81,9 @@ pub use rollout_store::{ SQL_TABLE_NAME, }; // Schema declaration lives in the API crate: it is part of the wire contract. -pub use lance_context_api::{ColumnSpec, ColumnType, SchemaSpec, ID_COLUMN}; +pub use lance_context_api::{ + trace_schema, ColumnSpec, ColumnType, SchemaSpec, TraceRecord, TraceStore, ID_COLUMN, +}; pub use storage::{ create_local_dir_if_needed, join_uri, remove_dataset, validate_store_name, MAX_STORE_NAME_LEN, }; diff --git a/crates/lance-context-core/tests/trace_store.rs b/crates/lance-context-core/tests/trace_store.rs new file mode 100644 index 00000000..22f7bd4d --- /dev/null +++ b/crates/lance-context-core/tests/trace_store.rs @@ -0,0 +1,181 @@ +use std::collections::BTreeSet; + +use lance_context_api::MAX_BATCH_GET_IDS; +use lance_context_core::{ + trace_schema, GenericStore, GenericStoreOptions, TraceRecord, TraceStore, +}; +use serde_json::json; +use tempfile::TempDir; + +fn turn(id: &str, session: &str, turn_id: i64, content: Option<&str>) -> TraceRecord { + TraceRecord { + id: id.into(), + session_id: session.into(), + turn_id, + role: "assistant".into(), + content: content.map(str::to_owned), + content_type: "text/plain".into(), + source: Some("model_calls".into()), + metadata: Some(json!({"parsed":true})), + } +} + +#[tokio::test] +async fn late_turns_variants_and_replays_survive_merge_and_restart() { + let dir = TempDir::new().unwrap(); + let uri = dir + .path() + .join("trace.lance") + .to_string_lossy() + .into_owned(); + let backend = GenericStore::open( + &uri, + trace_schema(), + GenericStoreOptions { + seal_on_add: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let store = TraceStore::new(backend).unwrap(); + let later = turn("later", "s", 2, Some("C")); + let early = turn("early", "s", 0, Some("A")); + let variant = turn("branch", "s", 2, Some("different C")); + let other = turn("other-session", "other", 0, Some("A")); + store.add(std::slice::from_ref(&later)).await.unwrap(); + store + .add(&[early.clone(), variant.clone(), other.clone()]) + .await + .unwrap(); + // Replay both before and after sealing: identity, not arrival order, decides + // which logical rows survive. Equal positions are allowed across branches. + store.add(&[later.clone(), early.clone()]).await.unwrap(); + let ids = vec![ + "branch".into(), + "missing".into(), + "early".into(), + "later".into(), + "branch".into(), + ]; + assert_eq!( + store.existing_ids(&ids).await.unwrap(), + ["branch", "early", "later"] + ); + assert_eq!( + store.get_many(&ids).await.unwrap(), + [variant.clone(), early.clone(), later.clone()] + ); + let mut backend = store.into_generic(); + backend.cleanup_wal().await.unwrap(); + backend.create_id_index().await.unwrap(); + backend.close().await.unwrap(); + let backend = GenericStore::open_existing(&uri, Default::default()) + .await + .unwrap(); + let store = TraceStore::new(backend).unwrap(); + store.add(&[later.clone(), early.clone()]).await.unwrap(); + assert_eq!(store.get_many(&ids).await.unwrap(), [variant, early, later]); + let rows = store.as_generic().list(None, None).await.unwrap(); + assert_eq!(rows.len(), 4); + assert_eq!( + rows.iter() + .map(|r| r["id"].as_str().unwrap()) + .collect::>() + .len(), + 4 + ); + assert_eq!(store.get_many(&[other.id.clone()]).await.unwrap(), [other]); + store.into_generic().close().await.unwrap(); + assert_eq!( + std::fs::read_dir(dir.path()).unwrap().count(), + 1, + "trace creates one dataset" + ); +} + +#[tokio::test] +async fn batch_probe_uses_generic_limits_projection_and_wal_visibility() { + let dir = TempDir::new().unwrap(); + let backend = GenericStore::open( + dir.path().join("trace.lance").to_str().unwrap(), + trace_schema(), + GenericStoreOptions::default(), + ) + .await + .unwrap(); + let store = TraceStore::new(backend).unwrap(); + let records: Vec<_> = (0..MAX_BATCH_GET_IDS) + .map(|i| turn(&format!("id-{i}"), "s", i as i64, Some("payload"))) + .collect(); + let ids: Vec<_> = records.iter().map(|r| r.id.clone()).collect(); + assert_eq!(store.add(&records).await.unwrap().count, records.len()); + assert!(store.existing_ids(&ids).await.unwrap().is_empty()); + store.flush().await.unwrap(); + assert_eq!(store.existing_ids(&ids).await.unwrap(), ids); + assert_eq!(store.get_many(&ids).await.unwrap(), records); + let projected = store + .as_generic() + .get_many(&ids[..1], Some(&["id".into()])) + .await + .unwrap(); + assert_eq!( + projected, + vec![json!({"id":"id-0"}).as_object().unwrap().clone()] + ); + assert!(store.existing_ids(&[]).await.unwrap().is_empty()); + assert!(store.get_many(&[]).await.unwrap().is_empty()); + let too_many = vec!["id-0".into(); MAX_BATCH_GET_IDS + 1]; + assert!(store.existing_ids(&too_many).await.is_err()); + assert!(store.get_many(&too_many).await.is_err()); + store.into_generic().close().await.unwrap(); +} + +#[tokio::test] +async fn invalid_batch_writes_nothing_and_mismatched_schema_is_rejected() { + let dir = TempDir::new().unwrap(); + let backend = GenericStore::open( + dir.path().join("trace.lance").to_str().unwrap(), + trace_schema(), + GenericStoreOptions { + seal_on_add: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let store = TraceStore::new(backend).unwrap(); + let valid = turn("valid", "s", 0, None); + let invalid = turn("invalid", "s", -1, None); + assert!(store.add(&[valid.clone(), invalid]).await.is_err()); + assert!(store + .existing_ids(&["valid".into()]) + .await + .unwrap() + .is_empty()); + let mut empty = turn("empty", "s", 1, Some("")); + empty.metadata = Some(serde_json::Value::Null); + store.add(&[valid.clone(), empty.clone()]).await.unwrap(); + assert_eq!( + store + .get_many(&["valid".into(), "empty".into()]) + .await + .unwrap(), + [valid, empty] + ); + store.into_generic().close().await.unwrap(); + let wrong = lance_context_api::SchemaSpec::new(vec![( + "id".into(), + lance_context_api::ColumnSpec::required(lance_context_api::ColumnType::String { + large: false, + }), + )]); + let backend = GenericStore::open( + dir.path().join("wrong.lance").to_str().unwrap(), + wrong, + Default::default(), + ) + .await + .unwrap(); + assert!(TraceStore::new(backend).is_err()); +} diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs index f76a2657..ed6f6ef6 100644 --- a/crates/lance-context-server/src/routes/generic.rs +++ b/crates/lance-context-server/src/routes/generic.rs @@ -365,6 +365,60 @@ mod tests { value.as_object().unwrap().clone() } + #[tokio::test] + async fn trace_schema_and_batch_dedup_round_trip_through_existing_http_api() { + use lance_context_api::{trace_schema, TraceRecord, TraceStore}; + use lance_context_client::RemoteGenericStore; + + let (state, _dir) = test_state().await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let app = crate::routes::router().with_state(state); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let backend = RemoteGenericStore::connect_or_create( + &url, + &CreateGenericStoreRequest { + name: "trace-turns".into(), + schema: trace_schema(), + storage_options: None, + seal_on_add: true, + }, + ) + .await + .unwrap(); + let store = TraceStore::new(backend).unwrap(); + let record: TraceRecord = serde_json::from_value(serde_json::json!({ + "id":"turn/雪'", "session_id":"session", "turn_id":3, + "role":"tool", "content":"tool output", "metadata":{"tool_name":"search"} + })) + .unwrap(); + assert_eq!( + store + .add(std::slice::from_ref(&record)) + .await + .unwrap() + .count, + 1 + ); + let ids = vec![record.id.clone(), "absent".into(), record.id.clone()]; + assert_eq!(store.existing_ids(&ids).await.unwrap(), [record.id.clone()]); + assert_eq!(store.get_many(&ids).await.unwrap(), [record.clone()]); + store.add(std::slice::from_ref(&record)).await.unwrap(); + store.flush().await.unwrap(); + let reopened = TraceStore::new( + RemoteGenericStore::connect(&url, "trace-turns") + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(reopened.get_many(&ids).await.unwrap(), [record]); + assert!(reopened + .existing_ids(&vec!["id".into(); MAX_BATCH_GET_IDS + 1]) + .await + .is_err()); + server.abort(); + } + #[tokio::test] async fn batch_get_round_trips_through_http_and_rust_client() { use lance_context_api::GenericStoreApi; diff --git a/crates/lance-context/src/lib.rs b/crates/lance-context/src/lib.rs index 7f0c39d0..0447afd9 100644 --- a/crates/lance-context/src/lib.rs +++ b/crates/lance-context/src/lib.rs @@ -18,17 +18,17 @@ pub use lance_context_core::{ }; pub use lance_context_api::{ - AddDatagenEventsRequest, AddDatagenEventsResponse, AddRecordRequest, AddRecordsResponse, - AddRolloutRequest, AddRolloutsResponse, AddRowsRequest, AddRowsResponse, ColumnSpec, - ColumnType, CompactRequest, CompactResponse, CompactStatsResponse, ContextError, ContextResult, - ContextStoreApi, CreateDatagenStoreRequest, CreateGenericStoreRequest, + trace_schema, AddDatagenEventsRequest, AddDatagenEventsResponse, AddRecordRequest, + AddRecordsResponse, AddRolloutRequest, AddRolloutsResponse, AddRowsRequest, AddRowsResponse, + ColumnSpec, ColumnType, CompactRequest, CompactResponse, CompactStatsResponse, ContextError, + ContextResult, ContextStoreApi, CreateDatagenStoreRequest, CreateGenericStoreRequest, CreateRolloutStoreRequest, DatagenEventDto, DatagenFailureBucketDto, DatagenFailureDto, DatagenFieldStateDto, DatagenRootItemStatusesResponse, DatagenRunOverviewDto, DatagenStepCursorDto, DatagenStoreApi, DatagenStreamPositionDto, DatagenValueDto, DeleteRecordResponse, FoldedDatagenItemDto, GenericStoreApi, GenericStoreInfo, RecordDto, RelationshipDto, RetrieveRequest, RetrieveResponse, RetrieveResultDto, RolloutRecordDto, - RolloutStoreApi, SchemaSpec, SearchResultDto, UpsertRecordRequest, UpsertRecordResponse, - ID_COLUMN, + RolloutStoreApi, SchemaSpec, SearchResultDto, TraceRecord, TraceStore, UpsertRecordRequest, + UpsertRecordResponse, ID_COLUMN, }; #[cfg(feature = "remote")] diff --git a/docs/trace-records.md b/docs/trace-records.md new file mode 100644 index 00000000..43c0b3e9 --- /dev/null +++ b/docs/trace-records.md @@ -0,0 +1,118 @@ +# Trace records + +`TraceRecord` stores one conversation turn per row in one Lance dataset. It is +a built-in schema and typed adapter over `GenericStore`: it uses the existing +WAL, BTree ID index, merge recovery, batch lookup, and HTTP routes. There is no +second copy of the turns, separate reference table, or per-turn checkpoint store. + +| Field | Type | Meaning | +| --- | --- | --- | +| `id` | required string | Stable storage identity, reused on retries | +| `session_id` | required string | Conversation grouping key | +| `turn_id` | required int64, nonnegative | Zero-based turn order in the observed conversation | +| `role` | required string | User, assistant, system, tool, or a source-specific role | +| `content` | nullable large string | Exact content of this turn; null differs from empty | +| `content_type` | required string | MIME type; the typed record defaults to `text/plain` | +| `source` | nullable string | Source dataset or system | +| `metadata` | nullable large string | JSON-encoded source-specific annotations | + +The typed Rust `metadata` field is `Option`; `to_row` and +`from_row` convert it to/from the physical JSON-text column. Whole-call token +arrays, prompts, and responses should not be copied into every turn's metadata. +No embeddings, model calls, or training labels are generated implicitly. + +## Identity and reconstruction + +The caller assigns `id`. For deduplicated observations, derive it deterministically +from the source/tenant scope, session ID, turn ID, role, content type, and exact +content. Use unambiguous serialization and preserve null versus empty content. +Turn IDs come from the input's order, not arrival time. Receiving turns 0–2 after +turns 0–4 must produce the same IDs for the shared prefix. + +`(session_id, turn_id)` is **not unique**. Different content at the same position +uses different IDs, retaining both observations. A session without variants can +be reconstructed by sorting on `turn_id`. When there are branches or truncated +inputs, positions alone do not establish a global trajectory or identify which +variants appeared in a particular call. The application must preserve that +association explicitly if required; the schema does not invent branch links. +Source deletions do not automatically retract a turn shared by multiple calls. + +## Embedded Rust + +```rust,no_run +use lance_context_core::{ + trace_schema, GenericStore, GenericStoreOptions, TraceRecord, TraceStore, +}; + +# async fn example() -> Result<(), Box> { +let backend = GenericStore::open( + "az://container/trace-turns.lance", + trace_schema(), + GenericStoreOptions { seal_on_add: true, ..Default::default() }, +).await?; +let traces = TraceStore::new(backend)?; +let record = TraceRecord { + id: "stable-content-derived-id".into(), + session_id: "session-42".into(), + turn_id: 0, + role: "user".into(), + content: Some("Hello".into()), + content_type: "text/plain".into(), + source: Some("model_calls".into()), + metadata: None, +}; +let ids = vec![record.id.clone()]; +let present = traces.existing_ids(&ids).await?; +if present.is_empty() { + traces.add(&[record]).await?; +} +let records = traces.get_many(&ids).await?; +let mut backend = traces.into_generic(); +backend.cleanup_wal().await?; +backend.create_id_index().await?; +backend.close().await?; +# Ok(()) +# } +``` + +`TraceStore::new` checks the persisted schema before typed access. The same +adapter accepts core `GenericStore`, the local/remote `lance_context::GenericStore`, +or `lance_context_client::RemoteGenericStore`. Their maintenance and filtered +projection APIs remain accessible through `as_generic` / `into_generic`. + +## Batch deduplication and durability + +`get_many` returns complete typed records. `existing_ids` uses the **same batch +lookup introduced for GenericStore**, projecting only `id`; it does not fetch +content or metadata. Both accept up to 1,024 input IDs (including duplicates), +return found IDs once in first-requested order, and omit missing IDs. Larger +requests must be split into bounded batches by the caller. Reads resolve the +base and flushed WAL together, including newest-ID resolution during merges. + +An existence probe followed by `add` is not a cross-writer atomic insert-if-absent +operation. Coordinate writers when immutable conflict rejection is required. +`add` validates the complete typed batch before writing and inherits generic +newest-write-wins semantics. Reusing an ID with changed content replaces that +logical record; it does not preserve a variant or merge metadata/provenance. + +`seal_on_add: true` makes the append visible before acknowledgement; otherwise, +call `flush` before depending on a previous write being visible to dedup probes. +Only advance the source checkpoint after durable destination acknowledgement. +The returned base version is not a per-append commit token or a complete source +snapshot. Checkpoints and source scheduling belong to the ingestion service. + +## HTTP and remote Rust + +Create a generic store with `schema: trace_schema()` and the required +`seal_on_add` setting, then use the existing endpoints: + +- `POST /api/v1/generic/{name}/rows`: batch append encoded trace rows. +- `POST /api/v1/generic/{name}/get-rows`: batch lookup; use + `{"ids":["a","b"],"columns":["id"]}` for deduplication probes. +- `POST /api/v1/generic/{name}/flush`: publish deferred appends. + +Rust clients can wrap `RemoteGenericStore::connect_or_create` with +`TraceStore::new`, so typed encoding, batch reads, and ID-only probes are identical +for local and remote storage. Raw generic HTTP writes enforce the physical schema; +the additional nonempty-identity/nonnegative-turn checks are performed by the +typed adapter. Raw callers must apply the same semantic validation. From 462f6be35b78f5cb48a9cd306b392ef9a90abdaa Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Sun, 4 Oct 2026 18:30:40 +0000 Subject: [PATCH 2/4] test: allow native Lance future layout in trace integration tests --- crates/lance-context-core/tests/trace_store.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/lance-context-core/tests/trace_store.rs b/crates/lance-context-core/tests/trace_store.rs index 22f7bd4d..1451c9b6 100644 --- a/crates/lance-context-core/tests/trace_store.rs +++ b/crates/lance-context-core/tests/trace_store.rs @@ -1,3 +1,5 @@ +#![recursion_limit = "256"] + use std::collections::BTreeSet; use lance_context_api::MAX_BATCH_GET_IDS; From 1a29923cfce0c00ae1a3c63adeb14db73e3bbe2c Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Sun, 4 Oct 2026 18:31:06 +0000 Subject: [PATCH 3/4] test: borrow the single trace lookup key --- crates/lance-context-core/tests/trace_store.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/lance-context-core/tests/trace_store.rs b/crates/lance-context-core/tests/trace_store.rs index 1451c9b6..10b4e67e 100644 --- a/crates/lance-context-core/tests/trace_store.rs +++ b/crates/lance-context-core/tests/trace_store.rs @@ -87,7 +87,13 @@ async fn late_turns_variants_and_replays_survive_merge_and_restart() { .len(), 4 ); - assert_eq!(store.get_many(&[other.id.clone()]).await.unwrap(), [other]); + assert_eq!( + store + .get_many(std::slice::from_ref(&other.id)) + .await + .unwrap(), + [other] + ); store.into_generic().close().await.unwrap(); assert_eq!( std::fs::read_dir(dir.path()).unwrap().count(), From 44910a1e652c44c10943b50fd8bf2ff1ff8ec83d Mon Sep 17 00:00:00 2001 From: Beinan Wang Date: Sun, 4 Oct 2026 18:42:49 +0000 Subject: [PATCH 4/4] test: borrow expected trace records in HTTP assertions --- crates/lance-context-server/src/routes/generic.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs index ed6f6ef6..27177607 100644 --- a/crates/lance-context-server/src/routes/generic.rs +++ b/crates/lance-context-server/src/routes/generic.rs @@ -401,8 +401,14 @@ mod tests { 1 ); let ids = vec![record.id.clone(), "absent".into(), record.id.clone()]; - assert_eq!(store.existing_ids(&ids).await.unwrap(), [record.id.clone()]); - assert_eq!(store.get_many(&ids).await.unwrap(), [record.clone()]); + assert_eq!( + store.existing_ids(&ids).await.unwrap(), + std::slice::from_ref(&record.id) + ); + assert_eq!( + store.get_many(&ids).await.unwrap(), + std::slice::from_ref(&record) + ); store.add(std::slice::from_ref(&record)).await.unwrap(); store.flush().await.unwrap(); let reopened = TraceStore::new(