"""Generic post-push hook framework — Phase 7.4. Mirror of :mod:`muse.core.merge_hooks` but for the ``muse push`` lifecycle. Domain plugins use it to dispatch *out-of-band* side-effects after a successful push — typically anchoring commits in an external trust root (ICP canister, blockchain, AIR ledger). The first consumer is :class:`muse.plugins.knowtation.icp_push_hook.KnowtationICPPushHook` which writes attestation records to the immutable ICP canister ``dejku-syaaa-aaaaa-qgy3q-cai`` after a knowtation vault push. Design contract (identical to merge_hooks) ------------------------------------------ 1. **Out-of-band by default.** Hooks performing network I/O MUST spawn a background thread and return immediately with :attr:`PushHookResult.out_of_band` set to ``True``. 2. **Errors are non-fatal.** Any exception inside the hook is caught and reported in :attr:`PushHookResult.error`. The push command never fails because of a hook failure. 3. **Domain isolation.** Each plugin registers under a domain string; the registry never invokes a hook for a domain that did not register it. 4. **Thread-safe registry.** Module-level singleton serialises registration so concurrent registrations cannot lose entries. 5. **Idempotent.** Re-running the same hook on the same push payload MUST produce the same effect — relevant for the ICP canister's "already exists" rejection path. """ from __future__ import annotations import logging import pathlib import threading from dataclasses import dataclass, field from typing import Protocol, runtime_checkable logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Result type # --------------------------------------------------------------------------- @dataclass(frozen=True) class PushHookResult: """Result of a single post-push hook invocation. Attributes: hook_name: Stable identifier — typically ``"-"``. Used for logging and for the ``push_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. By contract every ICP push hook returns ``out_of_band=True``. skipped: ``True`` when the hook chose not to fire because of a documented condition (e.g. ICP unreachable, MUSE_ICP_ENABLED=0). Distinct from ``fired=False`` (no opt-in) and ``error`` (failure). reason: Short machine-readable token explaining ``skipped=True``. Example values: ``"ICP_UNREACHABLE"``, ``"DISABLED"``, ``"NO_ATTESTATION_IN_COMMIT"``. Empty when not skipped. message: Human-readable status (≤ 200 chars). Surfaced in ``muse push`` text output. error: ``None`` on success. When non-``None``, the hook encountered an error that was caught by the registry and logged at WARNING. anchored_count: Number of attestation records the hook successfully anchored. ``0`` when nothing to anchor or when ``skipped=True``. failed_count: Number of attestation records the hook failed to anchor (e.g. transient ICP errors). ``failed_count > 0`` does NOT mean the hook itself failed — the records are simply not yet anchored and may be retried later. """ hook_name: str fired: bool out_of_band: bool skipped: bool reason: str message: str error: str | None anchored_count: int = 0 failed_count: int = 0 # --------------------------------------------------------------------------- # Plugin protocol # --------------------------------------------------------------------------- @runtime_checkable class PushHookPlugin(Protocol): """Structural type for a post-push hook provider. Implementations MUST: * Be safe to call from any thread. * Return a :class:`PushHookResult` with ``out_of_band=True`` if any I/O is dispatched asynchronously. * Never raise — exceptional conditions belong in :attr:`PushHookResult.error`. """ def run_post_push( self, repo_path: pathlib.Path, push_result: dict[str, object], domain: str, config: dict[str, object] | None = None, ) -> PushHookResult: """Invoke the hook for a single push operation. Args: repo_path: Repository root. push_result: Dict describing what was pushed: ``branches`` (list of branch names), ``commits`` (list of commit IDs newly pushed to the remote — domain-specific), ``remote`` (remote name string). Hooks MUST treat any non-documented keys as opaque. domain: Domain string the hook is registered under. config: Domain-specific configuration dict (e.g. ICP canister ID, identity PEM path). Hooks MUST validate any field they read; unrecognised keys are ignored. Returns: A :class:`PushHookResult` describing what the hook did. Implementations MUST NOT raise. """ ... # --------------------------------------------------------------------------- # Registry # --------------------------------------------------------------------------- class PushHookRegistry: """Thread-safe registry mapping domain names to push hook plugins. Use :func:`get_push_hook_registry` to obtain the process-wide singleton. Construct a fresh ``PushHookRegistry()`` for tests that need isolation. """ def __init__(self) -> None: """Initialise an empty registry with its own lock.""" self._plugins: dict[str, PushHookPlugin] = {} self._lock: threading.Lock = threading.Lock() def register(self, domain: str, plugin: PushHookPlugin) -> None: """Register *plugin* under *domain*, replacing any prior entry. Args: domain: Non-empty domain name string. plugin: An object implementing :class:`PushHookPlugin`. 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) -> PushHookPlugin | None: """Return the registered plugin for *domain* or ``None``. Args: domain: Domain name string. Returns: The registered :class:`PushHookPlugin`, or ``None`` when no plugin has been registered for *domain*. """ with self._lock: return self._plugins.get(domain) def unregister(self, domain: str) -> bool: """Remove the plugin registered under *domain*, if any. Returns: ``True`` if a plugin was removed, ``False`` otherwise. """ with self._lock: return self._plugins.pop(domain, None) is not None def run_post_push( self, repo_path: pathlib.Path, domain: str, push_result: dict[str, object], config: dict[str, object] | None = None, ) -> list[PushHookResult]: """Invoke the registered hook for *domain* and return its result(s). 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). Errors raised inside the plugin's :meth:`PushHookPlugin.run_post_push` are caught, logged at WARNING level, and returned as a synthetic :class:`PushHookResult` with the exception text in :attr:`PushHookResult.error` — the push call site never sees the exception. Args: repo_path: Repository root. domain: Domain name string used to look up the plugin. push_result: See :meth:`PushHookPlugin.run_post_push`. config: Domain-specific configuration dict. Returns: A list of :class:`PushHookResult` (length 0 or 1 today). """ plugin = self.get(domain) if plugin is None: return [] try: result = plugin.run_post_push(repo_path, push_result, domain, config) except Exception as exc: # noqa: BLE001 — see docstring contract logger.warning( "push_hooks: domain=%s plugin=%s raised %s: %s", domain, type(plugin).__name__, type(exc).__name__, exc, ) return [PushHookResult( hook_name=f"{domain}-unknown", fired=False, out_of_band=False, skipped=False, reason="", message="hook raised an exception (caught)", error=f"{type(exc).__name__}: {exc}", )] if not isinstance(result, PushHookResult): logger.warning( "push_hooks: domain=%s plugin=%s returned non-PushHookResult: %r", domain, type(plugin).__name__, type(result).__name__, ) return [PushHookResult( hook_name=f"{domain}-unknown", fired=False, out_of_band=False, skipped=False, reason="", message="hook returned wrong type (ignored)", error=f"expected PushHookResult, got {type(result).__name__}", )] return [result] # --------------------------------------------------------------------------- # Singleton accessor # --------------------------------------------------------------------------- _REGISTRY_LOCK: threading.Lock = threading.Lock() _REGISTRY: PushHookRegistry | None = None def get_push_hook_registry() -> PushHookRegistry: """Return the process-wide :class:`PushHookRegistry` singleton. 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:`PushHookRegistry` rather than mutating the singleton. Returns: The :class:`PushHookRegistry` shared by all callers. """ global _REGISTRY if _REGISTRY is not None: return _REGISTRY with _REGISTRY_LOCK: if _REGISTRY is None: _REGISTRY = PushHookRegistry() return _REGISTRY __all__ = [ "PushHookPlugin", "PushHookRegistry", "PushHookResult", "get_push_hook_registry", ]