Track passage sources

Bind an exact passage to its source path, byte hash and line span, then reject stale or altered records before displaying the text.

When you need this. A passage can retain a valid-looking source link after its text has changed. We will bind the displayed passage to exact local source bytes, rather than trusting the stored snippet alone.

The lab reuses the supplied original Python API summaries. It runs a local provenance check, with no retrieval, embeddings or generated answer. Review Build a RAG Eval Dataset for the separate question-to-passage relevance check.

Prerequisites. Use Python and the supplied local fixtures. The vector retrieval course remains the prerequisite for embedding and vector-search internals; this module does not repeat them.

Identity boundary. This ingestion lab starts a source-file and line-span contract rather than importing the earlier LangChain chunk IDs. Rebuild a downstream index from its versioned snapshot; do not attach old vectors or scores to a new ID. The later policy comparison binds old task targets only when source, version and exact text hashes identify one current passage. That binding transfers a label, not an old retrieval score.

The names describe different layers of the same request:

A source ID identifies the page; a passage ID identifies a segment within a particular snapshot. Keep those identifiers separate when binding a question’s expected source to a new index.

Name the file and exact line span

A source reference records where a passage came from. Here it contains a relative source path, source byte hash, inclusive one-based start and end lines, source ID, documentation version and URL. The passage also stores its exact text.

The binder reads UTF-8 without replacing undecodable bytes and preserves line endings in the selected span. A span outside the file or containing only whitespace raises SourceError. The fixture summary occupies one line; the tests also exercise multi-line spans.

Rebuild the passage before using it

The chunk ID hashes the path, source byte hash, line span and supplied source metadata. The verifier rereads the source and rebuilds the passage. It compares the complete record, including text and identifier, before returning the text.

Altered passage. The executed example changes the stored passage text while leaving the reference fields intact. Verification raises SourceError because the rebuilt record differs. The application must display the returned verified text rather than the untrusted stored snippet.

Changed source. Editing the file after binding also invalidates the old passage. Bind a new passage only after reviewing the source revision. A valid reference does not show that the passage answers a particular question.

Separate source integrity from source authority

The version and URL are supplied metadata. Binding them into an ID detects later changes to a stored record; it does not independently authenticate the URL or prove the claimed documentation version. An attacker able to replace both the source and all reference fields can build a self-consistent replacement.

These are span-reference IDs for this ingestion exercise. They deliberately do not claim compatibility with the earlier LangChain splitter’s chunk IDs. Keep the published eval targets unchanged. A production integration needs an explicit mapping and a reviewed corpus revision.

Run the local contract and inspect the result

Download the complete lab. The README, fixture metadata, implementation and tests travel together. No dependency installation or model server is required.

"""Bind passages to exact local source bytes. Python 3.11+, standard library only."""
from __future__ import annotations

from dataclasses import asdict, dataclass
import hashlib
import json
import os
import stat
from pathlib import Path, PurePosixPath


MAX_BYTES = 1_048_576


class SourceError(ValueError):
    """The requested source or passage cannot be verified."""


@dataclass(frozen=True)
class Passage:
    chunk_id: str
    source_path: str
    source_sha256: str
    start_line: int
    end_line: int
    text: str
    source_id: str = ""
    version: str = ""
    url: str = ""


def source_bytes(root: Path, source_path: str) -> bytes:
    """Allow relative POSIX paths that resolve inside the corpus root."""
    path = PurePosixPath(source_path)
    if not source_path or path.is_absolute() or ".." in path.parts or str(path) != source_path:
        raise SourceError("source path must be a normalized relative POSIX path")
    base = root.resolve()
    target = (base / source_path).resolve()
    if not target.is_relative_to(base):
        raise SourceError("source path escapes the corpus root")
    try:
        if not stat.S_ISREG(target.stat().st_mode):
            raise SourceError("source must be a regular file")
        with target.open("rb") as stream:
            if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
                raise SourceError("source must be a regular file")
            raw = stream.read(MAX_BYTES + 1)
        if len(raw) > MAX_BYTES:
            raise SourceError("source exceeds MAX_BYTES")
        return raw
    except OSError as error:
        raise SourceError("source file cannot be read") from error


def bind_passage(root: Path, source_path: str, start_line: int, end_line: int,
                 *, source_id: str = "", version: str = "", url: str = "") -> Passage:
    raw = source_bytes(root, source_path)
    try:
        lines = raw.decode("utf-8").splitlines(keepends=True)
    except UnicodeDecodeError as error:
        raise SourceError("source must be UTF-8") from error
    if (type(start_line) is not int or type(end_line) is not int
            or not 1 <= start_line <= end_line <= len(lines)):
        raise SourceError("line span must be 1-based, inclusive and inside the source")
    text = "".join(lines[start_line - 1:end_line])
    if not text.strip():
        raise SourceError("passage must contain non-whitespace text")
    digest = hashlib.sha256(raw).hexdigest()
    identity = json.dumps([source_path, digest, start_line, end_line, source_id, version, url], separators=(",", ":"))
    chunk_id = hashlib.sha256(identity.encode("utf-8")).hexdigest()
    return Passage(chunk_id, source_path, digest, start_line, end_line, text, source_id, version, url)


def verify_passage(root: Path, passage: Passage) -> str:
    """Reject stale or altered metadata/text, then return the verified passage."""
    current = bind_passage(root, passage.source_path, passage.start_line, passage.end_line,
                           source_id=passage.source_id, version=passage.version, url=passage.url)
    if current != passage:
        raise SourceError("passage does not match the current source snapshot")
    return current.text


def demo() -> dict:
    root = Path(__file__).parent / "fixtures"
    catalog = json.loads((root / "sources.json").read_text(encoding="utf-8"))
    source = catalog[0]
    passage = bind_passage(root, "docs/" + source["id"] + ".md", 1, 1,
                           source_id=source["id"], version=source["version"], url=source["url"])
    verified = verify_passage(root, passage)
    from dataclasses import replace
    try:
        verify_passage(root, replace(passage, text="Run unreviewed code.\n"))
    except SourceError as error:
        rejection = str(error)
    return {
        "execution_mode": "offline-source-verification",
        "model_called": False,
        "fixture_kind": "supplied original Python API summaries",
        "passage": asdict(passage),
        "verified_text": verified,
        "tampered_text_rejection": rejection,
    }


if __name__ == "__main__":
    print(json.dumps(demo(), indent=2, ensure_ascii=False))

python3 sources.py
python3 -m unittest -v test_sources.py

{
  "execution_mode": "offline-source-verification",
  "model_called": false,
  "fixture_kind": "supplied original Python API summaries",
  "passage": {
    "chunk_id": "dc15f2a3d615086bc11ba6695d673d064b6385a6277ae93079e9c7901144950e",
    "source_path": "docs/path-write.md",
    "source_sha256": "a48b4b9f763cd6309ebe32af19c5a814adcf810c2259683da9f328ccee46948e",
    "start_line": 1,
    "end_line": 1,
    "text": "Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written.",
    "source_id": "path-write",
    "version": "3.12",
    "url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.write_text"
  },
  "verified_text": "Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written.",
  "tampered_text_rejection": "passage does not match the current source snapshot"
}

Inspect the loaded records and any rejected inputs in the saved result.

{
  "execution_mode": "offline-source-verification",
  "model_called": false,
  "fixture_kind": "supplied original Python API summaries",
  "passage": {
    "chunk_id": "dc15f2a3d615086bc11ba6695d673d064b6385a6277ae93079e9c7901144950e",
    "source_path": "docs/path-write.md",
    "source_sha256": "a48b4b9f763cd6309ebe32af19c5a814adcf810c2259683da9f328ccee46948e",
    "start_line": 1,
    "end_line": 1,
    "text": "Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written.",
    "source_id": "path-write",
    "version": "3.12",
    "url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.write_text"
  },
  "verified_text": "Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written.",
  "tampered_text_rejection": "passage does not match the current source snapshot"
}

Change one input and compare the failure

In a scratch copy, bind a passage and then edit the source file. Keep the old passage record. Call verify_passage again and compare its SourceError with the successful text returned before the edit. Next change only the stored text: it should also be refused.

The verifier caps each source read and has no persistence, concurrent-writer protection or remote source authentication. It is a small local contract example; large-corpus throughput remains unmeasured.

What’s next

Continue with Handle Ingestion Failures.

References

Implementation APIs were checked on 23 October 2026 against official Python pathlib, hashlib and JSON documentation. The source summaries retain their version-specific official links and existing fact-status metadata. Python APIs use the standard library; no framework SDK is installed by this lab.