"""MuseHub domain manifest builder for the Knowtation domain plugin. The *hub manifest* is a JSON document that registers the Knowtation domain on the MuseHub marketplace. It is distinct from the vault-snapshot manifests in :mod:`~muse.plugins.knowtation.manifest`. Two representations of the manifest are maintained in sync: 1. **Static file** — :data:`_MANIFEST_FILE` (``manifest.json`` alongside this module) — content-addressed by its SHA-256 hash. The hash is stable across runs, making reproducible builds easy to verify. 2. **Dynamic builder** — :func:`build_manifest` — derives the same payload from the live :class:`~muse.plugins.knowtation.plugin.KnowtationPlugin` schema so that the static file and the live schema can be kept in sync via tests. Public API ---------- - :data:`MANIFEST_FILE` — absolute path to the static ``manifest.json``. - :func:`load_manifest` — parse and return the static file as a dict. - :func:`build_manifest` — derive the manifest from the live plugin schema. - :func:`manifest_hash` — compute the SHA-256 content hash of a manifest dict. - :func:`assert_manifest_in_sync` — raise if static and dynamic differ. """ from __future__ import annotations import hashlib import json import pathlib from typing import Any # --------------------------------------------------------------------------- # Paths # --------------------------------------------------------------------------- #: Absolute path to the static MuseHub manifest JSON file shipped alongside #: this module. MANIFEST_FILE: pathlib.Path = pathlib.Path(__file__).with_name("manifest.json") # Fields that must be present in a valid hub manifest. _REQUIRED_FIELDS: frozenset[str] = frozenset({ "slug", "display_name", "description", "viewer_type", "merge_semantics", "version", "dimensions", }) # Fields that must be present in each dimension entry. _REQUIRED_DIMENSION_FIELDS: frozenset[str] = frozenset({ "name", "kind", "element_type", "description", "independent_merge", }) # Expected dimension names for the knowtation domain (order-insensitive). EXPECTED_DIMENSIONS: frozenset[str] = frozenset({ "notes", "frontmatter", "sections", "links", "entities", "attachments", }) # --------------------------------------------------------------------------- # Load / build # --------------------------------------------------------------------------- def load_manifest() -> dict[str, Any]: """Parse and return the static ``manifest.json`` as a plain dict. Returns: Parsed manifest dictionary. Raises: FileNotFoundError: When ``manifest.json`` does not exist. json.JSONDecodeError: When the file is not valid JSON. """ return json.loads(MANIFEST_FILE.read_text(encoding="utf-8")) def build_manifest() -> dict[str, Any]: """Derive a MuseHub hub manifest from the live ``KnowtationPlugin`` schema. Reads the static ``manifest.json`` for fields that cannot be derived from the plugin schema alone (``viewer_type``, ``artifact_types``, ``supported_commands``, ``display_name``, ``version``), then overrides ``dimensions`` and ``description`` from the live schema to keep both in sync. Returns: A dict in the same shape as :func:`load_manifest`. Raises: NotImplementedError: When the plugin's ``schema()`` method raises. """ from muse.plugins.knowtation.plugin import KnowtationPlugin static = load_manifest() plugin = KnowtationPlugin() schema = plugin.schema() # Rebuild dimensions from the live schema — preserves descriptions from # the static file when names match so human-written text is not lost. static_dims_by_name = {d["name"]: d for d in static.get("dimensions", [])} live_dims: list[dict[str, Any]] = [] for dim in schema.get("dimensions", []): name = dim["name"] static_desc = static_dims_by_name.get(name, {}).get("description", "") live_dims.append({ "name": name, "kind": dim["schema"]["kind"], "element_type": ( dim["schema"].get("node_type") or dim["schema"].get("element_type") or dim["schema"].get("key_type") or "" ), "description": static_desc, "independent_merge": dim.get("independent_merge", False), }) return { **static, "description": schema.get("description", static.get("description", "")), "merge_semantics": static.get("merge_semantics", "ot"), "dimensions": live_dims, } # --------------------------------------------------------------------------- # Hashing # --------------------------------------------------------------------------- def manifest_hash(manifest: dict[str, Any]) -> str: """Return the SHA-256 content hash of *manifest* as ``'sha256:'``. Serialises to JSON with sorted keys and no extra whitespace for determinism, then hashes the UTF-8 bytes. Args: manifest: Manifest dict (from :func:`load_manifest` or :func:`build_manifest`). Returns: Hash string in ``'sha256:<64-hex-chars>'`` format. """ canonical = json.dumps(manifest, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() return f"sha256:{digest}" # --------------------------------------------------------------------------- # Validation helpers # --------------------------------------------------------------------------- def validate_manifest(manifest: dict[str, Any]) -> list[str]: """Return a list of validation errors for *manifest*. Checks required top-level fields, required per-dimension fields, and that the set of dimension names matches :data:`EXPECTED_DIMENSIONS`. Args: manifest: Manifest dict. Returns: List of human-readable error strings; empty list means valid. """ errors: list[str] = [] for key in sorted(_REQUIRED_FIELDS): if key not in manifest: errors.append(f"Missing required field: '{key}'") dims = manifest.get("dimensions") if not isinstance(dims, list) or not dims: errors.append("'dimensions' must be a non-empty list") return errors dim_names: set[str] = set() for i, dim in enumerate(dims): if not isinstance(dim, dict): errors.append(f"dimensions[{i}] is not a dict") continue for key in sorted(_REQUIRED_DIMENSION_FIELDS): if key not in dim: errors.append(f"dimensions[{i}] missing required field '{key}'") dim_names.add(dim.get("name", "")) missing_dims = EXPECTED_DIMENSIONS - dim_names if missing_dims: errors.append( f"Missing expected dimensions: {sorted(missing_dims)}" ) return errors def assert_manifest_in_sync() -> None: """Raise ``AssertionError`` when static and dynamic manifests diverge. Compares the SHA-256 hash of the statically declared ``manifest.json`` dimension list against the one computed from the live plugin schema. Only the dimension names are compared (not descriptions) to avoid false failures from editorial updates to human-written text. Raises: AssertionError: When the set of dimension names differs. """ static = load_manifest() dynamic = build_manifest() static_names = {d["name"] for d in static.get("dimensions", [])} dynamic_names = {d["name"] for d in dynamic.get("dimensions", [])} if static_names != dynamic_names: only_static = static_names - dynamic_names only_dynamic = dynamic_names - static_names parts = [] if only_static: parts.append(f"only in static: {sorted(only_static)}") if only_dynamic: parts.append(f"only in dynamic: {sorted(only_dynamic)}") raise AssertionError( "manifest.json dimensions out of sync with live plugin schema: " + "; ".join(parts) )