# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-downstream-context-invalidation-when-memory-forget-semantics-are-non-atomic-across-a-propagation-frontier.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-downstream-context-invalidation-when-memory-forget-semantics-are-non-atomic-across-a-propagation-frontier.warrant.json
# Markdown SHA-256:   208ef2669e129b0cf09751e06932f47238ae64711dc29417e206e4cdddb79f61
# Code AST SHA-256:   1c384aac964af9ca9a39eccb5790f7bfca1efda6ac104b7c7193cad3d61b03fb
# Verified Classes:   StaleDerivationError, Artifact, MemoryStore
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Dict, Optional


class StaleDerivationError(RuntimeError):
    """Raised when a consumer acts on a derived artifact older than a forget."""


@dataclass
class Artifact:
    payload: str
    source_versions: Dict[str, int] = field(default_factory=dict)


class MemoryStore:
    def __init__(self) -> None:
        self._version: int = 0
        self._records: Dict[str, int] = {}
        self._forgets: Dict[str, int] = {}

    def write(self, key: str) -> int:
        self._version += 1
        self._records[key] = self._version
        return self._version

    def get_with_version(self, key: str) -> tuple[str, int]:
        if key not in self._records:
            raise KeyError(key)
        return (f"value-of-{key}", self._records[key])

    def forget(self, key: str) -> int:
        self._version += 1
        self._forgets[key] = self._version
        self._records.pop(key, None)
        return self._version

    def forget_version(self, key: str) -> Optional[int]:
        return self._forgets.get(key)


def derive(store: MemoryStore, key: str) -> Artifact:
    value, version = store.get_with_version(key)
    return Artifact(payload=f"summary({value})", source_versions={key: version})


def assert_fresh(store: MemoryStore, artifact: Artifact) -> None:
    for key, src_version in artifact.source_versions.items():
        fv = store.forget_version(key)
        if fv is not None and src_version < fv:
            raise StaleDerivationError(
                f"artifact derived from {key}@v{src_version} is stale; "
                f"forget committed at v{fv}"
            )


def consume(store: MemoryStore, artifact: Artifact) -> str:
    assert_fresh(store, artifact)
    return artifact.payload


if __name__ == "__main__":
    store = MemoryStore()
    store.write("F")
    art = derive(store, "F")
    assert consume(store, art) == "summary(value-of-F)"
    store.forget("F")
    try:
        consume(store, art)
    except StaleDerivationError as exc:
        print(f"correctly rejected stale artifact: {exc}")