From 54ae3942a808d92c1dce298de679782b99cdb153 Mon Sep 17 00:00:00 2001 From: Amidwestnoob Date: Wed, 19 Aug 2026 16:05:29 -0500 Subject: [PATCH] feat: add conversation materialization --- docs/.gitignore | 1 + docs/conversation-materialization.md | 172 ++++++++ openkb/conversation/__init__.py | 65 +++ openkb/conversation/authority.py | 53 +++ openkb/conversation/manifest.py | 471 ++++++++++++++++++++++ openkb/conversation/materialize.py | 532 +++++++++++++++++++++++++ openkb/conversation/models.py | 244 ++++++++++++ tests/test_conversation_authority.py | 176 ++++++++ tests/test_conversation_manifest.py | 389 ++++++++++++++++++ tests/test_conversation_materialize.py | 434 ++++++++++++++++++++ 10 files changed, 2537 insertions(+) create mode 100644 docs/conversation-materialization.md create mode 100644 openkb/conversation/__init__.py create mode 100644 openkb/conversation/authority.py create mode 100644 openkb/conversation/manifest.py create mode 100644 openkb/conversation/materialize.py create mode 100644 openkb/conversation/models.py create mode 100644 tests/test_conversation_authority.py create mode 100644 tests/test_conversation_manifest.py create mode 100644 tests/test_conversation_materialize.py diff --git a/docs/.gitignore b/docs/.gitignore index 0abcf25c6..759b40585 100644 --- a/docs/.gitignore +++ b/docs/.gitignore @@ -6,3 +6,4 @@ * !.gitignore !golden-principles.md +!conversation-materialization.md diff --git a/docs/conversation-materialization.md b/docs/conversation-materialization.md new file mode 100644 index 000000000..d91209d97 --- /dev/null +++ b/docs/conversation-materialization.md @@ -0,0 +1,172 @@ +# Conversation materialization + +## Purpose + +This package converts caller-supplied closed conversation windows into immutable source documents. + +The caller remains the source of truth. The source documents are a read-only projection. Each turn keeps its conversation identity, speaker, authority, time, text, and source anchor. + +The package has no source-specific rules. The caller selects the input window, cleaning hooks, limits, and authority values. + +## Data model + +`ConversationTurn` contains: + +- conversation ID +- turn ID +- speaker +- authority +- timestamp +- text +- source anchor + +Authority values are `first_party`, `assistant`, and `document`. These values come from public OpenKB claims rules. + +`ConversationWindow` contains a caller-defined window ID, source ID, creation time, closed state, and ordered turns. OpenKB does not calculate a delay. OpenKB does not use a local time zone. + +Each timestamp must use ISO format and include a UTC offset. OpenKB normalizes timestamps to UTC for sorting. When two normalized times are equal, OpenKB keeps the caller order. Compile selection normalizes each creation time to UTC before it applies the source, window, segment, and path tie breaks. + +## Cleaning and limits + +`clean_text()` normalizes line endings, removes NUL characters, and runs pure caller hooks in order. + +The default materializer fails when one turn exceeds `max_turn_chars`. It also fails when one complete rendered source document exceeds `max_document_chars`. The document limit includes headers, evidence fields, turn fields, text, and line breaks. The materializer never truncates by default. + +A caller can use `truncate_with_digest()` as an explicit choice. The result keeps the beginning and end. It records the original character count and the SHA-256 digest of the omitted middle. + +The splitter stops only at turn boundaries. It preserves UTC order and caller order for equal UTC times. After each split, it verifies that the flattened turn identities exactly match the ordered input identities. A coverage error stops materialization before a file or manifest mutation. + +## Closed-window identity + +The manifest stores one fingerprint for the canonical closed-window bytes. It also stores one fingerprint for the active document set. + +A normal replay must match both fingerprints. A change to a turn, creation time, rendered bytes, segment count, segment path, or document hash fails. An empty closed window also gets a fingerprint and a window record. A later non-empty replay of that identity fails. + +Only `recover_uncompiled()` can replace the active document set. The canonical closed-window fingerprint must still match. + +Path tokens contain a readable prefix and the full SHA-256 digest of the opaque ID. Different IDs cannot collide only because their readable forms are the same. + +## Immutable documents and manifest rules + +Each materialized document contains: + +- window ID +- source ID +- creation time +- segment index and count +- evidence rules +- turn ID +- speaker +- authority +- timestamp +- source anchor +- text + +The manifest stores relative paths and lowercase hexadecimal SHA-256 values. Each document record also stores the full `(conversation_id, turn_id)` identities. + +Manifest loading rejects: + +- malformed timestamps +- malformed hashes +- corrupt UTF-8 manifest text +- duplicate document paths +- duplicate window records +- conflicting window document sets +- unknown completed hashes +- receipts for incomplete hashes + +A replay with the same bytes is a no-op. A replay with different bytes fails. `verify_manifest()` checks every active file. + +`select_compile_inputs()` sorts by normalized UTC creation time, source ID, window ID, segment index, and relative path. It skips completed hashes and returns the caller limit. The optional `filename` value selects a non-default manifest. The function does not call a compiler. + +`check_compile_budget()` validates positive values and requires: + +`batch_limit * document_timeout < scheduler_guard` + +## Compile receipts + +`mark_compiled()` can store optional receipt metadata for each content hash. The allowed metadata is: + +- compile time +- duration in seconds +- result SHA-256 + +The manifest JSON is deterministic. Receipt values do not contain source-specific fields. + +## Recovery + +`recover_uncompiled()` uses a smaller document limit. It selects turns with the full `(conversation_id, turn_id)` identity. Equal turn IDs from different conversations cannot cross-select. + +Recovery does not replace a compiled document. It writes the new recovery documents first. It then writes the new manifest. After both writes are durable, it removes the replaced uncompiled files from the active source tree. The manifest and compile selector cannot return the old files. + +If a pending turn cannot fit the recovery limit, recovery stops before mutation. The pending source file and manifest remain unchanged. + +Residual risk: A process stop after a new recovery file write and before the manifest write can leave an untracked recovery file. The file is not selectable because it is not in the manifest. Automatic cleanup of this crash orphan is outside this change. + +Recovery does not rewrite wiki output. + +## Shared locking + +Manifest read-modify-write functions use the shared OpenKB mutation lock. They use unlocked helpers only while the public function holds that lock. + +`apply_verified_claims()` uses the same lock rule. The function reads, merges, and writes one page while it holds the lock. Concurrent claim updates cannot overwrite each other. + +## Verified claims + +`apply_verified_claims()` accepts caller-verified claim objects. It calls `openkb.claims.merge_claims()`. It uses the shared frontmatter helpers and atomic writer. + +The function keeps the page body and claim history. Repeated input is idempotent. Public claim rules apply: + +- A first-party claim can supersede an older assistant proposal when the claim rules permit it. +- An assistant claim cannot remain validated. +- An assistant claim cannot supersede a first-party claim. + +## Deployment examples + +The following values are aggregate deployment examples. They are not code defaults: + +- One deployment processed 280 conversations into 537 documents. +- Initial materialization used a 240,000-character document limit. +- Recovery used a 40,000-character document limit. +- A later live materializer used a 120,000-character document limit and a 40,000-character turn limit. +- One document exceeded 600 seconds and completed in 885 seconds. +- The validated operating point used a 1,200-second document timeout, a two-document batch, and a 3,600-second scheduler guard. + +These values show why callers must set limits for their own workload. + +## Relation to public OpenKB work + +This change is stacked on PR #223. It uses the claims API from that change. + +This change does not depend on PR #179. It does not copy the ingest-bundle pipeline. A future PR #179 adapter can call this materializer. + +The package does not add live input, source-specific filters, source data, scheduler wiring, or model calls. + +## Synthetic example + +```python +window = ConversationWindow( + window_id="example-window", + source_id="example-source", + created_at="2026-01-01T00:00:00Z", + closed=True, + turns=( + ConversationTurn( + conversation_id="example-conversation", + turn_id="turn-1", + speaker="first-party", + authority="first_party", + timestamp="2026-01-01T00:00:00Z", + text="A synthetic value.", + source_anchor="example:turn-1", + ), + ), +) + +materialize_window( + window, + output_dir, + max_turn_chars=40000, + max_document_chars=120000, +) +``` diff --git a/openkb/conversation/__init__.py b/openkb/conversation/__init__.py new file mode 100644 index 000000000..a22abc93d --- /dev/null +++ b/openkb/conversation/__init__.py @@ -0,0 +1,65 @@ +"""Transport-neutral closed conversation materialization.""" + +from openkb.conversation.authority import apply_verified_claims +from openkb.conversation.manifest import ( + Manifest, + ManifestConflictError, + ManifestError, + check_compile_budget, + load_manifest, + mark_compiled, + register_documents, + save_manifest, + select_compile_inputs, + verify_manifest, +) +from openkb.conversation.materialize import ( + DuplicateTurnError, + OversizeTurnError, + clean_text, + deduplicate_turns, + materialize_window, + recover_uncompiled, + render_document, + split_turns, + truncate_with_digest, +) +from openkb.conversation.models import ( + CompileReceipt, + CompileReceiptMetadata, + ConversationTurn, + ConversationWindow, + DocumentRef, + TruncationResult, + WindowRecord, +) + +__all__ = [ + "CompileReceipt", + "CompileReceiptMetadata", + "ConversationTurn", + "ConversationWindow", + "DocumentRef", + "TruncationResult", + "WindowRecord", + "DuplicateTurnError", + "OversizeTurnError", + "Manifest", + "ManifestError", + "ManifestConflictError", + "apply_verified_claims", + "check_compile_budget", + "clean_text", + "deduplicate_turns", + "load_manifest", + "mark_compiled", + "materialize_window", + "recover_uncompiled", + "register_documents", + "render_document", + "save_manifest", + "select_compile_inputs", + "split_turns", + "truncate_with_digest", + "verify_manifest", +] diff --git a/openkb/conversation/authority.py b/openkb/conversation/authority.py new file mode 100644 index 000000000..8f844ebda --- /dev/null +++ b/openkb/conversation/authority.py @@ -0,0 +1,53 @@ +"""Apply caller-verified claims through the public claims API.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Mapping, Sequence + +from openkb import frontmatter +from openkb.claims import merge_claims +from openkb.locks import atomic_write_text, kb_ingest_lock + + +def _default_openkb_dir(page_path: Path) -> Path: + for parent in page_path.parents: + candidate = parent / ".openkb" + if candidate.is_dir(): + return candidate + return page_path.parent / ".openkb" + + +def _apply_verified_claims_unlocked( + path: Path, + claims: Sequence[Mapping[str, object]], +) -> list[dict]: + text = path.read_text(encoding="utf-8") + parts = frontmatter.split(text) + if parts is None: + raise ValueError("page must contain valid frontmatter") + fm_block, body = parts + existing = frontmatter.parse(text).get("claims") + merged = merge_claims(existing if isinstance(existing, list) else [], list(claims)) + if merged: + fm_block = frontmatter.set_json_line(fm_block, "claims", merged) + else: + fm_block = frontmatter.drop_line(fm_block, "claims") + atomic_write_text(path, fm_block + body) + return merged + + +def apply_verified_claims( + page_path: Path | str, + claims: Sequence[Mapping[str, object]], + *, + openkb_dir: Path | str | None = None, +) -> list[dict]: + """Merge verified claims under the shared mutation lock.""" + path = Path(page_path).resolve() + lock_dir = Path(openkb_dir) if openkb_dir is not None else _default_openkb_dir(path) + with kb_ingest_lock(lock_dir): + return _apply_verified_claims_unlocked(path, claims) + + +apply_claims = apply_verified_claims diff --git a/openkb/conversation/manifest.py b/openkb/conversation/manifest.py new file mode 100644 index 000000000..0c8f2807c --- /dev/null +++ b/openkb/conversation/manifest.py @@ -0,0 +1,471 @@ +"""Replay-safe manifests for conversation source documents.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +from openkb.conversation.models import ( + CompileReceipt, + CompileReceiptMetadata, + DocumentRef, + WindowRecord, + utc_timestamp, + validate_sha256, +) +from openkb.locks import atomic_write_text, kb_ingest_lock, kb_read_lock + +MANIFEST_FILENAME = "manifest.json" + + +class ManifestError(ValueError): + """Base error for invalid or conflicting manifest state.""" + + +class ManifestConflictError(ManifestError): + """Raised when immutable manifest state conflicts with new input.""" + + +def _document_set_sha256(documents: Iterable[DocumentRef]) -> str: + """Return a deterministic hash for one active document set.""" + ordered = sorted(documents, key=lambda ref: ref.relative_path) + payload = [ + { + "relative_path": ref.relative_path, + "content_sha256": ref.content_sha256, + "turn_identities": [list(identity) for identity in ref.turn_identities], + } + for ref in ordered + ] + encoded = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +@dataclass(frozen=True, slots=True) +class Manifest: + """The manifest state for one materialization root.""" + + documents: tuple[DocumentRef, ...] = () + windows: tuple[WindowRecord, ...] = () + completed_hashes: tuple[str, ...] = () + compile_receipts: tuple[CompileReceipt, ...] = () + version: int = 1 + + def __post_init__(self) -> None: + documents = tuple(self.documents) + windows = tuple(self.windows) + completed = tuple(self.completed_hashes) + receipts = tuple(self.compile_receipts) + if self.version != 1: + raise ManifestError("manifest version is not supported") + if len({ref.relative_path for ref in documents}) != len(documents): + raise ManifestError("manifest contains duplicate document paths") + if len({record.identity for record in windows}) != len(windows): + raise ManifestError("manifest contains duplicate window records") + known_hashes = {ref.content_sha256 for ref in documents} + if len(set(completed)) != len(completed): + raise ManifestError("manifest contains duplicate completed hashes") + for digest in completed: + validate_sha256(digest, "completed hash") + if not set(completed).issubset(known_hashes): + raise ManifestError("manifest contains an unknown completed hash") + if len({receipt.content_sha256 for receipt in receipts}) != len(receipts): + raise ManifestError("manifest contains duplicate compile receipts") + if not {receipt.content_sha256 for receipt in receipts}.issubset(set(completed)): + raise ManifestError("manifest contains a receipt for an incomplete hash") + + documents_by_window: dict[tuple[str, str], list[DocumentRef]] = {} + for ref in documents: + documents_by_window.setdefault((ref.source_id, ref.window_id), []).append(ref) + windows_by_identity = {record.identity: record for record in windows} + if set(documents_by_window) - set(windows_by_identity): + raise ManifestError("manifest document has no window record") + for identity, record in windows_by_identity.items(): + refs = sorted(documents_by_window.get(identity, []), key=lambda ref: ref.relative_path) + if any(ref.created_at != record.created_at for ref in refs): + raise ManifestError("manifest window creation times conflict") + if tuple(ref.relative_path for ref in refs) != record.document_paths: + raise ManifestError("manifest window document paths conflict") + if tuple(ref.content_sha256 for ref in refs) != record.document_hashes: + raise ManifestError("manifest window document hashes conflict") + if _document_set_sha256(refs) != record.materialization_sha256: + raise ManifestError("manifest window materialization hash conflicts") + + object.__setattr__(self, "documents", documents) + object.__setattr__(self, "windows", windows) + object.__setattr__(self, "completed_hashes", completed) + object.__setattr__(self, "compile_receipts", receipts) + + +def manifest_path(root: Path | str, filename: str = MANIFEST_FILENAME) -> Path: + """Return the manifest path below *root*.""" + path = Path(root) / filename + if Path(filename).is_absolute() or ".." in Path(filename).parts: + raise ValueError("manifest filename must stay below the materialization root") + return path + + +def _lock_dir(root: Path | str) -> Path: + return Path(root) / ".openkb" + + +def _required_str(item: dict, key: str) -> str: + value = item.get(key) + if not isinstance(value, str): + raise ManifestError(f"manifest field must be a string: {key}") + return value + + +def _required_int(item: dict, key: str) -> int: + value = item.get(key) + if isinstance(value, bool) or not isinstance(value, int): + raise ManifestError(f"manifest field must be an integer: {key}") + return value + + +def _parse_document(item: object) -> DocumentRef: + if not isinstance(item, dict): + raise ManifestError("manifest document entry is not an object") + turn_ids = item.get("turn_ids") + identities = item.get("turn_identities") + if not isinstance(turn_ids, list) or any(not isinstance(value, str) for value in turn_ids): + raise ManifestError("manifest turn_ids is invalid") + if not isinstance(identities, list): + raise ManifestError("manifest turn_identities is invalid") + parsed_identities: list[tuple[str, str]] = [] + for identity in identities: + if ( + not isinstance(identity, list) + or len(identity) != 2 + or any(not isinstance(value, str) for value in identity) + ): + raise ManifestError("manifest turn identity is invalid") + parsed_identities.append((identity[0], identity[1])) + try: + return DocumentRef( + relative_path=_required_str(item, "relative_path"), + content_sha256=_required_str(item, "content_sha256"), + window_id=_required_str(item, "window_id"), + source_id=_required_str(item, "source_id"), + created_at=_required_str(item, "created_at"), + segment_index=_required_int(item, "segment_index"), + segment_count=_required_int(item, "segment_count"), + turn_ids=tuple(turn_ids), + turn_identities=tuple(parsed_identities), + ) + except (TypeError, ValueError) as exc: + raise ManifestError("manifest document entry is invalid") from exc + + +def _parse_window(item: object) -> WindowRecord: + if not isinstance(item, dict): + raise ManifestError("manifest window entry is not an object") + paths = item.get("document_paths") + hashes = item.get("document_hashes") + if not isinstance(paths, list) or any(not isinstance(value, str) for value in paths): + raise ManifestError("manifest window document_paths is invalid") + if not isinstance(hashes, list) or any(not isinstance(value, str) for value in hashes): + raise ManifestError("manifest window document_hashes is invalid") + try: + return WindowRecord( + source_id=_required_str(item, "source_id"), + window_id=_required_str(item, "window_id"), + created_at=_required_str(item, "created_at"), + window_sha256=_required_str(item, "window_sha256"), + materialization_sha256=_required_str(item, "materialization_sha256"), + document_paths=tuple(paths), + document_hashes=tuple(hashes), + ) + except (TypeError, ValueError) as exc: + raise ManifestError("manifest window entry is invalid") from exc + + +def _parse_receipts(raw: object) -> tuple[CompileReceipt, ...]: + if not isinstance(raw, dict): + raise ManifestError("manifest compile_receipts is invalid") + receipts: list[CompileReceipt] = [] + allowed = {"compiled_at", "duration_seconds", "result_sha256"} + for content_sha256, value in raw.items(): + if not isinstance(content_sha256, str) or not isinstance(value, dict): + raise ManifestError("manifest compile receipt is invalid") + if set(value) - allowed or "compiled_at" not in value: + raise ManifestError("manifest compile receipt fields are invalid") + compiled_at = value.get("compiled_at") + duration = value.get("duration_seconds") + result_sha256 = value.get("result_sha256") + if not isinstance(compiled_at, str): + raise ManifestError("manifest compiled_at is invalid") + if duration is not None and ( + isinstance(duration, bool) or not isinstance(duration, (int, float)) + ): + raise ManifestError("manifest duration_seconds is invalid") + if result_sha256 is not None and not isinstance(result_sha256, str): + raise ManifestError("manifest result_sha256 is invalid") + try: + metadata = CompileReceiptMetadata( + compiled_at=compiled_at, + duration_seconds=duration, + result_sha256=result_sha256, + ) + receipts.append(CompileReceipt(content_sha256, metadata)) + except (TypeError, ValueError) as exc: + raise ManifestError("manifest compile receipt is invalid") from exc + return tuple(receipts) + + +def _load_manifest_unlocked(root: Path | str, filename: str = MANIFEST_FILENAME) -> Manifest: + path = manifest_path(root, filename) + if not path.exists(): + return Manifest() + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ManifestError(f"cannot read manifest: {path.name}") from exc + if not isinstance(raw, dict) or raw.get("version") != 1: + raise ManifestError("manifest version is not supported") + documents_raw = raw.get("documents") + windows_raw = raw.get("windows") + completed_raw = raw.get("completed_hashes") + if not isinstance(documents_raw, list): + raise ManifestError("manifest documents is invalid") + if not isinstance(windows_raw, list): + raise ManifestError("manifest windows is invalid") + if not isinstance(completed_raw, list) or any( + not isinstance(value, str) for value in completed_raw + ): + raise ManifestError("manifest completed_hashes is invalid") + try: + return Manifest( + documents=tuple(_parse_document(item) for item in documents_raw), + windows=tuple(_parse_window(item) for item in windows_raw), + completed_hashes=tuple(completed_raw), + compile_receipts=_parse_receipts(raw.get("compile_receipts", {})), + ) + except ManifestError: + raise + except (TypeError, ValueError) as exc: + raise ManifestError("manifest is invalid") from exc + + +def load_manifest(root: Path | str, filename: str = MANIFEST_FILENAME) -> Manifest: + """Load one manifest while holding the shared read lock.""" + with kb_read_lock(_lock_dir(root)): + return _load_manifest_unlocked(root, filename) + + +def _document_dict(ref: DocumentRef) -> dict[str, object]: + return { + "relative_path": ref.relative_path, + "content_sha256": ref.content_sha256, + "window_id": ref.window_id, + "source_id": ref.source_id, + "created_at": ref.created_at, + "segment_index": ref.segment_index, + "segment_count": ref.segment_count, + "turn_ids": list(ref.turn_ids), + "turn_identities": [list(identity) for identity in ref.turn_identities], + } + + +def _window_dict(record: WindowRecord) -> dict[str, object]: + return { + "source_id": record.source_id, + "window_id": record.window_id, + "created_at": record.created_at, + "window_sha256": record.window_sha256, + "materialization_sha256": record.materialization_sha256, + "document_paths": list(record.document_paths), + "document_hashes": list(record.document_hashes), + } + + +def _receipt_dict(metadata: CompileReceiptMetadata) -> dict[str, object]: + result: dict[str, object] = {"compiled_at": metadata.compiled_at} + if metadata.duration_seconds is not None: + result["duration_seconds"] = metadata.duration_seconds + if metadata.result_sha256 is not None: + result["result_sha256"] = metadata.result_sha256 + return result + + +def _manifest_text(manifest: Manifest) -> str: + documents = sorted(manifest.documents, key=lambda item: item.relative_path) + windows = sorted(manifest.windows, key=lambda item: item.identity) + receipts = { + receipt.content_sha256: _receipt_dict(receipt.metadata) + for receipt in sorted(manifest.compile_receipts, key=lambda item: item.content_sha256) + } + payload = { + "version": manifest.version, + "documents": [_document_dict(ref) for ref in documents], + "windows": [_window_dict(record) for record in windows], + "completed_hashes": sorted(manifest.completed_hashes), + "compile_receipts": receipts, + } + return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n" + + +def _save_manifest_unlocked( + root: Path | str, + manifest: Manifest, + filename: str = MANIFEST_FILENAME, +) -> None: + atomic_write_text(manifest_path(root, filename), _manifest_text(manifest)) + + +def save_manifest(root: Path | str, manifest: Manifest, filename: str = MANIFEST_FILENAME) -> None: + """Write one manifest while holding the shared mutation lock.""" + with kb_ingest_lock(_lock_dir(root)): + _save_manifest_unlocked(root, manifest, filename) + + +def _merge_documents( + existing: Iterable[DocumentRef], new: Iterable[DocumentRef] +) -> tuple[DocumentRef, ...]: + by_path = {ref.relative_path: ref for ref in existing} + for ref in new: + old = by_path.get(ref.relative_path) + if old is not None and old != ref: + raise ManifestConflictError( + f"document path has conflicting metadata: {ref.relative_path}" + ) + if old is None: + by_path[ref.relative_path] = ref + return tuple(sorted(by_path.values(), key=lambda item: item.relative_path)) + + +def _merge_windows( + existing: Iterable[WindowRecord], new: Iterable[WindowRecord] +) -> tuple[WindowRecord, ...]: + by_identity = {record.identity: record for record in existing} + for record in new: + old = by_identity.get(record.identity) + if old is not None and old != record: + raise ManifestConflictError("window identity has conflicting immutable state") + if old is None: + by_identity[record.identity] = record + return tuple(sorted(by_identity.values(), key=lambda item: item.identity)) + + +def register_documents( + root: Path | str, + documents: Iterable[DocumentRef], + *, + window_records: Iterable[WindowRecord] = (), + filename: str = MANIFEST_FILENAME, +) -> Manifest: + """Register documents and windows under one mutation lock.""" + with kb_ingest_lock(_lock_dir(root)): + current = _load_manifest_unlocked(root, filename) + merged = Manifest( + documents=_merge_documents(current.documents, documents), + windows=_merge_windows(current.windows, window_records), + completed_hashes=current.completed_hashes, + compile_receipts=current.compile_receipts, + ) + _save_manifest_unlocked(root, merged, filename) + return merged + + +def verify_manifest(root: Path | str, filename: str = MANIFEST_FILENAME) -> bool: + """Verify every file hash stored in the manifest.""" + with kb_read_lock(_lock_dir(root)): + manifest = _load_manifest_unlocked(root, filename) + root_path = Path(root).resolve() + for ref in manifest.documents: + path = (root_path / ref.relative_path).resolve() + if not path.is_relative_to(root_path) or not path.is_file(): + raise ManifestError(f"manifest file is missing: {ref.relative_path}") + digest = hashlib.sha256(path.read_bytes()).hexdigest() + if digest != ref.content_sha256: + raise ManifestError(f"manifest hash mismatch: {ref.relative_path}") + return True + + +def mark_compiled( + root: Path | str, + documents: Iterable[DocumentRef | str], + *, + receipt: CompileReceiptMetadata | None = None, + filename: str = MANIFEST_FILENAME, +) -> Manifest: + """Mark hashes as compiled and store optional public receipt metadata.""" + with kb_ingest_lock(_lock_dir(root)): + current = _load_manifest_unlocked(root, filename) + known = {ref.content_sha256 for ref in current.documents} + requested: set[str] = set() + for item in documents: + digest = item.content_sha256 if isinstance(item, DocumentRef) else item + validate_sha256(digest, "completed hash") + requested.add(digest) + if requested - known: + raise ManifestError("cannot mark an unknown document hash as compiled") + receipts = {item.content_sha256: item for item in current.compile_receipts} + if receipt is not None: + for digest in requested: + incoming = CompileReceipt(digest, receipt) + old = receipts.get(digest) + if old is not None and old != incoming: + raise ManifestConflictError("compile receipt conflicts with stored metadata") + receipts[digest] = incoming + merged = Manifest( + documents=current.documents, + windows=current.windows, + completed_hashes=tuple(sorted(set(current.completed_hashes) | requested)), + compile_receipts=tuple(sorted(receipts.values(), key=lambda item: item.content_sha256)), + ) + _save_manifest_unlocked(root, merged, filename) + return merged + + +def select_compile_inputs( + root_or_manifest: Path | str | Manifest, + limit: int, + *, + completed_hashes: Iterable[str] = (), + filename: str = MANIFEST_FILENAME, +) -> tuple[DocumentRef, ...]: + """Return the oldest pending document references up to *limit*.""" + if limit <= 0: + raise ValueError("limit must be positive") + manifest = ( + load_manifest(root_or_manifest, filename) + if not isinstance(root_or_manifest, Manifest) + else root_or_manifest + ) + completed = set(manifest.completed_hashes) + for digest in completed_hashes: + validate_sha256(digest, "completed hash") + completed.add(digest) + ordered = sorted( + (ref for ref in manifest.documents if ref.content_sha256 not in completed), + key=lambda ref: ( + utc_timestamp(ref.created_at, "created_at"), + ref.source_id, + ref.window_id, + ref.segment_index, + ref.relative_path, + ), + ) + return tuple(ordered[:limit]) + + +def check_compile_budget( + batch_limit: int, + document_timeout: float, + scheduler_guard: float, +) -> bool: + """Validate a bounded compile budget without starting any work.""" + if batch_limit <= 0 or document_timeout <= 0 or scheduler_guard <= 0: + raise ValueError("compile budget values must be positive") + if batch_limit * document_timeout >= scheduler_guard: + raise ValueError("batch_limit * document_timeout must be less than scheduler_guard") + return True diff --git a/openkb/conversation/materialize.py b/openkb/conversation/materialize.py new file mode 100644 index 000000000..878cc6dcf --- /dev/null +++ b/openkb/conversation/materialize.py @@ -0,0 +1,532 @@ +"""Materialize closed conversation windows as immutable source documents.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import replace +from pathlib import Path +from typing import Callable, Iterable, Sequence + +from openkb.conversation.manifest import ( + Manifest, + ManifestConflictError, + _document_set_sha256, + _load_manifest_unlocked, + _save_manifest_unlocked, +) +from openkb.conversation.models import ( + ConversationTurn, + ConversationWindow, + DocumentRef, + TruncationResult, + WindowRecord, + utc_timestamp, +) +from openkb.locks import atomic_write_bytes, kb_ingest_lock + +TextHook = Callable[[str], str] + + +class OversizeTurnError(ValueError): + """Raised when one turn cannot fit the requested limits.""" + + +class DuplicateTurnError(ValueError): + """Raised when one turn identity has conflicting fields.""" + + +def clean_text(text: str, hooks: Iterable[TextHook] = ()) -> str: + """Normalize line endings, remove NUL characters, and run pure hooks.""" + if not isinstance(text, str): + raise TypeError("text must be a string") + cleaned = text.replace("\r\n", "\n").replace("\r", "\n").replace("\x00", "") + for hook in hooks: + cleaned = hook(cleaned) + if not isinstance(cleaned, str): + raise TypeError("cleaning hooks must return strings") + return cleaned + + +def truncate_with_digest(text: str, max_chars: int) -> TruncationResult: + """Keep the text edges and record the digest of the omitted middle.""" + if max_chars < 0: + raise ValueError("max_chars must not be negative") + original_count = len(text) + if original_count <= max_chars: + omitted = "" + return TruncationResult( + text=text, + original_char_count=original_count, + omitted_middle_sha256=hashlib.sha256(omitted.encode("utf-8")).hexdigest(), + omitted_middle_char_count=0, + ) + head_count = max_chars // 2 + tail_count = max_chars - head_count + tail_start = original_count - tail_count + omitted = text[head_count:tail_start] + return TruncationResult( + text=text[:head_count] + text[tail_start:], + original_char_count=original_count, + omitted_middle_sha256=hashlib.sha256(omitted.encode("utf-8")).hexdigest(), + omitted_middle_char_count=len(omitted), + ) + + +def deduplicate_turns(turns: Iterable[ConversationTurn]) -> tuple[ConversationTurn, ...]: + """Deduplicate turns and preserve caller order for equal UTC times.""" + unique: dict[tuple[str, str], tuple[int, ConversationTurn]] = {} + for position, turn in enumerate(turns): + old = unique.get(turn.identity) + if old is not None: + if old[1].canonical_fields != turn.canonical_fields: + raise DuplicateTurnError( + f"turn identity has conflicting fields: {turn.conversation_id}/{turn.turn_id}" + ) + continue + unique[turn.identity] = (position, turn) + return tuple( + turn + for _, turn in sorted( + unique.values(), + key=lambda item: (utc_timestamp(item[1].timestamp), item[0]), + ) + ) + + +def _field(name: str, value: str | int) -> str: + return f"{name}: {json.dumps(value, ensure_ascii=False)}" + + +def render_document( + window: ConversationWindow, + turns: Sequence[ConversationTurn], + *, + segment_index: int, + segment_count: int, +) -> bytes: + """Render one deterministic source document.""" + lines = [ + "# Conversation source document", + "", + _field("window_id", window.window_id), + _field("source_id", window.source_id), + _field("created_at", window.created_at), + _field("segment_index", segment_index), + _field("segment_count", segment_count), + "", + "## Evidence rules", + "", + "- The caller closed the window before materialization.", + "- Each turn remains one evidence unit.", + "- The caller supplied the authority and source anchor values.", + "- The source document is a read-only projection of caller evidence.", + ] + for turn in turns: + lines.extend( + [ + "", + f"## Turn {turn.turn_id}", + "", + _field("turn_id", turn.turn_id), + _field("speaker", turn.speaker), + _field("authority", turn.authority), + _field("timestamp", turn.timestamp), + _field("source_anchor", turn.source_anchor), + "text:", + turn.text, + ] + ) + return ("\n".join(lines) + "\n").encode("utf-8") + + +def _rendered_char_count( + window: ConversationWindow, + turns: Sequence[ConversationTurn], + *, + segment_index: int, + segment_count: int, +) -> int: + return len( + render_document( + window, + turns, + segment_index=segment_index, + segment_count=segment_count, + ).decode("utf-8") + ) + + +def _require_complete_split( + turns: Sequence[ConversationTurn], + segments: Sequence[Sequence[ConversationTurn]], +) -> None: + expected = tuple(turn.identity for turn in turns) + actual = tuple(turn.identity for segment in segments for turn in segment) + if actual != expected: + raise RuntimeError("split turn coverage does not match the ordered input") + + +def split_turns( + turns: Sequence[ConversationTurn], + *, + window: ConversationWindow, + max_turn_chars: int, + max_document_chars: int, +) -> tuple[tuple[ConversationTurn, ...], ...]: + """Split at turn boundaries and limit final rendered document characters.""" + if max_turn_chars <= 0 or max_document_chars <= 0: + raise ValueError("materialization limits must be positive") + if not turns: + empty_segments: tuple[tuple[ConversationTurn, ...], ...] = () + _require_complete_split(turns, empty_segments) + return empty_segments + for turn in turns: + if len(turn.text) > max_turn_chars: + raise OversizeTurnError(f"turn exceeds max_turn_chars: {turn.turn_id}") + + segments: list[tuple[ConversationTurn, ...]] = [] + current: list[ConversationTurn] = [] + count_hint = len(turns) + for turn in turns: + candidate = [*current, turn] + segment_index = len(segments) + 1 + if ( + _rendered_char_count( + window, + candidate, + segment_index=segment_index, + segment_count=count_hint, + ) + <= max_document_chars + ): + current = candidate + continue + if current: + segments.append(tuple(current)) + current = [turn] + segment_index = len(segments) + 1 + else: + current = [turn] + if ( + _rendered_char_count( + window, + current, + segment_index=segment_index, + segment_count=count_hint, + ) + > max_document_chars + ): + raise OversizeTurnError(f"turn cannot fit the rendered document limit: {turn.turn_id}") + if current: + segments.append(tuple(current)) + result = tuple(segments) + _require_complete_split(turns, result) + return result + + +def _path_token(value: str) -> str: + prefix = re.sub(r"[^A-Za-z0-9._-]", "_", value).strip(".")[:48] or "value" + digest = hashlib.sha256(value.encode("utf-8")).hexdigest() + return f"{prefix}-{digest}" + + +def _window_sha256(window: ConversationWindow, turns: Sequence[ConversationTurn]) -> str: + payload = { + "source_id": window.source_id, + "window_id": window.window_id, + "created_at": window.created_at, + "closed": window.closed, + "turns": [ + { + "conversation_id": turn.conversation_id, + "turn_id": turn.turn_id, + "speaker": turn.speaker, + "authority": turn.authority, + "timestamp": turn.timestamp, + "text": turn.text, + "source_anchor": turn.source_anchor, + } + for turn in turns + ], + } + encoded = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _window_record( + window: ConversationWindow, + window_sha256: str, + documents: Sequence[DocumentRef], +) -> WindowRecord: + ordered = sorted(documents, key=lambda ref: ref.relative_path) + return WindowRecord( + source_id=window.source_id, + window_id=window.window_id, + created_at=window.created_at, + window_sha256=window_sha256, + materialization_sha256=_document_set_sha256(ordered), + document_paths=tuple(ref.relative_path for ref in ordered), + document_hashes=tuple(ref.content_sha256 for ref in ordered), + ) + + +def _refs_and_bytes( + window: ConversationWindow, + segments: Sequence[Sequence[ConversationTurn]], + *, + max_document_chars: int, + directory: str = "conversations", +) -> tuple[tuple[DocumentRef, bytes], ...]: + segment_count = len(segments) + rendered: list[tuple[Sequence[ConversationTurn], bytes, str]] = [] + for index, turns in enumerate(segments, start=1): + content = render_document( + window, + turns, + segment_index=index, + segment_count=segment_count, + ) + if len(content.decode("utf-8")) > max_document_chars: + raise OversizeTurnError("rendered document exceeds max_document_chars") + rendered.append((turns, content, hashlib.sha256(content).hexdigest())) + + base = Path(directory) + if directory == "recovery" and rendered: + generation_payload = json.dumps( + [digest for _, _, digest in rendered], + separators=(",", ":"), + ).encode("utf-8") + base /= hashlib.sha256(generation_payload).hexdigest() + + result: list[tuple[DocumentRef, bytes]] = [] + for index, (turns, content, digest) in enumerate(rendered, start=1): + relative = ( + base + / _path_token(window.source_id) + / _path_token(window.window_id) + / f"segment-{index:04d}-of-{segment_count:04d}.md" + ).as_posix() + result.append( + ( + DocumentRef( + relative_path=relative, + content_sha256=digest, + window_id=window.window_id, + source_id=window.source_id, + created_at=window.created_at, + segment_index=index, + segment_count=segment_count, + turn_ids=tuple(turn.turn_id for turn in turns), + turn_identities=tuple(turn.identity for turn in turns), + ), + content, + ) + ) + return tuple(result) + + +def _write_documents_unlocked( + root: Path, + documents: Sequence[tuple[DocumentRef, bytes]], + window_record: WindowRecord, + *, + manifest_filename: str, + replace_paths: Iterable[str] = (), + allow_window_replacement: bool = False, +) -> Manifest: + current = _load_manifest_unlocked(root, manifest_filename) + current_windows = {record.identity: record for record in current.windows} + old_window = current_windows.get(window_record.identity) + if old_window is None and allow_window_replacement: + raise ManifestConflictError("recovery cannot find the closed window record") + if old_window is not None: + if allow_window_replacement: + if old_window.window_sha256 != window_record.window_sha256: + raise ManifestConflictError( + "closed window bytes conflict with the stored fingerprint" + ) + elif old_window != window_record: + raise ManifestConflictError("closed window replay changes bytes or the segment set") + + removed = set(replace_paths) + refs_by_path = { + ref.relative_path: ref for ref in current.documents if ref.relative_path not in removed + } + removed_refs = [ref for ref in current.documents if ref.relative_path in removed] + if any(ref.content_sha256 in current.completed_hashes for ref in removed_refs): + raise ManifestConflictError("recovery cannot replace a compiled document") + + root_resolved = root.resolve() + for ref, content in documents: + path = (root_resolved / ref.relative_path).resolve() + if not path.is_relative_to(root_resolved): + raise ValueError("document path escapes the materialization root") + if hashlib.sha256(content).hexdigest() != ref.content_sha256: + raise ValueError("document hash does not match rendered bytes") + old = refs_by_path.get(ref.relative_path) + if old is not None and old != ref: + raise ManifestConflictError( + f"document path has conflicting metadata: {ref.relative_path}" + ) + if path.exists() and hashlib.sha256(path.read_bytes()).hexdigest() != ref.content_sha256: + raise ManifestConflictError(f"document replay has different bytes: {ref.relative_path}") + + for ref, content in documents: + path = root_resolved / ref.relative_path + if not path.exists(): + atomic_write_bytes(path, content) + refs_by_path[ref.relative_path] = ref + + current_windows[window_record.identity] = window_record + merged = Manifest( + documents=tuple(sorted(refs_by_path.values(), key=lambda ref: ref.relative_path)), + windows=tuple(sorted(current_windows.values(), key=lambda record: record.identity)), + completed_hashes=current.completed_hashes, + compile_receipts=current.compile_receipts, + ) + _save_manifest_unlocked(root, merged, manifest_filename) + return merged + + +def materialize_window( + window: ConversationWindow, + output_dir: Path | str, + *, + max_turn_chars: int, + max_document_chars: int, + hooks: Iterable[TextHook] = (), + manifest_filename: str = "manifest.json", +) -> tuple[DocumentRef, ...]: + """Materialize one closed window exactly once.""" + if not window.closed: + raise ValueError("only closed windows can be materialized") + root = Path(output_dir) + cleaned = tuple( + replace(turn, text=clean_text(turn.text, hooks)) for turn in deduplicate_turns(window.turns) + ) + segments = split_turns( + cleaned, + window=window, + max_turn_chars=max_turn_chars, + max_document_chars=max_document_chars, + ) + documents = _refs_and_bytes( + window, + segments, + max_document_chars=max_document_chars, + ) + refs = tuple(ref for ref, _ in documents) + record = _window_record(window, _window_sha256(window, cleaned), refs) + root.mkdir(parents=True, exist_ok=True) + with kb_ingest_lock(root / ".openkb"): + _write_documents_unlocked( + root, + documents, + record, + manifest_filename=manifest_filename, + ) + return refs + + +def _prune_empty_parents(path: Path, stop: Path) -> None: + current = path + while current != stop and current.is_relative_to(stop): + try: + current.rmdir() + except OSError: + return + current = current.parent + + +def recover_uncompiled( + window: ConversationWindow, + output_dir: Path | str, + *, + max_turn_chars: int, + max_document_chars: int, + recovery_max_document_chars: int, + hooks: Iterable[TextHook] = (), + manifest_filename: str = "manifest.json", +) -> tuple[DocumentRef, ...]: + """Replace only uncompiled documents with documents that use a smaller limit.""" + if not window.closed: + raise ValueError("only closed windows can be recovered") + if recovery_max_document_chars <= 0 or recovery_max_document_chars >= max_document_chars: + raise ValueError("recovery_max_document_chars must be smaller than max_document_chars") + root = Path(output_dir) + cleaned = tuple( + replace(turn, text=clean_text(turn.text, hooks)) for turn in deduplicate_turns(window.turns) + ) + window_sha256 = _window_sha256(window, cleaned) + root_resolved = root.resolve() + with kb_ingest_lock(root / ".openkb"): + manifest = _load_manifest_unlocked(root, manifest_filename) + identity = (window.source_id, window.window_id) + stored_record = next( + (record for record in manifest.windows if record.identity == identity), None + ) + if stored_record is None: + raise ManifestConflictError("recovery cannot find the closed window record") + if stored_record.window_sha256 != window_sha256: + raise ManifestConflictError("closed window bytes conflict with the stored fingerprint") + completed = set(manifest.completed_hashes) + active_refs = [ + ref for ref in manifest.documents if (ref.source_id, ref.window_id) == identity + ] + pending = [ref for ref in active_refs if ref.content_sha256 not in completed] + if not pending: + return () + pending_identities = { + turn_identity for ref in pending for turn_identity in ref.turn_identities + } + turns = tuple(turn for turn in cleaned if turn.identity in pending_identities) + if {turn.identity for turn in turns} != pending_identities: + raise ManifestConflictError("recovery input does not match pending turn identities") + segments = split_turns( + turns, + window=window, + max_turn_chars=max_turn_chars, + max_document_chars=recovery_max_document_chars, + ) + documents = _refs_and_bytes( + window, + segments, + max_document_chars=recovery_max_document_chars, + directory="recovery", + ) + new_refs = tuple(ref for ref, _ in documents) + retained = [ref for ref in active_refs if ref.content_sha256 in completed] + new_record = _window_record(window, window_sha256, [*retained, *new_refs]) + replaced_paths = {ref.relative_path for ref in pending} + _write_documents_unlocked( + root, + documents, + new_record, + manifest_filename=manifest_filename, + replace_paths=replaced_paths, + allow_window_replacement=True, + ) + + new_paths = {ref.relative_path for ref in new_refs} + for ref in pending: + if ref.relative_path in new_paths: + continue + if ref.content_sha256 in completed: + raise ManifestConflictError("recovery cannot remove a compiled document") + stale_path = (root_resolved / ref.relative_path).resolve() + if not stale_path.is_relative_to(root_resolved): + raise ManifestConflictError("stale document path escapes the materialization root") + stale_path.unlink(missing_ok=True) + _prune_empty_parents(stale_path.parent, root_resolved) + return new_refs + + +materialize = materialize_window +recover = recover_uncompiled diff --git a/openkb/conversation/models.py b/openkb/conversation/models.py new file mode 100644 index 000000000..caac7cfb4 --- /dev/null +++ b/openkb/conversation/models.py @@ -0,0 +1,244 @@ +"""Typed records for closed conversation materialization.""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import PurePosixPath + +from openkb.claims import AUTHORITY_ROLES + +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") + + +def validate_sha256(value: str, field_name: str = "SHA-256") -> str: + """Return a valid lowercase SHA-256 value.""" + if not isinstance(value, str) or _SHA256_RE.fullmatch(value) is None: + raise ValueError(f"{field_name} must be 64 lowercase hexadecimal characters") + return value + + +def utc_timestamp(value: str, field_name: str = "timestamp") -> datetime: + """Parse one ISO timestamp and normalize it to UTC.""" + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{field_name} must be a non-empty ISO timestamp") + candidate = value.strip() + normalized = candidate[:-1] + "+00:00" if candidate.endswith("Z") else candidate + try: + parsed = datetime.fromisoformat(normalized) + except ValueError as exc: + raise ValueError(f"{field_name} must be a valid ISO timestamp") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise ValueError(f"{field_name} must include a UTC offset") + return parsed.astimezone(timezone.utc) + + +def _validate_relative_path(value: str) -> str: + if not isinstance(value, str) or not value: + raise ValueError("relative_path must be a non-empty relative path") + path = PurePosixPath(value) + if path.is_absolute() or ".." in path.parts or "\\" in value: + raise ValueError("relative_path must stay below the materialization root") + return value + + +@dataclass(frozen=True, slots=True) +class ConversationTurn: + """One caller-supplied conversation turn.""" + + conversation_id: str + turn_id: str + speaker: str + authority: str + timestamp: str + text: str + source_anchor: str + + def __post_init__(self) -> None: + for name in ("conversation_id", "turn_id", "speaker", "source_anchor"): + value = getattr(self, name) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{name} must be a non-empty string") + utc_timestamp(self.timestamp) + if not isinstance(self.text, str): + raise TypeError("text must be a string") + if self.authority not in AUTHORITY_ROLES: + raise ValueError(f"authority must be one of {sorted(AUTHORITY_ROLES)}") + + @property + def identity(self) -> tuple[str, str]: + """Return the stable identity for this turn.""" + return self.conversation_id, self.turn_id + + @property + def canonical_fields(self) -> tuple[str, ...]: + """Return fields used to compare duplicate turn identities.""" + return ( + self.conversation_id, + self.turn_id, + self.speaker, + self.authority, + self.timestamp, + self.text, + self.source_anchor, + ) + + +@dataclass(frozen=True, slots=True) +class ConversationWindow: + """A caller-defined conversation window.""" + + window_id: str + source_id: str + created_at: str + closed: bool + turns: tuple[ConversationTurn, ...] + + def __post_init__(self) -> None: + for name in ("window_id", "source_id"): + value = getattr(self, name) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{name} must be a non-empty string") + utc_timestamp(self.created_at, "created_at") + if not isinstance(self.closed, bool): + raise TypeError("closed must be a boolean") + turns = tuple(self.turns) + if any(not isinstance(turn, ConversationTurn) for turn in turns): + raise TypeError("turns must contain ConversationTurn values") + object.__setattr__(self, "turns", turns) + + +@dataclass(frozen=True, slots=True) +class DocumentRef: + """Metadata for one immutable materialized document.""" + + relative_path: str + content_sha256: str + window_id: str + source_id: str + created_at: str + segment_index: int + segment_count: int + turn_ids: tuple[str, ...] + turn_identities: tuple[tuple[str, str], ...] + + def __post_init__(self) -> None: + _validate_relative_path(self.relative_path) + validate_sha256(self.content_sha256, "content_sha256") + for name in ("window_id", "source_id"): + value = getattr(self, name) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{name} must be a non-empty string") + utc_timestamp(self.created_at, "created_at") + if self.segment_index < 1 or self.segment_count < 1: + raise ValueError("segment indexes must be positive") + if self.segment_index > self.segment_count: + raise ValueError("segment_index cannot exceed segment_count") + turn_ids = tuple(self.turn_ids) + identities = tuple(tuple(identity) for identity in self.turn_identities) + if any(not isinstance(turn_id, str) or not turn_id for turn_id in turn_ids): + raise ValueError("turn_ids must contain non-empty strings") + if len(identities) != len(turn_ids): + raise ValueError("turn_identities must match turn_ids") + for turn_id, identity in zip(turn_ids, identities): + if len(identity) != 2 or any( + not isinstance(value, str) or not value for value in identity + ): + raise ValueError("each turn identity must contain two non-empty strings") + if identity[1] != turn_id: + raise ValueError("each turn identity must match its turn ID") + object.__setattr__(self, "turn_ids", turn_ids) + object.__setattr__(self, "turn_identities", identities) + + +@dataclass(frozen=True, slots=True) +class WindowRecord: + """Immutable identity and active document set for one closed window.""" + + source_id: str + window_id: str + created_at: str + window_sha256: str + materialization_sha256: str + document_paths: tuple[str, ...] + document_hashes: tuple[str, ...] + + def __post_init__(self) -> None: + for name in ("source_id", "window_id"): + value = getattr(self, name) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{name} must be a non-empty string") + utc_timestamp(self.created_at, "created_at") + validate_sha256(self.window_sha256, "window_sha256") + validate_sha256(self.materialization_sha256, "materialization_sha256") + paths = tuple(self.document_paths) + hashes = tuple(self.document_hashes) + if len(paths) != len(hashes): + raise ValueError("window document paths and hashes must have the same length") + for path in paths: + _validate_relative_path(path) + for digest in hashes: + validate_sha256(digest, "document_hash") + object.__setattr__(self, "document_paths", paths) + object.__setattr__(self, "document_hashes", hashes) + + @property + def identity(self) -> tuple[str, str]: + """Return the stable identity for this window.""" + return self.source_id, self.window_id + + +@dataclass(frozen=True, slots=True) +class CompileReceiptMetadata: + """Optional public metadata for one completed compile operation.""" + + compiled_at: str + duration_seconds: float | None = None + result_sha256: str | None = None + + def __post_init__(self) -> None: + utc_timestamp(self.compiled_at, "compiled_at") + if self.duration_seconds is not None: + if ( + isinstance(self.duration_seconds, bool) + or not isinstance(self.duration_seconds, (int, float)) + or not math.isfinite(self.duration_seconds) + or self.duration_seconds < 0 + ): + raise ValueError("duration_seconds must be a finite non-negative number") + object.__setattr__(self, "duration_seconds", float(self.duration_seconds)) + if self.result_sha256 is not None: + validate_sha256(self.result_sha256, "result_sha256") + + +@dataclass(frozen=True, slots=True) +class CompileReceipt: + """Compile metadata stored by document content hash.""" + + content_sha256: str + metadata: CompileReceiptMetadata + + def __post_init__(self) -> None: + validate_sha256(self.content_sha256, "content_sha256") + if not isinstance(self.metadata, CompileReceiptMetadata): + raise TypeError("metadata must be CompileReceiptMetadata") + + +@dataclass(frozen=True, slots=True) +class TruncationResult: + """Result of an explicit middle truncation.""" + + text: str + original_char_count: int + omitted_middle_sha256: str + omitted_middle_char_count: int + + def __post_init__(self) -> None: + validate_sha256(self.omitted_middle_sha256, "omitted_middle_sha256") + + @property + def omitted_middle_digest(self) -> str: + """Return the omitted-middle digest with a descriptive alias.""" + return self.omitted_middle_sha256 diff --git a/tests/test_conversation_authority.py b/tests/test_conversation_authority.py new file mode 100644 index 000000000..899c17881 --- /dev/null +++ b/tests/test_conversation_authority.py @@ -0,0 +1,176 @@ +"""Tests for caller-verified claim application.""" + +from __future__ import annotations + +import json +import threading + +import pytest + +import openkb.conversation.authority as authority_module +from openkb.claims import claim_id +from openkb.conversation import apply_verified_claims +from openkb.frontmatter import parse + + +def claim( + text: str, + *, + source_anchor: str, + authority: str, + status: str, + as_of: str = "2026-01-01", + supersedes: list[str] | None = None, +) -> dict: + return { + "text": text, + "as_of": as_of, + "status": status, + "source_anchor": source_anchor, + "authority": authority, + "supersedes": supersedes or [], + } + + +def write_page(path, claims): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + f"---\ntype: Concept\nclaims: {json.dumps(claims)}\n---\n\nSynthetic page body.\n", + encoding="utf-8", + ) + + +def test_first_party_claim_supersedes_assistant_proposal_and_body_stays(tmp_path): + old = claim( + "Synthetic device uses revision A.", + source_anchor="fixture:old", + authority="assistant", + status="proposed", + ) + old_id = claim_id(old["text"], old["as_of"], old["source_anchor"]) + page = tmp_path / "wiki" / "concepts" / "page.md" + write_page(page, [old]) + replacement = claim( + "Synthetic device uses revision B.", + source_anchor="fixture:new", + authority="first_party", + status="validated", + as_of="2026-01-02", + supersedes=[old_id], + ) + merged = apply_verified_claims(page, [replacement], openkb_dir=tmp_path / ".openkb") + assert page.read_text(encoding="utf-8").endswith("Synthetic page body.\n") + by_text = {item["text"]: item for item in merged} + assert by_text[old["text"]]["status"] == "superseded" + assert by_text[replacement["text"]]["status"] == "validated" + assert by_text[old["text"]]["superseded_by"] + + +def test_assistant_claim_cannot_remain_validated_or_supersede_first_party(tmp_path): + old = claim( + "Synthetic value is 10.", + source_anchor="fixture:first-party", + authority="first_party", + status="validated", + ) + old_id = claim_id(old["text"], old["as_of"], old["source_anchor"]) + page = tmp_path / "wiki" / "concepts" / "page.md" + write_page(page, [old]) + assistant = claim( + "Assistant says synthetic value is 12.", + source_anchor="fixture:assistant", + authority="assistant", + status="validated", + as_of="2026-01-02", + supersedes=[old_id], + ) + merged = apply_verified_claims(page, [assistant], openkb_dir=tmp_path / ".openkb") + by_text = {item["text"]: item for item in merged} + assert by_text[assistant["text"]]["status"] == "proposed" + assert by_text[old["text"]]["status"] == "validated" + assert by_text[old["text"]]["superseded_by"] == [] + + +def test_replaying_verified_claims_is_idempotent(tmp_path): + page = tmp_path / "wiki" / "concepts" / "page.md" + write_page(page, []) + incoming = [ + claim( + "Synthetic value is 10.", + source_anchor="fixture:value", + authority="first_party", + status="validated", + ) + ] + first = apply_verified_claims(page, incoming, openkb_dir=tmp_path / ".openkb") + first_bytes = page.read_bytes() + second = apply_verified_claims(page, incoming, openkb_dir=tmp_path / ".openkb") + assert second == first + assert page.read_bytes() == first_bytes + assert parse(page.read_text(encoding="utf-8"))["claims"] + + +def test_shared_lock_prevents_lost_concurrent_claim_updates(tmp_path, monkeypatch): + page = tmp_path / "wiki" / "concepts" / "page.md" + write_page(page, []) + first_claim = claim( + "Synthetic claim A.", + source_anchor="fixture:a", + authority="first_party", + status="validated", + ) + second_claim = claim( + "Synthetic claim B.", + source_anchor="fixture:b", + authority="first_party", + status="validated", + ) + original_merge = authority_module.merge_claims + first_merge_entered = threading.Event() + release_first_merge = threading.Event() + + def delayed_merge(existing, new): + if new and new[0].get("text") == first_claim["text"]: + first_merge_entered.set() + assert release_first_merge.wait(timeout=2) + return original_merge(existing, new) + + monkeypatch.setattr(authority_module, "merge_claims", delayed_merge) + errors: list[BaseException] = [] + second_done = threading.Event() + + def apply_one(incoming, done=None): + try: + apply_verified_claims( + page, + [incoming], + openkb_dir=tmp_path / ".openkb", + ) + except BaseException as exc: + errors.append(exc) + finally: + if done is not None: + done.set() + + first_thread = threading.Thread(target=apply_one, args=(first_claim,)) + second_thread = threading.Thread(target=apply_one, args=(second_claim, second_done)) + first_thread.start() + assert first_merge_entered.wait(timeout=2) + second_thread.start() + assert not second_done.wait(timeout=0.1) + release_first_merge.set() + first_thread.join(timeout=2) + second_thread.join(timeout=2) + assert errors == [] + stored = parse(page.read_text(encoding="utf-8"))["claims"] + assert {item["text"] for item in stored} == { + first_claim["text"], + second_claim["text"], + } + + +def test_invalid_page_frontmatter_is_rejected(tmp_path): + page = tmp_path / "page.md" + page.write_text("No frontmatter", encoding="utf-8") + with pytest.raises(ValueError, match="frontmatter"): + apply_verified_claims(page, []) diff --git a/tests/test_conversation_manifest.py b/tests/test_conversation_manifest.py new file mode 100644 index 000000000..319a21af2 --- /dev/null +++ b/tests/test_conversation_manifest.py @@ -0,0 +1,389 @@ +"""Tests for replay-safe conversation manifests.""" + +from __future__ import annotations + +import copy +import hashlib +import json +import threading + +import pytest + +import openkb.conversation.manifest as manifest_module +from openkb.conversation import ( + CompileReceiptMetadata, + ConversationTurn, + ConversationWindow, + DocumentRef, + ManifestError, + WindowRecord, + check_compile_budget, + load_manifest, + mark_compiled, + materialize_window, + register_documents, + select_compile_inputs, + verify_manifest, +) + + +def window(source: str, name: str, created: str, turn_id: str) -> ConversationWindow: + return ConversationWindow( + window_id=name, + source_id=source, + created_at=created, + closed=True, + turns=( + ConversationTurn( + conversation_id=f"conversation-{name}", + turn_id=turn_id, + speaker="first-party", + authority="first_party", + timestamp=created, + text="synthetic value", + source_anchor=f"fixture:{turn_id}", + ), + ), + ) + + +def materialize(root, item: ConversationWindow): + return materialize_window( + item, + root, + max_turn_chars=100, + max_document_chars=2000, + ) + + +def read_raw_manifest(root): + return json.loads((root / "manifest.json").read_text(encoding="utf-8")) + + +def write_raw_manifest(root, raw): + (root / "manifest.json").write_text(json.dumps(raw), encoding="utf-8") + + +def test_manifest_verification_detects_changed_files(tmp_path): + refs = materialize( + tmp_path, + window("source-a", "window-a", "2026-01-02T00:00:00Z", "turn-a"), + ) + assert verify_manifest(tmp_path) is True + (tmp_path / refs[0].relative_path).write_text("changed", encoding="utf-8") + with pytest.raises(ManifestError, match="hash mismatch"): + verify_manifest(tmp_path) + + +def test_compile_selection_uses_utc_time_then_stable_ties(tmp_path): + utc_earlier = materialize( + tmp_path, + window("source-b", "window-b", "2026-01-01T00:30:00+01:00", "turn-b"), + ) + utc_later = materialize( + tmp_path, + window("source-a", "window-a", "2025-12-31T23:45:00Z", "turn-a"), + ) + selected = select_compile_inputs(tmp_path, 2) + assert [ref.window_id for ref in selected] == ["window-b", "window-a"] + + mark_compiled(tmp_path, utc_earlier) + selected = select_compile_inputs(tmp_path, 2) + assert [ref.window_id for ref in selected] == ["window-a"] + assert utc_later[0].content_sha256 not in load_manifest(tmp_path).completed_hashes + + +def test_compile_receipt_is_optional_deterministic_and_keyed_by_hash(tmp_path): + refs = materialize( + tmp_path, + window("source-a", "window-a", "2026-01-01T00:00:00Z", "turn-a"), + ) + receipt = CompileReceiptMetadata( + compiled_at="2026-01-01T01:00:00Z", + duration_seconds=12.5, + result_sha256="a" * 64, + ) + manifest = mark_compiled(tmp_path, refs, receipt=receipt) + assert manifest.compile_receipts[0].content_sha256 == refs[0].content_sha256 + assert manifest.compile_receipts[0].metadata == receipt + first_bytes = (tmp_path / "manifest.json").read_bytes() + assert mark_compiled(tmp_path, refs, receipt=receipt) == manifest + assert (tmp_path / "manifest.json").read_bytes() == first_bytes + + raw = read_raw_manifest(tmp_path) + stored = raw["compile_receipts"][refs[0].content_sha256] + assert set(stored) == {"compiled_at", "duration_seconds", "result_sha256"} + + +def test_compile_receipt_conflict_is_rejected(tmp_path): + refs = materialize( + tmp_path, + window("source-a", "window-a", "2026-01-01T00:00:00Z", "turn-a"), + ) + mark_compiled( + tmp_path, + refs, + receipt=CompileReceiptMetadata(compiled_at="2026-01-01T01:00:00Z"), + ) + with pytest.raises(ValueError, match="receipt conflicts"): + mark_compiled( + tmp_path, + refs, + receipt=CompileReceiptMetadata(compiled_at="2026-01-01T02:00:00Z"), + ) + + +def test_sha256_requires_lowercase_hexadecimal(): + with pytest.raises(ValueError, match="lowercase hexadecimal"): + DocumentRef( + relative_path="fixture.md", + content_sha256="z" * 64, + window_id="window", + source_id="source", + created_at="2026-01-01T00:00:00Z", + segment_index=1, + segment_count=1, + turn_ids=("turn",), + turn_identities=(("conversation", "turn"),), + ) + with pytest.raises(ValueError, match="lowercase hexadecimal"): + CompileReceiptMetadata( + compiled_at="2026-01-01T00:00:00Z", + result_sha256="A" * 64, + ) + with pytest.raises(ValueError, match="materialization root"): + DocumentRef( + relative_path="..\\escape.md", + content_sha256="a" * 64, + window_id="window", + source_id="source", + created_at="2026-01-01T00:00:00Z", + segment_index=1, + segment_count=1, + turn_ids=("turn",), + turn_identities=(("conversation", "turn"),), + ) + + +def valid_raw_manifest(tmp_path): + materialize( + tmp_path, + window("source-a", "window-a", "2026-01-01T00:00:00Z", "turn-a"), + ) + return read_raw_manifest(tmp_path) + + +def test_manifest_load_rejects_duplicate_document_paths(tmp_path): + raw = valid_raw_manifest(tmp_path) + raw["documents"].append(copy.deepcopy(raw["documents"][0])) + write_raw_manifest(tmp_path, raw) + with pytest.raises(ManifestError, match="duplicate document paths"): + load_manifest(tmp_path) + + +def test_manifest_load_rejects_conflicting_duplicate_windows(tmp_path): + raw = valid_raw_manifest(tmp_path) + duplicate = copy.deepcopy(raw["windows"][0]) + duplicate["window_sha256"] = "f" * 64 + raw["windows"].append(duplicate) + write_raw_manifest(tmp_path, raw) + with pytest.raises(ManifestError, match="duplicate window records"): + load_manifest(tmp_path) + + +def test_manifest_load_rejects_unknown_completed_hashes(tmp_path): + raw = valid_raw_manifest(tmp_path) + raw["completed_hashes"] = ["f" * 64] + write_raw_manifest(tmp_path, raw) + with pytest.raises(ManifestError, match="unknown completed hash"): + load_manifest(tmp_path) + + +def test_manifest_load_rejects_malformed_hashes_and_timestamps(tmp_path): + valid = valid_raw_manifest(tmp_path) + raw = copy.deepcopy(valid) + raw["documents"][0]["content_sha256"] = "g" * 64 + write_raw_manifest(tmp_path, raw) + with pytest.raises(ManifestError, match="document entry is invalid"): + load_manifest(tmp_path) + + raw = copy.deepcopy(valid) + raw["documents"][0]["created_at"] = "not-a-time" + write_raw_manifest(tmp_path, raw) + with pytest.raises(ManifestError, match="document entry is invalid"): + load_manifest(tmp_path) + + +def test_manifest_load_wraps_corrupt_unicode(tmp_path): + (tmp_path / "manifest.json").write_bytes(b"\xff") + with pytest.raises(ManifestError, match="cannot read manifest"): + load_manifest(tmp_path) + + +def test_compile_selection_uses_custom_manifest_filename(tmp_path): + item = window("source-a", "window-a", "2026-01-01T00:00:00Z", "turn-a") + refs = materialize_window( + item, + tmp_path, + max_turn_chars=100, + max_document_chars=2000, + manifest_filename="conversation-state.json", + ) + assert select_compile_inputs(tmp_path, 10) == () + assert ( + select_compile_inputs( + tmp_path, + 10, + filename="conversation-state.json", + ) + == refs + ) + + +def test_mark_compiled_lock_prevents_lost_concurrent_updates(tmp_path, monkeypatch): + first = materialize( + tmp_path, + window("source-a", "window-a", "2026-01-01T00:00:00Z", "turn-a"), + )[0] + second = materialize( + tmp_path, + window("source-b", "window-b", "2026-01-01T00:00:01Z", "turn-b"), + )[0] + original_save = manifest_module._save_manifest_unlocked + first_save_entered = threading.Event() + release_first_save = threading.Event() + blocked = False + + def delayed_save(root, manifest, filename="manifest.json"): + nonlocal blocked + completed = set(manifest.completed_hashes) + if ( + first.content_sha256 in completed + and second.content_sha256 not in completed + and not blocked + ): + blocked = True + first_save_entered.set() + assert release_first_save.wait(timeout=2) + original_save(root, manifest, filename) + + monkeypatch.setattr(manifest_module, "_save_manifest_unlocked", delayed_save) + errors: list[BaseException] = [] + second_done = threading.Event() + + def run_first(): + try: + mark_compiled(tmp_path, [first]) + except BaseException as exc: + errors.append(exc) + + def run_second(): + try: + mark_compiled(tmp_path, [second]) + except BaseException as exc: + errors.append(exc) + finally: + second_done.set() + + first_thread = threading.Thread(target=run_first) + second_thread = threading.Thread(target=run_second) + first_thread.start() + assert first_save_entered.wait(timeout=2) + second_thread.start() + assert not second_done.wait(timeout=0.1) + release_first_save.set() + first_thread.join(timeout=2) + second_thread.join(timeout=2) + assert errors == [] + assert not first_thread.is_alive() + assert not second_thread.is_alive() + assert set(load_manifest(tmp_path).completed_hashes) == { + first.content_sha256, + second.content_sha256, + } + + +def manual_document(name: str) -> tuple[DocumentRef, WindowRecord]: + content_hash = hashlib.sha256(name.encode("utf-8")).hexdigest() + ref = DocumentRef( + relative_path=f"fixtures/{name}.md", + content_sha256=content_hash, + window_id=f"window-{name}", + source_id=f"source-{name}", + created_at="2026-01-01T00:00:00Z", + segment_index=1, + segment_count=1, + turn_ids=(f"turn-{name}",), + turn_identities=((f"conversation-{name}", f"turn-{name}"),), + ) + record = WindowRecord( + source_id=ref.source_id, + window_id=ref.window_id, + created_at=ref.created_at, + window_sha256=hashlib.sha256(f"window-{name}".encode("utf-8")).hexdigest(), + materialization_sha256=manifest_module._document_set_sha256((ref,)), + document_paths=(ref.relative_path,), + document_hashes=(ref.content_sha256,), + ) + return ref, record + + +def test_register_documents_lock_prevents_lost_concurrent_updates(tmp_path, monkeypatch): + first_ref, first_record = manual_document("a") + second_ref, second_record = manual_document("b") + original_save = manifest_module._save_manifest_unlocked + first_save_entered = threading.Event() + release_first_save = threading.Event() + blocked = False + + def delayed_save(root, manifest, filename="manifest.json"): + nonlocal blocked + paths = {ref.relative_path for ref in manifest.documents} + if ( + first_ref.relative_path in paths + and second_ref.relative_path not in paths + and not blocked + ): + blocked = True + first_save_entered.set() + assert release_first_save.wait(timeout=2) + original_save(root, manifest, filename) + + monkeypatch.setattr(manifest_module, "_save_manifest_unlocked", delayed_save) + errors: list[BaseException] = [] + second_done = threading.Event() + + def register(ref, record, done=None): + try: + register_documents(tmp_path, [ref], window_records=[record]) + except BaseException as exc: + errors.append(exc) + finally: + if done is not None: + done.set() + + first_thread = threading.Thread(target=register, args=(first_ref, first_record)) + second_thread = threading.Thread( + target=register, + args=(second_ref, second_record, second_done), + ) + first_thread.start() + assert first_save_entered.wait(timeout=2) + second_thread.start() + assert not second_done.wait(timeout=0.1) + release_first_save.set() + first_thread.join(timeout=2) + second_thread.join(timeout=2) + assert errors == [] + assert {ref.relative_path for ref in load_manifest(tmp_path).documents} == { + first_ref.relative_path, + second_ref.relative_path, + } + + +def test_compile_budget_requires_positive_safe_values(): + assert check_compile_budget(2, 1200, 3600) is True + with pytest.raises(ValueError): + check_compile_budget(0, 1200, 3600) + with pytest.raises(ValueError, match="less than"): + check_compile_budget(2, 1800, 3600) diff --git a/tests/test_conversation_materialize.py b/tests/test_conversation_materialize.py new file mode 100644 index 000000000..2ddcfce89 --- /dev/null +++ b/tests/test_conversation_materialize.py @@ -0,0 +1,434 @@ +"""Tests for deterministic conversation materialization.""" + +from __future__ import annotations + +import hashlib + +import pytest + +from openkb.conversation import ( + ConversationTurn, + ConversationWindow, + DuplicateTurnError, + ManifestConflictError, + OversizeTurnError, + clean_text, + deduplicate_turns, + load_manifest, + mark_compiled, + materialize_window, + recover_uncompiled, + render_document, + select_compile_inputs, + split_turns, + truncate_with_digest, + verify_manifest, +) + + +def make_turn( + turn_id: str, + text: str, + *, + conversation_id: str = "fixture-conversation", + timestamp: str = "2026-01-01T00:00:00Z", + speaker: str = "first-party", + authority: str = "first_party", +) -> ConversationTurn: + return ConversationTurn( + conversation_id=conversation_id, + turn_id=turn_id, + speaker=speaker, + authority=authority, + timestamp=timestamp, + text=text, + source_anchor=f"fixture:{conversation_id}:{turn_id}", + ) + + +def make_window( + *turns: ConversationTurn, + closed: bool = True, + source_id: str = "fixture-source", + window_id: str = "fixture-window", + created_at: str = "2026-01-01T00:00:00Z", +) -> ConversationWindow: + return ConversationWindow( + window_id=window_id, + source_id=source_id, + created_at=created_at, + closed=closed, + turns=tuple(turns), + ) + + +def rendered_chars( + window: ConversationWindow, + turns: tuple[ConversationTurn, ...], + *, + segment_index: int = 1, + segment_count: int = 1, +) -> int: + return len( + render_document( + window, + turns, + segment_index=segment_index, + segment_count=segment_count, + ).decode("utf-8") + ) + + +def test_cleaning_normalizes_line_endings_and_runs_hooks_in_order(): + calls: list[str] = [] + + def first(value: str) -> str: + calls.append("first") + return value + " one" + + def second(value: str) -> str: + calls.append("second") + return value + " two" + + assert clean_text("a\r\nb\rc\x00", (first, second)) == "a\nb\nc one two" + assert calls == ["first", "second"] + + +def test_explicit_truncation_keeps_edges_and_records_digest(): + result = truncate_with_digest("0123456789", 6) + assert result.text == "012789" + assert result.original_char_count == 10 + assert result.omitted_middle_char_count == 4 + assert result.omitted_middle_sha256 == hashlib.sha256(b"3456").hexdigest() + + +def test_default_oversize_handling_uses_final_rendered_characters(tmp_path): + window = make_window(make_turn("turn-1", "x")) + with pytest.raises(OversizeTurnError): + materialize_window( + window, + tmp_path, + max_turn_chars=1, + max_document_chars=1, + ) + + +def test_first_turn_fails_when_header_fits_but_complete_document_does_not(): + turn = make_turn("turn-1", "x") + window = make_window(turn) + header_chars = rendered_chars(window, ()) + assert header_chars < rendered_chars(window, (turn,)) + with pytest.raises(OversizeTurnError, match="turn-1"): + split_turns( + (turn,), + window=window, + max_turn_chars=1, + max_document_chars=header_chars, + ) + + +def test_exact_duplicate_turns_deduplicate_and_conflicts_fail(tmp_path): + first = make_turn("turn-1", "same") + assert deduplicate_turns((first, first)) == (first,) + with pytest.raises(DuplicateTurnError): + materialize_window( + make_window(first, make_turn("turn-1", "different")), + tmp_path, + max_turn_chars=20, + max_document_chars=2000, + ) + + +def test_timestamp_sort_uses_utc_and_preserves_caller_order_for_ties(): + earlier = make_turn( + "turn-b", + "earlier", + conversation_id="conversation-b", + timestamp="2026-01-01T00:30:00+01:00", + ) + later = make_turn( + "turn-a", + "later", + conversation_id="conversation-a", + timestamp="2025-12-31T23:45:00Z", + ) + assert deduplicate_turns((later, earlier)) == (earlier, later) + + tie_first = make_turn( + "turn-z", + "first", + timestamp="2026-01-01T01:00:00+01:00", + ) + tie_second = make_turn( + "turn-a", + "second", + timestamp="2026-01-01T00:00:00Z", + ) + assert deduplicate_turns((tie_first, tie_second)) == (tie_first, tie_second) + + +def test_malformed_turn_and_window_timestamps_are_rejected(): + with pytest.raises(ValueError, match="ISO timestamp"): + make_turn("turn-1", "text", timestamp="not-a-time") + with pytest.raises(ValueError, match="UTC offset"): + make_turn("turn-1", "text", timestamp="2026-01-01T00:00:00") + with pytest.raises(ValueError, match="created_at"): + make_window(created_at="not-a-time") + + +def test_splitting_preserves_order_and_limits_rendered_documents(): + turns = ( + make_turn("turn-1", "aa", timestamp="2026-01-01T00:00:01Z"), + make_turn("turn-2", "bb", timestamp="2026-01-01T00:00:02Z"), + make_turn("turn-3", "cc", timestamp="2026-01-01T00:00:03Z"), + ) + window = make_window(*turns) + limit = max( + rendered_chars(window, (turn,), segment_index=index, segment_count=3) + for index, turn in enumerate(turns, start=1) + ) + segments = split_turns( + turns, + window=window, + max_turn_chars=2, + max_document_chars=limit, + ) + assert [[turn.turn_id for turn in segment] for segment in segments] == [ + ["turn-1"], + ["turn-2"], + ["turn-3"], + ] + assert [turn.identity for segment in segments for turn in segment] == [ + turn.identity for turn in turns + ] + for index, segment in enumerate(segments, start=1): + assert ( + rendered_chars( + window, + segment, + segment_index=index, + segment_count=len(segments), + ) + <= limit + ) + + +def test_rendering_contains_all_evidence_fields(): + window = make_window(make_turn("turn-1", "synthetic text")) + rendered = render_document(window, window.turns, segment_index=1, segment_count=1).decode() + for value in ( + "fixture-window", + "fixture-source", + "turn-1", + "first-party", + "first_party", + "2026-01-01T00:00:00Z", + "fixture:fixture-conversation:turn-1", + "synthetic text", + ): + assert value in rendered + assert "Evidence rules" in rendered + + +def test_closed_window_replay_rejects_changed_bytes_and_segment_set(tmp_path): + turns = ( + make_turn("turn-1", "aaaa", timestamp="2026-01-01T00:00:01Z"), + make_turn("turn-2", "bbbb", timestamp="2026-01-01T00:00:02Z"), + ) + window = make_window(*turns) + one_segment_limit = rendered_chars(window, turns) + refs = materialize_window( + window, + tmp_path, + max_turn_chars=10, + max_document_chars=one_segment_limit, + ) + path = tmp_path / refs[0].relative_path + first_bytes = path.read_bytes() + assert ( + materialize_window( + window, + tmp_path, + max_turn_chars=10, + max_document_chars=one_segment_limit, + ) + == refs + ) + assert path.read_bytes() == first_bytes + + changed = make_window(make_turn("turn-1", "changed"), turns[1]) + with pytest.raises(ManifestConflictError, match="closed window"): + materialize_window( + changed, + tmp_path, + max_turn_chars=20, + max_document_chars=2000, + ) + + split_limit = max( + rendered_chars(window, (turn,), segment_index=index, segment_count=2) + for index, turn in enumerate(turns, start=1) + ) + assert split_limit < one_segment_limit + with pytest.raises(ManifestConflictError, match="segment set"): + materialize_window( + window, + tmp_path, + max_turn_chars=10, + max_document_chars=split_limit, + ) + + +def test_empty_closed_window_is_recorded_and_immutable(tmp_path): + empty = make_window() + assert ( + materialize_window( + empty, + tmp_path, + max_turn_chars=20, + max_document_chars=2000, + ) + == () + ) + manifest = load_manifest(tmp_path) + assert manifest.documents == () + assert len(manifest.windows) == 1 + assert ( + materialize_window( + empty, + tmp_path, + max_turn_chars=20, + max_document_chars=2000, + ) + == () + ) + + changed = make_window(make_turn("turn-1", "new")) + with pytest.raises(ManifestConflictError, match="closed window"): + materialize_window( + changed, + tmp_path, + max_turn_chars=20, + max_document_chars=2000, + ) + + +def test_recovery_uses_full_identity_removes_stale_files_and_keeps_compiled(tmp_path): + first = make_turn( + "same-turn", + "aa", + conversation_id="conversation-a", + timestamp="2026-01-01T00:00:01Z", + ) + second = make_turn( + "same-turn", + "bb", + conversation_id="conversation-b", + timestamp="2026-01-01T00:00:02Z", + ) + window = make_window(first, second) + single_sizes = [ + rendered_chars(window, (turn,), segment_index=index, segment_count=2) + for index, turn in enumerate((first, second), start=1) + ] + original_limit = max(single_sizes) + 5 + assert rendered_chars(window, (first, second)) > original_limit + original = materialize_window( + window, + tmp_path, + max_turn_chars=2, + max_document_chars=original_limit, + ) + assert len(original) == 2 + compiled_path = tmp_path / original[0].relative_path + stale_path = tmp_path / original[1].relative_path + compiled_bytes = compiled_path.read_bytes() + mark_compiled(tmp_path, [original[0]]) + + recovery_limit = rendered_chars(window, (second,)) + assert recovery_limit < original_limit + recovered = recover_uncompiled( + window, + tmp_path, + max_turn_chars=2, + max_document_chars=original_limit, + recovery_max_document_chars=recovery_limit, + ) + assert len(recovered) == 1 + assert recovered[0].turn_identities == (("conversation-b", "same-turn"),) + assert [ref.turn_identities for ref in select_compile_inputs(tmp_path, 10)] == [ + (("conversation-b", "same-turn"),) + ] + assert compiled_path.read_bytes() == compiled_bytes + assert not stale_path.exists() + assert not any(path == stale_path for path in tmp_path.rglob("*.md")) + assert verify_manifest(tmp_path) is True + + +def test_recovery_oversize_keeps_pending_source_and_manifest_unchanged(tmp_path): + turn = make_turn("turn-1", "x") + window = make_window(turn) + complete_chars = rendered_chars(window, (turn,)) + header_chars = rendered_chars(window, ()) + assert header_chars < complete_chars + original = materialize_window( + window, + tmp_path, + max_turn_chars=1, + max_document_chars=complete_chars, + ) + pending_path = tmp_path / original[0].relative_path + pending_bytes = pending_path.read_bytes() + manifest_path = tmp_path / "manifest.json" + manifest_bytes = manifest_path.read_bytes() + + with pytest.raises(OversizeTurnError, match="turn-1"): + recover_uncompiled( + window, + tmp_path, + max_turn_chars=1, + max_document_chars=complete_chars, + recovery_max_document_chars=header_chars, + ) + + assert pending_path.read_bytes() == pending_bytes + assert manifest_path.read_bytes() == manifest_bytes + assert select_compile_inputs(tmp_path, 10) == original + + +def test_collision_resistant_path_tokens_separate_opaque_ids(tmp_path): + first = make_window( + make_turn("turn-1", "x", conversation_id="conversation-a"), + source_id="opaque/a", + window_id="window/a", + ) + second = make_window( + make_turn("turn-1", "x", conversation_id="conversation-b"), + source_id="opaque?a", + window_id="window?a", + ) + first_ref = materialize_window( + first, + tmp_path, + max_turn_chars=2, + max_document_chars=2000, + )[0] + second_ref = materialize_window( + second, + tmp_path, + max_turn_chars=2, + max_document_chars=2000, + )[0] + assert first_ref.relative_path != second_ref.relative_path + assert (tmp_path / first_ref.relative_path).is_file() + assert (tmp_path / second_ref.relative_path).is_file() + + +def test_open_window_is_rejected(tmp_path): + with pytest.raises(ValueError, match="closed"): + materialize_window( + make_window(make_turn("turn-1", "text"), closed=False), + tmp_path, + max_turn_chars=20, + max_document_chars=2000, + )