fix(corpus): harden canonical chunk and frontmatter invariants

This commit is contained in:
2026-07-12 03:50:32 +02:00
parent 015715d092
commit 11e7ee9ea6
5 changed files with 186 additions and 53 deletions
@@ -37,3 +37,23 @@
configuration before ingestion is wired. configuration before ingestion is wired.
- Character limits use Python Unicode code points (`len`), not UTF-8 bytes or tokenizer tokens; - Character limits use Python Unicode code points (`len`), not UTF-8 bytes or tokenizer tokens;
this is recorded in the chunk-policy metadata and tested with non-ASCII content. this is recorded in the chunk-policy metadata and tested with non-ASCII content.
## Review hardening follow-up
- Chunk IDs now bind the canonical document identity, document content hash, ordinal, chunk hash,
and a canonical SHA-256 fingerprint of every `ChunkPolicy` field. Identical content in separate
documents and same-version policies with different limits cannot collide.
- Boundary-aware slicing now retains separators in the slices. Concatenating every chunk exactly
reconstructs the canonical document for repeated spaces, tabs, blank lines, Markdown hard
breaks, fenced code, whitespace-only input, Unicode, and overlong tokens; every slice remains
within `max_chars`.
- Frontmatter uses a bounded `SafeLoader` variant: duplicate keys, anchors/aliases, structures
deeper than 20 nodes, and documents larger than 1000 composed nodes are rejected. YAML parse,
JSON type, credential-safety, and resulting canonical-model errors attributable to frontmatter
map to `PermanentNormalizationError(reason="invalid_frontmatter")`; invalid pipeline policy
remains a programmer-facing `ValueError`.
- Follow-up TDD evidence: the expanded focused suite first reported 11 expected failures against
the prior implementation, then passed **45/45** across normalization, chunking, and manifest
invariants.
- Follow-up full verification: **586 passed, 5 deselected**. Targeted Ruff is clean. Full Ruff
continues to report the same **34 unrelated pre-existing** violations in legacy tests.
+50 -3
View File
@@ -3,7 +3,7 @@ import hashlib
import pytest import pytest
from tht.corpus.chunk import ChunkPolicy, chunk from tht.corpus.chunk import ChunkPolicy, chunk
from tht.corpus.models import CanonicalDocument from tht.corpus.models import CanonicalDocument, CorpusManifest
def document(content: str) -> CanonicalDocument: def document(content: str) -> CanonicalDocument:
@@ -23,6 +23,16 @@ def document(content: str) -> CanonicalDocument:
) )
def other_document(content: str) -> CanonicalDocument:
return document(content).model_copy(
update={
"document_id": "doc:def",
"source_id": "source:b",
"source_uri": "https://host/b.md",
}
)
def test_chunk_ids_are_stable_for_same_content_and_repeat_runs(): def test_chunk_ids_are_stable_for_same_content_and_repeat_runs():
policy = ChunkPolicy(version="paragraph:v1", max_chars=8) policy = ChunkPolicy(version="paragraph:v1", max_chars=8)
first = chunk(document("A\n\nB"), policy) first = chunk(document("A\n\nB"), policy)
@@ -39,12 +49,47 @@ def test_policy_version_changes_ids_without_changing_boundaries():
assert [item.chunk_id for item in first] != [item.chunk_id for item in second] assert [item.chunk_id for item in first] != [item.chunk_id for item in second]
def test_same_policy_version_with_different_boundary_config_changes_ids():
doc = document("alpha beta")
first = chunk(doc, ChunkPolicy(version="paragraph:v1", max_chars=6))
second = chunk(doc, ChunkPolicy(version="paragraph:v1", max_chars=7))
assert first[0].chunk_id != second[0].chunk_id
def test_identical_content_in_different_documents_cannot_collide_in_manifest():
policy = ChunkPolicy(version="paragraph:v1", max_chars=20)
first = document("same")
second = other_document("same")
chunks = [*chunk(first, policy), *chunk(second, policy)]
manifest = CorpusManifest(
pipeline_version="pipe:v1", documents=[first, second], chunks=chunks
)
assert len({item.chunk_id for item in manifest.chunks}) == 2
def test_long_non_ascii_tokens_are_hard_split_by_unicode_characters(): def test_long_non_ascii_tokens_are_hard_split_by_unicode_characters():
chunks = chunk(document("ééééé世界"), ChunkPolicy(version="chars:v1", max_chars=3)) chunks = chunk(document("ééééé世界"), ChunkPolicy(version="chars:v1", max_chars=3))
assert [item.content for item in chunks] == ["ééé", "éé世", "界"] assert [item.content for item in chunks] == ["ééé", "éé世", "界"]
assert all(len(item.content) <= 3 for item in chunks) assert all(len(item.content) <= 3 for item in chunks)
@pytest.mark.parametrize(
"content",
[
"alpha beta\tgamma\n\ndelta",
"line with markdown hard break \nnext line\n```\na b\n```",
" \t\n\n \n",
"supercalifragilisticexpialidocious",
"é 世界\r\nnext",
],
)
def test_chunks_preserve_every_character_and_respect_max_chars(content):
doc = document(content)
chunks = chunk(doc, ChunkPolicy(version="exact:v1", max_chars=9))
assert "".join(item.content for item in chunks) == doc.content
assert all(0 < len(item.content) <= 9 for item in chunks)
def test_chunks_have_contiguous_ordinals_hashes_and_provenance_metadata(): def test_chunks_have_contiguous_ordinals_hashes_and_provenance_metadata():
doc = document("alpha beta gamma") doc = document("alpha beta gamma")
chunks = chunk(doc, ChunkPolicy(version="words:v1", max_chars=7)) chunks = chunk(doc, ChunkPolicy(version="words:v1", max_chars=7))
@@ -54,13 +99,15 @@ def test_chunks_have_contiguous_ordinals_hashes_and_provenance_metadata():
assert item.source_uri == doc.source_uri assert item.source_uri == doc.source_uri
assert item.document_id == doc.document_id assert item.document_id == doc.document_id
assert item.pipeline_version == doc.pipeline_version assert item.pipeline_version == doc.pipeline_version
assert item.metadata["chunk_policy"] == {"max_chars": 7, "version": "words:v1"} assert item.metadata["chunk_policy"]["max_chars"] == 7
assert item.metadata["chunk_policy"]["version"] == "words:v1"
assert item.metadata["chunk_policy"]["fingerprint"].startswith("sha256:")
assert item.metadata["document"] == {"owner": "docs"} assert item.metadata["document"] == {"owner": "docs"}
assert item.content_hash == "sha256:" + hashlib.sha256(item.content.encode()).hexdigest() assert item.content_hash == "sha256:" + hashlib.sha256(item.content.encode()).hexdigest()
def test_duplicate_chunk_content_cannot_collide_across_ordinals(): def test_duplicate_chunk_content_cannot_collide_across_ordinals():
chunks = chunk(document("same\n\nsame"), ChunkPolicy(version="paragraph:v1", max_chars=8)) chunks = chunk(document("samesame"), ChunkPolicy(version="paragraph:v1", max_chars=4))
assert [item.content for item in chunks] == ["same", "same"] assert [item.content for item in chunks] == ["same", "same"]
assert chunks[0].chunk_id != chunks[1].chunk_id assert chunks[0].chunk_id != chunks[1].chunk_id
+23
View File
@@ -52,6 +52,29 @@ def test_frontmatter_can_end_at_eof_without_inventing_content():
assert document.content == "" assert document.content == ""
@pytest.mark.parametrize(
"frontmatter",
[
"title: first\ntitle: second",
"title: &shared value\ncopy: *shared",
"nested: " + "[" * 25 + "x" + "]" * 25,
"items: [" + ",".join("x" for _ in range(1100)) + "]",
"api_key: secret",
],
)
def test_rejects_unsafe_frontmatter_as_typed_permanent_error(frontmatter):
raw = f"---\n{frontmatter}\n---\nbody".encode()
with pytest.raises(PermanentNormalizationError) as caught:
normalize(acquired(raw), "pipe:v1")
assert caught.value.reason == "invalid_frontmatter"
def test_pipeline_policy_errors_are_not_misclassified_as_bad_frontmatter():
with pytest.raises(ValueError, match="pipeline_version") as caught:
normalize(acquired(b"---\ntitle: valid\n---\nbody"), "")
assert not isinstance(caught.value, PermanentNormalizationError)
@pytest.mark.parametrize( @pytest.mark.parametrize(
("content", "media_type", "reason"), ("content", "media_type", "reason"),
[ [
+31 -36
View File
@@ -1,8 +1,9 @@
"""Versioned deterministic chunking for canonical corpus documents.""" """Versioned deterministic chunking for canonical corpus documents."""
import hashlib import hashlib
import json
import re import re
from dataclasses import dataclass from dataclasses import asdict, dataclass
from tht.corpus.models import CanonicalChunk, CanonicalDocument from tht.corpus.models import CanonicalChunk, CanonicalDocument
@@ -23,62 +24,56 @@ def _hash(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest() return hashlib.sha256(text.encode("utf-8")).hexdigest()
def _hard_split(text: str, maximum: int) -> list[str]:
return [text[start : start + maximum] for start in range(0, len(text), maximum)]
def _split_block(block: str, maximum: int) -> list[str]:
if len(block) <= maximum:
return [block]
tokens = re.findall(r"\S+", block)
chunks: list[str] = []
current = ""
for token in tokens:
if len(token) > maximum:
if current:
chunks.append(current)
current = ""
chunks.extend(_hard_split(token, maximum))
continue
candidate = f"{current} {token}" if current else token
if len(candidate) <= maximum:
current = candidate
else:
chunks.append(current)
current = token
if current:
chunks.append(current)
return chunks
def _contents(content: str, maximum: int) -> list[str]: def _contents(content: str, maximum: int) -> list[str]:
if not content:
return []
result: list[str] = [] result: list[str] = []
for block in re.split(r"\n{2,}", content): start = 0
if block: while start < len(content):
result.extend(_split_block(block, maximum)) end = min(start + maximum, len(content))
if end < len(content):
boundaries = list(re.finditer(r"\s+", content[start:end]))
if boundaries:
end = start + boundaries[-1].end()
result.append(content[start:end])
start = end
return result return result
def _policy_fingerprint(policy: ChunkPolicy) -> str:
serialized = json.dumps(asdict(policy), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
return f"sha256:{_hash(serialized)}"
def chunk(document: CanonicalDocument, policy: ChunkPolicy) -> list[CanonicalChunk]: def chunk(document: CanonicalDocument, policy: ChunkPolicy) -> list[CanonicalChunk]:
"""Split canonical text with stable character-count boundaries and identifiers.""" """Split canonical text with stable character-count boundaries and identifiers."""
chunks: list[CanonicalChunk] = [] chunks: list[CanonicalChunk] = []
policy_fingerprint = _policy_fingerprint(policy)
for ordinal, content in enumerate(_contents(document.content, policy.max_chars)): for ordinal, content in enumerate(_contents(document.content, policy.max_chars)):
identifier = _hash(f"{document.content_hash}:{ordinal}:{policy.version}") chunk_hash = f"sha256:{_hash(content)}"
identifier = _hash(
":".join(
(
document.document_id,
document.content_hash,
policy_fingerprint,
str(ordinal),
chunk_hash,
)
)
)
chunks.append( chunks.append(
CanonicalChunk( CanonicalChunk(
chunk_id=f"chunk:{identifier}", chunk_id=f"chunk:{identifier}",
document_id=document.document_id, document_id=document.document_id,
ordinal=ordinal, ordinal=ordinal,
content=content, content=content,
content_hash=f"sha256:{_hash(content)}", content_hash=chunk_hash,
source_uri=document.source_uri, source_uri=document.source_uri,
pipeline_version=document.pipeline_version, pipeline_version=document.pipeline_version,
metadata={ metadata={
"chunk_policy": { "chunk_policy": {
"version": policy.version, "version": policy.version,
"max_chars": policy.max_chars, "max_chars": policy.max_chars,
"fingerprint": policy_fingerprint,
}, },
"document": document.model_dump(mode="json")["metadata"], "document": document.model_dump(mode="json")["metadata"],
"source_fingerprint": document.source_fingerprint, "source_fingerprint": document.source_fingerprint,
+62 -14
View File
@@ -7,6 +7,8 @@ from collections.abc import Mapping
import yaml import yaml
from pydantic import JsonValue, TypeAdapter, ValidationError from pydantic import JsonValue, TypeAdapter, ValidationError
from yaml.events import AliasEvent
from yaml.nodes import MappingNode
from tht.corpus.models import CanonicalDocument from tht.corpus.models import CanonicalDocument
from tht.ports.evidence import AcquiredDocument, canonical_provenance_uri from tht.ports.evidence import AcquiredDocument, canonical_provenance_uri
@@ -16,6 +18,47 @@ MAX_DOCUMENT_BYTES = 10 * 1024 * 1024
_CHARSET = re.compile(r"(?:^|;)\s*charset\s*=\s*[\"']?([^;\s\"']+)", re.IGNORECASE) _CHARSET = re.compile(r"(?:^|;)\s*charset\s*=\s*[\"']?([^;\s\"']+)", re.IGNORECASE)
_FRONTMATTER = re.compile(r"\A---\n(.*?)\n---(?:\n|\Z)", re.DOTALL) _FRONTMATTER = re.compile(r"\A---\n(.*?)\n---(?:\n|\Z)", re.DOTALL)
_JSON_OBJECT = TypeAdapter(dict[str, JsonValue]) _JSON_OBJECT = TypeAdapter(dict[str, JsonValue])
_MAX_FRONTMATTER_DEPTH = 20
_MAX_FRONTMATTER_NODES = 1000
class _FrontmatterLoader(yaml.SafeLoader):
"""SafeLoader with bounded structure and no YAML graph features."""
def __init__(self, stream) -> None:
super().__init__(stream)
self._depth = 0
self._nodes = 0
def compose_node(self, parent, index):
event = self.peek_event()
if isinstance(event, AliasEvent) or getattr(event, "anchor", None) is not None:
raise yaml.constructor.ConstructorError(None, None, "aliases are not allowed")
self._depth += 1
self._nodes += 1
if self._depth > _MAX_FRONTMATTER_DEPTH or self._nodes > _MAX_FRONTMATTER_NODES:
raise yaml.constructor.ConstructorError(None, None, "frontmatter is too complex")
try:
return super().compose_node(parent, index)
finally:
self._depth -= 1
def construct_mapping(self, node, deep=False):
if not isinstance(node, MappingNode):
return super().construct_mapping(node, deep=deep)
seen: set[object] = set()
for key_node, _ in node.value:
key = self.construct_object(key_node, deep=deep)
try:
duplicate = key in seen
seen.add(key)
except TypeError as error:
raise yaml.constructor.ConstructorError(
None, None, "mapping keys must be scalar"
) from error
if duplicate:
raise yaml.constructor.ConstructorError(None, None, "duplicate mapping key")
return super().construct_mapping(node, deep=deep)
class PermanentNormalizationError(ValueError): class PermanentNormalizationError(ValueError):
@@ -55,7 +98,7 @@ def _frontmatter(text: str) -> tuple[dict[str, JsonValue], str]:
if match is None: if match is None:
return {}, text return {}, text
try: try:
loaded = yaml.safe_load(match.group(1)) loaded = yaml.load(match.group(1), Loader=_FrontmatterLoader)
if loaded is None: if loaded is None:
loaded = {} loaded = {}
if not isinstance(loaded, Mapping): if not isinstance(loaded, Mapping):
@@ -84,16 +127,21 @@ def normalize(acquired: AcquiredDocument, pipeline_version: str) -> CanonicalDoc
if frontmatter: if frontmatter:
metadata["frontmatter"] = frontmatter metadata["frontmatter"] = frontmatter
return CanonicalDocument( try:
document_id=f"doc:{_sha256(identity)}", return CanonicalDocument(
source_id=acquired.source.source_id, document_id=f"doc:{_sha256(identity)}",
source_uri=source_uri, source_id=acquired.source.source_id,
source_fingerprint=acquired.source.fingerprint, source_uri=source_uri,
content_hash=f"sha256:{_sha256(content)}", source_fingerprint=acquired.source.fingerprint,
title=str(frontmatter.get("title", "")), content_hash=f"sha256:{_sha256(content)}",
content=content, title=str(frontmatter.get("title", "")),
media_type=media_type, content=content,
modified_at=acquired.source.modified_at, media_type=media_type,
pipeline_version=pipeline_version, modified_at=acquired.source.modified_at,
metadata=metadata, pipeline_version=pipeline_version,
) metadata=metadata,
)
except ValidationError as error:
if frontmatter:
raise PermanentNormalizationError("invalid_frontmatter") from error
raise