fix(mosaic): register canonical framework skills

docs/PRD.md

@@ -79,6 +79,29 @@ Jarvis (v0.2.0) is a self-hosted AI assistant with a Python FastAPI backend and
 
 ---
 
+## Compaction Refresh Trust Lifecycle (M1, #827–#830)
+
+### Problem and objective
+
+Context compaction, session replacement, and same-PID runtime reloads can leave a previously VERIFIED runtime lease attached to stale directives. M1 must revoke that authority mechanically for Claude (including Claudex) and Pi without trusting caller-asserted identity or forking the external broker state machine.
+
+### Requirements
+
+1. `CR-REQ-01`: Claude `PreCompact` and `SessionStart` with matcher `compact`, plus Pi `session_before_compact` and the first post-`session_compact` `context`, SHALL independently revoke the active broker lease.
+2. `CR-REQ-02`: Runtime generation increases—including same-PID Pi reload/new/resume/fork and Claude resume/clear—SHALL monotonically replace the prior broker incarnation and inherit no VERIFIED lease.
+3. `CR-REQ-03`: A fired observer that cannot confirm broker revocation SHALL fail closed through lifecycle cancellation, a private local generation fence, and/or a runtime-local tool latch. The existing all-tools broker gate remains authoritative.
+4. `CR-REQ-04`: The lease TTL SHALL remain monotonic and capped at 300 seconds. If both observers are missed, within-TTL consequential actions remain allowed and after-TTL actions are denied. This named bounded residual stale window SHALL be documented without claiming a mutator-action bound inside the window.
+5. `CR-REQ-05`: Hook descendants SHALL use the broker-minted session and owner-only current-generation state inherited from register-before-exec. Caller-minted sessions and parallel lease state machines remain forbidden.
+
+### Acceptance criteria
+
+1. `AC-CR-01`: Real-socket tests prove each Claude observer revokes, Pi lifecycle tests prove both observer paths, and Claudex isolated settings preserve and install the mandatory hooks.
+2. `AC-CR-02`: A same-PID generation test proves the old generation is stale and the replacement generation is UNVERIFIED across reload/resume/fork-equivalent lifecycle events.
+3. `AC-CR-03`: RED-first T12b/T30 evidence explicitly reports dual-hook miss within TTL as **ALLOWED** and after TTL as **DENIED**.
+4. `AC-CR-04`: Attributable executable coverage is at least 85%, the full repository suite is green on deterministic main, and independent code/security review completes before merge.
+
+---
+
 ## Fleet Declarative Configuration Management Workstream (FCM, #758)
 
 ### Problem and objective

docs/SITEMAP.md

@@ -3,9 +3,11 @@
 ## Compaction refresh lease broker
 
 - [Internal broker protocol](architecture/lease-broker-protocol.md) — kernel identity, ancestry and generation invariants, framed requests, responses, and persisted cycle bindings.
-- [Broker operations](guides/lease-broker-operations.md) — protected paths, startup, fail-closed recovery posture, distinct-principal deployment, and residual risk.
+- [Broker operations](guides/lease-broker-operations.md) — protected paths, startup, constrained recovery, fail-closed posture, distinct-principal deployment, and residual risk.
+- [Constrained recovery skill](../packages/mosaic/framework/skills/mosaic-context-refresh/SKILL.md) — source-resident thin wrapper, receipt scope, C4 replay boundary, and T-C middle-drop disclosure.
 - [Lease-broker security notes](architecture/lease-broker-security.md) — identity, whole-class authorization, threat boundaries, and coordinator review requirements.
 - [Whole mutator-class gate](architecture/mutator-class-gate.md) — default-deny policy, revoke-first/promote-last state machine, TTL, runtime adapters, and T-B/T-C assurance boundary.
+- [Compaction revocation lifecycle](architecture/compaction-revocation.md) — Claude/Pi observer matrix, same-PID generation rollover, failure fencing, and the named bounded residual stale window.
 
 ## CLI and skill management
 

docs/architecture/compaction-revocation.md

@@ -0,0 +1,43 @@
+# Compaction observer revocation and runtime generations
+
+WI-3 connects Claude and Pi compaction/session lifecycle events to the existing authenticated lease-broker state machine. It does not add a second lease store or let runtime hooks assert identity. Each observer inherits the broker-minted session, resolves the current private runtime generation, and sends the existing `revoke_lease` action over the authenticated Unix socket.
+
+## Observer matrix
+
+| Runtime          | Lifecycle signal                                                 | Action                                                                                                                       |
+| ---------------- | ---------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------- |
+| Claude / Claudex | `PreCompact`                                                     | Revoke the current lease before compaction. A non-zero hook result blocks the lifecycle transition.                          |
+| Claude / Claudex | `SessionStart` with matcher `compact`                            | Revoke again after compacted context starts.                                                                                 |
+| Claude / Claudex | `SessionStart` with matcher `resume\|clear`                      | Atomically advance the private generation, then revoke the replacement incarnation.                                          |
+| Pi               | `session_before_compact`                                         | Revoke before compaction; return `{ cancel: true }` if revocation cannot be confirmed.                                       |
+| Pi               | `session_compact` then the first `context`                       | Arm and run an independent post-compaction revoke. A failed post observer blocks later tools locally until a retry succeeds. |
+| Pi               | `session_start` with reason `reload`, `new`, `resume`, or `fork` | Atomically advance the private generation, then revoke the replacement incarnation before reuse.                             |
+
+The first observer that reaches the broker deletes pending promotion tokens and makes the lease `UNVERIFIED`. The second compaction observer is deliberate redundancy, not a prerequisite for the first. Claudex receives the same mandatory hooks in its isolated `CLAUDE_CONFIG_DIR`; hook merging preserves unrelated isolated settings and rejects malformed or symlinked settings fail-closed.
+
+## Private generation authority
+
+`launch-runtime.py` still registers before `exec`, preserving the kernel-authenticated PID/starttime anchor. It now also creates `generation-<broker-session>.state` beside the broker socket. The file is owner-only mode `0600` under the broker's mode-`0700` directory. Hook descendants read that file instead of relying only on an immutable inherited environment value.
+
+Generation changes use an exclusive file lock, validate owner/type/mode/size, increment monotonically, truncate and write the complete new value, and `fsync` before contacting the broker. Therefore reload, new-session, resume, and fork events may retain the same PID/starttime while still becoming a new broker incarnation. The higher generation causes the broker to atomically discard prior tokens and lease authority; the replacement generation inherits no VERIFIED lease.
+
+If an observer fires while broker transport is unavailable, `revoke-lease.py` advances the private generation as a local fence before returning non-zero. Every later all-tools gate reads that higher value. When the broker is reachable again, authentication of that value performs the same old-generation revocation before authorization. Pi also keeps a process-local post-compaction/rollover failure latch that blocks tool calls. An unsafe or unreadable generation file itself makes both lifecycle revocation and tool authorization fail closed.
+
+## Threat contract and stopping boundary
+
+### BOUNDED RESIDUAL STALE WINDOW
+
+If **both** pre- and post-compaction observers are missed entirely, no revocation signal exists. During the remaining unexpired lease, **within-TTL consequential actions are allowed**. Their count and timing are **bounded by lease expiry, not by the mutator gate**. WI-3 makes no claim that it bounds mutator actions inside this stale interval. The broker's monotonic lease TTL is capped at 300 seconds; after expiry, the next consequential tool is denied with `LEASE_EXPIRED`.
+
+This is the named D2-v5 T-A residual. It is distinct from an observer that fires but cannot contact the broker: the latter creates a local generation fence and fails closed. It is also distinct from T-C total rot, where the lifecycle observers and the all-tools gate are both absent or replaced. Server-side branch protection, required CI, and independent review remain the irreducible backstop for T-C.
+
+| Condition                                                       | Result                                                                                                                |
+| --------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------- |
+| Either compaction observer succeeds                             | Existing lease and pending promotion tokens are revoked immediately.                                                  |
+| Observer runs but broker confirmation fails                     | Lifecycle transition is denied where supported; local generation fence and runtime latch prevent inherited authority. |
+| Both observers are missed, lease unexpired                      | **ALLOWED** inside the bounded residual stale window. No within-window mutator bound is claimed.                      |
+| Both observers are missed, lease expired                        | **DENIED** by monotonic TTL expiry.                                                                                   |
+| Generation advances on reload/new/resume/fork                   | Prior incarnation revoked; replacement starts `UNVERIFIED`.                                                           |
+| Lifecycle observers and all-tools gate both fail or are removed | T-C total-hook-miss residual; protected-branch controls remain required.                                              |
+
+The permanent T12b/T30 acceptance case prints both required outcomes: dual-hook miss within TTL is **ALLOWED**, and the same lease after TTL is **DENIED**. Separate real-socket tests prove each Claude observer and same-PID generation rollover; Pi lifecycle tests exercise pre/post observers, all four replacement reasons, and local failure closure.

docs/architecture/lease-broker-protocol.md

@@ -13,12 +13,17 @@ Each connection carries exactly one UTF-8 JSON object followed by one newline, c
 - `mint_token`: authenticated identity plus `binding` containing exactly `compaction_epoch`, `request_epoch`, `h_source`, `h_payload`, and `schema_version`.
 - `consume_token`: authenticated identity plus `token`.
 - `begin_verification`: authenticated identity, runtime (`claude` or `pi`), cycle `binding`, and a TTL no greater than 300 seconds. The broker revokes existing authority first, enters `PENDING_VERIFICATION`, and returns a single-use promotion token.
+- `begin_recovery`: the constrained recovery entrypoint. It rejects caller-provided receipt/challenge fields and delegates to the same `begin_verification` transition, but reports `PENDING_DELIVERY` and marks the volatile cycle as recovery-owned.
+- `complete_recovery`: authenticated identity only. It rejects caller-provided receipt/challenge fields, obtains the current recovery challenge only from broker state, and delegates to the same trusted-observer → evidence → consume → promote sequence. An observation failure revokes recovery authority; retry starts a fresh challenge.
+
+The daemon owns a second protected production observer socket (mode `0600`) unless a private `--test-observer-file` fixture is selected. That transport accepts only the exact `record_runtime_observation` schema after kernel `SO_PEERCRED` plus the existing anchor/ancestry authentication; it validates the pending runtime/generation before storing one finalized assistant entry for the in-process `RuntimeReceiptObserver`. It is **not** a broker request action. Claude sends its latest assistant entry from the Stop-hook transport; Pi sends only finalized `message_end` assistant content. The public broker socket continues to reject request-supplied `latest_assistant_message` in begin, observe, and complete paths.
+
 - `promote_lease`: authenticated identity plus the exact pending promotion token. The broker commits token consumption before making `VERIFIED` visible.
-- `revoke_lease`: authenticated observer signal; deletes pending tokens and makes the session `UNVERIFIED` immediately.
+- `revoke_lease`: authenticated observer signal; deletes pending tokens and makes the session `UNVERIFIED` immediately. WI-3 Claude/Pi hooks send this existing action; `runtime` and bounded `reason` fields are diagnostic input only and never identity authority.
 - `authorize_tool`: authenticated identity, runtime, and exact runtime-reported tool name. The broker returns an explicit allow/deny decision from the whole-class policy and current lease.
 
-A higher generation for the same anchor atomically replaces the stored incarnation and deletes all prior tokens and lease authority for that session. A lower generation is stale. Tokens are 256-bit values from the operating-system cryptographic RNG and are single use. At most 256 pending tokens may be persisted; another mint fails with `TOKEN_CAPACITY` before mutation. Successful consumption deletes the token, while a replay still fails with `TOKEN_REPLAY`. Live v1 token records retain the existing `consumed: false` schema.
+A higher generation for the same anchor atomically replaces the stored incarnation and deletes all prior tokens and lease authority for that session. A lower generation is stale. Runtime descendants resolve the current generation from an owner-only, locked generation file created by the register-before-exec launcher; reload/new/resume/fork observers advance and `fsync` it before broker revocation. This supports generation replacement even when PID/starttime do not change. Tokens are 256-bit values from the operating-system cryptographic RNG and are single use. At most 256 pending tokens may be persisted; another mint fails with `TOKEN_CAPACITY` before mutation. Successful consumption deletes the token, while a replay still fails with `TOKEN_REPLAY`. Live v1 token records retain the existing `consumed: false` schema.
 
-VERIFIED leases are volatile and monotonic-time bounded: broker restart, generation change, explicit observer revocation, or expiry returns the session to `UNVERIFIED`. `begin_verification` always revokes before minting a new prerequisite. `promote_lease` is valid only from the matching pending cycle; persistence failure rolls token and lease state back, while post-rename durability uncertainty terminates the broker. The WI-1 token is the atomic promotion prerequisite substrate. A later receipt implementation must satisfy that prerequisite but cannot replace the mechanical mutator gate as safety authority.
+VERIFIED leases are volatile and monotonic-time bounded: broker restart, generation change, explicit observer revocation, or expiry returns the session to `UNVERIFIED`. `begin_verification` always revokes before minting a new prerequisite. `begin_recovery` reuses that exact transition and mints a new challenge, so a normal-path receipt/challenge cannot be replayed through recovery. `promote_lease` is valid only from the matching pending cycle; persistence failure rolls token and lease state back, while post-rename durability uncertainty terminates the broker. The WI-1 token is the atomic promotion prerequisite substrate. Receipt evidence is a T-A delivery/liveness prerequisite only; it cannot replace the mechanical mutator gate as safety authority. A tail-preserving middle drop is not receipt-detectable and remains the disclosed T-C/server-side residual.
 
 State replacement serializes and enforces the 4 MiB maximum before opening a temporary file, then uses a mode-`0600` temporary file, `fsync`, atomic rename, and parent-directory `fsync`. Every broker mutation snapshots the prior v1 state. A commit failure before rename restores that snapshot and leaves durable state unchanged. A failure after rename makes durability uncertain, so the store is poisoned without rolling memory back and the daemon terminates rather than serving with divergent state. Existing state is opened without following symlinks, must be a bounded regular file at mode `0600`, and is fully schema- and invariant-validated before use. Persisted tokens must be unconsumed, match their session's current generation, and remain within the 256-token cap. Session identity is uniquely keyed by `(anchor_pid,anchor_starttime)`; duplicate logical sessions for one anchor refuse startup. State integrity or mode failures refuse startup. The daemon does not log session IDs or tokens.

docs/architecture/lease-broker-security.md

@@ -2,13 +2,25 @@
 
 - Trusted identity comes only from Linux `SO_PEERCRED` plus `/proc` starttime, never request identity fields.
 - Descendant authorization is anchored to `(pid,starttime)` and uses a complete second starttime pass to fail closed on disappearance or PID-reuse races.
-- Runtime generations are monotonic per anchor; a bump revokes prior-incarnation tokens before persistence commits.
+- Runtime generations are monotonic per anchor; a bump revokes prior-incarnation tokens before persistence commits. WI-3 stores the live generation in an owner-only locked file so same-PID Pi reload/new/resume/fork and Claude resume/clear transitions cannot inherit a VERIFIED lease.
 - Session IDs and cycle tokens use the OS cryptographic RNG. `Math.random` and model output are not token sources.
 - Framing and persistence failures fail closed. Sensitive tokens are not logged.
 - Built-in `0700`/`0600` filesystem modes provide same-principal hardening only, not socket authenticity against the same UID. WI-1 provides no distinct-principal isolation. That stronger deployment requires an external protected proxy, ACL, or service boundary, and the boundary must preserve authenticated client identity for the broker's `SO_PEERCRED` and ancestry authorization rather than substituting a shared proxy identity.
 - WI-2 whole-class authorization denies every consequential, unknown, and custom tool while UNVERIFIED; it does not inspect shell strings or trust wrapper selection. First-class Claude/Pi, both Claudex dispatch modes, PRDY, QA remediation, coord, orchestrator, and fleet starts converge on broker register-before-exec; Claudex additionally installs the mandatory all-tools hook inside its preserved isolated config and fails closed on unsafe settings.
 - The permanent `check-runtime-launches.py` suite/CI guard scans production source for direct literal, absolute-path, process-API, command-array, and dynamic Claude/Pi launches. It has no bypass allowlist: an unrecognized launch form fails CI until routed through the common boundary.
 - WI-2 promotion consumes a WI-1 cycle token before VERIFIED becomes visible. Observer revocation, runtime-generation replacement, broker restart, and monotonic TTL expiry remove authority.
-- Receipt observation, payload construction, compaction observers, and constrained recovery implementation remain later surfaces. A receipt can become a promotion prerequisite but is never the safety mechanism.
+- WI-3 wires redundant Claude `PreCompact`/`SessionStart(compact)` and Pi `session_before_compact`/post-`session_compact` `context` observers to that same revoke action. If broker confirmation fails after an observer fires, the revoker advances the private generation as a local fence; subsequent authorization revokes the stale broker incarnation before any consequential allow.
+- Dual observer absence while a lease remains live is the named **bounded residual stale window**: consequential tools remain allowed until monotonic expiry, with no claimed within-window action bound. After expiry they are denied. Total observer-plus-gate absence remains T-C.
+- Receipt observation, payload construction, and constrained recovery implementation remain later surfaces. A receipt can become a promotion prerequisite but is never the safety mechanism.
+
+## Named residual: promote-lease-lost-ACK (WI-3 D2-v5)
+
+A valid `promote_lease` can leave a session `VERIFIED` in the broker while the client never learns of it. This is a named, bounded D2-v5 T-A residual — an **authority-observability divergence, not an authority divergence, not an ALLOW-risk, and not a retry double-apply**. It is disclosed here, not laundered.
+
+**Window — where it can occur.** The broker commits token consumption and durable `VERIFIED` state _before_ the success reply becomes visible (see the promotion order in `lease-broker-protocol.md`). The residual is confined to the interval after that commit+fsync when the broker→client reply or peer-ACK is lost — for example an extreme-contention send failure or peer disconnect after `handle()` has already mutated and persisted state (the #838 fail-closed transport path). The lease mutation is already durable broker-side; only the acknowledgement to the client is lost. No uncommitted or partially-applied state is involved: the commit either happened (and is authoritative) or it did not (and no lease exists).
+
+**Fail-safe direction — the client can only under-claim.** Broker intent is the ceiling; client authority is always ≤ broker intent, never more. Client-side authority-belief is granted only by a _received_ acknowledgement; a lost acknowledgement conveys nothing, so the client cannot conclude "verified" and continues to treat itself as `UNVERIFIED` (it re-verifies or recovers). If the client retries `promote_lease` with the same token, the token is already consumed and the broker rejects the retry (`PROMOTION_TOKEN_MISMATCH` / `INVALID_LEASE_TRANSITION`); there is no double-apply. The committed `VERIFIED` state the broker holds is authority the lease _legitimately earned_ from a real promotion — the broker authorizing consequential tools under it is correct, not inflation. Divergence is therefore strictly toward _less_ client authority than the broker granted; it never produces authority the broker did not grant.
+
+**Bound — TTL plus the observer/gen-bump revoke backstop, self-healing.** The orphaned `VERIFIED` lease is indistinguishable to the broker from any other legitimately verified lease, so the identical D2-v5 revocation backstops dispose of it: any compaction observer (`PreCompact` / `SessionStart(compact)` for Claude; `session_before_compact` / post-`session_compact` `context` for Pi), any same-PID runtime-generation bump (reload/new/resume/fork), broker restart, or monotonic-time expiry returns the session to `UNVERIFIED`. Monotonic TTL expiry (capped at 300 seconds) is **unconditional** — it requires no observer at all — so the maximum exposure of the orphaned lease is one TTL, ≤ 300 s, after which the next consequential tool is denied with `LEASE_EXPIRED`. Any observer that fires shortens the window further. The residual self-heals: "≥1 observer fires OR expiry ⇒ revoke" catches the lost-ACK lease on the same terms as every other stale lease. As with the dual-observer-miss stale window, WI-3 makes no claim that the mutator gate bounds actions inside the residual interval; the interval is bounded by TTL and the revoke backstop, and the server-side branch-protection / required-CI / independent-review line remains the irreducible backstop for protected-repository mutations.
 
 Coordinator security review must rerun the real socket/peercred and mutator-gate acceptance suites on an unrestricted Linux runner and obtain the mandated independent Opus-SECREV review before integration.

docs/architecture/mutator-class-gate.md

@@ -21,13 +21,13 @@ The authenticated broker is the sole lease writer:
 3. Token consumption commits before the volatile lease becomes VERIFIED. Promotion is last and cannot be reached directly from UNVERIFIED.
 4. `revoke_lease`, a runtime-generation increase, broker restart, or monotonic expiry removes mutator authority.
 
-The initial TTL is capped at the ratified 300-second maximum. A caller may request a shorter positive TTL but cannot lengthen the maximum. Dual compaction-hook miss within an unexpired lease remains the ratified bounded T-A residual; once either observer revokes or TTL expires, the next consequential tool is denied.
+The initial TTL is capped at the ratified 300-second maximum. A caller may request a shorter positive TTL but cannot lengthen the maximum. WI-3 installs the [compaction observer and generation lifecycle](compaction-revocation.md). Dual compaction-hook miss within an unexpired lease remains the ratified bounded T-A residual: consequential tools are allowed until expiry, with no claimed within-window action bound; once either observer revokes or TTL expires, the next consequential tool is denied.
 
 A receipt is only a future promotion prerequisite. It is not an obedience, residency, or safety proof and never replaces this mechanical gate.
 
 ## Runtime adapters
 
-`launch-runtime.py` registers itself with the broker and then `exec`s Claude or Pi so PID/starttime remain the authenticated parent anchor. It exports only the broker-minted session ID and current generation to descendants.
+`launch-runtime.py` registers itself with the broker and then `exec`s Claude or Pi so PID/starttime remain the authenticated parent anchor. It exports the broker-minted session ID and an owner-only generation-file reference to descendants; lifecycle hooks advance that file for same-PID replacement generations.
 
 - Claude installs `mutator-gate.py` as an all-tools (`.*`) `PreToolUse` hook.
 - `mosaic claudex` and `mosaic yolo claudex` preserve their isolated `CLAUDE_CONFIG_DIR`, merge the mandatory hook into that isolated `settings.json`, and use the same register-before-exec launcher. Malformed or symlinked isolated settings deny launch.

docs/compaction-refresh/governance/GATE0-PROBE3-EXEC-AMENDMENT.md

@@ -1,109 +0,0 @@
-# Gate0 Charter Amendment — Probe-3 (D4) Evidence-Class EXECUTION Authorization
-
-**Status:** DRAFT (revision 2) — supersedes prior candidate pin `8fca5b7a`; pending Mos re-scope-verify BEFORE
-re-pin is ratified. Revised per Mos adjudication directive (2026-07-18): §3 = D4-focused Option A; §1 = explicit binding model.
-**Amendment type:** Additive durable-authority amendment (does NOT mutate any ratified pinned doc in place).
-**Milestone:** 188 — Compaction-Refresh Mechanism · **Issue:** Gitea #827 (WI-0 Gate0) · gates WI-3 #830 merge.
-
----
-
-## 1. Authorized source (why this amendment is valid)
-
-Per the Mosaic governance principle (OpenBrain `f0cb61b6`, permanent; fail-closed / DO-178C): a durable
-Gate0 execution-hold clears ONLY by **(i) explicit user (owner) authorization** OR **(ii) a durable-authority
-amendment from an authorized source**. A runtime coordinator's verbal GO does NOT qualify.
-
-- **Owner authorization (i):** Jason (owner / north-star) directed in `#mos` **2026-07-18 17:28Z** — *"amend the
-  charter"* (option **B** from the WI-3 escalation), quoting the escalation back. This is the direct-user
-  authorization artifact that homelab's durable-Gate0 objection (2026-07-18 06:34:56Z) explicitly named as a
-  valid resolution condition (*"user artifact OR auth amendment"*). Relayed + authenticity-adjudicated by Mos
-  (`web1:mos-claude`). Owner outranks any peer-lane durable authority.
-- **Amendment (ii):** this document, authorized BY that owner directive.
-
-Both qualifying conditions are therefore satisfied. Homelab's stated condition is met by the owner artifact;
-homelab is NOTIFIED as a transparency step (it holds durable-Gate0 standing and raised the original catch),
-but its co-sign is not a gate (owner > peer).
-
-**Binding model (explicit — how this SHA is authorized).** The authority binding for this amendment's pinned
-bytes is the composition of two distinct acts:
-1. **Owner class-authorization (PRE-SHA):** Jason (owner) authorized the *class of action* — "amend the
-   charter" (option B), `#mos` 2026-07-18 17:28Z. This occurred **before** these amendment bytes (and hence
-   this SHA256) existed; the owner authorized the amendment, not a specific hash.
-2. **Independent-adjudicator byte-verification:** Mos (`web1:mos-claude`), acting as the independent
-   adjudicator (author ≠ verifier — MS-LEAD authored, Mos verifies), scope-verifies the **exact bytes** of
-   this document and confirms they express only the owner-authorized narrow scope.
-
-There is therefore **NO single "Jason-signs-the-SHA" artifact, and none is required** under this binding
-model: owner-authorized-class + independent-adjudicator-byte-verify **is** the binding. This is the durable,
-inspectable chain of custody for the pin.
-
-## 2. Charter provisions amended (sha-pinned, unmutated)
-
-This amendment attaches to — and does NOT rewrite — the ratified Gate0 charter provisions:
-
-- `compaction-refresh-BUILD-BRIEF.md` §5 / R4 "Gate0 = BUILD-ADMISSION GATE", specifically the item:
-  *"Same-PID `runtime_generation` bump on reload/resume/fork revokes prior lease (D4)."*
-  sha256 `89fdbc27ed0e5050dc7b52f3ef2ddaea691edf17fd89d51b15e26fb5ed47171b` (intact/unchanged).
-- `compaction-refresh-SPEC-RATIFICATION.md` R4.
-  sha256 `bac58319c9c4028b5b40e1129e0033cdb5a6b7b02033c25f06f4cb77d7779c67` (intact/unchanged).
-
-The two pinned docs remain byte-identical; this amendment is the delta of record.
-
-## 3. What is authorized (NARROW — D4-focused harness ONLY)
-
-Adjudicator ruling (Mos, `web1:mos-claude`): "D4-only" against the existing runner is non-executable — the
-only runner (`pi_gate0_run.py`) drives P2+P5+P6+D4 in one flow, and D4 intrinsically requires a
-promoted-to-VERIFIED baseline (spec L219: "P3 immediately follows the valid P2 promotion so reload must
-revoke a genuinely VERIFIED prior gen") to revoke a genuine prior generation. Therefore authorize EXACTLY the
-minimal executable form (Option A) required as WI-3 (#830) merge-gate (2):
-
-- **D4-focused harness:** spawns **ONLY** `p3_generation_broker.py` (not the full `pi_gate0_run.py` runner).
-- **Minimal promotion-to-VERIFIED, as a FIXTURE PRECONDITION ONLY:** established solely so the D4 revoke has a
-  genuine `VERIFIED` target. This is **NOT** authorization to bank P2 as an independent evidence class.
-- **D4 same-PID `runtime_generation`-bump revoke observation:** new generation → `MUTATOR_UNVERIFIED`;
-  prior (superseded) generation → `STALE_GENERATION`.
-- **Explicitly EXCLUDED:** P5 (missing / oversize / hash-invalidation) and P6 (atomic-observation) — not
-  needed for the WI-3 D4 gate and **not authorized** by this amendment.
-- **Isolation / integrity (per §4/§5):** 3× repeated, isolated runs; hardened deterministic harness;
-  non-destructive; launches broker/socket/state artifacts in an isolated fixture ONLY; **does NOT touch
-  production**, mutates no live lease/broker/deployment, creates no durable side effect outside the fixture;
-  a **FAIL** returns to planner — no retry-launder.
-
-**Intrinsic-precondition boundary (verbatim):** the promotion step is a D4 test fixture, not a P2
-evidence-gathering authorization; P5/P6 remain unauthorized.
-
-## 4. What is NOT authorized (all other holds remain in force)
-
-- This is **NOT** a blanket Gate0-execution release. Every other Gate0 execution item and every standing
-  execution hold remains exactly as-is.
-- No live mutation, migration, canary, deployment, systemd, tmux-fleet, or connector/socket/Hermes action is
-  authorized by this amendment.
-- Authorizes producing THIS evidence class once (3× isolation is the evidence-integrity requirement, not a
-  license to re-run on failure): a **FAIL** returns to planner — **no retry-launder**, no severity-laundering.
-
-## 5. Preserved fail-closed / DO-178C invariants (verbatim, unchanged)
-
-- Producing probe evidence **EXECUTES** the Gate0 mechanism (launches processes, creates socket/state
-  artifacts, exercises revocation transitions); "isolated + non-destructive" does **not** make it
-  non-execution. This amendment authorizes that execution for the narrow class in §3 — it does not redefine
-  execution as non-execution.
-- A sha-pinned DURABLE authority OUTRANKS a runtime coordinator's verbal GO; when they conflict the durable
-  artifact wins. This amendment derives its authority from the OWNER, not from a coordinator.
-- Card-advancement gates and execution-holds are distinct; this amendment lifts ONLY the §3 execution-hold,
-  not any advancement/review/SECREV gate. WI-3's independent-review + Opus-SECREV + green-suite gates stand.
-
-## 6. Effect on WI-3 (#830) merge chain
-
-On this amendment being pinned (Mos-verified + Jason-shown): a fresh lane (ms-rev-826) executes the §3
-probe-3 evidence class (3× isolation) → PASS/FAIL reported to Mos. **On PASS**, and only after re-confirming
-WI-3 head `f4008307` UNMOVED + FIRE-PRECONDITION-0 (fresh `mosaic-context-refresh` attestation for
-mosaic-100 + planner-opus), the existing chain runs: CI queue-guard → push → PR → mirror RoR → relay full-40
-head + pr/ci# → Mos independent 6-check → squash `closes #830` → chain WI-4→7. **On FAIL**: planner-return.
-
----
-
-**Amendment author:** MS-LEAD (`web1:mosaic-100`), Gate0 governance interface / charter custodian.
-**Pin protocol:** this file's sha256 becomes the amendment pin ONLY after Mos re-scope-verify. Reported to Mos
-before any ratification/publication, per the transparency checkpoint on a durable-authority change. Ratification
-and durable publication (to a homelab-inspectable provider ref) are the adjudicator's (Mos), NOT the author's
-(MS-LEAD) — the author does not self-ratify and does not publish.

docs/compaction-refresh/governance/OWNER-WINDOW.md

@@ -1,49 +0,0 @@
-# Gate0 Probe-3 Amendment — Owner Transparency Window (durable record)
-
-**Purpose.** Durable evidence (defined duration + closure) that the owner-transparency window on the
-probe-3 amendment **scope-correction** has a bounded, inspectable lifecycle — per homelab's GATE-A
-requirement (2026-07-18 18:13:14Z). This is NOT the owner's core authorization (that is durable in the
-amendment §1); it is the objection-window on the narrowing disclosed afterward.
-
-## What was disclosed
-The owner (Jason) explicitly authorized the amendment **class** — "amend the charter" (option B),
-`#mos` 2026-07-18 17:28Z (amendment §1). My initial narrowing was "D4-only"; on adjudication that proved
-non-executable (D4 intrinsically needs a promoted-to-VERIFIED baseline). The **corrected** scope
-(rev 2, pin `9ac9ff87`, §3) adds a *minimal promotion-to-VERIFIED as a test fixture only* (P5/P6 still
-excluded, P2 not banked as evidence). Because this adds a fixture step beyond the "D4-only" I first told
-the owner, I disclosed the correction to him for transparency.
-
-## Window (bounded, durable)
-- **Opened:** 2026-07-18T18:00:25Z  — anchored to the adjudicator's Discord scope-correction disclosure to the
-  owner (message id `1528099140100296736`, snowflake-decoded; not an informal estimate).
-- **Duration:** 2 hours (defined).
-- **Scheduled closure:** 2026-07-18T20:00:25Z.
-- **Closure condition (fail-closed):**
-  - No owner objection received by scheduled closure → window **CLOSED-CLEAR** (the disclosed correction
-    stands; it is within the already-granted class authorization).
-  - Any owner objection before closure → window does **NOT** clear → re-adjudicate the corrected scope.
-  - An affirmative owner "proceed" before closure → **early CLOSED-CLEAR** (recorded here).
-
-## Closure attestation
-Closure will be confirmed by a dated adjudicator tick appended below (and to the orchestration ledger)
-at/after 2026-07-18T20:00:25Z. Until that attestation exists, GATE A is OPEN and probe-3 execution stays HELD.
-
-- [x] **CLOSED-CLEAR attestation — recorded 2026-07-18T20:08Z by adjudicator Mos (`web1:mos-claude`).**
-
-### CLOSED-CLEAR attestation (2026-07-18T20:08Z)
-
-The owner-transparency window on the probe-3 amendment scope-correction is **CLOSED-CLEAR**.
-
-- **Scheduled closure:** 2026-07-18T20:00:25Z — reached (authoritative UTC at attestation: 2026-07-18T20:07Z, past closure).
-- **Closure condition met:** *No owner objection received by scheduled closure.* Verified fail-closed against the
-  **durable Discord #mos record** (not memory): the channel's last message is the scope-correction disclosure itself
-  (`id 1528099140100296736`, 2026-07-18T18:00:25.709Z). From window-open (18:00:25Z) through closure (20:00:25Z) and
-  to this attestation, there are **zero owner (jetrich) messages** — no objection, no affirmative early-GO. The last
-  owner message preceding the window was the class authorization *"Amend the charter"* (17:28Z, amendment §1).
-- **Effect:** the disclosed corrected scope (rev 2, pin `9ac9ff87`, §3 — minimal promotion-to-VERIFIED as a D4 test
-  fixture only; P5/P6 excluded; P2 not banked as evidence) **stands**, within the already-granted class authorization.
-- **GATE A is now CLOSED.** Probe-3 execution proceeds ONLY under the full remaining chain: GATE-B fresh §3-review PASS
-  (concurred @839d156f, author≠reviewer) + adjudicator explicit FIRE GO. No other Gate0 hold is lifted by this tick.
-
-*Adjudicator: Mos (`web1:mos-claude`). This record is durable and provider-inspectable alongside the
-amendment (`9ac9ff87`) and its RATIFICATION.md.*

docs/compaction-refresh/governance/RATIFICATION.md

@@ -1,37 +0,0 @@
-# Gate0 Probe-3 (D4) Amendment — Adjudicator Ratification Record
-
-**Ratifies pin:** `9ac9ff873fad41a6e15763cc89cb94d0bc4a6b0cf9b6770561d1781b03f63276`
-(`GATE0-PROBE3-EXEC-AMENDMENT.md`, 7699 bytes, this branch).
-**Supersedes:** prior candidate pin `8fca5b7a` (never executed against; withdrawn on scope + durability grounds).
-
-## Re-scope-verify — PASSED
-Verified by **Mos (`web1:mos-claude`)**, acting as independent adjudicator (author ≠ verifier: MS-LEAD
-authored, Mos verified), 2026-07-18. SHA256 recomputed = pin (exact). Scope checks, all PASS:
-- §3 = Option A, D4-focused: harness spawns ONLY `p3_generation_broker.py`; minimal promotion-to-VERIFIED
-  is a FIXTURE PRECONDITION only (not P2 evidence-banking); D4 gen-bump revoke observation
-  (new→`MUTATOR_UNVERIFIED`, prior→`STALE_GENERATION`); **P5/P6 explicitly excluded + unauthorized**;
-  verbatim intrinsic-precondition boundary line present.
-- §1 binding model explicit; §2 charter provisions pinned unmutated; §4/§5 invariants intact.
-- No scope creep beyond the owner-authorized narrow class.
-
-## Authority binding
-- **Owner class-authorization (PRE-SHA):** Jason (owner) — "amend the charter" (option B), `#mos`
-  2026-07-18 17:28Z. Authorized the amendment class before these bytes existed.
-- **Independent-adjudicator byte-verification:** Mos scope-verified the exact bytes.
-- No single "Jason-signs-the-SHA" artifact exists or is required under this binding model.
-
-## Round-trip integrity (this branch/commit)
-All three files fetched back via the Gitea raw API and SHA-matched (each > 200B, no not-found sentinel):
-- `GATE0-PROBE3-EXEC-AMENDMENT.md` → `9ac9ff87…f63276` ✓
-- `charter-BUILD-BRIEF.md`         → `89fdbc27…71b` ✓
-- `charter-SPEC-RATIFICATION.md`   → `bac58319…67` ✓
-
-## Status: RATIFIED — probe-3 execution remains HELD
-Publication here closes the governance-infra gap (the charter provisions were previously local-only,
-not inspectable). Probe-3 execution stays HELD pending both: (1) homelab independent verification of
-these published bytes; (2) owner transparency window. On both clearing → ms-rev-826 builds + runs the
-§3 D4-focused harness (3× isolation, non-destructive) → PASS/FAIL to Mos → WI-3 (#830) chain per §6.
-
-*Note: the amendment file's own status line reads "DRAFT (revision 2)"; that internal text is
-superseded by THIS external ratification record (the pinned bytes are deliberately not edited, so the
-SHA stays stable). Ratification status lives here, not in the pinned artifact.*

docs/compaction-refresh/governance/charter-BUILD-BRIEF.md

@@ -1,106 +0,0 @@
-# Compaction-Refresh Mechanism — BUILD BRIEF (Mos → MS-LEAD)
-
-**Status:** STAGED — do NOT dispatch until PR #826 (#824 skill-CLI bridge) MERGES. Mos routes this to
-MS-LEAD at #824 land. PLAN phase COMPLETE + RATIFIED; this is the BUILD-execution brief.
-**Routed by:** Mos (main orchestrator). **Owner:** MS-LEAD (mosaic-100). **Runtime scope M1:** Claude + Pi.
-
-## 1. AUTHORITY (sha256-pinned — build AGAINST these, do not re-derive)
-| Artifact | Path | SHA-256 |
-|---|---|---|
-| **Ratification record (SSOT)** | `~/agent-work/reviews/compaction-refresh-SPEC-RATIFICATION.md` | `bac58319c9c4028b5b40e1129e0033cdb5a6b7b02033c25f06f4cb77d7779c67` |
-| Ratified SPEC v5 | `~/agent-work/reviews/compaction-refresh-SPEC-v5.md` | `a6d07ade835758e8488ca10d3b0631caf0beb93ea3a6733631f151b0c2f01433` |
-| sol FINAL red-team (GO) | `~/agent-work/reviews/compaction-refresh-SPEC-v5-redteam-sol.md` | `3da326a4ea91767b731e128a93b13194e8002358101e30de3fcb8ca2f8f54faa` |
-
-The coder MUST verify these hashes before building. The in-context working spec is DERIVED and
-compaction-fragile — the sha256'd records above are the recovery SSOT (durable-authority discipline).
-
-## 2. TARGET & SCOPE
-- **Framework-native `packages/mosaic/…`** in `mosaicstack/stack`. NOT jarvis-brain. NOT
-  `~/.config/mosaic/` directly (that tree is WIPED on framework upgrade — durable home is SOURCE).
-- **M1 = Claude + Pi orchestrators only.** Codex/other runtimes = scope-note only.
-- **Hard dependency (now satisfied at dispatch):** the `mosaic skill` register/list + install/upgrade
-  symlink auto-sync (#824 / PR #826) — so the durable `mosaic-context-refresh` skill ships REACHABLE
-  (auto-symlinked into `~/.claude/skills/`). Do not dispatch before #826 merges.
-
-## 3. DELIVERABLES (locked architecture — from the ratification record; do NOT redesign)
-1. **Authenticated external lease broker** bound to kernel `SO_PEERCRED` `(pid, starttime,
-   runtime_generation)` — the only unforgeable identity on a same-uid tmux fleet. Broker **MINTS** the
-   logical `session_id` at first peercred contact (NEVER caller-asserted). Distinct-principal /
-   protected socket.
-2. **Whole mutator-class gate** (not per-wrapper): no consequential mutator succeeds without a valid
-   VERIFIED lease. **Revoke-first / promote-last.**
-3. **Compaction observers → revoke:** Claude `PreCompact` + `SessionStart(matcher=compact)`; Pi
-   `session_before_compact`/`context` equivalents. Any `runtime_generation` bump (reload/resume/fork,
-   same PID included) auto-revokes the prior incarnation.
-4. **Verbatim-hashed normative fragments:** `B_payload` (exact fragments + deterministic metadata) and
-   `H_payload = SHA256(domain_sep ‖ length_framed(B_payload))` — NO self-reference (byte-identical
-   across the Claude and Pi builders).
-5. **Receipt-challenge protocol (promotion prerequisite, NOT the safety mechanism):** broker mints a
-   single-use `receipt_challenge` per cycle bound to `(session_id, runtime_generation,
-   compaction_epoch, request_epoch, H_source, H_payload, schema_version)`. **COMPUTE = broker**
-   (observes latest-assistant message, computes the binding incl `H(latest-assistant-message)`);
-   **COPY = model** (verbatim copy of challenge + `H_payload` — an LLM cannot hash its own output;
-   T29). Observed in the EXACT latest assistant message (Claude: latest assistant entry + pending
-   challenge, NO transcript grep; Pi: `message_end`, NOT `after_provider_response`). Single-use,
-   consumed before the tool batch. Broker transition order (mandatory): `revoke → build
-   B_payload/H_payload → PENDING_DELIVERY(mint unique challenge) → deliver → observe exact receipt →
-   evidence commit → consume challenge → promote VERIFIED (last)`.
-6. **Constrained recovery command** = the single ungated mutator; the `mosaic-context-refresh` skill
-   is its wrapper (uses a newly-minted challenge + identical protocol; cannot replay normal-path
-   receipt). Durable skill lands here, symlink-reachable via #824.
-7. **T-C server-side line = branch protection** (the irreducible guarantee; client gate is
-   window-narrowing only). If ops-config rather than code, DOCUMENT the required posture explicitly.
-
-## 4. BINDING BUILD CONDITIONS R1–R6 (MANDATORY — verbatim from ratification record)
-- **R1 (honesty):** state that the receipt detects ABSENT or PREFIX-TRUNCATED terminal token, but a
-  MIDDLE-DROP preserving the tail is a **T-C contract violation, NOT receipt-detectable** (covered by
-  server-side). No over-claim.
-- **R2 (atomicity):** evidence-commit → consume-challenge → promote-VERIFIED is ONE atomic broker
-  transaction; a crash leaves neither VERIFIED-with-live-challenge nor consumed-with-ambiguous-promote;
-  recovery mints a NEW challenge, never reuses.
-- **R3 (parsing):** exact single current-cycle receipt parse — no unbounded grep; reject
-  quoted/tool-output receipts, wrong generation/epoch, unknown/stale challenge, multiple receipts.
-- **R4 (Gate0 = BUILD-ADMISSION GATE):** see §5 — runtime evidence BEFORE build admission; a failed
-  Gate0 item RETURNS the design to planner review, it is NOT waived by the GO.
-- **R5 (TTL):** 300 s MAX lease TTL + soak-tighten; soak may only SHORTEN, never lengthen. Mos-accepted.
-- **R6 (receipt semantics):** receipt = T-A delivery/liveness ONLY, never obedience/safety/residency.
-  Permanent invariant.
-
-## 5. R4 GATE0 — RUNTIME EVIDENCE BEFORE BUILD ADMISSION (WI-0, do this FIRST)
-Structure the build like gitwatch: **WI-0 is a Gate0 runtime-evidence probe; no feature build admitted
-until every item produces POSITIVE runtime evidence** (not a design assertion). A failed item RETURNS
-to planner — do not paper over:
-- Launcher `exec`/parent topology + supported-hook ancestry (D1); broker authenticates the launcher
-  chain, rejects sibling-substitution.
-- Pi lifecycle: last-position invariant (last-or-closed), per-tool nonce → tool-call-id map (D5).
-- Same-PID `runtime_generation` bump on reload/resume/fork revokes prior lease (D4).
-- Broker socket authenticity posture (protected / distinct-principal); `SO_PEERCRED` returns the true
-  `(pid, starttime)`.
-- Source-invalidation fail-closed (missing/oversize/hash-mismatch fragment → no promotion).
-- Claude `additionalContext` + Pi `context` inject atomically (A-v5-1) — the transport assumption
-  T27 rests on; verify or class the gap T-C.
-
-## 6. ACCEPTANCE TESTS (red-first TDD)
-Carry all v4 ACs + T12b; add **T24–T30**: T24 hash-construction no-self-reference byte-identical across
-builders · T25 replay rejected · T26 transcript-stale-match cannot promote · T27 partial-delivery
-(prefix-trunc/tail-only/middle-drop/malformed) none promote · T28 single-use (no renew/reopen) · T29
-model copies not computes · T30 dual-hook-miss matches AMENDED threat table within+after TTL.
-Acceptance = full green suite + Gate0 evidence pack.
-
-## 7. REVIEW & MERGE DISCIPLINE
-- **This IS an Opus-SECREV-mandatory surface** — UNLIKE #824 (local same-uid FS). The broker is an
-  authentication/identity/authorization mechanism (peercred binding, session_id minting, mutator-class
-  gating, receipt protocol). Security review = **Opus-SECREV, no GPT/terra substitute** on the
-  broker/gate/receipt/socket surfaces. Functional/non-security parts may take GPT review.
-- **Author ≠ reviewer**, independent review, no self-merge, exact-head RoR (reviewed-SHA=merged-SHA),
-  `closes #<issue>` close-keyword, full 40-char head. Never edit tests to pass / never force-merge red
-  / never `--no-verify`.
-- **Mos merges** (per ratification record) after Opus-SECREV + independent review pass + green suite +
-  Gate0 pack. Decompose into work-items (WI-0 Gate0 first); likely a milestone, not one monolithic PR.
-
-## 8. SEQUENCING
-#826 (#824 bridge) MERGES → Mos routes this brief → MS-LEAD decomposes (WI-0 Gate0 first) → build M1
-(Claude+Pi) → Gate0 evidence pack (failed item → planner) → Opus-SECREV + independent review → Mos
-merges → durable `mosaic-context-refresh` skill lands symlink-reachable → jarvis-brain `CLAUDE.md`
-mandate PR aligns to the shipped receipt/hash contract. Interim skill + CLAUDE.md mandate stay LIVE
-until the mechanism ships.

docs/compaction-refresh/governance/charter-SPEC-RATIFICATION.md

@@ -1,75 +0,0 @@
-# Compaction-Refresh Mechanism — SPEC RATIFICATION RECORD
-
-**Ratifier:** Mos (main fleet orchestrator, `mos-claude`)
-**Date:** 2026-07-17 ~22:2xZ
-**Decision:** ✅ **RATIFIED — GO WITH BINDING CONDITIONS R1–R6.** Architecture converged v1→v5 via
-oppositional adversarial planning (planner-opus authors / planner-sol red-teams). No v6 required.
-
-## Authenticated authority artifacts (sha256-pinned)
-
-| Artifact | Path | SHA-256 |
-|---|---|---|
-| Ratified SPEC v5 | `~/agent-work/reviews/compaction-refresh-SPEC-v5.md` | `a6d07ade835758e8488ca10d3b0631caf0beb93ea3a6733631f151b0c2f01433` |
-| sol FINAL red-team (GO) | `~/agent-work/reviews/compaction-refresh-SPEC-v5-redteam-sol.md` | `3da326a4ea91767b731e128a93b13194e8002358101e30de3fcb8ca2f8f54faa` |
-| (base) SPEC v4 | `~/agent-work/reviews/compaction-refresh-SPEC-v4.md` | `a5e9c261a613974fc0d853d69ad76f616f5f1c3b6b63a436daecdd15b4a72b80` |
-| (base) sol v4 red-team | `~/agent-work/reviews/compaction-refresh-SPEC-v4-redteam-sol.md` | `1e76ee5942241d9520b9ff8f39a628488578ec2d9b91cc09b9f0e7cc6091ef2f` |
-| Ratification-Input (R1-R6 + rulings, provenance-clean) | `~/agent-work/reviews/compaction-refresh-SPEC-RATIFICATION-INPUT.md` | `d01411a5c966f513e992a941deef0ff5cc6f29d37adab97ea2fbd3c5e6aef266` |
-
-## Convergence trail
-v1 NO-GO (9 architectural) → v2 NO-GO (7 enforcement, arch accepted) → v3 NO-GO (5 surgical, arch
-locked) → v4 NO-GO (narrow; echo-lock narrowing APPROVED) → **v5 GO-with-conditions**.
-
-## Locked architecture (do not redesign)
-Authenticated external **lease broker** bound to kernel `SO_PEERCRED` (pid, starttime,
-runtime_generation) — the only unforgeable identity fact on a same-uid tmux fleet. Broker MINTS the
-logical session_id at first peercred contact (never caller-asserted). **Revoke-first / promote-last.**
-Whole **mutator-class gate** (not per-wrapper). Verbatim-hashed normative fragments. Single
-**constrained recovery command** = the only ungated mutator; the `mosaic-context-refresh` skill is its
-wrapper. **Threat tiers:** T-A honest-stale → client gate · T-B compromised-tool → mutator-class · T-C
-fully-rotted → **server-side branch-protection = the irreducible line** (client gate is
-window-narrowing, NOT a complete guarantee).
-
-**Delivery-receipt (echo-lock, sol §8-final):** broker mints a single-use `receipt_challenge` per
-cycle bound to (session, runtime_generation, epochs, H_source, H_payload); **COMPUTE = broker**
-(observes + computes the full binding incl H(latest-assistant-message) as the cycle-audit fact),
-**COPY = model** (copies challenge + H_payload verbatim — model never computes a hash; sol T29
-copy-not-compute preserved). Receipt is a **one-cycle T-A delivery/liveness proof ONLY — never an
-obedience, residency, or safety proof.** The mechanical mutator-class gate remains THE safety
-mechanism; §8 intact for T-B/T-C.
-
-## Binding ratification conditions (R1–R6) — MANDATORY for build acceptance
-- **R1 (honesty / T27 amendment):** state honestly that the receipt detects an ABSENT or
-  PREFIX-TRUNCATED terminal token, but a MIDDLE-DROP that preserves the tail token is a **T-C
-  contract violation, NOT receipt-detectable** (covered by server-side, not the receipt). No
-  over-claim. Fold as a doc amendment — no v6.
-- **R2 (atomicity):** evidence-commit → consume-challenge → promote-VERIFIED is ONE atomic broker
-  transaction. A crash leaves NEITHER a VERIFIED-with-live-challenge NOR a
-  consumed-with-ambiguous-promotion state; recovery mints a **NEW** challenge, never reuses one.
-- **R3 (parsing):** exact single current-cycle receipt parse — no unbounded grep; reject
-  quoted/tool-output receipts, wrong generation/epoch, unknown/stale challenge, and multiple receipts.
-- **R4 (Gate0 runtime evidence — BUILD-ADMISSION GATE):** launcher-ancestry/D1,
-  sibling-substitution rejection, Pi last-or-closed, same-PID runtime_generation bump,
-  nonce→tool-call-id map, socket/authenticity posture, and source-invalidation fail-closed require
-  **runtime Gate0 evidence BEFORE build admission**, not design assertion. **A failed Gate0 item
-  RETURNS the design to planner review — it is NOT waived by this GO.**
-- **R5 (TTL — RATIFIER ACCEPTED):** proposed **300s MAX lease TTL + soak-tighten** is **ACCEPTED by
-  Mos (ratifier)**. Safe-direction default: soak data may only SHORTEN it, never lengthen. Operator
-  (Jason) may tighten at will; flagged for operator awareness, non-blocking (not an escalation
-  trigger — a proposed max that only tightens).
-- **R6 (receipt semantics — REAFFIRMED):** receipt remains T-A delivery/liveness ONLY, never
-  obedience/safety proof. Reaffirmed as a permanent invariant of the design.
-
-## Build sequencing (PLAN-ONLY until these clear)
-1. **HARD DEP:** `mosaic skill` register/unregister/list + generic install/upgrade symlink auto-sync
-   (mosaicstack/stack **#824**, in flight on ms-824) MUST land first — else the durable
-   `mosaic-context-refresh` skill ships unreachable (no `~/.claude/skills/` bridge symlink).
-2. Then MS-LEAD builds the mechanism in **packages/mosaic** (M1 = Claude + Pi orchestrators) against
-   THIS ratified spec + R1–R6, producing R4 Gate0 evidence. Author≠reviewer independent review; no
-   self-merge; Mos merges.
-3. The durable `mosaic-context-refresh` skill (+ jarvis-brain CLAUDE.md mandate PR) lands aligned to
-   the ratified receipt/hash contract, AFTER the #824 bridge exists.
-
-**Interim protection (live now, until the mechanism ships):** `mosaic-context-refresh` skill
-(fail-closed 7-line residency attestation) + jarvis-brain CLAUDE.md "Directive Freshness After
-Compaction" mandate — self-run manual re-read + attestation. Symlink hand-created this session; loads
-and attests 7/7.

docs/compaction-refresh/probes/p5_receipt_replay.py

@@ -0,0 +1,228 @@
+#!/usr/bin/env python3
+"""P5 Gate0 replay probe; BUILT ONLY, execution is Mos-gated.
+
+Run only under fresh-executor authorization:
+  python3 -I -S -B docs/compaction-refresh/probes/p5_receipt_replay.py
+
+Each of the default three isolated runs launches the shipped lease-broker daemon
+in a distinct private temporary directory.  This driver never changes broker
+state directly and does not replace the promote gate: every transition is sent
+over the daemon's real Unix socket.  It proves the shipped order is
+PENDING_DELIVERY -> observe/evidence commit -> consume -> VERIFIED and that a
+consumed challenge cannot be replayed or reopen/renew its lease.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import importlib.util
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+
+HERE = Path(__file__).resolve().parent
+REPOSITORY = HERE.parents[2]
+TOOLS = REPOSITORY / "packages/mosaic/framework/tools/lease-broker"
+DAEMON = TOOLS / "daemon.py"
+FRAGMENTS = TOOLS / "normative_fragments.py"
+
+
+def load_shipped_fragments():
+    if not FRAGMENTS.is_file():
+        raise RuntimeError(f"shipped normative construction missing: {FRAGMENTS}")
+    spec = importlib.util.spec_from_file_location("p5_shipped_normative_fragments", FRAGMENTS)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("unable to load shipped normative construction")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]:
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(3.0)
+        connection.connect(str(socket_path))
+        connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
+        connection.shutdown(socket.SHUT_WR)
+        response = bytearray()
+        while True:
+            chunk = connection.recv(4096)
+            if not chunk:
+                break
+            response.extend(chunk)
+    if not response.endswith(b"\n") or response.count(b"\n") != 1:
+        raise AssertionError(f"unframed broker reply: {bytes(response)!r}")
+    parsed = json.loads(response[:-1])
+    if not isinstance(parsed, dict):
+        raise AssertionError(f"non-object broker reply: {parsed!r}")
+    return parsed
+
+
+def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None:
+    deadline = time.monotonic() + 5.0
+    while time.monotonic() < deadline:
+        if socket_path.exists():
+            return
+        if process.poll() is not None:
+            output = process.stdout.read() if process.stdout is not None else ""
+            raise RuntimeError(f"shipped daemon exited before READY: {output}")
+        time.sleep(0.02)
+    raise TimeoutError("shipped daemon did not create private probe socket")
+
+
+def expect_refused(reply: dict[str, object], code: str) -> None:
+    if reply != {"ok": False, "code": code}:
+        raise AssertionError(f"expected refusal {code}, got {reply!r}")
+
+
+def run_once(index: int) -> str:
+    fragments = load_shipped_fragments()
+    root = Path(tempfile.mkdtemp(prefix=f"mosaic-p5-replay-{index}-"))
+    os.chmod(root, 0o700)
+    socket_path = root / "broker.sock"
+    state_path = root / "state.json"
+    observer_path = root / "test-observer.json"
+    process = subprocess.Popen(
+        [
+            sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(socket_path),
+            "--state", str(state_path), "--test-observer-file", str(observer_path),
+        ],
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    try:
+        wait_ready(process, socket_path)
+        registered = request(socket_path, {"action": "register_anchor", "runtime_generation": 1})
+        if registered.get("ok") is not True or not isinstance(registered.get("session_id"), str):
+            raise AssertionError(f"registration failed: {registered!r}")
+        session_id = registered["session_id"]
+        construction = fragments.build_payload(
+            manifest_version=1,
+            generator_version="p5-replay-probe",
+            fragments=[
+                fragments.NormativeFragment(
+                    "authority/probe",
+                    b"P5 shipped transition driver\n",
+                    "63537df1a6cb0d80195a96757ab11d629e5b5e1f23be167218b84cb195b1c1d6",
+                ),
+            ],
+        )
+        if construction.injectionDecision != "ACCEPTED" or not construction.promotion:
+            raise AssertionError("shipped normative construction refused P5 fixture")
+        binding = {
+            "compaction_epoch": index,
+            "request_epoch": index + 100,
+            "h_source": construction.h_source,
+            "h_payload": construction.h_payload,
+            "schema_version": 1,
+        }
+        construction_request = {
+            "manifest_version": 1,
+            "generator_version": "p5-replay-probe",
+            "fragments": [{
+                "source_id": "authority/probe",
+                "content_base64": base64.b64encode(b"P5 shipped transition driver\n").decode("ascii"),
+                "expected_sha256": "63537df1a6cb0d80195a96757ab11d629e5b5e1f23be167218b84cb195b1c1d6",
+            }],
+        }
+        pending = request(socket_path, {
+            "action": "begin_verification",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "runtime": "pi",
+            "binding": binding,
+            "construction": construction_request,
+        })
+        if pending.get("ok") is not True or pending.get("state") != "PENDING_VERIFICATION":
+            raise AssertionError(f"shipped pending-delivery transition failed: {pending!r}")
+        challenge = pending.get("receipt_challenge")
+        receipt = pending.get("receipt")
+        if not isinstance(challenge, str) or not isinstance(receipt, str):
+            raise AssertionError(f"shipped broker did not mint a receipt challenge: {pending!r}")
+
+        # Promotion before observation/evidence/consumption is forbidden.
+        expect_refused(request(socket_path, {
+            "action": "promote_lease",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+        }), "INVALID_LEASE_TRANSITION")
+
+        observer_path.write_text(json.dumps({
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "latest_assistant_message": receipt,
+        }), encoding="utf-8")
+        os.chmod(observer_path, 0o600)
+        observed = request(socket_path, {
+            "action": "observe_receipt",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+        })
+        if observed.get("ok") is not True or observed.get("state") != "PENDING_PROMOTION":
+            raise AssertionError(f"shipped evidence transition failed: {observed!r}")
+        durable = json.loads(state_path.read_text(encoding="utf-8"))
+        evidence = durable["tokens"][challenge].get("evidence")
+        if not isinstance(evidence, dict) or not isinstance(evidence.get("h_latest_assistant"), str):
+            raise AssertionError("shipped receipt evidence was not committed before consume/promote")
+
+        promoted = request(socket_path, {
+            "action": "promote_lease",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+        })
+        if promoted.get("ok") is not True or promoted.get("state") != "VERIFIED":
+            raise AssertionError(f"shipped consume-before-promote transition failed: {promoted!r}")
+
+        # T25/T28: the actual consumed challenge, re-presented through the
+        # shipped daemon, can neither be observed again nor re-promote/reopen.
+        expect_refused(request(socket_path, {
+            "action": "observe_receipt",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+        }), "RECEIPT_REPLAY")
+        expect_refused(request(socket_path, {
+            "action": "promote_lease",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+        }), "RECEIPT_REPLAY")
+        return challenge
+    finally:
+        if process.poll() is None:
+            process.terminate()
+            try:
+                process.wait(timeout=3.0)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+        shutil.rmtree(root, ignore_errors=True)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--runs", type=int, default=3)
+    arguments = parser.parse_args()
+    if arguments.runs != 3:
+        raise SystemExit("P5 requires exactly three isolated runs")
+    challenges = [run_once(index) for index in range(arguments.runs)]
+    if len(set(challenges)) != arguments.runs:
+        raise AssertionError("separate shipped cycles did not mint unique challenges")
+    print("P5 receipt replay probe PASS: 3 isolated shipped-daemon runs")
+
+
+if __name__ == "__main__":
+    main()

docs/compaction-refresh/probes/p6_constrained_recovery.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""P6 constrained-recovery probe; BUILT ONLY and Mos-gated.
+
+DO NOT self-fire. Under Mos authorization only:
+  python3 -I -S -B docs/compaction-refresh/probes/p6_constrained_recovery.py
+
+The default three isolated runs launch the shipped daemon plus its production
+observer transport on private sockets. The driver invokes the shipped recovery
+command and adapter gate identity; it never resets broker state, mocks promote,
+or taps a live model-output stream.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import importlib.util
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+
+HERE = Path(__file__).resolve().parent
+REPOSITORY = HERE.parents[2]
+TOOLS = REPOSITORY / "packages/mosaic/framework/tools/lease-broker"
+DAEMON = TOOLS / "daemon.py"
+GATE = TOOLS / "mutator-gate.py"
+RECOVERY_COMMAND = TOOLS / "recover-context.py"
+OBSERVER_CLIENT = TOOLS / "receipt-observer-client.py"
+FRAGMENTS = TOOLS / "normative_fragments.py"
+CLAUDE_SETTINGS = REPOSITORY / "packages/mosaic/framework/runtime/claude/settings.json"
+PI_EXTENSION = REPOSITORY / "packages/mosaic/framework/runtime/pi/mosaic-extension.ts"
+
+
+def load_shipped_fragments():
+    spec = importlib.util.spec_from_file_location("p6_shipped_fragments", FRAGMENTS)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("shipped normative construction unavailable")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]:
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(3.0)
+        connection.connect(str(socket_path))
+        connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
+        connection.shutdown(socket.SHUT_WR)
+        response = bytearray()
+        while True:
+            chunk = connection.recv(4096)
+            if not chunk:
+                break
+            response.extend(chunk)
+    if not response.endswith(b"\n") or response.count(b"\n") != 1:
+        raise AssertionError(f"unframed broker reply: {bytes(response)!r}")
+    reply = json.loads(response[:-1])
+    if not isinstance(reply, dict):
+        raise AssertionError("broker reply is not an object")
+    return reply
+
+
+def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None:
+    deadline = time.monotonic() + 5.0
+    while time.monotonic() < deadline:
+        if socket_path.exists():
+            return
+        if process.poll() is not None:
+            output = process.stdout.read() if process.stdout is not None else ""
+            raise RuntimeError(f"shipped daemon exited before READY: {output}")
+        time.sleep(0.02)
+    raise TimeoutError("shipped daemon did not create private probe socket")
+
+
+def run_json(command: list[str], environment: dict[str, str], input_value: object | None = None) -> dict[str, object]:
+    completed = subprocess.run(
+        command,
+        input=None if input_value is None else json.dumps(input_value),
+        text=True,
+        capture_output=True,
+        env=environment,
+        check=False,
+    )
+    if not completed.stdout.endswith("\n"):
+        raise AssertionError(f"command omitted framed result: {completed.stderr!r}")
+    reply = json.loads(completed.stdout)
+    if not isinstance(reply, dict):
+        raise AssertionError("command result is not an object")
+    return reply
+
+
+def gate_recovery(runtime: str, phase: str, environment: dict[str, str]) -> None:
+    command = [sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", runtime]
+    if runtime == "claude":
+        command.extend(["--recovery-command", str(RECOVERY_COMMAND)])
+        recovery_invocation = (
+            f"python3 {RECOVERY_COMMAND} begin --construction /tmp/p6.json "
+            "--compaction-epoch 1 --request-epoch 1"
+            if phase == "begin"
+            else f"python3 {RECOVERY_COMMAND} complete"
+        )
+        value = {"tool_name": "Bash", "tool_input": {"command": recovery_invocation}}
+    else:
+        value = {"tool_name": "mosaic_context_recover"}
+    completed = subprocess.run(command, input=json.dumps(value), text=True, capture_output=True, env=environment, check=False)
+    if completed.returncode != 0:
+        raise AssertionError(f"{runtime} recovery invocation remained gated: {completed.stderr!r}")
+
+
+def record_production_observation(runtime: str, message: str, root: Path, environment: dict[str, str]) -> None:
+    command = [sys.executable, "-I", "-S", "-B", str(OBSERVER_CLIENT), "--runtime", runtime]
+    if runtime == "claude":
+        transcript = root / "claude-transcript.jsonl"
+        transcript.write_text(json.dumps({"message": {"role": "assistant", "content": message}}) + "\n", encoding="utf-8")
+        payload = {"transcript_path": str(transcript)}
+        command.append("--latest-entry")
+    else:
+        payload = {"latest_assistant_message": message}
+    completed = subprocess.run(command, input=json.dumps(payload), text=True, capture_output=True, env=environment, check=False)
+    if completed.returncode != 0:
+        raise AssertionError(f"{runtime} production observer transport refused: {completed.stderr!r}")
+
+
+def run_once(index: int, runtime: str) -> None:
+    # Parity guard: drive the shipped command and the repaired adapter/observer
+    # bytes, not a shadow receipt or promotion implementation.
+    recovery_source = RECOVERY_COMMAND.read_text(encoding="utf-8")
+    if '"action": "begin_recovery"' not in recovery_source or '"action": "complete_recovery"' not in recovery_source:
+        raise AssertionError("P6 parity guard: recovery command no longer drives shipped broker entrypoints")
+    gate_source = GATE.read_text(encoding="utf-8")
+    if "--recovery-command" not in CLAUDE_SETTINGS.read_text(encoding="utf-8"):
+        raise AssertionError("P6 parity guard: Claude recovery mapping is missing")
+    if "_SHELL_ACTIVE" not in gate_source or "argv[1] != str(recovery_command)" not in gate_source:
+        raise AssertionError("P6 parity guard: Claude mapping is not literal-only")
+    if "const RECOVERY_TOOL = 'mosaic_context_recover'" not in PI_EXTENSION.read_text(encoding="utf-8"):
+        raise AssertionError("P6 parity guard: Pi recovery tool mapping is missing")
+
+    fragments = load_shipped_fragments()
+    root = Path(tempfile.mkdtemp(prefix=f"mosaic-p6-recovery-{index}-"))
+    os.chmod(root, 0o700)
+    socket_path = root / "broker.sock"
+    observer_socket = root / "observer.sock"
+    state_path = root / "state.json"
+    construction_path = root / "construction.json"
+    content = b"P6 constrained recovery fixture\n"
+    construction = {
+        "manifest_version": 1,
+        "generator_version": "p6-constrained-recovery",
+        "fragments": [{
+            "source_id": "authority/p6",
+            "content_base64": base64.b64encode(content).decode("ascii"),
+            "expected_sha256": hashlib.sha256(content).hexdigest(),
+        }],
+    }
+    construction_path.write_text(json.dumps(construction), encoding="utf-8")
+    os.chmod(construction_path, 0o600)
+    process = subprocess.Popen(
+        [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(socket_path),
+         "--state", str(state_path), "--observer-socket", str(observer_socket)],
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    try:
+        wait_ready(process, socket_path)
+        registered = request(socket_path, {"action": "register_anchor", "runtime_generation": 1})
+        session_id = registered.get("session_id")
+        if registered.get("ok") is not True or not isinstance(session_id, str):
+            raise AssertionError(f"broker anchor registration failed: {registered!r}")
+        built = fragments.build_payload_from_wire(construction)
+        normal = request(socket_path, {
+            "action": "begin_verification", "session_id": session_id, "runtime_generation": 1,
+            "runtime": runtime, "construction": construction,
+            "binding": {"compaction_epoch": index, "request_epoch": index + 100,
+                        "h_source": built.h_source, "h_payload": built.h_payload, "schema_version": 1},
+        })
+        normal_challenge = normal.get("receipt_challenge")
+        normal_receipt = normal.get("receipt")
+        if not isinstance(normal_challenge, str) or not isinstance(normal_receipt, str):
+            raise AssertionError("normal path did not mint a receipt challenge")
+        environment = {
+            **os.environ,
+            "MOSAIC_LEASE_BROKER_SOCKET": str(socket_path),
+            "MOSAIC_RECEIPT_OBSERVER_SOCKET": str(observer_socket),
+            "MOSAIC_LEASE_SESSION_ID": session_id,
+            "MOSAIC_RUNTIME_GENERATION": "1",
+            "MOSAIC_LEASE_RUNTIME": runtime,
+        }
+        gate_recovery(runtime, "begin", environment)
+        recovery = run_json([
+            sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "begin", "--construction", str(construction_path),
+            "--compaction-epoch", str(index + 10), "--request-epoch", str(index + 110),
+        ], environment)
+        challenge = recovery.get("receipt_challenge")
+        receipt = recovery.get("receipt")
+        if recovery.get("state") != "PENDING_DELIVERY" or not isinstance(challenge, str) or not isinstance(receipt, str):
+            raise AssertionError(f"recovery command did not drive pending delivery: {recovery!r}")
+        if challenge == normal_challenge:
+            raise AssertionError("recovery reused a normal-path challenge")
+
+        # C4: production observer content is still exact-current-cycle only.
+        record_production_observation(runtime, normal_receipt, root, environment)
+        refused = run_json([sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "complete"], environment)
+        if refused.get("ok") is not False or refused.get("code") != "RECEIPT_MISMATCH":
+            raise AssertionError(f"normal-path receipt replay was not refused: {refused!r}")
+
+        gate_recovery(runtime, "begin", environment)
+        recovery = run_json([
+            sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "begin", "--construction", str(construction_path),
+            "--compaction-epoch", str(index + 20), "--request-epoch", str(index + 120),
+        ], environment)
+        receipt = recovery.get("receipt")
+        if recovery.get("state") != "PENDING_DELIVERY" or not isinstance(receipt, str):
+            raise AssertionError(f"fresh recovery retry did not pend: {recovery!r}")
+        record_production_observation(runtime, receipt, root, environment)
+        gate_recovery(runtime, "complete", environment)
+        promoted = run_json([sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "complete"], environment)
+        if promoted.get("ok") is not True or promoted.get("state") != "VERIFIED":
+            raise AssertionError(f"recovery consume-before-promote failed: {promoted!r}")
+        replay = run_json([sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "complete"], environment)
+        if replay.get("ok") is not False or replay.get("code") != "INVALID_LEASE_TRANSITION":
+            raise AssertionError(f"consumed recovery challenge re-promoted: {replay!r}")
+    finally:
+        if process.poll() is None:
+            process.terminate()
+            try:
+                process.wait(timeout=3.0)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+        shutil.rmtree(root, ignore_errors=True)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--runs", type=int, default=3)
+    arguments = parser.parse_args()
+    if arguments.runs != 3:
+        raise SystemExit("P6 requires exactly three isolated runs")
+    for index, runtime in enumerate(("pi", "claude", "pi")):
+        run_once(index, runtime)
+    print("P6 constrained recovery probe PASS: 3 isolated shipped recovery-command runs")
+
+
+if __name__ == "__main__":
+    main()

docs/guides/lease-broker-operations.md

@@ -17,7 +17,9 @@ export MOSAIC_LEASE_BROKER_SOCKET=/run/user/1000/mosaic-lease/broker.sock
 mosaic claude # or: mosaic claudex, mosaic yolo claudex, mosaic pi
 ```
 
-The wrapper obtains a broker-minted session ID and `exec`s the runtime without changing its PID/starttime anchor. The all-tools Claude `PreToolUse` hook and Pi `tool_call` handler inherit that identity. Claudex retains its isolated proxy environment and config directory; Mosaic merges the mandatory all-tools hook into that isolated `settings.json` before invoking the same wrapper. PRDY init/update, QA remediation, coord, orchestrator, and fleet launchers also converge on this boundary. Broker registration failure, unsafe isolated settings, or missing identity denies launch/tool execution fail-closed; broker timeout/unavailability and malformed replies also block tools.
+The wrapper obtains a broker-minted session ID, creates a private `generation-<session>.state` file beside the socket, and `exec`s the runtime without changing its PID/starttime anchor. The all-tools Claude `PreToolUse` hook and Pi `tool_call` handler inherit that identity and read the current generation from the file. Claudex retains its isolated proxy environment and config directory; Mosaic merges the mandatory all-tools and compaction-lifecycle hooks into that isolated `settings.json` before invoking the same wrapper. PRDY init/update, QA remediation, coord, orchestrator, and fleet launchers also converge on this boundary. Broker registration failure, unsafe isolated settings, unsafe generation state, or missing identity denies launch/tool execution fail-closed; broker timeout/unavailability and malformed replies also block tools.
+
+Claude `PreCompact` and `SessionStart(compact)` hooks and Pi pre-/post-compaction handlers invoke `revoke-lease.py`. Pi `session_start` reload/new/resume/fork and Claude resume/clear advance the locked generation before revocation, so a replacement session inherits no lease even when PID/starttime stay unchanged. Do not invoke the revoker manually as a way to restore authority; it only removes authority. If a lifecycle hook reports failure, stop consequential work and repair broker/generation-state availability before re-verification.
 
 Run the permanent launch inventory locally with:
 
@@ -29,7 +31,13 @@ The same check runs in the Mosaic package test suite and therefore in root CI. A
 
 Clients must complete the request boundary before waiting for a reply. After sending the single JSON object and its terminating newline, the client **MUST half-close the socket's write side** (`shutdown(SHUT_WR)` in POSIX clients; `socket.end()` in Node) and only then await the response. Merely calling `write()` and waiting is invalid: the broker waits for EOF to enforce the exact-one-frame contract and fails closed at its one-second deadline. Do not replace `end()` with `write()` in client helpers. A delayed second frame remains malformed and is rejected.
 
-There is no automated recovery workflow yet. `mosaic_context_recover` is reserved as the only unverified mutator class, but its fixed payload/receipt implementation lands in a later WI. After a crash, preserve the protected state file and restart only after verifying that no broker owns the socket. Restart intentionally clears all volatile VERIFIED leases. A leftover socket requires an operator to verify the owning service is stopped and remove that exact socket deliberately. Corrupt, oversized, symlinked, or non-regular state fails closed; do not overwrite it. Preserve it for incident review and establish new state only through an explicit operational decision, which invalidates prior sessions and tokens.
+`mosaic_context_recover` is the only unverified mutator class. Its durable `mosaic-context-refresh` skill is a thin wrapper over `tools/lease-broker/recover-context.py`: `begin` has the broker rebuild the validated `B_payload`/`H_payload`, revoke first, and mint a new `PENDING_DELIVERY` receipt challenge; `complete` accepts neither receipt text nor a challenge argument. Claude maps only the exact direct recovery executable/validated arguments to this exempt tool identity; ordinary `Bash` remains gated. Pi exposes only the `mosaic_context_recover` custom tool; ordinary `bash` and all other tools remain gated. A normal-path receipt cannot be replayed through recovery because each retry begins a distinct recovery cycle and recovery completion cannot receive caller-presented evidence.
+
+Production daemon startup creates a separate private observer socket unless a test-only `--test-observer-file` fixture is selected. Claude's Stop hook sends its exact latest assistant entry and Pi's `message_end` handler sends only finalized assistant content to that authenticated transport; the broker public socket never accepts message text. This is byte-build and private out-of-process harness wiring only: do not activate it against a live daemon, live socket, systemd service, tmux session, or model-output stream outside the controlled integration procedure.
+
+Receipt honesty is load-bearing: absent, malformed, prefix-truncated, and observable adapter-mutated terminal receipts do not promote. A tail-only case is non-promoting only where the concrete terminal payload is malformed or observably incomplete. A tail-preserving middle drop is **not receipt-detectable**; it is the disclosed T-C injection-contract residual deferred to WI-7 server-side evidence. The receipt remains a T-A delivery/liveness prerequisite, never a safety, obedience, or residency proof. The framework skill is source-resident and bridge-projected on install/upgrade; do not hand-create a live runtime symlink.
+
+After a runtime exits, its `generation-<session>.state` file may be removed only after verifying that no process for that broker-minted session remains; stale files carry no lease authority but should be retained during incident analysis. After a broker crash, preserve the protected state file and restart only after verifying that no broker owns the socket. Restart intentionally clears all volatile VERIFIED leases. A leftover socket requires an operator to verify the owning service is stopped and remove that exact socket deliberately. Corrupt, oversized, symlinked, or non-regular state fails closed; do not overwrite it. Preserve it for incident review and establish new state only through an explicit operational decision, which invalidates prior sessions and tokens.
 
 ## Security posture
 

docs/reports/compaction-refresh/830-documentation-checklist.md

@@ -0,0 +1,34 @@
+# #830 Documentation Completion Checklist
+
+## Required artifacts
+
+- [x] `docs/PRD.md` contains the M1 compaction-refresh trust-lifecycle requirements and acceptance criteria.
+- [x] Operator behavior and recovery are documented in `docs/guides/lease-broker-operations.md`.
+- [x] Developer architecture and protocol behavior are documented in `docs/architecture/compaction-revocation.md`, `lease-broker-protocol.md`, and `mutator-class-gate.md`.
+- [x] Security boundaries and residuals are documented in `docs/architecture/lease-broker-security.md` and `compaction-revocation.md`.
+- [x] `docs/SITEMAP.md` links the new architecture page.
+- [x] User-guide changes are not applicable: observers are mandatory internal runtime controls with no end-user workflow.
+- [x] OpenAPI/endpoint changes are not applicable: the broker remains an internal Unix-socket protocol, not a public HTTP API.
+
+## Contract coverage
+
+- [x] Claude and Claudex lifecycle signals, matchers, commands, and fail-closed behavior are documented.
+- [x] Pi pre-/post-compaction signals and session replacement reasons are documented.
+- [x] Private generation-file ownership, monotonic update, same-PID replacement, and failure fencing are documented.
+- [x] `revoke_lease` input purpose, broker response state, and denial behavior are documented.
+- [x] T12b/T30 explicitly names the bounded residual stale window and reports within-TTL **ALLOWED** / after-TTL **DENIED**.
+- [x] Documentation explicitly disclaims a within-window mutator-action bound.
+- [x] T-A, T-C, same-principal, and protected-branch boundaries are retained.
+
+## Structure and review
+
+- [x] New architecture content is under `docs/architecture/`.
+- [x] This report is under `docs/reports/compaction-refresh/`.
+- [x] Session evidence is under `docs/scratchpads/`.
+- [x] Documentation changes are in the same logical change set as code and tests.
+- [ ] Independent exact-head code and Opus security reviews pending coordinator sequencing after the deterministic-main rebase gate.
+
+## Publishing
+
+- [x] Canonical documentation remains in-repository.
+- [x] No external publishing target is required for this internal M1 control.

docs/scratchpads/830-compaction-revoke.md

@@ -0,0 +1,95 @@
+# WI-3 Scratchpad — Compaction revocation and runtime-generation rollover
+
+- **Issue:** Gitea #830
+- **Branch:** `feat/830-compaction-revoke`
+- **Base HEAD:** `abd2791f59b3f06f46dd08e55298ced72f6aa7c2`
+- **Role:** sol author/build lane only; terra CODE and Opus SECREV are coordinator-owned.
+
+## Mission prompt
+
+Implement BUILD-BRIEF Deliverable 3.3 and D4 on merged WI-1/WI-2 under `packages/mosaic/`. Claude `PreCompact` and `SessionStart(matcher=compact)` plus Pi `session_before_compact`/`context` equivalents must revoke the active lease through the existing broker state machine. Any `runtime_generation` bump—including same-PID reload/resume/fork—must auto-revoke the prior incarnation so the new generation inherits no prior lease. M1 is Claude + Pi only.
+
+Honor amended D2-v5 exactly: hard fail-closure when at least one observer fires or after lease expiry; both observers missing within TTL is an explicitly named bounded residual stale window (maximum 300 seconds, soak-tighten only), with no claim that the mutator gate bounds actions inside that window; total gate-hook miss is T-C. T12b/T30 must report both the within-TTL ALLOWED outcome and after-TTL DENIED outcome.
+
+## Session start verification
+
+- Worktree is clean on `feat/830-compaction-revoke` at exact required base `abd2791f59b3f06f46dd08e55298ced72f6aa7c2`; `origin/main` is the same SHA and includes merged WI-2 atop WI-1.
+- Authority SHA-256 verified:
+  - BUILD-BRIEF: `89fdbc27ed0e5050dc7b52f3ef2ddaea691edf17fd89d51b15e26fb5ed47171b`
+  - SPEC-v5: `a6d07ade835758e8488ca10d3b0631caf0beb93ea3a6733631f151b0c2f01433`
+  - Ratification: `bac58319c9c4028b5b40e1129e0033cdb5a6b7b02033c25f06f4cb77d7779c67`
+  - sol red-team: `3da326a4ea91767b731e128a93b13194e8002358101e30de3fcb8ca2f8f54faa`
+- WI-0 evidence pack SHA-256 `5d418306fcc597fd514e500bee40d1509f0bf467e46ee13fc5c280ed8274759d` read directly. Probe P3 is **PASS**: real Pi retained the same PID/starttime through reload/fork/new/resume while generations advanced and a prior VERIFIED generation was revoked.
+- P6 planner-return ruling SHA-256 `b7bbb6ea6e8d9a5c3366993642ab4e4f65b961af04936dcac20bfbcdcbaf1a09` read directly: feature WI admission is GO with the exact-delivery empirical compatibility fact and disclosed T-C middle-drop residual; no receipt redesign.
+
+## Plan and budget
+
+1. RED real-socket acceptance for T12b/T30, each Claude observer, same-PID generation rollover, Claude/Claudex hook wiring, and Pi lifecycle wiring.
+2. Add one broker client executable for observer revocation plus a private monotonic generation-file helper shared by launcher, gate, and revoker.
+3. Wire Claude `PreCompact`, `SessionStart(compact)`, and resume/clear generation rollover; merge equivalent mandatory hooks into isolated Claudex settings.
+4. Wire Pi pre/post compaction observers and reload/new/resume/fork generation rollover with local fail-closed tool blocking if lifecycle revocation fails.
+5. Document the D2-v5 bounded stale window without claiming the mutator gate bounds within-TTL actions; update protocol/security/operations/sitemap/checklist.
+6. Run focused real tests, independently measured executable coverage ≥85%, full repository gates, commit/push, open an unmerged `closes #830` PR, and hand off for terra CODE + mandatory Opus SECREV.
+
+Working estimate: **35K tokens**. No explicit hard cap was supplied; reduce refactor breadth before touching locked broker authority/state-machine semantics.
+
+## RED evidence
+
+- New T12b/T30 test already reports the inherited primitive honestly: within-TTL **ALLOWED**, after-TTL **DENIED**. The complete AC remains RED because the mandatory threat-contract document is absent.
+- Focused real-socket suite is RED with 7 expected failures: missing revoker executable (both Claude observers + generation bump), missing Claude/Pi wiring, missing isolated Claudex observers, and missing D2-v5 disclosure.
+- Branch-focused Python suite is RED on the wished generation initializer/resolver interfaces and missing `lease_generation.py` / `revoke-lease.py`.
+- Pi lifecycle suite is RED because the wished standalone `lease-lifecycle.ts` observer/generation module does not exist.
+
+## Locked discipline
+
+- RED-first T12b/T30 and observer/generation tests; test commit precedes implementation.
+- Reuse broker `revoke_lease`; do not fork identity, lease, or transition authority.
+- Preserve revoke-first/promote-last and WI-1/WI-2 reviewed state machine.
+- ≥85% attributable executable coverage with real tests.
+- No author self-review, no merge, no `--no-verify`.
+
+## Local implementation complete (push held)
+
+Implemented on the WI-3 base `abd2791f59b3f06f46dd08e55298ced72f6aa7c2` without changing the reviewed broker state machine:
+
+- Added `revoke-lease.py`, which authenticates through the existing broker session/generation and invokes `revoke_lease`. A fired observer that cannot confirm broker revocation advances the private generation as a local fence before returning non-zero.
+- Added `lease_generation.py`: owner/type/mode/size validation, no-follow opens, exclusive bump lock, monotonic `int64` generation, write-all + `fsync`, and fail-closed exhaustion/corruption handling.
+- `launch-runtime.py` creates `generation-<broker-session>.state` mode `0600` beside the socket before `exec`; `mutator-gate.py` resolves that current file value on every tool check.
+- Claude settings and isolated Claudex settings now preserve/install `PreCompact`, `SessionStart(compact)`, and resume/clear rollover hooks in addition to the global all-tools gate.
+- Pi now registers tested `session_before_compact`, `session_compact`→first `context`, and `session_start(reload|new|resume|fork)` handlers. Failed pre-compact revocation cancels compaction; failed post-compact/rollover revocation latches local all-tool denial.
+- Added PRD requirements, architecture/security/protocol/operations updates, sitemap entry, and the ignored-by-default documentation checklist (force-add required at commit).
+
+### Acceptance and coverage evidence
+
+- Focused acceptance: `19/19`; T12b/T30 prints within-TTL **ALLOWED** and after-TTL **DENIED**.
+- Pi lifecycle: `8/8`, with **100% statements/branches/functions/lines** attributable coverage.
+- New Python generation/revoker: `24/24`, **99% branch-aware aggregate coverage** (`lease_generation.py` 98%, `revoke-lease.py` 100%).
+- Mosaic package: `1399/1399`; framework shell Python `24/24`, launch guard `12/12`, permanent launch inventory `14 gated/14 total`.
+- Existing lease-broker real-socket acceptance: `37/37` within the package run.
+- Full repository: `43/43` Turbo tasks green; gateway `628 passed / 12 skipped`; Mosaic `1399/1399`.
+- Root typecheck: `42/42`; lint: `23/23`; format and `git diff --check` green.
+- Initial direct package test without first building the package reproduced the known missing-`dist/cli.js` harness condition; the canonical root Turbo test (which schedules `@mosaicstack/mosaic#build`) and explicit package build+test are green. No test was weakened.
+
+### Review evidence
+
+- Codex uncommitted code review: **APPROVE**, confidence `0.88`, zero findings. Its read-only sandbox could not rerun Vitest, but the author-side focused and full suites above were green.
+- Codex uncommitted security review: risk **none**, confidence `0.91`, zero findings.
+- Coordinator-mandated fresh exact-head terra CODE and Opus SECREV remain pending after rebase/push clearance; these local reviews do not replace that final gate.
+
+### Hold and residuals
+
+- **DO NOT PUSH OR OPEN A PR YET.** Coordinator requires flake-fix #838 to land, then WI-3 must rebase onto deterministic-green `main` before push.
+- Merge remains gated on #838, #827 Probe 3, and combined GO.
+- Named residual retained verbatim: when both observers are entirely missed, within-TTL consequential actions remain allowed; only lease expiry denies after the bounded stale window. No within-window mutator-action bound is claimed.
+
+## Deterministic-main rebase evidence
+
+- Fetched and confirmed `origin/main` at `8dfcf1903e385f977121069f798f476eb671fffc` (`#838` bounded broker deadlines, empty-read fail-closure, and de-flaked acceptance client).
+- Linear rebase completed. The only content conflict was `packages/mosaic/src/mutator-gate/runtime_tools_unittest.py`; resolution retained #838's `subprocess`/`threading` deadline regressions and WI-3's `stat` generation-state coverage. No authority/state-machine choice was ambiguous.
+- `packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts` auto-merged on top of #838's shared `requestBrokerReply` helper. No inline socket/`JSON.parse` client was resurrected.
+- Verified WI-3 has zero diff from `origin/main` for #838-owned `daemon.py`, `broker-test-client.ts`, `lease-broker.acceptance.spec.ts`, `vitest.config.ts`, and `packages/mosaic/package.json`; bounded deadlines and the de-flaked harness are preserved byte-for-byte.
+- Required verbose acceptance command: **2 files / 56 tests green**. T12b/T30 still prints within-TTL **ALLOWED** and after-TTL **DENIED**.
+- Full Mosaic package after explicit build: **74 files / 1408 tests green**; deadline unit `2/2`, runtime tools `25/25`, launch guard `12/12`, inventory `14/14`.
+- Full repository: **43/43 Turbo tasks green**. Root typecheck `42/42`, lint `23/23`, format and diff checks green.
+- Attributable coverage remains Python **99%** branch-aware and Pi lifecycle **100%** statements/branches/functions/lines.
+- Push and PR remain held pending combined GO and all WI-3 merge gates. The coordinator-owned promote-lease-lost-ACK SPEC amendment/backstop is acknowledged as a future merge prerequisite and was not retro-expanded into this core rebase/build.

docs/scratchpads/832-receipt-challenge-protocol.md

@@ -0,0 +1,13 @@
+# #832 Receipt-challenge protocol — build scratchpad
+
+- **Objective:** Deliver WI-5 receipt-challenge protocol ACs T25, T26, T28, and T29 only.
+- **Authority:** BUILD-BRIEF, SPEC-v5, ratification, and red-team hashes verified in STEP-0.
+- **Base:** `e522b22fa4492861b0fcd4a956a8795c54eb9bfe` (`origin/main`).
+- **Constraints:** Byte-build only: no live broker/socket/systemd/tmux mutation. No PR, self-review, or probe fire. T27/T30 are out of scope.
+- **Plan:**
+  1. Add red-first deterministic T26/T29 in-build tests that call shipped normative construction and broker path.
+  2. Add an unexecuted, isolated P5 out-of-process replay harness that drives the shipped daemon and asserts consume-before-promote for T25/T28.
+  3. Implement the broker-minted receipt challenge and exact receipt observation/consume/promote path.
+  4. Run unit, framework-shell, compile, lint, and type checks; push after the required queue guard; report to `mosaic-100`.
+- **Risks:** The standalone harness must drive the real daemon without a divergent fixture. If that is impossible, stop and flag Mos.
+- **Evidence:** Initial RED recorded in `/home/hermes/agent-work/reviews/832-wi5-red-receipt-challenge.log`; initial green checks passed. Remediation RED recorded in `/home/hermes/agent-work/reviews/832-wi5-remediation-red.log` before observer/payload implementation; remediation green passed. Remediation-2 RED recorded in `/home/hermes/agent-work/reviews/832-wi5-remediation2-red.log`: each rejected begin restored prior VERIFIED authority. Remediation-2 GREEN: receipt unittest (5: all `INVALID_CONSTRUCTION`, `PAYLOAD_CONSTRUCTION_REFUSED`, and `PAYLOAD_BINDING_MISMATCH` cases preserve UNVERIFIED and deny the next mutator), normative-fragments unittest (5), state-store regression (10), full mutator-gate acceptance (20, including the real begin → observer → consume → promote path), `py_compile`, Mosaic package lint/typecheck, and targeted Prettier check. The P5 harness remains unfired. Coverage tooling remains unavailable (`python3 -m coverage`: module not installed). Push pending.

docs/scratchpads/833-constrained-recovery-command.md

@@ -0,0 +1,15 @@
+# #833 constrained recovery command — build scratchpad
+
+- **Objective:** Deliver WI-6 plus Mos-ruled B1/B2 and R2 Claude-only literal-argv repair: no shell-active recovery mapping bypass, unchanged Pi gate/B2 observer, AC-1/C4 preservation, and an unfired P6 probe.
+- **Authority:** STEP-0 SHA-256 verified 4/4 against the supplied BUILD-BRIEF, SPEC-v5, ratification, and red-team records.
+- **Base:** exact `07553ead337a70a9241f826d27571650262b289c`; new branch `feat/833-constrained-recovery-command`; merge-base assertion passed before any commit.
+- **Constraints:** No rebase/pull during build; no live install/symlink or live broker/socket/tmux/systemd/model-stream activation; no self-review, PR, merge, push, or P6 fire. `docs/TASKS.md` is orchestrator-owned and will not be modified.
+- **Plan:**
+  1. Add red-first unit tests against the recovery broker entrypoint for fresh recovery challenge, normal-receipt replay refusal, observable partial-delivery refusal, and the explicit middle-drop negative capability; commit the RED test and preserve its command output.
+  2. Implement the recovery command as a thin driver over shared WI-5 broker transitions and the trusted observer seam; it never accepts caller receipt text.
+  3. Add the source-resident skill under `packages/mosaic/framework/skills/`, plus a tmp-only #824 bridge projection test.
+  4. Build an unfired, default-3-run P6 standalone real-socket driver; it is not added to package scripts and will not be run.
+  5. Run targeted broker/mutator/receipt suites, lint, and format check; report head to `mosaic-100` and stop.
+- **Risks:** The observer can only represent an exact latest message. Tail-preserving middle-drop is intentionally not claimed receipt-detectable (T-C residual deferred to WI-7 server evidence).
+- **Evidence:** Original RED test committed at `3b5513bd6efad0c06b599fc759a66cff4286db04`; expected `UNKNOWN_ACTION` is logged in `/home/hermes/agent-work/reviews/833-wi6-red.log`. B1/B2 repair RED is `f4beedc3e7ac2e142dcbdeefb0e5ee40c20d9b86` in `/home/hermes/agent-work/reviews/833-wi6-repair-red.log`. R2 adversarial RED is committed at `65e2bd71cf360b6f45c86eec0b06ae24494832d8` in `/home/hermes/agent-work/reviews/833-wi6-repair-R2-red.log` before the literal-only gate source: the private real-gate/real-daemon battery covers every argv position (executable, path, phase, each flag, each value) for command substitution, backticks, parameter/arithmetic expansion, brace/tilde, process substitution, glob, redirects, control operations, embedded newline, and quotes. P6 remains rebuilt and unfired. The ordinary package Vitest/lint/typecheck/root-format commands cannot resolve their executables in this intentionally dependency-free fresh worktree (`node_modules` absent); no install/symlink workaround was used.
+- **Budget:** No explicit task token cap was supplied; scope is fixed to WI-6 and no unrelated behavior will be added.

packages/mosaic/framework/framework-manifest.txt

@@ -35,6 +35,8 @@ CONTRIBUTING.md
 defaults/**
 examples/**
 guides/**
+# Shipped framework subtree — canonical skills are upgrade-reconciled.
+skills/**
 install.sh
 install.ps1
 LICENSE
@@ -67,6 +69,9 @@ policy/**
 memory/**
 sources/**
 credentials/**
+# Operator-authored/customized skills live separately from canonical skills/ and
+# must remain structurally unprunable even as skills/** is framework-owned.
+skills-local/**
 # Secret-bearing operator file INSIDE the framework-owned tools/ subtree.
 # Listed explicitly so the deny-wins rule carves it out of tools/**.
 tools/_lib/credentials.json

packages/mosaic/framework/runtime/claude/settings.json

@@ -1,13 +1,44 @@
 {
   "model": "opus",
   "hooks": {
+    "PreCompact": [
+      {
+        "matcher": ".*",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 \"$HOME/.config/mosaic/tools/lease-broker/revoke-lease.py\" --runtime claude --reason pre-compact"
+          }
+        ]
+      }
+    ],
+    "SessionStart": [
+      {
+        "matcher": "compact",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 \"$HOME/.config/mosaic/tools/lease-broker/revoke-lease.py\" --runtime claude --reason session-start-compact"
+          }
+        ]
+      },
+      {
+        "matcher": "resume|clear",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 \"$HOME/.config/mosaic/tools/lease-broker/revoke-lease.py\" --runtime claude --reason session-start-rollover --bump-generation"
+          }
+        ]
+      }
+    ],
     "PreToolUse": [
       {
         "matcher": ".*",
         "hooks": [
           {
             "type": "command",
-            "command": "python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude",
+            "command": "python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude --recovery-command ~/.config/mosaic/tools/lease-broker/recover-context.py",
             "timeout": 3
           }
         ]
@@ -48,6 +79,11 @@
     "Stop": [
       {
         "hooks": [
+          {
+            "type": "command",
+            "command": "python3 ~/.config/mosaic/tools/lease-broker/receipt-observer-client.py --runtime claude --latest-entry",
+            "timeout": 3
+          },
           {
             "type": "command",
             "command": "~/.config/mosaic/tools/qa/reflect-stop-hook.sh",

packages/mosaic/framework/runtime/pi/lease-lifecycle.ts

@@ -0,0 +1,93 @@
+export type LeaseLifecycleRunner = (args: string[]) => boolean;
+
+type LifecycleEvent = {
+  reason?: unknown;
+  toolName?: unknown;
+};
+
+type LifecycleHandler = (
+  event: LifecycleEvent,
+  context: Record<string, unknown>,
+) => unknown | Promise<unknown>;
+
+export interface LeaseLifecyclePiApi {
+  on(event: string, handler: LifecycleHandler): void;
+}
+
+const ROLLOVER_REASONS = new Set(['reload', 'new', 'resume', 'fork']);
+
+function eventReason(event: LifecycleEvent): string {
+  return typeof event.reason === 'string' && event.reason.length > 0 ? event.reason : 'unknown';
+}
+
+/**
+ * Register redundant Pi compaction observers and same-PID generation rollover.
+ *
+ * A failed pre-compaction observer cancels compaction. A failed post-compaction
+ * observer or generation rollover locally blocks later tools in addition to the
+ * broker-backed all-tools gate.
+ */
+export function registerLeaseLifecycleHooks(
+  pi: LeaseLifecyclePiApi,
+  runRevoker: LeaseLifecycleRunner,
+): void {
+  let postCompactReason: string | null = null;
+  let postCompactFailure = false;
+  let rolloverFailure = false;
+
+  pi.on('session_before_compact', async (event) => {
+    const reason = eventReason(event);
+    const revoked = runRevoker([
+      '--runtime',
+      'pi',
+      '--reason',
+      `pi-session-before-compact:${reason}`,
+    ]);
+    if (!revoked) return { cancel: true };
+    return undefined;
+  });
+
+  pi.on('session_compact', async (event) => {
+    postCompactReason = eventReason(event);
+  });
+
+  pi.on('context', async () => {
+    if (postCompactReason === null) return undefined;
+    const reason = postCompactReason;
+    const revoked = runRevoker([
+      '--runtime',
+      'pi',
+      '--reason',
+      `pi-context-after-compact:${reason}`,
+    ]);
+    if (revoked) {
+      postCompactReason = null;
+      postCompactFailure = false;
+    } else {
+      postCompactFailure = true;
+    }
+    return undefined;
+  });
+
+  pi.on('session_start', async (event) => {
+    const reason = eventReason(event);
+    if (!ROLLOVER_REASONS.has(reason)) return undefined;
+    const revoked = runRevoker([
+      '--runtime',
+      'pi',
+      '--reason',
+      `pi-session-start:${reason}`,
+      '--bump-generation',
+    ]);
+    rolloverFailure = !revoked;
+    return undefined;
+  });
+
+  pi.on('tool_call', async () => {
+    if (!postCompactFailure && !rolloverFailure) return undefined;
+    return {
+      block: true,
+      reason: 'BLOCKED: Mosaic lease lifecycle revoke failed; runtime remains UNVERIFIED.',
+    };
+  });
+}

packages/mosaic/framework/runtime/pi/mosaic-extension.ts

@@ -22,6 +22,7 @@ import {
 import { join, basename } from 'node:path';
 import { homedir } from 'node:os';
 import { execSync, spawnSync } from 'node:child_process';
+import { registerLeaseLifecycleHooks, type LeaseLifecyclePiApi } from './lease-lifecycle.js';
 
 // ---------------------------------------------------------------------------
 // Config
@@ -29,6 +30,15 @@ import { execSync, spawnSync } from 'node:child_process';
 
 const MOSAIC_HOME = process.env['MOSAIC_HOME'] ?? join(homedir(), '.config', 'mosaic');
 const MUTATOR_GATE = join(MOSAIC_HOME, 'tools', 'lease-broker', 'mutator-gate.py');
+const LEASE_REVOKER = join(MOSAIC_HOME, 'tools', 'lease-broker', 'revoke-lease.py');
+const RECOVERY_COMMAND = join(MOSAIC_HOME, 'tools', 'lease-broker', 'recover-context.py');
+const RECEIPT_OBSERVER_CLIENT = join(
+  MOSAIC_HOME,
+  'tools',
+  'lease-broker',
+  'receipt-observer-client.py',
+);
+const RECOVERY_TOOL = 'mosaic_context_recover';
 
 // ---------------------------------------------------------------------------
 // Helpers
@@ -107,6 +117,15 @@ function nowIso(): string {
   return new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
 }
 
+function runPiLeaseRevoker(args: string[]): boolean {
+  const result = spawnSync('python3', [LEASE_REVOKER, ...args], {
+    encoding: 'utf8',
+    timeout: 2_000,
+    env: process.env,
+  });
+  return result.status === 0;
+}
+
 function checkPiMutatorGate(toolName: string): { block: true; reason: string } | undefined {
   const result = spawnSync('python3', [MUTATOR_GATE, '--runtime', 'pi'], {
     input: `${JSON.stringify({ tool_name: toolName })}\n`,
@@ -124,6 +143,78 @@ function checkPiMutatorGate(toolName: string): { block: true; reason: string } |
   };
 }
 
+function checkPiRecoveryGate(): { block: true; reason: string } | undefined {
+  return checkPiMutatorGate(RECOVERY_TOOL);
+}
+
+function assistantMessageText(message: unknown): string | undefined {
+  if (typeof message !== 'object' || message === null) return undefined;
+  const value = message as { role?: unknown; content?: unknown };
+  if (value.role !== 'assistant') return undefined;
+  if (typeof value.content === 'string') return value.content;
+  if (!Array.isArray(value.content)) return undefined;
+  const text: string[] = [];
+  for (const part of value.content) {
+    if (typeof part !== 'object' || part === null) return undefined;
+    const typed = part as { type?: unknown; text?: unknown };
+    if (typed.type !== 'text' || typeof typed.text !== 'string') return undefined;
+    text.push(typed.text);
+  }
+  return text.join('');
+}
+
+function recordPiMessageEnd(message: unknown): void {
+  const latestAssistantMessage = assistantMessageText(message);
+  if (latestAssistantMessage === undefined) return;
+  // This sends finalized Pi message_end content only to the daemon-owned
+  // authenticated observer transport, never to the public broker request API.
+  spawnSync('python3', [RECEIPT_OBSERVER_CLIENT, '--runtime', 'pi'], {
+    input: `${JSON.stringify({ latest_assistant_message: latestAssistantMessage })}\n`,
+    encoding: 'utf8',
+    timeout: 2_000,
+    env: process.env,
+  });
+}
+
+function runPiRecoveryCommand(params: {
+  phase: 'begin' | 'complete';
+  construction?: string;
+  compactionEpoch?: number;
+  requestEpoch?: number;
+}): { content: Array<{ type: 'text'; text: string }> } {
+  const args = [RECOVERY_COMMAND, params.phase];
+  if (params.phase === 'begin') {
+    if (
+      typeof params.construction !== 'string' ||
+      !Number.isInteger(params.compactionEpoch) ||
+      !Number.isInteger(params.requestEpoch) ||
+      params.compactionEpoch < 0 ||
+      params.requestEpoch < 0
+    ) {
+      return {
+        content: [
+          { type: 'text', text: 'Recovery begin requires construction and non-negative epochs.' },
+        ],
+      };
+    }
+    args.push(
+      '--construction',
+      params.construction,
+      '--compaction-epoch',
+      String(params.compactionEpoch),
+      '--request-epoch',
+      String(params.requestEpoch),
+    );
+  }
+  const result = spawnSync('python3', args, {
+    encoding: 'utf8',
+    timeout: 3_000,
+    env: process.env,
+  });
+  const output = result.status === 0 ? String(result.stdout ?? '') : String(result.stderr ?? '');
+  return { content: [{ type: 'text', text: output || 'Constrained recovery refused.' }] };
+}
+
 // ---------------------------------------------------------------------------
 // Mission detection
 // ---------------------------------------------------------------------------
@@ -268,11 +359,40 @@ export default function register(pi: ExtensionAPI) {
   let hbModel: string | null = null;
   let hbTimer: ReturnType<typeof setInterval> | null = null;
 
+  // ── Compaction observers and same-PID generation rollover ─────────────
+  registerLeaseLifecycleHooks(pi as unknown as LeaseLifecyclePiApi, runPiLeaseRevoker);
+
   // ── Whole mutator-class authorization gate ────────────────────────────
   // Every Pi tool, including unknown/custom tools, reaches the broker-backed
   // class gate before execution. Broker/script failure blocks fail-closed.
   pi.on('tool_call', async (event) => checkPiMutatorGate(event.toolName));
 
+  // Pi records only a finalized assistant entry at message_end. It never uses
+  // after_provider_response, which occurs before stream consumption.
+  pi.on('message_end', async (event) => {
+    recordPiMessageEnd((event as unknown as { message?: unknown }).message);
+  });
+
+  // The recovery custom tool is the only Pi invocation that maps to the
+  // broker's exempt RECOVERY_TOOL identity. It is not a Bash exception.
+  pi.registerTool({
+    name: RECOVERY_TOOL,
+    label: 'Mosaic Context Recovery',
+    description:
+      'Run the constrained broker-backed context recovery flow. This is the sole ungated mutator.',
+    parameters: Type.Object({
+      phase: Type.Union([Type.Literal('begin'), Type.Literal('complete')]),
+      construction: Type.Optional(Type.String()),
+      compactionEpoch: Type.Optional(Type.Integer({ minimum: 0 })),
+      requestEpoch: Type.Optional(Type.Integer({ minimum: 0 })),
+    }),
+    async execute(_toolCallId, params) {
+      const blocked = checkPiRecoveryGate();
+      if (blocked !== undefined) return { content: [{ type: 'text', text: blocked.reason }] };
+      return runPiRecoveryCommand(params);
+    },
+  });
+
   // ── Session Start ─────────────────────────────────────────────────────
   pi.on('session_start', async (_event, ctx) => {
     sessionCwd = process.cwd();

packages/mosaic/framework/skills/mosaic-context-refresh/SKILL.md

@@ -0,0 +1,65 @@
+---
+name: mosaic-context-refresh
+description: Run the constrained Mosaic context-recovery flow after compaction or directive-loss. This is a thin wrapper over the broker-backed recovery command; it never treats a receipt as a safety or residency proof.
+---
+
+# mosaic-context-refresh
+
+Use this only after compaction, session resume, or confirmed directive drift. It invokes the
+**single ungated mutator**, `tools/lease-broker/recover-context.py`; every other consequential
+mutator remains behind the verified lease gate.
+
+## Wrapper procedure
+
+1. The runtime supplies the exact validated normative-fragment construction and the current
+   compaction/request epochs.
+   - **Claude:** invoke only this direct command shape (no shell composition):
+
+     ```bash
+     python3 /absolute/path/to/mosaic/tools/lease-broker/recover-context.py begin --construction /absolute/path/to/mosaic-context-refresh-construction.json --compaction-epoch 0 --request-epoch 0
+     ```
+
+     This is a literal argv template: replace the recover-context.py path and construction JSON path
+     with the literal absolute paths for your install, then replace each epoch with literal decimal
+     digits. Do not use variables, quoting, globs, redirects,
+     shell operators, substitutions, or line continuations. Claude's all-tools gate maps only this
+     fully literal recovery shape to `mosaic_context_recover`; ordinary `Bash` remains gated.
+
+   - **Pi:** call the registered `mosaic_context_recover` tool with `phase: "begin"`,
+     `construction`, `compactionEpoch`, and `requestEpoch`. It is the exact broker-exempt tool name;
+     Pi `bash` and every other tool remain gated.
+
+   Both forms delegate to the shipped WI-5 broker transition: revoke first, build the canonical
+   `B_payload`/`H_payload`, enter `PENDING_DELIVERY`, and mint a fresh one-time challenge. They print
+   the terminal receipt envelope to deliver exactly as returned.
+
+2. The current assistant message copies that one terminal receipt verbatim. It does not compute a
+   hash, add prose, quote a prior receipt, or present a caller-supplied receipt/challenge.
+3. The production trusted-observer transport records that finalized assistant entry before completion:
+   - **Claude** selects the latest assistant entry at its `Stop` hook.
+   - **Pi** records only finalized assistant content at `message_end` (never
+     `after_provider_response`).
+
+   Then invoke completion with the same adapter form: Claude runs
+   `python3 /absolute/path/to/mosaic/tools/lease-broker/recover-context.py complete`; Pi calls
+   `mosaic_context_recover` with `phase: "complete"`. Completion supplies no receipt or challenge
+   argument. The broker observes the exact latest assistant entry, commits evidence, consumes its own
+   fresh challenge, and promotes VERIFIED last. If observation is absent, malformed, stale, or
+   duplicated, recovery remains UNVERIFIED and a retry begins a new cycle.
+
+## Scope and honesty
+
+- A receipt from the normal verification path cannot be replayed through recovery: recovery mints a
+  distinct current challenge and does not accept caller-provided receipt text as evidence.
+- Observable absent, malformed, prefix-truncated, and adapter-mutated terminal receipts do not
+  promote. “Tail-only” is non-promoting only when the delivered terminal bytes are concretely
+  malformed or incomplete.
+- **Negative capability:** a tail-preserving middle drop is not represented as receipt-detectable.
+  It is a T-C injection-contract residual deferred to WI-7 server-side evidence; do not claim this
+  skill or receipt catches it.
+- The receipt is a T-A delivery/liveness prerequisite only. It never proves obedience, comprehension,
+  durable residency, or safety; the whole mutator-class gate and server-side branch protection retain
+  those roles.
+
+This source-resident skill is projected by the Mosaic skill bridge after framework install/upgrade.
+Do not create a live symlink manually.

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

@@ -6,10 +6,12 @@ from __future__ import annotations
 import argparse
 import copy
 import errno
+import hmac
 from concurrent.futures import ThreadPoolExecutor
 import json
 import os
 import secrets
+import select
 import signal
 import socket
 import stat
@@ -20,6 +22,19 @@ import time
 from pathlib import Path
 from typing import Final
 
+# This script is loaded both as an executable and through isolated stdlib tests.
+# Keep its co-located receipt implementation importable in both modes.
+_MODULE_DIRECTORY = str(Path(__file__).resolve().parent)
+if _MODULE_DIRECTORY not in sys.path:
+    sys.path.insert(0, _MODULE_DIRECTORY)
+from normative_fragments import build_payload_from_wire
+from receipt_challenge import is_verbatim_receipt, latest_assistant_digest, receipt_for
+from receipt_observer import (
+    FileTestReceiptObserver,
+    ReceiptObserver,
+    RuntimeReceiptObserver,
+)
+
 MAX_FRAME: Final = 64 * 1024
 MAX_STATE: Final = 4 * 1024 * 1024
 MAX_PENDING_TOKENS: Final = 256
@@ -90,6 +105,14 @@ def valid_binding(binding: object) -> bool:
     )
 
 
+def valid_receipt_evidence(evidence: object) -> bool:
+    return (
+        isinstance(evidence, dict)
+        and set(evidence) == {"h_latest_assistant"}
+        and is_hex_256(evidence["h_latest_assistant"])
+    )
+
+
 def validate_state(value: object) -> dict[str, object]:
     if not isinstance(value, dict) or set(value) != {"version", "sessions", "tokens"}:
         raise BrokerFailure("STATE_INTEGRITY")
@@ -124,7 +147,11 @@ def validate_state(value: object) -> dict[str, object]:
     for token_value, token in tokens.items():
         if not is_hex_256(token_value) or not isinstance(token, dict):
             raise BrokerFailure("STATE_INTEGRITY")
-        if set(token) != {"session_id", "runtime_generation", "binding", "consumed"}:
+        token_fields = set(token)
+        if token_fields not in (
+            {"session_id", "runtime_generation", "binding", "consumed"},
+            {"session_id", "runtime_generation", "binding", "consumed", "evidence"},
+        ):
             raise BrokerFailure("STATE_INTEGRITY")
         session_id = token["session_id"]
         generation = token["runtime_generation"]
@@ -137,6 +164,9 @@ def validate_state(value: object) -> dict[str, object]:
             raise BrokerFailure("STATE_INTEGRITY")
         if not valid_binding(token["binding"]) or token["consumed"] is not False:
             raise BrokerFailure("STATE_INTEGRITY")
+        evidence = token.get("evidence")
+        if evidence is not None and not valid_receipt_evidence(evidence):
+            raise BrokerFailure("STATE_INTEGRITY")
     return value
 
 
@@ -275,8 +305,15 @@ class StateStore:
 
 
 class Broker:
-    def __init__(self, store: StateStore) -> None:
+    def __init__(self, store: StateStore, observer: ReceiptObserver | None = None) -> None:
         self.store = store
+        # Production construction always has a transport-capable observer. Test
+        # fixtures may inject their controlled observer explicitly.
+        self.observer: ReceiptObserver = observer if observer is not None else RuntimeReceiptObserver()
+        # Set only by begin_verification after its mandatory revoke-first fence.
+        # It is preserved if later cycle admission is refused; all other broker
+        # actions retain the normal snapshot rollback behavior.
+        self._rejected_cycle_fence: tuple[dict[str, object], dict[str, dict[str, object]]] | None = None
         # VERIFIED authority is deliberately volatile: broker restart revokes all
         # leases while preserving WI-1 identity and pending-token integrity.
         self.leases: dict[str, dict[str, object]] = {}
@@ -348,6 +385,9 @@ class Broker:
             "runtime_generation": generation,
             "binding": copy.deepcopy(binding),
             "consumed": False,
+            # Receipt evidence is durably committed before this challenge may
+            # be consumed and promotion made externally visible.
+            "evidence": None,
         }
         return token
 
@@ -357,16 +397,51 @@ class Broker:
             raise BrokerFailure("STATE_INTEGRITY")
         lease["state"] = LEASE_VERIFIED
 
+    def record_runtime_observation(
+        self, peer_pid: int, request: dict[str, object]
+    ) -> dict[str, object]:
+        """Record only an authenticated adapter's finalized assistant message.
+
+        This method is deliberately unreachable through ``Broker.handle`` and
+        its public broker socket. The production observer socket calls it after
+        SO_PEERCRED/ancestry authentication, preserving the S1 rule that a
+        broker request can never carry ``latest_assistant_message``.
+        """
+
+        required = {
+            "action", "session_id", "runtime_generation", "runtime", "latest_assistant_message"
+        }
+        if set(request) != required or request.get("action") != "record_runtime_observation":
+            raise BrokerFailure("INVALID_OBSERVATION")
+        runtime = request.get("runtime")
+        message = request.get("latest_assistant_message")
+        if runtime not in READ_ONLY_TOOLS or not isinstance(message, str) or len(message.encode("utf-8")) > MAX_FRAME:
+            raise BrokerFailure("INVALID_OBSERVATION")
+        session_id, _ = self.authenticate(peer_pid, request)
+        generation = request["runtime_generation"]
+        lease = self.leases.get(session_id)
+        if (
+            not isinstance(lease, dict)
+            or lease.get("state") != LEASE_PENDING
+            or lease.get("runtime") != runtime
+            or lease.get("runtime_generation") != generation
+            or not isinstance(self.observer, RuntimeReceiptObserver)
+        ):
+            raise BrokerFailure("OBSERVATION_UNAVAILABLE")
+        self.observer.record_latest_assistant_message(session_id, runtime, generation, message)
+        return {"ok": True}
+
     def handle(self, peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
         if self.store.poisoned:
             raise StateCommitUncertain()
         previous = copy.deepcopy(self.store.value)
         previous_leases = copy.deepcopy(self.leases)
+        self._rejected_cycle_fence = None
         try:
             response = self._handle(peer, request)
             if self.store.value != previous:
                 self.store.commit()
-            if request.get("action") == "promote_lease":
+            if request.get("action") in {"promote_lease", "complete_recovery"}:
                 session_id = request.get("session_id")
                 if not isinstance(session_id, str):
                     raise BrokerFailure("INVALID_IDENTITY")
@@ -376,9 +451,18 @@ class Broker:
         except StateCommitUncertain:
             raise
         except Exception:
-            self.store.value = previous
-            self.leases = previous_leases
+            fence = self._rejected_cycle_fence
+            if fence is None:
+                self.store.value = previous
+                self.leases = previous_leases
+            else:
+                # A refused re-verification must never resurrect the preceding
+                # VERIFIED authority. Preserve only this post-revoke fence;
+                # every unrelated partial-write failure still rolls back.
+                self.store.value, self.leases = fence
             raise
+        finally:
+            self._rejected_cycle_fence = None
 
     def _handle(self, peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
         peer_pid, peer_uid, peer_gid = peer
@@ -426,10 +510,76 @@ class Broker:
                 raise BrokerFailure("TOKEN_REPLAY")
             del self.store.tokens()[token_value]
             return {"ok": True}
+        if action == "begin_recovery":
+            # Recovery is the one ungated mutator, but it is not a second
+            # receipt protocol. It delegates to this exact normal-path
+            # transition, then marks its volatile pending cycle so completion
+            # can obtain the broker-minted challenge internally. Caller-supplied
+            # receipt text is never an input to recovery.
+            if any(field in request for field in ("receipt", "latest_assistant_message", "receipt_challenge")):
+                raise BrokerFailure("INVALID_RECOVERY_REQUEST")
+            normal_request = dict(request)
+            normal_request["action"] = "begin_verification"
+            response = self._handle(peer, normal_request)
+            session_id = response.get("session_id")
+            if not isinstance(session_id, str):
+                # begin_verification deliberately does not return identity;
+                # recover it only after its authenticated shared transition.
+                candidate = request.get("session_id")
+                if not isinstance(candidate, str):
+                    raise BrokerFailure("INVALID_IDENTITY")
+                session_id = candidate
+            lease = self.leases.get(session_id)
+            if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING:
+                raise BrokerFailure("STATE_INTEGRITY")
+            lease["cycle_kind"] = "recovery"
+            response["state"] = "PENDING_DELIVERY"
+            return response
+        if action == "complete_recovery":
+            if any(field in request for field in ("receipt", "latest_assistant_message", "receipt_challenge")):
+                raise BrokerFailure("INVALID_RECOVERY_REQUEST")
+            session_id, _ = self.authenticate(peer_pid, request)
+            lease = self.leases.get(session_id)
+            challenge = lease.get("receipt_challenge") if isinstance(lease, dict) else None
+            if (
+                not isinstance(lease, dict)
+                or lease.get("cycle_kind") != "recovery"
+                or lease.get("state") != LEASE_PENDING
+                or not isinstance(challenge, str)
+            ):
+                raise BrokerFailure("INVALID_LEASE_TRANSITION")
+            try:
+                # Reuse the shipped observe -> evidence commit -> consume ->
+                # promote transition. The recovery caller supplies neither a
+                # normal-path receipt nor a challenge; the trusted observer
+                # and current broker cycle remain the sole evidence authority.
+                observed_request = {
+                    "action": "observe_receipt",
+                    "session_id": session_id,
+                    "runtime_generation": request["runtime_generation"],
+                    "receipt_challenge": challenge,
+                }
+                self._handle(peer, observed_request)
+                return self._handle(peer, {
+                    "action": "promote_lease",
+                    "session_id": session_id,
+                    "runtime_generation": request["runtime_generation"],
+                    "receipt_challenge": challenge,
+                })
+            except Exception:
+                # A malformed, absent, or stale observed receipt must leave
+                # no pending recovery capability or live lease. A retry mints
+                # a new challenge through the shared begin transition.
+                self.revoke_session_authority(session_id)
+                self._rejected_cycle_fence = (
+                    copy.deepcopy(self.store.value), copy.deepcopy(self.leases)
+                )
+                raise
         if action == "begin_verification":
             session_id, _ = self.authenticate(peer_pid, request)
             runtime = request.get("runtime")
             binding = request.get("binding")
+            construction = request.get("construction")
             ttl_seconds = request.get("ttl_seconds", MAX_LEASE_TTL_SECONDS)
             if runtime not in READ_ONLY_TOOLS:
                 raise BrokerFailure("INVALID_RUNTIME")
@@ -443,47 +593,117 @@ class Broker:
                 raise BrokerFailure("INVALID_LEASE_TTL")
             # Revoke-first is a broker operation, not advisory adapter order.
             self.revoke_session_authority(session_id)
-            token = self.mint_token(session_id, request["runtime_generation"], binding)
+            # If subsequent construction admission rejects, handle() restores
+            # this fence rather than the pre-cycle VERIFIED snapshot.
+            self._rejected_cycle_fence = (
+                copy.deepcopy(self.store.value), copy.deepcopy(self.leases)
+            )
+            try:
+                constructed = build_payload_from_wire(construction)
+            except ValueError as exc:
+                raise BrokerFailure("INVALID_CONSTRUCTION") from exc
+            if (
+                constructed.injectionDecision != "ACCEPTED"
+                or not constructed.promotion
+                or not isinstance(constructed.h_source, str)
+                or not isinstance(constructed.h_payload, str)
+            ):
+                raise BrokerFailure("PAYLOAD_CONSTRUCTION_REFUSED")
+            if (
+                not hmac.compare_digest(binding["h_source"], constructed.h_source)
+                or not hmac.compare_digest(binding["h_payload"], constructed.h_payload)
+            ):
+                raise BrokerFailure("PAYLOAD_BINDING_MISMATCH")
+            cycle_binding = copy.deepcopy(binding)
+            cycle_binding["runtime_generation"] = request["runtime_generation"]
+            challenge = self.mint_token(session_id, request["runtime_generation"], binding)
             self.leases[session_id] = {
                 "state": LEASE_PENDING,
                 "runtime": runtime,
                 "runtime_generation": request["runtime_generation"],
                 "binding": copy.deepcopy(binding),
-                "promotion_token": token,
+                "receipt_challenge": challenge,
                 "ttl_seconds": ttl_seconds,
             }
+            # The broker constructs both the bound challenge and the delivery
+            # text. The model's role is an exact copy, never hashing its output.
             return {
                 "ok": True,
                 "state": LEASE_PENDING,
-                "promotion_token": token,
+                "receipt_challenge": challenge,
+                "receipt": receipt_for(challenge, cycle_binding),
+                "binding": cycle_binding,
             }
-        if action == "promote_lease":
+        if action == "observe_receipt":
             session_id, _ = self.authenticate(peer_pid, request)
-            promotion_token = request.get("promotion_token")
-            lease = self.leases.get(session_id)
-            if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING:
-                raise BrokerFailure("INVALID_LEASE_TRANSITION")
-            expected_token = lease.get("promotion_token")
-            if (
-                not isinstance(promotion_token, str)
-                or not isinstance(expected_token, str)
-                or not secrets.compare_digest(promotion_token, expected_token)
-            ):
-                raise BrokerFailure("PROMOTION_TOKEN_MISMATCH")
-            token = self.store.tokens().get(promotion_token)
+            challenge = request.get("receipt_challenge")
+            if "latest_assistant_message" in request:
+                raise BrokerFailure("INVALID_RECEIPT")
+            if not isinstance(challenge, str):
+                raise BrokerFailure("INVALID_RECEIPT")
+            token = self.store.tokens().get(challenge)
             if (
                 not isinstance(token, dict)
                 or token.get("session_id") != session_id
                 or token.get("runtime_generation") != request.get("runtime_generation")
-                or token.get("binding") != lease.get("binding")
                 or token.get("consumed") is not False
             ):
-                raise BrokerFailure("PROMOTION_TOKEN_INVALID")
-            del self.store.tokens()[promotion_token]
+                raise BrokerFailure("RECEIPT_REPLAY")
+            lease = self.leases.get(session_id)
+            if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING:
+                raise BrokerFailure("RECEIPT_REPLAY")
+            expected_challenge = lease.get("receipt_challenge")
+            binding = lease.get("binding")
+            if (
+                not isinstance(expected_challenge, str)
+                or not secrets.compare_digest(challenge, expected_challenge)
+                or not isinstance(binding, dict)
+                or token.get("binding") != binding
+            ):
+                raise BrokerFailure("RECEIPT_REPLAY")
+            receipt_binding = copy.deepcopy(binding)
+            receipt_binding["runtime_generation"] = request["runtime_generation"]
+            message = self.observer.observe_latest_assistant_message(
+                session_id, str(lease["runtime"]), request["runtime_generation"], receipt_binding
+            )
+            if not isinstance(message, str):
+                raise BrokerFailure("RECEIPT_OBSERVATION_UNAVAILABLE")
+            # This accepts one exact current-cycle assistant entry only. It is
+            # deliberately not a transcript search and rejects quoted/extra text.
+            if not is_verbatim_receipt(message, challenge, receipt_binding):
+                raise BrokerFailure("RECEIPT_MISMATCH")
+            evidence = {"h_latest_assistant": latest_assistant_digest(message)}
+            token["evidence"] = evidence
             lease["state"] = LEASE_PENDING_PROMOTION
+            lease["evidence"] = copy.deepcopy(evidence)
+            return {"ok": True, "state": LEASE_PENDING_PROMOTION}
+        if action == "promote_lease":
+            session_id, _ = self.authenticate(peer_pid, request)
+            challenge = request.get("receipt_challenge")
+            if not isinstance(challenge, str):
+                raise BrokerFailure("INVALID_RECEIPT")
+            lease = self.leases.get(session_id)
+            if not isinstance(lease, dict) or lease.get("state") == LEASE_UNVERIFIED:
+                raise BrokerFailure("INVALID_LEASE_TRANSITION")
+            token = self.store.tokens().get(challenge)
+            if (
+                not isinstance(token, dict)
+                or token.get("session_id") != session_id
+                or token.get("runtime_generation") != request.get("runtime_generation")
+                or token.get("consumed") is not False
+            ):
+                raise BrokerFailure("RECEIPT_REPLAY")
+            if lease.get("state") != LEASE_PENDING_PROMOTION:
+                raise BrokerFailure("INVALID_LEASE_TRANSITION")
+            expected_challenge = lease.get("receipt_challenge")
+            if not isinstance(expected_challenge, str) or not secrets.compare_digest(challenge, expected_challenge):
+                raise BrokerFailure("RECEIPT_REPLAY")
+            if token.get("binding") != lease.get("binding") or not valid_receipt_evidence(token.get("evidence")):
+                raise BrokerFailure("PROMOTION_TOKEN_INVALID")
+            del self.store.tokens()[challenge]
             lease["expires_at"] = time.monotonic() + int(lease["ttl_seconds"])
-            # handle() commits token consumption before finish_promotion() makes
-            # VERIFIED externally visible: promote-last by construction.
+            # handle() commits the evidence-backed consumption before
+            # finish_promotion() makes VERIFIED externally visible: promote-last.
             return {"ok": True, "state": LEASE_PENDING_PROMOTION}
         if action == "revoke_lease":
             session_id, _ = self.authenticate(peer_pid, request)
@@ -604,12 +824,65 @@ def handle_connection(
             return
 
 
-def serve(socket_path: Path, state_path: Path) -> None:
+def handle_runtime_observation_connection(
+    connection: socket.socket,
+    broker: Broker,
+    broker_lock: threading.Lock,
+) -> None:
+    """Serve the authenticated production observer transport, never the broker API."""
+
+    with connection:
+        read_deadline = time.monotonic() + READ_DEADLINE_SECONDS
+        try:
+            raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
+            peer = struct.unpack("3i", raw)
+            request = read_frame(connection, read_deadline)
+        except BrokerFailure as exc:
+            reply = {"ok": False, "code": exc.code}
+        except OSError:
+            return
+        else:
+            acquired = broker_lock.acquire(timeout=HANDLE_QUEUE_TIMEOUT_SECONDS)
+            if not acquired:
+                reply = {"ok": False, "code": "BROKER_BUSY"}
+            else:
+                try:
+                    try:
+                        reply = broker.record_runtime_observation(peer[0], request)
+                    except BrokerFailure as exc:
+                        reply = {"ok": False, "code": exc.code}
+                finally:
+                    broker_lock.release()
+        try:
+            connection.settimeout(SEND_TIMEOUT_SECONDS)
+            connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode())
+        except OSError:
+            return
+
+
+def serve(
+    socket_path: Path,
+    state_path: Path,
+    observer: ReceiptObserver | None = None,
+    observer_socket_path: Path | None = None,
+) -> None:
     secure_parent(socket_path)
     if socket_path.exists() or socket_path.is_symlink():
         raise BrokerFailure("SOCKET_ALREADY_EXISTS")
+    runtime_observer = observer is None
+    if runtime_observer:
+        observer = RuntimeReceiptObserver()
+        observer_socket_path = observer_socket_path or socket_path.with_name("receipt-observer.sock")
+        if observer_socket_path == socket_path:
+            raise BrokerFailure("OBSERVER_SOCKET_CONFLICT")
+        secure_parent(observer_socket_path)
+        if observer_socket_path.exists() or observer_socket_path.is_symlink():
+            raise BrokerFailure("OBSERVER_SOCKET_ALREADY_EXISTS")
+    elif observer_socket_path is not None:
+        raise BrokerFailure("TEST_OBSERVER_SOCKET_CONFLICT")
+
     store = StateStore(state_path)
-    broker = Broker(store)
+    broker = Broker(store, observer)
     broker_lock = threading.Lock()
     slots = threading.BoundedSemaphore(MAX_IN_FLIGHT_CONNECTIONS)
     fatal_lock = threading.Lock()
@@ -622,16 +895,28 @@ def serve(socket_path: Path, state_path: Path) -> None:
     server.bind(str(socket_path))
     os.chmod(socket_path, 0o600)
     owned = (socket_path.stat().st_dev, socket_path.stat().st_ino)
+    observer_server: socket.socket | None = None
+    observer_owned: tuple[int, int] | None = None
+    if runtime_observer and observer_socket_path is not None:
+        observer_server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        observer_server.bind(str(observer_socket_path))
+        os.chmod(observer_socket_path, 0o600)
+        observer_owned = (observer_socket_path.stat().st_dev, observer_socket_path.stat().st_ino)
     stopping = False
 
     def stop(_signum: int, _frame: object) -> None:
         nonlocal stopping
         stopping = True
         server.close()
+        if observer_server is not None:
+            observer_server.close()
 
-    def process_connection(connection: socket.socket) -> None:
+    def process_connection(connection: socket.socket, is_observer: bool) -> None:
         try:
-            handle_connection(connection, broker, broker_lock)
+            if is_observer:
+                handle_runtime_observation_connection(connection, broker, broker_lock)
+            else:
+                handle_connection(connection, broker, broker_lock)
         except Exception as exc:
             with fatal_lock:
                 if not fatal_errors:
@@ -646,51 +931,73 @@ def serve(socket_path: Path, state_path: Path) -> None:
     signal.signal(signal.SIGTERM, stop)
     signal.signal(signal.SIGINT, stop)
     server.listen(MAX_IN_FLIGHT_CONNECTIONS)
-    server.settimeout(0.1)
+    server.setblocking(False)
+    if observer_server is not None:
+        observer_server.listen(MAX_IN_FLIGHT_CONNECTIONS)
+        observer_server.setblocking(False)
     print("READY", flush=True)
     try:
         while not stopping:
             failure = fatal_error()
             if failure is not None:
                 raise failure
-            if not slots.acquire(timeout=0.1):
-                continue
+            listeners = [server, *([observer_server] if observer_server is not None else [])]
             try:
-                connection, _ = server.accept()
-            except socket.timeout:
-                slots.release()
-                continue
-            except OSError:
-                slots.release()
-                failure = fatal_error()
-                if failure is not None:
-                    raise failure
+                ready, _, _ = select.select(listeners, [], [], 0.1)
+            except (OSError, ValueError):
                 if stopping:
                     break
                 raise
-            try:
-                executor.submit(process_connection, connection)
-            except Exception:
-                slots.release()
-                connection.close()
-                raise
+            for listener in ready:
+                if not slots.acquire(blocking=False):
+                    continue
+                try:
+                    connection, _ = listener.accept()
+                except BlockingIOError:
+                    slots.release()
+                    continue
+                except OSError:
+                    slots.release()
+                    if stopping:
+                        break
+                    raise
+                try:
+                    executor.submit(process_connection, connection, listener is observer_server)
+                except Exception:
+                    slots.release()
+                    connection.close()
+                    raise
     finally:
         server.close()
+        if observer_server is not None:
+            observer_server.close()
         executor.shutdown(wait=True)
-        try:
-            current = socket_path.stat()
-            if (current.st_dev, current.st_ino) == owned:
-                socket_path.unlink()
-        except FileNotFoundError:
-            pass
+        for path, inode in ((socket_path, owned), (observer_socket_path, observer_owned)):
+            if path is None or inode is None:
+                continue
+            try:
+                current = path.stat()
+                if (current.st_dev, current.st_ino) == inode:
+                    path.unlink()
+            except FileNotFoundError:
+                pass
 
 
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("--socket", required=True, type=Path)
     parser.add_argument("--state", required=True, type=Path)
+    parser.add_argument("--observer-socket", type=Path)
+    parser.add_argument("--test-observer-file", type=Path)
     arguments = parser.parse_args()
-    serve(arguments.socket, arguments.state)
+    if arguments.observer_socket is not None and arguments.test_observer_file is not None:
+        raise BrokerFailure("TEST_OBSERVER_SOCKET_CONFLICT")
+    observer = (
+        FileTestReceiptObserver(arguments.test_observer_file)
+        if arguments.test_observer_file is not None
+        else None
+    )
+    serve(arguments.socket, arguments.state, observer, arguments.observer_socket)
 
 
 if __name__ == "__main__":

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

@@ -12,6 +12,8 @@ from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import Final
 
+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"
@@ -44,6 +46,7 @@ def main(
     environ: Mapping[str, str] | None = None,
     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,
 ) -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
@@ -83,6 +86,8 @@ def main(
             or any(character not in "0123456789abcdef" for character in session_id)
         ):
             raise ValueError("registration refused")
+        generation_file = socket_path.parent / f"generation-{session_id}.state"
+        initialize_generation(generation_file, generation)
     except (KeyError, ValueError, OSError, json.JSONDecodeError):
         print("Mosaic lease broker registration failed; runtime launch denied.", file=sys.stderr)
         return 1
@@ -90,7 +95,14 @@ def main(
     environment = dict(source_environment)
     environment["MOSAIC_LEASE_SESSION_ID"] = session_id
     environment["MOSAIC_RUNTIME_GENERATION"] = str(generation)
+    environment["MOSAIC_LEASE_GENERATION_FILE"] = str(generation_file)
     environment["MOSAIC_LEASE_RUNTIME"] = arguments.runtime
+    # Matches daemon.py's production default; deployments using a distinct
+    # observer socket may set this authenticated transport path explicitly.
+    environment.setdefault(
+        "MOSAIC_RECEIPT_OBSERVER_SOCKET",
+        str(socket_path.with_name("receipt-observer.sock")),
+    )
     try:
         execute(command[0], command, environment)
     except OSError:

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

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""Private monotonic runtime-generation state shared by runtime hook processes."""
+
+from __future__ import annotations
+
+import fcntl
+import os
+import stat
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Final
+
+MAX_GENERATION: Final = (1 << 63) - 1
+MAX_GENERATION_BYTES: Final = 32
+
+
+def parse_generation(value: object) -> int:
+    if not isinstance(value, str) or not value.isascii() or not value.isdigit():
+        raise ValueError("invalid runtime generation")
+    generation = int(value)
+    if generation < 0 or generation > MAX_GENERATION:
+        raise ValueError("invalid runtime generation")
+    return generation
+
+
+def _validate_descriptor(descriptor: int) -> None:
+    metadata = os.fstat(descriptor)
+    if not stat.S_ISREG(metadata.st_mode):
+        raise ValueError("runtime generation state is not a regular file")
+    if metadata.st_uid != os.geteuid():
+        raise ValueError("runtime generation state has the wrong owner")
+    if stat.S_IMODE(metadata.st_mode) & 0o077:
+        raise ValueError("runtime generation state permissions are not private")
+    if metadata.st_size > MAX_GENERATION_BYTES:
+        raise ValueError("runtime generation state is oversized")
+
+
+def _read_descriptor(descriptor: int) -> int:
+    os.lseek(descriptor, 0, os.SEEK_SET)
+    raw = os.read(descriptor, MAX_GENERATION_BYTES + 1)
+    if len(raw) > MAX_GENERATION_BYTES:
+        raise ValueError("runtime generation state is oversized")
+    try:
+        return parse_generation(raw.decode("ascii").strip())
+    except UnicodeDecodeError as exc:
+        raise ValueError("invalid runtime generation state") from exc
+
+
+def _write_descriptor(descriptor: int, generation: int) -> None:
+    payload = f"{generation}\n".encode("ascii")
+    os.lseek(descriptor, 0, os.SEEK_SET)
+    os.ftruncate(descriptor, 0)
+    remaining = memoryview(payload)
+    while remaining:
+        written = os.write(descriptor, remaining)
+        if written <= 0:
+            raise OSError("runtime generation write made no progress")
+        remaining = remaining[written:]
+    os.fsync(descriptor)
+
+
+def initialize_runtime_generation(path: Path, generation: int) -> None:
+    if generation < 0 or generation > MAX_GENERATION:
+        raise ValueError("invalid runtime generation")
+    descriptor = os.open(
+        path,
+        os.O_RDWR | os.O_CREAT | os.O_TRUNC | os.O_CLOEXEC | os.O_NOFOLLOW,
+        0o600,
+    )
+    try:
+        os.fchmod(descriptor, 0o600)
+        _validate_descriptor(descriptor)
+        _write_descriptor(descriptor, generation)
+    finally:
+        os.close(descriptor)
+
+
+def read_runtime_generation(environ: Mapping[str, str]) -> int:
+    state_path = environ.get("MOSAIC_LEASE_GENERATION_FILE")
+    if not state_path:
+        return parse_generation(environ["MOSAIC_RUNTIME_GENERATION"])
+    descriptor = os.open(state_path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
+    try:
+        _validate_descriptor(descriptor)
+        return _read_descriptor(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def bump_runtime_generation(environ: Mapping[str, str]) -> int:
+    state_path = environ.get("MOSAIC_LEASE_GENERATION_FILE")
+    if not state_path:
+        raise ValueError("runtime generation file is required for same-PID rollover")
+    descriptor = os.open(state_path, os.O_RDWR | os.O_CLOEXEC | os.O_NOFOLLOW)
+    try:
+        fcntl.flock(descriptor, fcntl.LOCK_EX)
+        _validate_descriptor(descriptor)
+        current = _read_descriptor(descriptor)
+        if current >= MAX_GENERATION:
+            raise ValueError("runtime generation exhausted")
+        generation = current + 1
+        _write_descriptor(descriptor, generation)
+        return generation
+    finally:
+        os.close(descriptor)

packages/mosaic/framework/tools/lease-broker/mutator-gate.py

@@ -8,12 +8,23 @@ import json
 import os
 import socket
 import sys
+import re
 from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import BinaryIO, 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 lease_generation import read_runtime_generation
+
 MAX_FRAME: Final = 64 * 1024
 BROKER_TIMEOUT_SECONDS: Final = 1.5
+RECOVERY_TOOL: Final = "mosaic_context_recover"
+_LITERAL_ABSOLUTE_PATH: Final = re.compile(r"/[A-Za-z0-9._/-]+\Z")
+_SHELL_ACTIVE: Final = frozenset("$`~*?[]{}<>;|&" + '"' + "'" + "\\" + "\n\r\t")
 
 
 def deny(code: str) -> int:
@@ -21,7 +32,7 @@ def deny(code: str) -> int:
     return 2
 
 
-def read_tool_name(stream: BinaryIO | None = None) -> str:
+def read_tool_request(stream: BinaryIO | None = None) -> dict[str, object]:
     source = sys.stdin.buffer if stream is None else stream
     raw = source.read(MAX_FRAME + 1)
     if len(raw) > MAX_FRAME:
@@ -32,7 +43,61 @@ def read_tool_name(stream: BinaryIO | None = None) -> str:
     tool_name = value.get("tool_name")
     if not isinstance(tool_name, str) or not tool_name or len(tool_name) > 256:
         raise ValueError("INVALID_GATE_INPUT")
-    return tool_name
+    return value
+
+
+def read_tool_name(stream: BinaryIO | None = None) -> str:
+    """Backward-compatible strict extraction for callers that need only the name."""
+
+    return str(read_tool_request(stream)["tool_name"])
+
+
+def recovery_invocation_name(request: dict[str, object], recovery_command: Path | None) -> str:
+    """Map only a byte-literal Claude recovery argv to ``RECOVERY_TOOL``.
+
+    Claude's Bash tool evaluates its raw command with a real shell. Therefore
+    the gate never attempts a second shell parser: any quote, expansion,
+    redirection, operator, glob, newline, or non-space whitespace is refused
+    before tokenizing. The remaining plain-space split is an exact argv proof,
+    not a best-effort interpretation of shell syntax.
+    """
+
+    tool_name = request["tool_name"]
+    if tool_name != "Bash" or recovery_command is None:
+        return str(tool_name)
+    tool_input = request.get("tool_input")
+    if not isinstance(tool_input, dict) or set(tool_input) != {"command"}:
+        return str(tool_name)
+    command = tool_input.get("command")
+    if (
+        not isinstance(command, str)
+        or not command
+        or any(character in _SHELL_ACTIVE for character in command)
+        or command.startswith(" ")
+        or command.endswith(" ")
+        or "  " in command
+    ):
+        return str(tool_name)
+    argv = command.split(" ")
+    if " ".join(argv) != command or len(argv) < 3 or argv[0] != "python3":
+        return str(tool_name)
+    if argv[1] != str(recovery_command):
+        return str(tool_name)
+    phase = argv[2]
+    if phase == "complete" and len(argv) == 3:
+        return RECOVERY_TOOL
+    if phase != "begin" or len(argv) != 9:
+        return str(tool_name)
+    if argv[3::2] != ["--construction", "--compaction-epoch", "--request-epoch"]:
+        return str(tool_name)
+    construction, compaction_epoch, request_epoch = argv[4::2]
+    if (
+        _LITERAL_ABSOLUTE_PATH.fullmatch(construction) is None
+        or not compaction_epoch.isdecimal()
+        or not request_epoch.isdecimal()
+    ):
+        return str(tool_name)
+    return RECOVERY_TOOL
 
 
 def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
@@ -64,19 +129,20 @@ def main(
     environ: Mapping[str, str] | None = None,
     stream: BinaryIO | None = None,
     request: Callable[[Path, dict[str, object]], dict[str, object]] = broker_request,
+    resolve_generation: Callable[[Mapping[str, str]], int] = read_runtime_generation,
 ) -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
+    parser.add_argument("--recovery-command", type=Path)
     arguments = parser.parse_args(argv)
     source_environment = os.environ if environ is None else environ
 
     try:
-        tool_name = read_tool_name(stream)
+        request_input = read_tool_request(stream)
+        tool_name = recovery_invocation_name(request_input, arguments.recovery_command)
         socket_value = source_environment["MOSAIC_LEASE_BROKER_SOCKET"]
         session_id = source_environment["MOSAIC_LEASE_SESSION_ID"]
-        generation = int(source_environment["MOSAIC_RUNTIME_GENERATION"])
-        if generation < 0:
-            raise ValueError("INVALID_GENERATION")
+        generation = resolve_generation(source_environment)
         reply = request(
             Path(socket_value),
             {

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

@@ -0,0 +1,215 @@
+#!/usr/bin/env python3
+"""Fail-closed construction of verbatim-hashed normative fragments.
+
+This is the single construction path for the Claude and Pi adapters.  The
+payload contains only versioned source metadata and exact validated source
+bytes.  ``h_payload`` is derived afterwards and is deliberately not representable
+as a payload input, preventing cryptographic self-reference.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import struct
+from collections.abc import Sequence
+from typing import Final
+
+MAX_FRAGMENT_BYTES: Final = 64 * 1024
+HASH_DOMAIN_SEPARATOR: Final = b"MOSAIC/H_PAYLOAD/v1\x00"
+SOURCE_DOMAIN_SEPARATOR: Final = b"MOSAIC/H_SOURCE/v1\x00"
+_PAYLOAD_VERSION_LABEL: Final = b"MOSAIC/B_PAYLOAD/v1"
+
+
+class NormativeFragment:
+    """A source identity, its expected digest, and its exact resolved bytes."""
+
+    def __init__(self, source_id: str, content: bytes | None, expected_sha256: str) -> None:
+        self.source_id = source_id
+        self.content = content
+        self.expected_sha256 = expected_sha256
+
+
+class ConstructionResult:
+    """A source-admission decision and, only when admitted, derived payload values.
+
+    ``promotion`` means the construction has produced the only values a later
+    receipt protocol may use to attempt promotion. It never performs broker
+    promotion itself. A REFUSED result has no payload/hash values, so it cannot
+    advance to that later protocol.
+    """
+
+    def __init__(
+        self,
+        injection_decision: str,
+        promotion: bool,
+        source_reason: str | None,
+        b_payload: bytes | None,
+        h_payload: str | None,
+        h_source: str | None,
+    ) -> None:
+        self.injectionDecision = injection_decision
+        self.promotion = promotion
+        self.source_reason = source_reason
+        self.b_payload = b_payload
+        self.h_payload = h_payload
+        self.h_source = h_source
+
+
+def length_frame(parts: Sequence[bytes]) -> bytes:
+    """Encode a finite ordered byte sequence with unambiguous 64-bit framing."""
+
+    framed = bytearray(struct.pack(">Q", len(parts)))
+    for part in parts:
+        if not isinstance(part, bytes):
+            raise TypeError("length framing requires bytes")
+        framed.extend(struct.pack(">Q", len(part)))
+        framed.extend(part)
+    return bytes(framed)
+
+
+def _sha256_hex(payload: bytes) -> str:
+    return hashlib.sha256(payload).hexdigest()
+
+
+def _valid_digest(value: object) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) == 64
+        and all(character in "0123456789abcdef" for character in value)
+    )
+
+
+def _source_reason(fragment: NormativeFragment) -> str | None:
+    if not isinstance(fragment.source_id, str) or not fragment.source_id:
+        return "missing"
+    if fragment.content is None:
+        return "missing"
+    if not isinstance(fragment.content, bytes):
+        return "missing"
+    if len(fragment.content) > MAX_FRAGMENT_BYTES:
+        return "oversize"
+    if not _valid_digest(fragment.expected_sha256):
+        return "hash-mismatch"
+    actual = _sha256_hex(fragment.content)
+    if not hmac.compare_digest(actual, fragment.expected_sha256):
+        return "hash-mismatch"
+    return None
+
+
+def _refused(reason: str) -> ConstructionResult:
+    return ConstructionResult(
+        injection_decision="REFUSED",
+        promotion=False,
+        source_reason=reason,
+        b_payload=None,
+        h_payload=None,
+        h_source=None,
+    )
+
+
+def build_payload(
+    *,
+    manifest_version: int,
+    generator_version: str,
+    fragments: Sequence[NormativeFragment],
+) -> ConstructionResult:
+    """Construct B_payload then H_payload after fail-closed source validation.
+
+    The sequence order is caller-supplied resolved-source order and is encoded
+    directly. Changing source order, source identity, metadata, or any exact
+    fragment byte therefore changes the framed B_payload and its derived hash.
+    """
+
+    if type(manifest_version) is not int or manifest_version < 0:
+        return _refused("missing")
+    if not isinstance(generator_version, str) or not generator_version:
+        return _refused("missing")
+
+    validated = list(fragments)
+    if not validated:
+        return _refused("missing")
+    for fragment in validated:
+        if not isinstance(fragment, NormativeFragment):
+            return _refused("missing")
+        reason = _source_reason(fragment)
+        if reason is not None:
+            return _refused(reason)
+
+    source_identities = [
+        length_frame([
+            fragment.source_id.encode("utf-8"),
+            fragment.expected_sha256.encode("ascii"),
+        ])
+        for fragment in validated
+    ]
+    h_source = _sha256_hex(SOURCE_DOMAIN_SEPARATOR + length_frame(source_identities))
+    payload_parts = [
+        _PAYLOAD_VERSION_LABEL,
+        str(manifest_version).encode("ascii"),
+        generator_version.encode("utf-8"),
+        *source_identities,
+        *(fragment.content for fragment in validated),
+    ]
+    # h_payload is intentionally not an argument or field in payload_parts.
+    b_payload = length_frame(payload_parts)
+    h_payload = _sha256_hex(HASH_DOMAIN_SEPARATOR + length_frame([b_payload]))
+    return ConstructionResult(
+        injection_decision="ACCEPTED",
+        promotion=True,
+        source_reason=None,
+        b_payload=b_payload,
+        h_payload=h_payload,
+        h_source=h_source,
+    )
+
+
+def build_payload_from_wire(value: object) -> ConstructionResult:
+    """Decode a bounded broker request then invoke the sole payload builder.
+
+    Wire inputs contain only source bytes and their claimed source digests. The
+    authoritative ``build_payload`` implementation remains the only code that
+    admits those bytes and derives ``h_source``/``h_payload``.
+    """
+
+    if not isinstance(value, dict) or set(value) != {
+        "manifest_version", "generator_version", "fragments"
+    }:
+        raise ValueError("invalid construction")
+    raw_fragments = value["fragments"]
+    if not isinstance(raw_fragments, list):
+        raise ValueError("invalid construction")
+    fragments: list[NormativeFragment] = []
+    for raw_fragment in raw_fragments:
+        if not isinstance(raw_fragment, dict) or set(raw_fragment) != {
+            "source_id", "content_base64", "expected_sha256"
+        }:
+            raise ValueError("invalid construction")
+        source_id = raw_fragment["source_id"]
+        encoded = raw_fragment["content_base64"]
+        expected_sha256 = raw_fragment["expected_sha256"]
+        if not isinstance(source_id, str) or not isinstance(encoded, str) or not isinstance(expected_sha256, str):
+            raise ValueError("invalid construction")
+        try:
+            content = base64.b64decode(encoded.encode("ascii"), validate=True)
+        except (UnicodeEncodeError, ValueError) as exc:
+            raise ValueError("invalid construction") from exc
+        fragments.append(NormativeFragment(source_id, content, expected_sha256))
+    return build_payload(
+        manifest_version=value["manifest_version"],
+        generator_version=value["generator_version"],
+        fragments=fragments,
+    )
+
+
+def build_for_claude(**kwargs: object) -> ConstructionResult:
+    """Claude adapter entrypoint; delegates to the sole shared constructor."""
+
+    return build_payload(**kwargs)  # type: ignore[arg-type]
+
+
+def build_for_pi(**kwargs: object) -> ConstructionResult:
+    """Pi adapter entrypoint; delegates to the sole shared constructor."""
+
+    return build_payload(**kwargs)  # type: ignore[arg-type]

packages/mosaic/framework/tools/lease-broker/receipt-observer-client.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""Authenticated adapter-to-daemon transport for finalized assistant receipts.
+
+This is not a broker request client. It writes only to the daemon-owned observer
+socket, which authenticates SO_PEERCRED/ancestry before retaining a message for
+the broker's ReceiptObserver seam.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import socket
+import stat
+import sys
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Final
+
+
+MAX_FRAME: Final = 64 * 1024
+BROKER_TIMEOUT_SECONDS: Final = 1.5
+MAX_TRANSCRIPT_BYTES: Final = 4 * 1024 * 1024
+
+
+def read_json(stream: object) -> dict[str, object]:
+    raw = getattr(stream, "buffer", stream).read(MAX_FRAME + 1)
+    if not isinstance(raw, bytes) or len(raw) > MAX_FRAME:
+        raise ValueError("invalid observer input")
+    value = json.loads(raw)
+    if not isinstance(value, dict):
+        raise ValueError("invalid observer input")
+    return value
+
+
+def assistant_text(entry: object) -> str | None:
+    if not isinstance(entry, dict):
+        return None
+    message = entry.get("message", entry)
+    if not isinstance(message, dict) or message.get("role") != "assistant":
+        return None
+    content = message.get("content")
+    if isinstance(content, str):
+        return content
+    if not isinstance(content, list):
+        return None
+    parts: list[str] = []
+    for item in content:
+        if not isinstance(item, dict) or item.get("type") != "text" or not isinstance(item.get("text"), str):
+            return None
+        parts.append(item["text"])
+    return "".join(parts)
+
+
+def claude_latest_entry(value: dict[str, object]) -> str:
+    transcript_path = value.get("transcript_path")
+    if not isinstance(transcript_path, str) or not transcript_path:
+        raise ValueError("invalid Claude observer input")
+    path = Path(transcript_path)
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(path, flags)
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_TRANSCRIPT_BYTES:
+            raise ValueError("unsafe Claude transcript")
+        raw = os.read(descriptor, MAX_TRANSCRIPT_BYTES + 1)
+    finally:
+        os.close(descriptor)
+    if len(raw) > MAX_TRANSCRIPT_BYTES:
+        raise ValueError("oversized Claude transcript")
+    for line in reversed(raw.decode("utf-8").splitlines()):
+        try:
+            text = assistant_text(json.loads(line))
+        except json.JSONDecodeError as exc:
+            raise ValueError("invalid Claude transcript") from exc
+        if text is not None:
+            return text
+    raise ValueError("Claude transcript has no assistant entry")
+
+
+def pi_message_end(value: dict[str, object]) -> str:
+    if set(value) != {"latest_assistant_message"} or not isinstance(value["latest_assistant_message"], str):
+        raise ValueError("invalid Pi observer input")
+    return value["latest_assistant_message"]
+
+
+def observer_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
+    payload = (json.dumps(request, separators=(",", ":")) + "\n").encode()
+    if len(payload) > MAX_FRAME:
+        raise ValueError("observer request too large")
+    response = bytearray()
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(BROKER_TIMEOUT_SECONDS)
+        connection.connect(str(socket_path))
+        connection.sendall(payload)
+        connection.shutdown(socket.SHUT_WR)
+        while len(response) <= MAX_FRAME:
+            chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(response)))
+            if not chunk:
+                break
+            response.extend(chunk)
+    if len(response) > MAX_FRAME or not response.endswith(b"\n"):
+        raise ValueError("invalid observer reply")
+    value = json.loads(response)
+    if not isinstance(value, dict):
+        raise ValueError("invalid observer reply")
+    return value
+
+
+def main(argv: Sequence[str] | None = None, *, environ: Mapping[str, str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
+    parser.add_argument("--latest-entry", action="store_true")
+    arguments = parser.parse_args(argv)
+    source_environment = os.environ if environ is None else environ
+    try:
+        source = read_json(sys.stdin)
+        if arguments.runtime == "claude":
+            if not arguments.latest_entry:
+                raise ValueError("Claude observer requires --latest-entry")
+            message = claude_latest_entry(source)
+        else:
+            if arguments.latest_entry:
+                raise ValueError("Pi observer is message_end only")
+            message = pi_message_end(source)
+        if len(message.encode("utf-8")) > MAX_FRAME:
+            raise ValueError("assistant message too large")
+        reply = observer_request(Path(source_environment["MOSAIC_RECEIPT_OBSERVER_SOCKET"]), {
+            "action": "record_runtime_observation",
+            "session_id": source_environment["MOSAIC_LEASE_SESSION_ID"],
+            "runtime_generation": int(source_environment["MOSAIC_RUNTIME_GENERATION"]),
+            "runtime": arguments.runtime,
+            "latest_assistant_message": message,
+        })
+    except (KeyError, OSError, ValueError, json.JSONDecodeError) as error:
+        print(f"Mosaic receipt observer refused: {error}", file=sys.stderr)
+        return 2
+    return 0 if reply == {"ok": True} else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+"""Broker-side construction and verification for one-time receipt challenges."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import struct
+from typing import Final
+
+
+LATEST_ASSISTANT_DOMAIN_SEPARATOR: Final = b"MOSAIC/H_LATEST_ASSISTANT/v1\x00"
+
+
+def _length_frame(value: bytes) -> bytes:
+    return struct.pack(">Q", len(value)) + value
+
+
+def receipt_for(challenge: str, binding: dict[str, object]) -> str:
+    """Return the sole receipt text a model may copy for this broker cycle."""
+
+    return (
+        "MOSAIC-RECEIPT{"
+        f"challenge={challenge}; "
+        f"H_payload={binding['h_payload']}; "
+        f"gen={binding['runtime_generation']}; "
+        f"cep={binding['compaction_epoch']}"
+        "}"
+    )
+
+
+def is_verbatim_receipt(message: str, challenge: str, binding: dict[str, object]) -> bool:
+    """Require the exact one current-cycle receipt, not a transcript substring."""
+
+    expected = receipt_for(challenge, binding)
+    return hmac.compare_digest(message, expected)
+
+
+def latest_assistant_digest(message: str) -> str:
+    """Record the broker-computed digest of the exact observed assistant entry."""
+
+    encoded = message.encode("utf-8")
+    return hashlib.sha256(LATEST_ASSISTANT_DOMAIN_SEPARATOR + _length_frame(encoded)).hexdigest()

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

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""Trusted latest-assistant-message observer boundary for receipt promotion.
+
+Production adapters deliver finalized assistant content over the daemon-owned
+observer socket after the daemon authenticates their peer against the broker's
+kernel-anchored session identity. The broker request protocol never accepts
+assistant-message content. Claude supplies its latest assistant entry; Pi
+supplies finalized assistant content at ``message_end``.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+from pathlib import Path
+from typing import Protocol
+
+
+class ReceiptObserver(Protocol):
+    def observe_latest_assistant_message(
+        self,
+        session_id: str,
+        runtime: str,
+        runtime_generation: int,
+        binding: dict[str, object],
+    ) -> str | None: ...
+
+
+class UnavailableReceiptObserver:
+    """Fail-closed only for direct unit construction without daemon transport."""
+
+    def observe_latest_assistant_message(
+        self,
+        _session_id: str,
+        _runtime: str,
+        _runtime_generation: int,
+        _binding: dict[str, object],
+    ) -> str | None:
+        return None
+
+
+class RuntimeReceiptObserver:
+    """Daemon-owned production observer populated only by authenticated adapters."""
+
+    def __init__(self) -> None:
+        self._messages: dict[tuple[str, str, int], str] = {}
+
+    def record_latest_assistant_message(
+        self,
+        session_id: str,
+        runtime: str,
+        runtime_generation: int,
+        message: str,
+    ) -> None:
+        self._messages[(session_id, runtime, runtime_generation)] = message
+
+    def observe_latest_assistant_message(
+        self,
+        session_id: str,
+        runtime: str,
+        runtime_generation: int,
+        _binding: dict[str, object],
+    ) -> str | None:
+        return self._messages.get((session_id, runtime, runtime_generation))
+
+
+class TestReceiptObserver:
+    """Deterministic controlled observer used only by byte-build tests."""
+
+    def __init__(self) -> None:
+        self._messages: dict[tuple[str, int], str] = {}
+
+    def record_latest_assistant_message(
+        self, session_id: str, runtime_generation: int, message: str
+    ) -> None:
+        self._messages[(session_id, runtime_generation)] = message
+
+    def observe_latest_assistant_message(
+        self,
+        session_id: str,
+        _runtime: str,
+        runtime_generation: int,
+        _binding: dict[str, object],
+    ) -> str | None:
+        return self._messages.get((session_id, runtime_generation))
+
+
+class FileTestReceiptObserver:
+    """Private fixture-file observer for isolated out-of-process test drivers only."""
+
+    def __init__(self, path: Path) -> None:
+        self.path = path
+
+    def observe_latest_assistant_message(
+        self,
+        session_id: str,
+        _runtime: str,
+        runtime_generation: int,
+        _binding: dict[str, object],
+    ) -> str | None:
+        flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+        try:
+            descriptor = os.open(self.path, flags)
+        except OSError:
+            return None
+        try:
+            metadata = os.fstat(descriptor)
+            if (
+                not stat.S_ISREG(metadata.st_mode)
+                or stat.S_IMODE(metadata.st_mode) != 0o600
+                or metadata.st_uid != os.geteuid()
+                or metadata.st_size > 64 * 1024
+            ):
+                return None
+            raw = os.read(descriptor, 64 * 1024 + 1)
+        finally:
+            os.close(descriptor)
+        if len(raw) > 64 * 1024:
+            return None
+        try:
+            value = json.loads(raw)
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            return None
+        if (
+            not isinstance(value, dict)
+            or set(value) != {"session_id", "runtime_generation", "latest_assistant_message"}
+            or value["session_id"] != session_id
+            or value["runtime_generation"] != runtime_generation
+            or not isinstance(value["latest_assistant_message"], str)
+        ):
+            return None
+        return value["latest_assistant_message"]

packages/mosaic/framework/tools/lease-broker/recover-context.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""Constrained recovery command: the sole ungated Mosaic mutator.
+
+This is deliberately a thin client of the broker's recovery entrypoint.  It
+never accepts receipt text or a caller-provided challenge: the broker mints the
+fresh challenge, delivers its exact receipt envelope, and later asks the
+trusted ReceiptObserver seam to observe that same pending cycle.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import socket
+import sys
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Final
+
+# The out-of-process recovery command is intentionally runnable with `python
+# -I`; locate its shipped construction module without caller-controlled paths.
+_MODULE_DIRECTORY = str(Path(__file__).resolve().parent)
+if _MODULE_DIRECTORY not in sys.path:
+    sys.path.insert(0, _MODULE_DIRECTORY)
+from normative_fragments import build_payload_from_wire
+
+MAX_FRAME: Final = 64 * 1024
+BROKER_TIMEOUT_SECONDS: Final = 1.5
+
+
+def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
+    payload = (json.dumps(request, separators=(",", ":")) + "\n").encode()
+    if len(payload) > MAX_FRAME:
+        raise ValueError("recovery request too large")
+    response = bytearray()
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(BROKER_TIMEOUT_SECONDS)
+        connection.connect(str(socket_path))
+        connection.sendall(payload)
+        connection.shutdown(socket.SHUT_WR)
+        while len(response) <= MAX_FRAME:
+            chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(response)))
+            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 load_construction(path: Path) -> dict[str, object]:
+    raw = path.read_bytes()
+    if len(raw) > MAX_FRAME:
+        raise ValueError("construction exceeds broker frame limit")
+    value = json.loads(raw)
+    if not isinstance(value, dict):
+        raise ValueError("construction must be an object")
+    return value
+
+
+def identity(environ: Mapping[str, str]) -> tuple[Path, str, int, str]:
+    socket_path = Path(environ["MOSAIC_LEASE_BROKER_SOCKET"])
+    session_id = environ["MOSAIC_LEASE_SESSION_ID"]
+    generation = int(environ["MOSAIC_RUNTIME_GENERATION"])
+    runtime = environ["MOSAIC_LEASE_RUNTIME"]
+    if generation < 0 or runtime not in {"claude", "pi"}:
+        raise ValueError("invalid runtime identity")
+    return socket_path, session_id, generation, runtime
+
+
+def begin(
+    construction_path: Path,
+    compaction_epoch: int,
+    request_epoch: int,
+    environ: Mapping[str, str],
+) -> dict[str, object]:
+    if compaction_epoch < 0 or request_epoch < 0:
+        raise ValueError("epochs must be non-negative")
+    construction = load_construction(construction_path)
+    # Invoke the shared WI-5 construction before asking the broker to repeat
+    # its authoritative admission/build. No digest or receipt enters via CLI.
+    built = build_payload_from_wire(construction)
+    if (
+        built.injectionDecision != "ACCEPTED"
+        or not built.promotion
+        or not isinstance(built.h_source, str)
+        or not isinstance(built.h_payload, str)
+    ):
+        raise ValueError("payload construction refused")
+    socket_path, session_id, generation, runtime = identity(environ)
+    return broker_request(socket_path, {
+        "action": "begin_recovery",
+        "session_id": session_id,
+        "runtime_generation": generation,
+        "runtime": runtime,
+        "binding": {
+            "compaction_epoch": compaction_epoch,
+            "request_epoch": request_epoch,
+            "h_source": built.h_source,
+            "h_payload": built.h_payload,
+            "schema_version": 1,
+        },
+        "construction": construction,
+    })
+
+
+def complete(environ: Mapping[str, str]) -> dict[str, object]:
+    socket_path, session_id, generation, _runtime = identity(environ)
+    # No receipt or challenge argument exists: recovery completion can only use
+    # the broker's current recovery cycle and its trusted observer seam.
+    return broker_request(socket_path, {
+        "action": "complete_recovery",
+        "session_id": session_id,
+        "runtime_generation": generation,
+    })
+
+
+def main(argv: Sequence[str] | None = None, *, environ: Mapping[str, str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    subcommands = parser.add_subparsers(dest="phase", required=True)
+    begin_parser = subcommands.add_parser("begin", help="mint and deliver a fresh recovery receipt")
+    begin_parser.add_argument("--construction", required=True, type=Path)
+    begin_parser.add_argument("--compaction-epoch", required=True, type=int)
+    begin_parser.add_argument("--request-epoch", required=True, type=int)
+    subcommands.add_parser("complete", help="observe and promote only the current recovery receipt")
+    arguments = parser.parse_args(argv)
+    source_environment = os.environ if environ is None else environ
+    try:
+        reply = (
+            begin(
+                arguments.construction,
+                arguments.compaction_epoch,
+                arguments.request_epoch,
+                source_environment,
+            )
+            if arguments.phase == "begin"
+            else complete(source_environment)
+        )
+    except (KeyError, OSError, ValueError, json.JSONDecodeError) as error:
+        print(f"Mosaic constrained recovery refused: {error}", file=sys.stderr)
+        return 2
+    print(json.dumps(reply, separators=(",", ":")))
+    return 0 if reply.get("ok") is True else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+"""Revoke a runtime lease from a compaction or lifecycle observer."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import socket
+import sys
+from collections.abc import Callable, Mapping, Sequence
+from pathlib import Path
+from typing import Final
+
+from lease_generation import bump_runtime_generation, read_runtime_generation
+
+MAX_FRAME: Final = 64 * 1024
+BROKER_TIMEOUT_SECONDS: Final = 1.5
+
+
+def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
+    payload = (json.dumps(request, separators=(",", ":")) + "\n").encode()
+    if len(payload) > MAX_FRAME:
+        raise ValueError("invalid revoke request")
+    response = bytearray()
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(BROKER_TIMEOUT_SECONDS)
+        connection.connect(str(socket_path))
+        connection.sendall(payload)
+        connection.shutdown(socket.SHUT_WR)
+        while len(response) <= MAX_FRAME:
+            chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(response)))
+            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 main(
+    argv: Sequence[str] | None = None,
+    *,
+    environ: Mapping[str, str] | None = None,
+    request: Callable[[Path, dict[str, object]], dict[str, object]] = broker_request,
+) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
+    parser.add_argument("--reason", required=True)
+    parser.add_argument("--bump-generation", action="store_true")
+    arguments = parser.parse_args(argv)
+    source_environment = os.environ if environ is None else environ
+
+    try:
+        if not arguments.reason or len(arguments.reason) > 128:
+            raise ValueError("invalid revoke reason")
+        socket_path = Path(source_environment["MOSAIC_LEASE_BROKER_SOCKET"])
+        session_id = source_environment["MOSAIC_LEASE_SESSION_ID"]
+        if (
+            len(session_id) != 64
+            or any(character not in "0123456789abcdef" for character in session_id)
+        ):
+            raise ValueError("invalid broker session")
+        generation = (
+            bump_runtime_generation(source_environment)
+            if arguments.bump_generation
+            else read_runtime_generation(source_environment)
+        )
+        reply = request(
+            socket_path,
+            {
+                "action": "revoke_lease",
+                "session_id": session_id,
+                "runtime_generation": generation,
+                "reason": arguments.reason,
+                "runtime": arguments.runtime,
+            },
+        )
+        if reply.get("ok") is not True or reply.get("state") != "UNVERIFIED":
+            raise ValueError("revocation refused")
+    except (KeyError, ValueError, OSError, json.JSONDecodeError):
+        # A fired observer must remain fail-closed even if the broker transport
+        # is unavailable. Advancing the private local generation fences every
+        # later tool check; the broker revokes the old lease when it next sees
+        # that higher generation. Explicit rollover already advanced it above.
+        if not arguments.bump_generation:
+            try:
+                bump_runtime_generation(source_environment)
+            except (KeyError, ValueError, OSError):
+                pass
+        print("Mosaic lease revocation failed; lifecycle transition denied.", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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/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"
+    "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"
   },
   "dependencies": {
     "@mosaicstack/brain": "workspace:*",

packages/mosaic/src/commands/claudex.spec.ts

@@ -601,15 +601,23 @@ describe('ensureClaudexMutatorGateSettings', () => {
 
       const settings = JSON.parse(readFileSync(join(root, 'settings.json'), 'utf8')) as {
         preserved: boolean;
-        hooks: { PreToolUse: Array<{ hooks: Array<{ command: string }> }> };
+        hooks: Record<string, Array<{ matcher: string; hooks: Array<{ command: string }> }>>;
       };
-      const commands = settings.hooks.PreToolUse.flatMap((entry) =>
+      const commands = settings.hooks['PreToolUse']!.flatMap((entry) =>
         entry.hooks.map((hook) => hook.command),
       );
       expect(settings.preserved).toBe(true);
       expect(commands).toContain(
         'python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude',
       );
+      expect(
+        settings.hooks['PreCompact']?.some((entry) =>
+          entry.hooks.some((hook) => hook.command.includes('revoke-lease.py')),
+        ),
+      ).toBe(true);
+      expect(settings.hooks['SessionStart']?.map((entry) => entry.matcher)).toEqual(
+        expect.arrayContaining(['compact', 'resume|clear']),
+      );
       expect(lstatSync(join(root, 'settings.json')).mode & 0o777).toBe(0o600);
     } finally {
       rmSync(root, { recursive: true, force: true });

packages/mosaic/src/commands/claudex.ts

@@ -287,17 +287,19 @@ export function buildClaudexEnv(
 
 const CLAUDEX_MUTATOR_GATE_COMMAND =
   'python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude';
+const CLAUDEX_PRE_COMPACT_COMMAND =
+  'python3 ~/.config/mosaic/tools/lease-broker/revoke-lease.py --runtime claude --reason pre-compact';
+const CLAUDEX_POST_COMPACT_COMMAND =
+  'python3 ~/.config/mosaic/tools/lease-broker/revoke-lease.py --runtime claude --reason session-start-compact';
+const CLAUDEX_ROLLOVER_COMMAND =
+  'python3 ~/.config/mosaic/tools/lease-broker/revoke-lease.py --runtime claude --reason session-start-rollover --bump-generation';
 
-const CLAUDEX_MUTATOR_GATE_HOOK = {
-  matcher: '.*',
-  hooks: [
-    {
-      type: 'command',
-      command: CLAUDEX_MUTATOR_GATE_COMMAND,
-      timeout: 3,
-    },
-  ],
-};
+const CLAUDEX_MANDATORY_LEASE_HOOKS = [
+  { event: 'PreToolUse', matcher: '.*', command: CLAUDEX_MUTATOR_GATE_COMMAND },
+  { event: 'PreCompact', matcher: '.*', command: CLAUDEX_PRE_COMPACT_COMMAND },
+  { event: 'SessionStart', matcher: 'compact', command: CLAUDEX_POST_COMPACT_COMMAND },
+  { event: 'SessionStart', matcher: 'resume|clear', command: CLAUDEX_ROLLOVER_COMMAND },
+] as const;
 
 function isRecord(value: unknown): value is Record<string, unknown> {
   return typeof value === 'object' && value !== null && !Array.isArray(value);
@@ -331,26 +333,31 @@ export function ensureClaudexMutatorGateSettings(configDir: string): void {
     throw new Error('claudex: isolated settings hooks must be an object (fail closed).');
   }
   const hooks = hooksValue ?? {};
-  const preToolUseValue = hooks['PreToolUse'];
-  if (preToolUseValue !== undefined && !Array.isArray(preToolUseValue)) {
-    throw new Error('claudex: isolated PreToolUse hooks must be an array (fail closed).');
+  for (const mandatory of CLAUDEX_MANDATORY_LEASE_HOOKS) {
+    const eventValue = hooks[mandatory.event];
+    if (eventValue !== undefined && !Array.isArray(eventValue)) {
+      throw new Error(`claudex: isolated ${mandatory.event} hooks must be an array (fail closed).`);
+    }
+    const eventHooks = eventValue ?? [];
+    const hookPresent = eventHooks.some(
+      (entry) =>
+        isRecord(entry) &&
+        entry['matcher'] === mandatory.matcher &&
+        Array.isArray(entry['hooks']) &&
+        entry['hooks'].some(
+          (hook) =>
+            isRecord(hook) && hook['type'] === 'command' && hook['command'] === mandatory.command,
+        ),
+    );
+    if (!hookPresent) {
+      eventHooks.unshift({
+        matcher: mandatory.matcher,
+        hooks: [{ type: 'command', command: mandatory.command, timeout: 3 }],
+      });
+    }
+    hooks[mandatory.event] = eventHooks;
   }
-  const preToolUse = preToolUseValue ?? [];
-  const gatePresent = preToolUse.some(
-    (entry) =>
-      isRecord(entry) &&
-      entry['matcher'] === '.*' &&
-      Array.isArray(entry['hooks']) &&
-      entry['hooks'].some(
-        (hook) =>
-          isRecord(hook) &&
-          hook['type'] === 'command' &&
-          hook['command'] === CLAUDEX_MUTATOR_GATE_COMMAND,
-      ),
-  );
-  if (!gatePresent) preToolUse.unshift(CLAUDEX_MUTATOR_GATE_HOOK);
 
-  hooks['PreToolUse'] = preToolUse;
   settings['hooks'] = hooks;
   writeFileSync(settingsPath, `${JSON.stringify(settings, null, 2)}\n`, { mode: 0o600 });
   chmodSync(settingsPath, 0o600);

packages/mosaic/src/commands/skill.spec.ts

@@ -8,6 +8,7 @@ import {
   mkdirSync,
   mkdtempSync,
   readlinkSync,
+  readFileSync,
   rmSync,
   symlinkSync,
   writeFileSync,
@@ -26,6 +27,9 @@ import {
 const LEGACY_SYNC_SCRIPT = fileURLToPath(
   new URL('../../framework/tools/_scripts/mosaic-sync-skills', import.meta.url),
 );
+const CONTEXT_REFRESH_SOURCE_SKILL = fileURLToPath(
+  new URL('../../framework/skills/mosaic-context-refresh/SKILL.md', import.meta.url),
+);
 
 describe('Claude skill bridge', () => {
   let root: string;
@@ -360,6 +364,25 @@ describe('Claude skill bridge', () => {
   });
 
   describe('syncClaudeSkills', () => {
+    it('projects the durable context-refresh skill through the #824 bridge in a tmp root', () => {
+      expect(existsSync(CONTEXT_REFRESH_SOURCE_SKILL)).toBe(true);
+      const source = readFileSync(CONTEXT_REFRESH_SOURCE_SKILL, 'utf8');
+      expect(source).toContain('recover-context.py');
+      expect(source).toContain('middle drop is not represented as receipt-detectable');
+      const skillDir = createSkill('mosaic-context-refresh');
+      writeFileSync(join(skillDir, 'SKILL.md'), source);
+
+      const result = syncClaudeSkills(paths);
+
+      expect(result).toEqual({
+        registered: ['mosaic-context-refresh'],
+        repaired: [],
+        unchanged: [],
+        conflicts: [],
+      });
+      expectCorrectLink('mosaic-context-refresh');
+    });
+
     it('generically creates every missing canonical link and repairs managed broken links', () => {
       createSkill('added-after-setup');
       createSkill('another-new-skill');

packages/mosaic/src/framework/manifest-parity.spec.ts

@@ -78,6 +78,7 @@ const PROBE_PATHS = [
   'install.sh',
   'framework-manifest.txt',
   'guides/E2E-DELIVERY.md',
+  'skills/mosaic-context-refresh/SKILL.md',
   'tools/git/pr-create.sh',
   'tools/_lib/manifest.sh',
   'defaults/SOUL.md',
@@ -97,6 +98,7 @@ const PROBE_PATHS = [
   'memory/note.md',
   'sources/skills/x.md',
   'credentials/c.json',
+  'skills-local/custom/SKILL.md',
   'tools/_lib/credentials.json',
   'fleet/roster.yaml',
   'fleet/roster.json',

packages/mosaic/src/framework/manifest.spec.ts

@@ -304,6 +304,11 @@ describe('manifest completeness against shipped framework tree', () => {
     expect(misclassified).toEqual([]);
   });
 
+  it('shipped canonical skills are framework-owned while local skills are operator-owned', () => {
+    expect(resolveOwnership(manifest, 'skills/mosaic-context-refresh/SKILL.md')).toBe('framework');
+    expect(resolveOwnership(manifest, 'skills-local/custom/SKILL.md')).toBe('operator');
+  });
+
   it('the operator-owned surface from #791 resolves to operator', () => {
     const operatorPaths = [
       'agents/coder0.conf',
@@ -313,6 +318,7 @@ describe('manifest completeness against shipped framework tree', () => {
       'SOUL.local.md',
       'USER.local.md',
       'STANDARDS.local.md',
+      'skills-local/custom/SKILL.md',
       'tools/_lib/credentials.json',
       'fleet/roster.yaml',
       'fleet/roster.json',

packages/mosaic/src/lease-broker/broker-test-client.ts

@@ -1,4 +1,5 @@
 import { createHash } from 'node:crypto';
+import { chmod, writeFile } from 'node:fs/promises';
 import { createConnection, type Socket } from 'node:net';
 
 const DEFAULT_TIMEOUT_MS = 3_000;
@@ -185,3 +186,51 @@ export async function requestBrokerReply<T extends object>(
     options,
   );
 }
+
+export interface ReceiptChallengeCycle {
+  sessionId: string;
+  runtimeGeneration: number;
+  receiptChallenge: string;
+  receipt: string;
+}
+
+export interface ReceiptChallengeReply {
+  ok: boolean;
+  code?: string;
+  state?: 'UNVERIFIED' | 'PENDING_VERIFICATION' | 'PENDING_PROMOTION' | 'VERIFIED';
+}
+
+/**
+ * Complete the shipped begin -> trusted-observer -> consume -> promote path.
+ * The private fixture is read by the daemon's injected test observer; the
+ * observation request itself never carries assistant-message content.
+ */
+export async function observeAndPromoteReceiptChallenge(
+  socketPath: string,
+  observerFixturePath: string,
+  cycle: ReceiptChallengeCycle,
+): Promise<ReceiptChallengeReply> {
+  await writeFile(
+    observerFixturePath,
+    `${JSON.stringify({
+      session_id: cycle.sessionId,
+      runtime_generation: cycle.runtimeGeneration,
+      latest_assistant_message: cycle.receipt,
+    })}\n`,
+    { encoding: 'utf8', mode: 0o600 },
+  );
+  await chmod(observerFixturePath, 0o600);
+  const observed = await requestBrokerReply<ReceiptChallengeReply>(socketPath, {
+    action: 'observe_receipt',
+    session_id: cycle.sessionId,
+    runtime_generation: cycle.runtimeGeneration,
+    receipt_challenge: cycle.receiptChallenge,
+  });
+  if (observed.ok !== true || observed.state !== 'PENDING_PROMOTION') return observed;
+  return await requestBrokerReply<ReceiptChallengeReply>(socketPath, {
+    action: 'promote_lease',
+    session_id: cycle.sessionId,
+    runtime_generation: cycle.runtimeGeneration,
+    receipt_challenge: cycle.receiptChallenge,
+  });
+}

packages/mosaic/src/lease-broker/context_recovery_unittest.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""RED-first WI-6 contracts against the shipped constrained recovery entrypoint."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import importlib.util
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+
+TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
+DAEMON_PATH = TOOLS / "daemon.py"
+FRAGMENTS_PATH = TOOLS / "normative_fragments.py"
+OBSERVER_PATH = TOOLS / "receipt_observer.py"
+
+
+def load_module(name: str, path: Path):
+    assert path.is_file(), f"shipped module is missing: {path}"
+    spec = importlib.util.spec_from_file_location(name, path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"unable to load {name}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+DAEMON = load_module("lease_broker_recovery_daemon", DAEMON_PATH)
+FRAGMENTS = load_module("lease_broker_recovery_fragments", FRAGMENTS_PATH)
+OBSERVER = load_module("lease_broker_recovery_observer", OBSERVER_PATH)
+
+
+class RecoveryFixture(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        root = Path(self.temporary.name)
+        os.chmod(root, 0o700)
+        self.peer = (os.getpid(), os.getuid(), os.getgid())
+        self.observer = OBSERVER.TestReceiptObserver()
+        self.broker = DAEMON.Broker(DAEMON.StateStore(root / "state.json"), observer=self.observer)
+        registered = self.broker.handle(self.peer, {
+            "action": "register_anchor",
+            "runtime_generation": 11,
+        })
+        self.session_id = registered["session_id"]
+
+    def tearDown(self) -> None:
+        self.temporary.cleanup()
+
+    def construction_and_binding(self) -> tuple[dict[str, object], dict[str, object]]:
+        content = b"Mosaic recovery authority\n"
+        construction = {
+            "manifest_version": 1,
+            "generator_version": "wi6-recovery-test",
+            "fragments": [{
+                "source_id": "authority/recovery",
+                "content_base64": base64.b64encode(content).decode("ascii"),
+                "expected_sha256": hashlib.sha256(content).hexdigest(),
+            }],
+        }
+        built = FRAGMENTS.build_payload_from_wire(construction)
+        self.assertEqual(built.injectionDecision, "ACCEPTED")
+        self.assertTrue(built.promotion)
+        return construction, {
+            "compaction_epoch": 17,
+            "request_epoch": 23,
+            "h_source": built.h_source,
+            "h_payload": built.h_payload,
+            "schema_version": 1,
+        }
+
+    def begin_normal(self) -> dict[str, object]:
+        construction, binding = self.construction_and_binding()
+        return self.broker.handle(self.peer, {
+            "action": "begin_verification",
+            "session_id": self.session_id,
+            "runtime_generation": 11,
+            "runtime": "pi",
+            "binding": binding,
+            "construction": construction,
+        })
+
+    def begin_recovery(self) -> dict[str, object]:
+        construction, binding = self.construction_and_binding()
+        return self.broker.handle(self.peer, {
+            "action": "begin_recovery",
+            "session_id": self.session_id,
+            "runtime_generation": 11,
+            "runtime": "pi",
+            "binding": binding,
+            "construction": construction,
+        })
+
+    def complete_recovery(self) -> dict[str, object]:
+        return self.broker.handle(self.peer, {
+            "action": "complete_recovery",
+            "session_id": self.session_id,
+            "runtime_generation": 11,
+        })
+
+    def assert_recovery_refused_unverified(self, code: str) -> None:
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, code):
+            self.complete_recovery()
+        self.assertEqual(self.broker.leases[self.session_id]["state"], DAEMON.LEASE_UNVERIFIED)
+        denied = self.broker.handle(self.peer, {
+            "action": "authorize_tool",
+            "session_id": self.session_id,
+            "runtime_generation": 11,
+            "runtime": "pi",
+            "tool_name": "bash",
+        })
+        self.assertEqual(denied["decision"], "deny")
+
+
+class ConstrainedRecoveryContractTest(RecoveryFixture):
+    def test_recovery_mints_a_fresh_challenge_distinct_from_normal_path(self) -> None:
+        normal = self.begin_normal()
+        recovery = self.begin_recovery()
+
+        self.assertEqual(recovery["state"], "PENDING_DELIVERY")
+        self.assertNotEqual(normal["receipt_challenge"], recovery["receipt_challenge"])
+        self.assertIn(recovery["receipt_challenge"], recovery["receipt"])
+
+    def test_c4_normal_path_receipt_cannot_be_replayed_through_recovery(self) -> None:
+        normal = self.begin_normal()
+        recovery = self.begin_recovery()
+        self.assertNotEqual(normal["receipt_challenge"], recovery["receipt_challenge"])
+
+        self.observer.record_latest_assistant_message(self.session_id, 11, normal["receipt"])
+        self.assert_recovery_refused_unverified("RECEIPT_MISMATCH")
+
+    def test_t27_observable_partial_delivery_variants_never_promote(self) -> None:
+        variants = {
+            "absent": None,
+            "malformed": "MOSAIC-RECEIPT{malformed}",
+            "prefix-truncated": None,
+            "observable-adapter-mutation": None,
+            # Tail-only is represented only by this concrete malformed/incomplete
+            # delivery. It is not a category-wide tail-only detection claim.
+            "tail-only-malformed": "H_payload=tail-only",
+        }
+        for name, observed in variants.items():
+            with self.subTest(name=name):
+                recovery = self.begin_recovery()
+                if name == "prefix-truncated":
+                    observed = recovery["receipt"][:-1]
+                elif name == "observable-adapter-mutation":
+                    observed = recovery["receipt"].replace("H_payload=", "H_payload=0", 1)
+                if observed is not None:
+                    self.observer.record_latest_assistant_message(self.session_id, 11, observed)
+                self.assert_recovery_refused_unverified(
+                    "RECEIPT_OBSERVATION_UNAVAILABLE" if observed is None else "RECEIPT_MISMATCH"
+                )
+
+    def test_negative_capability_tail_preserving_middle_drop_is_not_receipt_detectable(self) -> None:
+        recovery = self.begin_recovery()
+
+        # The observer seam receives the exact terminal message, not the delivered
+        # payload bytes. A middle drop that preserves this tail is therefore T-C
+        # and deliberately NOT represented as receipt-detectable; WI-7 server-side
+        # evidence owns that residual. This is not an assertion that it is caught.
+        self.observer.record_latest_assistant_message(self.session_id, 11, recovery["receipt"])
+        promoted = self.complete_recovery()
+        self.assertEqual(promoted["state"], DAEMON.LEASE_VERIFIED)
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/mosaic/src/lease-broker/framework_skill_portability_unittest.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""RED-first framework-firewall and portability contracts for shipped skills."""
+
+from __future__ import annotations
+
+import importlib.util
+import re
+import unittest
+from pathlib import Path
+
+
+MOSAIC = Path(__file__).parents[2]
+SKILLS = MOSAIC / "framework/skills"
+REFRESH_SKILL = SKILLS / "mosaic-context-refresh/SKILL.md"
+GATE_PATH = MOSAIC / "framework/tools/lease-broker/mutator-gate.py"
+OPERATOR_HOME = re.compile(r"/home/[^/\s]+/")
+RECOVERY_PLACEHOLDER = "/absolute/path/to/mosaic/tools/lease-broker/recover-context.py"
+CONSTRUCTION_PLACEHOLDER = "/absolute/path/to/mosaic-context-refresh-construction.json"
+
+
+def load_gate():
+    spec = importlib.util.spec_from_file_location("framework_skill_portability_gate", GATE_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("unable to load mutator gate")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+GATE = load_gate()
+
+
+class FrameworkSkillPortabilityTest(unittest.TestCase):
+    def test_shipped_framework_skills_contain_no_operator_home_path(self) -> None:
+        offenders = [
+            str(path.relative_to(SKILLS))
+            for path in SKILLS.rglob("SKILL.md")
+            if OPERATOR_HOME.search(path.read_text(encoding="utf-8"))
+        ]
+        self.assertEqual(offenders, [])
+
+    def test_shipped_recovery_template_resolves_to_a_literal_install_path_and_admits(self) -> None:
+        source = REFRESH_SKILL.read_text(encoding="utf-8")
+        self.assertIn(RECOVERY_PLACEHOLDER, source)
+        self.assertIn(CONSTRUCTION_PLACEHOLDER, source)
+        resolved_recovery = "/opt/mosaic/tools/lease-broker/recover-context.py"
+        resolved_construction = "/opt/mosaic/recovery/construction.json"
+        rendered = source.replace(RECOVERY_PLACEHOLDER, resolved_recovery).replace(
+            CONSTRUCTION_PLACEHOLDER, resolved_construction
+        )
+        match = re.search(r"```bash\s*\n\s*(.*?)\n\s*```", rendered, flags=re.DOTALL)
+        self.assertIsNotNone(match)
+        command = match.group(1) if match is not None else ""
+        self.assertEqual(
+            GATE.recovery_invocation_name(
+                {"tool_name": "Bash", "tool_input": {"command": command}},
+                Path(resolved_recovery),
+            ),
+            GATE.RECOVERY_TOOL,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/mosaic/src/lease-broker/normative_fragments_unittest.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+"""RED-first contract tests for verbatim-hashed normative fragments."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import tempfile
+import unittest
+from pathlib import Path
+
+
+MODULE_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/normative_fragments.py"
+
+
+def shipped_module():
+    # Each test reaches the shipped implementation; no test doubles or local
+    # reimplementation of construction are allowed on this admission surface.
+    assert MODULE_PATH.is_file(), f"shipped construction module is missing: {MODULE_PATH}"
+    spec = importlib.util.spec_from_file_location("normative_fragments", MODULE_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("unable to load normative fragment construction")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def fragment(module, source_id: str, content: bytes):
+    return module.NormativeFragment(
+        source_id=source_id,
+        content=content,
+        expected_sha256=hashlib.sha256(content).hexdigest(),
+    )
+
+
+def valid_fragments(module):
+    return [
+        fragment(module, "authority/constitution", b"Constitution\n"),
+        fragment(module, "authority/runtime", b"Runtime\n"),
+    ]
+
+
+class NormativeFragmentsTest(unittest.TestCase):
+    def test_t24_claude_and_pi_builders_are_byte_identical_and_one_way(self) -> None:
+        module = shipped_module()
+        fragments = valid_fragments(module)
+
+        claude = module.build_for_claude(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=fragments,
+        )
+        pi = module.build_for_pi(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=fragments,
+        )
+
+        self.assertEqual(claude.injectionDecision, "ACCEPTED")
+        self.assertTrue(claude.promotion)
+        self.assertEqual(claude.b_payload, pi.b_payload)
+        self.assertEqual(claude.h_payload, pi.h_payload)
+        self.assertEqual(
+            claude.h_payload,
+            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + module.length_frame([claude.b_payload])).hexdigest(),
+        )
+        self.assertNotIn(b"h_payload", claude.b_payload)
+
+        mutated_fragment = module.build_for_claude(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=[
+                fragment(module, "authority/constitution", b"Constitution changed\n"),
+                fragment(module, "authority/runtime", b"Runtime\n"),
+            ],
+        )
+        mutated_metadata = module.build_for_claude(
+            manifest_version=2,
+            generator_version="wi4-test",
+            fragments=fragments,
+        )
+        self.assertNotEqual(claude.h_payload, mutated_fragment.h_payload)
+        self.assertNotEqual(claude.h_payload, mutated_metadata.h_payload)
+
+    def test_length_framing_and_domain_separation_prevent_ambiguous_construction(self) -> None:
+        module = shipped_module()
+        left = module.length_frame([b"ab", b"c"])
+        right = module.length_frame([b"a", b"bc"])
+
+        self.assertNotEqual(left, right)
+        self.assertNotEqual(
+            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(),
+            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + right).hexdigest(),
+        )
+        self.assertNotEqual(
+            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(),
+            hashlib.sha256(b"other-context\x00" + left).hexdigest(),
+        )
+
+    def test_source_invalidation_missing_refuses_real_construction_and_promotion(self) -> None:
+        module = shipped_module()
+        missing = module.NormativeFragment(
+            source_id="authority/missing",
+            content=None,
+            expected_sha256=hashlib.sha256(b"missing").hexdigest(),
+        )
+
+        result = module.build_for_claude(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=[missing],
+        )
+
+        self.assertEqual(result.injectionDecision, "REFUSED")
+        self.assertFalse(result.promotion)
+        self.assertEqual(result.source_reason, "missing")
+        self.assertIsNone(result.b_payload)
+        self.assertIsNone(result.h_payload)
+
+    def test_source_invalidation_oversize_refuses_real_construction_and_promotion(self) -> None:
+        module = shipped_module()
+        content = b"x" * (module.MAX_FRAGMENT_BYTES + 1)
+        oversize = fragment(module, "authority/oversize", content)
+
+        result = module.build_for_pi(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=[oversize],
+        )
+
+        self.assertEqual(result.injectionDecision, "REFUSED")
+        self.assertFalse(result.promotion)
+        self.assertEqual(result.source_reason, "oversize")
+        self.assertIsNone(result.b_payload)
+        self.assertIsNone(result.h_payload)
+
+    def test_source_invalidation_hash_mismatch_refuses_real_construction_and_promotion(self) -> None:
+        module = shipped_module()
+        mismatch = module.NormativeFragment(
+            source_id="authority/hash-mismatch",
+            content=b"trusted bytes",
+            expected_sha256=hashlib.sha256(b"different bytes").hexdigest(),
+        )
+
+        result = module.build_for_claude(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=[mismatch],
+        )
+
+        self.assertEqual(result.injectionDecision, "REFUSED")
+        self.assertFalse(result.promotion)
+        self.assertEqual(result.source_reason, "hash-mismatch")
+        self.assertIsNone(result.b_payload)
+        self.assertIsNone(result.h_payload)
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/mosaic/src/lease-broker/receipt_challenge_unittest.py

@@ -0,0 +1,227 @@
+#!/usr/bin/env python3
+"""RED-first contracts for the shipped receipt challenge and observer seam."""
+
+from __future__ import annotations
+
+import base64
+import copy
+import hashlib
+import importlib.util
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+
+TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
+DAEMON_PATH = TOOLS / "daemon.py"
+FRAGMENTS_PATH = TOOLS / "normative_fragments.py"
+OBSERVER_PATH = TOOLS / "receipt_observer.py"
+
+
+def load_module(name: str, path: Path):
+    assert path.is_file(), f"shipped module is missing: {path}"
+    spec = importlib.util.spec_from_file_location(name, path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"unable to load {name}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+DAEMON = load_module("lease_broker_receipt_daemon", DAEMON_PATH)
+FRAGMENTS = load_module("lease_broker_normative_fragments", FRAGMENTS_PATH)
+
+
+class BrokerFixture(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        root = Path(self.temporary.name)
+        os.chmod(root, 0o700)
+        self.peer = (os.getpid(), os.getuid(), os.getgid())
+        self.broker = DAEMON.Broker(DAEMON.StateStore(root / "state.json"))
+        registered = self.broker.handle(self.peer, {
+            "action": "register_anchor",
+            "runtime_generation": 7,
+        })
+        self.session_id = registered["session_id"]
+        self.assertIsInstance(self.session_id, str)
+
+    def tearDown(self) -> None:
+        self.temporary.cleanup()
+
+    def construction(self) -> tuple[dict[str, object], dict[str, object]]:
+        content = b"Constitution\n"
+        expected_sha256 = hashlib.sha256(content).hexdigest()
+        construction = {
+            "manifest_version": 1,
+            "generator_version": "wi5-receipt-test",
+            "fragments": [{
+                "source_id": "authority/constitution",
+                "content_base64": base64.b64encode(content).decode("ascii"),
+                "expected_sha256": expected_sha256,
+            }],
+        }
+        result = FRAGMENTS.build_payload(
+            manifest_version=construction["manifest_version"],
+            generator_version=construction["generator_version"],
+            fragments=[FRAGMENTS.NormativeFragment("authority/constitution", content, expected_sha256)],
+        )
+        self.assertEqual(result.injectionDecision, "ACCEPTED")
+        self.assertTrue(result.promotion)
+        return construction, {
+            "compaction_epoch": 3,
+            "request_epoch": 8,
+            "h_source": result.h_source,
+            "h_payload": result.h_payload,
+            "schema_version": 1,
+        }
+
+    def begin(self, binding: dict[str, object], construction: dict[str, object]) -> dict[str, object]:
+        response = self.broker.handle(self.peer, {
+            "action": "begin_verification",
+            "session_id": self.session_id,
+            "runtime_generation": 7,
+            "runtime": "pi",
+            "binding": binding,
+            "construction": construction,
+        })
+        self.assertEqual(response["state"], DAEMON.LEASE_PENDING)
+        self.assertIsInstance(response.get("receipt_challenge"), str)
+        self.assertIsInstance(response.get("receipt"), str)
+        return response
+
+
+class BuildPayloadAdmissionTest(BrokerFixture):
+    def test_b3_forged_h_source_or_h_payload_is_refused_against_shipped_build_payload(self) -> None:
+        construction, trusted = self.construction()
+        for field in ("h_source", "h_payload"):
+            with self.subTest(field=field):
+                forged = dict(trusted)
+                forged[field] = "f" * 64
+                with self.assertRaisesRegex(DAEMON.BrokerFailure, "PAYLOAD_BINDING_MISMATCH"):
+                    self.begin(forged, construction)
+
+
+class ReceiptObserverTest(BrokerFixture):
+    def setUp(self) -> None:
+        super().setUp()
+        observers = load_module("lease_broker_test_observer", OBSERVER_PATH)
+        self.observer = observers.TestReceiptObserver()
+        self.broker = DAEMON.Broker(self.broker.store, observer=self.observer)
+
+    def record(self, message: str) -> None:
+        self.observer.record_latest_assistant_message(self.session_id, 7, message)
+
+    def observe(self, challenge: str, **untrusted: object) -> dict[str, object]:
+        return self.broker.handle(self.peer, {
+            "action": "observe_receipt",
+            "session_id": self.session_id,
+            "runtime_generation": 7,
+            "receipt_challenge": challenge,
+            **untrusted,
+        })
+
+    def promote(self, challenge: str) -> dict[str, object]:
+        return self.broker.handle(self.peer, {
+            "action": "promote_lease",
+            "session_id": self.session_id,
+            "runtime_generation": 7,
+            "receipt_challenge": challenge,
+        })
+
+    def test_b2_echoed_request_observation_is_refused_but_observer_source_promotes(self) -> None:
+        construction, binding = self.construction()
+        cycle = self.begin(binding, construction)
+        challenge = cycle["receipt_challenge"]
+        receipt = cycle["receipt"]
+
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_RECEIPT"):
+            self.observe(challenge, latest_assistant_message=receipt)
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_OBSERVATION_UNAVAILABLE"):
+            self.observe(challenge)
+
+        self.record(receipt)
+        self.assertEqual(self.observe(challenge)["state"], DAEMON.LEASE_PENDING_PROMOTION)
+        self.assertEqual(self.promote(challenge)["state"], DAEMON.LEASE_VERIFIED)
+        self.assertEqual(self.broker.handle(self.peer, {
+            "action": "authorize_tool",
+            "session_id": self.session_id,
+            "runtime_generation": 7,
+            "runtime": "pi",
+            "tool_name": "bash",
+        })["decision"], "allow")
+
+    def test_rejected_begin_keeps_revoke_first_fence_for_all_construction_refusals(self) -> None:
+        construction, binding = self.construction()
+        refusal_cases = {
+            "INVALID_CONSTRUCTION": {"bad": "construction"},
+            "PAYLOAD_CONSTRUCTION_REFUSED": {
+                **construction,
+                "fragments": [{
+                    **construction["fragments"][0],
+                    "expected_sha256": "0" * 64,
+                }],
+            },
+            "PAYLOAD_BINDING_MISMATCH": None,
+        }
+        for expected_code, rejected_construction in refusal_cases.items():
+            with self.subTest(expected_code=expected_code):
+                verified = self.begin(binding, construction)
+                self.record(verified["receipt"])
+                self.observe(verified["receipt_challenge"])
+                self.assertEqual(self.promote(verified["receipt_challenge"])["state"], DAEMON.LEASE_VERIFIED)
+
+                rejected_binding = copy.deepcopy(binding)
+                if expected_code == "PAYLOAD_BINDING_MISMATCH":
+                    rejected_binding["h_payload"] = "f" * 64
+                    rejected_construction = construction
+                with self.assertRaisesRegex(DAEMON.BrokerFailure, expected_code):
+                    self.begin(rejected_binding, rejected_construction)
+
+                self.assertEqual(
+                    self.broker.leases[self.session_id]["state"], DAEMON.LEASE_UNVERIFIED
+                )
+                denied = self.broker.handle(self.peer, {
+                    "action": "authorize_tool",
+                    "session_id": self.session_id,
+                    "runtime_generation": 7,
+                    "runtime": "pi",
+                    "tool_name": "bash",
+                })
+                self.assertEqual(denied["decision"], "deny")
+                self.assertEqual(denied["state"], DAEMON.LEASE_UNVERIFIED)
+
+    def test_t26_stale_epoch_receipt_cannot_promote_against_shipped_binding(self) -> None:
+        construction, stale_binding = self.construction()
+        stale = self.begin(stale_binding, construction)
+        current_binding = dict(stale_binding)
+        current_binding["compaction_epoch"] = 4
+        current_binding["request_epoch"] = 9
+        current = self.begin(current_binding, construction)
+        self.assertNotEqual(stale["receipt_challenge"], current["receipt_challenge"])
+
+        self.record(stale["receipt"])
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
+            self.observe(current["receipt_challenge"])
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
+            self.promote(current["receipt_challenge"])
+
+    def test_t29_altered_model_hash_cannot_promote_against_shipped_binding(self) -> None:
+        construction, binding = self.construction()
+        cycle = self.begin(binding, construction)
+        expected = cycle["receipt"]
+        altered_hash = "f" * 64
+        self.assertNotEqual(altered_hash, cycle["binding"]["h_payload"])
+        altered = expected.replace(cycle["binding"]["h_payload"], altered_hash, 1)
+        self.assertNotEqual(altered, expected)
+
+        self.record(altered)
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
+            self.observe(cycle["receipt_challenge"])
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
+            self.promote(cycle["receipt_challenge"])
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/mosaic/src/lease-broker/recovery_b1_adversarial_unittest.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+"""RED-first adversarial Claude B1 gate contracts.
+
+This private harness drives the shipped gate and daemon out of process.  It
+never contacts a live broker/runtime and executes a shell payload only after a
+regression has already (incorrectly) received the recovery exemption.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+
+TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
+FRAMEWORK = Path(__file__).parents[2] / "framework"
+DAEMON = TOOLS / "daemon.py"
+GATE = TOOLS / "mutator-gate.py"
+RECOVERY = TOOLS / "recover-context.py"
+SKILL = FRAMEWORK / "skills/mosaic-context-refresh/SKILL.md"
+
+
+def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]:
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(3.0)
+        connection.connect(str(socket_path))
+        connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
+        connection.shutdown(socket.SHUT_WR)
+        response = bytearray()
+        while True:
+            chunk = connection.recv(4096)
+            if not chunk:
+                break
+            response.extend(chunk)
+    if not response.endswith(b"\n") or response.count(b"\n") != 1:
+        raise AssertionError(f"unframed broker response: {bytes(response)!r}")
+    reply = json.loads(response[:-1])
+    if not isinstance(reply, dict):
+        raise AssertionError("broker response is not an object")
+    return reply
+
+
+def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None:
+    deadline = time.monotonic() + 5.0
+    while time.monotonic() < deadline:
+        if socket_path.exists():
+            return
+        if process.poll() is not None:
+            output = process.stdout.read() if process.stdout is not None else ""
+            raise RuntimeError(f"daemon exited before READY: {output}")
+        time.sleep(0.02)
+    raise TimeoutError("daemon did not create private socket")
+
+
+class ClaudeRecoveryGateAdversarialTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name)
+        os.chmod(self.root, 0o700)
+        self.socket = self.root / "broker.sock"
+        self.daemon = subprocess.Popen(
+            [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(self.socket), "--state", str(self.root / "state.json")],
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+        )
+        wait_ready(self.daemon, self.socket)
+        registered = request(self.socket, {"action": "register_anchor", "runtime_generation": 1})
+        self.session_id = registered["session_id"]
+        self.environment = {
+            **os.environ,
+            "MOSAIC_LEASE_BROKER_SOCKET": str(self.socket),
+            "MOSAIC_LEASE_SESSION_ID": self.session_id,
+            "MOSAIC_RUNTIME_GENERATION": "1",
+            "MOSAIC_LEASE_RUNTIME": "claude",
+        }
+        # This private path is only a gate-classifier identity; if a regression
+        # blesses a payload, bash invokes no installed/live recovery command.
+        self.recovery_path = self.root / "recover-context.py"
+        self.canonical = (
+            f"python3 {self.recovery_path} begin "
+            f"--construction {self.root / 'construction.json'} --compaction-epoch 0 --request-epoch 0"
+        )
+
+    def tearDown(self) -> None:
+        if self.daemon.poll() is None:
+            self.daemon.terminate()
+            try:
+                self.daemon.wait(timeout=3.0)
+            except subprocess.TimeoutExpired:
+                self.daemon.kill()
+                self.daemon.wait()
+        if self.daemon.stdout is not None:
+            self.daemon.stdout.close()
+        self.temporary.cleanup()
+
+    def gate(
+        self, command: str, recovery_command: Path | str | None = None
+    ) -> subprocess.CompletedProcess[str]:
+        configured_recovery = self.recovery_path if recovery_command is None else recovery_command
+        return subprocess.run(
+            [
+                sys.executable,
+                "-I",
+                "-S",
+                "-B",
+                str(GATE),
+                "--runtime",
+                "claude",
+                "--recovery-command",
+                str(configured_recovery),
+            ],
+            input=json.dumps({"tool_name": "Bash", "tool_input": {"command": command}}),
+            text=True,
+            capture_output=True,
+            env=self.environment,
+            check=False,
+        )
+
+    def command_for(self, position: str, value: str) -> str:
+        argv = [
+            "python3",
+            str(self.recovery_path),
+            "begin",
+            "--construction",
+            str(self.root / "construction.json"),
+            "--compaction-epoch",
+            "0",
+            "--request-epoch",
+            "0",
+        ]
+        positions = {
+            "executable": 0,
+            "path": 1,
+            "phase": 2,
+            "construction_flag": 3,
+            "construction": 4,
+            "compaction_epoch_flag": 5,
+            "compaction_epoch": 6,
+            "request_epoch_flag": 7,
+            "request_epoch": 8,
+        }
+        try:
+            argv[positions[position]] = value
+        except KeyError as exc:
+            raise AssertionError(f"unknown argv position {position}") from exc
+        return " ".join(argv)
+
+    def test_canonical_literal_and_shipped_skill_invocation_are_ungated(self) -> None:
+        self.assertEqual(self.gate(self.canonical).returncode, 0)
+        source = SKILL.read_text(encoding="utf-8")
+        recovery_placeholder = "/absolute/path/to/mosaic/tools/lease-broker/recover-context.py"
+        construction_placeholder = "/absolute/path/to/mosaic-context-refresh-construction.json"
+        self.assertIn(recovery_placeholder, source)
+        self.assertIn(construction_placeholder, source)
+        resolved_recovery = "/opt/mosaic/tools/lease-broker/recover-context.py"
+        rendered = source.replace(recovery_placeholder, resolved_recovery).replace(
+            construction_placeholder, "/opt/mosaic/recovery/construction.json"
+        )
+        match = re.search(r"```bash\s*\n\s*(.*?)\n\s*```", rendered, flags=re.DOTALL)
+        self.assertIsNotNone(match)
+        shipped = match.group(1) if match is not None else ""
+        self.assertEqual(self.gate(shipped, resolved_recovery).returncode, 0)
+
+    def test_every_shell_active_vector_in_every_argv_position_falls_through_to_bash_deny(self) -> None:
+        positions = (
+            "executable",
+            "path",
+            "phase",
+            "construction_flag",
+            "construction",
+            "compaction_epoch_flag",
+            "compaction_epoch",
+            "request_epoch_flag",
+            "request_epoch",
+        )
+        marker = self.root / "PWNED"
+        marker_vector = f"$(touch${{IFS}}{marker})"
+        vectors = {
+            "command-substitution": marker_vector,
+            "backtick": f"`touch${{IFS}}{marker}`",
+            "process-substitution": f"<(touch${{IFS}}{marker})",
+            "parameter-expansion": "${IFS}",
+            "home-expansion": "${HOME}",
+            "arithmetic-expansion": "$((1+1))",
+            "brace-expansion": "{a,b}",
+            "tilde-expansion": "~",
+            "glob": "*",
+            "redirection": ">",
+            "semicolon": f";touch${{IFS}}{marker}",
+            "and": f"&&touch${{IFS}}{marker}",
+            "pipe": f"|touch${{IFS}}{marker}",
+            "embedded-newline": "literal\nnext",
+            "quoting-trick": "'literal'",
+        }
+        for position in positions:
+            for kind, vector in vectors.items():
+                with self.subTest(position=position, kind=kind):
+                    marker.unlink(missing_ok=True)
+                    command = self.command_for(position, vector)
+                    gated = self.gate(command)
+                    if gated.returncode == 0 and kind in {"command-substitution", "backtick", "process-substitution", "semicolon", "and", "pipe"}:
+                        subprocess.run(["bash", "-c", command], cwd=self.root, env=self.environment, check=False)
+                    self.assertEqual(gated.returncode, 2, f"unexpected recovery exemption: {command!r}")
+                    self.assertFalse(marker.exists(), f"shell payload executed: {command!r}")
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/mosaic/src/lease-broker/recovery_runtime_unittest.py

@@ -0,0 +1,282 @@
+#!/usr/bin/env python3
+"""RED-first B1/B2 runtime-boundary contracts for constrained recovery.
+
+Every runtime process in these tests is a fresh child against a private daemon
+and Unix sockets. They do not activate a live Mosaic daemon or model stream.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+
+TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
+FRAMEWORK = Path(__file__).parents[2] / "framework"
+DAEMON = TOOLS / "daemon.py"
+GATE = TOOLS / "mutator-gate.py"
+RECOVERY = TOOLS / "recover-context.py"
+OBSERVER_CLIENT = TOOLS / "receipt-observer-client.py"
+CLAUDE_SETTINGS = FRAMEWORK / "runtime/claude/settings.json"
+PI_EXTENSION = FRAMEWORK / "runtime/pi/mosaic-extension.ts"
+
+
+def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]:
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(3.0)
+        connection.connect(str(socket_path))
+        connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
+        connection.shutdown(socket.SHUT_WR)
+        response = bytearray()
+        while True:
+            chunk = connection.recv(4096)
+            if not chunk:
+                break
+            response.extend(chunk)
+    if not response.endswith(b"\n") or response.count(b"\n") != 1:
+        raise AssertionError(f"unframed broker response: {bytes(response)!r}")
+    reply = json.loads(response[:-1])
+    if not isinstance(reply, dict):
+        raise AssertionError("broker response is not an object")
+    return reply
+
+
+def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None:
+    deadline = time.monotonic() + 5.0
+    while time.monotonic() < deadline:
+        if socket_path.exists():
+            return
+        if process.poll() is not None:
+            output = process.stdout.read() if process.stdout is not None else ""
+            raise RuntimeError(f"daemon exited before READY: {output}")
+        time.sleep(0.02)
+    raise TimeoutError("daemon did not create private broker socket")
+
+
+class RuntimeBoundaryFixture(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name)
+        os.chmod(self.root, 0o700)
+        self.socket = self.root / "broker.sock"
+        self.observer_socket = self.root / "observer.sock"
+        self.state = self.root / "state.json"
+        self.children: list[subprocess.Popen[str]] = []
+
+    def tearDown(self) -> None:
+        for child in self.children:
+            if child.poll() is None:
+                child.terminate()
+                try:
+                    child.wait(timeout=3.0)
+                except subprocess.TimeoutExpired:
+                    child.kill()
+                    child.wait()
+            if child.stdout is not None:
+                child.stdout.close()
+        self.temporary.cleanup()
+
+    def start_daemon(self, *, production_observer: bool) -> None:
+        arguments = [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(self.socket), "--state", str(self.state)]
+        if production_observer:
+            arguments.extend(["--observer-socket", str(self.observer_socket)])
+        process = subprocess.Popen(
+            arguments,
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+        )
+        self.children.append(process)
+        wait_ready(process, self.socket)
+
+    def register(self) -> str:
+        reply = request(self.socket, {"action": "register_anchor", "runtime_generation": 1})
+        session_id = reply.get("session_id")
+        self.assertTrue(reply.get("ok"))
+        self.assertIsInstance(session_id, str)
+        return session_id
+
+    def environment(self, session_id: str) -> dict[str, str]:
+        return {
+            **os.environ,
+            "MOSAIC_LEASE_BROKER_SOCKET": str(self.socket),
+            "MOSAIC_RECEIPT_OBSERVER_SOCKET": str(self.observer_socket),
+            "MOSAIC_LEASE_SESSION_ID": session_id,
+            "MOSAIC_RUNTIME_GENERATION": "1",
+            "MOSAIC_LEASE_RUNTIME": "pi",
+        }
+
+    def construction(self) -> Path:
+        content = b"WI-6 B2 production observer fixture\n"
+        path = self.root / "construction.json"
+        path.write_text(json.dumps({
+            "manifest_version": 1,
+            "generator_version": "wi6-repair-runtime-boundary",
+            "fragments": [{
+                "source_id": "authority/wi6-repair",
+                "content_base64": base64.b64encode(content).decode("ascii"),
+                "expected_sha256": hashlib.sha256(content).hexdigest(),
+            }],
+        }), encoding="utf-8")
+        os.chmod(path, 0o600)
+        return path
+
+
+class RecoveryRuntimeBoundaryTest(RuntimeBoundaryFixture):
+    def test_b1_claude_exact_recovery_command_is_invocable_unverified_but_bash_is_not(self) -> None:
+        self.start_daemon(production_observer=False)
+        session_id = self.register()
+        environment = self.environment(session_id)
+        recovery_command = f"python3 {RECOVERY} complete"
+        mapped = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)],
+            input=json.dumps({"tool_name": "Bash", "tool_input": {"command": recovery_command}}),
+            text=True,
+            capture_output=True,
+            env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"},
+            check=False,
+        )
+        self.assertEqual(mapped.returncode, 0, mapped.stderr)
+        mapped_begin = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)],
+            input=json.dumps({
+                "tool_name": "Bash",
+                "tool_input": {
+                    "command": (
+                        f"python3 {RECOVERY} begin --construction /tmp/construction.json "
+                        "--compaction-epoch 1 --request-epoch 1"
+                    )
+                },
+            }),
+            text=True,
+            capture_output=True,
+            env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"},
+            check=False,
+        )
+        self.assertEqual(mapped_begin.returncode, 0, mapped_begin.stderr)
+        ordinary_bash = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)],
+            input=json.dumps({"tool_name": "Bash", "tool_input": {"command": "echo not-recovery"}}),
+            text=True,
+            capture_output=True,
+            env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"},
+            check=False,
+        )
+        self.assertEqual(ordinary_bash.returncode, 2)
+
+    def test_b1_pi_registers_only_the_broker_exempt_recovery_tool(self) -> None:
+        extension = PI_EXTENSION.read_text(encoding="utf-8")
+        self.assertIn("const RECOVERY_TOOL = 'mosaic_context_recover'", extension)
+        self.assertIn("name: RECOVERY_TOOL", extension)
+        self.assertIn("checkPiMutatorGate(RECOVERY_TOOL)", extension)
+        self.assertNotIn("toolName === 'bash' ? RECOVERY_TOOL", extension)
+
+    def test_b2_production_observer_promotes_over_private_transport_and_rejects_broker_supplied_message(self) -> None:
+        self.start_daemon(production_observer=True)
+        session_id = self.register()
+        environment = self.environment(session_id)
+        begin = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(RECOVERY), "begin", "--construction", str(self.construction()), "--compaction-epoch", "1", "--request-epoch", "1"],
+            text=True,
+            capture_output=True,
+            env=environment,
+            check=False,
+        )
+        self.assertEqual(begin.returncode, 0, begin.stderr)
+        cycle = json.loads(begin.stdout)
+        receipt = cycle.get("receipt")
+        self.assertIsInstance(receipt, str)
+
+        # S1: production transport is not a broker request field. The public
+        # broker endpoint keeps rejecting caller-supplied assistant evidence.
+        rejected_begin = request(self.socket, {
+            "action": "begin_recovery",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt": receipt,
+        })
+        self.assertEqual(rejected_begin, {"ok": False, "code": "INVALID_RECOVERY_REQUEST"})
+        rejected_observe = request(self.socket, {
+            "action": "observe_receipt",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": cycle["receipt_challenge"],
+            "latest_assistant_message": receipt,
+        })
+        self.assertEqual(rejected_observe, {"ok": False, "code": "INVALID_RECEIPT"})
+        rejected_complete = request(self.socket, {
+            "action": "complete_recovery",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "latest_assistant_message": receipt,
+        })
+        self.assertEqual(rejected_complete, {"ok": False, "code": "INVALID_RECOVERY_REQUEST"})
+
+        recorded = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(OBSERVER_CLIENT), "--runtime", "pi"],
+            input=json.dumps({"latest_assistant_message": receipt}),
+            text=True,
+            capture_output=True,
+            env=environment,
+            check=False,
+        )
+        self.assertEqual(recorded.returncode, 0, recorded.stderr)
+        complete = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(RECOVERY), "complete"],
+            text=True,
+            capture_output=True,
+            env=environment,
+            check=False,
+        )
+        self.assertEqual(complete.returncode, 0, complete.stderr)
+        self.assertEqual(json.loads(complete.stdout).get("state"), "VERIFIED")
+
+        # The independent Claude transport selects one latest assistant entry
+        # from its hook transcript; it is not a Pi/message_end fallback.
+        claude_environment = {**environment, "MOSAIC_LEASE_RUNTIME": "claude"}
+        claude_begin = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(RECOVERY), "begin", "--construction", str(self.construction()), "--compaction-epoch", "2", "--request-epoch", "2"],
+            text=True,
+            capture_output=True,
+            env=claude_environment,
+            check=False,
+        )
+        self.assertEqual(claude_begin.returncode, 0, claude_begin.stderr)
+        claude_receipt = json.loads(claude_begin.stdout)["receipt"]
+        transcript = self.root / "claude-transcript.jsonl"
+        transcript.write_text(
+            json.dumps({"message": {"role": "assistant", "content": claude_receipt}}) + "\n",
+            encoding="utf-8",
+        )
+        claude_recorded = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(OBSERVER_CLIENT), "--runtime", "claude", "--latest-entry"],
+            input=json.dumps({"transcript_path": str(transcript)}),
+            text=True,
+            capture_output=True,
+            env=claude_environment,
+            check=False,
+        )
+        self.assertEqual(claude_recorded.returncode, 0, claude_recorded.stderr)
+        claude_complete = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(RECOVERY), "complete"],
+            text=True,
+            capture_output=True,
+            env=claude_environment,
+            check=False,
+        )
+        self.assertEqual(claude_complete.returncode, 0, claude_complete.stderr)
+        self.assertEqual(json.loads(claude_complete.stdout).get("state"), "VERIFIED")
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts

@@ -6,42 +6,73 @@ import { spawn, spawnSync, type ChildProcess } from 'node:child_process';
 import { afterEach, describe, expect, test } from 'vitest';
 
 import { launchClaudex, type ClaudexHarnessAdapter } from '../commands/claudex.js';
-import { requestBrokerReply } from '../lease-broker/broker-test-client.js';
+import {
+  observeAndPromoteReceiptChallenge,
+  requestBrokerReply,
+} from '../lease-broker/broker-test-client.js';
 
 interface BrokerReply {
   ok: boolean;
   code?: string;
   decision?: 'allow' | 'deny';
-  state?: 'UNVERIFIED' | 'PENDING_VERIFICATION' | 'VERIFIED';
+  state?: 'UNVERIFIED' | 'PENDING_VERIFICATION' | 'PENDING_PROMOTION' | 'VERIFIED';
   session_id?: string;
-  promotion_token?: string;
+  receipt_challenge?: string;
+  receipt?: string;
+}
+
+interface PendingReceiptCycle {
+  sessionId: string;
+  runtimeGeneration: number;
+  receiptChallenge: string;
+  receipt: string;
 }
 
 interface BrokerPaths {
   socket: string;
+  observerFixture: string;
 }
 
 const frameworkRoot = new URL('../../framework/', import.meta.url).pathname;
+const repositoryRoot = new URL('../../../../', import.meta.url).pathname;
 const daemonPath = join(frameworkRoot, 'tools/lease-broker/daemon.py');
 const gatePath = join(frameworkRoot, 'tools/lease-broker/mutator-gate.py');
 const launchGuardPath = join(frameworkRoot, 'tools/lease-broker/check-runtime-launches.py');
 const launcherPath = join(frameworkRoot, 'tools/lease-broker/launch-runtime.py');
+const revokerPath = join(frameworkRoot, 'tools/lease-broker/revoke-lease.py');
+const compactionThreatPath = join(repositoryRoot, 'docs/architecture/compaction-revocation.md');
 const claudeSettingsPath = join(frameworkRoot, 'runtime/claude/settings.json');
 const piExtensionPath = join(frameworkRoot, 'runtime/pi/mosaic-extension.ts');
+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');
 const children: ChildProcess[] = [];
 const temporaryRoots: string[] = [];
 
+const construction = {
+  manifest_version: 1,
+  generator_version: 'mutator-gate-acceptance',
+  fragments: [
+    {
+      source_id: 'authority/mutator-gate-acceptance',
+      content_base64: 'bXV0YXRvci1nYXRlIGFjY2VwdGFuY2UK',
+      expected_sha256: 'cc3de191821d48037f60b4d006fce74b5dd394d39fb5c8bf681c28889ff0e623',
+    },
+  ],
+};
+
 const binding = (compaction_epoch = 1) => ({
   compaction_epoch,
   request_epoch: 0,
-  h_source: 'a'.repeat(64),
-  h_payload: 'b'.repeat(64),
+  h_source: '3c8fc6733d6a2bdc001ed9277d636d7cfac037ae48ed7d54203180a3839dc7a6',
+  h_payload: '42da889037c29c3a41397df0f86465ffd121bdb8cd18ad0d7c3df259ab502d3e',
   schema_version: 1,
 });
 
+const pendingReceiptCycles = new Map<string, PendingReceiptCycle>();
+const observerFixtures = new Map<string, string>();
+
 async function request(socketPath: string, requestValue: object): Promise<BrokerReply> {
   return await requestBrokerReply<BrokerReply>(socketPath, requestValue);
 }
@@ -50,9 +81,18 @@ async function startBroker(): Promise<BrokerPaths> {
   const root = await mkdtemp(join(tmpdir(), 'mosaic-mutator-gate-'));
   await chmod(root, 0o700);
   const socket = join(root, 'broker.sock');
+  const observerFixture = join(root, 'test-observer.json');
   const child = spawn(
     'python3',
-    [daemonPath, '--socket', socket, '--state', join(root, 'state.json')],
+    [
+      daemonPath,
+      '--socket',
+      socket,
+      '--state',
+      join(root, 'state.json'),
+      '--test-observer-file',
+      observerFixture,
+    ],
     {
       stdio: ['ignore', 'pipe', 'pipe'],
     },
@@ -68,7 +108,8 @@ async function startBroker(): Promise<BrokerPaths> {
     );
     child.stdout?.once('data', () => resolve());
   });
-  return { socket };
+  observerFixtures.set(socket, observerFixture);
+  return { socket, observerFixture };
 }
 
 interface RuntimeLaunchEntry {
@@ -162,28 +203,43 @@ async function beginVerification(
   ttl_seconds = 300,
   compactionEpoch = 1,
 ): Promise<BrokerReply> {
-  return await request(socket, {
+  const reply = await request(socket, {
     action: 'begin_verification',
     session_id,
     runtime_generation,
     runtime,
     ttl_seconds,
     binding: binding(compactionEpoch),
+    construction,
   });
+  if (typeof reply.receipt_challenge === 'string' && typeof reply.receipt === 'string') {
+    pendingReceiptCycles.set(reply.receipt_challenge, {
+      sessionId: session_id,
+      runtimeGeneration: runtime_generation,
+      receiptChallenge: reply.receipt_challenge,
+      receipt: reply.receipt,
+    });
+  }
+  return reply;
 }
 
 async function promote(
   socket: string,
   session_id: string,
-  promotion_token: string,
+  receipt_challenge: string,
   runtime_generation = 1,
 ): Promise<BrokerReply> {
-  return await request(socket, {
-    action: 'promote_lease',
-    session_id,
-    runtime_generation,
-    promotion_token,
-  });
+  const cycle = pendingReceiptCycles.get(receipt_challenge);
+  const observerFixture = observerFixtures.get(socket);
+  if (cycle === undefined || observerFixture === undefined) {
+    return await request(socket, {
+      action: 'promote_lease',
+      session_id,
+      runtime_generation,
+      receipt_challenge,
+    });
+  }
+  return await observeAndPromoteReceiptChallenge(socket, observerFixture, cycle);
 }
 
 async function authorize(
@@ -245,13 +301,13 @@ describe('whole mutator-class lease gate', () => {
 
     const pending = await beginVerification(socket, sessionId, 'claude');
     expect(pending).toMatchObject({ ok: true, state: 'PENDING_VERIFICATION' });
-    expect(pending.promotion_token).toMatch(/^[a-f0-9]{64}$/);
+    expect(pending.receipt_challenge).toMatch(/^[a-f0-9]{64}$/);
     expect(await authorize(socket, sessionId, 'claude', 'Write')).toMatchObject({
       ok: false,
       decision: 'deny',
     });
 
-    expect(await promote(socket, sessionId, pending.promotion_token!)).toMatchObject({
+    expect(await promote(socket, sessionId, pending.receipt_challenge!)).toMatchObject({
       ok: true,
       state: 'VERIFIED',
     });
@@ -267,9 +323,9 @@ describe('whole mutator-class lease gate', () => {
       ok: false,
       decision: 'deny',
     });
-    expect(await promote(socket, sessionId, pending.promotion_token!)).toMatchObject({
+    expect(await promote(socket, sessionId, pending.receipt_challenge!)).toMatchObject({
       ok: false,
-      code: 'PROMOTION_TOKEN_MISMATCH',
+      code: 'RECEIPT_REPLAY',
     });
   });
 
@@ -393,11 +449,220 @@ describe('whole mutator-class lease gate', () => {
     ).toMatchObject({ ok: false, code: 'INVALID_TOOL' });
   });
 
+  test('T12b/T30 reports the dual-observer-miss residual within and after TTL', async () => {
+    const { socket } = await startBroker();
+    const sessionId = await register(socket);
+    const pending = await beginVerification(socket, sessionId, 'claude', 1, 1);
+    await promote(socket, sessionId, pending.receipt_challenge!);
+
+    // Intentionally invoke neither compaction observer: this is the amended
+    // D2-v5 bounded residual, not a fail-closed path.
+    const withinTtl = await authorize(socket, sessionId, 'claude', 'Bash');
+    expect(withinTtl).toMatchObject({ ok: true, decision: 'allow', state: 'VERIFIED' });
+    console.info('T12b/T30 dual-hook-miss within-TTL: ALLOWED (bounded residual stale window)');
+
+    await new Promise((resolve) => setTimeout(resolve, 1_100));
+    const afterTtl = await authorize(socket, sessionId, 'claude', 'Bash');
+    expect(afterTtl).toMatchObject({
+      ok: false,
+      code: 'LEASE_EXPIRED',
+      decision: 'deny',
+    });
+    console.info('T12b/T30 dual-hook-miss after-TTL: DENIED (lease expiry)');
+
+    const threatContract = await readFile(compactionThreatPath, 'utf8');
+    expect(threatContract).toContain('BOUNDED RESIDUAL STALE WINDOW');
+    expect(threatContract).toContain('within-TTL consequential actions are allowed');
+    expect(threatContract).toContain('bounded by lease expiry, not by the mutator gate');
+  });
+
+  test.each([
+    { observer: 'Claude PreCompact', reason: 'pre-compact' },
+    { observer: 'Claude SessionStart(compact)', reason: 'session-start-compact' },
+  ])('$observer revokes a verified lease through the broker path', async ({ reason }) => {
+    const { socket } = await startBroker();
+    const sessionId = await register(socket);
+    const pending = await beginVerification(socket, sessionId, 'claude');
+    await promote(socket, sessionId, pending.receipt_challenge!);
+
+    const revoked = spawnSync('python3', [revokerPath, '--runtime', 'claude', '--reason', reason], {
+      encoding: 'utf8',
+      env: {
+        ...process.env,
+        MOSAIC_LEASE_BROKER_SOCKET: socket,
+        MOSAIC_LEASE_SESSION_ID: sessionId,
+        MOSAIC_RUNTIME_GENERATION: '1',
+      },
+    });
+    expect(revoked.status, revoked.stderr).toBe(0);
+    expect(await authorize(socket, sessionId, 'claude', 'Write')).toMatchObject({
+      ok: false,
+      code: 'MUTATOR_UNVERIFIED',
+      decision: 'deny',
+    });
+  });
+
+  test('promote-lease-lost-ACK orphaned VERIFIED lease is caught by observer revoke and by monotonic TTL expiry (D2-v5 backstop)', async () => {
+    const { socket } = await startBroker();
+    const observerSessionId = await register(socket);
+    const observerPending = await beginVerification(socket, observerSessionId, 'claude');
+    await promote(socket, observerSessionId, observerPending.receipt_challenge!);
+
+    expect(await authorize(socket, observerSessionId, 'claude', 'Bash')).toMatchObject({
+      ok: true,
+      decision: 'allow',
+      state: 'VERIFIED',
+    });
+    const retriedPromotion = await promote(
+      socket,
+      observerSessionId,
+      observerPending.receipt_challenge!,
+    );
+    expect(retriedPromotion.ok).toBe(false);
+    expect(retriedPromotion.code).toBe('RECEIPT_REPLAY');
+
+    const revoked = spawnSync(
+      'python3',
+      [revokerPath, '--runtime', 'claude', '--reason', 'session-start-compact'],
+      {
+        encoding: 'utf8',
+        env: {
+          ...process.env,
+          MOSAIC_LEASE_BROKER_SOCKET: socket,
+          MOSAIC_LEASE_SESSION_ID: observerSessionId,
+          MOSAIC_RUNTIME_GENERATION: '1',
+        },
+      },
+    );
+    expect(revoked.status, revoked.stderr).toBe(0);
+    expect(await authorize(socket, observerSessionId, 'claude', 'Write')).toMatchObject({
+      ok: false,
+      code: 'MUTATOR_UNVERIFIED',
+      decision: 'deny',
+    });
+
+    const { socket: expirySocket } = await startBroker();
+    const expirySessionId = await register(expirySocket);
+    expect(expirySessionId).not.toBe(observerSessionId);
+    const expiryPending = await beginVerification(expirySocket, expirySessionId, 'claude', 1, 1);
+    await promote(expirySocket, expirySessionId, expiryPending.receipt_challenge!);
+    expect(await authorize(expirySocket, expirySessionId, 'claude', 'Bash')).toMatchObject({
+      ok: true,
+      decision: 'allow',
+      state: 'VERIFIED',
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 1_100));
+    expect(await authorize(expirySocket, expirySessionId, 'claude', 'Bash')).toMatchObject({
+      ok: false,
+      code: 'LEASE_EXPIRED',
+      decision: 'deny',
+    });
+  });
+
+  test('a fired observer fences the old lease even while broker transport is unavailable', async () => {
+    const { socket } = await startBroker();
+    const sessionId = await register(socket);
+    const pending = await beginVerification(socket, sessionId, 'claude');
+    await promote(socket, sessionId, pending.receipt_challenge!);
+
+    const root = await mkdtemp(join(tmpdir(), 'mosaic-observer-fence-'));
+    temporaryRoots.push(root);
+    const generationFile = join(root, 'runtime.generation');
+    await writeFile(generationFile, '1\n', { mode: 0o600 });
+    const failedObserver = spawnSync(
+      'python3',
+      [revokerPath, '--runtime', 'claude', '--reason', 'session-start-compact'],
+      {
+        encoding: 'utf8',
+        env: {
+          ...process.env,
+          MOSAIC_LEASE_BROKER_SOCKET: join(root, 'unavailable.sock'),
+          MOSAIC_LEASE_SESSION_ID: sessionId,
+          MOSAIC_RUNTIME_GENERATION: '1',
+          MOSAIC_LEASE_GENERATION_FILE: generationFile,
+        },
+      },
+    );
+    expect(failedObserver.status).toBe(2);
+    expect(await readFile(generationFile, 'utf8')).toBe('2\n');
+    expect(await authorize(socket, sessionId, 'claude', 'Write', 2)).toMatchObject({
+      ok: false,
+      code: 'MUTATOR_UNVERIFIED',
+      decision: 'deny',
+    });
+  });
+
+  test('same-PID runtime-generation bump revokes the prior incarnation automatically', async () => {
+    const { socket } = await startBroker();
+    const sessionId = await register(socket);
+    const pending = await beginVerification(socket, sessionId, 'pi');
+    await promote(socket, sessionId, pending.receipt_challenge!);
+    const anchorPid = process.pid;
+
+    const root = await mkdtemp(join(tmpdir(), 'mosaic-generation-bump-'));
+    temporaryRoots.push(root);
+    const generationFile = join(root, 'runtime.generation');
+    await writeFile(generationFile, '1\n', { mode: 0o600 });
+    const bumped = spawnSync(
+      'python3',
+      [revokerPath, '--runtime', 'pi', '--reason', 'session-start-resume', '--bump-generation'],
+      {
+        encoding: 'utf8',
+        env: {
+          ...process.env,
+          MOSAIC_LEASE_BROKER_SOCKET: socket,
+          MOSAIC_LEASE_SESSION_ID: sessionId,
+          MOSAIC_RUNTIME_GENERATION: '1',
+          MOSAIC_LEASE_GENERATION_FILE: generationFile,
+        },
+      },
+    );
+    expect(bumped.status, bumped.stderr).toBe(0);
+    expect(await readFile(generationFile, 'utf8')).toBe('2\n');
+    expect(process.pid).toBe(anchorPid);
+    expect(await authorize(socket, sessionId, 'pi', 'bash', 2)).toMatchObject({
+      ok: false,
+      code: 'MUTATOR_UNVERIFIED',
+      decision: 'deny',
+    });
+    expect(await authorize(socket, sessionId, 'pi', 'bash', 1)).toMatchObject({
+      ok: false,
+      code: 'STALE_GENERATION',
+    });
+  });
+
+  test('Claude and Pi compaction observer wiring is complete and fail-closed', async () => {
+    const settings = JSON.parse(await readFile(claudeSettingsPath, 'utf8')) as {
+      hooks: Record<string, Array<{ matcher?: string; hooks: Array<{ command: string }> }>>;
+    };
+    expect(
+      settings.hooks['PreCompact']?.some((entry) =>
+        entry.hooks.some((hook) => hook.command.includes('revoke-lease.py')),
+      ),
+    ).toBe(true);
+    expect(
+      settings.hooks['SessionStart']?.some(
+        (entry) =>
+          entry.matcher === 'compact' &&
+          entry.hooks.some((hook) => hook.command.includes('revoke-lease.py')),
+      ),
+    ).toBe(true);
+
+    const piExtension = await readFile(piExtensionPath, 'utf8');
+    const piLifecycle = await readFile(piLifecyclePath, 'utf8');
+    expect(piExtension).toContain('registerLeaseLifecycleHooks');
+    expect(piLifecycle).toContain("pi.on('session_before_compact'");
+    expect(piLifecycle).toContain("pi.on('session_compact'");
+    expect(piLifecycle).toContain("pi.on('context'");
+    expect(piLifecycle).toContain('--bump-generation');
+  });
+
   test('observer revocation and monotonic TTL expiry deny the next mutator', async () => {
     const { socket } = await startBroker();
     const sessionId = await register(socket);
     const pending = await beginVerification(socket, sessionId, 'claude', 1, 1);
-    await promote(socket, sessionId, pending.promotion_token!);
+    await promote(socket, sessionId, pending.receipt_challenge!);
 
     expect(await authorize(socket, sessionId, 'claude', 'Bash')).toMatchObject({
       ok: true,
@@ -411,7 +676,7 @@ describe('whole mutator-class lease gate', () => {
     });
 
     const refreshed = await beginVerification(socket, sessionId, 'claude', 1, 300, 2);
-    await promote(socket, sessionId, refreshed.promotion_token!);
+    await promote(socket, sessionId, refreshed.receipt_challenge!);
     expect(
       await request(socket, {
         action: 'revoke_lease',
@@ -431,7 +696,7 @@ describe('whole mutator-class lease gate', () => {
     const { socket } = await startBroker();
     const sessionId = await register(socket);
     const pending = await beginVerification(socket, sessionId, 'pi');
-    await promote(socket, sessionId, pending.promotion_token!);
+    await promote(socket, sessionId, pending.receipt_challenge!);
 
     expect(await authorize(socket, sessionId, 'pi', 'bash', 2)).toMatchObject({
       ok: false,
@@ -542,6 +807,12 @@ hook_present = any(
     item.get("matcher") == ".*" and any("mutator-gate.py" in hook.get("command", "") for hook in item.get("hooks", []))
     for item in pre_tool
 )
+pre_compact = settings.get("hooks", {}).get("PreCompact", [])
+session_start = settings.get("hooks", {}).get("SessionStart", [])
+observers_present = (
+    any(any("revoke-lease.py" in hook.get("command", "") for hook in item.get("hooks", [])) for item in pre_compact)
+    and any(item.get("matcher") == "compact" and any("revoke-lease.py" in hook.get("command", "") for hook in item.get("hooks", [])) for item in session_start)
+)
 denied = subprocess.run(
     ["python3", ${JSON.stringify(gatePath)}, "--runtime", "claude"],
     input=json.dumps({"tool_name": "Bash"}) + "\\n",
@@ -553,11 +824,12 @@ is_yolo = "--dangerously-skip-permissions" in sys.argv[1:]
 result = {
     "session_id": session_id,
     "hook_present": hook_present,
+    "observers_present": observers_present,
     "denied": denied,
     "is_yolo": is_yolo,
 }
 print(json.dumps(result))
-raise SystemExit(0 if len(session_id) == 64 and hook_present and denied else 1)
+raise SystemExit(0 if len(session_id) == 64 and hook_present and observers_present and denied else 1)
 `;
       await writeFile(fakeClaude, probe, { mode: 0o700 });
       await chmod(fakeClaude, 0o700);
@@ -621,6 +893,7 @@ raise SystemExit(0 if len(session_id) == 64 and hook_present and denied else 1)
       expect(JSON.parse(String(execution!.stdout))).toEqual({
         session_id: expect.stringMatching(/^[a-f0-9]{64}$/),
         hook_present: true,
+        observers_present: true,
         denied: true,
         is_yolo: yolo,
       });
@@ -636,7 +909,7 @@ raise SystemExit(0 if len(session_id) == 64 and hook_present and denied else 1)
     expect(runRuntimeGate(socket, sessionId, 'pi', 'unknown_custom_tool').status).toBe(2);
 
     const pending = await beginVerification(socket, sessionId, 'claude');
-    await promote(socket, sessionId, pending.promotion_token!);
+    await promote(socket, sessionId, pending.receipt_challenge!);
     expect(runRuntimeGate(socket, sessionId, 'claude', 'Bash').status).toBe(0);
 
     const settings = JSON.parse(await readFile(claudeSettingsPath, 'utf8')) as {

packages/mosaic/src/mutator-gate/pi-compaction-lifecycle.spec.ts

@@ -0,0 +1,127 @@
+import { describe, expect, test } from 'vitest';
+
+type LifecycleRunner = (args: string[]) => boolean;
+type LifecycleRegister = (api: unknown, runner: LifecycleRunner) => void;
+type Handler = (event: Record<string, unknown>, ctx: Record<string, unknown>) => unknown;
+
+const lifecycleModuleUrl = new URL('../../framework/runtime/pi/lease-lifecycle.ts', import.meta.url)
+  .href;
+const { registerLeaseLifecycleHooks } = (await import(lifecycleModuleUrl)) as {
+  registerLeaseLifecycleHooks: LifecycleRegister;
+};
+
+function fakePi() {
+  const handlers = new Map<string, Handler[]>();
+  return {
+    handlers,
+    api: {
+      on(event: string, handler: Handler) {
+        handlers.set(event, [...(handlers.get(event) ?? []), handler]);
+      },
+    },
+    async emit(event: string, value: Record<string, unknown> = {}) {
+      const results = [];
+      for (const handler of handlers.get(event) ?? []) {
+        results.push(await handler(value, {}));
+      }
+      return results;
+    },
+  };
+}
+
+describe('Pi compaction and runtime-generation lease lifecycle', () => {
+  test('pre-compaction and first post-compaction context independently revoke', async () => {
+    const pi = fakePi();
+    const calls: string[][] = [];
+    registerLeaseLifecycleHooks(pi.api as never, (args) => {
+      calls.push(args);
+      return true;
+    });
+
+    expect(await pi.emit('session_before_compact', { reason: 'threshold' })).toEqual([undefined]);
+    expect(calls.at(-1)).toEqual([
+      '--runtime',
+      'pi',
+      '--reason',
+      'pi-session-before-compact:threshold',
+    ]);
+
+    await pi.emit('session_compact', { reason: 'threshold' });
+    await pi.emit('context', { messages: [] });
+    expect(calls.at(-1)).toEqual([
+      '--runtime',
+      'pi',
+      '--reason',
+      'pi-context-after-compact:threshold',
+    ]);
+    const afterFirstContext = calls.length;
+    await pi.emit('context', { messages: [] });
+    expect(calls).toHaveLength(afterFirstContext);
+  });
+
+  test.each(['reload', 'new', 'resume', 'fork'])(
+    '%s bumps generation before reuse',
+    async (reason) => {
+      const pi = fakePi();
+      const calls: string[][] = [];
+      registerLeaseLifecycleHooks(pi.api as never, (args) => {
+        calls.push(args);
+        return true;
+      });
+
+      await pi.emit('session_start', { reason });
+      expect(calls).toEqual([
+        ['--runtime', 'pi', '--reason', `pi-session-start:${reason}`, '--bump-generation'],
+      ]);
+    },
+  );
+
+  test('startup leaves the launcher generation intact and tools locally open', async () => {
+    const pi = fakePi();
+    const calls: string[][] = [];
+    registerLeaseLifecycleHooks(pi.api as never, (args) => {
+      calls.push(args);
+      return true;
+    });
+
+    await pi.emit('session_start', { reason: 'startup' });
+    await pi.emit('session_start');
+    expect(calls).toEqual([]);
+    expect(await pi.emit('tool_call', { toolName: 'bash' })).toEqual([undefined]);
+  });
+
+  test('failed post-compaction revoke blocks tools until context retries successfully', async () => {
+    const pi = fakePi();
+    const outcomes = [false, true];
+    registerLeaseLifecycleHooks(pi.api as never, () => outcomes.shift() ?? true);
+
+    await pi.emit('session_compact');
+    await pi.emit('context');
+    expect(await pi.emit('tool_call', { toolName: 'bash' })).toEqual([
+      {
+        block: true,
+        reason: expect.stringContaining('lease lifecycle revoke failed'),
+      },
+    ]);
+
+    await pi.emit('context');
+    expect(await pi.emit('tool_call', { toolName: 'bash' })).toEqual([undefined]);
+  });
+
+  test('failed lifecycle revoke cancels compaction and closes later tool calls', async () => {
+    const pi = fakePi();
+    registerLeaseLifecycleHooks(pi.api as never, () => false);
+
+    const compact = await pi.emit('session_before_compact', { reason: 'manual' });
+    expect(compact).toEqual([{ cancel: true }]);
+
+    await pi.emit('session_start', { reason: 'resume' });
+    const tool = await pi.emit('tool_call', { toolName: 'bash' });
+    expect(tool).toEqual([
+      {
+        block: true,
+        reason: expect.stringContaining('lease lifecycle revoke failed'),
+      },
+    ]);
+  });
+});

packages/mosaic/src/mutator-gate/runtime_tools_unittest.py

@@ -9,6 +9,7 @@ import json
 import os
 import runpy
 import socket
+import stat
 import subprocess
 import sys
 import tempfile
@@ -16,10 +17,13 @@ import threading
 import unittest
 from contextlib import redirect_stderr
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import patch
 
 
 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):
@@ -78,6 +82,9 @@ class LaunchRuntimeTest(unittest.TestCase):
         def execute(command: str, argv: list[str], environment: dict[str, str]) -> None:
             calls["execute"] = (command, argv, environment)
 
+        def initialize_generation(path: Path, generation: int) -> None:
+            calls["generation"] = (path, generation)
+
         result = LAUNCHER.main(
             ["--runtime", "claude", "--", "claude", "--print", "hello"],
             environ={
@@ -87,6 +94,7 @@ class LaunchRuntimeTest(unittest.TestCase):
             },
             request=request,
             execute=execute,
+            initialize_generation=initialize_generation,
         )
 
         self.assertEqual(result, 0)
@@ -101,6 +109,14 @@ class LaunchRuntimeTest(unittest.TestCase):
         self.assertEqual(environment["MOSAIC_LEASE_SESSION_ID"], session_id)
         self.assertEqual(environment["MOSAIC_RUNTIME_GENERATION"], "7")
         self.assertEqual(environment["MOSAIC_LEASE_RUNTIME"], "claude")
+        self.assertEqual(
+            environment["MOSAIC_LEASE_GENERATION_FILE"],
+            f"/run/test/generation-{session_id}.state",
+        )
+        self.assertEqual(
+            calls["generation"],
+            (Path(f"/run/test/generation-{session_id}.state"), 7),
+        )
         self.assertEqual(environment["PRESERVED"], "yes")
 
     def test_dangerous_claude_mode_is_owned_and_injected_by_the_wrapper(self) -> None:
@@ -110,6 +126,7 @@ class LaunchRuntimeTest(unittest.TestCase):
             environ={"MOSAIC_LEASE_BROKER_SOCKET": "/broker"},
             request=lambda *_args: {"ok": True, "session_id": "e" * 64},
             execute=lambda *args: executed.append(args),
+            initialize_generation=lambda *_args: None,
         )
         self.assertEqual(result, 0)
         self.assertEqual(
@@ -135,6 +152,7 @@ class LaunchRuntimeTest(unittest.TestCase):
             environ={"MOSAIC_LEASE_BROKER_SOCKET": "/broker"},
             request=lambda *_args: {"ok": True, "session_id": "f" * 64},
             execute=lambda *args: executed.append(args),
+            initialize_generation=lambda *_args: None,
         )
         self.assertEqual(result, 0)
         self.assertEqual(executed[0][0:2], ("pi", ["pi", "--print", "hello"]))
@@ -174,6 +192,21 @@ class LaunchRuntimeTest(unittest.TestCase):
                 self.assertEqual(result, 1)
                 self.assertEqual(executed, [])
 
+    def test_generation_initialization_failure_denies_before_exec(self) -> None:
+        with redirect_stderr(io.StringIO()):
+            self.assertEqual(
+                LAUNCHER.main(
+                    ["--runtime", "pi", "--", "pi"],
+                    environ={"MOSAIC_LEASE_BROKER_SOCKET": "/broker"},
+                    request=lambda *_args: {"ok": True, "session_id": "a" * 64},
+                    execute=lambda *_args: self.fail("must not execute"),
+                    initialize_generation=lambda *_args: (_ for _ in ()).throw(
+                        OSError("unsafe state")
+                    ),
+                ),
+                1,
+            )
+
     def test_registration_exceptions_fail_closed(self) -> None:
         failures = [ValueError("bad"), OSError("down"), json.JSONDecodeError("bad", "x", 0)]
         for failure in failures:
@@ -199,6 +232,7 @@ class LaunchRuntimeTest(unittest.TestCase):
                     environ={"MOSAIC_LEASE_BROKER_SOCKET": "/x"},
                     request=lambda *_args: {"ok": True, "session_id": "c" * 64},
                     execute=lambda *_args: (_ for _ in ()).throw(OSError("missing")),
+                    initialize_generation=lambda *_args: None,
                 ),
                 1,
             )
@@ -284,6 +318,23 @@ class ExecutableEntrypointTest(unittest.TestCase):
             runpy.run_path(str(TOOLS_DIR / "launch-runtime.py"), run_name="__main__")
         self.assertEqual(raised.exception.code, 64)
 
+    def test_revoker_entrypoint_denies_when_identity_environment_is_absent(self) -> None:
+        with patch.object(
+            sys,
+            "argv",
+            [
+                str(TOOLS_DIR / "revoke-lease.py"),
+                "--runtime",
+                "claude",
+                "--reason",
+                "pre-compact",
+            ],
+        ), patch.dict(os.environ, {}, clear=True), redirect_stderr(
+            io.StringIO()
+        ), self.assertRaises(SystemExit) as raised:
+            runpy.run_path(str(TOOLS_DIR / "revoke-lease.py"), run_name="__main__")
+        self.assertEqual(raised.exception.code, 2)
+
     def test_gate_entrypoint_denies_when_identity_environment_is_absent(self) -> None:
         class Stdin:
             buffer = io.BytesIO(b'{"tool_name":"Bash"}')
@@ -325,6 +376,19 @@ class MutatorGateTest(unittest.TestCase):
             )
         return result, stderr.getvalue(), calls
 
+    def test_generation_file_is_the_effective_generation_authority(self) -> None:
+        calls: list[dict[str, object]] = []
+        result = GATE.main(
+            ["--runtime", "pi"],
+            environ=self.environment(),
+            stream=io.BytesIO(b'{"tool_name":"bash"}'),
+            request=lambda _path, payload: calls.append(payload)
+            or {"ok": True, "decision": "allow"},
+            resolve_generation=lambda _environment: 9,
+        )
+        self.assertEqual(result, 0)
+        self.assertEqual(calls[0]["runtime_generation"], 9)
+
     def test_allow_and_denial_decisions(self) -> None:
         allowed, allowed_stderr, calls = self.run_main()
         self.assertEqual(allowed, 0)
@@ -432,5 +496,252 @@ class MutatorGateTest(unittest.TestCase):
                         self.assertEqual(fake.shutdown_how, socket.SHUT_WR)
 
 
+class LeaseRevocationTest(unittest.TestCase):
+    def load_modules(self):
+        generation_path = TOOLS_DIR / "lease_generation.py"
+        revoker_path = TOOLS_DIR / "revoke-lease.py"
+        self.assertTrue(generation_path.is_file(), "lease_generation.py must be shipped")
+        self.assertTrue(revoker_path.is_file(), "revoke-lease.py must be shipped")
+        return (
+            load_tool("lease_generation_test", "lease_generation.py"),
+            load_tool("lease_revoker_test", "revoke-lease.py"),
+        )
+
+    def test_generation_parser_and_descriptor_security_rejections(self) -> None:
+        generation, _ = self.load_modules()
+        for value in (None, "", "-1", "no", "é", str(generation.MAX_GENERATION + 1)):
+            with self.subTest(value=value), self.assertRaises(ValueError):
+                generation.parse_generation(value)
+        for value in (-1, generation.MAX_GENERATION + 1):
+            with self.subTest(initialize=value), tempfile.TemporaryDirectory() as directory, self.assertRaises(ValueError):
+                generation.initialize_runtime_generation(Path(directory) / "state", value)
+
+        with patch.object(
+            generation.os,
+            "fstat",
+            return_value=SimpleNamespace(st_mode=stat.S_IFDIR | 0o700, st_uid=os.geteuid(), st_size=0),
+        ), self.assertRaises(ValueError):
+            generation._validate_descriptor(4)
+        with patch.object(
+            generation.os,
+            "fstat",
+            return_value=SimpleNamespace(st_mode=stat.S_IFREG | 0o600, st_uid=os.geteuid() + 1, st_size=0),
+        ), self.assertRaises(ValueError):
+            generation._validate_descriptor(4)
+        with patch.object(
+            generation.os,
+            "fstat",
+            return_value=SimpleNamespace(st_mode=stat.S_IFREG | 0o644, st_uid=os.geteuid(), st_size=0),
+        ), self.assertRaises(ValueError):
+            generation._validate_descriptor(4)
+        with patch.object(
+            generation.os,
+            "fstat",
+            return_value=SimpleNamespace(
+                st_mode=stat.S_IFREG | 0o600,
+                st_uid=os.geteuid(),
+                st_size=generation.MAX_GENERATION_BYTES + 1,
+            ),
+        ), self.assertRaises(ValueError):
+            generation._validate_descriptor(4)
+
+    def test_generation_file_is_private_monotonic_and_rejects_unsafe_state(self) -> None:
+        generation, _ = self.load_modules()
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "runtime.generation"
+            generation.initialize_runtime_generation(path, 4)
+            self.assertEqual(generation.read_runtime_generation({
+                "MOSAIC_RUNTIME_GENERATION": "1",
+                "MOSAIC_LEASE_GENERATION_FILE": str(path),
+            }), 4)
+            self.assertEqual(generation.bump_runtime_generation({
+                "MOSAIC_RUNTIME_GENERATION": "1",
+                "MOSAIC_LEASE_GENERATION_FILE": str(path),
+            }), 5)
+            self.assertEqual(path.read_text(), "5\n")
+            self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
+
+            path.write_text("bad\n")
+            with self.assertRaises(ValueError):
+                generation.read_runtime_generation({
+                    "MOSAIC_RUNTIME_GENERATION": "1",
+                    "MOSAIC_LEASE_GENERATION_FILE": str(path),
+                })
+            path.write_bytes(b"\xff\n")
+            with self.assertRaises(ValueError):
+                generation.read_runtime_generation({
+                    "MOSAIC_LEASE_GENERATION_FILE": str(path),
+                })
+            path.write_text(f"{generation.MAX_GENERATION}\n")
+            with self.assertRaises(ValueError):
+                generation.bump_runtime_generation({
+                    "MOSAIC_LEASE_GENERATION_FILE": str(path),
+                })
+            with self.assertRaises(ValueError):
+                generation.bump_runtime_generation({"MOSAIC_RUNTIME_GENERATION": "1"})
+
+    def test_generation_write_must_make_progress(self) -> None:
+        generation, _ = self.load_modules()
+        with tempfile.TemporaryDirectory() as directory, patch.object(
+            generation.os, "write", return_value=0
+        ), self.assertRaises(OSError):
+            generation.initialize_runtime_generation(Path(directory) / "state", 1)
+
+    def test_revoker_reuses_broker_revoke_and_optional_generation_bump(self) -> None:
+        _, revoker = self.load_modules()
+        with tempfile.TemporaryDirectory() as directory:
+            generation_file = Path(directory) / "runtime.generation"
+            generation_file.write_text("2\n")
+            generation_file.chmod(0o600)
+            environment = {
+                "MOSAIC_LEASE_BROKER_SOCKET": "/broker",
+                "MOSAIC_LEASE_SESSION_ID": "a" * 64,
+                "MOSAIC_RUNTIME_GENERATION": "2",
+                "MOSAIC_LEASE_GENERATION_FILE": str(generation_file),
+            }
+            calls: list[tuple[Path, dict[str, object]]] = []
+            result = revoker.main(
+                ["--runtime", "pi", "--reason", "session-start-resume", "--bump-generation"],
+                environ=environment,
+                request=lambda path, payload: calls.append((path, payload))
+                or {"ok": True, "state": "UNVERIFIED"},
+            )
+            self.assertEqual(result, 0)
+            self.assertEqual(generation_file.read_text(), "3\n")
+            self.assertEqual(calls[0][0], Path("/broker"))
+            self.assertEqual(calls[0][1], {
+                "action": "revoke_lease",
+                "session_id": "a" * 64,
+                "runtime_generation": 3,
+                "reason": "session-start-resume",
+                "runtime": "pi",
+            })
+
+            calls.clear()
+            self.assertEqual(
+                revoker.main(
+                    ["--runtime", "pi", "--reason", "pi-context-after-compact"],
+                    environ=environment,
+                    request=lambda path, payload: calls.append((path, payload))
+                    or {"ok": True, "state": "UNVERIFIED"},
+                ),
+                0,
+            )
+            self.assertEqual(calls[0][1]["runtime_generation"], 3)
+
+    def test_revoker_broker_framing_and_shape_validation(self) -> None:
+        _, revoker = self.load_modules()
+        with self.assertRaises(ValueError):
+            revoker.broker_request(Path("/broker"), {"reason": "x" * revoker.MAX_FRAME})
+        replies = [
+            (b'{"ok":true,"state":"UNVERIFIED"}\n', {"ok": True, "state": "UNVERIFIED"}),
+            (b'{"ok":true}', ValueError),
+            (b'[]\n', ValueError),
+            (b'x' * (revoker.MAX_FRAME + 1), ValueError),
+        ]
+        for wire_reply, expected in replies:
+            with self.subTest(size=len(wire_reply)):
+                fake = FakeSocket(wire_reply)
+                with patch.object(revoker.socket, "socket", return_value=fake):
+                    if isinstance(expected, type) and issubclass(expected, Exception):
+                        with self.assertRaises(expected):
+                            revoker.broker_request(Path("/broker"), {"action": "revoke_lease"})
+                    else:
+                        self.assertEqual(
+                            revoker.broker_request(Path("/broker"), {"action": "revoke_lease"}),
+                            expected,
+                        )
+                        self.assertEqual(fake.timeout, revoker.BROKER_TIMEOUT_SECONDS)
+                        self.assertEqual(fake.connected, "/broker")
+                        self.assertEqual(fake.shutdown_how, socket.SHUT_WR)
+
+    def test_failed_observer_revocation_advances_the_local_generation_fence(self) -> None:
+        _, revoker = self.load_modules()
+        with tempfile.TemporaryDirectory() as directory:
+            generation_file = Path(directory) / "runtime.generation"
+            generation_file.write_text("8\n")
+            generation_file.chmod(0o600)
+            environment = {
+                "MOSAIC_LEASE_BROKER_SOCKET": "/broker",
+                "MOSAIC_LEASE_SESSION_ID": "a" * 64,
+                "MOSAIC_RUNTIME_GENERATION": "8",
+                "MOSAIC_LEASE_GENERATION_FILE": str(generation_file),
+            }
+            with redirect_stderr(io.StringIO()):
+                result = revoker.main(
+                    ["--runtime", "claude", "--reason", "session-start-compact"],
+                    environ=environment,
+                    request=lambda *_args: (_ for _ in ()).throw(OSError("down")),
+                )
+            self.assertEqual(result, 2)
+            self.assertEqual(generation_file.read_text(), "9\n")
+
+    def test_revoker_fails_closed_on_identity_reply_and_transport_errors(self) -> None:
+        _, revoker = self.load_modules()
+        good = {
+            "MOSAIC_LEASE_BROKER_SOCKET": "/broker",
+            "MOSAIC_LEASE_SESSION_ID": "a" * 64,
+            "MOSAIC_RUNTIME_GENERATION": "1",
+        }
+        malformed_session = {**good, "MOSAIC_LEASE_SESSION_ID": "not-a-session"}
+        cases = [
+            ({}, lambda *_args: {"ok": True, "state": "UNVERIFIED"}),
+            (malformed_session, lambda *_args: {"ok": True, "state": "UNVERIFIED"}),
+            (good, lambda *_args: {"ok": False, "state": "UNVERIFIED"}),
+            (good, lambda *_args: {"ok": True, "state": "VERIFIED"}),
+            (good, lambda *_args: (_ for _ in ()).throw(OSError("down"))),
+            (
+                good,
+                lambda *_args: (_ for _ in ()).throw(
+                    json.JSONDecodeError("bad", "x", 0)
+                ),
+            ),
+        ]
+        for environment, request in cases:
+            with self.subTest(environment=environment), redirect_stderr(io.StringIO()):
+                self.assertEqual(
+                    revoker.main(
+                        ["--runtime", "claude", "--reason", "pre-compact"],
+                        environ=environment,
+                        request=request,
+                    ),
+                    2,
+                )
+
+        with redirect_stderr(io.StringIO()):
+            self.assertEqual(
+                revoker.main(
+                    ["--runtime", "claude", "--reason", "x" * 129],
+                    environ=good,
+                    request=lambda *_args: self.fail("invalid reason reached broker"),
+                ),
+                2,
+            )
+
+        with tempfile.TemporaryDirectory() as directory:
+            generation_file = Path(directory) / "runtime.generation"
+            generation_file.write_text("1\n")
+            generation_file.chmod(0o600)
+            with redirect_stderr(io.StringIO()):
+                self.assertEqual(
+                    revoker.main(
+                        [
+                            "--runtime",
+                            "pi",
+                            "--reason",
+                            "session-start-resume",
+                            "--bump-generation",
+                        ],
+                        environ={
+                            **good,
+                            "MOSAIC_LEASE_GENERATION_FILE": str(generation_file),
+                        },
+                        request=lambda *_args: (_ for _ in ()).throw(OSError("down")),
+                    ),
+                    2,
+                )
+            self.assertEqual(generation_file.read_text(), "2\n")
+
+
 if __name__ == "__main__":
     unittest.main()