feat(coordination): Overlay model + OverlayEngine.draft (WP-0008 T2)

Overlay value type (id, target, base_rev, body, state) recorded as an
OVERLAY_CREATED decision-log event (coordination-canonical); get()/open_overlays()
reconstruct from the fold. 4 tests green, pyflakes clean. (ADR-05, blueprint §8.2)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-15 11:56:53 +02:00
parent 92d5774baf
commit d797bc5ee4
4 changed files with 129 additions and 2 deletions

View File

@@ -6,5 +6,13 @@ from shard_wiki.coordination.decision_log import (
DecisionLog,
EventType,
)
from shard_wiki.coordination.overlay import Overlay, OverlayEngine
__all__ = ["DecisionLog", "DecisionEvent", "EventType", "CoordinationState"]
__all__ = [
"DecisionLog",
"DecisionEvent",
"EventType",
"CoordinationState",
"Overlay",
"OverlayEngine",
]

View File

@@ -0,0 +1,77 @@
"""Overlay engine — overlay-before-mutation (FederationRequirements ADR-05, blueprint §8.2).
An overlay is a non-destructive local edit against a page. It is **coordination-canonical**: an
``OVERLAY_CREATED`` event in the decision log (§8.1), not a mutable side file. The current set
of open overlays is the log fold. ``draft`` records one; ``apply`` (T4) resolves it under drift.
"""
from __future__ import annotations
import uuid
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
from shard_wiki.coordination.decision_log import DecisionLog, EventType
from shard_wiki.model import Identity
from shard_wiki.provenance import OverlayState
__all__ = ["Overlay", "OverlayEngine"]
@dataclass(frozen=True, slots=True)
class Overlay:
"""A non-destructive edit: the proposed ``body`` for ``target``, authored against
``base_rev`` (the shard revision seen at draft time, for drift detection)."""
overlay_id: str
target: Identity
base_rev: str | None
body: str
state: OverlayState = OverlayState.DRAFT
def to_payload(self) -> dict[str, Any]:
return {
"overlay_id": self.overlay_id,
"target_shard": self.target.shard,
"target_key": self.target.key,
"base_rev": self.base_rev,
"body": self.body,
"state": self.state.value,
}
@classmethod
def from_payload(cls, payload: Mapping[str, Any]) -> Overlay:
return cls(
overlay_id=payload["overlay_id"],
target=Identity(payload["target_shard"], payload["target_key"]),
base_rev=payload["base_rev"],
body=payload["body"],
state=OverlayState(payload.get("state", OverlayState.DRAFT.value)),
)
class OverlayEngine:
def __init__(self, space: str, log: DecisionLog) -> None:
self.space = space
self.log = log
def draft(
self,
target: Identity,
body: str,
base_rev: str | None,
actor: str | None = None,
) -> Overlay:
"""Create a draft overlay and record it in the decision log (coordination-canonical)."""
overlay = Overlay(uuid.uuid4().hex, target, base_rev, body)
self.log.append(self.space, EventType.OVERLAY_CREATED, overlay.to_payload(), actor=actor)
return overlay
def get(self, overlay_id: str) -> Overlay | None:
payload = self.log.fold(self.space).open_overlays.get(overlay_id)
return Overlay.from_payload(payload) if payload is not None else None
def open_overlays(self) -> tuple[Overlay, ...]:
state = self.log.fold(self.space)
return tuple(Overlay.from_payload(p) for p in state.open_overlays.values())