Local Runner

The built-in Runner: it schedules Stages, keeping its state as files on Storage — no database (ADR-0010) — and streams records so memory stays bounded (ADR-0033).

runner

LocalRunner — single-node DAG execution with checkpoint/resume (ADR-0003, 0010).

Run state is files-on-Storage, no database: a run manifest, per-Stage checkpoint files, and a write-ahead log, all under <storage_uri>/_latence/runs/<run_id>/. Checkpoints are written batch-granularly and streamed to a temp file that is atomically published temp-write→rename (WAL discipline, ADR-0033), so the write transient is O(batch) not O(all-records) and a run killed mid-Stage leaves either the previous checkpoint or the complete new one — never a torn write. Re-running:

• a Stage whose checkpoint exists is skipped (resume);
• a fully-completed pipeline is a no-op.

Stages execute in topological order over their depends_on edges; for the S1 spine (Source→Parse→Export) the order is linear, but the DAG machinery is real.

RunRecords dataclass

RunRecords(
    runner: LocalRunner,
    outputs: dict[str, list[StageOutput]],
    storage: Storage,
    pipeline: Pipeline,
    run_id: str,
)

One run's record substrate for :class:~latence_core.stage_inputs.StageInputs (#69).

The live retained outputs map first, the Stage's on-disk checkpoint as the fallback — the two-source read that makes doc-level release transparent. Reads go through the engine's public :meth:LocalRunner.stage_records / :meth:LocalRunner.stage_record_count rather than around them, so a subclass that intercepts the checkpoint stream (the test seam) still sees every read, and a second Runner can construct one of these over its own substrate map.

records

records(stage_name: str) -> Iterator[StageOutput]

One Stage's output — from the live map if retained, else streamed from its checkpoint.

Source code in packages/latence-core/src/latence_core/runner.py

def records(self, stage_name: str) -> Iterator[StageOutput]:
    """One Stage's output — from the live map if retained, else streamed from its checkpoint."""
    return self.runner.stage_records(
        stage_name,
        self.outputs,
        storage=self.storage,
        pipeline=self.pipeline,
        run_id=self.run_id,
    )

count

count(stage_name: str) -> int

How many records one Stage produced — a len or the checkpoint's line count.

Source code in packages/latence-core/src/latence_core/runner.py

def count(self, stage_name: str) -> int:
    """How many records one Stage produced — a ``len`` or the checkpoint's line count."""
    return self.runner.stage_record_count(
        stage_name,
        self.outputs,
        storage=self.storage,
        pipeline=self.pipeline,
        run_id=self.run_id,
    )

carrier

carrier(stage: Stage) -> type[StageOutput] | type[Record]

The record type a Stage checkpoints (Record for a mixed-carrier Stage).

Source code in packages/latence-core/src/latence_core/runner.py

def carrier(self, stage: Stage) -> type[StageOutput] | type[Record]:
    """The record type a Stage checkpoints (``Record`` for a mixed-carrier Stage)."""
    return run_store.stage_output_type(stage, self.pipeline)

LocalRunner

LocalRunner(
    registry: ProviderRegistry | None = None,
    *,
    span_sink: SpanSink | None = None,
    lineage: OpenLineageEmitter | None = None
)

Executes a Pipeline on the local node with resumable file-based state.

Source code in packages/latence-core/src/latence_core/runner.py

def __init__(
    self,
    registry: ProviderRegistry | None = None,
    *,
    span_sink: SpanSink | None = None,
    lineage: OpenLineageEmitter | None = None,
) -> None:
    self._registry = registry or ProviderRegistry()
    # #185: the optional OpenLineage seam. ``None`` (the default) means OFF — no event is
    # built and no socket is opened, so a default run is byte- and behaviour-identical to a
    # pre-#185 run. An operator who points it at a lineage endpoint
    # (``build_lineage_emitter(url)``) gets per-Stage START/COMPLETE/FAIL RunEvents with the
    # Stage's input/output datasets + schema facets, in the standard vocabulary every catalog
    # (Marquez/Atlan/Collibra/Dataplex) already consumes. Counts only, never content; an
    # emission failure is logged and swallowed — lineage never fails a run.
    self._lineage = lineage
    # The run-scoped lineage handle, set for the duration of ``run()`` (mirrors the
    # ``_reuse_overlay`` pattern) so ``_run`` can emit per-Stage events without
    # threading a parameter through the whole executor. ``None`` ⇒ lineage off.
    self._lineage_run: RunLineage | None = None
    # #73: the optional tracing sink. ``None`` ⇒ each run's RunTracer defaults to the
    # silent-by-default LoggingSpanSink, so a normal run's output is unchanged; an operator
    # (or a downstream tracer, or a test) passes a CollectingSpanSink / their own SpanSink to
    # capture the per-Stage spans. Kept a thin seam — no OTel dep in core (ADR-0016).
    self._span_sink = span_sink
    # Per-Stage relation fan-out-guard tally, keyed by Stage name: (output-capped docs,
    # relations dropped [ADR-0033 / #63], input-capped docs, mentions dropped [#68]). A
    # RelationExtractor Provider exposes ``fanout_capped_documents``/
    # ``fanout_dropped_relations`` (output cap) and ``mention_capped_documents``/
    # ``mention_dropped_mentions`` (input cap) int attributes; the Runner reads them after
    # the Stage runs and folds them into the Quality Report's RelationQuality. A Provider
    # without them contributes (0, 0, 0, 0).
    self._fanout_by_stage: dict[str, tuple[int, int, int, int]] = {}
    # #69 + C6: when True (the default), a Stage's records are STREAMED from its Provider
    # straight into its checkpoint and never retained in the in-memory ``outputs`` dict at all
    # — the finalized checkpoint is the source of truth (ADR-0017) and every later reader
    # (a downstream Stage's gather, an accumulator's stream, the Quality Report) reads it back
    # batch-by-batch through :meth:`_stage_records`. That is what completes ADR-0033's bound:
    # the executor's per-Stage transient is O(checkpoint batch), not O(the Stage's records).
    # Setting it False materialises and retains every Stage's output (the pre-#69 behaviour) —
    # a parity/debug knob; the on-disk output, resume, and report are byte-identical either
    # way, only peak memory differs.
    self._release_doc_level_outputs = True
    # W17 (ADR-0043): the cross-run doc-level REUSE OVERLAY. A delta excludes an unchanged
    # document at Source (so its doc-level Stages never run) and hands the Runner that
    # document's persisted doc-level records; the Runner streams them into the CORPUS-LEVEL
    # accumulators (Profiling/Disambiguation/Graph) and the post-assembly per-chunk
    # transforms (Context Enrichment) / a doc-level Export ALONGSIDE the freshly-extracted
    # records, so those Stages see the whole corpus (reused ∪ fresh) exactly as if every
    # document had been re-extracted — while the doc-level Stages ran only on the changed
    # documents. The overlay is NEVER fed to a doc-level Stage (parse→…→redact) — those
    # gather via ``StageInputs.stream``/``gathered``, which are overlay-blind — so a reused
    # document is never
    # re-parsed/-extracted. Empty by default (``run(reuse=None)``), so a non-delta run is
    # byte-identical. WHERE the overlay folds, in what ORDER, and what happens to a carrier
    # no stream claims are the ADR-0061 §5 rule owned by
    # :class:`~latence_core.reuse_overlay.ReuseOverlay` (Wave 3-G).
    self._reuse_overlay = ReuseOverlay()
    # H-F1: the Quality Report half of the Runner, extracted into its own cohesive unit
    # (``quality_report_builder``). The engine composes it — passing itself as the
    # ``StageRecordReader`` (its public ``stage_records`` + ``read_findings``) and its live
    # ``_fanout_by_stage`` tally — so the report build is the same byte-identical logic, now a
    # relocatable module reused by both Runners through the public interface rather than via a
    # private reach-through. ``_build_report`` / ``_final_records`` below are thin delegators.
    self._report_builder = QualityReportBuilder(self, fanout_by_stage=self._fanout_by_stage)
    # C3: the Stage-execution module. Everything it takes to run ONE Stage — provider load,
    # device routing, skip-with-flag, the dispatch, the Provider tally drains (AFTER the
    # record stream is consumed), typed-error categorisation and the full StageMetrics
    # assembly — now lives behind ``execute(stage, ctx) -> StageOutcome``. This loop supplies
    # only scheduling + the files-on-Storage substrate; a second Runner supplies its own and
    # gets identical metrics for free, because it shares this object (see ``stage_execution``).
    self._stage_execution = StageExecution(
        self, self._registry, fanout_by_stage=self._fanout_by_stage
    )

quality_report_builder property

quality_report_builder: QualityReportBuilder

The engine's composed :class:~latence_core.quality_report_builder.QualityReportBuilder.

Public so a second Runner reuses the same report seam this engine uses — bound to this engine's live reader (stage_records / read_findings) and its _fanout_by_stage tally, so a report built through it is byte-identical to the local Runner's. This is the interface the Airflow adapter builds its report through, instead of reaching into the private _build_report / _final_records (the audit's THEME F fix, H-F1).

stage_execution property

stage_execution: StageExecution

The Stage-execution module every Runner runs its Stages through (C3).

Public because it IS the Runner seam: a second Runner (Airflow, Delta, an adopter's own) calls stage_execution.execute(stage, ctx) and gets device routing, the skip-with-flag posture, the Provider tally drains and a fully-populated :class:~latence_core.quality.StageMetrics — identical to this engine's, because it is the same code, not a re-implementation. Bound to this engine's registry, its dispatch and its live fan-out tally, so a Quality Report built off either Runner carries the same columns (ADR-0003: the identical Stages under a pluggable Runner).

stage_inputs

stage_inputs(
    stage: Stage,
    outputs: dict[str, list[StageOutput]],
    *,
    storage: Storage,
    pipeline: Pipeline,
    run_id: str
) -> StageInputs

This Stage's input, resolved against the Capability's declared plan (Wave 4-I).

The one call a Runner makes to answer "what does this Stage receive, how much of it is there, and is it wired correctly?" — replacing the executor's four-way shape branch and the per-shape counting rules. Public because it is the other half of the :class:~latence_core.stage_execution.StageContext a second Runner builds: an adapter hands its own substrate map in as outputs and gets the identical input semantics, rather than re-deriving which Capabilities stream and which gather.

Bound to this run's reuse overlay, so a delta's reused records fold in exactly where ADR-0061 §5 says they do.

Source code in packages/latence-core/src/latence_core/runner.py

def stage_inputs(
    self,
    stage: Stage,
    outputs: dict[str, list[StageOutput]],
    *,
    storage: Storage,
    pipeline: Pipeline,
    run_id: str,
) -> StageInputs:
    """This Stage's input, resolved against the Capability's declared plan (Wave 4-I).

    The one call a Runner makes to answer "what does this Stage receive, how much of it is
    there, and is it wired correctly?" — replacing the executor's four-way shape branch and the
    per-shape counting rules. Public because it is the other half of the
    :class:`~latence_core.stage_execution.StageContext` a second Runner builds: an adapter
    hands its own substrate map in as ``outputs`` and gets the identical input semantics,
    rather than re-deriving which Capabilities stream and which gather.

    Bound to this run's reuse overlay, so a delta's reused records fold in exactly where
    ADR-0061 §5 says they do.
    """
    return StageInputs(
        stage,
        pipeline,
        RunRecords(self, outputs, storage, pipeline, run_id),
        overlay=self._reuse_overlay,
    )

dispatch_stage

dispatch_stage(
    stage: Stage,
    provider: object,
    storage: Storage,
    pipeline: Pipeline,
    run_id: str,
    inputs: StageInputs,
) -> Iterator[StageOutput]

The :class:~latence_core.stage_execution.StageDispatch half this engine supplies.

An internal collaborator of :class:~latence_core.stage_execution.StageExecution, not part of its interface — a Runner adapter never calls this. Public only because the executor is a separate module and reaching a private name across it is exactly the reach-through this refactor removes.

Source code in packages/latence-core/src/latence_core/runner.py

def dispatch_stage(
    self,
    stage: Stage,
    provider: object,
    storage: Storage,
    pipeline: Pipeline,
    run_id: str,
    inputs: StageInputs,
) -> Iterator[StageOutput]:
    """The :class:`~latence_core.stage_execution.StageDispatch` half this engine supplies.

    An internal collaborator of :class:`~latence_core.stage_execution.StageExecution`, not part
    of its interface — a Runner adapter never calls this. Public only because the executor is a
    separate module and reaching a private name across it is exactly the reach-through this
    refactor removes.
    """
    return self._dispatch_stage(stage, provider, storage, pipeline, run_id, inputs)

run_dir

run_dir(pipeline: Pipeline, run_id: str) -> str

The run's state/artifact directory on Storage (<uri>/_latence/runs/<run_id>).

Public so a second Runner reusing this engine's report path can land the run's data artifacts where the shared report builder expects them, without reaching a private method (H-F1). The local Runner's own file-state (manifest, checkpoints, WAL) still lives here.

The layout itself is owned by :class:~latence_core.run_store.RunLayout (Wave 2-E) — every path method below delegates there, so this engine and every other consumer mint byte-identical locations from one decision.

Source code in packages/latence-core/src/latence_core/runner.py

def run_dir(self, pipeline: Pipeline, run_id: str) -> str:
    """The run's state/artifact directory on Storage (``<uri>/_latence/runs/<run_id>``).

    Public so a second Runner reusing this engine's report path can land the run's data
    artifacts where the shared report builder expects them, without reaching a private method
    (H-F1). The local Runner's own file-state (manifest, checkpoints, WAL) still lives here.

    The layout itself is owned by :class:`~latence_core.run_store.RunLayout` (Wave 2-E) —
    every path method below delegates there, so this engine and every other consumer mint
    byte-identical locations from one decision.
    """
    return self._layout(pipeline, run_id).run_dir

report_uri

report_uri(pipeline: Pipeline, run_id: str) -> str

The run's Quality Report location on Storage.

Public so a second Runner persists the report the shared :class:~latence_core.quality_report_builder.QualityReportBuilder produced to the same place an adopter reads it regardless of Runner — no private reach-through (H-F1).

Source code in packages/latence-core/src/latence_core/runner.py

def report_uri(self, pipeline: Pipeline, run_id: str) -> str:
    """The run's Quality Report location on Storage.

    Public so a second Runner persists the report the shared
    :class:`~latence_core.quality_report_builder.QualityReportBuilder` produced to the same
    place an adopter reads it regardless of Runner — no private reach-through (H-F1).
    """
    return self._layout(pipeline, run_id).report_uri

read_type_vocabulary

read_type_vocabulary(
    storage: Storage,
    pipeline: Pipeline,
    run_id: str,
    stage_name: str,
) -> TypeVocabulary | None

Read back a run's persisted canonical type vocabulary; None when none was written.

The seam an incremental delta run (T4) reads its predecessor's vocabulary through before seeding the next consolidation — so a delta aliases a new type INTO an existing canonical cluster instead of re-electing (and thereby renaming) one.

Source code in packages/latence-core/src/latence_core/runner.py

def read_type_vocabulary(
    self, storage: Storage, pipeline: Pipeline, run_id: str, stage_name: str
) -> TypeVocabulary | None:
    """Read back a run's persisted canonical type vocabulary; ``None`` when none was written.

    The seam an incremental delta run (T4) reads its predecessor's vocabulary through before
    seeding the next consolidation — so a delta aliases a new type INTO an existing canonical
    cluster instead of re-electing (and thereby renaming) one.
    """
    uri = self._type_vocabulary_uri(pipeline, run_id, stage_name)
    if not storage.exists(uri):
        return None
    return TypeVocabulary.model_validate_json(storage.read_text(uri))

read_findings

read_findings(
    storage: Storage,
    pipeline: Pipeline,
    run_id: str,
    stage_name: str,
) -> list[ScreeningFinding]

Read a Screening Stage's persisted findings sidecar (empty if none written).

Public because it is one half of the :class:~latence_core.quality_report_builder.StageRecordReader seam the shared :class:~latence_core.quality_report_builder.QualityReportBuilder reads to build the Screening rollup — the report builder no longer reaches into a private method for it (H-F1). The executor's own screening path uses the same reader.

Source code in packages/latence-core/src/latence_core/runner.py

def read_findings(
    self, storage: Storage, pipeline: Pipeline, run_id: str, stage_name: str
) -> list[ScreeningFinding]:
    """Read a Screening Stage's persisted findings sidecar (empty if none written).

    Public because it is one half of the
    :class:`~latence_core.quality_report_builder.StageRecordReader` seam the shared
    :class:`~latence_core.quality_report_builder.QualityReportBuilder` reads to build the
    Screening rollup — the report builder no longer reaches into a private method for it
    (H-F1). The executor's own screening path uses the same reader.
    """
    uri = self._findings_uri(pipeline, run_id, stage_name)
    if not storage.exists(uri):
        return []
    return [
        ScreeningFinding.model_validate_json(line)
        for line in storage.read_text(uri).splitlines()
        if line.strip()
    ]

stage_records

stage_records(
    stage_name: str,
    outputs: Mapping[str, list[StageOutput]],
    *,
    storage: Storage,
    pipeline: Pipeline,
    run_id: str
) -> Iterator[StageOutput]

Public reader half of the report-builder seam — delegates to :meth:_stage_records.

The shared :class:~latence_core.quality_report_builder.QualityReportBuilder (and a second Runner reusing it) reads each Stage's records back through this public method instead of the private _stage_records — the report concern is a real, relocatable unit that composes with the engine through its interface, never a # noqa: SLF001 reach-through (H-F1).

Source code in packages/latence-core/src/latence_core/runner.py

def stage_records(
    self,
    stage_name: str,
    outputs: Mapping[str, list[StageOutput]],
    *,
    storage: Storage,
    pipeline: Pipeline,
    run_id: str,
) -> Iterator[StageOutput]:
    """Public reader half of the report-builder seam — delegates to :meth:`_stage_records`.

    The shared :class:`~latence_core.quality_report_builder.QualityReportBuilder` (and a second
    Runner reusing it) reads each Stage's records back through this public method instead of the
    private ``_stage_records`` — the report concern is a real, relocatable unit that composes
    with the engine through its interface, never a ``# noqa: SLF001`` reach-through (H-F1).
    """
    yield from self._stage_records(
        stage_name, outputs, storage=storage, pipeline=pipeline, run_id=run_id
    )

stage_record_count

stage_record_count(
    stage_name: str,
    outputs: Mapping[str, list[StageOutput]],
    *,
    storage: Storage,
    pipeline: Pipeline,
    run_id: str
) -> int

How many records one Stage produced — from the live dict, else its checkpoint's lines.

The counting companion to :meth:stage_records, and public for the same reason: the input resolution (:class:~latence_core.stage_inputs.StageInputs) reads a streamed Stage's records_in through it rather than around it. A checkpoint is newline-delimited model_dump_json (one record per non-blank line, blank lines skipped on read), so counting lines equals counting decoded records — including a fused Stage's mixed-carrier checkpoint, where each line is still exactly one record. A Stage with no checkpoint (not run, or genuinely empty) counts 0.

Source code in packages/latence-core/src/latence_core/runner.py

def stage_record_count(
    self,
    stage_name: str,
    outputs: Mapping[str, list[StageOutput]],
    *,
    storage: Storage,
    pipeline: Pipeline,
    run_id: str,
) -> int:
    """How many records one Stage produced — from the live dict, else its checkpoint's lines.

    The counting companion to :meth:`stage_records`, and public for the same reason: the input
    resolution (:class:`~latence_core.stage_inputs.StageInputs`) reads a streamed Stage's
    ``records_in`` through it rather than around it. A checkpoint is newline-delimited
    ``model_dump_json`` (one record per non-blank line, blank lines skipped on read), so
    counting lines equals counting decoded records — including a fused Stage's mixed-carrier
    checkpoint, where each line is still exactly one record. A Stage with no checkpoint (not
    run, or genuinely empty) counts 0.
    """
    if stage_name in outputs:
        return len(outputs[stage_name])
    return run_store.count_checkpoint_records(
        storage, self._checkpoint_uri(pipeline, run_id, stage_name)
    )

scrub_sensitive

scrub_sensitive(value: object) -> object

Recursively redact sensitive config keys in a JSON-able manifest structure.

Walks the pipeline.model_dump(mode="json") tree and replaces the value of any mapping key that looks like a secret (see _SENSITIVE_KEY_TOKENS) with a fixed placeholder, keeping the key so the manifest still documents that a secret was configured — just not its value (issue #30). Lists and nested mappings are scrubbed in place-of-copy; scalars pass through. Env-var references the operator might legitimately keep (e.g. "$OPENAI_API_KEY") are redacted too — the manifest never needs the secret's value, referenced or inlined.

Source code in packages/latence-core/src/latence_core/runner.py

def scrub_sensitive(value: object) -> object:
    """Recursively redact sensitive config keys in a JSON-able manifest structure.

    Walks the ``pipeline.model_dump(mode="json")`` tree and replaces the *value* of
    any mapping key that looks like a secret (see ``_SENSITIVE_KEY_TOKENS``) with a
    fixed placeholder, keeping the key so the manifest still documents that a secret
    was configured — just not its value (issue #30). Lists and nested mappings are
    scrubbed in place-of-copy; scalars pass through. Env-var *references* the operator
    might legitimately keep (e.g. ``"$OPENAI_API_KEY"``) are redacted too — the
    manifest never needs the secret's value, referenced or inlined.
    """
    if isinstance(value, dict):
        scrubbed: dict[str, object] = {}
        for key, val in value.items():
            if isinstance(key, str) and _is_sensitive_key(key):
                scrubbed[key] = _REDACTED
            else:
                scrubbed[key] = scrub_sensitive(val)
        return scrubbed
    if isinstance(value, list):
        return [scrub_sensitive(item) for item in value]
    return value

topological_order

topological_order(stages: list[Stage]) -> list[Stage]

Kahn's algorithm over depends_on edges. Raises on a cycle.

Source code in packages/latence-core/src/latence_core/runner.py

def topological_order(stages: list[Stage]) -> list[Stage]:
    """Kahn's algorithm over ``depends_on`` edges. Raises on a cycle."""
    by_name = {s.name: s for s in stages}
    indegree = {s.name: len(s.depends_on) for s in stages}
    # Deterministic: process ready nodes in declared order.
    ready = [s.name for s in stages if indegree[s.name] == 0]
    order: list[str] = []
    while ready:
        name = ready.pop(0)
        order.append(name)
        for s in stages:
            if name in s.depends_on:
                indegree[s.name] -= 1
                if indegree[s.name] == 0:
                    ready.append(s.name)
    if len(order) != len(stages):
        remaining = sorted(set(by_name) - set(order))
        msg = f"Pipeline has a dependency cycle among stages: {remaining}"
        # A dependency cycle is a malformed-Pipeline config error (#73). ConfigError is a
        # ValueError subclass, so the pre-taxonomy ``except ValueError`` callers keep catching it.
        raise ConfigError(msg)
    return [by_name[n] for n in order]