shard-wiki/src/shard_wiki/engine/extension.py

"""Typed-extension runtime — the engine framework (WikiEngineCoreArchitecture §4, E-3/E-9).

Everything beyond the kernel's c2-minimum is an :class:`Extension`: it declares a typed
contract (id, provided capabilities, declared types, bound hooks, dependencies, conflicts) and
the runtime **composes** an activation set deterministically, **rejecting impossible profiles**
(unmet deps / conflicts / type collisions) — the §6.5 capability-as-data discipline applied to
extensions. Extension structure is **verified at registration** (mirrors §6.6 conformance):
bad ids or non-callable hook handlers are refused, so the framework acts on verified data.

Hooks are dispatched in a declared, deterministic order (dependency-topological, ties by id):
*transform* hooks chain a payload through handlers; *collect* hooks gather contributions.
"""

from __future__ import annotations

from collections.abc import Callable, Iterable, Mapping
from enum import Enum
from typing import Any, ClassVar

__all__ = ["Hook", "Extension", "ExtensionError", "ExtensionRuntime", "ActiveExtensions"]


class ExtensionError(ValueError):
    """Raised when an extension is malformed or an activation set is impossible (§6.5)."""


class Hook(Enum):
    # transform hooks: each handler takes (payload, ctx) and returns the next payload
    ON_WRITE = "on_write"      # transform a draft before persist
    ON_READ = "on_read"        # transform a page on read
    ON_RESOLVE = "on_resolve"  # transform a name resolution
    ON_RENDER = "on_render"    # produce a derived representation
    # collect hooks: each handler takes (payload, ctx) and returns a contribution
    ON_LINK = "on_link"        # contribute link/transclusion edges
    ON_QUERY = "on_query"      # answer a query
    ON_PROFILE = "on_profile"  # contribute capability-profile positions (E-5)


_TRANSFORM = frozenset({Hook.ON_WRITE, Hook.ON_READ, Hook.ON_RESOLVE, Hook.ON_RENDER})
_COLLECT = frozenset({Hook.ON_LINK, Hook.ON_QUERY, Hook.ON_PROFILE})


class Extension:
    """Base class for a typed extension. Subclasses set the class vars and override
    :meth:`hooks` to bind handlers (signature ``handler(payload, ctx) -> result``)."""

    id: ClassVar[str] = ""
    provides: ClassVar[tuple[str, ...]] = ()
    declares_types: ClassVar[tuple[str, ...]] = ()
    depends_on: ClassVar[tuple[str, ...]] = ()
    conflicts_with: ClassVar[tuple[str, ...]] = ()

    def hooks(self) -> Mapping[Hook, Callable[[Any, Any], Any]]:
        return {}


class ActiveExtensions:
    """A composed, ordered activation set with deterministic hook dispatch."""

    def __init__(self, ordered: list[Extension]) -> None:
        self._ordered = ordered
        self.ids: tuple[str, ...] = tuple(e.id for e in ordered)
        self._tables: dict[Hook, list[tuple[str, Callable[[Any, Any], Any]]]] = {}
        for ext in ordered:
            for hook, fn in ext.hooks().items():
                self._tables.setdefault(hook, []).append((ext.id, fn))

    def handlers(self, hook: Hook) -> tuple[str, ...]:
        """The extension ids bound to ``hook``, in dispatch order (for introspection)."""
        return tuple(eid for eid, _ in self._tables.get(hook, ()))

    def dispatch_transform(self, hook: Hook, payload: Any, ctx: Any = None) -> Any:
        if hook not in _TRANSFORM:
            raise ExtensionError(f"{hook} is not a transform hook")
        for _eid, fn in self._tables.get(hook, ()):
            payload = fn(payload, ctx)
        return payload

    def dispatch_collect(self, hook: Hook, payload: Any = None, ctx: Any = None) -> list[Any]:
        if hook not in _COLLECT:
            raise ExtensionError(f"{hook} is not a collect hook")
        return [fn(payload, ctx) for _eid, fn in self._tables.get(hook, ())]


class ExtensionRuntime:
    def __init__(self) -> None:
        self._registered: dict[str, Extension] = {}

    def register(self, ext: Extension) -> Extension:
        """Register an extension after structural verification (mirrors §6.6)."""
        if not ext.id or not ext.id.startswith("ext."):
            raise ExtensionError(f"extension id must be 'ext.<name>', got {ext.id!r}")
        if ext.id in self._registered:
            raise ExtensionError(f"duplicate extension id: {ext.id}")
        bound = ext.hooks()
        for hook, fn in bound.items():
            if not isinstance(hook, Hook):
                raise ExtensionError(f"{ext.id}: hook key {hook!r} is not a Hook")
            if not callable(fn):
                raise ExtensionError(f"{ext.id}: handler for {hook} is not callable")
        self._registered[ext.id] = ext
        return ext

    def activate(self, ids: Iterable[str]) -> ActiveExtensions:
        """Compose an activation set: dependency closure → conflict/type checks → deterministic
        order. Raises :class:`ExtensionError` on an impossible profile."""
        requested = set(ids)
        unknown = requested - self._registered.keys()
        if unknown:
            raise ExtensionError(f"unknown extensions: {sorted(unknown)}")

        # dependency closure
        active: set[str] = set()
        frontier = list(requested)
        while frontier:
            eid = frontier.pop()
            if eid in active:
                continue
            ext = self._registered.get(eid)
            if ext is None:
                raise ExtensionError(f"unmet dependency: {eid}")
            active.add(eid)
            frontier.extend(d for d in ext.depends_on if d not in active)

        exts = [self._registered[e] for e in active]

        # conflicts
        for ext in exts:
            clash = active & set(ext.conflicts_with)
            if clash:
                raise ExtensionError(f"{ext.id} conflicts with active {sorted(clash)}")

        # type collisions (two active extensions claiming the same type id)
        owner: dict[str, str] = {}
        for ext in exts:
            for t in ext.declares_types:
                if t in owner:
                    raise ExtensionError(
                        f"type collision on {t!r}: {owner[t]} and {ext.id}"
                    )
                owner[t] = ext.id

        return ActiveExtensions(self._topo_order(exts))

    def _topo_order(self, exts: list[Extension]) -> list[Extension]:
        """Dependencies before dependents; ties broken by id (deterministic)."""
        by_id = {e.id: e for e in exts}
        ordered: list[Extension] = []
        placed: set[str] = set()

        def visit(ext: Extension) -> None:
            if ext.id in placed:
                return
            for dep in sorted(d for d in ext.depends_on if d in by_id):
                visit(by_id[dep])
            placed.add(ext.id)
            ordered.append(ext)

        for ext in sorted(exts, key=lambda e: e.id):
            visit(ext)
        return ordered