feat(union): overlay-aware resolution (WP-0008 T5)

resolve() layers open overlays onto canonical pages (overlay_state=DRAFT always
surfaced; overlaid body projected when policy.show_drafts); draft-only edits make
a not-yet-existing page resolvable. Never hides an unapplied overlay (I-4). Policy
gains show_drafts. 4 tests green, pyflakes clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-15 13:29:06 +02:00
parent 7d00ae758e
commit 4be2f190a0
4 changed files with 96 additions and 7 deletions

View File

@@ -31,6 +31,9 @@ class Policy:
canonical_source: CanonicalSource = CanonicalSource.CHORUS
designated_shard: str | None = None
# Whether an unapplied overlay's body is projected over its canonical page on read. Either
# way the overlay is never *hidden* — overlay_state is always surfaced in provenance.
show_drafts: bool = True
DEFAULT_POLICY = Policy()

View File

@@ -17,9 +17,10 @@ from dataclasses import dataclass
from enum import Enum
from shard_wiki.adapters import ShardAdapter
from shard_wiki.coordination import DecisionLog
from shard_wiki.model import Page
from shard_wiki.coordination import DecisionLog, Overlay
from shard_wiki.model import Identity, Page
from shard_wiki.policy import DEFAULT_POLICY, CanonicalSource, Policy
from shard_wiki.provenance import OverlayState, ProvenanceEnvelope, Staleness
__all__ = ["ResolutionKind", "Resolution", "UnionGraph"]
@@ -77,21 +78,28 @@ class UnionGraph:
return pages
def resolve(self, name: str) -> Resolution:
# 1. alias redirect (coordination-canonical, via the log fold)
state = self.log.fold(self.space)
overlays = {
Overlay.from_payload(p).target: Overlay.from_payload(p)
for p in state.open_overlays.values()
}
# 1. alias redirect (coordination-canonical, via the log fold)
target = state.resolve_alias(name)
if target is not None and ":" in target:
shard_id, _, key = target.partition(":")
shard = self._shard(shard_id)
if shard is not None:
try:
page = shard.read(key)
page = self._with_overlay(shard.read(key), overlays)
return Resolution(name, ResolutionKind.SINGLE, (page,))
except KeyError:
pass # alias dangles → fall through to normal resolution
# 2/3. union lookup by key across shards
pages = self._read_all(name)
# 2/3. union lookup by key across shards, then layer open overlays
pages = [self._with_overlay(p, overlays) for p in self._read_all(name)]
pages.extend(self._draft_only_pages(name, pages, overlays))
if not pages:
return Resolution(name, ResolutionKind.RED_LINK)
if len(pages) == 1:
@@ -102,6 +110,34 @@ class UnionGraph:
marked = tuple(_with_divergence(p, ordered) for p in ordered)
return Resolution(name, ResolutionKind.CHORUS, marked)
def _with_overlay(self, page: Page, overlays: dict[Identity, Overlay]) -> Page:
"""Mark a canonical page that has an open overlay (overlay_state DRAFT; project the
overlaid body where policy shows drafts) — never hides the overlay (I-4)."""
overlay = overlays.get(page.identity)
if overlay is None:
return page
env = dataclasses.replace(page.envelope, overlay_state=OverlayState.DRAFT)
body = overlay.body if self.policy.show_drafts else page.body
return dataclasses.replace(page, body=body, envelope=env)
def _draft_only_pages(
self, name: str, existing: list[Page], overlays: dict[Identity, Overlay]
) -> list[Page]:
"""Drafts that create a not-yet-existing page on an attached shard become resolvable."""
have = {p.identity for p in existing}
out: list[Page] = []
for identity, overlay in overlays.items():
if identity.key != name or identity in have or self._shard(identity.shard) is None:
continue
env = ProvenanceEnvelope(
source_shard=identity.shard,
staleness=Staleness.FRESH,
overlay_state=OverlayState.DRAFT,
)
body = overlay.body if self.policy.show_drafts else ""
out.append(Page(identity=identity, body=body, envelope=env))
return out
def _order_for_policy(self, pages: list[Page]) -> list[Page]:
if (
self.policy.canonical_source is CanonicalSource.DESIGNATED_CANONICAL