Ingest raw interaction logs into lance-context records (batch + streaming)

[dcfocus]

Summary

Upstream half of the post-training data pipeline. Add an ingestion layer that loads raw agent/interaction logs (chat transcripts, tool-call logs, RL rollout dumps) into faithful ContextRecords tagged source="raw", supporting both batch and streaming ingestion. The downstream half — curating those records into trainable datasets and exporting them — is tracked in #96.

raw logs ──(this issue) ingest/normalize──▶ ContextRecords ──(#96) curate + export──▶ trainable dataset
                                            (source="raw", faithful)

Motivation

lance-context is not where raw training data is generated. But to use it as a curation/staging layer for post-training (#96), raw logs must first land in the store faithfully (lossless) so they can be re-curated into different versioned cuts later — distilling on the way in would throw away the raw and defeat the reproducibility/versioning advantage.

Today the only entry points are add / add_many / upsert of already-shaped records. There is no adapter that maps common raw log formats into ContextRecords, and no documented batch-vs-streaming ingestion path.

Proposed work

1. Ingestion adapters / normalization — map common raw formats into ContextRecord:

   • chat transcripts (OpenAI-style messages: role/content)
   • tool-call / function-call logs (tool name, args, result) → role + content_type + state_metadata
   • RL rollout dumps (prompt, one or more responses, reward) → samples linked via relationships / external-id prefix
   • generic JSONL / Parquet / Arrow with a user-supplied field mapping
   • preserve provenance: original ids → external_id, timestamps → created_at, source="raw", plus session_id / run_id / tenant / bot_id tags

2. Batch ingestion — bulk-load a file / dataset / directory (JSONL, Parquet, Arrow) into records via chunked atomic appends (build on add_many and the batch-write path, Add batch write API for atomic multi-record appends #54). Deferred-embedding compatible (Support deferred embedding workflows for bulk ingestion #88): ingest text now, attach embeddings later.

3. Streaming ingestion — incremental append for live/continuous sources (a queue, a tail, a generator), and for chunked reads of very large dumps without materializing the whole input in memory.

   • This is a genuinely new execution model, not a wrapper over add_many: add_many eagerly materializes the entire input before writing and embeds the whole batch in one call (see code-reading note below), so it cannot stream an unbounded source or a dump larger than memory.
   • Real work: bounded-memory chunked reads; configurable commit cadence (records per Lance version / flush interval) — and since each commit is a new version, compaction pressure must be managed; resumability / checkpointing (a cursor/offset so an interrupted ingest resumes without re-appending); partial-failure semantics.
   • Uniqueness validation must not be O(n) per chunk. The current append path full-scans the dataset on every write to check id/external_id uniqueness (validate_unique_ids), so naive per-chunk add_many is O(n²). Streaming must validate via the id index / a bulk-upsert path instead.
   • Mirrors the streaming export in Curate lance-context records into trainable datasets (SFT/preference/rollout export) #96 so the pipeline is streaming-capable end to end.

4. Idempotent ingestion — re-ingesting the same source must not duplicate rows. Note add_many rejects duplicate external_ids rather than merging them, so this needs chunked upsert/merge semantics or an explicit bulk-upsert path keyed on external_id. (This is identity dedup; semantic dedup/decontamination is curation and lives in Curate lance-context records into trainable datasets (SFT/preference/rollout export) #96.)

5. Examples + docs

   • batch: ingest a chat-log JSONL dump into a context
   • streaming: tail a rollout/event stream into a context with a bounded flush interval and checkpoint/resume

Non-goals

• Not a new write path. Reuses add_many / upsert / deferred-embedding as the storage backend; this issue is the adapters + streaming runtime on top, not a replacement for the existing append/upsert APIs.
• No semantic dedup, decontamination, quality filtering, consolidation, or reward computation here — those are curation steps and belong to Curate lance-context records into trainable datasets (SFT/preference/rollout export) #96. Ingestion stays faithful and lossless.
• Bundles no models (no auto-embedding decisions beyond the existing provider registry / deferred-embedding paths).

Acceptance criteria

• A user can batch-load chat / tool-call / rollout logs from JSONL/Parquet/Arrow into ContextRecords with preserved provenance and source="raw".
• A user can stream-ingest from a continuous source (and from a dump larger than memory) without loading the entire input, with a configurable commit cadence.
• Streaming ingest resumes from a checkpoint after interruption without duplicating rows, and does not degrade to a full-store scan per chunk.
• Re-ingesting the same source by external_id is idempotent (no duplicate rows).
• Adapters preserve the same provenance/state fields across the Python, Rust/core, and REST surfaces (field parity).
• Ingested records are deferred-embedding compatible (text first, embeddings later).
• Tests cover batch + streaming paths, provenance mapping, field parity, resumability, and idempotency.

Notes

This is the interface contract with #96: ingestion produces faithful ContextRecords; curation/export consumes them. The ContextRecord schema (and the export manifest defined in #96) is the shared boundary — defined once, referenced by both issues. Builds on existing primitives (add_many, batch write #54, deferred embedding #88, embedding provider registry #87) rather than introducing a separate raw-data table.

Implementation audit (2026-06-23)

• Validated against the current repo: the issue is legitimate; there are no raw-log ingestion adapters and no documented batch-vs-streaming ingestion layer today.
• Existing substrate is real: add_many batch appends exists (Add batch write API for atomic multi-record appends #54 closed), upsert by external_id exists for single-record idempotent replacement, deferred embeddings exist (Support deferred embedding workflows for bulk ingestion #88 closed), and the embedding provider registry exists (feat: pluggable embedding provider registry #87 merged).
• Clarification for idempotency: idempotent batch/stream re-ingestion remains part of this issue. add_many rejects duplicate external_ids rather than merging them, so ingestion adapters need chunked upsert/merge semantics or an explicit bulk-upsert path.
• Clarification for field parity: adapters must preserve provenance and state consistently across Python, Rust/core DTOs, and REST. Some surfaces do not expose exactly the same add/upsert fields today, so field parity should be included in acceptance tests.
• Keep ingestion faithful and lossless; reward/verifier scoring, summarization, semantic dedup, and training-specific consolidation belong in Curate lance-context records into trainable datasets (SFT/preference/rollout export) #96.

Streaming design notes (code-reading, 2026-06-23)

Two concrete findings from the current code that shape the streaming design and distinguish this issue from the existing "store records" APIs:

• add_many is batch-only, not streaming. The Python wrapper materializes the full input into a list before writing (python/python/lance_context/api.py:657-717) and embeds the whole batch in one provider call (api.py:722-737); the core takes a &[ContextRecord] slice and writes a single version (crates/lance-context-core/src/store.rs:364, :373). There is no streaming path at any layer — bounded-memory streaming is the net-new capability here.
• Uniqueness validation is a full table scan per append. validate_unique_ids lists the entire dataset on every add/add_many (crates/lance-context-core/src/store.rs:670 → list_with_options(None, None, ...) at :699) to reject duplicate id/external_id. High-frequency streaming commits via this path would be O(n²); the streaming ingester must validate through the id index or a bulk-upsert path rather than the current full-scan validation.