feat(#204): deterministic chunkset hash for auto-stale detection

Adds the pure-domain hash function that summarises a chunkset for
stale-detection purposes. Recorded on each DocumentStoreLink at push
time (#203 ships the column slot); compared against the recomputed
current hash to flip a link to Stale when the source has drifted.

domain/hashing.py
- chunkset_hash(chunks: Iterable[ChunkResult]) -> str
- SHA-256 over (text, source_page, headings) per chunk
- Excludes bboxes / doc_items / token_count by design
- 0x1F separator between chunks defends against the join-attack
  (split A+B vs concat AB)

Tests
- 9 tests: determinism, sensitivity per included field, exclusion of
  rendering-only fields, join-attack resistance, order sensitivity,
  empty-input invariant
- Locked fixture: a hand-built 3-chunk input has a fixed expected hash;
  CI fails loud if anyone changes the canonical inputs without updating
  the fixture deliberately (and the release notes)

Service integration (recompute on chunk write, set on push) lands with
#205 once chunks are first-class — direct integration on the legacy
chunks_json path is deliberately deferred to keep #204 focused.

Refs #204
This commit is contained in:
Pier-Jean Malandrino 2026-04-29 15:40:49 +02:00
parent d81f36d899
commit 4b59ef2330
2 changed files with 169 additions and 0 deletions

View file

@ -0,0 +1,67 @@
"""Deterministic hashing for chunksets — substrate for auto-stale detection (#204).
A `chunkset_hash` summarises the content of a list of chunks for a
document. The hash is recorded on each `DocumentStoreLink` at push time
(#203 ships the column slot). When chunks change, recomputing the hash
and comparing against the stored value tells us whether the link has
gone stale.
Why a hash and not, say, an updated_at?
- Idempotent re-pipelines on identical input bump `updated_at` without
semantic change. A content hash is the only signal that survives
that.
- It is also content-addressed: two different docs that happen to have
the same chunkset get the same hash. Useful for de-duplication
further down the road.
Inputs and exclusions are pinned. Any change to the canonical inputs
re-flips every existing link to Stale once that is a deliberate
release-note event, not a silent migration.
This module is pure: in / out. No I/O. No randomness. No dates.
"""
from __future__ import annotations
import hashlib
import json
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Iterable
from domain.value_objects import ChunkResult
# Byte separator inserted between chunks so concatenating two chunks does
# not yield the same hash as a single chunk with the joined text. \x1f is
# the Unicode "Information Separator One" — semantically appropriate and
# safe inside arbitrary text.
_CHUNK_SEPARATOR = b"\x1f"
def chunkset_hash(chunks: Iterable[ChunkResult]) -> str:
"""Return a deterministic SHA-256 hex digest over a chunkset.
Hashed inputs (per chunk, in order):
- text (str)
- source_page (int | None)
- headings (list[str], order preserved)
Excluded:
- bboxes / doc_items (rendering artefacts; do not affect retrieval)
- token_count (derived; unstable across tokenizers)
The exclusion list is intentional. Bumping it changes every link's
hash and triggers a one-time corpus-wide flip to `Stale`.
"""
h = hashlib.sha256()
for chunk in chunks:
payload = {
"t": chunk.text,
"p": chunk.source_page,
"h": list(chunk.headings or []),
}
h.update(_CHUNK_SEPARATOR)
h.update(json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode())
return h.hexdigest()

View file

@ -0,0 +1,102 @@
"""Tests for the chunkset hash function (#204)."""
from __future__ import annotations
from domain.hashing import chunkset_hash
from domain.value_objects import ChunkBbox, ChunkDocItem, ChunkResult
def _chunk(text: str, *, page: int | None = 1, headings=()) -> ChunkResult:
return ChunkResult(text=text, headings=list(headings), source_page=page)
def test_empty_chunkset_returns_empty_sha256() -> None:
"""Empty input has a stable, well-known hash."""
h = chunkset_hash([])
# SHA-256 of nothing is the well-known constant.
assert h == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
def test_determinism_across_invocations() -> None:
chunks = [_chunk("a"), _chunk("b"), _chunk("c")]
assert chunkset_hash(chunks) == chunkset_hash(chunks)
def test_text_change_changes_hash() -> None:
base = [_chunk("alpha")]
edited = [_chunk("alpha!")]
assert chunkset_hash(base) != chunkset_hash(edited)
def test_page_change_changes_hash() -> None:
a = [_chunk("alpha", page=1)]
b = [_chunk("alpha", page=2)]
assert chunkset_hash(a) != chunkset_hash(b)
def test_headings_change_changes_hash() -> None:
a = [_chunk("alpha", headings=("Section A",))]
b = [_chunk("alpha", headings=("Section B",))]
assert chunkset_hash(a) != chunkset_hash(b)
def test_excluded_fields_do_not_change_hash() -> None:
"""token_count, bboxes, doc_items are deliberately excluded."""
a = ChunkResult(text="alpha", headings=[], source_page=1, token_count=100, bboxes=[])
b = ChunkResult(
text="alpha",
headings=[],
source_page=1,
token_count=999, # different
bboxes=[ChunkBbox(page=1, bbox=[0, 0, 10, 10])], # different
doc_items=[ChunkDocItem(self_ref="#/x", label="text")], # different
)
assert chunkset_hash([a]) == chunkset_hash([b])
def test_join_attack_resistance() -> None:
"""Splitting one chunk into two with the same combined content must
produce a different hash from the original single-chunk version."""
one = [_chunk("HelloWorld")]
two = [_chunk("Hello"), _chunk("World")]
assert chunkset_hash(one) != chunkset_hash(two)
def test_order_matters() -> None:
a = [_chunk("alpha"), _chunk("bravo")]
b = [_chunk("bravo"), _chunk("alpha")]
assert chunkset_hash(a) != chunkset_hash(b)
def test_locked_fixture_three_chunks() -> None:
"""Locked fixture: a hand-built 3-chunk input has a fixed expected
hash. CI fails loud if anyone changes the canonical input list
without updating the fixture deliberately.
The expected hash below was computed once with the function as
pinned in this commit. To regenerate after a deliberate canonical
change, run:
python -c 'from domain.hashing import chunkset_hash; \\
from domain.value_objects import ChunkResult; \\
print(chunkset_hash([
ChunkResult(text="Intro paragraph.", source_page=1,
headings=["Intro"]),
ChunkResult(text="Body of section A.", source_page=2,
headings=["A"]),
ChunkResult(text="Conclusion.", source_page=3,
headings=["Conclusion"]),
]))'
"""
chunks = [
ChunkResult(text="Intro paragraph.", source_page=1, headings=["Intro"]),
ChunkResult(text="Body of section A.", source_page=2, headings=["A"]),
ChunkResult(text="Conclusion.", source_page=3, headings=["Conclusion"]),
]
expected = "6ac365ae403e53675e57884b69b0629684f2209c39730093231caa11a40e5225"
actual = chunkset_hash(chunks)
assert actual == expected, (
f"Hash drift detected. Expected {expected}, got {actual}. "
"If you intentionally changed the canonical inputs, update this "
"fixture and document the breaking change in the release notes."
)