shard-wiki/src/shard_wiki/space.py

"""InformationSpace — the thin orchestrator entry tying the slice together (blueprint §3 L6).

A root entity / information space: shards attach to it (conformance-gated, TSD §A.2), a
coordination decision log records its canonical decisions, and a derived union resolves names to
pages. This is the L6 consumer surface for the foundation slice (attach → resolve → read);
a network API is a later workplan.
"""

from __future__ import annotations

from shard_wiki.adapters import ShardAdapter, assert_conformant
from shard_wiki.coordination import (
    ApplyResult,
    DecisionLog,
    EventType,
    Overlay,
    OverlayEngine,
)
from shard_wiki.model import Page
from shard_wiki.policy import DEFAULT_POLICY, Policy
from shard_wiki.union import Resolution, UnionGraph

__all__ = ["InformationSpace"]


class InformationSpace:
    def __init__(self, space_id: str, policy: Policy = DEFAULT_POLICY) -> None:
        self.space_id = space_id
        self.log = DecisionLog()
        self.union = UnionGraph(space_id, log=self.log, policy=policy)
        self.overlays = OverlayEngine(space_id, self.log)

    def attach(self, adapter: ShardAdapter) -> None:
        """Attach a shard — only if it passes conformance (verified profile, I-3/§6.6)."""
        assert_conformant(adapter)
        self.union.attach(adapter)

    def alias(self, name: str, target: str, actor: str | None = None) -> None:
        """Record a coordination-canonical alias (``name`` → ``"shard:key"``) in the log."""
        self.log.append(
            self.space_id, EventType.ALIAS_SET, {"alias": name, "target": target}, actor=actor
        )

    def resolve(self, name: str) -> Resolution:
        return self.union.resolve(name)

    def read(self, name: str) -> Page:
        """Resolve and return the page (or the canonical pick of a chorus). KeyError if red-link."""
        return self.union.resolve(name).single()

    def overlay(self, name: str, body: str, actor: str | None = None) -> Overlay:
        """Draft a non-destructive overlay against the resolved page (overlay-before-mutation)."""
        page = self.read(name)
        return self.overlays.draft(page.identity, body, page.envelope.source_rev, actor=actor)

    def apply_overlay(self, overlay_id: str) -> ApplyResult:
        """Apply a draft overlay to its target shard (apply-under-drift, §8.6)."""
        overlay = self.overlays.get(overlay_id)
        if overlay is None:
            raise KeyError(overlay_id)
        adapter = self.union.shard(overlay.target.shard)
        if adapter is None:
            raise KeyError(overlay.target.shard)
        return self.overlays.apply(overlay_id, adapter)

    def edit(self, name: str, body: str, actor: str | None = None) -> ApplyResult:
        """Edit a page through the one principled path: draft an overlay, then apply it. A
        write-through-capable target fast-forwards (write-through); a read-only target keeps the
        draft as local truth (I-5: overlay before mutation, always)."""
        overlay = self.overlay(name, body, actor=actor)
        return self.apply_overlay(overlay.overlay_id)