"""Knowtation pre-merge hooks — Phase 4.4. Implements :class:`muse.core.merge_hooks.MergeHookPlugin` for the Knowtation domain. Today the only hook is the consolidation kick: when the user passes ``muse merge --consolidate``, the hook fires a fire-and-forget HTTP ``POST /api/v1/memory/consolidate`` against the local Knowtation hub (``hub/server.mjs``) and returns immediately. The merge engine never waits on the response — per Q3 of the Knowtation domain plan, consolidation runs *external process today, declarative pre-merge hook later*, so the merge UX is never blocked on LLM latency. Security -------- The HTTP target URL is built **server-side only** from a port number that has been validated as an integer in the IANA-allowed range (``1`` – ``65535``). The port is read from the environment variable ``KNOWTATION_HUB_PORT`` (default ``3000``). No segment of the URL is ever interpolated from user-controlled string input; an attacker controlling the environment variable can at worst cause the hook to emit an error result (``KNOWTATION_HUB_PORT=oops``) or reach a different TCP port on ``localhost`` (``KNOWTATION_HUB_PORT=8080``). They cannot inject shell metacharacters, redirect to a remote host, smuggle additional URL paths, or escalate to anything beyond a single ``POST`` request to ``http://localhost:/api/v1/memory/consolidate`` with an empty JSON body. Errors ------ Any exception raised by :func:`urllib.request.urlopen` (connection refused, timeout, DNS error, malformed response) is caught inside the background thread, logged at ``WARNING`` level, and never propagates to the merge process. The synchronous :class:`MergeHookResult` returned by :meth:`KnowtationMergeHook.pre_merge_hook` always carries ``out_of_band=True`` when the hook successfully *dispatched* the work, even though the underlying HTTP call may still fail asynchronously. """ from __future__ import annotations import json import logging import os import pathlib import threading import urllib.error import urllib.request from muse.core.merge_hooks import ( MergeHookPlugin, MergeHookResult, get_hook_registry, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Configuration constants # --------------------------------------------------------------------------- #: Stable identifier for the consolidation hook. Used by both the #: registry log lines and the ``muse merge --consolidate`` JSON output. HOOK_NAME: str = "knowtation-consolidate" #: Environment variable read for the Knowtation hub TCP port. Default #: ``3000`` matches the canonical hub port used by Phase 4.4 fixtures #: and the ``hub/server.mjs`` default. ENV_PORT: str = "KNOWTATION_HUB_PORT" #: Default port when ``KNOWTATION_HUB_PORT`` is unset or empty. DEFAULT_PORT: int = 3000 #: Lowest legal TCP port number (excluding port ``0``, which is #: reserved by IANA for "any free port" and is not a real listener). PORT_MIN: int = 1 #: Highest legal TCP port number per RFC 6335. PORT_MAX: int = 65535 #: Hard timeout (seconds) on the background HTTP call. The merge has #: already returned by the time this fires; the ceiling exists only to #: ensure the daemon thread does not leak resources for a hung server. HTTP_TIMEOUT_SECONDS: float = 30.0 #: API path on the hub. Constructed as a constant string so callers #: cannot influence it. API_PATH: str = "/api/v1/memory/consolidate" # --------------------------------------------------------------------------- # Port validation # --------------------------------------------------------------------------- def _resolve_port() -> tuple[int | None, str | None]: """Read and validate the configured Knowtation hub port. Reads :data:`ENV_PORT` from :data:`os.environ`. When the variable is missing or empty, returns :data:`DEFAULT_PORT`. Otherwise, the value must parse as a base-10 integer in ``[PORT_MIN, PORT_MAX]``. Any other input — including strings with shell metacharacters such as ``"3000; rm -rf /"`` — fails the integer parse and returns an error tuple. Returns: ``(port, None)`` on success, where *port* is an ``int`` in the valid TCP range. ``(None, error_message)`` on failure, where *error_message* is suitable for inclusion in a :class:`MergeHookResult.error`. """ raw = os.environ.get(ENV_PORT, "") if not raw or not raw.strip(): return DEFAULT_PORT, None raw = raw.strip() try: port = int(raw, 10) except ValueError: return None, ( f"{ENV_PORT}={raw!r} is not a valid integer " f"(expected an integer in [{PORT_MIN}, {PORT_MAX}])" ) if port < PORT_MIN or port > PORT_MAX: return None, ( f"{ENV_PORT}={port} is out of range " f"(must be in [{PORT_MIN}, {PORT_MAX}])" ) return port, None # --------------------------------------------------------------------------- # Background HTTP worker # --------------------------------------------------------------------------- def _post_consolidate_in_background(url: str) -> threading.Thread: """Spawn a daemon thread that POSTs ``{}`` to *url* and returns it. The thread is started before the function returns. The thread is marked as a daemon so it does not block process exit; the merge command can return immediately while this thread is still running. Args: url: Pre-validated ``http://localhost:/api/v1/memory/consolidate`` URL. Callers MUST construct the URL via :func:`_resolve_port`; this function performs no further validation. Returns: The started :class:`threading.Thread`. Returned for testability — the merge engine itself ignores it. """ def _worker() -> None: body = b"{}" request = urllib.request.Request( url=url, data=body, method="POST", headers={ "Content-Type": "application/json", "User-Agent": "muse-merge-hook/1.0", "Content-Length": str(len(body)), }, ) try: with urllib.request.urlopen( # noqa: S310 — URL is server-built request, timeout=HTTP_TIMEOUT_SECONDS, ) as response: logger.debug( "knowtation-consolidate: POST %s -> %s", url, getattr(response, "status", "unknown"), ) except urllib.error.URLError as exc: logger.warning( "knowtation-consolidate: POST %s failed (URLError): %s", url, exc, ) except TimeoutError as exc: logger.warning( "knowtation-consolidate: POST %s timed out after %.1fs: %s", url, HTTP_TIMEOUT_SECONDS, exc, ) except Exception as exc: # noqa: BLE001 — must not propagate logger.warning( "knowtation-consolidate: POST %s raised %s: %s", url, type(exc).__name__, exc, ) thread = threading.Thread( target=_worker, name="knowtation-consolidate-hook", daemon=True, ) thread.start() return thread # --------------------------------------------------------------------------- # Public hook plugin # --------------------------------------------------------------------------- class KnowtationMergeHook: """Pre-merge hook that fires the Knowtation consolidation daemon. Implements :class:`muse.core.merge_hooks.MergeHookPlugin`. Behaviour: * ``consolidate=False`` (default) — return a no-op result with ``fired=False`` and ``out_of_band=False``. The hook does no I/O and reads no environment variables. * ``consolidate=True`` — read and validate :data:`ENV_PORT`, build ``http://localhost:/api/v1/memory/consolidate`` from the port int (no string interpolation from user input), spawn a daemon thread that POSTs an empty JSON body to that URL, and return ``fired=True, out_of_band=True``. Any subsequent failure inside the daemon thread is logged but never raised. The hook itself never raises — port validation failures and exceptions inside the dispatch path are converted to a :class:`MergeHookResult` with the failure message in the ``error`` field. """ def pre_merge_hook( self, root: pathlib.Path, ours_commit_id: str, theirs_commit_id: str, branch: str, *, consolidate: bool = False, ) -> MergeHookResult: """Dispatch the consolidation request and return a status record. Args: root: Repository root directory (currently unused; reserved for future hooks that need to read ``.muse/`` state). ours_commit_id: HEAD commit before merge (currently unused; reserved for richer payloads in a later phase). theirs_commit_id: HEAD of the merged-in branch (currently unused; reserved for richer payloads in a later phase). branch: Name of the merged-in branch (currently unused; reserved for richer payloads in a later phase). consolidate: When ``False``, return a no-op :class:`MergeHookResult` with ``fired=False`` and do no I/O. When ``True``, dispatch the consolidation HTTP POST in a background thread and return immediately. Returns: A :class:`MergeHookResult` describing the dispatch outcome. Never raises. """ del root, ours_commit_id, theirs_commit_id, branch if not consolidate: return MergeHookResult( hook_name=HOOK_NAME, fired=False, out_of_band=False, message="--consolidate not requested", error=None, ) port, port_error = _resolve_port() if port is None: return MergeHookResult( hook_name=HOOK_NAME, fired=False, out_of_band=False, message="port validation failed; consolidation skipped", error=port_error, ) url = f"http://localhost:{port}{API_PATH}" try: _post_consolidate_in_background(url) except Exception as exc: # noqa: BLE001 — must not propagate logger.warning( "knowtation-consolidate: failed to spawn background thread: %s", exc, ) return MergeHookResult( hook_name=HOOK_NAME, fired=False, out_of_band=False, message="failed to dispatch background HTTP call", error=f"{type(exc).__name__}: {exc}", ) return MergeHookResult( hook_name=HOOK_NAME, fired=True, out_of_band=True, message=f"dispatched POST {API_PATH} to localhost:{port} (background, {HTTP_TIMEOUT_SECONDS:.0f}s timeout)", error=None, ) # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register_knowtation_hook() -> None: """Register a :class:`KnowtationMergeHook` instance under ``"knowtation"``. Idempotent — calling this function more than once replaces the previous registration with a fresh instance, so it is safe to call eagerly at import time as well as lazily on first need. """ get_hook_registry().register("knowtation", KnowtationMergeHook()) # Expose a parsable JSON-shaped hook result body — used in CLI tests to # document the wire shape without redefining it. def hook_result_to_json(result: MergeHookResult) -> str: """Render a :class:`MergeHookResult` as a JSON string. Helper used by the merge CLI to embed hook results in the ``--format json`` output as the ``consolidation_hook`` field. Args: result: A :class:`MergeHookResult` returned from a hook. Returns: A compact JSON string with keys ``hook_name``, ``fired``, ``out_of_band``, ``message``, ``error``. """ return json.dumps( { "hook_name": result.hook_name, "fired": result.fired, "out_of_band": result.out_of_band, "message": result.message, "error": result.error, }, sort_keys=True, ) __all__ = [ "API_PATH", "DEFAULT_PORT", "ENV_PORT", "HOOK_NAME", "HTTP_TIMEOUT_SECONDS", "KnowtationMergeHook", "PORT_MAX", "PORT_MIN", "hook_result_to_json", "register_knowtation_hook", ]