Format workers¶

Architecture

Telha's core binary never parses a PDF, a spreadsheet, or a web page itself. It hands a job to a Python format worker over gRPC and receives back a graph projection: nodes, edges, spans, and canonical text. This page documents the worker protocol shared by every format, the IngestionResultBuilder that makes that protocol hard to misuse, and the five projections built on top of it.

Overview & purpose¶

Ingestion in Telha is deliberately split across a process boundary. The core engine (Rust, core-engine/) owns storage, temporal semantics, and the job queue; format-specific parsing (docling, polars, Playwright, tree-sitter, extract-msg) lives in Python workers under workers/, each a small package built on a shared harness, workers_common. A worker's only contract with core is the WorkerService gRPC surface: poll for work, submit a projection or a typed failure, heartbeat to keep a lease alive.

This split exists because the parsing ecosystems Telha depends on (docling's torch-backed stack, Playwright's Chromium, tree-sitter grammars) are heavyweight, version-churny, and not something a storage-and-temporal-semantics binary should link against. Keeping them out-of-process also means a worker crash or an OOM on a 500MB PDF never takes down the engine; the job simply re-enters the lease cycle.

Design¶

The WorkerService contract¶

service WorkerService {
  rpc PollJobs(PollJobsRequest) returns (stream JobPayload);
  rpc SubmitIngestionResult(IngestionResult) returns (SubmitResponse);
  rpc Heartbeat(HeartbeatRequest) returns (HeartbeatResponse);
}

PollJobs(worker_id, formats[]) is a long-poll stream, exempt from the normal request deadline; a worker declares the formats it can handle ("pdf", "web", "csv", …) and core routes matching jobs to it. Each JobPayload carries job_id, tenant_id, organization_id, kind, a JSON payload_json, a lease_deadline (µs), and attempts (the 0-based prior-failure counter). SubmitIngestionResult posts an IngestionResult{job_id, success, result_json, failure_reason, permanent} and gets back a SubmitResponse{accepted, node_ids, lease_deadline}; the returned node_ids are the logical ids core assigned to the nodes in that submission, in submission order, letting a streaming worker wire cross-partial edges. A background Heartbeat(worker_id, job_ids[]) extends leases for in-flight jobs every HEARTBEAT_INTERVAL_SECS = 60.

sequenceDiagram
    participant W as Worker (workers_common.WorkerConsumer)
    participant C as Core (WorkerService)
    W->>C: PollJobs(worker_id, formats)
    C-->>W: stream JobPayload
    loop every 60s while jobs in flight
        W->>C: Heartbeat(worker_id, job_ids)
        C-->>W: extended job_ids, new lease_deadline
    end
    W->>C: SubmitIngestionResult(result_json)
    C-->>W: SubmitResponse(accepted, node_ids, lease_deadline)

workers_common: one builder, five projections¶

Every worker constructs its submission through IngestionResultBuilder (workers_common/result.py) rather than hand-assembling JSON. The builder's job is to make the wire protocol hard to get wrong:

• add_span_text(text, kind=) / add_text(text) append canonical text and, for the former, register a byte-range span.
• set_canonical_text(text) + add_span_range(start, end, kind=) is the alternative for workers that already know their full text up front (tabular, code) and just need to carve spans out of it.
• add_node(labels, properties, spans=, valid_time=, id=) adds a node referencing previously added spans; an optional deterministic id lets core upsert instead of version-churn (email PERSON, code FILE).
• attach_span(node_ref, span_ref) links a span after the node exists.
• add_relationship(edge_type, from_ref=|from_id=, to_ref=|to_id=, properties=) adds an edge, addressing either a node in this submission (*_ref) or an existing node by id (*_id, needed for cross-job targets like email's IN_REPLY_TO).
• set_content_hash(hex) and add_child_job(format, name, options=, content=) support the fetch/fan-out formats (web recrawl dedup, email attachment routing, code manifest fan-out).
• add_ambiguity(node_ref, kind, candidates, detector_confidence=, applied_candidate_confidence=, severity_hint=) records a detected temporal ambiguity, gated on options.clarify.
• build() / to_json() produce the submission dict / UTF-8 JSON bytes for IngestionResult.result_json.

All text passes through normalize_canonical: NFC + LF-only line endings, applied once in the builder so byte offsets are stable across every worker and every platform. This is what lets a citation computed by one worker resolve identically regardless of which OS produced the source bytes.

builder = IngestionResultBuilder()
span = builder.add_span_text("Alpha owes Beta.", kind="body")
node = builder.add_node(["CONTRACT"], {"name": "Alpha"}, spans=[span])
payload = builder.build()

For streaming workers (tabular, code, large web crawls), a builder is constructed per partial with base_offset set to the previous partial's byte_length: span offsets stay global into the accumulated canonical text even though each partial's canonicalText carries only the newly appended slice, so partials concatenate byte-exactly server-side.

The consumer loop¶

WorkerConsumer.run() (workers_common/consumer.py) is the shared poll/dispatch/heartbeat loop every worker package runs unmodified. A handler is either a plain function returning bytes (single-shot submission) or a generator yielding submission dicts (ids = yield builder.build()) for multi-submit streaming: the consumer assigns seq, defaults each yielded dict to partial: True unless it explicitly sets partial: False (which finalizes the job), and feeds each partial's server-assigned node_ids back into the generator via send(). If the generator exhausts without an explicit final dict, the consumer submits an empty finalizer.

Failure taxonomy¶

class WorkerFailure(Exception):
    kind = "transient"
    permanent = False

Handlers raise a typed subclass; anything else escaping a handler is wrapped as transient. The wire failure_reason is "{kind}: {detail}", capped at 512 bytes.

Permanent kinds dead-letter immediately in core; retryable kinds re-enter the backoff cycle, bounded by jobs.max_attempts.

Data model¶

Wire submission shape¶

{
  "v": 1,
  "canonicalText": "...",
  "spans": [{"start": 0, "end": 17, "kind": "body"}],
  "nodes": [
    {"labels": ["CONTRACT"], "properties": {"name": "Alpha"}, "spanRefs": [0]}
  ],
  "relationships": [
    {"type": "REFERS_TO", "fromRef": 0, "toId": "...", "properties": {}}
  ],
  "stats": {"rows": 1200, "dirty_counts": {"date": 4}},
  "contentHash": "blake3-hex",
  "childJobs": [{"format": "docling", "name": "attachment:abc123", "contentB64": "..."}],
  "ambiguities": [
    {
      "nodeRef": 0,
      "kind": "conflicting_dates",
      "candidates": [{"label": "2026-05-01 (effective)", "validFrom": 1746057600000000}],
      "detectorConfidence": 0.7,
      "appliedCandidateConfidence": 0.55
    }
  ]
}

One subsection per projection¶

Algorithms & invariants¶

Determinism¶

flowchart LR
    BYTES["Identical input bytes"] --> PARSE["Deterministic parse<br/>(no wall clock, no randomness)"]
    PARSE --> PROJECT["Deterministic projection"]
    PROJECT --> HASH["Identical submission hash"]

Temporal ambiguity detection¶

Docling, tabular, and email each carry a §15 addendum defining document-specific rules for detecting temporally ambiguous facts (conflicting effective dates, relative-date phrases, missing dates, day-first/month-first cell readings). All three share the same contract:

Each rule produces one or more labeled candidates (best guess first), a detectorConfidence, and an appliedCandidateConfidence; core applies the first candidate's interval to the fact version. Every detector is deterministic: docling and email anchor relative-date phrases to document dates, never to "today," and tabular's day-first/month-first tie-break votes off unambiguous cells seen earlier in the same deterministic chunk order.

Streaming submission chaining¶

Configuration¶

HEARTBEAT_INTERVAL_SECS = 60 is a workers_common constant, not a per-worker config value.

As-built notes¶

Reconciled against each spec's §14 changelog:

• Tabular multi-submit streaming shipped (v0.3): the original v0.2 deviation (single submission per job, because core had no partial-submission protocol) was retired once the seq/partial protocol landed. Row spans are global offsets via base_offset chaining exactly as described above.
• Docling's failure-exit-code mapping and rlimit backstop are as specced in v0.2; RLIMIT_AS is POSIX-only, and the golden corpus is five docling-core fixtures rather than raw PDFs since the torch-backed parse stack is an optional extra.
• Web's URL_REF dedup is per-submission, not per-tenant, in v1 (v0.2 deviation): workers cannot query the graph, and true per-tenant dedup needs deterministic node identity/upsert, deferred.
• Email's heuristic threading tier is deferred (v0.2 pin 4): the worker sees one message per job and cannot query the graph for subject/time-window matches. subject_normalized is still emitted as a property so a future core-side pass can key on it. IN_REPLY_TO resolution instead relies on self-healing dangling edges to deterministic parent ids (pin 3), which the spec notes is simpler than pending-ref properties and satisfies retroactive resolution by construction.
• Email attachment child-job naming changed to content-hash-derived in v0.4 (PR-067), replacing the v0.2 message-scoped naming, so identical attachment bytes across any number of messages collapse into one shared source document.
• Code's cross-file CALLS resolution was not implemented (v0.2 pin 3): the spec's §3 aspiration was "same-ingestion resolvable," but because each file is its own child job without sibling file contents, only same-file resolution shipped. This is recorded as a v1 limitation, not silently dropped.
• Code's MODULE nodes are rust-only (v0.2 pin 4): a Python file's module identity is its FILE node; mod foo; projects as an IMPORTS reference, not a declaration.

Related¶

• Architecture › Job queue & leases
• Architecture › Ingestion & provenance
• Architecture › Clarification engine
• Architecture › gRPC contracts
• Concepts › Provenance & spans