"""Generic pre-merge hook framework — Phase 4.4. A small extension point that lets domain plugins register *out-of-band* side-effects fired around a ``muse merge`` invocation. The intent is **non-blocking**: hooks may kick off background work (HTTP requests to a local daemon, file-system probes, IPC messages) but the merge engine must never wait on them, and a hook crash must never abort the merge. The first consumer is the Knowtation domain (see :mod:`muse.plugins.knowtation.hooks`), which uses this framework to fire ``POST /api/v1/memory/consolidate`` against the local Knowtation hub when the user passes ``muse merge --consolidate``. Per the :doc:`Q3 recommendation `, consolidation is *external process today, declarative pre-merge hook later* — the merge UX is never blocked on LLM latency. Design contract --------------- 1. **Out-of-band by default.** Hooks that perform any I/O whose latency is not bounded by the merge engine itself must spawn a background thread (or otherwise dispatch async work) and return immediately with :attr:`MergeHookResult.out_of_band` set to ``True``. 2. **Errors are non-fatal.** Any exception inside a hook implementation is captured by :meth:`MergeHookRegistry.run_pre_merge` and reported in :attr:`MergeHookResult.error`. The merge caller never sees the exception and never fails because of a hook failure. 3. **Domains are isolated.** Each registered plugin has its own domain string key. The registry never invokes a hook for a domain that did not register it. 4. **Thread-safe.** The module-level singleton uses an internal :class:`threading.Lock` to serialise registration so concurrent registrations (e.g. lazy-loaded plugins) cannot lose entries. """ from __future__ import annotations import logging import pathlib import threading from dataclasses import dataclass from typing import Protocol, runtime_checkable logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Result type # --------------------------------------------------------------------------- @dataclass(frozen=True) class MergeHookResult: """Result of a single pre-merge hook invocation. Frozen dataclass so callers cannot mutate the result after the hook returns; this guarantees the merge caller and the audit log see the same value the hook produced. Attributes: hook_name: Stable identifier for the hook (``"-"``, e.g. ``"knowtation-consolidate"``). Used for logging and for the ``consolidation_hook`` field in JSON CLI output. fired: ``True`` when the hook actually executed any logic. ``False`` when the hook chose to no-op (e.g. an opt-in flag was not set, or a precondition failed without being an error). out_of_band: ``True`` when the hook dispatched its real work to a background thread (or otherwise asynchronously) and the merge engine MUST NOT wait on it. By contract every Knowtation consolidation hook returns ``out_of_band=True``. message: Short human-readable status (≤ 200 chars). Surfaced in ``muse merge`` text output and in the JSON ``consolidation_hook.message`` field. error: ``None`` on success. When non-``None``, the hook encountered an error that was caught by the registry and logged at WARNING level. The merge proceeded anyway; this field exists for diagnostics, not flow-control. """ hook_name: str fired: bool out_of_band: bool message: str error: str | None # --------------------------------------------------------------------------- # Plugin protocol # --------------------------------------------------------------------------- @runtime_checkable class MergeHookPlugin(Protocol): """Structural type for a pre-merge hook provider. A plugin is any object exposing :meth:`pre_merge_hook` with the signature below. Domain plugins typically expose this in a sub-module (``muse/plugins//hooks.py``) and register a singleton via the domain's ``register_*_hook()`` helper. Implementations MUST: * Be safe to call from any thread (the merge engine may invoke multiple hooks concurrently in the future, even though today they run serially). * Return a :class:`MergeHookResult` with ``out_of_band=True`` if any I/O is performed asynchronously. * Never raise — any exceptional condition belongs in :attr:`MergeHookResult.error`. """ def pre_merge_hook( self, root: pathlib.Path, ours_commit_id: str, theirs_commit_id: str, branch: str, *, consolidate: bool = False, ) -> "MergeHookResult": """Invoke the hook for a single merge operation. Args: root: Repository root directory (contains ``.muse/``). ours_commit_id: Commit ID of HEAD of the current branch immediately *before* the merge engine writes the merge commit. Hooks MUST treat this as opaque. theirs_commit_id: Commit ID of HEAD of the branch being merged in. Same opacity contract as *ours_commit_id*. branch: Name of the branch being merged in (the "theirs" branch). Already validated by the merge command. consolidate: Domain-specific feature flag — when ``False`` the hook should return a no-op result with ``fired=False``. Plumbed through from ``muse merge --consolidate``. Returns: A :class:`MergeHookResult` describing what the hook did. Implementations MUST NOT raise; convert exceptions to ``MergeHookResult(error=str(exc), ...)``. """ ... # --------------------------------------------------------------------------- # Registry # --------------------------------------------------------------------------- class MergeHookRegistry: """Thread-safe registry mapping domain names to hook plugins. Instances are usually obtained via :func:`get_hook_registry` so all callers share the same singleton. A fresh ``MergeHookRegistry()`` instance is useful in tests where isolation between cases matters more than process-wide visibility. """ def __init__(self) -> None: """Initialise an empty registry with its own lock.""" self._plugins: dict[str, MergeHookPlugin] = {} self._lock: threading.Lock = threading.Lock() def register(self, domain: str, plugin: MergeHookPlugin) -> None: """Register *plugin* under *domain*, replacing any prior entry. Args: domain: Domain name string (e.g. ``"knowtation"``). Used as the lookup key; case-sensitive. plugin: An object implementing the :class:`MergeHookPlugin` protocol. Not validated structurally — duck-typing is sufficient because :meth:`run_pre_merge` only calls :meth:`MergeHookPlugin.pre_merge_hook`. Raises: ValueError: If *domain* is empty or not a string. """ if not isinstance(domain, str) or not domain: raise ValueError("domain must be a non-empty string") with self._lock: self._plugins[domain] = plugin def get(self, domain: str) -> MergeHookPlugin | None: """Return the registered plugin for *domain* or ``None``. Args: domain: Domain name string (e.g. ``"knowtation"``). Returns: The :class:`MergeHookPlugin` registered for *domain*, or ``None`` when no plugin has been registered. Never raises. """ with self._lock: return self._plugins.get(domain) def unregister(self, domain: str) -> bool: """Remove the plugin registered under *domain*, if any. Useful in tests that need to restore a clean registry state between cases. Args: domain: Domain name string. Returns: ``True`` if a plugin was removed, ``False`` when no plugin was registered for *domain*. """ with self._lock: return self._plugins.pop(domain, None) is not None def run_pre_merge( self, root: pathlib.Path, domain: str, ours_commit_id: str, theirs_commit_id: str, branch: str, *, consolidate: bool = False, ) -> list[MergeHookResult]: """Invoke the registered hook for *domain* and return its result. Returns a list (rather than a single result) so future versions can fan out to multiple hooks per domain without breaking the caller signature. Today the list contains zero results (no plugin registered) or one result (the registered plugin's return value). The merge engine MUST NOT wait on any work the hook dispatched — results with :attr:`MergeHookResult.out_of_band` set to ``True`` carry the audit trail of the *dispatch*, not the eventual completion. Errors raised inside the plugin's :meth:`MergeHookPlugin.pre_merge_hook` are caught, logged at WARNING level, and returned as a synthetic :class:`MergeHookResult` with the exception text in :attr:`MergeHookResult.error` — the merge call site never sees the exception. Args: root: Repository root directory. domain: Domain name string used to look up the plugin. ours_commit_id: HEAD of the current branch before merge. theirs_commit_id: HEAD of the branch being merged in. branch: Name of the branch being merged in. consolidate: Domain-specific feature flag plumbed through from ``muse merge --consolidate``. Returns: A list of :class:`MergeHookResult` (length 0 or 1 today). Empty when no plugin is registered for *domain*. """ plugin = self.get(domain) if plugin is None: return [] try: result = plugin.pre_merge_hook( root, ours_commit_id, theirs_commit_id, branch, consolidate=consolidate, ) except Exception as exc: # noqa: BLE001 — see docstring contract logger.warning( "merge_hooks: domain=%s plugin=%s raised %s: %s", domain, type(plugin).__name__, type(exc).__name__, exc, ) return [ MergeHookResult( hook_name=f"{domain}-unknown", fired=False, out_of_band=False, message="hook raised an exception (caught)", error=f"{type(exc).__name__}: {exc}", ) ] if not isinstance(result, MergeHookResult): logger.warning( "merge_hooks: domain=%s plugin=%s returned non-MergeHookResult: %r", domain, type(plugin).__name__, type(result).__name__, ) return [ MergeHookResult( hook_name=f"{domain}-unknown", fired=False, out_of_band=False, message="hook returned wrong type (ignored)", error=f"expected MergeHookResult, got {type(result).__name__}", ) ] return [result] # --------------------------------------------------------------------------- # Singleton accessor # --------------------------------------------------------------------------- _REGISTRY_LOCK: threading.Lock = threading.Lock() _REGISTRY: MergeHookRegistry | None = None def get_hook_registry() -> MergeHookRegistry: """Return the process-wide :class:`MergeHookRegistry` singleton. The singleton is constructed lazily under a module-level lock the first time this function is called, so concurrent callers in different threads always observe the same instance. Tests that need isolation should construct a fresh :class:`MergeHookRegistry` rather than mutating the singleton. Returns: The :class:`MergeHookRegistry` shared by all callers. """ global _REGISTRY if _REGISTRY is not None: return _REGISTRY with _REGISTRY_LOCK: if _REGISTRY is None: _REGISTRY = MergeHookRegistry() return _REGISTRY __all__ = [ "MergeHookPlugin", "MergeHookRegistry", "MergeHookResult", "get_hook_registry", ]