"""Offline source-to-citation provenance lab. This standard-library project intentionally supports only UTF-8 Markdown whose line-oriented parser can map extracted text back to exact canonical character spans. It demonstrates identities and publication gates; it is not a PDF/OCR, vector-search, identity-provider, distributed-job, or physical-erasure system. """ from __future__ import annotations import argparse import copy import hashlib import json import math import re import sys import unicodedata from dataclasses import asdict, dataclass, replace from pathlib import Path from typing import Any, Iterable, Sequence SCHEMA_VERSION = "1.0" PUBLIC_SCHEMA_VERSION = "provenance-public-v1" AUDIT_SCHEMA_VERSION = "provenance-audit-v1" ARTIFACT_SCHEMA_VERSION = "provenance-eval-v2" HARNESS_REVISION = "offline-provenance-harness-v2" COORDINATE_SPACE = "canonical-text-lf-nfc-char-v1" HEX_64 = re.compile(r"^[0-9a-f]{64}$") TOKEN_RE = re.compile(r"[A-Za-z0-9_]+|[\u4e00-\u9fff]") HEADING_RE = re.compile(r"^(#{1,6})[ \t]+(.+?)[ \t]*$") MIN_LEXICAL_SCORE = 4 MAX_FIXTURE_BYTES = 4_000_000 MAX_JSON_DEPTH = 64 MAX_SOURCE_BYTES = 100_000 MAX_SOURCE_EVENTS = 256 MAX_QUERIES = 256 MAX_LIST_ITEMS = 128 MAX_TOP_K = 20 MAX_PROCESSED_EVENTS = 10_000 class ContractError(ValueError): """Input, lineage, publication, or evidence contract violation.""" def _reject_constant(value: str) -> None: raise ContractError(f"JSON 不允许非有限数值：{value}") def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: # Avoid reflecting an untrusted key into terminal/error output. A # JSON-escaped lone surrogate cannot be emitted as strict UTF-8. raise ContractError("JSON 对象包含重复键") result[key] = value return result def _reject_invalid_unicode(value: Any) -> None: """Reject decoded JSON strings/keys that cannot become evidence UTF-8 bytes.""" if isinstance(value, str): try: value.encode("utf-8", errors="strict") except UnicodeEncodeError as exc: raise ContractError("JSON 含非法 Unicode") from exc return if isinstance(value, list): for item in value: _reject_invalid_unicode(item) return if isinstance(value, dict): for key, item in value.items(): _reject_invalid_unicode(key) _reject_invalid_unicode(item) def _reject_excessive_json_nesting(text: str) -> None: depth = 0 in_string = False escaped = False for character in text: if in_string: if escaped: escaped = False elif character == "\\": escaped = True elif character == '"': in_string = False continue if character == '"': in_string = True elif character in "[{": depth += 1 if depth > MAX_JSON_DEPTH: raise ContractError(f"JSON 容器嵌套不得超过 {MAX_JSON_DEPTH} 层") elif character in "]}": depth = max(0, depth - 1) def strict_json_loads(text: str) -> Any: if not isinstance(text, str): raise ContractError("JSON boundary 必须是文本") try: raw = text.encode("utf-8", errors="strict") except UnicodeEncodeError as exc: raise ContractError("JSON 含非法 Unicode") from exc if len(raw) > MAX_FIXTURE_BYTES: raise ContractError(f"JSON 不得超过 {MAX_FIXTURE_BYTES} UTF-8 bytes") _reject_excessive_json_nesting(text) try: parsed = json.loads( text, object_pairs_hook=_unique_object, parse_constant=_reject_constant, ) except ContractError: raise except json.JSONDecodeError as exc: raise ContractError(f"JSON 无法解析：{exc.msg}") from exc except RecursionError as exc: raise ContractError("JSON 嵌套超过解析器上限") from exc except ValueError as exc: # CPython can reject an oversized integer before allocating it. Keep # that parser-specific failure on the contract-error path. raise ContractError("JSON 数值超出解析器边界") from exc _reject_invalid_unicode(parsed) return parsed def canonical_json(value: Any) -> str: """Canonical JSON for this deliberately restricted teaching domain. This is not an RFC 8785 implementation. Values are first checked so the digest domain contains only JSON null/bool/int/string/list/object values; floats are rejected to avoid cross-runtime number ambiguity. """ def check(item: Any, path: str) -> None: if item is None or isinstance(item, bool): return if isinstance(item, str): try: item.encode("utf-8", errors="strict") except UnicodeEncodeError as exc: raise ContractError(f"{path} 含非法 Unicode") from exc return if isinstance(item, int) and not isinstance(item, bool): return if isinstance(item, float): if not math.isfinite(item): raise ContractError(f"{path} 含非有限数值") raise ContractError(f"{path} 含未纳入本项目规范化域的浮点数") if isinstance(item, list) or isinstance(item, tuple): for index, child in enumerate(item): check(child, f"{path}[{index}]") return if isinstance(item, dict): for key, child in item.items(): if not isinstance(key, str): raise ContractError(f"{path} 含非字符串键") try: key.encode("utf-8", errors="strict") except UnicodeEncodeError as exc: raise ContractError(f"{path} 含非法 Unicode 键") from exc check(child, f"{path}.{key}") return raise ContractError(f"{path} 含不支持的类型：{type(item).__name__}") check(value, "value") return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ) def digest_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def digest_text(value: str) -> str: try: return digest_bytes(value.encode("utf-8", errors="strict")) except UnicodeEncodeError as exc: raise ContractError("文本含非法 Unicode，不能计算 UTF-8 摘要") from exc def digest_object(value: Any) -> str: return digest_text(canonical_json(value)) def _exact_fields(value: Any, expected: set[str], label: str) -> dict[str, Any]: if not isinstance(value, dict): raise ContractError(f"{label} 必须是对象") actual = set(value) if actual != expected: missing = sorted(expected - actual) extra = sorted(actual - expected) raise ContractError(f"{label} 字段不匹配；缺失={missing}，多余={extra}") return value def _token(name: str, value: Any, maximum: int = 300) -> str: if not isinstance(value, str) or not value.strip(): raise ContractError(f"{name} 必须是非空字符串") cleaned = value.strip() if len(cleaned) > maximum or any(ord(char) < 32 for char in cleaned): raise ContractError(f"{name} 超长或包含控制字符") try: cleaned.encode("utf-8", errors="strict") except UnicodeEncodeError as exc: raise ContractError(f"{name} 含非法 Unicode") from exc return cleaned def _sha(name: str, value: Any) -> str: if not isinstance(value, str) or HEX_64.fullmatch(value) is None: raise ContractError(f"{name} 必须是完整的 64 位小写 SHA-256") return value def _positive_int(name: str, value: Any) -> int: if not isinstance(value, int) or isinstance(value, bool) or value <= 0: raise ContractError(f"{name} 必须是正整数") return value def _sorted_unique_strings( name: str, value: Any, *, nonempty: bool = True, maximum_items: int = MAX_LIST_ITEMS, ) -> tuple[str, ...]: if not isinstance(value, list): raise ContractError(f"{name} 必须是字符串数组") if len(value) > maximum_items: raise ContractError(f"{name} 不得超过 {maximum_items} 项") result = tuple(_token(f"{name}[]", item, 100) for item in value) if nonempty and not result: raise ContractError(f"{name} 不得为空") if tuple(sorted(set(result))) != result: raise ContractError(f"{name} 必须已排序且不得重复") return result def normalize_text(value: str) -> str: if not isinstance(value, str): raise ContractError("content 必须是字符串") if "\x00" in value: raise ContractError("content 不允许 NUL") try: encoded = value.encode("utf-8", errors="strict") except UnicodeEncodeError as exc: raise ContractError("content 含非法 Unicode") from exc if len(encoded) > MAX_SOURCE_BYTES: raise ContractError(f"content 不得超过 {MAX_SOURCE_BYTES} UTF-8 bytes") return unicodedata.normalize("NFC", value.replace("\r\n", "\n").replace("\r", "\n")) @dataclass(frozen=True) class PipelineContract: pipeline_revision: str normalizer_revision: str parser_revision: str parser_config: dict[str, Any] parser_config_sha256: str chunker_revision: str index_revision: str active_authorization_revision: str known_authorization_revisions: tuple[str, ...] pipeline_fingerprint: str @dataclass(frozen=True) class SourceEvent: event_id: str kind: str connector: str tenant_id: str document_id: str sequence: int source_uri: str source_version: str media_type: str content: str | None raw_sha256: str | None allowed_groups: tuple[str, ...] state_sha256: str @dataclass(frozen=True) class CanonicalRevision: tenant_id: str document_id: str source_event_id: str source_uri: str source_version: str raw_content: str raw_sha256: str canonical_text: str canonical_text_sha256: str normalizer_revision: str acl_snapshot_sha256: str allowed_groups: tuple[str, ...] canonical_revision_id: str @dataclass(frozen=True) class Element: element_id: str parse_revision_id: str kind: str order: int text: str text_sha256: str coordinate_space: str char_start: int char_end: int section_path: tuple[str, ...] @dataclass(frozen=True) class Chunk: chunk_id: str canonical_revision_id: str parse_revision_id: str chunker_revision: str element_id: str text: str content_sha256: str retrieval_text: str retrieval_sha256: str acl_snapshot_sha256: str coordinate_space: str char_start: int char_end: int @dataclass(frozen=True) class IndexEntry: index_entry_id: str tenant_id: str document_id: str canonical_revision_id: str parse_revision_id: str chunk_id: str element_id: str index_revision: str retrieval_text: str retrieval_sha256: str content_text: str content_sha256: str allowed_groups: tuple[str, ...] acl_snapshot_sha256: str coordinate_space: str char_start: int char_end: int @dataclass class DocumentState: tenant_id: str document_id: str source_sequence: int event_state_sha256: str canonical_revision_id: str | None allowed_groups: tuple[str, ...] acl_snapshot_sha256: str deleted: bool access_blocked: bool @dataclass class IndexGeneration: generation_id: str snapshot_state_sha256: str tombstone_state_sha256: str authorization_revision: str pipeline_fingerprint: str entries: tuple[IndexEntry, ...] entry_set_sha256: str manifest_sha256: str status: str errors: tuple[str, ...] def parse_contract(value: Any) -> PipelineContract: obj = _exact_fields( value, { "pipeline_revision", "normalizer_revision", "parser_revision", "parser_config", "chunker_revision", "index_revision", "authorization", }, "contract", ) parser_config = _exact_fields( obj["parser_config"], {"media_types", "coordinate_space", "line_oriented"}, "contract.parser_config", ) if parser_config["media_types"] != ["text/markdown"]: raise ContractError("本项目只声明 text/markdown") if parser_config["coordinate_space"] != COORDINATE_SPACE: raise ContractError("coordinate_space 与实现不匹配") if parser_config["line_oriented"] is not True: raise ContractError("line_oriented 必须为 true") authorization = _exact_fields( obj["authorization"], {"active_revision", "known_revisions"}, "contract.authorization", ) known = _sorted_unique_strings( "contract.authorization.known_revisions", authorization["known_revisions"] ) active = _token("contract.authorization.active_revision", authorization["active_revision"]) if active not in known: raise ContractError("active authorization revision 必须位于 known_revisions") base = { "pipeline_revision": _token("pipeline_revision", obj["pipeline_revision"]), "normalizer_revision": _token("normalizer_revision", obj["normalizer_revision"]), "parser_revision": _token("parser_revision", obj["parser_revision"]), "parser_config": parser_config, "chunker_revision": _token("chunker_revision", obj["chunker_revision"]), "index_revision": _token("index_revision", obj["index_revision"]), } return PipelineContract( pipeline_revision=base["pipeline_revision"], normalizer_revision=base["normalizer_revision"], parser_revision=base["parser_revision"], parser_config=copy.deepcopy(parser_config), parser_config_sha256=digest_object(parser_config), chunker_revision=base["chunker_revision"], index_revision=base["index_revision"], active_authorization_revision=active, known_authorization_revisions=known, pipeline_fingerprint=digest_object(base), ) def parse_source_event(value: Any) -> SourceEvent: obj = _exact_fields( value, { "event_id", "kind", "connector", "tenant_id", "document_id", "sequence", "source_uri", "source_version", "media_type", "content", "raw_sha256", "allowed_groups", }, "source_event", ) kind = _token("source_event.kind", obj["kind"]) if kind not in {"upsert", "delete"}: raise ContractError("source_event.kind 只能是 upsert/delete") media_type = _token("source_event.media_type", obj["media_type"]) if media_type != "text/markdown": raise ContractError("本项目只接受 text/markdown") allowed_groups = _sorted_unique_strings( "source_event.allowed_groups", obj["allowed_groups"], nonempty=kind == "upsert" ) content = obj["content"] raw_sha = obj["raw_sha256"] if kind == "upsert": if not isinstance(content, str): raise ContractError("upsert.content 必须是字符串") normalize_text(content) declared = _sha("source_event.raw_sha256", raw_sha) actual = digest_text(content) if declared != actual: raise ContractError("source_event.raw_sha256 与 UTF-8 原文不匹配") raw_sha = declared elif content is not None or raw_sha is not None or allowed_groups: raise ContractError("delete 必须令 content/raw_sha256 为 null 且 allowed_groups 为空") state = { "kind": kind, "connector": _token("connector", obj["connector"]), "tenant_id": _token("tenant_id", obj["tenant_id"]), "document_id": _token("document_id", obj["document_id"]), "source_uri": _token("source_uri", obj["source_uri"]), "source_version": _token("source_version", obj["source_version"]), "media_type": media_type, "content": content, "raw_sha256": raw_sha, "allowed_groups": list(allowed_groups), } return SourceEvent( event_id=_token("event_id", obj["event_id"]), kind=kind, connector=state["connector"], tenant_id=state["tenant_id"], document_id=state["document_id"], sequence=_positive_int("source_event.sequence", obj["sequence"]), source_uri=state["source_uri"], source_version=state["source_version"], media_type=media_type, content=content, raw_sha256=raw_sha, allowed_groups=allowed_groups, state_sha256=digest_object(state), ) def validate_query(value: Any) -> dict[str, Any]: obj = _exact_fields( value, { "query_id", "query", "tenant_id", "subject_groups", "authorization_revision", "top_k", }, "query", ) result = { "query_id": _token("query.query_id", obj["query_id"]), "query": _token("query.query", obj["query"], 2000), "tenant_id": _token("query.tenant_id", obj["tenant_id"]), "subject_groups": list( _sorted_unique_strings( "query.subject_groups", obj["subject_groups"], maximum_items=MAX_LIST_ITEMS, ) ), "authorization_revision": _token( "query.authorization_revision", obj["authorization_revision"] ), "top_k": _positive_int("query.top_k", obj["top_k"]), } if result["top_k"] > MAX_TOP_K: raise ContractError(f"query.top_k 不得超过 {MAX_TOP_K}") return result def load_fixture(path: Path) -> dict[str, Any]: try: with path.open("rb") as handle: raw_bytes = handle.read(MAX_FIXTURE_BYTES + 1) if len(raw_bytes) > MAX_FIXTURE_BYTES: raise ContractError( f"fixture 不得超过 {MAX_FIXTURE_BYTES} UTF-8 bytes" ) raw = raw_bytes.decode("utf-8", errors="strict") except ContractError: raise except (OSError, UnicodeError) as exc: raise ContractError(f"无法读取 fixture：{type(exc).__name__}") from exc fixture = _exact_fields( strict_json_loads(raw), {"schema_version", "contract", "source_events", "queries", "oracle"}, "fixture", ) if fixture["schema_version"] != SCHEMA_VERSION: raise ContractError(f"只支持 fixture schema {SCHEMA_VERSION}") contract = parse_contract(fixture["contract"]) if ( not isinstance(fixture["source_events"], list) or not 1 <= len(fixture["source_events"]) <= MAX_SOURCE_EVENTS ): raise ContractError(f"source_events 必须是 1..{MAX_SOURCE_EVENTS} 项的数组") events = [parse_source_event(item) for item in fixture["source_events"]] event_ids = [item.event_id for item in events] if len(event_ids) != len(set(event_ids)): raise ContractError("source_event.event_id 不得重复") if ( not isinstance(fixture["queries"], list) or not 1 <= len(fixture["queries"]) <= MAX_QUERIES ): raise ContractError(f"queries 必须是 1..{MAX_QUERIES} 项的数组") queries = [validate_query(item) for item in fixture["queries"]] query_ids = [item["query_id"] for item in queries] if len(query_ids) != len(set(query_ids)): raise ContractError("query_id 不得重复") if not isinstance(fixture["oracle"], list): raise ContractError("oracle 必须是数组") oracle: list[dict[str, Any]] = [] for item in fixture["oracle"]: entry = _exact_fields( item, { "query_id", "expected_status", "expected_documents", "forbidden_documents", "critical", "slice", }, "oracle[]", ) query_id = _token("oracle.query_id", entry["query_id"]) if query_id not in query_ids: raise ContractError(f"oracle 引用了未知 query_id：{query_id}") if not isinstance(entry["critical"], bool): raise ContractError("oracle.critical 必须是布尔值") expected_status = _token("oracle.expected_status", entry["expected_status"]) if expected_status not in {"answered", "insufficient_evidence"}: raise ContractError("oracle.expected_status 不受支持") oracle.append( { "query_id": query_id, "expected_status": expected_status, "expected_documents": list( _sorted_unique_strings( "oracle.expected_documents", entry["expected_documents"], nonempty=False, ) ), "forbidden_documents": list( _sorted_unique_strings( "oracle.forbidden_documents", entry["forbidden_documents"], nonempty=False, ) ), "critical": entry["critical"], "slice": _token("oracle.slice", entry["slice"]), } ) if sorted(item["query_id"] for item in oracle) != sorted(query_ids): raise ContractError("每个 query 必须恰有一个 oracle") result = { "schema_version": SCHEMA_VERSION, "contract": contract, "source_events": events, "queries": queries, "oracle": oracle, "fixture_sha256": digest_text(raw), } result["fixture_model_sha256"] = fixture_model_sha256(result) return result def fixture_model_sha256(fixture: dict[str, Any]) -> str: contract = fixture.get("contract") events = fixture.get("source_events") if not isinstance(contract, PipelineContract) or not isinstance(events, list): raise ContractError("fixture model 尚未通过类型化加载") return digest_object( { "schema_version": fixture.get("schema_version"), "contract": asdict(contract), "source_events": [asdict(item) for item in events], "queries": fixture.get("queries"), "oracle": fixture.get("oracle"), } ) def require_fixture_integrity(fixture: dict[str, Any]) -> None: supplied = fixture.get("fixture_model_sha256") if not isinstance(supplied, str) or HEX_64.fullmatch(supplied) is None: raise ContractError("fixture_model_sha256 缺失或非法") if supplied != fixture_model_sha256(fixture): raise ContractError("fixture 在严格加载后被修改；必须重新验证") def _acl_hash(groups: Iterable[str]) -> str: return digest_object({"allowed_groups": sorted(groups)}) def _canonical_revision(event: SourceEvent, contract: PipelineContract) -> CanonicalRevision: if event.content is None or event.raw_sha256 is None: raise ContractError("delete 不能生成 canonical revision") canonical = normalize_text(event.content) canonical_hash = digest_text(canonical) acl_hash = _acl_hash(event.allowed_groups) identity = { "tenant_id": event.tenant_id, "document_id": event.document_id, "source_uri": event.source_uri, "source_version": event.source_version, "raw_sha256": event.raw_sha256, "canonical_text_sha256": canonical_hash, "normalizer_revision": contract.normalizer_revision, "acl_snapshot_sha256": acl_hash, } revision_id = "can_" + digest_object(identity) return CanonicalRevision( tenant_id=event.tenant_id, document_id=event.document_id, source_event_id=event.event_id, source_uri=event.source_uri, source_version=event.source_version, raw_content=event.content, raw_sha256=event.raw_sha256, canonical_text=canonical, canonical_text_sha256=canonical_hash, normalizer_revision=contract.normalizer_revision, acl_snapshot_sha256=acl_hash, allowed_groups=event.allowed_groups, canonical_revision_id=revision_id, ) def _line_spans(text: str) -> Iterable[tuple[str, int, int]]: cursor = 0 for line in text.splitlines(keepends=True): body = line[:-1] if line.endswith("\n") else line yield body, cursor, cursor + len(body) cursor += len(line) if not text: return if text.endswith("\n"): return def parse_elements( revision: CanonicalRevision, contract: PipelineContract ) -> tuple[str, tuple[Element, ...]]: parse_identity = { "canonical_revision_id": revision.canonical_revision_id, "parser_revision": contract.parser_revision, "parser_config_sha256": contract.parser_config_sha256, } parse_revision_id = "par_" + digest_object(parse_identity) sections: list[str] = [] elements: list[Element] = [] for line, line_start, _line_end in _line_spans(revision.canonical_text): if not line.strip(): continue match = HEADING_RE.fullmatch(line) if match: level = len(match.group(1)) text = match.group(2) relative_start = match.start(2) kind = "heading" sections = sections[: level - 1] sections.append(text) section_path = tuple(sections) else: leading = len(line) - len(line.lstrip(" \t")) trailing = len(line.rstrip(" \t")) text = line[leading:trailing] relative_start = leading kind = "paragraph" section_path = tuple(sections) char_start = line_start + relative_start char_end = char_start + len(text) text_hash = digest_text(text) identity = { "parse_revision_id": parse_revision_id, "kind": kind, "coordinate_space": COORDINATE_SPACE, "char_start": char_start, "char_end": char_end, "text_sha256": text_hash, } elements.append( Element( element_id="el_" + digest_object(identity), parse_revision_id=parse_revision_id, kind=kind, order=len(elements) + 1, text=text, text_sha256=text_hash, coordinate_space=COORDINATE_SPACE, char_start=char_start, char_end=char_end, section_path=section_path, ) ) if not elements or not any(item.kind == "paragraph" for item in elements): raise ContractError("解析结果必须至少包含一个可引用段落") return parse_revision_id, tuple(elements) def build_chunks( revision: CanonicalRevision, parse_revision_id: str, elements: Sequence[Element], contract: PipelineContract, ) -> tuple[Chunk, ...]: chunks: list[Chunk] = [] for element in elements: if element.kind != "paragraph": continue context = " > ".join(element.section_path) retrieval_text = f"标题路径：{context}\n{element.text}" if context else element.text content_hash = digest_text(element.text) retrieval_hash = digest_text(retrieval_text) identity = { "canonical_revision_id": revision.canonical_revision_id, "parse_revision_id": parse_revision_id, "chunker_revision": contract.chunker_revision, "element_id": element.element_id, "content_sha256": content_hash, "acl_snapshot_sha256": revision.acl_snapshot_sha256, } chunks.append( Chunk( chunk_id="chk_" + digest_object(identity), canonical_revision_id=revision.canonical_revision_id, parse_revision_id=parse_revision_id, chunker_revision=contract.chunker_revision, element_id=element.element_id, text=element.text, content_sha256=content_hash, retrieval_text=retrieval_text, retrieval_sha256=retrieval_hash, acl_snapshot_sha256=revision.acl_snapshot_sha256, coordinate_space=COORDINATE_SPACE, char_start=element.char_start, char_end=element.char_end, ) ) return tuple(chunks) def build_entries( revision: CanonicalRevision, parse_revision_id: str, chunks: Sequence[Chunk], contract: PipelineContract, ) -> tuple[IndexEntry, ...]: entries: list[IndexEntry] = [] for chunk in chunks: identity = { "chunk_id": chunk.chunk_id, "retrieval_sha256": chunk.retrieval_sha256, "index_revision": contract.index_revision, "acl_snapshot_sha256": revision.acl_snapshot_sha256, } entries.append( IndexEntry( index_entry_id="idx_" + digest_object(identity), tenant_id=revision.tenant_id, document_id=revision.document_id, canonical_revision_id=revision.canonical_revision_id, parse_revision_id=parse_revision_id, chunk_id=chunk.chunk_id, element_id=chunk.element_id, index_revision=contract.index_revision, retrieval_text=chunk.retrieval_text, retrieval_sha256=chunk.retrieval_sha256, content_text=chunk.text, content_sha256=chunk.content_sha256, allowed_groups=revision.allowed_groups, acl_snapshot_sha256=revision.acl_snapshot_sha256, coordinate_space=chunk.coordinate_space, char_start=chunk.char_start, char_end=chunk.char_end, ) ) return tuple(entries) def _snapshot_rows(documents: dict[tuple[str, str], DocumentState]) -> list[dict[str, Any]]: return [ { "tenant_id": state.tenant_id, "document_id": state.document_id, "source_sequence": state.source_sequence, "canonical_revision_id": state.canonical_revision_id, "acl_snapshot_sha256": state.acl_snapshot_sha256, "deleted": state.deleted, } for _key, state in sorted(documents.items()) ] class ProvenanceEngine: def __init__(self, contract: PipelineContract): self.contract = contract self.active_authorization_revision = contract.active_authorization_revision self.documents: dict[tuple[str, str], DocumentState] = {} self.revisions: dict[str, CanonicalRevision] = {} self.elements: dict[str, tuple[Element, ...]] = {} self.chunks: dict[str, tuple[Chunk, ...]] = {} self.tombstones: dict[tuple[str, str], int] = {} self.generations: dict[str, IndexGeneration] = {} self.published_generation_id: str | None = None self.processed_event_sha256: dict[str, str] = {} def apply_event(self, event_or_value: SourceEvent | dict[str, Any]) -> str: event = ( event_or_value if isinstance(event_or_value, SourceEvent) else parse_source_event(event_or_value) ) event_fingerprint = digest_object(asdict(event)) previous_event_fingerprint = self.processed_event_sha256.get(event.event_id) if previous_event_fingerprint is not None: if previous_event_fingerprint == event_fingerprint: return "noop" raise ContractError("相同 event_id 不能绑定不同来源事件") if len(self.processed_event_sha256) >= MAX_PROCESSED_EVENTS: raise ContractError( "processed event 去重窗口已满；必须持久化/轮换后再接收新事件" ) def remember(action: str) -> str: self.processed_event_sha256[event.event_id] = event_fingerprint return action key = (event.tenant_id, event.document_id) previous = self.documents.get(key) if previous is not None: if event.sequence < previous.source_sequence: return remember("stale_ignored") if event.sequence == previous.source_sequence: if event.state_sha256 == previous.event_state_sha256: return remember("noop") raise ContractError("相同 source sequence 对应不同状态") if event.state_sha256 == previous.event_state_sha256: previous.source_sequence = event.sequence return remember("checkpoint_advanced") if event.kind == "delete": if previous is None: acl_hash = _acl_hash(()) self.documents[key] = DocumentState( event.tenant_id, event.document_id, event.sequence, event.state_sha256, None, (), acl_hash, True, True, ) else: previous.source_sequence = event.sequence previous.event_state_sha256 = event.state_sha256 previous.deleted = True previous.access_blocked = True previous.allowed_groups = () previous.acl_snapshot_sha256 = _acl_hash(()) self.tombstones[key] = event.sequence return remember("deleted") revision = _canonical_revision(event, self.contract) self.revisions.setdefault(revision.canonical_revision_id, revision) acl_changed = previous is not None and previous.acl_snapshot_sha256 != revision.acl_snapshot_sha256 resurrected = previous is not None and previous.deleted already_blocked = previous is not None and previous.access_blocked state = DocumentState( tenant_id=event.tenant_id, document_id=event.document_id, source_sequence=event.sequence, event_state_sha256=event.state_sha256, canonical_revision_id=revision.canonical_revision_id, allowed_groups=revision.allowed_groups, acl_snapshot_sha256=revision.acl_snapshot_sha256, deleted=False, access_blocked=acl_changed or resurrected or already_blocked, ) self.documents[key] = state return remember("upserted") def capture_snapshot(self) -> dict[str, Any]: rows = _snapshot_rows(self.documents) tombstones = [ {"tenant_id": key[0], "document_id": key[1], "sequence": sequence} for key, sequence in sorted(self.tombstones.items()) ] return { "documents": copy.deepcopy(rows), "snapshot_state_sha256": digest_object(rows), "tombstone_state_sha256": digest_object(tombstones), } def rotate_authorization(self, revision: str) -> None: revision = _token("authorization_revision", revision) if revision not in self.contract.known_authorization_revisions: raise ContractError("未知 authorization revision") self.active_authorization_revision = revision def stage_generation( self, *, authorization_revision: str | None = None, snapshot: dict[str, Any] | None = None, failure: str | None = None, ) -> IndexGeneration: auth = authorization_revision or self.active_authorization_revision if auth not in self.contract.known_authorization_revisions: raise ContractError("未知 authorization revision") captured = copy.deepcopy(snapshot) if snapshot is not None else self.capture_snapshot() _exact_fields( captured, {"documents", "snapshot_state_sha256", "tombstone_state_sha256"}, "snapshot", ) _sha("snapshot_state_sha256", captured["snapshot_state_sha256"]) _sha("tombstone_state_sha256", captured["tombstone_state_sha256"]) if digest_object(captured["documents"]) != captured["snapshot_state_sha256"]: raise ContractError("snapshot documents 与 snapshot hash 不匹配") entries: list[IndexEntry] = [] errors: list[str] = [] for row in captured["documents"]: row = _exact_fields( row, { "tenant_id", "document_id", "source_sequence", "canonical_revision_id", "acl_snapshot_sha256", "deleted", }, "snapshot.documents[]", ) if row["deleted"]: continue revision_id = row["canonical_revision_id"] if not isinstance(revision_id, str) or revision_id not in self.revisions: errors.append("missing_canonical_revision") continue revision = self.revisions[revision_id] parse_id, elements = parse_elements(revision, self.contract) chunks = build_chunks(revision, parse_id, elements, self.contract) document_entries = build_entries(revision, parse_id, chunks, self.contract) self.elements[revision_id] = elements self.chunks[revision_id] = chunks for entry in document_entries: entries.append(entry) if failure == "after_n_index_entries" and len(entries) == 1: errors.append("partial_index_build") break if errors: break entries.sort(key=lambda item: item.index_entry_id) entry_set_hash = digest_object([item.index_entry_id for item in entries]) generation_identity = { "snapshot_state_sha256": captured["snapshot_state_sha256"], "tombstone_state_sha256": captured["tombstone_state_sha256"], "authorization_revision": auth, "pipeline_fingerprint": self.contract.pipeline_fingerprint, "entry_set_sha256": entry_set_hash, } generation_id = "gen_" + digest_object(generation_identity) manifest = { **generation_identity, "generation_id": generation_id, "entry_ids": [item.index_entry_id for item in entries], } generation = IndexGeneration( generation_id=generation_id, snapshot_state_sha256=captured["snapshot_state_sha256"], tombstone_state_sha256=captured["tombstone_state_sha256"], authorization_revision=auth, pipeline_fingerprint=self.contract.pipeline_fingerprint, entries=tuple(entries), entry_set_sha256=entry_set_hash, manifest_sha256=digest_object(manifest), status="failed" if errors else "staged", errors=tuple(errors), ) self.generations[generation_id] = generation return generation def _entry_errors(self, entry: IndexEntry) -> list[str]: errors: list[str] = [] revision = self.revisions.get(entry.canonical_revision_id) if revision is None: return ["missing_canonical_revision"] if (entry.tenant_id, entry.document_id) != ( revision.tenant_id, revision.document_id, ): errors.append("entry_document_binding_mismatch") if ( entry.allowed_groups != revision.allowed_groups or entry.acl_snapshot_sha256 != revision.acl_snapshot_sha256 ): errors.append("entry_acl_revision_mismatch") if digest_text(revision.raw_content) != revision.raw_sha256: errors.append("canonical_raw_hash_mismatch") if digest_text(revision.canonical_text) != revision.canonical_text_sha256: errors.append("canonical_text_hash_mismatch") actual_text = revision.canonical_text[entry.char_start : entry.char_end] if entry.coordinate_space != COORDINATE_SPACE: errors.append("bad_coordinate_space") if not 0 <= entry.char_start < entry.char_end <= len(revision.canonical_text): errors.append("source_span_out_of_bounds") actual_text = "" if actual_text != entry.content_text: errors.append("source_span_text_mismatch") if digest_text(entry.content_text) != entry.content_sha256: errors.append("entry_content_hash_mismatch") if digest_text(entry.retrieval_text) != entry.retrieval_sha256: errors.append("entry_retrieval_hash_mismatch") if _acl_hash(entry.allowed_groups) != entry.acl_snapshot_sha256: errors.append("entry_acl_hash_mismatch") expected_id = "idx_" + digest_object( { "chunk_id": entry.chunk_id, "retrieval_sha256": entry.retrieval_sha256, "index_revision": entry.index_revision, "acl_snapshot_sha256": entry.acl_snapshot_sha256, } ) if expected_id != entry.index_entry_id: errors.append("index_entry_id_mismatch") matching_chunks = [ item for item in self.chunks.get(entry.canonical_revision_id, ()) if item.chunk_id == entry.chunk_id ] if len(matching_chunks) != 1: errors.append("chunk_lineage_missing") else: chunk = matching_chunks[0] if ( chunk.parse_revision_id != entry.parse_revision_id or chunk.element_id != entry.element_id or chunk.text != entry.content_text or chunk.content_sha256 != entry.content_sha256 or chunk.retrieval_text != entry.retrieval_text or chunk.retrieval_sha256 != entry.retrieval_sha256 or chunk.char_start != entry.char_start or chunk.char_end != entry.char_end ): errors.append("chunk_entry_binding_mismatch") return errors def _generation_identity_errors(self, generation: IndexGeneration) -> list[str]: errors: list[str] = [] entry_ids = sorted(item.index_entry_id for item in generation.entries) actual_entry_set = digest_object(entry_ids) if actual_entry_set != generation.entry_set_sha256: errors.append("entry_set_hash_mismatch") identity = { "snapshot_state_sha256": generation.snapshot_state_sha256, "tombstone_state_sha256": generation.tombstone_state_sha256, "authorization_revision": generation.authorization_revision, "pipeline_fingerprint": generation.pipeline_fingerprint, "entry_set_sha256": generation.entry_set_sha256, } expected_generation_id = "gen_" + digest_object(identity) if expected_generation_id != generation.generation_id: errors.append("generation_id_mismatch") manifest = { **identity, "generation_id": generation.generation_id, "entry_ids": entry_ids, } if digest_object(manifest) != generation.manifest_sha256: errors.append("generation_manifest_hash_mismatch") return errors def validate_generation(self, generation: IndexGeneration) -> list[str]: errors = list(generation.errors) current = self.capture_snapshot() if generation.snapshot_state_sha256 != current["snapshot_state_sha256"]: errors.append("stale_snapshot") if generation.tombstone_state_sha256 != current["tombstone_state_sha256"]: errors.append("stale_tombstone_state") if generation.authorization_revision != self.active_authorization_revision: errors.append("authorization_snapshot_mismatch") if generation.pipeline_fingerprint != self.contract.pipeline_fingerprint: errors.append("pipeline_fingerprint_mismatch") errors.extend(self._generation_identity_errors(generation)) active_documents = { (state.tenant_id, state.document_id) for state in self.documents.values() if not state.deleted } indexed_documents = { (entry.tenant_id, entry.document_id) for entry in generation.entries } if indexed_documents != active_documents: errors.append("active_document_coverage_mismatch") expected_entry_ids: list[str] = [] for state in self.documents.values(): if state.deleted or state.canonical_revision_id is None: continue revision = self.revisions.get(state.canonical_revision_id) if revision is None: errors.append("current_canonical_revision_missing") continue parse_id, elements = parse_elements(revision, self.contract) chunks = build_chunks(revision, parse_id, elements, self.contract) expected_entry_ids.extend( item.index_entry_id for item in build_entries(revision, parse_id, chunks, self.contract) ) if sorted(expected_entry_ids) != sorted( item.index_entry_id for item in generation.entries ): errors.append("current_entry_set_mismatch") for entry in generation.entries: errors.extend(self._entry_errors(entry)) return sorted(set(errors)) def publish_generation(self, generation_id: str) -> None: generation = self.generations.get(generation_id) if generation is None: raise ContractError("未知 index generation") if generation.status != "staged": raise ContractError("只有完整 staged generation 可以发布") errors = self.validate_generation(generation) if errors: generation.status = "blocked" generation.errors = tuple(errors) raise ContractError("generation 发布门禁失败：" + ",".join(errors)) previous_generation_id = self.published_generation_id if previous_generation_id is not None and previous_generation_id != generation_id: previous = self.generations.get(previous_generation_id) if previous is not None and previous.status == "published": previous.status = "superseded" generation.status = "published" self.published_generation_id = generation_id published_pairs = { (entry.tenant_id, entry.document_id): ( entry.canonical_revision_id, entry.acl_snapshot_sha256, ) for entry in generation.entries } for key, state in self.documents.items(): pair = published_pairs.get(key) if ( not state.deleted and pair is not None and pair == (state.canonical_revision_id, state.acl_snapshot_sha256) ): state.access_blocked = False def build_and_publish(self) -> str: generation = self.stage_generation() self.publish_generation(generation.generation_id) return generation.generation_id def _published(self) -> IndexGeneration: if self.published_generation_id is None: raise ContractError("尚无已发布 index generation") generation = self.generations[self.published_generation_id] if generation.status != "published": raise ContractError("published pointer 未指向已发布 generation") integrity_errors = self._generation_identity_errors(generation) for entry in generation.entries: integrity_errors.extend(self._entry_errors(entry)) if integrity_errors: raise ContractError( "published generation integrity failed：" + ",".join(sorted(set(integrity_errors))) ) return generation @staticmethod def _features(text: str) -> set[str]: units = [match.group(0).lower() for match in TOKEN_RE.finditer(text)] # 提取透明 token，并统一英文大小写。 result = set(units) # 先加入单个 token，集合会自动去重。 for left, right in zip(units, units[1:]): # 再查看相邻 token，构造最小的中文二字符特征。 if len(left) == 1 and len(right) == 1: # 只组合两个单字符，避免把英文词强行拼接。 result.add(left + right) # 加入二字符特征以保留局部顺序。 return result # 返回只用于教学检索的透明特征集合。 def _select_entries( self, query: dict[str, Any], generation: IndexGeneration ) -> tuple[list[IndexEntry], int, list[IndexEntry]]: subject_groups = set(query["subject_groups"]) # 从已验证 runtime query 取出主体的可信组集合。 visible: list[IndexEntry] = [] # 保存通过所有实时授权与生命周期检查的索引 entry。 filtered = 0 # 仅用于受保护审计的过滤计数。 for entry in generation.entries: # 对当前 published generation 的每个 entry 重新检查 live state。 state = self.documents.get((entry.tenant_id, entry.document_id)) # 读取对应 canonical document 的最新状态。 allowed = ( # 所有条件必须同时成立，任何缺失状态都 fail closed。 state is not None and not state.deleted and not state.access_blocked and entry.tenant_id == query["tenant_id"] and bool(subject_groups.intersection(entry.allowed_groups)) and state.acl_snapshot_sha256 == entry.acl_snapshot_sha256 ) if allowed: # 只有 live document、同 tenant、ACL 命中且快照未漂移时才可看见。 visible.append(entry) # 加入唯一允许被评分的集合。 else: # 未授权、已删除或快照漂移的 entry 都不会参与检索。 filtered += 1 # 只累计受保护诊断数，不写入公共响应。 query_features = self._features(query["query"]) # 解析 query 的透明教学特征。 ranked: list[tuple[int, IndexEntry]] = [] # 暂存“词法分数、entry”对。 for entry in visible: # 只遍历上面已完成实时安全复核的 entries。 score = len(query_features.intersection(self._features(entry.retrieval_text))) # 计算特征交集大小。 if score >= MIN_LEXICAL_SCORE: # 低于最小分数的 entry 不占用有限 top-k 窗口。 ranked.append((score, entry)) # 保存候选，稍后统一稳定排序。 ranked.sort(key=lambda item: (-item[0], item[1].index_entry_id)) # 分数降序、稳定 ID 升序打破平分。 selected = [entry for _score, entry in ranked[: query["top_k"]]] # 应用 query 中已经验证的 top-k 预算。 return visible, filtered, selected # 同时返回可见集合、内部计数和真正被引用的 entries。 def query( self, value: dict[str, Any], *, failure: str | None = None ) -> tuple[dict[str, Any], dict[str, Any]]: query = validate_query(value) # 只接受严格 runtime query，oracle 字段不能进入此 API。 if failure == "retrieval_unavailable": # 依赖不可用时不能绕过检索自由生成事实。 status = "dependency_unavailable" # 向公共层给出不泄露来源的通用状态。 claims: list[dict[str, Any]] = [] # 没有检索证据就不能输出事实 claim。 public = { # 形成最小公共投影，不包含 generation、ACL 或过滤计数。 "schema_version": PUBLIC_SCHEMA_VERSION, "query_id": query["query_id"], "status": status, "claims": claims, "trace_id": digest_object({"query": query, "status": status, "claims": claims}), } audit = { # 受保护审计可记录本次故障和已签发的授权 revision。 "schema_version": AUDIT_SCHEMA_VERSION, "visibility": "protected_audit", "trace_id": public["trace_id"], "authorization_revision": query["authorization_revision"], "index_generation_id": self.published_generation_id, "selected_entry_ids": [], "filter_summary": {"visible": 0, "filtered": 0}, "failure": "retrieval_unavailable", } return public, audit # 故障分支在不触发检索的前提下返回双投影。 generation = self._published() # 读取当前已发布 generation；未发布或失效状态会在内部拒绝。 if ( # query 与当前 generation 都必须绑定当前 active authorization revision。 query["authorization_revision"] != self.active_authorization_revision or generation.authorization_revision != self.active_authorization_revision ): status = "insufficient_evidence" # 不区分“授权快照错配”与无资料，避免公共侧信道。 claims = [] # 快照不一致时不能使用已发布 entry 生成 claim。 public = { # 仍以固定公共 schema 返回。 "schema_version": PUBLIC_SCHEMA_VERSION, "query_id": query["query_id"], "status": status, "claims": claims, "trace_id": digest_object({"query": query, "status": status, "claims": claims}), } audit = { "schema_version": AUDIT_SCHEMA_VERSION, "visibility": "protected_audit", "trace_id": public["trace_id"], "authorization_revision": query["authorization_revision"], "index_generation_id": generation.generation_id, "selected_entry_ids": [], "filter_summary": {"visible": 0, "filtered": len(generation.entries)}, "failure": "authorization_snapshot_mismatch", } return public, audit visible, filtered, selected = self._select_entries(query, generation) claims: list[dict[str, Any]] = [] for index, entry in enumerate(selected, start=1): revision = self.revisions[entry.canonical_revision_id] citation = { "document_id": entry.document_id, "source_uri": revision.source_uri, "source_version": revision.source_version, "raw_sha256": revision.raw_sha256, "canonical_revision_id": revision.canonical_revision_id, "parse_revision_id": entry.parse_revision_id, "element_id": entry.element_id, "chunk_id": entry.chunk_id, "index_entry_id": entry.index_entry_id, "coordinate_space": entry.coordinate_space, "char_start": entry.char_start, "char_end": entry.char_end, "span_sha256": digest_text(entry.content_text), } claims.append( { "claim_id": f"C{index}", "text": entry.content_text, "citations": [citation], } ) status = "answered" if claims else "insufficient_evidence" trace_id = digest_object( { "query": query, "status": status, "claims": claims, } ) public = { "schema_version": PUBLIC_SCHEMA_VERSION, "query_id": query["query_id"], "status": status, "claims": claims, "trace_id": trace_id, } audit = { "schema_version": AUDIT_SCHEMA_VERSION, "visibility": "protected_audit", "trace_id": trace_id, "authorization_revision": query["authorization_revision"], "index_generation_id": generation.generation_id, "selected_entry_ids": [entry.index_entry_id for entry in selected], "filter_summary": {"visible": len(visible), "filtered": filtered}, "failure": None, } evidence_errors = self.validate_evidence(query, public, audit) if evidence_errors: raise ContractError("evidence validation failed：" + ",".join(evidence_errors)) return public, audit def validate_evidence( self, query_value: dict[str, Any], public: Any, audit: Any, ) -> list[str]: errors: list[str] = [] try: query = validate_query(query_value) public_obj = _exact_fields( public, {"schema_version", "query_id", "status", "claims", "trace_id"}, "public", ) audit_obj = _exact_fields( audit, { "schema_version", "visibility", "trace_id", "authorization_revision", "index_generation_id", "selected_entry_ids", "filter_summary", "failure", }, "audit", ) except ContractError as exc: return [str(exc)] if public_obj["schema_version"] != PUBLIC_SCHEMA_VERSION: errors.append("public_schema_mismatch") if audit_obj["schema_version"] != AUDIT_SCHEMA_VERSION: errors.append("audit_schema_mismatch") if audit_obj["visibility"] != "protected_audit": errors.append("audit_visibility_mismatch") if public_obj["query_id"] != query["query_id"]: errors.append("query_id_mismatch") if public_obj["status"] not in { "answered", "insufficient_evidence", "dependency_unavailable", }: errors.append("public_status_invalid") if public_obj["trace_id"] != audit_obj["trace_id"]: errors.append("trace_id_mismatch") if audit_obj["authorization_revision"] != query["authorization_revision"]: errors.append("audit_authorization_revision_mismatch") generation_id = audit_obj["index_generation_id"] if ( not isinstance(generation_id, str) or not generation_id.startswith("gen_") or HEX_64.fullmatch(generation_id.removeprefix("gen_")) is None ): errors.append("index_generation_id_invalid") return sorted(set(errors)) generation = self.generations.get(generation_id) if generation is None or generation.status != "published": errors.append("generation_not_published") return sorted(set(errors)) if generation_id != self.published_generation_id: errors.append("generation_not_current") return sorted(set(errors)) selected_ids = audit_obj["selected_entry_ids"] if not isinstance(selected_ids, list) or any( not isinstance(item, str) for item in selected_ids ): errors.append("selected_entry_ids_invalid") return sorted(set(errors)) failure = audit_obj["failure"] if failure not in { None, "retrieval_unavailable", "authorization_snapshot_mismatch", }: errors.append("audit_failure_invalid") entry_by_id = {entry.index_entry_id: entry for entry in generation.entries} claims = public_obj["claims"] if not isinstance(claims, list): errors.append("claims_invalid") return sorted(set(errors)) if public_obj["status"] == "answered" and not claims: errors.append("answered_without_claims") if public_obj["status"] != "answered" and claims: errors.append("non_answered_with_claims") if failure is None: if query["authorization_revision"] != self.active_authorization_revision: errors.append("query_authorization_not_active") if generation.authorization_revision != self.active_authorization_revision: errors.append("generation_authorization_not_active") if generation.authorization_revision != audit_obj["authorization_revision"]: errors.append("generation_authorization_revision_mismatch") elif failure == "retrieval_unavailable": if public_obj["status"] != "dependency_unavailable": errors.append("retrieval_failure_status_mismatch") if claims or selected_ids: errors.append("retrieval_failure_must_not_return_evidence") if audit_obj["filter_summary"] != {"visible": 0, "filtered": 0}: errors.append("retrieval_failure_filter_summary_mismatch") elif failure == "authorization_snapshot_mismatch": authorization_is_stale = ( query["authorization_revision"] != self.active_authorization_revision or generation.authorization_revision != self.active_authorization_revision ) if not authorization_is_stale: errors.append("authorization_failure_without_mismatch") if public_obj["status"] != "insufficient_evidence": errors.append("authorization_failure_status_mismatch") if claims or selected_ids: errors.append("authorization_failure_must_not_return_evidence") if audit_obj["filter_summary"] != { "visible": 0, "filtered": len(generation.entries), }: errors.append("authorization_failure_filter_summary_mismatch") cited_entry_ids: list[str] = [] for claim in claims: if not isinstance(claim, dict) or set(claim) != {"claim_id", "text", "citations"}: errors.append("claim_schema_mismatch") continue if not isinstance(claim["text"], str) or not claim["text"]: errors.append("claim_text_invalid") continue citations = claim["citations"] if not isinstance(citations, list) or not citations: errors.append("citation_missing") continue for citation in citations: expected_fields = { "document_id", "source_uri", "source_version", "raw_sha256", "canonical_revision_id", "parse_revision_id", "element_id", "chunk_id", "index_entry_id", "coordinate_space", "char_start", "char_end", "span_sha256", } if not isinstance(citation, dict) or set(citation) != expected_fields: errors.append("citation_schema_mismatch") continue string_fields = expected_fields - {"char_start", "char_end"} if any( not isinstance(citation[field], str) or not citation[field] for field in string_fields ): errors.append("citation_string_field_invalid") continue citation_entry_id = citation["index_entry_id"] entry = entry_by_id.get(citation_entry_id) cited_entry_ids.append(citation_entry_id) if entry is None: errors.append("citation_entry_missing") continue if entry.index_entry_id not in selected_ids: errors.append("citation_not_selected") state = self.documents.get((entry.tenant_id, entry.document_id)) if ( state is None or state.deleted or state.access_blocked or entry.tenant_id != query["tenant_id"] or not set(query["subject_groups"]).intersection(entry.allowed_groups) or state.acl_snapshot_sha256 != entry.acl_snapshot_sha256 ): errors.append("citation_not_currently_authorized") revision = self.revisions.get(citation["canonical_revision_id"]) if revision is None: errors.append("citation_revision_missing") continue bindings = { "document_id": entry.document_id, "canonical_revision_id": entry.canonical_revision_id, "parse_revision_id": entry.parse_revision_id, "element_id": entry.element_id, "chunk_id": entry.chunk_id, "coordinate_space": entry.coordinate_space, "char_start": entry.char_start, "char_end": entry.char_end, } if any(citation[key] != expected for key, expected in bindings.items()): errors.append("citation_binding_mismatch") if citation["source_uri"] != revision.source_uri: errors.append("citation_source_uri_mismatch") if citation["source_version"] != revision.source_version: errors.append("citation_source_version_mismatch") if citation["raw_sha256"] != revision.raw_sha256: errors.append("citation_raw_hash_mismatch") start, end = citation["char_start"], citation["char_end"] if ( not isinstance(start, int) or isinstance(start, bool) or not isinstance(end, int) or isinstance(end, bool) or not 0 <= start < end <= len(revision.canonical_text) ): errors.append("citation_span_out_of_bounds") continue exact = revision.canonical_text[start:end] if exact != claim["text"]: errors.append("citation_does_not_support_claim") if digest_text(exact) != citation["span_sha256"]: errors.append("citation_span_hash_mismatch") if len(cited_entry_ids) != len(set(cited_entry_ids)): errors.append("duplicate_citation_entry") if sorted(cited_entry_ids) != sorted(selected_ids): errors.append("selected_citation_set_mismatch") elif cited_entry_ids != selected_ids: errors.append("selected_citation_order_mismatch") if failure is None: visible, filtered, expected_selected = self._select_entries(query, generation) expected_selected_ids = [item.index_entry_id for item in expected_selected] if selected_ids != expected_selected_ids: errors.append("selected_entries_not_recomputed") if audit_obj["filter_summary"] != { "visible": len(visible), "filtered": filtered, }: errors.append("filter_summary_not_recomputed") expected_trace = digest_object( { "query": query, "status": public_obj["status"], "claims": public_obj["claims"], } ) if public_obj["trace_id"] != expected_trace: errors.append("trace_binding_mismatch") return sorted(set(errors)) def reconcile(self) -> dict[str, int]: report = { "canonical_raw_hash_mismatch": 0, "canonical_text_hash_mismatch": 0, "canonical_revision_id_mismatch": 0, "element_integrity_mismatch": 0, "chunk_integrity_mismatch": 0, "index_entry_integrity_mismatch": 0, "published_generation_mismatch": 0, } for revision_id, revision in self.revisions.items(): if digest_text(revision.raw_content) != revision.raw_sha256: report["canonical_raw_hash_mismatch"] += 1 if digest_text(revision.canonical_text) != revision.canonical_text_sha256: report["canonical_text_hash_mismatch"] += 1 identity = { "tenant_id": revision.tenant_id, "document_id": revision.document_id, "source_uri": revision.source_uri, "source_version": revision.source_version, "raw_sha256": revision.raw_sha256, "canonical_text_sha256": revision.canonical_text_sha256, "normalizer_revision": revision.normalizer_revision, "acl_snapshot_sha256": revision.acl_snapshot_sha256, } if revision_id != "can_" + digest_object(identity): report["canonical_revision_id_mismatch"] += 1 for element in self.elements.get(revision_id, ()): exact = revision.canonical_text[element.char_start : element.char_end] expected_id = "el_" + digest_object( { "parse_revision_id": element.parse_revision_id, "kind": element.kind, "coordinate_space": element.coordinate_space, "char_start": element.char_start, "char_end": element.char_end, "text_sha256": element.text_sha256, } ) if ( exact != element.text or digest_text(element.text) != element.text_sha256 or expected_id != element.element_id ): report["element_integrity_mismatch"] += 1 for chunk in self.chunks.get(revision_id, ()): exact = revision.canonical_text[chunk.char_start : chunk.char_end] expected_id = "chk_" + digest_object( { "canonical_revision_id": chunk.canonical_revision_id, "parse_revision_id": chunk.parse_revision_id, "chunker_revision": chunk.chunker_revision, "element_id": chunk.element_id, "content_sha256": chunk.content_sha256, "acl_snapshot_sha256": chunk.acl_snapshot_sha256, } ) if ( exact != chunk.text or digest_text(chunk.text) != chunk.content_sha256 or digest_text(chunk.retrieval_text) != chunk.retrieval_sha256 or expected_id != chunk.chunk_id ): report["chunk_integrity_mismatch"] += 1 for generation in self.generations.values(): if self._generation_identity_errors(generation): report["published_generation_mismatch"] += 1 for entry in generation.entries: if self._entry_errors(entry): report["index_entry_integrity_mismatch"] += 1 if self.published_generation_id is not None: published = self.generations.get(self.published_generation_id) if published is None or published.status != "published": report["published_generation_mismatch"] += 1 return report def require_reconciled(self) -> dict[str, int]: report = self.reconcile() failures = {key: value for key, value in report.items() if value} if failures: raise ContractError("reconciliation failed：" + canonical_json(failures)) return report def initialize_fixture(fixture: dict[str, Any]) -> ProvenanceEngine: require_fixture_integrity(fixture) engine = ProvenanceEngine(fixture["contract"]) for event in fixture["source_events"]: engine.apply_event(event) engine.build_and_publish() engine.require_reconciled() return engine def evaluate_fixture( fixture: dict[str, Any], *, failure: str | None = None ) -> dict[str, Any]: engine = initialize_fixture(fixture) oracle_by_id = {item["query_id"]: item for item in fixture["oracle"]} cases: list[dict[str, Any]] = [] for query in fixture["queries"]: public, audit = engine.query(query, failure=failure) oracle = oracle_by_id[query["query_id"]] actual_documents = sorted( { citation["document_id"] for claim in public["claims"] for citation in claim["citations"] } ) errors = engine.validate_evidence(query, public, audit) if public["status"] != oracle["expected_status"]: errors.append("status_mismatch") if failure is None and actual_documents != oracle["expected_documents"]: errors.append("expected_documents_mismatch") if set(actual_documents).intersection(oracle["forbidden_documents"]): errors.append("forbidden_document_disclosed") cases.append( { "query_id": query["query_id"], "slice": oracle["slice"], "critical": oracle["critical"], "passed": not errors, "failure_codes": sorted(set(errors)), } ) generation = engine._published() passed = all(item["passed"] for item in cases) artifact = { "schema_version": ARTIFACT_SCHEMA_VERSION, "decision": "PASS" if passed else "BLOCK", "case_count": len(cases), "passed_case_count": sum(item["passed"] for item in cases), "fixture_sha256": fixture["fixture_sha256"], "fixture_model_sha256": fixture["fixture_model_sha256"], "pipeline_fingerprint": fixture["contract"].pipeline_fingerprint, "snapshot_state_sha256": generation.snapshot_state_sha256, "tombstone_state_sha256": generation.tombstone_state_sha256, "authorization_revision": generation.authorization_revision, "index_generation_id": generation.generation_id, "index_manifest_sha256": generation.manifest_sha256, "harness_revision": HARNESS_REVISION, "cases": cases, } artifact["artifact_sha256"] = digest_object(artifact) return artifact def validate_artifact(value: Any) -> list[str]: errors: list[str] = [] expected = { "schema_version", "decision", "case_count", "passed_case_count", "fixture_sha256", "fixture_model_sha256", "pipeline_fingerprint", "snapshot_state_sha256", "tombstone_state_sha256", "authorization_revision", "index_generation_id", "index_manifest_sha256", "harness_revision", "cases", "artifact_sha256", } try: artifact = _exact_fields(value, expected, "artifact") except ContractError as exc: return [str(exc)] if artifact["schema_version"] != ARTIFACT_SCHEMA_VERSION: errors.append("artifact_schema_mismatch") if artifact["harness_revision"] != HARNESS_REVISION: errors.append("harness_revision_mismatch") for field in ( "fixture_sha256", "fixture_model_sha256", "pipeline_fingerprint", "snapshot_state_sha256", "tombstone_state_sha256", "index_manifest_sha256", "artifact_sha256", ): if not isinstance(artifact[field], str) or HEX_64.fullmatch(artifact[field]) is None: errors.append(f"{field}_invalid") if ( not isinstance(artifact["case_count"], int) or isinstance(artifact["case_count"], bool) or artifact["case_count"] < 1 ): errors.append("case_count_invalid") if ( not isinstance(artifact["passed_case_count"], int) or isinstance(artifact["passed_case_count"], bool) or artifact["passed_case_count"] < 0 ): errors.append("passed_case_count_invalid") if not isinstance(artifact["authorization_revision"], str) or not artifact[ "authorization_revision" ]: errors.append("authorization_revision_invalid") if ( not isinstance(artifact["index_generation_id"], str) or not artifact["index_generation_id"].startswith("gen_") or HEX_64.fullmatch( artifact["index_generation_id"].removeprefix("gen_") ) is None ): errors.append("index_generation_id_invalid") cases = artifact["cases"] if not isinstance(cases, list) or not cases: errors.append("cases_invalid") else: expected_case_fields = { "query_id", "slice", "critical", "passed", "failure_codes", } seen_query_ids: set[str] = set() for index, item in enumerate(cases): if not isinstance(item, dict) or set(item) != expected_case_fields: errors.append(f"case_{index}_schema_mismatch") continue query_id = item["query_id"] if not isinstance(query_id, str) or not query_id or len(query_id) > 300: errors.append(f"case_{index}_query_id_invalid") elif query_id in seen_query_ids: errors.append("case_query_id_duplicate") else: seen_query_ids.add(query_id) if ( not isinstance(item["slice"], str) or not item["slice"] or len(item["slice"]) > 300 ): errors.append(f"case_{index}_slice_invalid") if not isinstance(item["critical"], bool): errors.append(f"case_{index}_critical_invalid") if not isinstance(item["passed"], bool): errors.append(f"case_{index}_passed_invalid") failure_codes = item["failure_codes"] if ( not isinstance(failure_codes, list) or any(not isinstance(code, str) or not code for code in failure_codes) or failure_codes != sorted(set(failure_codes)) ): errors.append(f"case_{index}_failure_codes_invalid") elif isinstance(item["passed"], bool) and item["passed"] != ( not failure_codes ): errors.append(f"case_{index}_pass_failure_mismatch") passed = sum(isinstance(item, dict) and item.get("passed") is True for item in cases) if artifact["case_count"] != len(cases): errors.append("case_count_mismatch") if artifact["passed_case_count"] != passed: errors.append("passed_case_count_mismatch") expected_decision = "PASS" if passed == len(cases) else "BLOCK" if artifact["decision"] != expected_decision: errors.append("decision_mismatch") unsigned = dict(artifact) supplied = unsigned.pop("artifact_sha256", None) if supplied != digest_object(unsigned): errors.append("artifact_hash_mismatch") return sorted(set(errors)) def _json_output(value: Any) -> None: print(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True, allow_nan=False)) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="离线 source-to-citation 证据链实验") parser.add_argument("--fixture", type=Path, required=True) subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("demo") ask = subparsers.add_parser("ask") ask.add_argument("--query-id", required=True) inspect = subparsers.add_parser("inspect") inspect.add_argument("--query-id", required=True) inspect.add_argument("--operator-view", action="store_true") evaluate = subparsers.add_parser("evaluate") evaluate.add_argument( "--failure", choices=["retrieval_unavailable"], default=None ) manifest = subparsers.add_parser("manifest") manifest.add_argument( "--operator-view", action="store_true", help="明确确认当前输出含 generation、授权与发布诊断", ) return parser def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) try: fixture = load_fixture(args.fixture) engine = initialize_fixture(fixture) if args.command == "demo": _json_output([engine.query(query)[0] for query in fixture["queries"]]) return 0 if args.command in {"ask", "inspect"}: query = next( (item for item in fixture["queries"] if item["query_id"] == args.query_id), None, ) if query is None: raise ContractError(f"未知 query_id：{args.query_id}") public, audit = engine.query(query) if args.command == "inspect": if not args.operator_view: raise ContractError("inspect 需要显式 --operator-view") _json_output({"public": public, "protected_audit": audit}) else: _json_output(public) return 0 if args.command == "evaluate": artifact = evaluate_fixture(fixture, failure=args.failure) _json_output(artifact) return 0 if artifact["decision"] == "PASS" else 1 if args.command == "manifest": if not args.operator_view: raise ContractError( "manifest 需要显式 --operator-view；该 flag 仅作教学确认，不替代真实鉴权" ) generation = engine._published() _json_output( { "generation_id": generation.generation_id, "snapshot_state_sha256": generation.snapshot_state_sha256, "tombstone_state_sha256": generation.tombstone_state_sha256, "authorization_revision": generation.authorization_revision, "pipeline_fingerprint": generation.pipeline_fingerprint, "entry_set_sha256": generation.entry_set_sha256, "manifest_sha256": generation.manifest_sha256, "entry_count": len(generation.entries), } ) return 0 raise ContractError("未知命令") except ContractError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())