feat(policy,union): policy leaf + UnionGraph resolution with chorus (WP-0007 T6)

policy/ leaf (CanonicalSource presets, default chorus). union/ UnionGraph:
identity-keyed resolve (alias-redirect via log fold → union lookup → chorus →
red-link); chorus records divergent peers in each page's provenance envelope
(union without erasure); designated-canonical orders the pick. Imports down only.
6 tests green. (blueprint §8.4, ADR-01)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/shard_wiki/policy/__init__.py

@@ -0,0 +1,36 @@
+"""policy/ — the configurable policy surface, a dependency-free leaf (blueprint §10, §11).
+
+Mechanism over policy (I-7): core mechanisms read policy *choices* from here; they never
+hard-code one. This leaf holds only the presets + a pure ``resolve``-style selector. Mechanism
+(how a choice is honoured) lives in ``coordination``/``union``/``projection``, never here.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+
+__all__ = ["CanonicalSource", "Policy", "DEFAULT_POLICY"]
+
+
+class CanonicalSource(Enum):
+    """Resolution policy over a divergent/equivalent set (FederationArchitecture T9). Detection
+    is core; this is only the *resolution* choice."""
+
+    CHORUS = "chorus"  # default: present all versions, none privileged
+    DESIGNATED_CANONICAL = "designated-canonical"
+    GIT_MERGE = "git-merge"
+    VOTE = "vote"
+    OVERLAY_ONLY = "overlay-only"
+
+
+@dataclass(frozen=True, slots=True)
+class Policy:
+    """A space's policy choices. Extended as the policy surface grows (freshness, compaction,
+    execution, tenant mapping — blueprint §10); the slice needs canonical-source + designation."""
+
+    canonical_source: CanonicalSource = CanonicalSource.CHORUS
+    designated_shard: str | None = None
+
+
+DEFAULT_POLICY = Policy()

src/shard_wiki/union/__init__.py

@@ -0,0 +1,8 @@
+"""union/ — the derived-tier union (resolution, equivalence, projection read path).
+
+Disposable/recomputable (I-2); imports down only and is imported by nothing.
+"""
+
+from shard_wiki.union.resolver import Resolution, ResolutionKind, UnionGraph
+
+__all__ = ["UnionGraph", "Resolution", "ResolutionKind"]

src/shard_wiki/union/resolver.py

@@ -0,0 +1,119 @@
+"""Union resolution — the derived-tier read path (CoreArchitectureBlueprint §8.4, ADR-01).
+
+A minimal :class:`UnionGraph` over ≥1 attached shard plus the decision-log fold. ``resolve``
+keys on **identity** (FederationRequirements ADR-01): alias redirect → exact union lookup →
+equivalence chorus → red-link. Ambiguity returns a **chorus set** (union without erasure, I-4):
+divergent peers are recorded in each page's provenance envelope rather than silently dropped.
+
+This is derived/disposable: it reads shards + the log fold; it stores nothing canonical. Per the
+dependency rule it may import down (model/adapters/coordination/policy/provenance) and is
+imported by nothing.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+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.policy import DEFAULT_POLICY, CanonicalSource, Policy
+
+__all__ = ["ResolutionKind", "Resolution", "UnionGraph"]
+
+
+class ResolutionKind(Enum):
+    SINGLE = "single"
+    CHORUS = "chorus"
+    RED_LINK = "red-link"
+
+
+@dataclass(frozen=True, slots=True)
+class Resolution:
+    name: str
+    kind: ResolutionKind
+    pages: tuple[Page, ...] = ()
+
+    @property
+    def is_red_link(self) -> bool:
+        return self.kind is ResolutionKind.RED_LINK
+
+    def single(self) -> Page:
+        """The one page (SINGLE), or the canonical pick of a CHORUS (first), else KeyError."""
+        if not self.pages:
+            raise KeyError(self.name)
+        return self.pages[0]
+
+
+class UnionGraph:
+    """Composes attached shards + the coordination-log fold into a resolvable union."""
+
+    def __init__(
+        self,
+        space: str,
+        log: DecisionLog | None = None,
+        policy: Policy = DEFAULT_POLICY,
+    ) -> None:
+        self.space = space
+        self.log = log or DecisionLog()
+        self.policy = policy
+        self._shards: list[ShardAdapter] = []
+
+    def attach(self, adapter: ShardAdapter) -> None:
+        self._shards.append(adapter)
+
+    def _shard(self, shard_id: str) -> ShardAdapter | None:
+        return next((s for s in self._shards if s.shard_id == shard_id), None)
+
+    def _read_all(self, key: str) -> list[Page]:
+        pages: list[Page] = []
+        for shard in self._shards:
+            try:
+                pages.append(shard.read(key))
+            except KeyError:
+                continue
+        return pages
+
+    def resolve(self, name: str) -> Resolution:
+        # 1. alias redirect (coordination-canonical, via the log fold)
+        state = self.log.fold(self.space)
+        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)
+                    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)
+        if not pages:
+            return Resolution(name, ResolutionKind.RED_LINK)
+        if len(pages) == 1:
+            return Resolution(name, ResolutionKind.SINGLE, (pages[0],))
+
+        # ambiguity → chorus, with divergence recorded (never erased, I-4)
+        ordered = self._order_for_policy(pages)
+        marked = tuple(_with_divergence(p, ordered) for p in ordered)
+        return Resolution(name, ResolutionKind.CHORUS, marked)
+
+    def _order_for_policy(self, pages: list[Page]) -> list[Page]:
+        if (
+            self.policy.canonical_source is CanonicalSource.DESIGNATED_CANONICAL
+            and self.policy.designated_shard is not None
+        ):
+            pages = sorted(
+                pages, key=lambda p: p.identity.shard != self.policy.designated_shard
+            )
+        return pages
+
+
+def _with_divergence(page: Page, group: list[Page]) -> Page:
+    peers = tuple(str(p.identity) for p in group if p.identity != page.identity)
+    new_env = dataclasses.replace(page.envelope, divergence=peers)
+    return dataclasses.replace(page, envelope=new_env)