feat(lease-broker): add the missing promotion client

The enforcement half of the lease broker ships and denies; the promotion half
has no production caller anywhere in the package. Verified across 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 mutator-gate denies
every mutator with MUTATOR_UNVERIFIED via a gate that nothing shipped can
satisfy. Runtimes that enforce the gate in-process (pi, via mosaic-extension's
tool_call hook) are bricked for mutators; runtimes whose gate is wired through a
settings hook escape only when that hook is absent — i.e. by being ungated.

This adds the client. It implements protocol steps 1, 4 and 5:

  1. begin_verification  -> mint a challenge, return the exact receipt text
  2. the MODEL emits that text verbatim as its entire latest message
  3. the runtime adapter ships that message to the observer socket
  4. observe_receipt      -> PENDING_PROMOTION
  5. promote_lease        -> VERIFIED

Step 2 is deliberately NOT implemented here, and that is the point.
is_verbatim_receipt uses hmac.compare_digest against the exact minted string —
explicitly "not a transcript substring" — which makes promotion a LIVENESS
PROOF: it 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 specifically looks. Emitting the receipt therefore belongs to the runtime
adapter, where a real model turn happens. A local diagnostic that posts its own
receipt exists in the operator's repo and is deliberately NOT shipped here.

The construction binds the exact normative source bytes, so a VERIFIED lease
means "this agent is running THIS law", not merely "this session id is known".
h_source/h_payload are derived by importing the framework's own
normative_fragments.build_payload rather than reimplementing it: the broker
derives them the same way and any divergence yields PAYLOAD_BINDING_MISMATCH.
There must be exactly one implementation.

session_identity() prefers the generation FILE over the env var, matching
lease_generation.py. Sending a generation higher than the broker's would revoke
the session's own authority (daemon.py:342-344), so it never guesses.

Verified end-to-end on sb-it-1-dt under a real lease-gated anchor: a mutator
denied rc=2 MUTATOR_UNVERIFIED, then begin -> observe -> promote -> VERIFIED,
then the same mutator allowed rc=0. Negative controls pass: a fresh session is
still denied, and an unrelated session still reads UNVERIFIED — promotion is
per-session and does not leak.

Still open: adapter wiring for step 2. Lazy promotion on first mutator attempt
avoids colliding with the Constitution's first-response mode declaration, since
compare_digest requires the receipt to be the WHOLE message.

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

@@ -0,0 +1,269 @@
+#!/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, so a VERIFIED lease
+means "this agent is running THIS law", not merely "this session id is known".
+``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.
+
+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 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. Sources absent on a given
+# deployment are simply not part of the binding — never fabricated.
+FRAGMENT_SOURCES: Final = (
+    "CONSTITUTION.md",
+    "AGENTS.md",
+    "SOUL.md",
+    "USER.md",
+    "STANDARDS.md",
+    "TOOLS.md",
+)
+
+
+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."""
+    sources = list(FRAGMENT_SOURCES) + [f"runtime/{runtime}/RUNTIME.md"]
+    wire_fragments: list[dict[str, str]] = []
+    objects: list[NormativeFragment] = []
+
+    for source_id in sources:
+        try:
+            content = (mosaic_home() / source_id).read_bytes()
+        except OSError:
+            continue
+        import hashlib
+
+        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 RuntimeError("no normative sources found — refusing to build an empty binding")
+
+    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())