Revert "feat(pi): wire lazy lease promotion into the mutator gate"

This reverts 939f2e04. Keeping the revert rather than dropping the commit,
because the failed attempt is the most useful record on this branch.

The wiring worked mechanically — verified with a live model on sb-it-1-dt: the
receipt was emitted verbatim as a whole message, and the broker token was minted
AND consumed, so observe_receipt and promote_lease both succeeded and the lease
reached VERIFIED.

It failed as a DESIGN, for reasons that are properties of the protocol rather
than of this wiring:

  * It puts control-plane traffic in the user-facing conversation channel. An
    operator asking "what model are you?" received a receipt string instead of an
    answer — the model tried a tool, was blocked, complied with the receipt
    instruction, and in one-shot mode that text turn BECAME the reply. Observed
    twice, non-deterministically.
  * The lease TTL is hard-capped at 300s (MAX_LEASE_TTL_SECONDS; ttl_seconds >
    cap raises INVALID_LEASE_TTL). Measured: allowed at T+0, LEASE_EXPIRED at
    T+310. So the visible cost recurs every five minutes of mutator activity.
  * Model compliance is not guaranteed — one run retried the command instead of
    emitting the receipt.

Any model emission is user-visible, so this is not fixable by better wiring; it
needs a design answer about how promotion is triggered and paid for. That is
under adversarial review (docs/scratchpads/lease-remediation/07-liveness-design-brief.md
in the operator's repo). Promotion triggering will return on its own branch once
that lands.

What remains here is independently sound and unblocked: harness-home isolation,
the immutable launch record, the skills relocation, the promotion client itself
(steps 1/4/5), and the #1087 prefix guard.

packages/mosaic/framework/tools/_scripts/mosaic-sync-skills

@@ -153,7 +153,24 @@ if [[ $link_only -eq 1 ]]; then
   exit 0
 fi
 
+# Skills are linked into the MOSAIC-OWNED harness homes, never a base install.
+# Paths mirror the config-dir env vars the launcher injects (HARNESS_HOME_ENV in
+# commands/launch.js):
+#   claude   CLAUDE_CONFIG_DIR      -> <home>/skills
+#   pi       PI_CODING_AGENT_DIR    -> <home>/skills   (replaces ~/.pi/agent)
+#   codex    CODEX_HOME             -> <home>/skills
+#   opencode XDG_CONFIG_HOME        -> <home>/opencode/skills  (XDG adds a level)
 link_targets=(
+  "$MOSAIC_HOME/.claude/skills"
+  "$MOSAIC_HOME/.codex/skills"
+  "$MOSAIC_HOME/.opencode/opencode/skills"
+  "$MOSAIC_HOME/.pi/skills"
+)
+
+# Pre-isolation installs planted the same symlink farm directly in the operator's
+# base installs. Those are now orphaned: the launcher no longer reads them, but
+# they persist and make a "clean" base install look mosaic-managed.
+legacy_link_targets=(
   "$HOME/.claude/skills"
   "$HOME/.codex/skills"
   "$HOME/.config/opencode/skills"
@@ -245,13 +262,72 @@ prune_stale_links_in_target() {
     # -m resolves lexical dangling targets too. If resolution fails, ownership
     # is unproven and the link must be preserved.
     resolved="$(readlink -m "$link_path" 2>/dev/null || true)"
-    if [[ -n "$resolved" && "$resolved" == "$canonical_real/"* ]]; then
+    # $canonical_real must be length-checked BEFORE use as a prefix: if it were
+    # ever empty, "$resolved" == "$canonical_real/"* collapses to == "/"* and
+    # matches every absolute path. Combined with the is_mosaic_skill_name skip
+    # above, that inverts the function precisely — it would delete exactly the
+    # FOREIGN symlinks and keep the mosaic ones. (#1087, reported by mos-claude.)
+    if [[ -n "$resolved" && -n "$canonical_real" && "$resolved" == "$canonical_real/"* ]]; then
       rm -f "$link_path"
       echo "[mosaic-skills] Removed stale retired skill link: $link_path"
     fi
   done < <(find "$target_dir" -mindepth 1 -maxdepth 1 -type l -print0)
 }
 
+# Remove mosaic-owned symlinks left in a base install by a pre-isolation sync.
+#
+# Ownership is proven by RESOLUTION, not by name: only links resolving inside the
+# canonical or local skills dirs are removed. Anything else — a real directory, a
+# link elsewhere, an unresolvable link — is left untouched. This mirrors the
+# refusal in commands/skill.js ("only symlinks pointing inside the Mosaic skills
+# directory are managed") and preserves e.g. codex's own `.system` dir.
+#
+# The directory itself is kept: mosaic-doctor warns when ~/.pi/agent/skills is
+# missing, and an empty dir is the correct end state, not an absent one.
+cleanup_legacy_target() {
+  local target_dir="$1"
+  local removed=0 kept=0
+
+  [[ -d "$target_dir" ]] || return 0
+
+  while IFS= read -r -d '' link_path; do
+    local resolved owned=0
+    resolved="$(readlink -m "$link_path" 2>/dev/null || true)"
+
+    # Guard the empty-prefix trap: an unset *_real would make "$resolved" == "/"*
+    # match every absolute path and delete foreign links.
+    if [[ -n "$resolved" ]]; then
+      if [[ -n "$canonical_real" && "$resolved" == "$canonical_real/"* ]]; then
+        owned=1
+      elif [[ -n "$local_real" && "$resolved" == "$local_real/"* ]]; then
+        owned=1
+      fi
+    fi
+
+    if [[ $owned -eq 1 ]]; then
+      rm -f "$link_path"
+      removed=$((removed + 1))
+    else
+      kept=$((kept + 1))
+    fi
+  done < <(find "$target_dir" -mindepth 1 -maxdepth 1 -type l -print0)
+
+  if [[ $removed -gt 0 ]]; then
+    echo "[mosaic-skills] Legacy cleanup: removed $removed mosaic symlink(s) from $target_dir (preserved $kept foreign)"
+  fi
+}
+
+for legacy in "${legacy_link_targets[@]}"; do
+  # Skip anything that is also a current target, so isolation can never
+  # self-destruct if the two lists ever overlap.
+  skip=0
+  for target in "${link_targets[@]}"; do
+    [[ "$legacy" == "$target" ]] && skip=1
+  done
+  [[ $skip -eq 1 ]] && continue
+  cleanup_legacy_target "$legacy"
+done
+
 for target in "${link_targets[@]}"; do
   mkdir -p "$target"
 

packages/mosaic/framework/tools/lease-broker/launch-runtime.py

@@ -8,7 +8,9 @@ import json
 import os
 import socket
 import sys
+import time
 from collections.abc import Callable, Mapping, Sequence
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Final
 
@@ -53,6 +55,48 @@ def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, o
     return value
 
 
+def _self_starttime() -> str | None:
+    """Field 22 of our own /proc stat — the anchor starttime the broker records.
+
+    Read past the comm field's parens, since a process name may contain them.
+    """
+    try:
+        raw = Path(f"/proc/{os.getpid()}/stat").read_text()
+        return raw.rsplit(")", 1)[1].split()[19]
+    except (OSError, IndexError, ValueError):
+        return None
+
+
+def _append_launch_record(environ: Mapping[str, str], record: dict[str, object]) -> None:
+    """Append one NDJSON event to the #797 Runtime Session Ledger.
+
+    `fleet/run/sessions/` is operator-classified in framework-manifest.txt and is
+    already covered by test-upgrade-manifest-guard.sh, so an upgrade can neither
+    overwrite nor prune it. Files 0600 under a 0700 dir, matching what that guard
+    asserts.
+
+    Never raises: a launch must not be denied over bookkeeping. But it also never
+    fails silently — a missing record is exactly the kind of gap that made the
+    2026-08-06 MUTATOR_UNVERIFIED investigation cost a day.
+    """
+    try:
+        mosaic_home = environ.get("MOSAIC_HOME") or str(Path.home() / ".config" / "mosaic")
+        directory = Path(mosaic_home) / "fleet" / "run" / "sessions"
+        directory.mkdir(parents=True, exist_ok=True)
+        os.chmod(directory, 0o700)
+        framed = {
+            "seq": time.time_ns() // 1_000_000,
+            "ts": datetime.now(timezone.utc).isoformat(),
+            **record,
+        }
+        path = directory / "events.ndjson"
+        descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
+        with os.fdopen(descriptor, "w") as handle:
+            handle.write(json.dumps(framed, separators=(",", ":")) + "\n")
+    except (OSError, ValueError, TypeError) as error:
+        print(f"[mosaic] WARNING: launch record not written: {error}", file=sys.stderr)
+
+
 def main(
     argv: Sequence[str] | None = None,
     *,
@@ -94,8 +138,9 @@ def main(
     # silent pass and never folded into the generic registration-failure
     # branch.
     try:
+        activation_capability = probe_activation_capability(source_environment)
         assert_activation_capability_matches(
-            probe_activation_capability(source_environment),
+            activation_capability,
             expected_activation_capability,
         )
     except VersionCouplingError as version_error:
@@ -128,6 +173,32 @@ def main(
         print("Mosaic lease broker registration failed; runtime launch denied.", file=sys.stderr)
         return 1
 
+    # Immutable launch record, half two. `mosaic` wrote `session.launch` with the
+    # config/provenance it knows; only this process knows the broker session id
+    # and the activation capability it just asserted. os.execvpe preserves the
+    # PID, so this PID is BOTH the anchor pid and the join key back to that
+    # record. Never fatal — bookkeeping must not deny a launch — but never
+    # silent either.
+    _append_launch_record(
+        source_environment,
+        {
+            "kind": "lease.register",
+            # Joins back to `mosaic`'s session.launch record. NOT pid: execRuntime()
+            # spawns rather than execs, so this process is a CHILD of mosaic with a
+            # different pid. This pid IS the broker anchor pid (os.execvpe below
+            # preserves it), which is a separate and still-useful fact.
+            "launch_id": source_environment.get("MOSAIC_LAUNCH_ID"),
+            "pid": os.getpid(),
+            "runtime": arguments.runtime,
+            "session_id": session_id,
+            "runtime_generation": generation,
+            "generation_file": str(generation_file),
+            "anchor_starttime": _self_starttime(),
+            "activation_capability": activation_capability,
+            "command": Path(command[0]).name,
+        },
+    )
+
     environment = dict(source_environment)
     environment["MOSAIC_LEASE_SESSION_ID"] = session_id
     environment["MOSAIC_RUNTIME_GENERATION"] = str(generation)

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())

packages/mosaic/src/commands/launch.ts

@@ -14,9 +14,11 @@ import {
   readdirSync,
   realpathSync,
   rmSync,
+  appendFileSync,
 } from 'node:fs';
+import { createHash, randomBytes } from 'node:crypto';
 import { createRequire } from 'node:module';
-import { homedir } from 'node:os';
+import { homedir, hostname } from 'node:os';
 import { join, dirname } from 'node:path';
 import type { Command } from 'commander';
 import {
@@ -42,6 +44,163 @@ const RUNTIME_LABELS: Record<RuntimeName, string> = {
   pi: 'Pi',
 };
 
+// ─── Harness home isolation ──────────────────────────────────────────────────
+// Mosaic-launched runtimes read config from a dedicated home under the mosaic
+// tree — never the operator's base install. A bare `claude` / `pi` therefore
+// keeps its own config AND its own auth, and stays a working break-glass no
+// matter what mosaic does to its own tree.
+//
+// These paths are manifest-UNKNOWN, which resolves to operator ownership
+// (framework-manifest.txt rule 3, #791), so a keep-mode `mosaic update` can
+// neither overwrite nor prune them. Overwrite-mode install still would.
+//
+// opencode has no dedicated config-dir variable and follows XDG, so isolating it
+// sets XDG_CONFIG_HOME for that process tree. That is blunter than the other
+// three: it also relocates XDG lookups for anything opencode spawns.
+const HARNESS_HOME_ENV: Record<RuntimeName, string> = {
+  claude: 'CLAUDE_CONFIG_DIR',
+  pi: 'PI_CODING_AGENT_DIR',
+  codex: 'CODEX_HOME',
+  opencode: 'XDG_CONFIG_HOME',
+};
+
+/** Dedicated mosaic-owned home for a runtime: ~/.config/mosaic/.<runtime> */
+function harnessHome(runtime: RuntimeName): string {
+  return join(MOSAIC_HOME, `.${runtime}`);
+}
+
+/**
+ * Env overlay pointing a runtime at its mosaic-owned home. The directory is
+ * created on demand so a first launch does not fail on a missing path.
+ */
+function harnessEnv(runtime: RuntimeName): Record<string, string> {
+  const key = HARNESS_HOME_ENV[runtime];
+  if (!key) return {};
+  const home = harnessHome(runtime);
+  mkdirSync(home, { recursive: true });
+  return { [key]: home };
+}
+
+// ─── Launch record (immutable provenance) ────────────────────────────────────
+// MANDATORY and MECHANICAL: every launch appends one record of what the agent
+// actually launched with, written before exec. No model involvement, no opt-out.
+//
+// WHY LAUNCH-TIME AND NOT INSPECT-LATER: pi rewrites its own argv to a bare
+// `pi`, so /proc/<pid>/cmdline DESTROYS the launch evidence. That has already
+// produced a confident wrong diagnosis ("this agent bypassed the launcher"),
+// disproved only by the parent process's argv and only because the parent had
+// not yet exited. A record written before exec is the only place this survives.
+//
+// Lands in fleet/run/sessions/ — the #797 Runtime Session Ledger path, already
+// operator-classified in framework-manifest.txt and already covered by
+// test-upgrade-manifest-guard.sh, so an upgrade can neither overwrite nor prune
+// it.
+//
+// CORRELATION is by an explicit MOSAIC_LAUNCH_ID, never by pid: execRuntime()
+// uses spawnSync, so the runtime is a CHILD with a different pid.
+// launch-runtime.py appends the matching `lease.register` event.
+//
+// NEVER records a credential value: env is captured as PRESENT NAMES ONLY, and
+// oversized argv values (the composed system prompt) become a digest + length.
+const LAUNCH_LEDGER_DIR = join(MOSAIC_HOME, 'fleet', 'run', 'sessions');
+
+const CLI_VERSION: string | null = (() => {
+  try {
+    // Resolved RELATIVELY: the package `exports` map does not expose
+    // package.json, so '@mosaicstack/mosaic/package.json' throws
+    // ERR_PACKAGE_PATH_NOT_EXPORTED. Same relative depth from src/ and dist/.
+    return (createRequire(import.meta.url)('../../package.json') as { version: string }).version;
+  } catch {
+    return null;
+  }
+})();
+
+interface NormativeFragmentDigest {
+  source_id: string;
+  sha256: string | null;
+  bytes: number | null;
+  missing?: boolean;
+}
+
+function sha256Of(value: string | Buffer): string {
+  return createHash('sha256').update(value).digest('hex');
+}
+
+/**
+ * Hash the normative sources injected into the agent. This is "what the agent
+ * IS" — and it is the same fragment set the lease broker hashes for promotion,
+ * so an unexpected digest here is a mechanically detectable red flag rather than
+ * a matter of judgement.
+ */
+function normativeFragmentDigests(runtime: RuntimeName): NormativeFragmentDigest[] {
+  const candidates: Array<[string, string]> = [
+    ['CONSTITUTION.md', join(MOSAIC_HOME, 'CONSTITUTION.md')],
+    ['AGENTS.md', join(MOSAIC_HOME, 'AGENTS.md')],
+    ['SOUL.md', join(MOSAIC_HOME, 'SOUL.md')],
+    ['USER.md', join(MOSAIC_HOME, 'USER.md')],
+    ['STANDARDS.md', join(MOSAIC_HOME, 'STANDARDS.md')],
+    ['TOOLS.md', join(MOSAIC_HOME, 'TOOLS.md')],
+    [`runtime/${runtime}/RUNTIME.md`, join(MOSAIC_HOME, 'runtime', runtime, 'RUNTIME.md')],
+  ];
+  return candidates.map(([sourceId, path]) => {
+    try {
+      const bytes = readFileSync(path);
+      return { source_id: sourceId, sha256: sha256Of(bytes), bytes: bytes.length };
+    } catch {
+      return { source_id: sourceId, sha256: null, bytes: null, missing: true };
+    }
+  });
+}
+
+/** argv with oversized values replaced by a digest, so the record stays small
+ * and never inlines injected content verbatim. */
+function redactArgv(argv: string[]): string[] {
+  return argv.map((a) =>
+    typeof a === 'string' && a.length > 256
+      ? `<redacted sha256:${sha256Of(a).slice(0, 16)} bytes:${a.length}>`
+      : a,
+  );
+}
+
+function recordLaunch(runtime: RuntimeName, cliArgs: string[], yolo: boolean): void {
+  try {
+    mkdirSync(LAUNCH_LEDGER_DIR, { recursive: true, mode: 0o700 });
+    // Correlation id for the lease.register half. Set into process.env so it
+    // propagates through every `...process.env` / `...baseEnv` spread below.
+    const launchId = `${Date.now().toString(36)}-${randomBytes(6).toString('hex')}`;
+    process.env['MOSAIC_LAUNCH_ID'] = launchId;
+    const record = {
+      seq: Date.now(),
+      kind: 'session.launch',
+      launch_id: launchId,
+      ts: new Date().toISOString(),
+      host: hostname(),
+      pid: process.pid,
+      runtime,
+      mode: yolo ? 'yolo' : 'normal',
+      cwd: process.cwd(),
+      cli_version: CLI_VERSION,
+      config_home: harnessHome(runtime),
+      config_home_isolated: true,
+      config_home_env: HARNESS_HOME_ENV[runtime] ?? null,
+      argv: redactArgv(cliArgs),
+      normative_fragments: normativeFragmentDigests(runtime),
+      // names only — values are never recorded
+      mosaic_env_present: Object.keys(process.env)
+        .filter((k) => k.startsWith('MOSAIC_'))
+        .sort(),
+    };
+    appendFileSync(join(LAUNCH_LEDGER_DIR, 'events.ndjson'), `${JSON.stringify(record)}\n`, {
+      mode: 0o600,
+    });
+  } catch (err) {
+    // Never block a launch on bookkeeping — but never fail silently either.
+    console.error(
+      `[mosaic] WARNING: launch record not written: ${err instanceof Error ? err.message : String(err)}`,
+    );
+  }
+}
+
 // ─── Pre-flight checks ──────────────────────────────────────────────────────
 
 function checkMosaicHome(): void {
@@ -105,11 +264,11 @@ interface SettingsAudit {
 
 function auditClaudeSettings(): SettingsAudit {
   const warnings: string[] = [];
-  const settingsPath = join(homedir(), '.claude', 'settings.json');
+  const settingsPath = join(harnessHome('claude'), 'settings.json');
   const settings = readJson(settingsPath);
 
   if (!settings) {
-    warnings.push('~/.claude/settings.json not found — hooks and plugins will be missing');
+    warnings.push(`${settingsPath} not found — hooks and plugins will be missing`);
     return { warnings };
   }
 
@@ -561,7 +720,9 @@ function skillRealPath(dir: string): string {
 /** Skill roots Pi auto-discovers natively (no `--skill` needed): its global
  * skills dir and the project-local one relative to the launch cwd. */
 function piNativeSkillRoots(cwd: string = process.cwd()): string[] {
-  return [join(homedir(), '.pi', 'agent', 'skills'), join(cwd, '.pi', 'skills')];
+  // PI_CODING_AGENT_DIR replaces ~/.pi/agent (not ~/.pi), so skills live at
+  // <home>/skills — there is no extra 'agent' segment under the isolated home.
+  return [join(harnessHome('pi'), 'skills'), join(cwd, '.pi', 'skills')];
 }
 
 /** Enumerate skill dirs under a set of roots, deduped by real path. A directory
@@ -764,12 +925,13 @@ function launchRuntime(runtime: RuntimeName, args: string[], yolo: boolean): nev
         cliArgs.push(...args);
       }
       console.log(`[mosaic] Launching ${label}${modeStr}${missionStr}...`);
+      recordLaunch('claude', cliArgs, yolo);
       execLeaseGatedRuntime('claude', cliArgs, process.env, yolo);
       break;
     }
 
     case 'codex': {
-      ensureRuntimeConfig('codex', join(homedir(), '.codex', 'instructions.md'));
+      ensureRuntimeConfig('codex', join(harnessHome('codex'), 'instructions.md'));
       const cliArgs = yolo ? ['--dangerously-bypass-approvals-and-sandbox'] : [];
       if (hasMissionNoArgs) {
         cliArgs.push(missionPrompt);
@@ -777,14 +939,17 @@ function launchRuntime(runtime: RuntimeName, args: string[], yolo: boolean): nev
         cliArgs.push(...args);
       }
       console.log(`[mosaic] Launching ${label}${modeStr}${missionStr}...`);
-      execRuntime('codex', cliArgs);
+      recordLaunch('codex', cliArgs, yolo);
+      execRuntime('codex', cliArgs, { ...process.env, ...harnessEnv('codex') });
       break;
     }
 
     case 'opencode': {
-      ensureRuntimeConfig('opencode', join(homedir(), '.config', 'opencode', 'AGENTS.md'));
+      // opencode follows XDG, so its config resolves to $XDG_CONFIG_HOME/opencode.
+      ensureRuntimeConfig('opencode', join(harnessHome('opencode'), 'opencode', 'AGENTS.md'));
       console.log(`[mosaic] Launching ${label}${modeStr}...`);
-      execRuntime('opencode', args);
+      recordLaunch('opencode', args, yolo);
+      execRuntime('opencode', args, { ...process.env, ...harnessEnv('opencode') });
       break;
     }
 
@@ -799,6 +964,7 @@ function launchRuntime(runtime: RuntimeName, args: string[], yolo: boolean): nev
         cliArgs.push(...args);
       }
       console.log(`[mosaic] Launching ${label}${modeStr}${missionStr}...`);
+      recordLaunch('pi', cliArgs, yolo);
       execLeaseGatedRuntime('pi', cliArgs);
       break;
     }
@@ -835,6 +1001,7 @@ function execLeaseGatedRuntime(
     [launcher, ...dangerousArgs, '--runtime', runtime, '--', runtime, ...args],
     {
       ...baseEnv,
+      ...harnessEnv(runtime),
       MOSAIC_LEASE_BROKER_SOCKET: defaultLeaseBrokerSocket(baseEnv),
       MOSAIC_RUNTIME_GENERATION: baseEnv['MOSAIC_RUNTIME_GENERATION'] ?? '1',
     },