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

#!/usr/bin/env python3
"""Lease promotion client — the half the enforcement toolkit never shipped.

The enforcement half (``daemon.py`` + ``mutator-gate.py``) ships and denies. The
promotion half has no production caller anywhere in the package: as of 0.0.48,
0.0.49 and 0.0.50-next.2207, ``begin_verification`` / ``observe_receipt`` /
``promote_lease`` are invoked only by ``broker-test-client.ts``, the acceptance
spec, unit tests, and two probes under ``docs/``. Consequence: **no lease on any
host can reach VERIFIED**, so every mutator is denied ``MUTATOR_UNVERIFIED`` by a
gate nothing can satisfy.

THE PROTOCOL (``daemon.py:578-754``)
------------------------------------
  1. ``begin_verification`` — broker revokes, mints a challenge, and returns the
     exact ``receipt`` text the MODEL must emit
  2. *the model emits that text verbatim as its ENTIRE latest message*
  3. the runtime adapter ships that message to the daemon-owned observer socket
  4. ``observe_receipt``  -> ``PENDING_PROMOTION``
  5. ``promote_lease``    -> ``VERIFIED``

THIS MODULE IMPLEMENTS 1, 4 AND 5 — NEVER 2
-------------------------------------------
Step 2 is the security property, not a formality. ``is_verbatim_receipt`` uses
``hmac.compare_digest`` against the exact minted string — explicitly "not a
transcript substring" (``receipt_challenge.py``). Promotion therefore requires a
live model that received the challenge in its context and echoed it exactly.

``receipt-observer-client.py`` will post ANY string as the latest assistant
message. A promotion client that posted its own receipt would satisfy the broker
while proving nothing — a gate-disabler indistinguishable from a working fix
unless someone looks for it. **This module never posts a receipt.** Emitting it
belongs to the runtime adapter, where a real model turn happens.

The construction binds the exact normative source bytes. ``h_source`` /
``h_payload`` are derived by the framework's own
``normative_fragments.build_payload`` rather than reimplemented: the broker
derives them the same way and any divergence yields ``PAYLOAD_BINDING_MISMATCH``.
There must be exactly one implementation.

WHAT THE BINDING DOES *NOT* PROVE
---------------------------------
It is tempting to read a VERIFIED lease as "this agent is running THIS law".
**It does not mean that**, and writing it down that way is how the belief spread.
The broker holds no reference copy of any normative source and never opens one;
it recomputes ``h_source`` / ``h_payload`` from the fragment bytes THIS CLIENT
sent and compares them to the binding THIS CLIENT sent (``daemon.py:602-616``).
Both sides of that comparison originate here, so it detects corruption in
transit and nothing else. What the binding actually asserts is "the client
claims these bytes, self-consistently".

Making it mean the stronger thing requires the broker to re-read the on-disk
sources itself, against a manifest the agent cannot rewrite — i.e. broker code
attestation under its own uid. Until then, do not cite a VERIFIED lease as
evidence of law integrity.

Usage
-----
    lease_promote.py --begin                 # prints the receipt the MODEL must emit
    lease_promote.py --complete <challenge>  # after the adapter observed it
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import os
import socket
import sys
from pathlib import Path
from typing import Final

# Isolated (`python -I`) adapter invocations must still import co-located
# framework modules; never depend on the caller's PYTHONPATH.
_MODULE_DIRECTORY = str(Path(__file__).resolve().parent)
if _MODULE_DIRECTORY not in sys.path:
    sys.path.insert(0, _MODULE_DIRECTORY)

from normative_fragments import NormativeFragment, build_payload  # noqa: E402

MAX_FRAME: Final = 64 * 1024
BROKER_TIMEOUT_SECONDS: Final = 3.0
SCHEMA_VERSION: Final = 1
MANIFEST_VERSION: Final = 1
GENERATOR_VERSION: Final = "mosaic/lease_promote@1"
DEFAULT_TTL_SECONDS: Final = 300

# Normative sources whose exact bytes bind the lease, in binding order. Order is
# load-bearing: ``h_source`` frames the resolved sequence, so reordering changes
# the derivation. Never fabricate a source that is not on disk.
FRAGMENT_SOURCES: Final = (
    "CONSTITUTION.md",
    "AGENTS.md",
    "SOUL.md",
    "USER.md",
    "STANDARDS.md",
    "TOOLS.md",
)

# Framework-owned sources, reconciled on every upgrade — `install.sh:76`
# FRAMEWORK_OWNED and `config/file-adapter.ts` FRAMEWORK_OWNED_FILES — plus the
# per-runtime contract shipped under `framework/runtime/<runtime>/`. A deployment
# missing one of these is broken, not minimal, so their absence is refused rather
# than silently dropped from the binding.
#
# SOUL.md and USER.md are deliberately excluded: install.sh does not seed them
# ("intentionally NOT seeded here — they are generated by `mosaic init`"), so a
# fresh install legitimately lacks both. TOOLS.md is user-seeded on first install
# only. Absence of those three is reported, not fatal.
REQUIRED_SOURCES: Final = frozenset({"CONSTITUTION.md", "AGENTS.md", "STANDARDS.md"})


class IncompleteBinding(RuntimeError):
    """A source that must bind this lease could not be read.

    **Never downgrade this to a skip.** The broker recomputes the hashes from the
    fragments it is sent, so an omitted fragment is internally consistent and
    ``PAYLOAD_BINDING_MISMATCH`` cannot fire — a partial law promotes exactly like
    a complete one, and nothing downstream can tell the difference. Dropping an
    unreadable source therefore does not degrade the binding, it forges a smaller
    one. Fail here, where the omission is still visible.
    """


def mosaic_home() -> Path:
    return Path(os.environ.get("MOSAIC_HOME") or Path.home() / ".config" / "mosaic")


def broker_socket() -> Path:
    value = os.environ.get("MOSAIC_LEASE_BROKER_SOCKET")
    if value:
        return Path(value)
    runtime_dir = os.environ.get("XDG_RUNTIME_DIR")
    if runtime_dir:
        return Path(runtime_dir) / "mosaic-lease" / "broker.sock"
    return Path(f"/run/user/{os.getuid()}/mosaic-lease/broker.sock")


def session_identity() -> tuple[str, int, str]:
    """Session id, CURRENT generation, runtime.

    The generation file wins over the env var, matching ``lease_generation.py``.
    Sending a generation HIGHER than the broker's would revoke this session's own
    authority (``daemon.py:342-344``), so this never guesses.
    """
    session_id = os.environ["MOSAIC_LEASE_SESSION_ID"]
    runtime = os.environ["MOSAIC_LEASE_RUNTIME"]
    state_file = os.environ.get("MOSAIC_LEASE_GENERATION_FILE")
    if state_file:
        try:
            return session_id, int(Path(state_file).read_text().strip()), runtime
        except (OSError, ValueError):
            pass
    return session_id, int(os.environ["MOSAIC_RUNTIME_GENERATION"]), runtime


def build_construction(runtime: str) -> tuple[dict[str, object], object]:
    """Assemble the wire construction and derive its hashes with the sole builder."""
    runtime_contract = f"runtime/{runtime}/RUNTIME.md"
    sources = list(FRAGMENT_SOURCES) + [runtime_contract]
    required = REQUIRED_SOURCES | {runtime_contract}
    wire_fragments: list[dict[str, str]] = []
    objects: list[NormativeFragment] = []
    absent: list[str] = []

    for source_id in sources:
        try:
            content = (mosaic_home() / source_id).read_bytes()
        except FileNotFoundError:
            # Genuinely not on disk. Legitimate only for operator-owned sources.
            if source_id in required:
                raise IncompleteBinding(
                    f"required normative source is absent: {source_id}"
                ) from None
            absent.append(source_id)
            continue
        except OSError as exc:
            # The path resolves but will not read — EACCES, EIO, EISDIR, ELOOP.
            # That is an anomaly for EVERY source, optional ones included: an
            # unreadable file is not an un-configured one, and treating it as
            # absent is what lets a permission change quietly shrink the law.
            raise IncompleteBinding(
                f"normative source is present but unreadable: {source_id} "
                f"({type(exc).__name__})"
            ) from exc

        digest = hashlib.sha256(content).hexdigest()
        wire_fragments.append(
            {
                "source_id": source_id,
                "content_base64": base64.b64encode(content).decode("ascii"),
                "expected_sha256": digest,
            }
        )
        objects.append(NormativeFragment(source_id, content, digest))

    if not wire_fragments:
        raise IncompleteBinding("no normative sources found — refusing an empty binding")

    # Absence is legitimate here but never invisible. The omission is already
    # baked into h_source (the framed source sequence differs), but nothing
    # compares h_source to an expected value, so this line is the only place a
    # human learns the binding was narrower than the full set.
    if absent:
        print(
            f"lease_promote: binding omits absent operator sources: {', '.join(absent)}",
            file=sys.stderr,
        )

    result = build_payload(
        manifest_version=MANIFEST_VERSION,
        generator_version=GENERATOR_VERSION,
        fragments=objects,
    )
    if result.injectionDecision != "ACCEPTED" or not result.promotion:
        raise RuntimeError(f"construction refused locally: {result.source_reason}")

    return (
        {
            "manifest_version": MANIFEST_VERSION,
            "generator_version": GENERATOR_VERSION,
            "fragments": wire_fragments,
        },
        result,
    )


def broker_request(payload: dict[str, object]) -> dict[str, object]:
    raw = (json.dumps(payload, separators=(",", ":")) + "\n").encode()
    if len(raw) > MAX_FRAME:
        raise ValueError(
            f"request too large ({len(raw)} bytes); broker frame cap is {MAX_FRAME}"
        )
    response = bytearray()
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
        connection.settimeout(BROKER_TIMEOUT_SECONDS)
        connection.connect(str(broker_socket()))
        connection.sendall(raw)
        connection.shutdown(socket.SHUT_WR)
        while len(response) <= MAX_FRAME:
            chunk = connection.recv(4096)
            if not chunk:
                break
            response.extend(chunk)
    if len(response) > MAX_FRAME or not response.endswith(b"\n"):
        raise ValueError("invalid broker reply")
    value = json.loads(response)
    if not isinstance(value, dict):
        raise ValueError("invalid broker reply")
    return value


def begin(
    ttl_seconds: int = DEFAULT_TTL_SECONDS,
    compaction_epoch: int = 0,
    request_epoch: int = 0,
) -> dict[str, object]:
    """Step 1. Returns the broker reply, including the exact ``receipt`` text."""
    session_id, generation, runtime = session_identity()
    construction, derived = build_construction(runtime)
    return broker_request(
        {
            "action": "begin_verification",
            "session_id": session_id,
            "runtime_generation": generation,
            "runtime": runtime,
            "ttl_seconds": ttl_seconds,
            "binding": {
                "compaction_epoch": compaction_epoch,
                "request_epoch": request_epoch,
                "h_source": derived.h_source,
                "h_payload": derived.h_payload,
                "schema_version": SCHEMA_VERSION,
            },
            "construction": construction,
        }
    )


def complete(challenge: str) -> dict[str, object]:
    """Steps 4-5. Assumes the model already emitted the receipt and the adapter
    shipped it to the observer socket."""
    session_id, generation, _ = session_identity()
    observed = broker_request(
        {
            "action": "observe_receipt",
            "session_id": session_id,
            "runtime_generation": generation,
            "receipt_challenge": challenge,
        }
    )
    if observed.get("ok") is not True or observed.get("state") != "PENDING_PROMOTION":
        return {"stage": "observe_receipt", **observed}
    promoted = broker_request(
        {
            "action": "promote_lease",
            "session_id": session_id,
            "runtime_generation": generation,
            "receipt_challenge": challenge,
        }
    )
    return {"stage": "promote_lease", **promoted}


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Mosaic lease promotion client.")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
        "--begin",
        action="store_true",
        help="mint a challenge; prints the receipt the MODEL must emit verbatim",
    )
    group.add_argument(
        "--complete",
        metavar="CHALLENGE",
        help="observe the emitted receipt and promote the lease",
    )
    parser.add_argument("--ttl-seconds", type=int, default=DEFAULT_TTL_SECONDS)
    arguments = parser.parse_args(argv)

    try:
        if arguments.begin:
            print(json.dumps(begin(ttl_seconds=arguments.ttl_seconds), indent=2))
        else:
            print(json.dumps(complete(arguments.complete), indent=2))
    except KeyError as exc:
        print(f"missing lease environment: {exc}; not a lease-gated session", file=sys.stderr)
        return 2
    except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
        print(f"{type(exc).__name__}: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())