diff --git a/src/basic_memory/indexing/wiki_projector.py b/src/basic_memory/indexing/wiki_projector.py new file mode 100644 index 000000000..267911210 --- /dev/null +++ b/src/basic_memory/indexing/wiki_projector.py @@ -0,0 +1,808 @@ +"""Deterministic, storage-neutral planning for the Basic Memory Wiki Projector.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from enum import StrEnum +from hashlib import sha256 +import json +from pathlib import PurePosixPath, PureWindowsPath +import unicodedata + +OKF_VERSION = "0.2" +WIKI_PROFILE = "wiki/1" +WIKI_PROJECTOR_VERSION = "wiki/1.0.0" +WIKI_PROJECTOR_NAME = "Basic Memory Wiki Projector" +WIKI_PROJECTOR_SOURCE = "wiki_projector" +RESERVED_WIKI_FILENAMES = frozenset({"index.md", "log.md"}) +WINDOWS_RESERVED_NAMES = frozenset( + {"CON", "PRN", "AUX", "NUL"} + | {f"COM{number}" for number in range(1, 10)} + | {f"LPT{number}" for number in range(1, 10)} +) + + +class WikiProjectionReason(StrEnum): + """Why a projector run was requested.""" + + accepted_note = "accepted_note" + project_created = "project_created" + import_rebuild = "import_rebuild" + manual_rebuild = "manual_rebuild" + + +class WikiChangeOperation(StrEnum): + """Accepted note operation represented in generated Wiki logs.""" + + created = "created" + updated = "updated" + moved = "moved" + deleted = "deleted" + + +class WikiProjectionState(StrEnum): + """User-visible state derived from a projector result or run ledger.""" + + current = "current" + updating = "updating" + partial = "partial" + conflicted = "conflicted" + failed = "failed" + + +@dataclass(frozen=True, slots=True) +class WikiProjectionRequest: + """Portable request consumed by local and Cloud projector adapters.""" + + project_id: str + through_partition_position: int + projector_version: str + reason: WikiProjectionReason + requested_scopes: tuple[str, ...] = () + + def __post_init__(self) -> None: + if not self.project_id.strip(): + raise ValueError("Wiki projection requires a project_id") + if self.through_partition_position < 0: + raise ValueError("Wiki projection position cannot be negative") + if self.projector_version != WIKI_PROJECTOR_VERSION: + raise ValueError(f"Wiki projection requires projector_version {WIKI_PROJECTOR_VERSION}") + normalized_scopes = tuple( + sorted({_normalize_scope(scope) for scope in self.requested_scopes}) + ) + object.__setattr__(self, "requested_scopes", normalized_scopes) + + @property + def is_full_rebuild(self) -> bool: + return self.reason in { + WikiProjectionReason.import_rebuild, + WikiProjectionReason.manual_rebuild, + } + + +@dataclass(frozen=True, slots=True) +class WikiSourceNote: + """One accepted, materialized note visible to a projector snapshot.""" + + path: str + permalink: str + title: str + note_type: str + checksum: str + + def __post_init__(self) -> None: + object.__setattr__(self, "path", _normalize_note_path(self.path)) + _validate_canonical_permalink(self.permalink, label=f"Wiki source note {self.path}") + if not self.title.strip(): + raise ValueError(f"Wiki source note {self.path} requires a title") + if not self.note_type.strip(): + raise ValueError(f"Wiki source note {self.path} requires a note_type") + if not self.checksum.strip(): + raise ValueError(f"Wiki source note {self.path} requires a checksum") + + +@dataclass(frozen=True, slots=True) +class WikiSourceChange: + """One accepted project-partition change used for materialization-aware logs.""" + + partition_position: int + operation: WikiChangeOperation + path: str + permalink: str + title: str + accepted_at: datetime + materialized: bool + source: str + previous_path: str | None = None + + def __post_init__(self) -> None: + if self.partition_position <= 0: + raise ValueError("Wiki source change position must be positive") + object.__setattr__(self, "path", _normalize_note_path(self.path)) + _validate_canonical_permalink(self.permalink, label=f"Wiki source change {self.path}") + if self.previous_path is not None: + object.__setattr__(self, "previous_path", _normalize_note_path(self.previous_path)) + if not self.title.strip(): + raise ValueError(f"Wiki source change {self.path} requires a title") + if self.accepted_at.tzinfo is None: + raise ValueError("Wiki source change accepted_at must be timezone-aware") + if not self.source.strip(): + raise ValueError("Wiki source change requires a source") + + +@dataclass(frozen=True, slots=True) +class WikiReservedDocument: + """Current accepted state for a path reserved to the Wiki Projector.""" + + path: str + checksum: str + content: bytes + projector_owned: bool + + def __post_init__(self) -> None: + normalized_path = _normalize_note_path(self.path) + if PurePosixPath(normalized_path).name.lower() not in RESERVED_WIKI_FILENAMES: + raise ValueError(f"Wiki reserved document has non-reserved path: {self.path}") + if not self.checksum.strip(): + raise ValueError(f"Wiki reserved document {self.path} requires a checksum") + object.__setattr__(self, "path", normalized_path) + + +@dataclass(frozen=True, slots=True) +class WikiProjectionSnapshot: + """Complete deterministic input needed to plan one projector run.""" + + project_id: str + project_name: str + source_partition_position: int + current_output_watermark: int + source_accepted_at: datetime + notes: tuple[WikiSourceNote, ...] + changes: tuple[WikiSourceChange, ...] + reserved_documents: tuple[WikiReservedDocument, ...] = () + + def __post_init__(self) -> None: + if not self.project_id.strip(): + raise ValueError("Wiki projection snapshot requires a project_id") + if not self.project_name.strip(): + raise ValueError("Wiki projection snapshot requires a project_name") + if self.source_partition_position < 0: + raise ValueError("Wiki snapshot source position cannot be negative") + if self.current_output_watermark < 0: + raise ValueError("Wiki output watermark cannot be negative") + if self.source_accepted_at.tzinfo is None: + raise ValueError("Wiki snapshot source_accepted_at must be timezone-aware") + _require_unique_paths(self.notes, label="source note") + _require_unique_paths( + self.reserved_documents, + label="reserved document", + case_sensitive=False, + ) + positions = [change.partition_position for change in self.changes] + if len(positions) != len(set(positions)): + raise ValueError("Wiki source changes require unique partition positions") + + +@dataclass(frozen=True, slots=True) +class WikiProjectionWrite: + """Checksum-protected canonical Markdown write planned for an adapter.""" + + path: str + content: bytes + checksum: str + expected_checksum: str | None + + +@dataclass(frozen=True, slots=True) +class WikiProjectionConflict: + """Reserved path the projector cannot safely claim or replace.""" + + path: str + reason: str + + +@dataclass(frozen=True, slots=True) +class WikiProjectionResult: + """Portable outcome recorded by local and Cloud run ledgers.""" + + source_watermark: int + output_watermark: int + created: int + updated: int + unchanged: int + conflicts: tuple[WikiProjectionConflict, ...] + warnings: tuple[str, ...] + pending_materialization: tuple[int, ...] + + @property + def state(self) -> WikiProjectionState: + if self.conflicts: + return WikiProjectionState.conflicted + if self.pending_materialization: + return WikiProjectionState.partial + if self.output_watermark < self.source_watermark: + return WikiProjectionState.updating + return WikiProjectionState.current + + +@dataclass(frozen=True, slots=True) +class WikiProjectionPlan: + """Pure projection result plus writes for a runtime adapter to execute.""" + + request: WikiProjectionRequest + writes: tuple[WikiProjectionWrite, ...] + unchanged_paths: tuple[str, ...] + result: WikiProjectionResult + + +def affected_wiki_scopes(*paths: str | None) -> tuple[str, ...]: + """Return root and every ancestor directory affected by note paths.""" + scopes = {""} + for path in paths: + if path is None: + continue + parent = PurePosixPath(_normalize_note_path(path)).parent + while parent != PurePosixPath("."): + scopes.add(parent.as_posix()) + parent = parent.parent + return tuple(sorted(scopes)) + + +def plan_wiki_projection( + request: WikiProjectionRequest, + snapshot: WikiProjectionSnapshot, +) -> WikiProjectionPlan: + """Plan deterministic OKF index/log writes without performing I/O.""" + if request.project_id != snapshot.project_id: + raise ValueError("Wiki projection request and snapshot project_id differ") + if request.through_partition_position < snapshot.current_output_watermark: + raise ValueError("Wiki projection request is older than the current output watermark") + if request.through_partition_position != snapshot.source_partition_position: + raise ValueError("Wiki projection requires an exact as-of source snapshot") + + changes = tuple( + sorted( + ( + change + for change in snapshot.changes + if change.partition_position <= request.through_partition_position + and not _is_projector_change(change) + ), + key=lambda change: change.partition_position, + ) + ) + pending = tuple( + change.partition_position + for change in changes + if not change.materialized and change.partition_position > snapshot.current_output_watermark + ) + if pending: + warning = ( + "Projection deferred until accepted note positions are materialized: " + + ", ".join(str(position) for position in pending) + ) + return WikiProjectionPlan( + request=request, + writes=(), + unchanged_paths=(), + result=WikiProjectionResult( + source_watermark=request.through_partition_position, + output_watermark=snapshot.current_output_watermark, + created=0, + updated=0, + unchanged=0, + conflicts=(), + warnings=(warning,), + pending_materialization=pending, + ), + ) + + new_changes = tuple( + change + for change in changes + if change.partition_position > snapshot.current_output_watermark + ) + projector_only_advance = ( + request.reason == WikiProjectionReason.accepted_note + and not new_changes + and any( + _is_projector_change(change) + and change.partition_position > snapshot.current_output_watermark + for change in snapshot.changes + ) + ) + notes = tuple( + note + for note in snapshot.notes + if PurePosixPath(note.path).name.lower() not in RESERVED_WIKI_FILENAMES + ) + scopes = _projection_scopes( + request, + snapshot, + notes, + changes, + new_changes, + repair_complete_projection=projector_only_advance, + ) + all_projection_paths: list[str | None] = [note.path for note in notes] + all_projection_paths.extend(document.path for document in snapshot.reserved_documents) + all_projection_paths.extend(change.path for change in changes) + all_projection_paths.extend( + change.previous_path for change in changes if change.previous_path is not None + ) + reserved_permalink_keys = { + _portable_path_key(_reserved_path(scope, filename).removesuffix(".md")) + for scope in affected_wiki_scopes(*all_projection_paths) + for filename in RESERVED_WIKI_FILENAMES + } + for note in notes: + if _portable_path_key(note.permalink) in reserved_permalink_keys: + raise ValueError( + "Wiki source note permalink collides with a generated document identity: " + f"{note.permalink}" + ) + for change in changes: + if ( + change.operation is not WikiChangeOperation.deleted + and _portable_path_key(change.permalink) in reserved_permalink_keys + ): + raise ValueError( + "Wiki source change permalink collides with a generated document identity: " + f"{change.permalink}" + ) + note_by_path = {_portable_path_key(note.path): note for note in notes} + scope_by_portable_path: dict[str, str] = {} + for scope in scopes: + portable_scope = _portable_path_key(scope) + if existing_note := note_by_path.get(portable_scope): + raise ValueError( + "Wiki projection scope collides with an existing source note path: " + f"{scope}, {existing_note.path}" + ) + if existing_scope := scope_by_portable_path.get(portable_scope): + raise ValueError( + "Wiki projection scopes must be unique when compared as portable paths: " + f"{existing_scope}, {scope}" + ) + scope_by_portable_path[portable_scope] = scope + existing_by_path = { + _portable_path_key(document.path): document for document in snapshot.reserved_documents + } + rendered: dict[str, bytes] = {} + for scope in scopes: + rendered[_reserved_path(scope, "index.md")] = _render_index( + snapshot=snapshot, + notes=notes, + scope=scope, + source_watermark=request.through_partition_position, + ) + rendered[_reserved_path(scope, "log.md")] = _render_log( + snapshot=snapshot, + changes=changes, + scope=scope, + source_watermark=request.through_partition_position, + ) + + conflicts = tuple( + WikiProjectionConflict( + path=path, + reason="reserved path is not owned by the Wiki Projector", + ) + for path in sorted(rendered) + if (existing := existing_by_path.get(_portable_path_key(path))) is not None + and not existing.projector_owned + ) + if conflicts: + # Indexes and logs describe one project watermark. Writing only the + # unblocked paths would publish a mixed projection that no ledger + # watermark could honestly represent, so conflict is all-or-nothing. + return WikiProjectionPlan( + request=request, + writes=(), + unchanged_paths=(), + result=WikiProjectionResult( + source_watermark=request.through_partition_position, + output_watermark=snapshot.current_output_watermark, + created=0, + updated=0, + unchanged=0, + conflicts=conflicts, + warnings=(), + pending_materialization=(), + ), + ) + + writes: list[WikiProjectionWrite] = [] + unchanged_paths: list[str] = [] + created = 0 + updated = 0 + for path, content in sorted(rendered.items()): + existing = existing_by_path.get(_portable_path_key(path)) + if existing is not None and ( + existing.content == content + or ( + projector_only_advance + and _without_projection_metadata(existing.content) + == _without_projection_metadata(content) + ) + ): + unchanged_paths.append(path) + continue + writes.append( + WikiProjectionWrite( + path=path, + content=content, + checksum=sha256(content).hexdigest(), + expected_checksum=existing.checksum if existing is not None else None, + ) + ) + if existing is None: + created += 1 + else: + updated += 1 + + return WikiProjectionPlan( + request=request, + writes=tuple(writes), + unchanged_paths=tuple(unchanged_paths), + result=WikiProjectionResult( + source_watermark=request.through_partition_position, + output_watermark=request.through_partition_position, + created=created, + updated=updated, + unchanged=len(unchanged_paths), + conflicts=(), + warnings=(), + pending_materialization=(), + ), + ) + + +def _projection_scopes( + request: WikiProjectionRequest, + snapshot: WikiProjectionSnapshot, + notes: tuple[WikiSourceNote, ...], + changes: tuple[WikiSourceChange, ...], + new_changes: tuple[WikiSourceChange, ...], + *, + repair_complete_projection: bool, +) -> tuple[str, ...]: + if request.is_full_rebuild or repair_complete_projection: + paths = [note.path for note in notes] + paths.extend(document.path for document in snapshot.reserved_documents) + paths.extend(change.path for change in changes) + paths.extend(change.previous_path for change in changes if change.previous_path is not None) + return affected_wiki_scopes(*paths) + paths = [change.path for change in new_changes] + paths.extend(change.previous_path for change in new_changes if change.previous_path is not None) + scopes = set(affected_wiki_scopes(*paths)) + for requested_scope in request.requested_scopes: + scope = PurePosixPath(requested_scope) + while scope != PurePosixPath("."): + scopes.add(scope.as_posix()) + scope = scope.parent + return tuple(sorted(scopes)) + + +def _without_projection_metadata(content: bytes) -> bytes: + frontmatter, separator, body = content.partition(b"\n---\n") + normalized_frontmatter = b"\n".join( + b" at:" + if line.startswith(b" at: ") + else b" source_watermark:" + if line.startswith(b" source_watermark: ") + else line + for line in frontmatter.split(b"\n") + ) + return normalized_frontmatter + separator + body + + +def _render_index( + *, + snapshot: WikiProjectionSnapshot, + notes: tuple[WikiSourceNote, ...], + scope: str, + source_watermark: int, +) -> bytes: + direct_notes = sorted( + (note for note in notes if _parent_scope(note.path) == scope), + key=lambda note: ( + note.title.casefold(), + note.path.casefold(), + note.title, + note.path, + ), + ) + child_scope_set: set[str] = set() + for note in notes: + if not _is_descendant(note.path, scope): + continue + child_scope = _direct_child_scope(scope, note.path) + if child_scope is not None: + child_scope_set.add(child_scope) + child_scopes = sorted(child_scope_set) + title = snapshot.project_name if not scope else _display_name(PurePosixPath(scope).name) + body: list[str] = [f"# {_escape_generated_markdown_text(title)}", ""] + if child_scopes: + body.extend(["## Sections", ""]) + body.extend( + "- " + f"[[{child_scope}/index|" + f"{_escape_generated_markdown_text(_display_name(PurePosixPath(child_scope).name))}]]" + for child_scope in child_scopes + ) + body.append("") + if direct_notes: + body.extend(["## Notes", ""]) + body.extend( + f"- [[{note.permalink}|{_escape_generated_markdown_text(note.title)}]]" + for note in direct_notes + ) + body.append("") + if not child_scopes and not direct_notes: + body.extend(["No concepts have been projected into this scope yet.", ""]) + return _render_document( + note_type="Index", + title=title, + source_watermark=source_watermark, + generated_at=snapshot.source_accepted_at, + body="\n".join(body), + include_okf_version=not scope, + ) + + +def _render_log( + *, + snapshot: WikiProjectionSnapshot, + changes: tuple[WikiSourceChange, ...], + scope: str, + source_watermark: int, +) -> bytes: + relevant = tuple( + sorted( + ( + change + for change in changes + if change.materialized + and ( + _is_descendant(change.path, scope) + or ( + change.previous_path is not None + and _is_descendant(change.previous_path, scope) + ) + ) + ), + key=lambda change: change.partition_position, + reverse=True, + ) + ) + title = ( + f"{snapshot.project_name} log" + if not scope + else f"{_display_name(PurePosixPath(scope).name)} log" + ) + body: list[str] = [f"# {_escape_generated_markdown_text(title)}", ""] + if relevant: + body.extend(_render_log_entry(change) for change in relevant) + body.append("") + else: + body.extend(["No accepted materialized changes have been recorded yet.", ""]) + return _render_document( + note_type="Log", + title=title, + source_watermark=source_watermark, + generated_at=snapshot.source_accepted_at, + body="\n".join(body), + include_okf_version=False, + ) + + +def _render_log_entry(change: WikiSourceChange) -> str: + timestamp = _isoformat_utc(change.accepted_at) + title = _escape_generated_markdown_text(change.title) + current_note = f"[[{change.permalink}|{title}]]" + match change.operation: + case WikiChangeOperation.created: + description = f"Created {current_note}" + case WikiChangeOperation.updated: + description = f"Updated {current_note}" + case WikiChangeOperation.moved: + if change.previous_path is None: + raise ValueError("Moved Wiki change requires previous_path") + description = f"Moved `{change.previous_path}` to {current_note}" + case WikiChangeOperation.deleted: + description = f"Deleted `{change.path}`" + return f"- {timestamp} — {description}" + + +def _render_document( + *, + note_type: str, + title: str, + source_watermark: int, + generated_at: datetime, + body: str, + include_okf_version: bool, +) -> bytes: + frontmatter = ["---", f"type: {note_type}"] + if include_okf_version: + frontmatter.append(f'okf_version: "{OKF_VERSION}"') + frontmatter.extend( + [ + f"title: {json.dumps(title, ensure_ascii=False)}", + "generated:", + f" by: {WIKI_PROJECTOR_NAME}", + f" at: {json.dumps(_isoformat_utc(generated_at))}", + "bm:", + f" profile: {WIKI_PROFILE}", + f' source_watermark: "{source_watermark}"', + "---", + body, + ] + ) + return ("\n".join(frontmatter).rstrip() + "\n").encode("utf-8") + + +def _is_projector_change(change: WikiSourceChange) -> bool: + return ( + change.source == WIKI_PROJECTOR_SOURCE + and PurePosixPath(change.path).name.lower() in RESERVED_WIKI_FILENAMES + ) + + +def _normalize_note_path(path: str) -> str: + normalized = _normalize_relative_path(path) + if not normalized or PurePosixPath(normalized).suffix.lower() != ".md": + raise ValueError(f"Wiki note path must be project-relative Markdown: {path}") + if "::" in normalized or any(character in normalized for character in "\r\n[]|`<>"): + raise ValueError(f"Wiki note path contains unsupported Markdown delimiters: {path}") + _validate_portable_path_components(normalized, path_kind="note path", source=path) + _reject_reserved_wiki_directory_components( + PurePosixPath(normalized).parts[:-1], + path_kind="note path", + source=path, + ) + return normalized + + +def _validate_canonical_permalink(permalink: str, *, label: str) -> None: + if not permalink.strip() or permalink != permalink.strip(): + raise ValueError(f"{label} requires a canonical permalink") + if "::" in permalink or any(character in permalink for character in "\r\n[]|`<>"): + raise ValueError(f"{label} has an unsafe canonical permalink") + + +def _normalize_scope(scope: str) -> str: + normalized = _normalize_relative_path(scope) + if not normalized: + return "" + if "::" in normalized or any(character in normalized for character in "\x00\r\n[]|`<>"): + raise ValueError(f"Wiki scope contains unsupported Markdown delimiters: {scope}") + _validate_portable_path_components(normalized, path_kind="scope", source=scope) + _reject_reserved_wiki_directory_components( + PurePosixPath(normalized).parts, + path_kind="scope", + source=scope, + ) + return normalized + + +def _reject_reserved_wiki_directory_components( + components: tuple[str, ...], + *, + path_kind: str, + source: str, +) -> None: + if any(component.casefold() in RESERVED_WIKI_FILENAMES for component in components): + raise ValueError(f"Wiki {path_kind} contains a reserved Wiki directory name: {source}") + + +def _validate_portable_path_components( + normalized: str, + *, + path_kind: str, + source: str, +) -> None: + for component in PurePosixPath(normalized).parts: + if any(unicodedata.category(character) == "Cc" for character in component): + raise ValueError(f"Wiki {path_kind} contains a control character: {source}") + if any(character in component for character in ':"?*'): + raise ValueError(f"Wiki {path_kind} contains a Windows-invalid character: {source}") + if component.endswith((".", " ")): + raise ValueError(f"Wiki {path_kind} contains a non-portable path component: {source}") + stem = component.split(".", 1)[0].upper() + if stem in WINDOWS_RESERVED_NAMES: + raise ValueError(f"Wiki {path_kind} contains a reserved device name: {source}") + + +def _normalize_relative_path(path: str) -> str: + if path != path.strip(): + raise ValueError(f"Wiki path must not contain boundary whitespace: {path}") + accepted_path = path + windows_path = PureWindowsPath(accepted_path) + if accepted_path.startswith(("/", "\\")) or windows_path.drive or windows_path.is_absolute(): + raise ValueError(f"Wiki path must be project-relative and normalized: {path}") + candidate = accepted_path.replace("\\", "/") + if not candidate: + return "" + if candidate.endswith("/"): + raise ValueError(f"Wiki path must be project-relative and normalized: {path}") + parsed = PurePosixPath(candidate) + if ( + parsed.is_absolute() + or any(part in {"", ".", ".."} for part in parsed.parts) + or parsed.as_posix() != candidate + ): + raise ValueError(f"Wiki path must be project-relative and normalized: {path}") + return parsed.as_posix() + + +def _escape_generated_markdown_text(value: str) -> str: + """Keep snapshot metadata from changing generated Markdown structure.""" + return value.translate( + str.maketrans( + { + "\r": " ", + "\n": " ", + "&": "&", + "\\": "\", + "[": "[", + "]": "]", + "|": "|", + "`": "`", + "<": "<", + ">": ">", + } + ) + ) + + +def _require_unique_paths( + values: tuple[object, ...], + *, + label: str, + case_sensitive: bool = True, +) -> None: + paths = [getattr(value, "path") for value in values] + if not case_sensitive: + paths = [_portable_path_key(path) for path in paths] + if len(paths) != len(set(paths)): + raise ValueError(f"Wiki projection snapshot has duplicate {label} paths") + + +def _portable_path_key(path: str) -> str: + """Compare paths the way normalization-insensitive filesystems do.""" + return unicodedata.normalize("NFC", path).casefold() + + +def _reserved_path(scope: str, filename: str) -> str: + return f"{scope}/{filename}" if scope else filename + + +def _parent_scope(path: str) -> str: + parent = PurePosixPath(path).parent + return "" if parent == PurePosixPath(".") else parent.as_posix() + + +def _is_descendant(path: str, scope: str) -> bool: + if not scope: + return True + return path == scope or path.startswith(f"{scope}/") + + +def _direct_child_scope(scope: str, note_path: str) -> str | None: + note_parent = _parent_scope(note_path) + if not note_parent or note_parent == scope: + return None + prefix = f"{scope}/" if scope else "" + child_name = note_parent[len(prefix) :].split("/", maxsplit=1)[0] + return f"{scope}/{child_name}" if scope else child_name + + +def _display_name(value: str) -> str: + return value.replace("-", " ").replace("_", " ").strip().title() + + +def _isoformat_utc(value: datetime) -> str: + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") diff --git a/tests/fixtures/wiki_projector/basic_projection.json b/tests/fixtures/wiki_projector/basic_projection.json new file mode 100644 index 000000000..0ef9090d1 --- /dev/null +++ b/tests/fixtures/wiki_projector/basic_projection.json @@ -0,0 +1,11 @@ +{ + "contract_version": "wiki/1.0.0", + "project_id": "project-88", + "through_partition_position": 3, + "expected_sha256": { + "guides/index.md": "c6481bd17c663a3c2595cb35cd22a03da46c6d55465a274482a91363b3af244a", + "guides/log.md": "a3bf317e661d40a156a7481478938b8a652d4a69184b06cfa8c1b8b5355307fb", + "index.md": "33a4af87c4c1a3d4e128f780e67a8a7084aa1796118fb7a5334bbec4a4e0728d", + "log.md": "8e21ce176e930556f6a39f30412af7c488f9724b4118d2946879b72d0eb6c2f3" + } +} diff --git a/tests/indexing/test_wiki_projector.py b/tests/indexing/test_wiki_projector.py new file mode 100644 index 000000000..3c23419be --- /dev/null +++ b/tests/indexing/test_wiki_projector.py @@ -0,0 +1,1130 @@ +"""Deterministic Wiki Projector contract and byte-output tests.""" + +from dataclasses import replace +from datetime import datetime, timezone +from hashlib import sha256 +import json +from pathlib import Path + +import pytest + +from basic_memory.indexing.wiki_projector import ( + WikiChangeOperation, + WikiProjectionReason, + WikiProjectionRequest, + WikiProjectionResult, + WikiProjectionSnapshot, + WikiProjectionState, + WIKI_PROJECTOR_VERSION, + WikiReservedDocument, + WikiSourceChange, + WikiSourceNote, + affected_wiki_scopes, + plan_wiki_projection, +) + +ACCEPTED_AT = datetime(2026, 8, 29, 18, 30, tzinfo=timezone.utc) + + +def _request( + *, + position: int = 3, + reason: WikiProjectionReason = WikiProjectionReason.accepted_note, + scopes: tuple[str, ...] = ("guides",), +) -> WikiProjectionRequest: + return WikiProjectionRequest( + project_id="project-88", + through_partition_position=position, + projector_version=WIKI_PROJECTOR_VERSION, + reason=reason, + requested_scopes=scopes, + ) + + +def _snapshot( + *, + output_watermark: int = 2, + materialized: bool = True, + reserved_documents: tuple[WikiReservedDocument, ...] = (), +) -> WikiProjectionSnapshot: + return WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=3, + current_output_watermark=output_watermark, + source_accepted_at=ACCEPTED_AT, + notes=( + WikiSourceNote( + path="overview.md", + permalink="overview", + title="Overview", + note_type="Note", + checksum="overview-checksum", + ), + WikiSourceNote( + path="guides/setup.md", + permalink="guides/setup", + title="Setup", + note_type="Guide", + checksum="setup-checksum", + ), + WikiSourceNote( + path="guides/deep/details.md", + permalink="guides/deep/details", + title="Details", + note_type="Guide", + checksum="details-checksum", + ), + ), + changes=( + WikiSourceChange( + partition_position=3, + operation=WikiChangeOperation.updated, + path="guides/setup.md", + permalink="guides/setup", + title="Setup", + accepted_at=ACCEPTED_AT, + materialized=materialized, + source="web", + ), + ), + reserved_documents=reserved_documents, + ) + + +def _reserved(path: str, content: bytes, *, owned: bool = True) -> WikiReservedDocument: + return WikiReservedDocument( + path=path, + checksum=sha256(content).hexdigest(), + content=content, + projector_owned=owned, + ) + + +def test_affected_scopes_include_root_and_move_ancestors() -> None: + assert affected_wiki_scopes("guides/old/setup.md", "reference/new/setup.md") == ( + "", + "guides", + "guides/old", + "reference", + "reference/new", + ) + + +def test_affected_scopes_ignore_missing_paths() -> None: + assert affected_wiki_scopes(None) == ("",) + + +def test_projection_request_rejects_invalid_contract_fields() -> None: + request = _request() + + with pytest.raises(ValueError, match="requires a project_id"): + replace(request, project_id=" ") + with pytest.raises(ValueError, match="cannot be negative"): + replace(request, through_partition_position=-1) + with pytest.raises(ValueError, match="requires projector_version wiki/1.0.0"): + replace(request, projector_version=" ") + with pytest.raises(ValueError, match="requires projector_version wiki/1.0.0"): + replace(request, projector_version="wiki/2.0.0") + + +def test_projection_request_normalizes_and_deduplicates_scopes() -> None: + request = _request(scopes=("guides\\deep", "guides/deep", "")) + + assert request.requested_scopes == ("", "guides/deep") + + +@pytest.mark.parametrize( + "scope", + ( + "bad|scope", + "bad::scope", + "bad\nscope", + "bad\x00scope", + "bad\x01scope", + "bad\x7fscope", + "bad:scope", + "CON", + "guides/trailing.", + " guides", + "guides ", + ), +) +def test_projection_request_rejects_nonportable_scopes(scope: str) -> None: + with pytest.raises(ValueError, match="Wiki (path|scope)"): + _request(scopes=(scope,)) + + +def test_source_note_rejects_missing_metadata() -> None: + note = WikiSourceNote( + path="note.md", + permalink="note", + title="Note", + note_type="Note", + checksum="checksum", + ) + + with pytest.raises(ValueError, match="requires a title"): + replace(note, title=" ") + with pytest.raises(ValueError, match="requires a note_type"): + replace(note, note_type=" ") + with pytest.raises(ValueError, match="requires a checksum"): + replace(note, checksum=" ") + with pytest.raises(ValueError, match="requires a canonical permalink"): + replace(note, permalink=" ") + with pytest.raises(ValueError, match="unsafe canonical permalink"): + replace(note, permalink="bad|target") + + +@pytest.mark.parametrize( + "path", + ( + "bad?/note.md", + "bad\x01/note.md", + "bad\x7f/note.md", + "NUL/note.md", + "trailing./note.md", + " note.md", + "note.md ", + ), +) +def test_source_note_rejects_nonportable_path_components(path: str) -> None: + with pytest.raises(ValueError, match="Wiki (path|note path)"): + WikiSourceNote( + path=path, + permalink="note", + title="Note", + note_type="Note", + checksum="checksum", + ) + + +@pytest.mark.parametrize("path", ("index.md/note.md", "guides/LOG.md/note.md")) +def test_source_note_rejects_reserved_wiki_directory_components(path: str) -> None: + with pytest.raises(ValueError, match="reserved Wiki directory name"): + WikiSourceNote( + path=path, + permalink="note", + title="Note", + note_type="Note", + checksum="checksum", + ) + + +@pytest.mark.parametrize("scope", ("index.md", "guides/LOG.md")) +def test_request_rejects_reserved_wiki_directory_scopes(scope: str) -> None: + with pytest.raises(ValueError, match="reserved Wiki directory name"): + _request(scopes=(scope,)) + + +def test_source_change_rejects_invalid_contract_fields() -> None: + change = WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.updated, + path="note.md", + permalink="note", + title="Note", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ) + + with pytest.raises(ValueError, match="position must be positive"): + replace(change, partition_position=0) + with pytest.raises(ValueError, match="requires a title"): + replace(change, title=" ") + with pytest.raises(ValueError, match="requires a canonical permalink"): + replace(change, permalink=" ") + with pytest.raises(ValueError, match="unsafe canonical permalink"): + replace(change, permalink="bad|target") + with pytest.raises(ValueError, match="timezone-aware"): + replace(change, accepted_at=ACCEPTED_AT.replace(tzinfo=None)) + with pytest.raises(ValueError, match="requires a source"): + replace(change, source=" ") + + +def test_source_change_normalizes_previous_path() -> None: + change = WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.moved, + path="new/note.md", + permalink="new/note", + previous_path="old\\note.md", + title="Note", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ) + + assert change.previous_path == "old/note.md" + + +def test_reserved_document_requires_a_reserved_path_and_checksum() -> None: + with pytest.raises(ValueError, match="non-reserved path"): + _reserved("note.md", b"content") + with pytest.raises(ValueError, match="requires a checksum"): + WikiReservedDocument( + path="index.md", + checksum=" ", + content=b"content", + projector_owned=True, + ) + + +def test_snapshot_rejects_invalid_contract_fields() -> None: + snapshot = _snapshot() + + with pytest.raises(ValueError, match="requires a project_id"): + replace(snapshot, project_id=" ") + with pytest.raises(ValueError, match="requires a project_name"): + replace(snapshot, project_name=" ") + with pytest.raises(ValueError, match="source position cannot be negative"): + replace(snapshot, source_partition_position=-1) + with pytest.raises(ValueError, match="cannot be negative"): + replace(snapshot, current_output_watermark=-1) + with pytest.raises(ValueError, match="timezone-aware"): + replace(snapshot, source_accepted_at=ACCEPTED_AT.replace(tzinfo=None)) + + +def test_snapshot_rejects_duplicate_note_paths_and_change_positions() -> None: + snapshot = _snapshot() + + with pytest.raises(ValueError, match="duplicate source note paths"): + replace(snapshot, notes=(snapshot.notes[0], snapshot.notes[0])) + with pytest.raises(ValueError, match="unique partition positions"): + replace(snapshot, changes=(snapshot.changes[0], snapshot.changes[0])) + + +def test_snapshot_rejects_case_folded_duplicate_reserved_paths() -> None: + lower = _reserved("guides/index.md", b"lower") + upper = _reserved("guides/Index.md", b"upper") + + with pytest.raises(ValueError, match="duplicate reserved document paths"): + replace(_snapshot(), reserved_documents=(lower, upper)) + + +@pytest.mark.parametrize( + ("first_scope", "second_scope"), + (("Foo", "foo"), ("caf\u00e9", "cafe\u0301")), +) +def test_projection_rejects_nonportable_duplicate_scopes( + first_scope: str, + second_scope: str, +) -> None: + snapshot = replace( + _snapshot(), + notes=( + WikiSourceNote( + path=f"{first_scope}/one.md", + permalink=f"{first_scope}/one", + title="One", + note_type="Note", + checksum="one-checksum", + ), + WikiSourceNote( + path=f"{second_scope}/two.md", + permalink=f"{second_scope}/two", + title="Two", + note_type="Note", + checksum="two-checksum", + ), + ), + ) + + with pytest.raises(ValueError, match="unique when compared as portable paths"): + plan_wiki_projection( + _request(reason=WikiProjectionReason.manual_rebuild, scopes=()), + snapshot, + ) + + +def test_snapshot_rejects_unicode_normalized_duplicate_reserved_paths() -> None: + composed = _reserved("caf\u00e9/index.md", b"composed") + decomposed = _reserved("cafe\u0301/index.md", b"decomposed") + + with pytest.raises(ValueError, match="duplicate reserved document paths"): + replace(_snapshot(), reserved_documents=(composed, decomposed)) + + +def test_projection_rejects_scope_that_collides_with_source_note_path() -> None: + with pytest.raises(ValueError, match="collides with an existing source note path"): + plan_wiki_projection(_request(scopes=("overview.md",)), _snapshot()) + + +def test_projection_result_reports_updating_when_output_lags_source() -> None: + result = WikiProjectionResult( + source_watermark=3, + output_watermark=2, + created=0, + updated=0, + unchanged=0, + conflicts=(), + warnings=(), + pending_materialization=(), + ) + + assert result.state == WikiProjectionState.updating + + +def test_projection_rejects_mismatched_project_and_stale_request() -> None: + with pytest.raises(ValueError, match="project_id differ"): + plan_wiki_projection(_request(), replace(_snapshot(), project_id="other")) + with pytest.raises(ValueError, match="older than"): + plan_wiki_projection(_request(position=2), _snapshot(output_watermark=3)) + + +def test_projection_renders_root_and_affected_directory_indexes_and_logs() -> None: + plan = plan_wiki_projection(_request(), _snapshot()) + + assert [write.path for write in plan.writes] == [ + "guides/index.md", + "guides/log.md", + "index.md", + "log.md", + ] + rendered = {write.path: write.content.decode() for write in plan.writes} + assert "[[guides/deep/index|Deep]]" in rendered["guides/index.md"] + assert "[[guides/setup|Setup]]" in rendered["guides/index.md"] + assert "[[guides/index|Guides]]" in rendered["index.md"] + assert "[[overview|Overview]]" in rendered["index.md"] + assert "Updated [[guides/setup|Setup]]" in rendered["guides/log.md"] + assert plan.result.source_watermark == 3 + assert plan.result.output_watermark == 3 + assert plan.result.created == 4 + assert plan.result.state == WikiProjectionState.current + + +def test_projection_bytes_match_the_shared_contract_fixture() -> None: + fixture_path = ( + Path(__file__).parents[1] / "fixtures" / "wiki_projector" / "basic_projection.json" + ) + fixture = json.loads(fixture_path.read_text()) + + plan = plan_wiki_projection(_request(), _snapshot()) + + assert fixture["contract_version"] == plan.request.projector_version + assert fixture["project_id"] == plan.request.project_id + assert fixture["through_partition_position"] == plan.request.through_partition_position + assert fixture["expected_sha256"] == {write.path: write.checksum for write in plan.writes} + + +def test_projection_is_a_byte_identical_noop_at_the_same_watermark() -> None: + first = plan_wiki_projection(_request(), _snapshot()) + existing = tuple(_reserved(write.path, write.content) for write in first.writes) + + replay = plan_wiki_projection( + _request(), + _snapshot(output_watermark=3, reserved_documents=existing), + ) + + assert replay.writes == () + assert replay.unchanged_paths == tuple(write.path for write in first.writes) + assert replay.result.unchanged == 4 + assert replay.result.state == WikiProjectionState.current + + +def test_full_rebuild_records_unchanged_and_updated_reserved_documents() -> None: + request = _request(reason=WikiProjectionReason.manual_rebuild, scopes=()) + first = plan_wiki_projection(request, _snapshot()) + first_by_path = {write.path: write for write in first.writes} + unchanged = first_by_path["index.md"] + stale = _reserved("log.md", b"stale\n") + snapshot = _snapshot( + reserved_documents=( + _reserved(unchanged.path, unchanged.content), + stale, + ) + ) + + replay = plan_wiki_projection(request, snapshot) + + assert replay.unchanged_paths == ("index.md",) + assert replay.result.unchanged == 1 + assert replay.result.updated == 1 + updated_log = next(write for write in replay.writes if write.path == "log.md") + assert updated_log.expected_checksum == stale.checksum + + +def test_pending_materialization_defers_all_bytes_without_advancing_output() -> None: + plan = plan_wiki_projection(_request(), _snapshot(materialized=False)) + + assert plan.writes == () + assert plan.result.output_watermark == 2 + assert plan.result.pending_materialization == (3,) + assert plan.result.state == WikiProjectionState.partial + + +def test_user_claimed_reserved_path_is_a_conflict_not_a_write() -> None: + claimed = _reserved("guides/index.md", b"# User index\n", owned=False) + + plan = plan_wiki_projection( + _request(), + _snapshot(reserved_documents=(claimed,)), + ) + + assert plan.writes == () + assert plan.result.conflicts[0].path == "guides/index.md" + assert plan.result.output_watermark == 2 + assert plan.result.state == WikiProjectionState.conflicted + + +def test_user_claimed_reserved_path_matches_case_insensitively() -> None: + claimed = _reserved("guides/Index.md", b"# User index\n", owned=False) + + plan = plan_wiki_projection( + _request(), + _snapshot(reserved_documents=(claimed,)), + ) + + assert plan.writes == () + assert plan.result.conflicts[0].path == "guides/index.md" + assert plan.result.output_watermark == 2 + assert plan.result.state == WikiProjectionState.conflicted + + +def test_projector_only_advance_preserves_complete_projection_bytes() -> None: + initial_snapshot = _snapshot() + initial = plan_wiki_projection( + _request(reason=WikiProjectionReason.manual_rebuild, scopes=()), + initial_snapshot, + ) + projector_change = WikiSourceChange( + partition_position=4, + operation=WikiChangeOperation.updated, + path="index.md", + permalink="index", + title="Project 88", + accepted_at=datetime(2026, 8, 29, 18, 31, tzinfo=timezone.utc), + materialized=True, + source="wiki_projector", + ) + snapshot = replace( + initial_snapshot, + source_partition_position=4, + current_output_watermark=3, + source_accepted_at=projector_change.accepted_at, + changes=(*initial_snapshot.changes, projector_change), + reserved_documents=tuple(_reserved(write.path, write.content) for write in initial.writes), + ) + + plan = plan_wiki_projection(_request(position=4, scopes=()), snapshot) + + assert plan.writes == () + assert plan.unchanged_paths == tuple(write.path for write in initial.writes) + assert plan.result.output_watermark == 4 + assert plan.result.state == WikiProjectionState.current + + +def test_projector_only_advance_repairs_missing_projection_document() -> None: + initial_snapshot = _snapshot() + initial = plan_wiki_projection( + _request(reason=WikiProjectionReason.manual_rebuild, scopes=()), + initial_snapshot, + ) + projector_change = WikiSourceChange( + partition_position=4, + operation=WikiChangeOperation.updated, + path="index.md", + permalink="index", + title="Project 88", + accepted_at=ACCEPTED_AT, + materialized=True, + source="wiki_projector", + ) + snapshot = replace( + initial_snapshot, + source_partition_position=4, + current_output_watermark=3, + changes=(*initial_snapshot.changes, projector_change), + reserved_documents=tuple( + _reserved(write.path, write.content) + for write in initial.writes + if write.path != "guides/deep/log.md" + ), + ) + + plan = plan_wiki_projection(_request(position=4, scopes=()), snapshot) + + assert tuple(write.path for write in plan.writes) == ("guides/deep/log.md",) + assert "Updated [[index|Project 88]]" not in plan.writes[0].content.decode() + + +def test_requested_scopes_cannot_omit_a_changed_note_scope() -> None: + changed_note = WikiSourceNote( + path="secret/note.md", + permalink="secret/note", + title="Secret note", + note_type="Note", + checksum="secret-checksum", + ) + changed = WikiSourceChange( + partition_position=3, + operation=WikiChangeOperation.updated, + path=changed_note.path, + permalink=changed_note.permalink, + title=changed_note.title, + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ) + snapshot = replace( + _snapshot(), + notes=(*_snapshot().notes, changed_note), + changes=(changed,), + ) + + plan = plan_wiki_projection(_request(scopes=("guides",)), snapshot) + + assert {write.path for write in plan.writes} >= { + "secret/index.md", + "secret/log.md", + } + assert plan.result.output_watermark == 3 + + +def test_full_rebuild_covers_every_note_directory() -> None: + request = _request( + reason=WikiProjectionReason.import_rebuild, + scopes=(), + ) + + plan = plan_wiki_projection(request, _snapshot()) + + assert {write.path for write in plan.writes} == { + "index.md", + "log.md", + "guides/index.md", + "guides/log.md", + "guides/deep/index.md", + "guides/deep/log.md", + } + + +def test_full_rebuild_covers_orphaned_reserved_document_scopes() -> None: + request = _request( + reason=WikiProjectionReason.manual_rebuild, + scopes=(), + ) + orphaned_index = _reserved("orphaned/index.md", b"# Stale index\n") + orphaned_log = _reserved("orphaned/log.md", b"# Stale log\n") + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=3, + current_output_watermark=2, + source_accepted_at=ACCEPTED_AT, + notes=(), + changes=( + WikiSourceChange( + partition_position=3, + operation=WikiChangeOperation.deleted, + path="orphaned/last-note.md", + permalink="orphaned/last-note", + title="Last note", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + reserved_documents=(orphaned_index, orphaned_log), + ) + + plan = plan_wiki_projection(request, snapshot) + + assert {write.path for write in plan.writes} == { + "index.md", + "log.md", + "orphaned/index.md", + "orphaned/log.md", + } + rendered = {write.path: write.content.decode() for write in plan.writes} + assert "No concepts have been projected" in rendered["orphaned/index.md"] + assert "Deleted `orphaned/last-note.md`" in rendered["orphaned/log.md"] + + +def test_projection_rejects_a_snapshot_ahead_of_the_requested_watermark() -> None: + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=2, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=(), + changes=( + WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.deleted, + path="included/note.md", + permalink="included/note", + title="Included", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + WikiSourceChange( + partition_position=2, + operation=WikiChangeOperation.deleted, + path="future/note.md", + permalink="future/note", + title="Future", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + ) + + with pytest.raises(ValueError, match="exact as-of source snapshot"): + plan_wiki_projection( + _request( + position=1, + reason=WikiProjectionReason.manual_rebuild, + scopes=(), + ), + snapshot, + ) + + +def test_moved_change_requires_previous_path() -> None: + with pytest.raises(ValueError, match="requires previous_path"): + plan_wiki_projection( + _request(), + WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=3, + current_output_watermark=2, + source_accepted_at=ACCEPTED_AT, + notes=(), + changes=( + WikiSourceChange( + partition_position=3, + operation=WikiChangeOperation.moved, + path="guides/new.md", + permalink="guides/new", + title="Moved", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + ), + ) + + +def test_created_and_moved_changes_render_in_the_log() -> None: + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=2, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=(), + changes=( + WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.created, + path="created.md", + permalink="created", + title="Created", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + WikiSourceChange( + partition_position=2, + operation=WikiChangeOperation.moved, + path="moved.md", + permalink="moved", + previous_path="old.md", + title="Moved", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + ) + + plan = plan_wiki_projection(_request(position=2, scopes=()), snapshot) + log = next(write.content.decode() for write in plan.writes if write.path == "log.md") + + assert "Created [[created|Created]]" in log + assert "Moved `old.md` to [[moved|Moved]]" in log + + +def test_log_preserves_ampersands_in_code_formatted_paths() -> None: + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=2, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=(), + changes=( + WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.moved, + path="new.md", + permalink="new", + previous_path="old&draft.md", + title="Moved", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + WikiSourceChange( + partition_position=2, + operation=WikiChangeOperation.deleted, + path="retired&archived.md", + permalink="retired-archived", + title="Deleted", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + ) + + plan = plan_wiki_projection(_request(position=2, scopes=()), snapshot) + log = next(write.content.decode() for write in plan.writes if write.path == "log.md") + + assert "Moved `old&draft.md` to [[new|Moved]]" in log + assert "Deleted `retired&archived.md`" in log + assert "&" not in log + + +def test_absolute_paths_are_rejected_at_the_contract_boundary() -> None: + with pytest.raises(ValueError, match="project-relative"): + WikiSourceNote( + path="/outside.md", + permalink="outside", + title="Outside", + note_type="Note", + checksum="checksum", + ) + + +@pytest.mark.parametrize("path", ("C:/outside.md", "C:\\outside.md")) +def test_windows_drive_paths_are_rejected_at_the_contract_boundary(path: str) -> None: + with pytest.raises(ValueError, match="project-relative"): + WikiSourceNote( + path=path, + permalink="outside", + title="Outside", + note_type="Note", + checksum="checksum", + ) + + +@pytest.mark.parametrize("path", ("notes//foo.md", "notes/./foo.md", "note.md/")) +def test_noncanonical_paths_are_rejected_at_the_contract_boundary(path: str) -> None: + with pytest.raises(ValueError, match="project-relative"): + WikiSourceNote( + path=path, + permalink="noncanonical", + title="Noncanonical", + note_type="Note", + checksum="checksum", + ) + + +@pytest.mark.parametrize( + "path", + ( + "notes/closing].md", + "notes/opening[.md", + "notes/alias|target.md", + "notes/code`span.md", + "notes/html None: + with pytest.raises(ValueError, match="unsupported Markdown delimiters"): + WikiSourceNote( + path=path, + permalink="unsupported", + title="Unsupported", + note_type="Note", + checksum="checksum", + ) + + +@pytest.mark.parametrize("path", ("", "note.txt")) +def test_non_markdown_note_paths_are_rejected(path: str) -> None: + with pytest.raises(ValueError, match="project-relative Markdown"): + WikiSourceNote( + path=path, + permalink="unsupported", + title="Unsupported", + note_type="Note", + checksum="checksum", + ) + + +def test_parent_segments_are_rejected_at_the_contract_boundary() -> None: + with pytest.raises(ValueError, match="project-relative and normalized"): + WikiSourceNote( + path="notes/../outside.md", + permalink="outside", + title="Outside", + note_type="Note", + checksum="checksum", + ) + + +def test_projection_order_is_deterministic_for_case_only_names() -> None: + notes = ( + WikiSourceNote( + path="foo.md", + permalink="foo", + title="same", + note_type="Note", + checksum="lower", + ), + WikiSourceNote( + path="Foo.md", + permalink="foo-1", + title="Same", + note_type="Note", + checksum="upper", + ), + ) + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=0, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=notes, + changes=(), + ) + reverse_snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=0, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=tuple(reversed(notes)), + changes=(), + ) + + first = plan_wiki_projection(_request(position=0, scopes=()), snapshot) + second = plan_wiki_projection(_request(position=0, scopes=()), reverse_snapshot) + + assert first.writes == second.writes + + +def test_projection_links_notes_by_their_canonical_permalinks() -> None: + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=0, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=( + WikiSourceNote( + path="foo bar.md", + permalink="foo-bar", + title="Spaced", + note_type="Note", + checksum="spaced", + ), + WikiSourceNote( + path="foo-bar.md", + permalink="foo-bar-1", + title="Hyphenated", + note_type="Note", + checksum="hyphenated", + ), + ), + changes=(), + ) + + plan = plan_wiki_projection(_request(position=0, scopes=()), snapshot) + index = next(write.content.decode() for write in plan.writes if write.path == "index.md") + + assert "[[foo-bar|Spaced]]" in index + assert "[[foo-bar-1|Hyphenated]]" in index + + +def test_projection_logs_preserve_each_changes_historical_permalink() -> None: + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=1, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=( + WikiSourceNote( + path="same.md", + permalink="new-note", + title="New note", + note_type="Note", + checksum="new-note", + ), + ), + changes=( + WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.created, + path="same.md", + permalink="old-note", + title="Old note", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + ) + + plan = plan_wiki_projection(_request(position=1, scopes=()), snapshot) + log = next(write.content.decode() for write in plan.writes if write.path == "log.md") + + assert "Created [[old-note|Old note]]" in log + assert "[[new-note|Old note]]" not in log + + +def test_projection_rejects_source_permalink_reserved_for_generated_index() -> None: + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=0, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=( + WikiSourceNote( + path="guides/topic.md", + permalink="guides/index", + title="Topic", + note_type="Note", + checksum="topic", + ), + ), + changes=(), + ) + + with pytest.raises(ValueError, match="generated document identity"): + plan_wiki_projection(_request(position=0, scopes=()), snapshot) + + +def test_projection_rejects_historical_permalink_reserved_for_generated_index() -> None: + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=2, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=(), + changes=( + WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.created, + path="guides/topic.md", + permalink="guides/index", + title="Deleted topic", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + WikiSourceChange( + partition_position=2, + operation=WikiChangeOperation.deleted, + path="guides/topic.md", + permalink="guides/index", + title="Deleted topic", + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + ) + + with pytest.raises(ValueError, match="generated document identity"): + plan_wiki_projection(_request(position=2, scopes=()), snapshot) + + +def test_projection_escapes_dynamic_markdown_structure() -> None: + injected_title = "Bad]]\n- relates_to [[evil" + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name=f"Project\n{injected_title}", + source_partition_position=1, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=( + WikiSourceNote( + path="safe-target.md", + permalink="safe-target", + title=injected_title, + note_type="Note", + checksum="unsafe", + ), + ), + changes=( + WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.updated, + path="safe-target.md", + permalink="safe-target", + title=injected_title, + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + ) + + plan = plan_wiki_projection(_request(position=1, scopes=()), snapshot) + rendered = {write.path: write.content.decode() for write in plan.writes} + + assert "\n- relates_to [[evil" not in rendered["index.md"] + assert "\n- relates_to [[evil" not in rendered["log.md"] + assert "[[safe-target|Bad]] - relates_to [[evil]]" in rendered["index.md"] + + +@pytest.mark.parametrize( + ("title", "escaped_title"), + ( + ("A & B", "A &amp; B"), + ("A ] B", "A &#93; B"), + ), +) +def test_projection_preserves_literal_entity_looking_titles( + title: str, + escaped_title: str, +) -> None: + snapshot = WikiProjectionSnapshot( + project_id="project-88", + project_name="Project 88", + source_partition_position=1, + current_output_watermark=0, + source_accepted_at=ACCEPTED_AT, + notes=( + WikiSourceNote( + path="entity-title.md", + permalink="entity-title", + title=title, + note_type="Note", + checksum="entity-title", + ), + ), + changes=( + WikiSourceChange( + partition_position=1, + operation=WikiChangeOperation.created, + path="entity-title.md", + permalink="entity-title", + title=title, + accepted_at=ACCEPTED_AT, + materialized=True, + source="web", + ), + ), + ) + + plan = plan_wiki_projection(_request(position=1, scopes=()), snapshot) + rendered = {write.path: write.content.decode() for write in plan.writes} + + assert f"[[entity-title|{escaped_title}]]" in rendered["index.md"]