feat(evidence): validate canonical curated corpus
This commit is contained in:
@@ -0,0 +1,256 @@
|
||||
"""Validation for the Git-reviewed Evidence authoring workspace."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
import yaml
|
||||
from pydantic import ValidationError, field_validator, model_validator
|
||||
|
||||
from tht.evidence.canonical import (
|
||||
CuratedEvidence,
|
||||
StrictModel,
|
||||
is_evidence_id,
|
||||
load_curated_tree,
|
||||
validate_source_file,
|
||||
)
|
||||
|
||||
MAX_AUTHORING_FILE_BYTES = 10 * 1024 * 1024
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ValidationFinding:
|
||||
severity: Literal["error", "warning"]
|
||||
code: str
|
||||
path: str
|
||||
message: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ValidationReport:
|
||||
findings: tuple[ValidationFinding, ...]
|
||||
|
||||
@property
|
||||
def publishable(self) -> bool:
|
||||
return not any(finding.severity == "error" for finding in self.findings)
|
||||
|
||||
|
||||
class ManifestSource(StrictModel):
|
||||
sha256: str
|
||||
units: tuple[str, ...]
|
||||
|
||||
@field_validator("sha256")
|
||||
@classmethod
|
||||
def _validate_sha256(cls, value: str) -> str:
|
||||
if not value.startswith("sha256:") or len(value) != 71:
|
||||
raise ValueError("sha256 must be a sha256 digest")
|
||||
try:
|
||||
int(value.removeprefix("sha256:"), 16)
|
||||
except ValueError as error:
|
||||
raise ValueError("sha256 must be a sha256 digest") from error
|
||||
return value
|
||||
|
||||
@field_validator("units")
|
||||
@classmethod
|
||||
def _validate_units(cls, value: tuple[str, ...]) -> tuple[str, ...]:
|
||||
if any(not is_evidence_id(unit) for unit in value):
|
||||
raise ValueError("units must use stable evidence identifiers")
|
||||
return value
|
||||
|
||||
|
||||
class EvidenceManifest(StrictModel):
|
||||
schema_version: Literal[1]
|
||||
pipeline_version: Literal["evidence-authoring-v1"]
|
||||
sources: dict[str, ManifestSource]
|
||||
orphans: tuple[str, ...]
|
||||
|
||||
@field_validator("sources")
|
||||
@classmethod
|
||||
def _validate_sources(cls, value: dict[str, ManifestSource]) -> dict[str, ManifestSource]:
|
||||
for source_path in value:
|
||||
validate_source_file(source_path)
|
||||
return value
|
||||
|
||||
@field_validator("orphans")
|
||||
@classmethod
|
||||
def _validate_orphans(cls, value: tuple[str, ...]) -> tuple[str, ...]:
|
||||
if any(not is_evidence_id(unit) for unit in value):
|
||||
raise ValueError("orphans must use stable evidence identifiers")
|
||||
return value
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_unit_membership(self) -> EvidenceManifest:
|
||||
seen: set[str] = set()
|
||||
for source in self.sources.values():
|
||||
duplicate = seen.intersection(source.units)
|
||||
if duplicate:
|
||||
raise ValueError("a unit may belong to only one source")
|
||||
seen.update(source.units)
|
||||
if len(set(self.orphans)) != len(self.orphans):
|
||||
raise ValueError("orphans must be unique")
|
||||
return self
|
||||
|
||||
|
||||
def load_manifest(path: Path) -> EvidenceManifest:
|
||||
"""Load the managed, versioned authoring manifest."""
|
||||
try:
|
||||
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeDecodeError, yaml.YAMLError) as error:
|
||||
raise ValueError("manifest cannot be read") from error
|
||||
try:
|
||||
return EvidenceManifest.model_validate(raw)
|
||||
except ValidationError as error:
|
||||
raise ValueError("manifest is invalid") from error
|
||||
|
||||
|
||||
def dump_manifest(manifest: EvidenceManifest) -> str:
|
||||
"""Serialize the manifest deterministically for Git review."""
|
||||
return yaml.safe_dump(manifest.model_dump(mode="json"), allow_unicode=True, sort_keys=True)
|
||||
|
||||
|
||||
def validate_workspace_evidence(workspace_root: Path) -> ValidationReport:
|
||||
"""Validate the curated corpus without writing the workspace."""
|
||||
evidence_root = workspace_root / "evidence"
|
||||
findings: list[ValidationFinding] = []
|
||||
manifest_path = evidence_root / "manifest.yaml"
|
||||
if not manifest_path.is_file():
|
||||
return ValidationReport((ValidationFinding(
|
||||
severity="error",
|
||||
code="manifest_missing",
|
||||
path="manifest.yaml",
|
||||
message="The managed Evidence manifest is missing.",
|
||||
),))
|
||||
try:
|
||||
manifest = load_manifest(manifest_path)
|
||||
except ValueError:
|
||||
return ValidationReport((ValidationFinding(
|
||||
severity="error",
|
||||
code="manifest_invalid",
|
||||
path="manifest.yaml",
|
||||
message="The managed Evidence manifest is invalid.",
|
||||
),))
|
||||
for orphan in manifest.orphans:
|
||||
findings.append(ValidationFinding(
|
||||
severity="error",
|
||||
code="orphaned_unit",
|
||||
path="manifest.yaml",
|
||||
message=f"Orphaned Evidence {orphan} must be resolved before publication.",
|
||||
))
|
||||
try:
|
||||
documents = load_curated_tree(evidence_root / "curated")
|
||||
except (OSError, ValidationError, ValueError):
|
||||
return ValidationReport(tuple(findings + [ValidationFinding(
|
||||
severity="error",
|
||||
code="curated_invalid",
|
||||
path="curated",
|
||||
message="A Curated Evidence document is invalid or cannot be read.",
|
||||
)]))
|
||||
source_texts = {
|
||||
source_path: _validate_manifest_source(evidence_root, source_path, source, findings)
|
||||
for source_path, source in manifest.sources.items()
|
||||
}
|
||||
seen_ids: set[str] = set()
|
||||
for evidence in documents:
|
||||
if evidence.id in seen_ids:
|
||||
findings.append(ValidationFinding(
|
||||
severity="error",
|
||||
code="duplicate_evidence_id",
|
||||
path="curated",
|
||||
message=f"Evidence id {evidence.id} appears more than once.",
|
||||
))
|
||||
seen_ids.add(evidence.id)
|
||||
findings.extend(_validate_unit(manifest, evidence, source_texts.get(evidence.provenance.source_file)))
|
||||
for source in manifest.sources.values():
|
||||
for unit in source.units:
|
||||
if unit not in seen_ids:
|
||||
findings.append(ValidationFinding(
|
||||
severity="error",
|
||||
code="manifest_unit_without_curated",
|
||||
path="manifest.yaml",
|
||||
message=f"Manifest Evidence {unit} has no Curated document.",
|
||||
))
|
||||
return ValidationReport(tuple(findings))
|
||||
|
||||
|
||||
def _validate_manifest_source(
|
||||
evidence_root: Path,
|
||||
source_path: str,
|
||||
manifest_source: ManifestSource,
|
||||
findings: list[ValidationFinding],
|
||||
) -> str | None:
|
||||
path = evidence_root / source_path
|
||||
if not path.is_file():
|
||||
findings.append(ValidationFinding("error", "source_missing", source_path,
|
||||
"The manifest source does not exist."))
|
||||
return None
|
||||
if path.stat().st_size > MAX_AUTHORING_FILE_BYTES:
|
||||
findings.append(ValidationFinding("error", "source_oversized", source_path,
|
||||
"The manifest source exceeds the authoring size limit."))
|
||||
return None
|
||||
try:
|
||||
source = _normalize(path.read_text(encoding="utf-8"))
|
||||
except UnicodeDecodeError:
|
||||
findings.append(ValidationFinding("error", "source_unreadable", source_path,
|
||||
"The manifest source is not valid UTF-8."))
|
||||
return None
|
||||
digest = "sha256:" + hashlib.sha256(source.encode("utf-8")).hexdigest()
|
||||
if digest != manifest_source.sha256:
|
||||
findings.append(ValidationFinding("error", "source_hash_mismatch", source_path,
|
||||
"The manifest hash does not match the normalized source."))
|
||||
return source
|
||||
|
||||
|
||||
def _validate_unit(
|
||||
manifest: EvidenceManifest, evidence: CuratedEvidence, source: str | None,
|
||||
) -> list[ValidationFinding]:
|
||||
path = evidence.provenance.source_file
|
||||
findings: list[ValidationFinding] = []
|
||||
manifest_source = manifest.sources.get(path)
|
||||
if manifest_source is None:
|
||||
findings.append(ValidationFinding(
|
||||
severity="error",
|
||||
code="manifest_source_missing",
|
||||
path="manifest.yaml",
|
||||
message="The manifest does not contain the provenance source.",
|
||||
))
|
||||
else:
|
||||
if manifest_source.sha256 != evidence.provenance.source_sha256:
|
||||
findings.append(ValidationFinding(
|
||||
severity="error",
|
||||
code="manifest_source_hash_mismatch",
|
||||
path="manifest.yaml",
|
||||
message="The manifest hash does not match the Evidence provenance.",
|
||||
))
|
||||
if evidence.id not in manifest_source.units:
|
||||
findings.append(ValidationFinding(
|
||||
severity="error",
|
||||
code="manifest_unit_missing",
|
||||
path="manifest.yaml",
|
||||
message="The manifest does not link the Evidence unit to its source.",
|
||||
))
|
||||
if source is None:
|
||||
return findings
|
||||
for excerpt in evidence.provenance.supporting_excerpts:
|
||||
if _normalize(excerpt) not in source:
|
||||
findings.append(ValidationFinding(
|
||||
severity="error",
|
||||
code="supporting_excerpt_missing",
|
||||
path=path,
|
||||
message="A supporting excerpt is absent from the normalized source.",
|
||||
))
|
||||
for item in evidence.review_items:
|
||||
findings.append(ValidationFinding(
|
||||
severity="error",
|
||||
code="unresolved_review_item",
|
||||
path=path,
|
||||
message=f"Review item {item.code} must be resolved before publication.",
|
||||
))
|
||||
return findings
|
||||
|
||||
|
||||
def _normalize(text: str) -> str:
|
||||
return unicodedata.normalize("NFC", text.replace("\r\n", "\n").replace("\r", "\n"))
|
||||
Reference in New Issue
Block a user