Ingestion¶

Get bytes into the graph, with proof

Every fact in Telha traces back to the exact bytes it came from. Ingestion is how those bytes enter the system: POST /v1/ingest accepts a document, either applies it immediately or hands it to a Python format worker, and every node it creates carries a provenance.spans pointer into the source's canonical text.

The sync/async split¶

format=json is parsed and projected by core itself, in the request, with no worker involved. Every other format is parsed by a Python format worker over gRPC, so the request only enqueues a job and returns immediately.

Connector formats (entra, sharepoint, exchange, slack, salesforce) follow the same asynchronous path through their own connector workers; see Connectors.

A re-submission of byte-identical content against the same source dedupes to 200 OK with {"deduplicated": true, "sourceId": ...} and writes nothing, for every format.

POST /v1/ingest¶

Auth: x-api-key (REST) or a connector token. Request body:

{
  "format": "json",
  "name": "risks-2026-07.json",
  "content": "...",
  "contentBase64": null,
  "options": {}
}

For format=web, both content and contentBase64 may be omitted: the name (the URL) is the request identity and the worker fetches the page itself. Every other format requires exactly one of content / contentBase64; supplying both, or neither, is 400 INVALID_CONTENT.

Responses¶

Worked example: JSON (synchronous, 201)¶

format=json needs nothing running besides telha serve itself:

curl -sS -X POST http://127.0.0.1:7625/v1/ingest \
  -H "x-api-key: $KEY" -H "Content-Type: application/json" \
  -d '{
    "format": "json",
    "name": "risks-2026-07.json",
    "content": "[{\"label\": \"RISK\", \"title\": \"Supplier delay\", \"severity\": 7, \"status\": \"open\"}]"
  }'

const res = await fetch("http://127.0.0.1:7625/v1/ingest", {
  method: "POST",
  headers: { "x-api-key": key, "Content-Type": "application/json" },
  body: JSON.stringify({
    format: "json",
    name: "risks-2026-07.json",
    content: JSON.stringify([
      { label: "RISK", title: "Supplier delay", severity: 7, status: "open" },
    ]),
  }),
});

import httpx, json

resp = httpx.post(
    "http://127.0.0.1:7625/v1/ingest",
    headers={"x-api-key": key},
    json={
        "format": "json",
        "name": "risks-2026-07.json",
        "content": json.dumps([
            {"label": "RISK", "title": "Supplier delay", "severity": 7, "status": "open"}
        ]),
    },
)

{
  "deduplicated": false,
  "sourceId": "b6b2...",
  "sourceVersionTx": 1751500000000000,
  "nodeIds": ["9c11..."],
  "edgeIds": []
}

Re-POST the exact same bytes and the response becomes 200 {"deduplicated": true, "sourceId": "b6b2..."} with no new rows.

Worked example: a PDF (asynchronous, 202 then poll)¶

pdf (and docx, pptx) route to docling_worker, so a worker must be running against the server's gRPC listener before the job can leave pending.

B64=$(base64 -w0 report.pdf)
curl -sS -X POST http://127.0.0.1:7625/v1/ingest \
  -H "x-api-key: $KEY" -H "Content-Type: application/json" \
  -d "{\"format\": \"pdf\", \"name\": \"report.pdf\", \"contentBase64\": \"$B64\"}"

{
  "deduplicated": false,
  "jobId": "0a1e...",
  "sourceId": "7cf4...",
  "statusUrl": "/v1/ingest/0a1e..."
}

curl -sS http://127.0.0.1:7625/v1/ingest/0a1e... -H "x-api-key: $KEY"

{
  "jobId": "0a1e...",
  "kind": "ingest:pdf",
  "state": "completed",
  "attempts": 1,
  "events": [
    { "at": 1751500000000000, "transition": "leased", "detail": null },
    { "at": 1751500001200000, "transition": "completed", "detail": null }
  ]
}

state is one of pending, leased, completed, dead. A job that never leaves pending almost always means no worker is consuming that job kind.

Sequence: the asynchronous path¶

sequenceDiagram
    autonumber
    participant C as Client
    participant Core as telha-core (REST)
    participant Q as Job queue (gRPC)
    participant W as docling_worker

    C->>Core: POST /v1/ingest (format=pdf, contentBase64)
    Core->>Core: content-hash dedup check
    Core->>Q: enqueue ingest:pdf job
    Core-->>C: 202 {jobId, sourceId, statusUrl}
    W->>Q: lease ingest:pdf job (gRPC worker protocol)
    W->>W: parse to canonical text + spans + nodes
    W->>Core: SubmitIngestionResult
    Core->>Core: write source version, blob, nodes/edges, span rows
    C->>Core: GET /v1/ingest/:jobId
    Core-->>C: 200 {state: "completed", events: [...]}

When a worker must be running¶

format=json needs nothing beyond telha serve. Every other format needs its worker started against the same gRPC listener, with a worker token minted from the operator's grpc.token_key:

export TELHA_GRPC_TARGET=127.0.0.1:7626
export TELHA_WORKER_TOKEN=$(telha api-key worker-token --tenant $TENANT --org $ORG)
python -m workers_common.run docling_worker   # or: tabular_worker, email_worker, code_worker, firecrawl_worker

Until the matching worker is running, jobs of that kind sit in pending forever; GET /v1/ingest/:jobId will keep reporting that state. See the deployment runbook for gRPC token setup.

content vs contentBase64¶

Exactly one is required for every format except web. content is a plain JSON string, use it for text you already have as a string (JSON payloads, small text files). contentBase64 carries arbitrary bytes, base64-encoded, required for binary formats (PDF, DOCX, PPTX, XLSX) and safe for anything else too. Requests are capped by the REST body limit (8 MiB); larger files need the streaming multi-submit protocol on the worker side, not a bigger request.

options passthrough¶

options is an opaque JSON object forwarded verbatim from the REST request into the worker's job payload (or, for format=json, read directly by the core bypass). Each format defines its own keys:

format=json options in detail¶

format=json accepts exactly one JSON object, or an array of objects (a batch, capped by ingest.json_max_records, default 10000). Anything else at the top level, bare scalars, arrays of scalars, mixed arrays, is 400 INVALID_CONTENT.

• options.label: the node label for created records. Defaults to the UPPER_SNAKE form of the request name with its extension stripped (so risks-2026-07.json becomes label RISKS_2026_07).
• options.nested: "flatten" (default, from ingest.json_nested) or "child".
  • flatten: nested objects become dotted property keys (address.city), recursing to ingest.json_max_depth (default 8). A dotted-key collision with a literal key is INVALID_CONTENT.
  • child: a nested object becomes a separate child node labeled UPPER_SNAKE(key), linked from the parent by a HAS_<KEY> edge. Arrays of objects are always child nodes under either policy (there is no meaningful flattening of an object array); each HAS_<KEY> edge carries an ord property for array position.

{
  "format": "json",
  "name": "contracts.json",
  "options": {"label": "CONTRACT", "nested": "child"},
  "content": "{\"title\": \"MSA\", \"parties\": [{\"name\": \"Acme\"}, {\"name\": \"Globex\"}]}"
}

This creates one CONTRACT node, two PARTIES child nodes, and two HAS_PARTIES edges (ord: 0, ord: 1) from the contract to each party.

Content-hash dedup¶

Every ingested source is identified by (tenant, source_id), where source_id is derived deterministically from the request name. Every ingested version carries a BLAKE3 content_hash of the raw bytes. Before writing anything, Telha checks the latest version of that source:

• Same source_id, same content_hash → no new version; the response is 200 {"deduplicated": true, "sourceId": ...}.
• Same source_id, different content_hash → a new version is written (the source's history grows; nothing is deleted).
• New source_id → a fresh source, version 1.

format=web skips this request-level check (the request body is just the URL, not the page content); dedup instead happens at submit time against the worker-reported hash of the fetched bytes, so an unchanged recrawl is still a no-op.

Source versions¶

Each ingested source accumulates versions over time, keyed by transaction time. A SourceVersion records: sourceId, the original name, format, the raw content_hash, the canonical_text_hash (the BLAKE3 of the extracted, normalized text, set once a worker's parse lands), size, the worker that produced it (core:json-bypass for the direct path), and the job_id that wrote it. Corrections never overwrite: re-ingesting under the same name with different bytes always adds a new version, and every prior version stays reconstructable at its own transaction time, the same append-only discipline as the rest of the tritemporal model.

Provenance spans¶

Every node a worker (or the JSON bypass) creates carries provenance.spans: a list of (sourceId, sourceVersionTx, start, end) references. The offsets are byte offsets into the canonical extracted text of that specific source version, never into the raw file bytes (meaningless for a PDF) and never dependent on a tokenizer (spans are forever; token boundaries change when the model does). Canonical text is normalized (NFC, LF-only newlines) before hashing or spanning so that offsets are stable across workers.

Spans are also written as a reverse index: a spans row per (sourceId, start, end) records which node ids that span produced, alongside a structural kind (paragraph, heading, cell, code, or body). This is the same span reference type (Architecture › Ingestion & provenance) that the memory planner packs into an evidence plan and that claim verification matches generated claims against, so a span created at ingestion time is the unit that a citation in a generated answer ultimately points back to.

For format=json, the canonical text is NDJSON: one compact JSON rendering per input record, LF-joined, and each record's line is that record's (and all its child nodes') span, kind body.

Streaming multi-submit for large files¶

A single worker parse can exceed what fits comfortably in one gRPC message. Rather than requiring one giant submission, a worker may stream a job's parse as a sequence of partial submissions, each carrying:

One source version per stream. The first partial pins the version: it writes the source-version row (with canonical_text_hash still absent) and allocates a provisional blob_id. Every partial appends its canonical-text bytes as chunks under that blob id (each chunk ≤ 1 MiB). Spans and node provenance in every partial use global byte offsets into the accumulated text, all pinned to that one version.

Finalize. The first partial: false submission appends its (possibly empty) tail, hashes the complete accumulated canonical text with BLAKE3, and rewrites the version row with the final canonical_text_hash (and the fetch-format content_hash override, if the worker reports one). Streamed chains skip submit-time dedup entirely, because the canonical hash isn't known until finalize.

Checkpoints and crash recovery. Between partials, core persists an IngestContinuation checkpoint on the leased job row: the expected next seq, the pinned version's transaction time, the blob id, the accumulated length, the next chunk index, and the node ids the last partial created. Every accepted partial extends the job's lease. The checkpoint is fenced by worker id and lease generation: if a worker crashes mid-stream, the job is reaped and re-leased, the stale checkpoint is discarded, and the next attempt starts a clean stream (a new pinned version; the abandoned half-written version is simply orphaned, identifiable by a missing canonical_text_hash). Retrying the last acknowledged seq after a lost network ack replays idempotently: the same node ids come back, nothing is written twice.

sequenceDiagram
    autonumber
    participant W as Worker
    participant Core as telha-core

    W->>Core: Submission seq=0, partial=true (chunk 1 of text)
    Core->>Core: pin source version + blob_id, checkpoint next_seq=1
    Core-->>W: accepted
    W->>Core: Submission seq=1, partial=true (chunk 2 of text)
    Core->>Core: append to blob, checkpoint next_seq=2
    Core-->>W: accepted
    W->>Core: Submission seq=2, partial=false (final chunk)
    Core->>Core: hash full canonical text to canonical_text_hash
    Core->>Core: finalize source version row
    Core-->>W: job complete

This is a worker/gRPC-level concern; the REST /v1/ingest request that triggers a large-file job looks identical to any other asynchronous submission, the streaming happens between the worker and core over the gRPC worker protocol, invisible to the REST client except that the job takes longer to reach completed.

Related¶

• REST reference - every /v1 endpoint.
• Generation & traces - how ingested spans become cited evidence.
• Errors - the full error code reference.
• Concepts › Provenance & spans - why spans are the unit of proof.
• Architecture › Ingestion & provenance - the full storage model.
• Architecture › Format workers - how each format worker parses and projects.
• Connectors - SharePoint, Exchange, Slack, Salesforce, Entra ingestion.