diff --git a/packages/mosaic/framework/tools/lease-broker/activation_version_gate.py b/packages/mosaic/framework/tools/lease-broker/activation_version_gate.py new file mode 100644 index 00000000..b1d151b2 --- /dev/null +++ b/packages/mosaic/framework/tools/lease-broker/activation_version_gate.py @@ -0,0 +1,180 @@ +#!/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 ` 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)) diff --git a/packages/mosaic/framework/tools/lease-broker/launch-runtime.py b/packages/mosaic/framework/tools/lease-broker/launch-runtime.py index f54aec46..c0fe8259 100644 --- a/packages/mosaic/framework/tools/lease-broker/launch-runtime.py +++ b/packages/mosaic/framework/tools/lease-broker/launch-runtime.py @@ -12,11 +12,24 @@ from collections.abc import Callable, Mapping, Sequence from pathlib import Path from typing import Final +from activation_version_gate import ( + EXPECTED_ACTIVATION_CAPABILITY, + ActivationCapability, + VersionCouplingError, + assert_activation_capability_matches, + default_probe_activation_capability, +) from lease_generation import initialize_runtime_generation MAX_FRAME: Final = 64 * 1024 BROKER_TIMEOUT_SECONDS: Final = 1.5 CLAUDE_DANGEROUS_FLAG: Final = "--dangerously-skip-permissions" +# Distinct, non-overlapping exit code for the C4 version-coupling gate (see +# `activation_version_gate.py`) — deliberately different from the `1` +# (broker registration failed closed) and `64` (usage error) codes already +# owned by this script, so a version-skew denial is unambiguous in caller +# logs/tests and is never confused with a broker-availability failure. +EXIT_VERSION_SKEW: Final = 65 def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]: @@ -47,6 +60,10 @@ def main( request: Callable[[Path, dict[str, object]], dict[str, object]] = broker_request, execute: Callable[[str, list[str], dict[str, str]], object] = os.execvpe, initialize_generation: Callable[[Path, int], None] = initialize_runtime_generation, + probe_activation_capability: Callable[ + [Mapping[str, str]], ActivationCapability | None + ] = default_probe_activation_capability, + expected_activation_capability: ActivationCapability = EXPECTED_ACTIVATION_CAPABILITY, ) -> int: parser = argparse.ArgumentParser() parser.add_argument("--runtime", required=True, choices=("claude", "pi")) @@ -66,6 +83,25 @@ def main( command = [command[0], CLAUDE_DANGEROUS_FLAG, *command[1:]] source_environment = os.environ if environ is None else environ + + # C4 version-coupling gate (#869 Point-1): before this ENFORCEMENT half + # chains into anything, assert that the ACTIVATION contract it is about + # to rely on (MOSAIC_LEASE_* injection, broker chaining) matches what + # this enforcement build expects. This is a build/deploy-defect check, + # not a broker-availability question, so it runs before — and + # independently of — broker registration below, and it FAILS LOUD: a + # clear stderr message plus a dedicated non-zero exit code, never a + # silent pass and never folded into the generic registration-failure + # branch. + try: + assert_activation_capability_matches( + probe_activation_capability(source_environment), + expected_activation_capability, + ) + except VersionCouplingError as version_error: + print(str(version_error), file=sys.stderr) + return EXIT_VERSION_SKEW + try: socket_path = Path(source_environment["MOSAIC_LEASE_BROKER_SOCKET"]) generation = int(source_environment.get("MOSAIC_RUNTIME_GENERATION", "1")) diff --git a/packages/mosaic/package.json b/packages/mosaic/package.json index de27c35a..2a719148 100644 --- a/packages/mosaic/package.json +++ b/packages/mosaic/package.json @@ -25,7 +25,7 @@ "lint": "eslint src", "typecheck": "tsc --noEmit", "test": "vitest run --passWithNoTests && pnpm run test:framework-shell", - "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/lease-broker/receipt_challenge_unittest.py && python3 src/lease-broker/context_recovery_unittest.py && python3 src/lease-broker/recovery_runtime_unittest.py && python3 src/lease-broker/recovery_b1_adversarial_unittest.py && python3 src/lease-broker/framework_skill_portability_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh && bash framework/tools/qa/test-deps-preflight.sh && bash framework/tools/git/test-pr-review-gitea-comment.sh && bash framework/tools/git/test-ci-queue-wait-branch-absent.sh && bash framework/tools/git/test-git-credential-mosaic.sh && bash framework/tools/git/test-gitea-token-identity.sh && bash framework/tools/_scripts/test-install-ordering-guard.sh" + "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/lease-broker/receipt_challenge_unittest.py && python3 src/lease-broker/context_recovery_unittest.py && python3 src/lease-broker/recovery_runtime_unittest.py && python3 src/lease-broker/recovery_b1_adversarial_unittest.py && python3 src/lease-broker/framework_skill_portability_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 src/mutator-gate/version_coupling_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh && bash framework/tools/qa/test-deps-preflight.sh && bash framework/tools/git/test-pr-review-gitea-comment.sh && bash framework/tools/git/test-ci-queue-wait-branch-absent.sh && bash framework/tools/git/test-git-credential-mosaic.sh && bash framework/tools/git/test-gitea-token-identity.sh && bash framework/tools/_scripts/test-install-ordering-guard.sh" }, "dependencies": { "@mosaicstack/brain": "workspace:*", diff --git a/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts b/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts index 4fb90ea6..97ad2a3c 100644 --- a/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts +++ b/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts @@ -47,6 +47,22 @@ const piLifecyclePath = join(frameworkRoot, 'runtime/pi/lease-lifecycle.ts'); const prdyInitPath = join(frameworkRoot, 'tools/prdy/prdy-init.sh'); const prdyUpdatePath = join(frameworkRoot, 'tools/prdy/prdy-update.sh'); const remediationHandlerPath = join(frameworkRoot, 'tools/qa/remediation-hook-handler.sh'); + +// C4 (#869 Point-1): launch-runtime.py now asserts, before anything else, +// that the CLI's advertised lease-activation capability (normally read via +// the hidden `mosaic __lease-capability` subcommand) matches what +// enforcement expects — see framework/tools/lease-broker/ +// activation_version_gate.py. This suite drives launch-runtime.py directly +// as a subprocess (never through the real `mosaic` CLI), so — exactly like +// the fake broker (daemon.py) and fake `claude` binaries already used +// below — it must supply a fake activation-capability probe rather than +// depend on a real `mosaic` binary being on PATH. `MOSAIC_LEASE_VERSION_PROBE_COMMAND` +// is launch-runtime.py's injection point for that fake; this literal +// {name, version} pair must be kept in sync with +// `EXPECTED_ACTIVATION_CAPABILITY` (activation_version_gate.py) and +// `LEASE_ACTIVATION_CAPABILITY` (lease-activation-probe.ts) — all three +// currently agree on v1. +const leaseCapabilityProbeStub = `python3 -c "import json; print(json.dumps({'name': 'lease-runtime-activation', 'version': 1}))"`; const children: ChildProcess[] = []; const temporaryRoots: string[] = []; @@ -184,6 +200,7 @@ raise SystemExit(0 if len(session_id) == 64 and denied else 1) MOSAIC_PRDY_RUNTIME: 'claude', MOSAIC_LEASE_BROKER_SOCKET: socket, MOSAIC_RUNTIME_GENERATION: '1', + MOSAIC_LEASE_VERSION_PROBE_COMMAND: leaseCapabilityProbeStub, }, }); } @@ -743,6 +760,7 @@ describe('whole mutator-class lease gate', () => { ...process.env, MOSAIC_LEASE_BROKER_SOCKET: socket, MOSAIC_RUNTIME_GENERATION: '1', + MOSAIC_LEASE_VERSION_PROBE_COMMAND: leaseCapabilityProbeStub, }, }, ); @@ -761,6 +779,7 @@ describe('whole mutator-class lease gate', () => { ...process.env, MOSAIC_LEASE_BROKER_SOCKET: join(tmpdir(), 'missing-mosaic-broker.sock'), MOSAIC_RUNTIME_GENERATION: '1', + MOSAIC_LEASE_VERSION_PROBE_COMMAND: leaseCapabilityProbeStub, }, }, ); @@ -878,6 +897,7 @@ raise SystemExit(0 if len(session_id) == 64 and hook_present and observers_prese PATH: `${binDir}:${process.env.PATH ?? ''}`, MOSAIC_LEASE_BROKER_SOCKET: socket, MOSAIC_RUNTIME_GENERATION: '1', + MOSAIC_LEASE_VERSION_PROBE_COMMAND: leaseCapabilityProbeStub, }, proxyGate: () => Promise.resolve({ diff --git a/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py b/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py index fa979ebc..25947b6a 100644 --- a/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py +++ b/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py @@ -39,6 +39,18 @@ LAUNCHER = load_tool("lease_runtime_launcher", "launch-runtime.py") GATE = load_tool("lease_mutator_gate", "mutator-gate.py") +def matching_activation_probe(*_args: object, **_kwargs: object) -> dict[str, object]: + """Fake activation-capability probe matching what enforcement expects + (C4, #869 Point-1). Injected into `LAUNCHER.main()` calls below that are + exercising OTHER branches (registration, exec, generation init, ...) so + the new version-coupling gate — which runs before those — never blocks + on host state (no real `mosaic` CLI on PATH in a test sandbox). The + version-coupling gate's OWN behavior (match/mismatch/absent) is covered + by its dedicated red-first tests in `version_coupling_unittest.py`.""" + + return dict(LAUNCHER.EXPECTED_ACTIVATION_CAPABILITY) + + class FakeSocket: def __init__(self, *chunks: bytes): self.chunks = list(chunks) @@ -95,6 +107,7 @@ class LaunchRuntimeTest(unittest.TestCase): request=request, execute=execute, initialize_generation=initialize_generation, + probe_activation_capability=matching_activation_probe, ) self.assertEqual(result, 0) @@ -127,6 +140,7 @@ class LaunchRuntimeTest(unittest.TestCase): request=lambda *_args: {"ok": True, "session_id": "e" * 64}, execute=lambda *args: executed.append(args), initialize_generation=lambda *_args: None, + probe_activation_capability=matching_activation_probe, ) self.assertEqual(result, 0) self.assertEqual( @@ -153,6 +167,7 @@ class LaunchRuntimeTest(unittest.TestCase): request=lambda *_args: {"ok": True, "session_id": "f" * 64}, execute=lambda *args: executed.append(args), initialize_generation=lambda *_args: None, + probe_activation_capability=matching_activation_probe, ) self.assertEqual(result, 0) self.assertEqual(executed[0][0:2], ("pi", ["pi", "--print", "hello"])) @@ -188,6 +203,7 @@ class LaunchRuntimeTest(unittest.TestCase): environ=environment, request=lambda *_args, value=reply: value, execute=lambda *args: executed.append(args), + probe_activation_capability=matching_activation_probe, ) self.assertEqual(result, 1) self.assertEqual(executed, []) @@ -203,6 +219,7 @@ class LaunchRuntimeTest(unittest.TestCase): initialize_generation=lambda *_args: (_ for _ in ()).throw( OSError("unsafe state") ), + probe_activation_capability=matching_activation_probe, ), 1, ) @@ -220,6 +237,7 @@ class LaunchRuntimeTest(unittest.TestCase): environ={"MOSAIC_LEASE_BROKER_SOCKET": "/x"}, request=request, execute=lambda *_args: self.fail("must not execute"), + probe_activation_capability=matching_activation_probe, ), 1, ) @@ -233,6 +251,7 @@ class LaunchRuntimeTest(unittest.TestCase): request=lambda *_args: {"ok": True, "session_id": "c" * 64}, execute=lambda *_args: (_ for _ in ()).throw(OSError("missing")), initialize_generation=lambda *_args: None, + probe_activation_capability=matching_activation_probe, ), 1, ) diff --git a/packages/mosaic/src/mutator-gate/version_coupling_unittest.py b/packages/mosaic/src/mutator-gate/version_coupling_unittest.py new file mode 100644 index 00000000..6151170e --- /dev/null +++ b/packages/mosaic/src/mutator-gate/version_coupling_unittest.py @@ -0,0 +1,287 @@ +#!/usr/bin/env python3 +"""Red-first tests for issue #869 Point-1 C4 — the enforcement/activation +version-coupling assertion at the `launch-runtime.py` seam. + +Root cause under test (#828 restated): the lease broker's ENFORCEMENT half +(this toolkit) and its ACTIVATION half (`execLeaseGatedRuntime()` in +`launch.ts`, chained through `launch-runtime.py`) shipped on different +channels and drifted. C1 (`lease-activation-probe.ts`) gave the activation +half a versioned, machine-checkable identity +(`LEASE_ACTIVATION_CAPABILITY`, printed via the hidden CLI subcommand +`mosaic __lease-capability`). C4 (this module + `activation_version_gate.py`) +is the assertion that actually USES that identity: enforcement must refuse +to proceed — loudly, with an actionable remediation message, never a +silent pass — unless the activation capability it observes exactly matches +what enforcement expects. + +Every case here drives the seam with injected fakes/stubs (a fake +`probe_activation_capability` callable at the `launch-runtime.py` level, or +a fake `run` transport at the `activation_version_gate` level) — never a +real broker, a real installed CLI, or a real `mosaic` binary on PATH. +""" + +from __future__ import annotations + +import importlib.util +import io +import subprocess +import sys +import unittest +from contextlib import redirect_stderr +from pathlib import Path + + +TOOLS_DIR = Path(__file__).parents[2] / "framework/tools/lease-broker" +if str(TOOLS_DIR) not in sys.path: + sys.path.insert(0, str(TOOLS_DIR)) + + +def load_tool(module_name: str, filename: str): + spec = importlib.util.spec_from_file_location(module_name, TOOLS_DIR / filename) + if spec is None or spec.loader is None: + raise RuntimeError(f"unable to load {filename}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +# Loaded under distinct module names from runtime_tools_unittest.py's own +# LAUNCHER/GATE loads — importlib.util.module_from_spec() gives each load a +# fresh module object regardless of name collisions, but distinct names keep +# tracebacks/debugging unambiguous when both files run in the same process. +LAUNCHER = load_tool("lease_runtime_launcher_version_coupling", "launch-runtime.py") +VERSION_GATE = load_tool("lease_activation_version_gate_test", "activation_version_gate.py") + + +def matching_capability() -> dict[str, object]: + return dict(VERSION_GATE.EXPECTED_ACTIVATION_CAPABILITY) + + +class AssertActivationCapabilityMatchesTest(unittest.TestCase): + """Unit-level coverage of `activation_version_gate.py`'s own assertion, + isolated from the launch-runtime.py seam it is wired into below.""" + + def test_matching_capability_passes_silently(self) -> None: + VERSION_GATE.assert_activation_capability_matches(matching_capability()) + # No exception is the assertion; nothing further to check. + + def test_absent_capability_fails_closed_not_silent_pass(self) -> None: + with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised: + VERSION_GATE.assert_activation_capability_matches(None) + message = str(raised.exception) + self.assertIn("#869", message) + self.assertIn("upgrade", message.lower()) + + def test_version_mismatch_message_is_actionable(self) -> None: + expected = {"name": "lease-runtime-activation", "version": 1} + mismatched = {"name": "lease-runtime-activation", "version": 2} + with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised: + VERSION_GATE.assert_activation_capability_matches(mismatched, expected) + message = str(raised.exception) + self.assertIn("v2", message) + self.assertIn("v1", message) + self.assertIn("#869", message) + self.assertIn("upgrade", message.lower()) + self.assertIn("version skew", message.lower()) + + def test_name_mismatch_fails_loud(self) -> None: + expected = {"name": "lease-runtime-activation", "version": 1} + mismatched = {"name": "some-other-capability", "version": 1} + with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised: + VERSION_GATE.assert_activation_capability_matches(mismatched, expected) + message = str(raised.exception) + self.assertIn("some-other-capability", message) + self.assertIn("lease-runtime-activation", message) + self.assertIn("#869", message) + + def test_reversed_drift_newer_activation_than_enforcement_expects_also_fails(self) -> None: + # A build/deploy where ACTIVATION shipped ahead of ENFORCEMENT is + # exactly as much version skew as the reverse (#828's actual shape + # was enforcement ahead of activation) — the assertion must not special + # case direction. + expected = {"name": "lease-runtime-activation", "version": 1} + newer_activation = {"name": "lease-runtime-activation", "version": 2} + with self.assertRaises(VERSION_GATE.VersionCouplingError): + VERSION_GATE.assert_activation_capability_matches(newer_activation, expected) + + +class ProbeActivationCapabilityTest(unittest.TestCase): + """Coverage of the probe's command resolution and fail-closed transport + handling — never spawns a real `mosaic` process.""" + + def test_returns_none_when_mosaic_is_not_resolvable_on_path(self) -> None: + result = VERSION_GATE.default_probe_activation_capability( + {"PATH": "/nonexistent-bin-dir-for-869-c4-test"} + ) + self.assertIsNone(result) + + def test_override_command_is_parsed_and_the_probe_subcommand_is_not_double_appended( + self, + ) -> None: + captured: list[list[str]] = [] + + class FakeCompleted: + returncode = 0 + stdout = '{"name": "lease-runtime-activation", "version": 1}' + + def fake_run(argv: list[str], **_kwargs: object) -> FakeCompleted: + captured.append(argv) + return FakeCompleted() + + result = VERSION_GATE.default_probe_activation_capability( + {VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic __lease-capability"}, + run=fake_run, + ) + self.assertEqual(result, {"name": "lease-runtime-activation", "version": 1}) + self.assertEqual(captured, [["/fake/mosaic", "__lease-capability"]]) + + def test_fails_closed_on_nonzero_exit_malformed_json_and_missing_fields(self) -> None: + class NonZeroExit: + returncode = 1 + stdout = '{"name": "lease-runtime-activation", "version": 1}' + + class MalformedOutput: + returncode = 0 + stdout = "not-json" + + class MissingVersion: + returncode = 0 + stdout = '{"name": "lease-runtime-activation"}' + + class WrongShapeVersion: + returncode = 0 + stdout = '{"name": "lease-runtime-activation", "version": "1"}' + + class BooleanVersion: + # bool is a subclass of int in Python; must not be accepted as + # a version number. + returncode = 0 + stdout = '{"name": "lease-runtime-activation", "version": true}' + + for fake in ( + NonZeroExit(), + MalformedOutput(), + MissingVersion(), + WrongShapeVersion(), + BooleanVersion(), + ): + with self.subTest(stdout=fake.stdout, returncode=fake.returncode): + result = VERSION_GATE.default_probe_activation_capability( + {VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic"}, + run=lambda *_a, fake=fake, **_kw: fake, + ) + self.assertIsNone(result) + + def test_fails_closed_on_timeout_and_transport_error(self) -> None: + def timeout_run(*_args: object, **_kwargs: object) -> None: + raise subprocess.TimeoutExpired(cmd="mosaic", timeout=2.0) + + def oserror_run(*_args: object, **_kwargs: object) -> None: + raise OSError("no such file or directory") + + for run_fake in (timeout_run, oserror_run): + with self.subTest(run=run_fake.__name__): + result = VERSION_GATE.default_probe_activation_capability( + {VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic"}, + run=run_fake, + ) + self.assertIsNone(result) + + +class LaunchRuntimeVersionCouplingSeamTest(unittest.TestCase): + """End-to-end (still fully faked) coverage of the seam as wired into + `launch-runtime.py`'s `main()` — the strongest natural enforcement point + per the C4 card, run before any broker registration.""" + + def _run(self, *, probe): + calls: dict[str, object] = {} + + def request(_path: Path, payload: dict[str, object]) -> dict[str, object]: + calls["registered"] = True + calls["request"] = payload + return {"ok": True, "session_id": "a" * 64} + + def execute(command: str, argv: list[str], environment: dict[str, str]) -> None: + calls["executed"] = (command, argv, environment) + + def initialize_generation(_path: Path, _generation: int) -> None: + calls["generation_initialized"] = True + + stderr = io.StringIO() + with redirect_stderr(stderr): + result = LAUNCHER.main( + ["--runtime", "claude", "--", "claude", "--print", "hello"], + environ={"MOSAIC_LEASE_BROKER_SOCKET": "/run/test/broker.sock"}, + request=request, + execute=execute, + initialize_generation=initialize_generation, + probe_activation_capability=probe, + ) + return result, stderr.getvalue(), calls + + def test_matching_activation_version_passes_and_the_gate_proceeds(self) -> None: + result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: matching_capability()) + self.assertEqual(result, 0) + self.assertEqual(stderr_text, "") + self.assertTrue(calls.get("registered")) + self.assertIn("executed", calls) + + def test_version_mismatch_fails_loud_denies_and_never_registers_or_execs(self) -> None: + expected = LAUNCHER.EXPECTED_ACTIVATION_CAPABILITY + mismatched = {"name": expected["name"], "version": expected["version"] + 1} + result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: mismatched) + + self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW) + self.assertNotEqual(result, 0) + self.assertIn("#869", stderr_text) + self.assertIn(f"v{mismatched['version']}", stderr_text) + self.assertIn(f"v{expected['version']}", stderr_text) + self.assertIn("upgrade", stderr_text.lower()) + # Never reaches broker registration or exec — the version gate is a + # hard stop, not advisory. + self.assertNotIn("registered", calls) + self.assertNotIn("executed", calls) + + def test_name_mismatch_fails_loud(self) -> None: + expected = LAUNCHER.EXPECTED_ACTIVATION_CAPABILITY + mismatched = {"name": "some-other-capability", "version": expected["version"]} + result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: mismatched) + + self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW) + self.assertIn("#869", stderr_text) + self.assertIn("some-other-capability", stderr_text) + self.assertNotIn("registered", calls) + self.assertNotIn("executed", calls) + + def test_absent_activation_capability_fails_closed_not_a_silent_pass(self) -> None: + result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: None) + + self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW) + self.assertNotEqual(result, 0) + self.assertIn("#869", stderr_text) + self.assertNotIn("registered", calls) + self.assertNotIn("executed", calls) + + def test_version_gate_runs_before_and_independently_of_broker_registration(self) -> None: + def request_must_not_be_called(*_args: object, **_kwargs: object) -> dict[str, object]: + self.fail("broker must not be contacted when activation version is mismatched") + + stderr = io.StringIO() + with redirect_stderr(stderr): + result = LAUNCHER.main( + ["--runtime", "claude", "--", "claude"], + environ={"MOSAIC_LEASE_BROKER_SOCKET": "/run/test/broker.sock"}, + request=request_must_not_be_called, + probe_activation_capability=lambda *_a, **_kw: None, + ) + self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW) + + def test_dedicated_exit_code_never_collides_with_usage_or_registration_codes(self) -> None: + # Distinctness guard: a version-skew denial must never be mistaken + # for the pre-existing usage error (64) or registration/exec + # fail-closed code (1) this script already owns. + self.assertNotIn(LAUNCHER.EXIT_VERSION_SKEW, (0, 1, 64)) + + +if __name__ == "__main__": + unittest.main()