feat(provenance): layered ProvenanceEnvelope + effective() leaf (WP-0007 T1)

Dependency-free leaf rail: ProvenanceEnvelope + SpanProvenanceDelta + effective()
(page envelope ⊕ span delta; zero-cost inheritance when uniform). Liveness/
Staleness/OverlayState enums. 6 tests green. (blueprint §7.3, §11)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-15 08:55:35 +02:00
parent 3b4f10a349
commit aca9bf30f9
3 changed files with 198 additions and 1 deletions

View File

@@ -0,0 +1,116 @@
"""Provenance — the cross-cutting leaf rail (CoreArchitectureBlueprint §7.3, §11).
Every artifact in the union carries a :class:`ProvenanceEnvelope`. To keep per-span cost near
zero when provenance is uniform, envelopes are **layered**: a page-level envelope plus optional
per-span deltas. The *effective* provenance of a span is ``page_envelope ⊕ span_delta`` — a span
with no delta inherits the page envelope at zero cost ("effective-vs-own", the same pattern the
page model uses for computed metadata).
This module is a **dependency-free leaf**: it imports nothing else in the package and contains
only stable data types plus the pure ``effective`` composition. Mechanism never lives here.
"""
from __future__ import annotations
import dataclasses
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
__all__ = [
"Liveness",
"Staleness",
"OverlayState",
"ProvenanceEnvelope",
"SpanProvenanceDelta",
"effective",
]
class Liveness(Enum):
"""Where a view sits on the live↔snapshot axis (blueprint §8, T18)."""
STATIC = "static"
CAPTURED_SNAPSHOT = "captured-snapshot"
LIVE_OVER_FILES = "live-over-files"
VIEW_TIME = "view-time"
IRREDUCIBLY_LIVE = "irreducibly-live"
class Staleness(Enum):
"""Freshness state of cached/projected content (blueprint §8.8). ``UNAVAILABLE`` is the
dead-shard state (FederationRequirements ADR-04)."""
LIVE = "live"
FRESH = "fresh"
STALE = "stale"
UNAVAILABLE = "unavailable"
class OverlayState(Enum):
"""Overlay lifecycle state of an artifact (FederationRequirements ADR-05)."""
NONE = "none"
DRAFT = "draft"
PATCH_PENDING = "patch-pending"
APPLIED = "applied"
@dataclass(frozen=True, slots=True)
class ProvenanceEnvelope:
"""The metadata every page/span carries — union without erasure (INTENT I-4).
This is the *page-level* (or fully-resolved *effective*) form. Per-span overrides are
expressed as a :class:`SpanProvenanceDelta` and composed via :func:`effective`.
"""
source_shard: str
liveness: Liveness = Liveness.STATIC
staleness: Staleness = Staleness.FRESH
overlay_state: OverlayState = OverlayState.NONE
source_rev: str | None = None
observed_at: datetime | None = None
authz_context: str | None = None
divergence: tuple[str, ...] = ()
lineage: str | None = None
@dataclass(frozen=True, slots=True)
class SpanProvenanceDelta:
"""A per-span override of a page envelope. Every field is optional; ``None`` (or, for
``divergence``, the sentinel default) means *inherit the page-level value*. A span with an
all-default delta — or no delta at all — costs nothing and resolves to the page envelope.
"""
source_shard: str | None = None
liveness: Liveness | None = None
staleness: Staleness | None = None
overlay_state: OverlayState | None = None
source_rev: str | None = None
observed_at: datetime | None = None
authz_context: str | None = None
# Sentinel: ``None`` means inherit; an explicit (possibly empty) tuple overrides.
divergence: tuple[str, ...] | None = None
lineage: str | None = None
def is_empty(self) -> bool:
"""True when the delta overrides nothing (the common, zero-cost case)."""
return all(getattr(self, f.name) is None for f in dataclasses.fields(self))
def effective(
base: ProvenanceEnvelope, delta: SpanProvenanceDelta | None = None
) -> ProvenanceEnvelope:
"""Compose a page envelope with an optional span delta → the span's effective provenance.
``effective(base, None)`` and ``effective(base, <empty delta>)`` both return ``base``
unchanged (zero-cost inheritance). Any non-``None`` delta field overrides ``base``.
"""
if delta is None or delta.is_empty():
return base
overrides = {
f.name: value
for f in dataclasses.fields(delta)
if (value := getattr(delta, f.name)) is not None
}
return dataclasses.replace(base, **overrides)