diff --git a/docs/api/utils.md b/docs/api/utils.md
index 50e80d1..d1e4002 100644
--- a/docs/api/utils.md
+++ b/docs/api/utils.md
@@ -22,7 +22,7 @@ Generates an **xxh64** digest (hex string, 16 characters) for arbitrary input, c
 def mkhash(x: Any) -> str
 ```
 
-The input is converted to a string and UTF-8 encoded before hashing. The digest is the content-addressed identity for sections: the per-section parquet store filename (`{hash}.parquet` in `.tablassert/store/`) and the section label in validation errors derive from it. User-facing progress labels truncate it to 8 characters for display.
+The input is converted to a string and UTF-8 encoded before hashing. `mkhash` is a general-purpose digest for arbitrary values, including section configuration; it is not itself a source-file hash. The build pipeline uses the section's `mkhash` value as one component of the build-time store key described below. Configuration-only validation uses this digest without reading source files; build-time progress and section-validation errors use the content-aware key, while source-file hashing errors retain the configuration digest as a fallback label. User-facing progress labels truncate the relevant key to 8 characters for display.
 
 The full 64-bit digest is used deliberately: a 32-bit hash would invite birthday collisions (~50% at ~77k sections) that could silently reuse another section's cached subgraph.
 
@@ -34,6 +34,38 @@ mkhash("hello")  # "26c7827d889f6da3"
 
 **Deterministic:** the same input always produces the same digest.
 
+## file_content_hash()
+
+Generates an **xxh64** digest of a local source file's raw bytes. It resolves the path and raises `SourceFileError` if the path does not exist, is not a regular file, or cannot be read.
+
+```python
+def file_content_hash(
+    path: Path,
+    *,
+    config: Path | None = None,
+    section_label: str | None = None,
+) -> str
+```
+
+This helper does not cache the digest. An edited source file therefore produces a different digest and, in a build, a different section store key. `config` and `section_label` are optional context included in a source-file error when available.
+
+## section_store_key()
+
+Returns the build-time content-aware store key for a section's cached parquet.
+
+```python
+def section_store_key(
+    section: Any,
+    local: Path | None = None,
+    *,
+    content_digest: str | None = None,
+) -> str
+```
+
+For a section with a local source, the key combines the section configuration digest with the source content digest and hashes the result into one 16-character xxh64 value. Callers may provide a precomputed `content_digest` to avoid rereading the file; otherwise `local` is passed to `file_content_hash`. If neither is provided, the function returns the section's plain `mkhash` value. The build pipeline uses the content-aware form for local sources, so changing one source invalidates the cached parquet for each section that reads it without invalidating sections backed by unchanged sources.
+
+The key formula is `mkhash(f"{mkhash(section)}:{content_digest}")`. Build-mode flags add their own filename suffixes (`.head`, `.release`, or `.qc`) after the key; they do not change the digest formula.
+
 ---
 
 ## namespace_uuid()  *(Rust extension: `tablassert.rs`)*
diff --git a/docs/cli.md b/docs/cli.md
index 50672fe..8609b16 100644
--- a/docs/cli.md
+++ b/docs/cli.md
@@ -185,8 +185,11 @@ identical at any worker count.
 
 Output is written to `rig.artifact_base_path` (created when missing) as `{name}_{version}.nodes.ndjson`,
 `{name}_{version}.edges.ndjson`, and `{name}_{version}.RIG.yaml`; intermediate parquet lands in
-`.tablassert/store/`. The RIG document is audited in memory before it is written: an invalid
-or incomplete RIG fails the build with `[rig-validation-failed]` and nothing is emitted. See
+`.tablassert/store/`. Build-time store keys incorporate the source-file content, so editing a source
+file rebuilds exactly the sections that read the changed content. The first build after this key-format
+upgrade rebuilds every section once because the keys have changed; orphaned old parquet files are not
+automatically deleted. The RIG document is audited in memory before it is written: an invalid or
+incomplete RIG fails the build with `[rig-validation-failed]` and nothing is emitted. See
 [Graph Configuration](configuration/graph.md).
 
 ??? info "Build progress & stages"
diff --git a/rust/src/lib.rs b/rust/src/lib.rs
index b599994..becbd32 100644
--- a/rust/src/lib.rs
+++ b/rust/src/lib.rs
@@ -1,3 +1,4 @@
+use pyo3::exceptions::PyOSError;
 use pyo3::prelude::*;
 
 // mimalloc returns memory to the OS far better than glibc malloc under heavy
@@ -5,7 +6,10 @@ use pyo3::prelude::*;
 // making millions of small String/Vec allocations).  Without it, per-thread
 // malloc arenas retain freed memory and inflate peak RSS several-fold.
 use mimalloc::MiMalloc;
-use xxhash_rust::xxh64::xxh64 as xxh64_digest;
+use std::fs::File;
+use std::io::Read;
+use std::path::PathBuf;
+use xxhash_rust::xxh64::{xxh64 as xxh64_digest, Xxh64};
 
 #[global_allocator]
 static GLOBAL: MiMalloc = MiMalloc;
@@ -26,6 +30,27 @@ fn xxh64(data: &str) -> String {
     format!("{:016x}", xxh64_digest(data.as_bytes(), 0))
 }
 
+/// XXH64 hex digest (seed 0) of a file's raw bytes, streamed in fixed-size chunks.
+#[pyfunction]
+fn xxh64_file(path: PathBuf) -> PyResult<String> {
+    const CHUNK_SIZE: usize = 8 * 1024 * 1024;
+
+    let mut file = File::open(&path)
+        .map_err(|error| PyOSError::new_err(format!("{}: {}", path.display(), error)))?;
+    let mut hasher = Xxh64::new(0);
+    let mut chunk = vec![0; CHUNK_SIZE];
+    loop {
+        let bytes_read = file
+            .read(&mut chunk)
+            .map_err(|error| PyOSError::new_err(format!("{}: {}", path.display(), error)))?;
+        if bytes_read == 0 {
+            break;
+        }
+        hasher.update(&chunk[..bytes_read]);
+    }
+    Ok(format!("{:016x}", hasher.digest()))
+}
+
 // Public re-exports of the fullmap `#[pyfunction]`s so Rust integration tests
 // (`rust/tests/`) and other non-Python embedders can drive the exact production
 // build/read path.  The `fullmap` module itself stays private; only these
@@ -48,12 +73,13 @@ fn rs(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add_function(wrap_pyfunction!(ndjson::dedup_ndjson, module)?)?;
     module.add_function(wrap_pyfunction!(uuid::namespace_uuid, module)?)?;
     module.add_function(wrap_pyfunction!(xxh64, module)?)?;
+    module.add_function(wrap_pyfunction!(xxh64_file, module)?)?;
     Ok(())
 }
 
 #[cfg(test)]
 mod tests {
-    use super::xxh64;
+    use super::{xxh64, xxh64_file};
 
     #[test]
     fn xxh64_matches_known_digests() {
@@ -63,4 +89,33 @@ mod tests {
         assert_eq!(xxh64("hello"), "26c7827d889f6da3");
         assert_eq!(xxh64(""), "ef46db3751d8e999");
     }
+
+    #[test]
+    fn xxh64_file_matches_known_digests() {
+        // WHY: pin file hashing to the same seed-0 digest as the string primitive,
+        // including the empty-file identity value and UTF-8 bytes.
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("hello");
+        std::fs::write(&path, b"hello").unwrap();
+        assert_eq!(xxh64_file(path).unwrap(), "26c7827d889f6da3");
+
+        let empty = dir.path().join("empty");
+        std::fs::write(&empty, b"").unwrap();
+        assert_eq!(xxh64_file(empty).unwrap(), "ef46db3751d8e999");
+    }
+
+    #[test]
+    fn xxh64_file_reports_io_failures_as_os_error() {
+        // WHY: callers must distinguish I/O failures from valid digests and see
+        // the offending path, including for missing files and directories.
+        let dir = tempfile::tempdir().unwrap();
+        let missing = dir.path().join("missing");
+        let error = xxh64_file(missing.clone()).unwrap_err();
+        assert!(error.to_string().contains(&missing.display().to_string()));
+
+        let error = xxh64_file(dir.path().to_path_buf()).unwrap_err();
+        assert!(error
+            .to_string()
+            .contains(&dir.path().display().to_string()));
+    }
 }
diff --git a/src/tablassert/cli.py b/src/tablassert/cli.py
index 6292f88..0db1473 100644
--- a/src/tablassert/cli.py
+++ b/src/tablassert/cli.py
@@ -75,6 +75,44 @@ def _section_store_path(h: str, head: bool = False, release: bool = False, qc: b
     return STORE / f"{h}{suffix}.parquet"
 
 
+def _section_store_key_for_build(
+    section: dict[str, Any], configuration_file: Path, content_hashes: dict[tuple[Path, int, int], str]
+) -> tuple[str, str]:
+    """Return a section's content-aware store key and config-only fallback label.
+
+    The memo belongs to one ``build_graph_pipeline`` call and is invalidated by any
+    ``(resolved path, mtime_ns, size)`` change. Progress and validation errors use the
+    content-aware key; the returned config-only hash labels ``SourceFileError`` when
+    hashing fails before a content-aware key can exist.
+    """
+    from tablassert.utils import file_content_hash, mkhash, section_store_key
+
+    config_hash: str = mkhash(section)
+    source: object = section.get("source")
+    local: object = source.get("local") if isinstance(source, dict) else None
+    if local is None:
+        return section_store_key(section), config_hash
+
+    resolved: Path = Path(str(local)).resolve()
+    config_value: object = section.get("config", "section")
+    section_label: str = f"{Path(str(config_value)).stem} · {config_hash[:8]}"
+    digest: str | None = None
+    try:
+        stat = resolved.stat()
+    except OSError:
+        # Reuse the contextual helper so missing/unreadable paths never leak a raw OS error.
+        digest = file_content_hash(resolved, config=configuration_file, section_label=section_label)
+        return section_store_key(section, content_digest=digest), config_hash
+
+    signature: tuple[Path, int, int] = (resolved, stat.st_mtime_ns, stat.st_size)
+    if digest is None:
+        digest = content_hashes.get(signature)
+    if digest is None:
+        digest = file_content_hash(resolved, config=configuration_file, section_label=section_label)
+        content_hashes[signature] = digest
+    return section_store_key(section, content_digest=digest), config_hash
+
+
 def _load_table_indexed(args: tuple[int, Path]) -> tuple[int, object]:
     """Load one table, tagged with its input index (multiprocessing worker).
 
@@ -184,7 +222,6 @@ def build_graph_pipeline(
     from tablassert.fullmap import fullmap_db_path
     from tablassert.lib import Tcode, compile_graph, compile_subgraph
     from tablassert.progress import flatten_pydantic_error, format_section_compact
-    from tablassert.utils import mkhash
 
     # Stage 1/6: load tables.
     progress.stage("Loading Tables")
@@ -223,8 +260,11 @@ def build_graph_pipeline(
     progress.stage("Building TCode")
     start, advance, _ = progress.section_loop(n, "TCode")
     tcode: list[Tcode] = []
+    # This memo is intentionally scoped to one build: a changed stat signature re-hashes,
+    # while repeated sections pointing to the same unchanged file hash only once.
+    content_hashes: dict[tuple[Path, int, int], str] = {}
     for s in sections:
-        h: str = mkhash(s)
+        h, _ = _section_store_key_for_build(s, configuration_file, content_hashes)
         start(f"{Path(str(s['config'])).stem} · {h[:8]}")
         # Mode flags change the cached parquet's content, so each combination caches
         # to a distinct file and can never quick-exit another mode's build.
diff --git a/src/tablassert/errors.py b/src/tablassert/errors.py
index 49ee188..c2fcf49 100644
--- a/src/tablassert/errors.py
+++ b/src/tablassert/errors.py
@@ -12,6 +12,7 @@
     "graph-validation-failed",
     "section-validation-failed",
     "babel-download-failed",
+    "source-file-unreadable",
     "resolve-bad-specs",
     "config-rows-and-row-slice-conflict",
     "comparison-bad-comparator-type",
@@ -185,3 +186,24 @@ def __init__(self, url: str, retries: int, last_error: BaseException) -> None:
             f"BABEL download failed after {retries} attempts: {url} (last error: {last_error}). Check network connectivity or pin a different BABEL version.",
             code="babel-download-failed",
         )
+
+
+class SourceFileError(TablassertError):
+    """A table section's ``source.local`` file could not be read for content-hashing.
+
+    Notes:
+        ``config`` and ``section_label`` are optional keyword context: the utils layer
+        that first notices the failure often lacks build context, but callers that have
+        it (the CLI's section loop) should pass them so the message names the exact
+        table config and section label (config stem + 8-char section digest).
+    """
+
+    def __init__(self, path: Path, detail: str, *, config: Path | None = None, section_label: str | None = None) -> None:
+        context: str = ""
+        if config is not None and section_label is not None:
+            context = f" (config {config}, section {section_label})"
+        elif config is not None:
+            context = f" (config {config})"
+        elif section_label is not None:
+            context = f" (section {section_label})"
+        super().__init__(f"Source file unreadable{context}: {path} — {detail}", code="source-file-unreadable")
diff --git a/src/tablassert/rs.pyi b/src/tablassert/rs.pyi
index 87c6571..911b152 100644
--- a/src/tablassert/rs.pyi
+++ b/src/tablassert/rs.pyi
@@ -19,3 +19,4 @@ def hydrate_sources(db: Path) -> list[str]: ...
 def lookup_fullmap_terms(db: Path, terms: list[str], return_format: str = "rows") -> list[dict[str, Any]]: ...
 def namespace_uuid(domain: str, values: list[str]) -> str: ...
 def xxh64(data: str) -> str: ...
+def xxh64_file(path: str) -> str: ...
diff --git a/src/tablassert/utils.py b/src/tablassert/utils.py
index bba7b18..fa82a1b 100644
--- a/src/tablassert/utils.py
+++ b/src/tablassert/utils.py
@@ -4,6 +4,7 @@
 from typing import Any
 
 from tablassert import rs
+from tablassert.errors import SourceFileError
 
 BASE: Path = Path("./.tablassert")
 STORE: Path = BASE / "store"
@@ -16,3 +17,37 @@ def mkhash(x: Any) -> str:
     # birthday collisions (~50% at ~77k sections) that would silently reuse another
     # section's cached subgraph; the full 16-hex 64-bit digest avoids that.
     return rs.xxh64(str(x))
+
+
+def file_content_hash(path: Path, *, config: Path | None = None, section_label: str | None = None) -> str:
+    """XXH64 hex digest of a source file's raw bytes, or a loud :class:`SourceFileError`.
+
+    Pure: no caching, so the digest always reflects the file's bytes at call time and an
+    edited source always yields a new section store key. ``config`` / ``section_label``
+    are optional build context forwarded to the error message when available.
+    """
+    resolved: Path = path.resolve()
+    if not resolved.exists():
+        raise SourceFileError(resolved, "no such file", config=config, section_label=section_label)
+    if not resolved.is_file():
+        raise SourceFileError(resolved, "not a regular file", config=config, section_label=section_label)
+    try:
+        return rs.xxh64_file(str(resolved))
+    except OSError as e:
+        raise SourceFileError(resolved, str(e), config=config, section_label=section_label) from e
+
+
+def section_store_key(section: Any, local: Path | None = None, *, content_digest: str | None = None) -> str:
+    """Content-addressed identity for a section's cached parquet store.
+
+    With a ``local`` source file or a precomputed ``content_digest`` the key mixes the
+    section config hash with the file's content digest, so editing the source file
+    invalidates the cache; without one the key is the plain section hash. Always a
+    single 16-hex xxh64 digest. ``content_digest`` lets a caller memoize file reads
+    without changing the key formula.
+    """
+    if content_digest is None:
+        if local is None:
+            return mkhash(section)
+        content_digest = file_content_hash(local)
+    return mkhash(f"{mkhash(section)}:{content_digest}")
diff --git a/tests/bench_file_hash_bench.py b/tests/bench_file_hash_bench.py
new file mode 100644
index 0000000..3c4729e
--- /dev/null
+++ b/tests/bench_file_hash_bench.py
@@ -0,0 +1,141 @@
+"""Opt-in benchmark harness for streaming source-file hashing.
+
+WHY: content-aware Stage 3 keys prevent stale cached builds, but the source file must
+be read cheaply enough that correctness does not become a build bottleneck. This
+harness measures the Rust XXH64 pass against a same-file chunked read floor and
+measures the actual Stage-3-style key derivation overhead against the former
+config-only ``mkhash(section)`` key.
+
+Usage (opt-in; this module is skipped before test collection by default)::
+
+    TABLASSERT_BENCH=1 uv run pytest tests/bench_file_hash_bench.py -s -n 0 -q
+
+The one-gigabyte CSV is generated in pytest's session temporary directory using a
+seeded, repeatable row pattern. Generation is streamed in 8 MiB blocks and the
+benchmark performs one run of each measurement.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.skipif(os.environ.get("TABLASSERT_BENCH") != "1", reason="benchmark harness; opt in with TABLASSERT_BENCH=1")
+
+if os.environ.get("TABLASSERT_BENCH") != "1":
+    pytest.skip("benchmark harness; opt in with TABLASSERT_BENCH=1", allow_module_level=True)
+
+_SEED = 6006
+_GIB = 1 << 30
+_CHUNK_BYTES = 8 * 1024 * 1024
+_FIXTURE_BYTES = 4 * 1024 * 1024
+
+
+def _seeded_csv_block() -> bytes:
+    """Return one deterministic 8 MiB block made from a seeded CSV row."""
+    import random
+
+    rng = random.Random(_SEED)
+    fields = [f"field_{index}_{rng.randrange(1_000_000_000)}" for index in range(8)]
+    row = ("subject,predicate,value," + ",".join(fields) + "\n").encode()
+    return (row * ((_CHUNK_BYTES // len(row)) + 1))[:_CHUNK_BYTES]
+
+
+def _write_seeded_csv(path: Path, size: int) -> int:
+    """Stream exactly ``size`` bytes of a repeatable CSV-shaped artifact."""
+    block = _seeded_csv_block()
+    written = 0
+    with path.open("wb") as handle:
+        while written < size:
+            chunk_size = min(len(block), size - written)
+            handle.write(block[:chunk_size])
+            written += chunk_size
+    return written
+
+
+@pytest.fixture(scope="session")
+def benchmark_files(tmp_path_factory: pytest.TempPathFactory) -> tuple[Path, Path]:
+    """Create the large throughput artifact and the small Stage 3 fixture once."""
+    root = tmp_path_factory.mktemp("file_hash_bench")
+    large = root / "seeded_1gib.csv"
+    fixture = root / "stage3_fixture.csv"
+    assert _write_seeded_csv(large, _GIB) == _GIB
+    assert _write_seeded_csv(fixture, _FIXTURE_BYTES) == _FIXTURE_BYTES
+    return large, fixture
+
+
+def _time_chunked_read(path: Path) -> tuple[float, int]:
+    """Read a file in the hashing primitive's chunk size and return time and bytes."""
+    started = time.perf_counter()
+    read_bytes = 0
+    with path.open("rb") as handle:
+        while chunk := handle.read(_CHUNK_BYTES):
+            read_bytes += len(chunk)
+    return time.perf_counter() - started, read_bytes
+
+
+def _gib_per_second(size: int, elapsed: float) -> float:
+    """Convert a byte count and wall time into binary GiB/s."""
+    return size / _GIB / elapsed
+
+
+def _stage3_key_timings(source: Path) -> tuple[float, float, list[str], list[str]]:
+    """Measure content-aware Stage 3 keys against config-only keys."""
+    from tablassert.cli import _section_store_key_for_build
+    from tablassert.utils import mkhash
+
+    sections = [{"config": Path("table.yaml"), "marker": f"section-{index}", "source": {"local": str(source)}} for index in range(8)]
+    memo: dict[tuple[Path, int, int], str] = {}
+    started = time.perf_counter()
+    content_keys = [_section_store_key_for_build(section, Path("graph.yaml"), memo)[0] for section in sections]
+    content_elapsed = time.perf_counter() - started
+
+    started = time.perf_counter()
+    config_keys = [mkhash(section) for section in sections]
+    config_elapsed = time.perf_counter() - started
+    return content_elapsed, config_elapsed, content_keys, config_keys
+
+
+def test_bench_file_hash(benchmark_files: tuple[Path, Path]) -> None:
+    """Measure file hashing, the raw-read floor, and Stage 3 key derivation.
+
+    WHY: a measured one-run result makes the performance tradeoff visible without
+    imposing a multi-gigabyte workload on normal CI, which skips this module entirely.
+    """
+    from tablassert import rs
+
+    large, fixture = benchmark_files
+    size = large.stat().st_size
+    assert size == _GIB, "benchmark artifact must be exactly 1 GiB"
+
+    started = time.perf_counter()
+    digest = rs.xxh64_file(str(large))
+    hash_elapsed = time.perf_counter() - started
+    assert large.stat().st_size == size
+    assert len(digest) == 16
+    assert all(character in "0123456789abcdef" for character in digest)
+
+    read_elapsed, read_bytes = _time_chunked_read(large)
+    assert read_bytes == size, "pure-read floor must exercise the complete benchmark artifact"
+
+    content_elapsed, config_elapsed, content_keys, config_keys = _stage3_key_timings(fixture)
+    assert len(content_keys) == len(config_keys) == 8
+    assert all(len(key) == 16 for key in content_keys + config_keys)
+    assert content_keys != config_keys, "content-aware keys must differ from config-only keys"
+    delta_ms_per_file = (content_elapsed - config_elapsed) * 1000 / len(content_keys)
+
+    print(
+        f"\nxxh64_file: size={size:,} bytes ({size / _GIB:.3f} GiB) digest={digest} "
+        f"wall_time={hash_elapsed:.3f}s throughput={_gib_per_second(size, hash_elapsed):.3f} GiB/s"
+    )
+    print(
+        f"pure read: bytes={read_bytes:,} ({read_bytes / _GIB:.3f} GiB) "
+        f"wall_time={read_elapsed:.3f}s throughput={_gib_per_second(read_bytes, read_elapsed):.3f} GiB/s"
+    )
+    print(
+        f"stage 3 keys: sections={len(content_keys)} shared_files=1 content_aware={content_elapsed * 1000:.3f}ms "
+        f"config_only_mkhash={config_elapsed * 1000:.3f}ms delta={delta_ms_per_file:.3f}ms/file"
+    )
diff --git a/tests/test_errors.py b/tests/test_errors.py
index 4a1466e..3c35c35 100644
--- a/tests/test_errors.py
+++ b/tests/test_errors.py
@@ -2,7 +2,7 @@
 
 from pathlib import Path
 
-from tablassert.errors import DOCS_URL, BabelDownloadError, GraphValidationError, QcRuntimeMissingError, SectionValidationError
+from tablassert.errors import DOCS_URL, BabelDownloadError, GraphValidationError, QcRuntimeMissingError, SectionValidationError, SourceFileError
 
 
 def test_qc_runtime_missing_error_code_and_docs_url() -> None:
@@ -47,3 +47,31 @@ def test_babel_download_error_code_and_docs_url() -> None:
     err: BabelDownloadError = BabelDownloadError("https://stars.renci.org/var/babel_outputs/x.gz", 5, RuntimeError("network down"))
     assert err.code == "babel-download-failed"
     assert str(err).endswith(DOCS_URL + "babel-download-failed")
+
+
+def test_source_file_error_carries_config_section_and_path() -> None:
+    """Guard: an unreadable source file names the config, the section, and the path.
+
+    The `source-file-unreadable` code plus docs URL tells a user which section's
+    ``source.local`` could not be hashed, and the message must carry the table config
+    path, the section label, and the offending path so they can find it without a
+    debugger.
+    """
+    err: SourceFileError = SourceFileError(Path("/data/missing.csv"), "no such file", config=Path("table.yaml"), section_label="table · 0123abcd")
+    assert err.code == "source-file-unreadable"
+    assert str(err).endswith(DOCS_URL + "source-file-unreadable")
+    assert "table.yaml" in str(err)
+    assert "table · 0123abcd" in str(err)
+    assert "/data/missing.csv" in str(err)
+
+
+def test_source_file_error_without_build_context() -> None:
+    """Guard: the error is still informative when raised without build context.
+
+    The utils layer that first notices the failure often lacks the config path and
+    section label; the message must still name the offending path and the OS detail.
+    """
+    err: SourceFileError = SourceFileError(Path("orphan.csv"), "permission denied")
+    assert err.code == "source-file-unreadable"
+    assert "orphan.csv" in str(err)
+    assert "permission denied" in str(err)
diff --git a/tests/test_rs.py b/tests/test_rs.py
index da62096..c703a30 100644
--- a/tests/test_rs.py
+++ b/tests/test_rs.py
@@ -3,12 +3,95 @@
 import itertools
 import json
 import random
+import time
 import uuid
 from collections.abc import Iterable
 from pathlib import Path
 from typing import Any
 from uuid import UUID
 
+import pytest
+
+
+def test_xxh64_file_matches_known_digests(tmp_path: Path) -> None:
+    """xxh64_file returns the same seed-0 digest as the in-memory primitive.
+
+    WHY: the streaming file primitive is a drop-in byte equivalent of the string
+    hash, so the pinned hello/empty digests catch any regression in seed, chunking,
+    or formatting.
+    """
+    from tablassert import rs
+
+    hello: Path = tmp_path / "hello.txt"
+    hello.write_bytes(b"hello")
+    assert rs.xxh64_file(str(hello)) == "26c7827d889f6da3"
+
+    empty: Path = tmp_path / "empty.bin"
+    empty.write_bytes(b"")
+    assert rs.xxh64_file(str(empty)) == "ef46db3751d8e999"
+
+
+def test_xxh64_file_binary_and_chunk_boundary(tmp_path: Path) -> None:
+    """Binary content and multi-chunk reads hash exactly.
+
+    WHY: chunking must preserve byte order and content (including NUL bytes and
+    non-text data), and the 8 MiB + 1 boundary exercises the streaming loop across
+    multiple reads instead of only small files. Expected digests are pinned from
+    the Python `xxhash` library (the independent oracle the Rust primitive
+    replaced, not installed in the dev env): binary=0915d8f748ad915d,
+    boundary=1a11076e494e1d0b.
+    """
+    from tablassert import rs
+
+    binary: Path = tmp_path / "binary.bin"
+    binary.write_bytes(b"\x00\xffbinary\n")
+    assert rs.xxh64_file(str(binary)) == "0915d8f748ad915d"
+
+    boundary: Path = tmp_path / "boundary.bin"
+    boundary.write_bytes(b"\x5a" * (8 * 1024 * 1024 + 1))
+    assert rs.xxh64_file(str(boundary)) == "1a11076e494e1d0b"
+
+
+def test_file_hash_smoke_xxh64_file(tmp_path: Path) -> None:
+    """A generated 4 MiB file hashes within the deliberately loose CI ceiling.
+
+    WHY: the opt-in GiB benchmark is skipped in normal CI, so this non-gated smoke
+    catches pathological regressions such as byte-at-a-time file reads without making
+    the default suite depend on benchmark-scale I/O.
+    """
+    from tablassert import rs
+
+    source: Path = tmp_path / "4mib.bin"
+    size = 4 * 1024 * 1024
+    pattern = b"tablassert-file-hash-smoke\n"
+    source.write_bytes(pattern * (size // len(pattern)) + pattern[: size % len(pattern)])
+    started = time.perf_counter()
+    digest = rs.xxh64_file(str(source))
+    elapsed = time.perf_counter() - started
+
+    assert source.stat().st_size == size
+    assert len(digest) == 16
+    assert all(character in "0123456789abcdef" for character in digest)
+    assert elapsed < 10, f"xxh64_file took {elapsed:.3f}s for a 4 MiB file"
+
+
+def test_xxh64_file_missing_and_directory_errors(tmp_path: Path) -> None:
+    """Missing files and directories raise OSError naming the path.
+
+    WHY: the API has no sentinel digest; every I/O failure must be loud and
+    traceable to the caller-provided path.
+    """
+    from tablassert import rs
+
+    missing: Path = tmp_path / "missing.bin"
+    with pytest.raises(OSError, match=str(missing)):
+        rs.xxh64_file(str(missing))
+
+    directory: Path = tmp_path / "dir"
+    directory.mkdir()
+    with pytest.raises(OSError, match=str(directory)):
+        rs.xxh64_file(str(directory))
+
 
 def test_namespace_uuid_returns_uuid() -> None:
     """namespace_uuid returns a UUID shaped string."""
diff --git a/tests/test_store_invalidation_e2e.py b/tests/test_store_invalidation_e2e.py
new file mode 100644
index 0000000..2a0d22e
--- /dev/null
+++ b/tests/test_store_invalidation_e2e.py
@@ -0,0 +1,147 @@
+"""End-to-end coverage for content-aware section-store invalidation."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import polars as pl
+import pytest
+
+from tablassert import rs
+from tablassert.cli import build_pipeline
+from tablassert.ingests import from_yaml, to_sections, to_yaml
+from tablassert.progress import PipelineProgress
+from tablassert.utils import mkhash
+
+
+def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> Path:
+    path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")
+    return path
+
+
+def _build_real_redb(root: Path) -> Path:
+    """Build the tiny real fullmap used by these invalidation smokes."""
+    root.mkdir(parents=True, exist_ok=True)
+    classes = _write_jsonl(root / "classes.ndjson", [{"id": "HGNC:1100", "equivalent_identifiers": [{"identifier": "NCBIGene:672"}]}])
+    synonyms = _write_jsonl(
+        root / "synonyms.ndjson",
+        [
+            {"curie": "HGNC:1100", "preferred_name": "BRCA1", "names": ["BRCA1", "brca1"], "types": ["Gene"], "taxa": ["NCBITaxon:9606"]},
+            {"curie": "HGNC:6871", "preferred_name": "MAPK1", "names": ["MAPK1", "mapk1"], "types": ["Gene"], "taxa": ["NCBITaxon:9606"]},
+        ],
+    )
+    output = root / "data" / "fullmap.redb"
+    rs.build_fullmap_db(output, [classes], [synonyms])
+    return output
+
+
+def _write_build_inputs(tmp_path: Path, rig_factory: Any, source_text: str) -> tuple[Path, Path, Path]:
+    source = tmp_path / "data.tsv"
+    source.write_text(source_text)
+    table = tmp_path / "table.yaml"
+    to_yaml(
+        table,
+        {
+            "template": {
+                "source": {"kind": "text", "local": str(source), "url": ["https://example.com/data.tsv"], "delimiter": "\t"},
+                "statement": {
+                    "subject": {"method": "column", "encoding": "A"},
+                    "predicate": "associated_with",
+                    "object": {"method": "column", "encoding": "B"},
+                },
+                "provenance": {"repo": "PMC", "publication": "PMC0000000"},
+            }
+        },
+    )
+    fullmap = _build_real_redb(tmp_path / "fullmap")
+    graph = tmp_path / "graph.yaml"
+    to_yaml(
+        graph,
+        {
+            "name": "INVALIDATION_KG",
+            "version": "1.0.0",
+            "tables": [str(table)],
+            "fullmap": str(fullmap),
+            "rig": rig_factory(tmp_path, infores_id="infores:invalidation-kg", source_info={"description": "store invalidation smoke graph"}),
+        },
+    )
+    return source, graph, table
+
+
+def _store_files() -> dict[str, Path]:
+    return {path.name: path for path in Path(".tablassert/store").glob("*.parquet")}
+
+
+def _build(graph: Path) -> None:
+    build_pipeline(graph, PipelineProgress(total_stages=6))
+
+
+def test_unchanged_build_reuses_store_and_preserves_outputs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, rig_factory: Any) -> None:
+    """An unchanged build must quick-exit the section cache without changing artifacts.
+
+    WHY: content hashing should invalidate stale source data, but a repeat build with
+    identical bytes must retain the existing parquet and byte-identical KGX output.
+    """
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / ".tablassert/store").mkdir(parents=True)
+    _, graph, _ = _write_build_inputs(tmp_path, rig_factory, "brca1\tmapk1\n")
+
+    _build(graph)
+    edges = tmp_path / "INVALIDATION_KG_1.0.0.edges.ndjson"
+    nodes = tmp_path / "INVALIDATION_KG_1.0.0.nodes.ndjson"
+    first_outputs = (nodes.read_bytes(), edges.read_bytes())
+    first_mtimes = {name: path.stat().st_mtime_ns for name, path in _store_files().items()}
+    assert first_mtimes
+
+    _build(graph)
+    assert (nodes.read_bytes(), edges.read_bytes()) == first_outputs
+    assert {name: path.stat().st_mtime_ns for name, path in _store_files().items()} == first_mtimes
+
+
+def test_source_edit_writes_new_store_and_new_edges(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, rig_factory: Any) -> None:
+    """Adding a source row rekeys the store and makes the row reach the final edges.
+
+    WHY: a config-only cache key would quick-exit the old parquet and silently omit
+    assertions added to an otherwise unchanged TSV configuration.
+    """
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / ".tablassert/store").mkdir(parents=True)
+    source, graph, _ = _write_build_inputs(tmp_path, rig_factory, "brca1\tmapk1\n")
+
+    _build(graph)
+    old_stores = _store_files()
+    source.write_text("brca1\tmapk1\nmapk1\tbrca1\n")
+    _build(graph)
+    new_stores = _store_files()
+
+    assert set(old_stores) < set(new_stores)
+    edge_rows = [json.loads(line) for line in (tmp_path / "INVALIDATION_KG_1.0.0.edges.ndjson").read_text().splitlines() if line]
+    assert {(row["subject"], row["object"]) for row in edge_rows} == {("HGNC:1100", "HGNC:6871"), ("HGNC:6871", "HGNC:1100")}
+
+
+def test_legacy_store_orphan_is_not_a_cache_hit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, rig_factory: Any) -> None:
+    """A legacy config-only parquet remains untouched while a content-aware store is built.
+
+    WHY: upgrading keying must orphan old stores safely; treating one as a hit could
+    expose stale or incompatible rows while overwriting user data would destroy it.
+    """
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / ".tablassert/store").mkdir(parents=True)
+    _, graph, table = _write_build_inputs(tmp_path, rig_factory, "brca1\tmapk1\n")
+
+    raw = from_yaml(table)
+    section = to_sections(raw, table)[0]  # type: ignore[index]
+    legacy = Path(".tablassert/store") / f"{mkhash(section)}.parquet"
+    pl.DataFrame({"legacy_marker": ["must not be read"]}).write_parquet(legacy)
+    marker_bytes = legacy.read_bytes()
+
+    _build(graph)
+    stores = _store_files()
+    assert legacy.read_bytes() == marker_bytes
+    assert len(stores) == 2
+    assert any(path != legacy for path in stores.values())
+    edges = (tmp_path / "INVALIDATION_KG_1.0.0.edges.ndjson").read_text()
+    assert "HGNC:1100" in edges
+    assert "HGNC:6871" in edges
diff --git a/tests/test_store_keying.py b/tests/test_store_keying.py
new file mode 100644
index 0000000..33fd338
--- /dev/null
+++ b/tests/test_store_keying.py
@@ -0,0 +1,166 @@
+"""Stage 3 content-aware store-key tests.
+
+These tests exercise the same key derivation seam used by ``build_graph_pipeline``
+without running the unrelated fullmap and graph compilation stages.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tablassert import cli
+from tablassert.errors import SourceFileError
+from tablassert.utils import section_store_key
+
+
+def _section(source: Path | None, marker: str = "same") -> dict[str, Any]:
+    """Build the smallest section-shaped mapping needed by Stage 3 keying."""
+    return {
+        "config": Path("table.yaml"),
+        "marker": marker,
+        "source": {"local": str(source)} if source is not None else {"url": ["https://example.org/data.tsv"]},
+    }
+
+
+def _key(section: dict[str, Any], tmp_path: Path, memo: dict[tuple[Path, int, int], str]) -> str:
+    """Derive a key through the exact private helper used by the build loop."""
+    return cli._section_store_key_for_build(section, tmp_path / "graph.yaml", memo)[0]
+
+
+def test_content_change_gives_new_store_filename(tmp_path: Path) -> None:
+    """Changing source bytes changes the Stage 3 parquet filename.
+
+    WHY: a config-only key would let ``Tcode.collect`` quick-exit with a parquet made
+    from stale source rows after the upstream file was edited.
+    """
+    source: Path = tmp_path / "data.tsv"
+    source.write_bytes(b"v1")
+    memo: dict[tuple[Path, int, int], str] = {}
+    first: str = _key(_section(source), tmp_path, memo)
+    source.write_bytes(b"v2")
+    second: str = _key(_section(source), tmp_path, memo)
+    assert first != second
+    assert len(second) == 16
+    assert (tmp_path / ".tablassert" / "store" / f"{second}.parquet").name == f"{second}.parquet"
+
+
+def test_utime_only_touch_keeps_store_filename(tmp_path: Path) -> None:
+    """Touching a source without changing bytes preserves its filename.
+
+    WHY: stat metadata is only the per-build memo invalidation signature; it must not
+    become part of the content-addressed store identity.
+    """
+    source: Path = tmp_path / "data.tsv"
+    source.write_bytes(b"same bytes")
+    memo: dict[tuple[Path, int, int], str] = {}
+    first: str = _key(_section(source), tmp_path, memo)
+    stat = source.stat()
+    os.utime(source, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000))
+    assert _key(_section(source), tmp_path, memo) == first
+
+
+def test_stat_change_rehashes_shared_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """A changed stat signature causes the per-build memo to read the file again.
+
+    WHY: the memo is an intra-build optimization, not a correctness cache; a source
+    replaced during a build must not keep using a digest associated with stale metadata.
+    """
+    source: Path = tmp_path / "data.tsv"
+    source.write_bytes(b"same bytes")
+    import tablassert.utils
+
+    real_hash = tablassert.utils.file_content_hash
+    calls: list[Path] = []
+
+    def counted_hash(path: Path, **kwargs: Any) -> str:
+        calls.append(path)
+        return real_hash(path, **kwargs)
+
+    monkeypatch.setattr(tablassert.utils, "file_content_hash", counted_hash)
+    memo: dict[tuple[Path, int, int], str] = {}
+    first: str = _key(_section(source), tmp_path, memo)
+    stat = source.stat()
+    os.utime(source, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000_000))
+    second: str = _key(_section(source), tmp_path, memo)
+    assert first == second
+    assert calls == [source.resolve(), source.resolve()]
+
+
+def test_config_only_change_rekeys(tmp_path: Path) -> None:
+    """Changing section configuration rekeys even when the source bytes are shared.
+
+    WHY: content awareness augments, rather than replaces, the section configuration
+    identity; two transformations over one file must never share a parquet.
+    """
+    source: Path = tmp_path / "data.tsv"
+    source.write_bytes(b"same bytes")
+    memo: dict[tuple[Path, int, int], str] = {}
+    first: str = _key(_section(source, marker="one"), tmp_path, memo)
+    second: str = _key(_section(source, marker="two"), tmp_path, memo)
+    assert first != second
+
+
+def test_shared_file_hashes_once_per_build(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Distinct sections sharing a file invoke the content hash once per build.
+
+    WHY: a graph can expand many sections over one downloaded source; Stage 3 should
+    avoid repeatedly streaming the same large file while retaining per-build scope.
+    """
+    source: Path = tmp_path / "data.tsv"
+    source.write_bytes(b"shared bytes")
+    import tablassert.utils
+
+    real_hash = tablassert.utils.file_content_hash
+    calls: list[Path] = []
+
+    def counted_hash(path: Path, **kwargs: Any) -> str:
+        calls.append(path)
+        return real_hash(path, **kwargs)
+
+    monkeypatch.setattr(tablassert.utils, "file_content_hash", counted_hash)
+    memo: dict[tuple[Path, int, int], str] = {}
+    _key(_section(source, marker="one"), tmp_path, memo)
+    _key(_section(source, marker="two"), tmp_path, memo)
+    assert calls == [source.resolve()]
+
+
+@pytest.mark.parametrize(("kind", "detail"), [("missing", "no such file"), ("directory", "not a regular file"), ("unreadable", "permission denied")])
+def test_source_errors_name_config_section_and_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, kind: str, detail: str) -> None:
+    """Missing, directory, and hashing-read failures are coded and contextualized.
+
+    WHY: Stage 3 is the first build stage that reads ``source.local`` for identity;
+    users need the configuration, section label, and resolved offending path at once.
+    """
+    source: Path = tmp_path / f"{kind}.tsv"
+    if kind == "directory":
+        source.mkdir()
+    elif kind == "unreadable":
+        source.write_bytes(b"bytes")
+
+        def fail_hash(path: Path, **kwargs: Any) -> str:
+            raise SourceFileError(path.resolve(), detail, **kwargs)
+
+        monkeypatch.setattr("tablassert.utils.file_content_hash", fail_hash)
+    section: dict[str, Any] = _section(source)
+    with pytest.raises(SourceFileError) as exc_info:
+        _key(section, tmp_path, {})
+    error = exc_info.value
+    assert error.code == "source-file-unreadable"
+    message = str(error)
+    assert "graph.yaml" in message
+    assert "table · " in message
+    assert str(source.resolve()) in message
+
+
+def test_no_local_uses_config_only_key(tmp_path: Path) -> None:
+    """A URL-only section keeps the config-only key and does not read a data file.
+
+    WHY: validation accepts URL-only/value sections without a local source, preserving
+    the prior Tcode validation behavior and cache identity.
+    """
+    section = _section(None)
+    assert _key(section, tmp_path, {}) == section_store_key(section)
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 5129459..d830cf3 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -1,7 +1,14 @@
 from __future__ import annotations
 
+import os
+from pathlib import Path
+from typing import Any
+
+import pytest
+
 from tablassert import rs
-from tablassert.utils import mkhash
+from tablassert.errors import SourceFileError
+from tablassert.utils import file_content_hash, mkhash, section_store_key
 
 
 def test_mkhash_deterministic() -> None:
@@ -47,3 +54,130 @@ def test_mkhash_is_full_xxh64_of_stringified_input() -> None:
     """
     assert mkhash("hello") == rs.xxh64("hello") == "26c7827d889f6da3"
     assert mkhash(42) == rs.xxh64("42")
+
+
+def test_file_content_hash_utf8_parity_with_rs_xxh64(tmp_path: Path) -> None:
+    """A file containing 'hello' hashes to the pinned digest 26c7827d889f6da3.
+
+    WHY: every section store key derives from this digest. If the streaming file hash
+    ever drifted from ``rs.xxh64`` over the same UTF-8 bytes, each cached section
+    store would be silently invalidated — or worse, collide with another section's.
+    """
+    source: Path = tmp_path / "hello.txt"
+    source.write_bytes(b"hello")  # ASCII == its UTF-8 encoding; parity with rs.xxh64("hello")
+    digest: str = file_content_hash(source)
+    assert digest == rs.xxh64("hello") == "26c7827d889f6da3"
+    assert digest == rs.xxh64_file(str(source.resolve()))
+
+
+def test_file_content_hash_resolves_relative_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Relative paths resolve against the cwd before hashing.
+
+    WHY: table configs name ``source.local`` relative to wherever the build runs; the
+    digest must be of the file the user meant, and errors must name the resolved path.
+    """
+    (tmp_path / "rel.csv").write_bytes(b"hello")
+    monkeypatch.chdir(tmp_path)
+    assert file_content_hash(Path("rel.csv")) == "26c7827d889f6da3"
+
+
+def test_file_content_hash_reflects_current_bytes(tmp_path: Path) -> None:
+    """Editing the file changes the digest on the very next call.
+
+    WHY: the helper is deliberately pure (no caching) — a memoized digest would let a
+    stale cache entry survive an edit to the source file.
+    """
+    source: Path = tmp_path / "data.csv"
+    source.write_bytes(b"v1")
+    first: str = file_content_hash(source)
+    source.write_bytes(b"v2")
+    assert file_content_hash(source) != first
+
+
+def test_file_content_hash_missing_path_raises(tmp_path: Path) -> None:
+    """A missing source file fails loudly, never with a sentinel digest.
+
+    WHY: hashing a nonexistent path must name the offending path so a user can fix
+    their ``source.local`` instead of debugging a wrong-cache-hit downstream.
+    """
+    with pytest.raises(SourceFileError) as excinfo:
+        file_content_hash(tmp_path / "nope.csv")
+    assert excinfo.value.code == "source-file-unreadable"
+    assert "nope.csv" in str(excinfo.value)
+
+
+def test_file_content_hash_directory_raises(tmp_path: Path) -> None:
+    """A directory passed as a source file is rejected before any read.
+
+    WHY: directories can be 'opened' on some platforms but hashing one is nonsense;
+    the failure must say the path is not a regular file, not surface a raw OS error.
+    """
+    with pytest.raises(SourceFileError) as excinfo:
+        file_content_hash(tmp_path)
+    assert excinfo.value.code == "source-file-unreadable"
+    assert "not a regular file" in str(excinfo.value)
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="root bypasses file permission bits")
+def test_file_content_hash_unreadable_file_raises_and_chains(tmp_path: Path) -> None:
+    """An unreadable file wraps the OS error, chained as the cause.
+
+    WHY: permission failures must surface as SourceFileError with the underlying
+    OSError preserved on ``__cause__`` so the traceback still shows the real errno.
+    """
+    source: Path = tmp_path / "locked.csv"
+    source.write_bytes(b"secret")
+    source.chmod(0)
+    try:
+        with pytest.raises(SourceFileError) as excinfo:
+            file_content_hash(source)
+    finally:
+        source.chmod(0o644)
+    assert excinfo.value.code == "source-file-unreadable"
+    assert isinstance(excinfo.value.__cause__, OSError)
+
+
+def test_section_store_key_without_local_is_plain_mkhash() -> None:
+    """Without a local file the store key is exactly ``mkhash(section)``.
+
+    WHY: back-compat — sections sourced purely by URL keep their existing cache
+    identity, so adopting content-aware keys does not invalidate every URL-sourced
+    cache on upgrade.
+    """
+    section: dict[str, Any] = {"config": "x", "source": {"url": ["https://example.org/a.csv"]}}
+    key: str = section_store_key(section)
+    assert key == mkhash(section)
+    assert len(key) == 16
+    int(key, 16)
+
+
+def test_section_store_key_with_local_mixes_content_digest(tmp_path: Path) -> None:
+    """With a local file the key is mkhash('<mkhash(section)>:<file digest>').
+
+    WHY: the local file's bytes are part of the section's identity; the key shape is
+    pinned so a drift in the mixing formula is caught as a cache-busting change, and
+    the output stays one 16-hex digest like every other store key.
+    """
+    source: Path = tmp_path / "s.csv"
+    source.write_bytes(b"hello")
+    section: dict[str, Any] = {"config": "x", "source": {"local": str(source)}}
+    expected: str = mkhash(f"{mkhash(section)}:{file_content_hash(source)}")
+    key: str = section_store_key(section, source)
+    assert key == expected
+    assert len(key) == 16
+    int(key, 16)
+
+
+def test_section_store_key_tracks_file_edits(tmp_path: Path) -> None:
+    """Editing the local source changes the store key; the config-only key differs too.
+
+    WHY: this is the entire point of content-aware keys — a stale parquet must never
+    survive an edit to the file it was built from.
+    """
+    source: Path = tmp_path / "s.csv"
+    source.write_bytes(b"v1")
+    section: dict[str, Any] = {"config": "x", "source": {"local": str(source)}}
+    first: str = section_store_key(section, source)
+    source.write_bytes(b"v2")
+    assert section_store_key(section, source) != first
+    assert section_store_key(section) != first