Ingestion & provenance¶

Architecture

Every fact Telha ingests must be traceable to the exact bytes it came from, below document granularity. This page documents how source identity and dedup work, how a worker's parse becomes graph nodes and edges in one atomic chain, the streaming protocol that lets a worker submit a large parse across several round trips, and the synchronous JSON bypass that skips the worker fleet entirely. Spans are the evidence unit claim verification and the memory planner both stand on; see Memory planner for how they get selected and packed.

Overview & purpose¶

Without pinned provenance, re-ingesting the same document duplicates the graph, "where did this fact come from" has no answer below document granularity, and byte offsets are ambiguous across six ingestion formats (raw file bytes are meaningless for PDF or other binary formats). The spec's answer is to make every ingestion produce a source version: identity is (tenant, source_id), each version keyed by transaction time and carrying a content_hash, and spans are byte offsets into that version's canonical extracted text, never into raw file bytes and never into a token-model's tokenization.

Two identity rules follow directly from that:

• Dedup is content-addressed. Same (source_id, content_hash) produces zero new rows; a new hash under the same source_id produces a new version. History is never overwritten.
• Spans are only ever comparable within one source version. A byte range [10, 40) in one version's canonical text has no defined relationship to the same range in another version, or in another source. The memory planner's dedup logic depends on this (spec, as-built note 1).

Design¶

Source identity¶

pub fn source_id_for(name: &str) -> Uuid {
    let h = xxhash_rust::xxh3::xxh3_128(format!("src:{name}").as_bytes());
    Uuid::from_u128(h)
}

source_id is deterministic in the caller-supplied name or URL: re-ingesting under the same name always addresses the same source, without a lookup. Connector-synced items (PR-065) instead derive source_id from (connector, tenant, external_id) via source_meta::connector_source_id, so a re-sync of the same external item addresses the same source regardless of what it is named on that sync. Content identity is separate: content_hash is BLAKE3 of the raw ingested bytes, computed fresh on every call to ingest().

The Ingestor and its three entry paths¶

pub struct Ingestor {
    engine: Arc<dyn StorageEngine>,
    nodes: Arc<NodeOps>,
    edges: Arc<EdgeOps>,
    jobs: Arc<JobQueue>,
    config: IngestConfig,
    clarify: Option<Arc<crate::clarify::ClarifyOps>>,
}

Ingestor::ingest(scope, format, name, content, options) is the single entry point for POST /v1/ingest. It always checks the GDPR erasure blocklist before any dedup match (a retired digest must refuse, never silently match masked content), then branches three ways:

1. Fetch formats (format == "web") skip request-level dedup entirely. The request bytes are the URL and crawl options, not the content; the worker fetches the real bytes and freshness is decided at submit time against the fetched content's hash.
2. format == "json" takes the synchronous bypass (json_bypass, see below) after the same dedup check as any non-fetch format.
3. Every other format enqueues an ingest:{format} job for a worker to pick up and returns IngestOutcome::Enqueued { job_id, source_id }.

The job payload¶

pub struct IngestJobPayload {
    pub v: u32,
    pub source_id: Uuid,
    pub format: String,
    pub name: String,
    pub content_b64: String,     // raw content, base64
    pub content_hash: [u8; 32],  // BLAKE3 of the raw bytes
    pub options: serde_json::Value, // format-specific knobs, passed through verbatim
}

The payload is JSON (not msgpack, unlike JobRecord's own envelope), stored as the job's opaque payload: Vec<u8>. content_b64 is flagged in the spec's §3.19 GDPR addendum as a first-class personal-data surface: terminal ingest jobs whose content is blocklisted have it deleted outright.

Data model¶

SourceVersion (sources CF value)¶

pub struct SourceVersion {
    pub source_id: Uuid,
    pub name: String,
    pub format: String,
    pub content_hash: [u8; 32],
    pub canonical_text_hash: Option<[u8; 32]>, // None until a worker submits its parse
    pub size: u64,
    pub worker: Option<String>,
    pub job_id: Uuid,
    pub blob_id: Option<Uuid>,          // provisional id for streamed (multi-partial) versions
    pub source_connector: Option<String>, // PR-065 delete-authority scope
    pub external_id: Option<String>,
    pub acl_json: Option<String>,        // advisory, floor-only ACL
    pub root_node_id: Option<Uuid>,      // projection root; the externalId lookup's answer
}

Keyed [scope][source_id 16][inv(tx_time_start) 8] per the storage-engine layout (56 bytes), so ScanRange::source_history returns a source's versions newest-first. canonical_text_hash starts None at job creation and lands once a worker's parse is applied; for streamed submissions it stays None until the stream's finalize step.

SpanRow (spans CF value): the reverse index¶

pub struct SpanRow {
    pub source_version_tx: Ts,
    pub linked_nodes: Vec<Uuid>,     // reverse index: span → nodes
    pub kind: String,                // paragraph | heading | cell | code | body
}

Keyed [scope][source_id 16][span_start 8][span_end 8] (64 bytes), not inverted: spans iterate in document order. This is the mirror of the forward-direction provenance every node already carries: a node's provenance.spans: Vec<SpanRef> points at the spans it came from; the spans CF's linked_nodes points back. Both are written in the same batch chain so they never drift.

SpanRef: the coordinate every consumer shares¶

pub struct SpanRef {
    pub source_id: Uuid,
    pub source_version_tx: u64, // pins the canonical text this offset is measured against
    pub start: u64,             // inclusive
    pub end: u64,                // exclusive
}

This struct lives in core-engine/src/model/mod.rs as part of Provenance, not in the ingest module, precisely because it is the shared coordinate system: node provenance, span-ledger rows, and the memory planner's PlannedSpan all reference spans through this exact four-field shape.

Canonical text storage: hash-addressed chunks¶

Canonical text (the worker's NFC/LF-normalized extraction) is stored as 1 MB chunks (BLOB_CHUNK = 1024 * 1024) directly in the sources CF, keyed by a blob_id that is either derived from the canonical text's BLAKE3 hash (single-shot submissions: blob_id = Uuid::from_slice(&canonical_hash[..16])) or an explicit provisional id (streamed submissions, allocated before the hash is known). Chunking under the same content-addressed scheme the storage-engine spec uses elsewhere gives blob dedup for free: two sources whose canonical text happens to match byte-for-byte share chunk storage.

Algorithms & invariants¶

Single-shot apply: apply_parts¶

For a non-streamed submission (seq == 0 && !partial), apply_parts runs a fixed sequence, writing each stage in its own atomic batch:

1. Compute canonical_hash = blake3(canonical_text). Check the erasure blocklist against both content_hash and canonical_hash (fetch formats only learn their real content hash here).
2. Submit-time dedup: if the latest stored version has the same content_hash and canonical_text_hash, and it was not written by this same job (a same-job retry must re-apply, not dedup, to support crash recovery), return SubmitOutcome { deduplicated: true, .. } with zero new rows.
3. Write the canonical text as hash-addressed chunks plus the source version row, in one batch.
4. write_nodes: create (or upsert, see below) every node, each carrying provenance.spans resolved from the submission's span_refs indices.
5. write_relationships: create every edge, endpoints resolved by submission index or by an existing logical id.
6. write_span_rows: the reverse-index rows, linked_nodes computed by scanning which node indices reference each span index.
7. enqueue_children: submission-requested follow-up jobs (crawl fan-out), each subject to the same request-level dedup as a top-level ingest() call.

The streaming multi-submit protocol¶

A worker may stream one job's parse as several submissions rather than one. Sequencing (seq, partial) lives in the Submission JSON itself, not in the gRPC wire envelope. ApplyOutcome is Complete(SubmitOutcome) or Partial(PartialOutcome), and apply_submission dispatches on the job's stored continuation plus the incoming seq/partial:

sequenceDiagram
    autonumber
    participant W as Worker
    participant Q as JobQueue (leased job row)
    participant I as Ingestor

    W->>I: Submission seq=0, partial=true
    Note over I: First partial pins ONE source version<br/>(tx fixed, canonical_text_hash absent)<br/>+ allocates provisional blob_id
    I->>I: append text chunks under blob_id<br/>write nodes/edges (global offsets)<br/>write span reverse-index rows
    I->>Q: checkpoint(IngestContinuation{next_seq:1,...}), extend lease
    I-->>W: PartialOutcome{seq:0, node_ids, lease_deadline}

    W->>I: Submission seq=1, partial=true
    Note over I: continuation.next_seq == 1: matches
    I->>I: append more text (offsets continue from<br/>accumulated length) + nodes/edges/spans
    I->>Q: checkpoint(next_seq:2), extend lease
    I-->>W: PartialOutcome{seq:1, node_ids, lease_deadline}

    W->>I: Submission seq=2, partial=false (finalize)
    Note over I: hash accumulated chunks via BLAKE3<br/>rewrite version row with canonical_text_hash
    I->>I: check blocklist against final canonical_hash
    I-->>W: ApplyOutcome::Complete(SubmitOutcome)

Key rules, all as documented in the spec's §16 addendum and implemented in apply_stream_part:

• One pinned source version per stream. The first partial writes the version row at a fixed tx, with canonical_text_hash absent and blob_id set to a freshly allocated Uuid::now_v7().
• Global offsets. Every partial's spans and node provenance use byte offsets into the accumulated canonical text across the whole stream, not offsets relative to that partial alone.
• Finalize re-hashes from storage, not from memory. The finalizing submission streams every stored chunk back through a blake3::Hasher to compute canonical_hash, then rewrites the version row. This makes the hash authoritative over what was actually persisted, not over what the worker's process happened to hold.
• Streamed chains skip submit-time dedup. The canonical hash is unknown until finalize, so there is nothing to dedup against mid-stream; request-level dedup (on the original content_hash) still applies to the job's inline formats.
• Duplicate seq is an idempotent replay. If the worker retries the previously-acknowledged seq after a lost ack, apply_submission matches submission.partial && submission.seq + 1 == continuation.next_seq, re-checkpoints to extend the lease, and returns the same node_ids without writing anything (PartialOutcome.replay = true).
• Wrong seq is OutOfSync. Any other mismatch, no continuation but seq != 0, or a seq that matches neither the expected value nor the replay case, returns IngestError::OutOfSync, which maps to gRPC ABORTED; the worker's clean recovery is to restart the job (a fresh lease generation clears the continuation automatically, per the job-queue spec).

The json_bypass synchronous path¶

format=json skips the worker fleet entirely. Core parses, projects, and applies within the POST /v1/ingest call itself:

let outcome = self.apply_parts(scope, &payload, Uuid::nil(), &submission, "core:json-bypass")?;
Ok(IngestOutcome::Completed { source_id, source_version_tx, node_ids, edge_ids })

job_id = Uuid::nil() marks the bypass: no job row is ever created, and the stored source version's worker field reads "core:json-bypass". The REST response is final, 201 with sourceId/sourceVersionTx/nodeIds/ edgeIds, nothing to poll. json_bypass::project_json builds the Submission in-process: canonical text is one NDJSON line per top-level record (compact serde_json rendering, LF-joined), each record's line is its own span of kind body, and nested objects flatten to dotted keys or become child nodes under ingest.json_nested (flatten default, or child, overridable per request via options.nested).

Deterministic node ids and the upsert ladder¶

Workers are tenant-blind: they cannot read the graph, yet some entities (email PERSON/MESSAGE, code FILE) need identity that survives across separate ingestion jobs. NodeIn.id (optional) lets a worker supply a client-minted deterministic UUID (uuid5 conventions documented per worker spec). write_nodes resolves it through NodeOps::upsert_node, whose outcome ladder is:

Edges may reference not-yet-ingested deterministic ids; EdgeOps does not validate the far endpoint at write time, so such edges dangle invisibly until the referenced node arrives (every read path validates endpoints). This is the mechanism behind the email worker's pending-reply linkage and the code worker's cross-file imports.

Configuration¶

The spec's §8 also names ingest.max_upload (200 MB) and ingest.canonical_chunk (1 MB); as built, the canonical chunk size is the compile-time constant BLOB_CHUNK = 1024 * 1024 in core-engine/src/ingest/mod.rs rather than a runtime IngestConfig field, and max_upload was not found as an IngestConfig key (upload size limits are enforced at the REST body-limit layer, outside this module). Treat those two as request/transport-layer settings, not [ingest] TOML keys, until the spec is reconciled.

The slice API¶

Downstream consumers (the memory planner, prompt assembly, the trace viewer) never touch the sources CF directly; they go through Ingestor's slice API:

canonical_slice is the one function the memory-planner spec calls out by name as the slice API its recall phase depends on (see Memory planner); byte offsets are worker-guaranteed to fall on UTF-8 boundaries, but a lossy conversion backstops corrupt offsets rather than failing the whole plan.

As-built notes¶

From spec §14 (Changelog), reconciled against core-engine/src/ingest/:

• v0.2 through v0.6 are all present in the code, not just the base v0.1 design: IngestJobPayload.options pass-through, submission stats/ contentHash/childJobs, submit-time dedup with the same-job exemption, the §15 JSON bypass, §16 multi-submit streaming, §17 deterministic node ids and inline child content, §18 shared attachment documents (content-hash named child jobs), and §3.19 GDPR erasure interaction (blocklist checks before every dedup match, blob redaction records, content_b64 scrubbing).
• The blocklist check runs before dedup on every path exactly as the §3.19 addendum requires: ingest() (request-level), apply_parts and apply_stream_part's finalize step (submit-time, against both content_hash and canonical_hash), and enqueue_children (child-enqueue path, where a blocklisted child is refused and logged while the parent submission proceeds rather than failing outright).
• ingest.max_upload and ingest.canonical_chunk are not IngestConfig fields in the code as read; see Configuration above. This is a documentation gap between spec §8 and the as-built IngestConfig struct, not a functional deviation, since the chunk size is fixed and still 1 MB.
• Not directly verifiable from core-engine/src/ingest/ alone: the spec's success metrics around a standalone "reverse-index consistency checker tool" and the six format workers' own canonical-text/span production (docling, web, tabular, code, email) live in worker-specific code and specs outside this module, and were not inspected for this page.

Related¶

• Storage engine & key layout · the sources and spans CF byte layouts.
• Job queue & leases · the ingest:{format} job kind and the continuation checkpoint this module writes.
• Memory planner · the consumer of spans as the evidence unit.
• GDPR erasure · blob redaction records and the erased-content blocklist.