stack/packages/mosaic/framework/tools/lease-broker/activation_version_gate.py

#!/usr/bin/env python3
"""Enforcement-side version-coupling gate (issue #869, Point-1 card C4).

Root cause this exists to guard against (#828 version skew, restated from
the C1 activation probe in ``lease-activation-probe.ts``): the lease
broker's ENFORCEMENT half (this toolkit — ``launch-runtime.py``,
``mutator-gate.py``, ``revoke-lease.py``) and its ACTIVATION half
(``execLeaseGatedRuntime()`` in ``launch.ts``, which chains the gated
runtime through ``launch-runtime.py`` and injects ``MOSAIC_LEASE_*``) ship
on different channels — an npm package and a framework/CLI reseed. C1 gave
the activation half a versioned, machine-checkable identity
(``LEASE_ACTIVATION_CAPABILITY``, printed by the CLI's hidden
``mosaic __lease-capability`` subcommand). That identity is inert on its
own: nothing yet asserted that ENFORCEMENT actually requires the version
ACTIVATION advertises. This module is that assertion, owned by the
enforcement side.

``EXPECTED_ACTIVATION_CAPABILITY`` below is this toolkit's own contract
declaration — bump it only when this toolkit's launch/gate seam starts
requiring a different activation contract (new env vars it depends on,
changed chaining behavior, etc.), independent of any package semver, for
the same reason C1's constant is: #828 happened precisely because a
version number that should have moved did not.

This module never talks to a real broker or a real installed CLI in its
own tests — both the probe's command resolution and its ``run`` transport
are injectable so tests can drive every branch with fakes/stubs (see
``version_coupling_unittest.py``).
"""

from __future__ import annotations

import json
import os
import shlex
import shutil
import subprocess
from collections.abc import Callable, Mapping
from typing import Final, TypedDict


class ActivationCapability(TypedDict):
    name: str
    version: int


# ENFORCEMENT-side expected activation contract. OWNED by this toolkit (the
# enforcement half). Mirrors — but is deliberately a SEPARATE constant from
# — `LEASE_ACTIVATION_CAPABILITY` in
# `packages/mosaic/src/commands/lease-activation-probe.ts` (the activation
# half's own declaration of what it implements). The two are compared at
# runtime by `assert_activation_capability_matches()`; drift between them is
# exactly the version-skew failure mode #828/#869 exist to catch, and must
# FAIL LOUD, never a silent pass and never a dead (always-true) gate.
EXPECTED_ACTIVATION_CAPABILITY: Final[ActivationCapability] = {
    "name": "lease-runtime-activation",
    "version": 1,
}

# Matches `LEASE_CAPABILITY_PROBE_COMMAND` in lease-activation-probe.ts —
# the hidden CLI subcommand that prints the activation half's advertised
# capability as compact JSON.
LEASE_CAPABILITY_PROBE_COMMAND: Final = "__lease-capability"

PROBE_TIMEOUT_SECONDS: Final = 2.0

# Override hook: a full shell-style command line (parsed with `shlex.split`)
# to run INSTEAD of resolving `mosaic` on PATH and appending the probe
# subcommand. Real deployments should never need this — `mosaic` is on PATH
# whenever a runtime was launched via `mosaic <cmd>` in the first place, the
# only real caller of this seam. It exists for integration tests that spawn
# `launch-runtime.py` directly (never through the real CLI) to supply a
# fake/stub CLI probe, matching the existing convention of those tests
# supplying a fake broker and a fake runtime binary rather than depending on
# host state.
MOSAIC_COMMAND_OVERRIDE_VAR: Final = "MOSAIC_LEASE_VERSION_PROBE_COMMAND"


class VersionCouplingError(Exception):
    """Raised when the activation capability is absent, unreadable, or does
    not match what enforcement expects. Callers MUST fail loud on this
    (non-zero exit, clear actionable stderr) — never swallow it into a
    silent pass, and never let its absence be treated as compatible."""


def _resolve_probe_command(environ: Mapping[str, str]) -> list[str] | None:
    override = environ.get(MOSAIC_COMMAND_OVERRIDE_VAR)
    if override:
        parsed = shlex.split(override)
        return parsed or None
    resolved = shutil.which("mosaic")
    if resolved is None:
        return None
    return [resolved, LEASE_CAPABILITY_PROBE_COMMAND]


def default_probe_activation_capability(
    environ: Mapping[str, str] | None = None,
    *,
    run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> ActivationCapability | None:
    """Real capability lookup: resolves and executes the CLI's hidden
    ``__lease-capability`` probe subcommand out-of-process (the same
    mechanism `defaultCapabilityProbe()` in lease-activation-probe.ts uses
    from the activation side) and parses its JSON stdout. Any failure to
    resolve a command, spawn it, have it exit zero, or produce a well-shaped
    ``{name, version}`` JSON object is treated as NO capability (``None``)
    — fail-closed, never a fabricated/guessed capability.
    """
    source_environment = os.environ if environ is None else environ
    command = _resolve_probe_command(source_environment)
    if command is None:
        return None
    try:
        completed = run(
            command,
            capture_output=True,
            text=True,
            timeout=PROBE_TIMEOUT_SECONDS,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired, ValueError):
        return None
    if completed.returncode != 0:
        return None
    try:
        parsed = json.loads(completed.stdout)
    except json.JSONDecodeError:
        return None
    if (
        not isinstance(parsed, dict)
        or not isinstance(parsed.get("name"), str)
        or not isinstance(parsed.get("version"), int)
        or isinstance(parsed.get("version"), bool)
    ):
        return None
    return {"name": parsed["name"], "version": parsed["version"]}


def format_mismatch_message(
    activation: ActivationCapability | None,
    expected: ActivationCapability,
) -> str:
    """Actionable, non-silent remediation message for either failure shape:
    absent/unreadable capability, or a present-but-incompatible one."""
    if activation is None:
        return (
            "Mosaic lease activation capability unreadable: enforcement "
            f"expects '{expected['name']}' v{expected['version']} but the "
            f"CLI's `mosaic {LEASE_CAPABILITY_PROBE_COMMAND}` probe produced "
            "no usable result (mosaic not on PATH, non-zero exit, or "
            "malformed output) — framework/CLI version skew; upgrade both "
            "as one unit; see #869."
        )
    if activation["name"] != expected["name"]:
        return (
            f"activation capability name '{activation['name']}' != "
            f"enforcement expects '{expected['name']}' — framework/CLI "
            "version skew; upgrade both as one unit; see #869"
        )
    return (
        f"activation capability v{activation['version']} != enforcement "
        f"expects v{expected['version']} — framework/CLI version skew; "
        "upgrade both as one unit; see #869"
    )


def assert_activation_capability_matches(
    activation: ActivationCapability | None,
    expected: ActivationCapability = EXPECTED_ACTIVATION_CAPABILITY,
) -> None:
    """Raise `VersionCouplingError` unless `activation` is present AND its
    `name`/`version` exactly match `expected`. Absence is treated the same
    as a mismatch — never a silent pass."""
    if (
        activation is None
        or activation.get("name") != expected["name"]
        or activation.get("version") != expected["version"]
    ):
        raise VersionCouplingError(format_mismatch_message(activation, expected))