diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml
index 4e6dd61..88852e8 100644
--- a/.github/workflows/validate.yml
+++ b/.github/workflows/validate.yml
@@ -60,6 +60,18 @@ jobs:
           python3 scripts/plan_repository_presentation.py verify-plan \
             --plan "${plan_path}"
 
+      - name: Prove reviewed general fleet convergence planning
+        shell: bash
+        run: |
+          plan_path="${RUNNER_TEMP}/fleet-convergence.plan.json"
+          python3 scripts/plan_fleet_convergence.py plan \
+            --manifest "examples/convergence/fleet.manifest.json" \
+            --catalog "examples/convergence/catalog.json" \
+            --observatory "examples/convergence/observatory.snapshot.json" \
+            --output "${plan_path}"
+          python3 scripts/plan_fleet_convergence.py verify-plan \
+            --plan "${plan_path}"
+
       - name: Run adversarial validator tests
         run: >-
           python3 -m unittest discover
@@ -76,4 +88,4 @@ jobs:
           set -euo pipefail
           while IFS= read -r document; do
             python3 -m json.tool "${document}" >/dev/null
-          done < <(find examples schemas -type f -name "*.json" -print | sort)
+          done < <(find contracts examples schemas -type f -name "*.json" -print | sort)
diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md
index adc88b7..3c08fa2 100644
--- a/ARCHITECTURE.md
+++ b/ARCHITECTURE.md
@@ -8,7 +8,7 @@ status: provisional
 owners:
   - egohygiene
 created: 2026-08-19
-updated: 2026-08-31
+updated: 2026-09-02
 governed_by:
   - architecture-architecture
 depends_on:
@@ -92,6 +92,29 @@ Egolint report, reviewed plan, immutable contract set, and rollback snapshot.
 An external authorized operator may consume that artifact only after the review
 boundary; application remains outside this slice.
 
+## Implemented fleet convergence slice
+
+```text
+Hygiene catalog + Holon manifest + Pace current/desired locks
+                         +
+          Observatory organization-health snapshot
+                         |
+                         v
+ deterministic drift -> ordered repository units -> exact review
+                         |
+                         v
+        bounded candidate verification -> one upgrade PR
+```
+
+`scripts/plan_fleet_convergence.py` activates the general inventory reader,
+desired-state resolver, drift engine, reconciliation planner, and a deliberately
+narrow GitHub adapter. Observatory supplies observation; Pace does not scan.
+Holon supplies exact generated-file plans and rendering; Pace verifies the
+resulting candidate rather than reproducing the materializer. The adapter can
+create one non-default branch and pull request only when the remote base still
+equals the represented commit and the local diff exactly equals the reviewed
+allowlist. Merge remains outside the adapter.
+
 ## Dependency rules
 
 - Sibling domain capabilities integrate through versioned public contracts, not direct access to internals.
@@ -115,11 +138,11 @@ The architecture favors independently usable local and self-hosted operation. Op
 
 ## Evidence and uncertainty
 
-- **Observed:** Pace owns a versioned dependency-lock validator plus a bounded
-  repository-presentation inventory, deterministic drift/plan projection,
-  exact review authorization, credential-free single-repository proposal,
-  adversarial tests, and least-privilege CI gate. General update resolution and
-  every consumer write/application path remain unimplemented.
+- **Observed:** Pace owns a versioned dependency-lock validator, a general
+  deterministic convergence planner, exact partial-fleet review records,
+  bounded candidate verification, an idempotent one-PR GitHub adapter, and the
+  specialized repository-presentation rollout planner. It consumes pinned
+  sibling contracts and does not merge or mutate a default branch.
 - **Decided for this draft:** The repository owns the bounded concern described here and participates through versioned contracts.
 - **Proposed:** Target systems and later roadmap phases remain proposals until accepted and implemented.
 - **Open question:** Which parts of this draft should become active in the first independently versioned release?
diff --git a/README.md b/README.md
index a23b1e1..1870b73 100644
--- a/README.md
+++ b/README.md
@@ -29,6 +29,28 @@ See the complete [lock and update policy](LOCK_POLICY.md), the
 [JSON Schema](schemas/pace-lock-v1.schema.json), and the
 [six-kind example](examples/pace.lock.json).
 
+## Plan reviewed fleet convergence
+
+The general convergence path consumes Hygiene catalog membership, Holon
+foundation intent, current and desired Pace locks, and Observatory's current
+represented state. It produces deterministic dependency-ordered drift before
+any write and requires an exact review for each repository-sized upgrade unit.
+
+```bash
+python3 scripts/plan_fleet_convergence.py plan \
+  --manifest "examples/convergence/fleet.manifest.json" \
+  --catalog "examples/convergence/catalog.json" \
+  --observatory "examples/convergence/observatory.snapshot.json" \
+  --output "/tmp/fleet-convergence.plan.json"
+
+python3 scripts/plan_fleet_convergence.py verify-plan \
+  --plan "/tmp/fleet-convergence.plan.json"
+```
+
+After review, `propose` prepares one credential-free PR request and `open-pr`
+accepts only an exact candidate tree on the still-current represented commit.
+See the [fleet convergence contract and operator guide](docs/fleet-convergence.md).
+
 ## Plan repository-presentation rollout
 
 Pace now owns a read-only, privacy-safe repository-presentation inventory and a
@@ -51,8 +73,8 @@ See the [rollout contract and operating guide](docs/repository-presentation-roll
 
 ## Current authority boundary
 
-Pace validates desired locks and can now turn a reviewed repository-presentation
-inventory into a deterministic no-write fleet plan. It may emit a credential-free
-single-repository proposal only after exact plan review and valid pinned Egolint
-evidence. It still does not edit consumer files, open pull requests, or apply
-changes; those remain separately authorized authority boundaries.
+Pace validates desired locks and can turn pinned catalog, Holon, lock, and
+Observatory inputs into deterministic no-write fleet plans. After an exact
+human review, it can verify a locally materialized candidate and open one
+bounded consumer pull request. It never scans repositories, renders Holon
+outputs, updates a default branch, or merges a pull request.
diff --git a/ROADMAP.md b/ROADMAP.md
index efb3fa5..12e6b2d 100644
--- a/ROADMAP.md
+++ b/ROADMAP.md
@@ -8,7 +8,7 @@ status: provisional
 owners:
   - egohygiene
 created: 2026-08-19
-updated: 2026-08-31
+updated: 2026-09-02
 governed_by:
   - architecture-roadmap
 depends_on:
@@ -40,7 +40,8 @@ updated: 2026-08-24
 > This evidence-reconciled snapshot is the issue-generation and visual-roadmap handoff. The longer-horizon strategy below remains canonical context; generated HTML, JSON, progress, issue plans, and commit lists are projections.
 
 **Lifecycle:** seed implementation  
-**Current gate:** Review issue #14's no-write repository-presentation fleet plan, then unblock and independently approve its three canaries before any consumer PR.  
+**Current gate:** Review PAC-02's general convergence contracts and bounded PR adapter, while issue #14 separately unblocks its three presentation canaries.
+
 **North-star outcome:** Reviewable fleet convergence from declared desired state, observed state, and explicit drift.
 
 ### Visual roadmap publication
@@ -78,72 +79,75 @@ issues: []
 
 <!-- roadmap-step
 id: PAC-Q02
-status: active
+status: complete
 depends_on: [PAC-Q01]
 issues: []
 -->
 #### PAC-Q02 — Capture observed state
 
-**State:** `active`  
+**State:** `complete`
+
 **Depends on:** `PAC-Q01`
 
 **Outcome:** Pace can record what is actually deployed without mutating it.
 
 **Exit criteria:**
 
-- [ ] An observed-state schema and collector are implemented.
-- [ ] Fixtures prove stable normalization of at least one representative fleet.
+- [x] Pace consumes Observatory's pinned organization-health schema instead of adding a second collector.
+- [x] Fixtures prove stable ingestion of represented fleet state and block stale or ambiguous evidence.
 
 **Current evidence:**
 
-- A bounded, privacy-safe repository-presentation inventory now records all 29 repositories and explicit adoption states.
-- General-purpose observed-state collection remains unimplemented.
+- Observatory PR #15 provides the read-only organization-health contract.
+- Pace pins that contract and binds the exact catalog bytes, snapshot ID, freshness, and represented commit into every plan.
 
 <!-- roadmap-step
 id: PAC-Q03
-status: planned
+status: complete
 depends_on: [PAC-Q02]
 issues: []
 -->
 #### PAC-Q03 — Render actionable drift
 
-**State:** `planned`  
+**State:** `complete`
+
 **Depends on:** `PAC-Q02`
 
 **Outcome:** Desired and observed states produce a deterministic, human-reviewable drift report.
 
 **Exit criteria:**
 
-- [ ] Additions, removals, and changes are distinguished.
-- [ ] Repeated runs on unchanged input are identical.
+- [x] Additions, removals, and changes are distinguished.
+- [x] Repeated runs on unchanged input are identical.
 
 **Current evidence:**
 
-- Issue #14 implements deterministic repository-presentation drift categories and an Observatory-safe projection.
-- The general cross-capability drift engine remains planned.
+- PAC-02 compares complete current and desired Pace lock entries and explains categorical risk.
+- The canonical plan digest binds inputs, dependency order, drift, risk, blockers, and rollback state.
 
 <!-- roadmap-step
 id: PAC-Q04
-status: planned
+status: complete
 depends_on: [PAC-Q03]
 issues: [2]
 -->
 #### PAC-Q04 — Create a reviewable convergence plan
 
-**State:** `planned`  
+**State:** `complete`
+
 **Depends on:** `PAC-Q03`
 
 **Outcome:** Issue #2 yields an ordered plan with explicit risk, ownership, and rollback information.
 
 **Exit criteria:**
 
-- [ ] The plan is generated without applying changes.
-- [ ] Each action links to the drift that caused it.
+- [x] The plan is generated without applying changes.
+- [x] Each action links to the drift that caused it.
 
 **Current evidence:**
 
-- Issue #14 now proves an exact no-write plan, plan digest, review record, supersession boundary, and one-repository proposal contract for repository presentation.
-- Issue #2 remains the broader convergence-planning gate.
+- PAC-02 emits topologically ordered repository units with full before/after lock entries, risk reasons, exact review records, and partial-fleet selection.
+- One reviewed unit can verify one bounded candidate and open one non-default-branch PR; merge remains human-owned.
 
 <!-- roadmap-step
 id: PAC-Q05
@@ -206,8 +210,8 @@ This roadmap describes capability evolution, not promised dates or an issue queu
 
 ## Phase 1: Define desired and observed state schemas
 
-**Status:** Desired dependency lock v1 is implemented and independently
-validated. Observed-state and drift schemas remain planned for Phase 2.
+**Status:** Desired dependency lock v1 and pinned Observatory ingestion are
+implemented and independently validated.
 
 **Outcome:** A bounded capability advances from documented intent to validated, independently usable behavior.
 
@@ -221,6 +225,9 @@ validated. Observed-state and drift schemas remain planned for Phase 2.
 
 ## Phase 2: Implement read-only drift
 
+**Status:** Implemented for current and desired Pace locks with explainable risk
+and fail-closed Observatory freshness.
+
 **Outcome:** A bounded capability advances from documented intent to validated, independently usable behavior.
 
 **Exit signals:**
@@ -233,6 +240,9 @@ validated. Observed-state and drift schemas remain planned for Phase 2.
 
 ## Phase 3: Generate reviewable reconciliation plans
 
+**Status:** Implemented with deterministic dependency order, exact-digest
+reviews, blockers, pause state, and partial-fleet approval.
+
 **Outcome:** A bounded capability advances from documented intent to validated, independently usable behavior.
 
 **Exit signals:**
@@ -245,6 +255,9 @@ validated. Observed-state and drift schemas remain planned for Phase 2.
 
 ## Phase 4: Apply bounded changes
 
+**Status:** Implemented only for one exact candidate tree and one pull request.
+Default-branch writes, force, and merge remain unavailable.
+
 **Outcome:** A bounded capability advances from documented intent to validated, independently usable behavior.
 
 **Exit signals:**
@@ -282,9 +295,9 @@ Optional managed services, enterprise controls, marketplaces, and the conversati
 ## Evidence and uncertainty
 
 - **Observed:** Pace owns `egohygiene.pace.lock/v1`, an offline validator,
-  and a bounded repository-presentation inventory/planner with explicit review,
-  proposal, rollback, privacy, and Observatory contracts. No updater or
-  repository mutation path exists.
+  general fleet convergence contracts, exact review and rollback anchors, a
+  bounded one-PR adapter, and a specialized repository-presentation planner.
+  No default-branch or merge mutation path exists.
 - **Decided for this draft:** The repository owns the bounded concern described here and participates through versioned contracts.
 - **Proposed:** Target systems and later roadmap phases remain proposals until accepted and implemented.
 - **Open question:** Which parts of this draft should become active in the first independently versioned release?
diff --git a/SYSTEM.md b/SYSTEM.md
index cd91e28..989d980 100644
--- a/SYSTEM.md
+++ b/SYSTEM.md
@@ -8,7 +8,7 @@ status: provisional
 owners:
   - egohygiene
 created: 2026-08-19
-updated: 2026-08-21
+updated: 2026-09-02
 governed_by:
   - architecture-system
 depends_on:
@@ -34,13 +34,13 @@ This document identifies Pace's logical systems and responsibilities. It answers
 | --- | --- | --- |
 | Desired-state lock contract | Active | Records immutable source identity, content digest, target ownership, compatibility, rollback, and bounded exceptions for six dependency kinds. |
 | Independent lock validator | Active | Validates lock structure and semantics offline without trusting an updater or receiving write authority. |
-| Repository inventory reader | Target | Owns its bounded portion of the repository adoption, reconciliation, synchronization, and conformance mechanism for the Ego Hygiene organization; exposes explicit inputs, outputs, failure states, and evidence. |
-| Desired-state resolver | Target | Owns its bounded portion of the repository adoption, reconciliation, synchronization, and conformance mechanism for the Ego Hygiene organization; exposes explicit inputs, outputs, failure states, and evidence. |
-| Drift engine | Target | Owns its bounded portion of the repository adoption, reconciliation, synchronization, and conformance mechanism for the Ego Hygiene organization; exposes explicit inputs, outputs, failure states, and evidence. |
-| Reconciliation planner | Target | Owns its bounded portion of the repository adoption, reconciliation, synchronization, and conformance mechanism for the Ego Hygiene organization; exposes explicit inputs, outputs, failure states, and evidence. |
-| Change applier | Target | Owns its bounded portion of the repository adoption, reconciliation, synchronization, and conformance mechanism for the Ego Hygiene organization; exposes explicit inputs, outputs, failure states, and evidence. |
+| Repository inventory reader | Active | Joins Hygiene membership to pinned Holon, Pace lock, and Observatory inputs without scanning repositories. |
+| Desired-state resolver | Active | Validates desired locks and binds repository intent to the matching foundation manifest. |
+| Drift engine | Active | Distinguishes lock additions, removals, and updates with deterministic risk reasons. |
+| Reconciliation planner | Active | Emits topologically ordered repository units with blockers, rollback anchors, source digests, and an exact review gate. |
+| Change applier | Active, bounded | Verifies one local candidate against one reviewed unit and opens one non-default-branch pull request; never merges. |
 | Override registry | Target | Owns its bounded portion of the repository adoption, reconciliation, synchronization, and conformance mechanism for the Ego Hygiene organization; exposes explicit inputs, outputs, failure states, and evidence. |
-| Reporting adapter | Target | Owns its bounded portion of the repository adoption, reconciliation, synchronization, and conformance mechanism for the Ego Hygiene organization; exposes explicit inputs, outputs, failure states, and evidence. |
+| Reporting adapter | Active | Publishes machine-readable fleet plans, PR proposals, and PR results with explicit coverage and failure state. |
 
 ## External systems
 
@@ -63,9 +63,10 @@ Systems fail closed at destructive, publication, privacy, and security boundarie
 
 ## Evidence and uncertainty
 
-- **Observed:** Desired-state lock validation is active. Repository inventory,
-  drift, planning, application, override storage, and reporting remain target
-  systems.
+- **Observed:** Desired-state lock validation and the general catalog/manifest/
+  lock/Observatory convergence path are active. Application is restricted to
+  one exact reviewed candidate PR. A durable override registry beyond explicit
+  paused manifest state and lock exceptions remains a target system.
 - **Decided for this draft:** The repository owns the bounded concern described here and participates through versioned contracts.
 - **Proposed:** Target systems and later roadmap phases remain proposals until accepted and implemented.
 - **Open question:** Which parts of this draft should become active in the first independently versioned release?
diff --git a/contracts/holon.foundation-materialization.v1.lock.json b/contracts/holon.foundation-materialization.v1.lock.json
new file mode 100644
index 0000000..8ff336c
--- /dev/null
+++ b/contracts/holon.foundation-materialization.v1.lock.json
@@ -0,0 +1,22 @@
+{
+  "contracts": [
+    "https://egohygiene.io/schemas/holon/foundation-manifest.v1.schema.json",
+    "holon.materialization-plan/v1"
+  ],
+  "source_commit": "cd15eea31300997ce05a99d1a0bedc9c3437a8cc",
+  "source_repository": "https://github.com/egohygiene/holon",
+  "sources": [
+    {
+      "path": "schemas/foundation-manifest.v1.schema.json",
+      "sha256": "405a88a0df1b2d77ed534bdc65e833d61764883f93f5ef2c6042858bed0c9069"
+    },
+    {
+      "path": "schemas/materialization-plan.v1.schema.json",
+      "sha256": "7be8cd77ff41540889470937956e19b142077aa4721d62e205b2c3c2183845bd"
+    }
+  ],
+  "versions": {
+    "foundation_manifest": "1.0.0",
+    "materialization_plan": "holon.materialization-plan/v1"
+  }
+}
diff --git a/contracts/hygiene.repository-catalog.v1.lock.json b/contracts/hygiene.repository-catalog.v1.lock.json
new file mode 100644
index 0000000..4b445f1
--- /dev/null
+++ b/contracts/hygiene.repository-catalog.v1.lock.json
@@ -0,0 +1,16 @@
+{
+  "contract": "egohygiene.hygiene.repository-catalog/v1",
+  "source_commit": "28f9d6c7519d820644572634ba4476614f418d83",
+  "source_repository": "https://github.com/egohygiene/hygiene",
+  "sources": [
+    {
+      "path": "catalog/repositories.yaml",
+      "sha256": "e1193c772640ceb8d3d569a1826d3931ee9de64b0a8357a94c9f03fbceab52e5"
+    },
+    {
+      "path": "schemas/repository-catalog.v1.schema.json",
+      "sha256": "3a06f19e8c0be6590223b04f580aef7ee9c9b23bcb11e8b3da3db6dd1e764724"
+    }
+  ],
+  "version": "1.0.0"
+}
diff --git a/contracts/observatory.organization-health.v1.lock.json b/contracts/observatory.organization-health.v1.lock.json
new file mode 100644
index 0000000..e70cb07
--- /dev/null
+++ b/contracts/observatory.organization-health.v1.lock.json
@@ -0,0 +1,12 @@
+{
+  "contract": "egohygiene.observatory.organization-health/v1",
+  "source_commit": "a4526ba65a4ec1110ad9f2f47dc866a0c6d2c5a3",
+  "source_repository": "https://github.com/egohygiene/observatory",
+  "sources": [
+    {
+      "path": "schemas/organization-health.v1.schema.json",
+      "sha256": "4f61ece1f744f5fc8dda519305c43de4a855f3368b8eb7508d066b3bc2103af9"
+    }
+  ],
+  "version": "1.0.0-alpha.1"
+}
diff --git a/docs/fleet-convergence.md b/docs/fleet-convergence.md
new file mode 100644
index 0000000..a020462
--- /dev/null
+++ b/docs/fleet-convergence.md
@@ -0,0 +1,158 @@
+# Reviewed fleet convergence
+
+Pace turns declared fleet intent and observed repository state into one bounded,
+reviewable upgrade pull request at a time. Planning is offline and has no
+credential input. GitHub access is available only in the final `open-pr`
+command, after an exact plan and one unit have been approved.
+
+## Contract ownership
+
+| Input | Owner | Pace use |
+| --- | --- | --- |
+| Repository catalog v1 | Hygiene | Defines current fleet membership, lifecycle, maturity, and visibility. |
+| Foundation manifest v1 | Holon | Defines repository class, capabilities, and immutable foundation intent. |
+| Pace lock v1 | Pace | Defines current and desired immutable sources, target ownership, compatibility, exceptions, and rollback. |
+| Organization-health v1 | Observatory | Supplies the current, freshness-qualified represented commit. |
+| Materialization plan v1 | Holon | Authorizes exact generated-target operations when lock drift touches generated paths. |
+
+The exact upstream schema revisions are recorded in `contracts/`. Pace does not
+scan repositories, reinterpret Observatory checks, resolve Holon capabilities,
+or render generated files.
+
+## Fleet manifest
+
+`egohygiene.pace.fleet-convergence-manifest/v1` joins those inputs without
+copying them. Paths are normalized, relative to the fleet manifest, and cannot
+escape its directory. Each repository declares:
+
+- one Holon foundation manifest;
+- an optional current Pace lock and one adoptable desired Pace lock;
+- an optional exact Holon materialization plan;
+- the consumer-local lock path and default branch;
+- explicit repository-unit dependencies; and
+- an `active` or `paused` state.
+
+Dependencies must be managed by the same manifest and form an acyclic graph.
+Repositories are topologically ordered with lexical tie-breaking. Repositories
+in the Hygiene catalog but outside this manifest remain visible as unmanaged;
+that is deliberate partial-fleet adoption, not implicit success.
+
+## Generate a no-write plan
+
+The checked example is a contract fixture, not a live consumer:
+
+```bash
+python3 scripts/plan_fleet_convergence.py plan \
+  --manifest "examples/convergence/fleet.manifest.json" \
+  --catalog "examples/convergence/catalog.json" \
+  --observatory "examples/convergence/observatory.snapshot.json" \
+  --output "/tmp/fleet-convergence.plan.json"
+
+python3 scripts/plan_fleet_convergence.py verify-plan \
+  --plan "/tmp/fleet-convergence.plan.json"
+```
+
+The planner requires the Observatory snapshot to bind the exact supplied
+catalog bytes. A repository is blocked if Observatory evidence is stale,
+unknown, mixed, or represents anything other than one full commit SHA. Pace
+never substitutes the current branch head for missing observation.
+
+Each repository is one bounded unit with one of four dispositions:
+
+| Disposition | Meaning |
+| --- | --- |
+| `ready_for_review` | Drift exists, observation is current, locks are valid, and generated changes have exact Holon coverage. |
+| `blocked` | A contract, freshness, ownership, or materialization precondition is missing. |
+| `paused` | The manifest explicitly holds the unit; regenerating creates a superseding plan digest. |
+| `no_change` | Current and desired locks are identical. |
+
+Changes distinguish additions, removals, and updates and retain complete before
+and after lock entries. Risk is explainable: removals, target ownership/path
+changes, and compatibility-contract changes are high risk; new adoption,
+provenance changes, migration/rollback changes, and exception changes are at
+least medium risk. All other metadata-only changes are low risk.
+
+The plan embeds every input needed to verify its drift again. Its
+`plan_digest` binds the canonical plan, unit ordering, source digests, current
+and desired locks, Holon operations, blockers, and risk.
+
+## Review and partial adoption
+
+Review is a separate document. `approved_units` may be a strict subset of the
+plan; `completed_units` records dependency units already adopted:
+
+```json
+{
+  "schema": "egohygiene.pace.fleet-convergence-review/v1",
+  "plan_digest": "<exact plan digest>",
+  "reviewer": "reviewer-name",
+  "reviewed_at": "2026-09-02T13:00:00Z",
+  "decision": "approved",
+  "approved_units": ["upgrade:egohygiene/example-tool:<digest>"],
+  "completed_units": []
+}
+```
+
+Pace refuses an unapproved, blocked, paused, dependency-incomplete, unknown, or
+tampered unit. Produce exactly one pull-request proposal:
+
+```bash
+python3 scripts/plan_fleet_convergence.py propose \
+  --plan "/tmp/fleet-convergence.plan.json" \
+  --review "/tmp/fleet-convergence.review.json" \
+  --unit "upgrade:egohygiene/example-tool:<digest>" \
+  --output "/tmp/example-tool.pull-request.json"
+```
+
+The proposal embeds the desired lock, rollback anchor, exact base commit,
+deterministic branch, PR text, and a closed candidate-path allowlist. It carries
+no credential.
+
+## Prepare and open one pull request
+
+Prepare a local checkout at the proposal's exact `base_commit`. Write the
+desired lock to `lock_path`. If generated targets change, run Holon's `render`
+with the exact plan embedded in the proposal; Pace requires every mutating
+Holon operation to correspond to a changed generated lock target.
+
+Then use a fine-grained token with contents and pull-request write access to
+that consumer only:
+
+```bash
+GITHUB_TOKEN="<short-lived token>" \
+python3 scripts/plan_fleet_convergence.py open-pr \
+  --proposal "/tmp/example-tool.pull-request.json" \
+  --candidate "/path/to/example-tool" \
+  --output "/tmp/example-tool.pull-request-result.json"
+```
+
+Before any write, Pace verifies:
+
+1. the proposal's internal digest and embedded review;
+2. the candidate worktree is exactly at the reviewed commit;
+3. every changed path is allowlisted and every required path changed;
+4. the lock equals the reviewed desired lock;
+5. generated bytes and Holon state match the reviewed Holon plan; and
+6. the remote default branch still equals the represented commit.
+
+The adapter creates blobs, one tree, one commit, one non-default branch, and one
+pull request. It never updates the default-branch ref. A retry reuses the same
+branch and PR only when the remote candidate tree is identical; a different
+tree or an advanced base is refused and requires fresh observation and a new
+plan.
+
+## Pause, retry, rollback, and recovery
+
+- **Pause:** set the unit to `paused` and regenerate. The new plan digest
+  supersedes any earlier unexecuted review.
+- **Retry:** reuse the unchanged proposal. Branch/tree equality makes retries
+  idempotent and prevents silent replacement.
+- **Partial adoption:** approve only selected units. A dependent unit is not
+  proposed until its unit IDs appear in `completed_units`.
+- **Rollback:** the proposal retains the full prior lock and Holon's plan ID and
+  prior generated digests. After a failed merge, collect a fresh Observatory
+  snapshot, make that prior lock the desired lock, use Holon's fail-closed
+  rollback output for generated paths, and review the resulting reverse plan.
+  There is no force rollback and no direct default-branch mutation.
+
+Pull-request merge remains a human authority boundary outside Pace.
diff --git a/examples/convergence/catalog.json b/examples/convergence/catalog.json
new file mode 100644
index 0000000..8a7b080
--- /dev/null
+++ b/examples/convergence/catalog.json
@@ -0,0 +1,31 @@
+{
+  "schema_version": "1.0.0",
+  "architecture_release": "architecture-v0.1.0",
+  "observed_at": "2026-09-02",
+  "organization": "egohygiene",
+  "canonical_repository": "hygiene",
+  "repositories": [
+    {
+      "name": "example-tool",
+      "full_name": "egohygiene/example-tool",
+      "visibility": "public",
+      "plane": "developer-runtime",
+      "lifecycle": "active",
+      "maturity": "fixture",
+      "owns": ["convergence fixture"],
+      "excludes": ["production state"],
+      "consumes": [],
+      "outputs": ["test evidence"],
+      "domains": [],
+      "source_url": "https://github.com/egohygiene/example-tool"
+    }
+  ],
+  "proposed_repositories": [],
+  "planes": [
+    {
+      "id": "developer-runtime",
+      "title": "Developer runtime"
+    }
+  ],
+  "repository_count": 1
+}
diff --git a/examples/convergence/example-tool.current.lock.json b/examples/convergence/example-tool.current.lock.json
new file mode 100644
index 0000000..a0d7695
--- /dev/null
+++ b/examples/convergence/example-tool.current.lock.json
@@ -0,0 +1,41 @@
+{
+  "$schema": "../../schemas/pace-lock-v1.schema.json",
+  "schema": "egohygiene.pace.lock/v1",
+  "repository": "egohygiene/example-tool",
+  "policy": {
+    "unknown_contract": "fail-closed",
+    "update_mode": "reviewed-pull-request",
+    "max_exception_days": 30,
+    "rollback": "restore-previous-lock-and-owned-targets"
+  },
+  "locks": [
+    {
+      "id": "schema-repository-report",
+      "kind": "schema",
+      "source": {
+        "owner": "egohygiene/relay",
+        "locator": "https://github.com/egohygiene/relay",
+        "version": "v1.0.0",
+        "reference_type": "content-digest",
+        "reference": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+        "digest": {
+          "algorithm": "sha256",
+          "value": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+        }
+      },
+      "target": {
+        "path": "schemas/repository-report.schema.json",
+        "management": "consumer-owned",
+        "owner": "egohygiene/example-tool",
+        "generator": null
+      },
+      "compatibility": {
+        "contract": "egohygiene.relay.repository-report/v1",
+        "accepted_major": 1,
+        "migration": "not-required",
+        "rollback": "restore-lock-only"
+      },
+      "exception": null
+    }
+  ]
+}
diff --git a/examples/convergence/example-tool.desired.lock.json b/examples/convergence/example-tool.desired.lock.json
new file mode 100644
index 0000000..0f2cefb
--- /dev/null
+++ b/examples/convergence/example-tool.desired.lock.json
@@ -0,0 +1,41 @@
+{
+  "$schema": "../../schemas/pace-lock-v1.schema.json",
+  "schema": "egohygiene.pace.lock/v1",
+  "repository": "egohygiene/example-tool",
+  "policy": {
+    "unknown_contract": "fail-closed",
+    "update_mode": "reviewed-pull-request",
+    "max_exception_days": 30,
+    "rollback": "restore-previous-lock-and-owned-targets"
+  },
+  "locks": [
+    {
+      "id": "schema-repository-report",
+      "kind": "schema",
+      "source": {
+        "owner": "egohygiene/relay",
+        "locator": "https://github.com/egohygiene/relay",
+        "version": "v1.1.0",
+        "reference_type": "content-digest",
+        "reference": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+        "digest": {
+          "algorithm": "sha256",
+          "value": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+        }
+      },
+      "target": {
+        "path": "schemas/repository-report.schema.json",
+        "management": "consumer-owned",
+        "owner": "egohygiene/example-tool",
+        "generator": null
+      },
+      "compatibility": {
+        "contract": "egohygiene.relay.repository-report/v1",
+        "accepted_major": 1,
+        "migration": "not-required",
+        "rollback": "restore-lock-only"
+      },
+      "exception": null
+    }
+  ]
+}
diff --git a/examples/convergence/example-tool.foundation-manifest.json b/examples/convergence/example-tool.foundation-manifest.json
new file mode 100644
index 0000000..df9f344
--- /dev/null
+++ b/examples/convergence/example-tool.foundation-manifest.json
@@ -0,0 +1,17 @@
+{
+  "schema_version": "1.0.0",
+  "repository": "egohygiene/example-tool",
+  "repository_class": "tool",
+  "security_level": "hardened",
+  "pins": {
+    "architecture": "egohygiene/hygiene@architecture-v0.1.0",
+    "foundation": "egohygiene/holon@foundation-v0.1.0"
+  },
+  "capabilities": {
+    "include": [],
+    "exclude": []
+  },
+  "sites": [],
+  "preserve_paths": ["README.md", "LICENSE"],
+  "parameters": {}
+}
diff --git a/examples/convergence/fleet.manifest.json b/examples/convergence/fleet.manifest.json
new file mode 100644
index 0000000..f318b7f
--- /dev/null
+++ b/examples/convergence/fleet.manifest.json
@@ -0,0 +1,24 @@
+{
+  "$schema": "../../schemas/fleet-convergence-manifest-v1.schema.json",
+  "schema": "egohygiene.pace.fleet-convergence-manifest/v1",
+  "organization": "egohygiene",
+  "policy": {
+    "unknown_state": "block",
+    "update_mode": "reviewed-pull-request",
+    "unit": "repository",
+    "partial_adoption": true
+  },
+  "repositories": [
+    {
+      "repository": "egohygiene/example-tool",
+      "foundation_manifest": "example-tool.foundation-manifest.json",
+      "current_lock": "example-tool.current.lock.json",
+      "desired_lock": "example-tool.desired.lock.json",
+      "holon_plan": null,
+      "lock_path": ".config/pace.lock.json",
+      "default_branch": "main",
+      "depends_on": [],
+      "state": "active"
+    }
+  ]
+}
diff --git a/examples/convergence/observatory.snapshot.json b/examples/convergence/observatory.snapshot.json
new file mode 100644
index 0000000..6c06dd4
--- /dev/null
+++ b/examples/convergence/observatory.snapshot.json
@@ -0,0 +1,137 @@
+{
+  "schema": "egohygiene.observatory.organization-health/v1",
+  "contract_version": "1.0.0-alpha.1",
+  "snapshot_id": "organization-health:sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
+  "as_of": "2026-09-02T12:00:00Z",
+  "generator": {
+    "name": "egohygiene/observatory:organization-health",
+    "version": "0.1.0"
+  },
+  "organization": "egohygiene",
+  "sources": {
+    "catalog": {
+      "schema_version": "1.0.0",
+      "architecture_release": "architecture-v0.1.0",
+      "observed_at": "2026-09-02",
+      "freshness": "unknown",
+      "freshness_reason": "the catalog contract does not declare a validity window",
+      "url": "https://example.invalid/hygiene/catalog/repositories.yaml",
+      "commit": "dddddddddddddddddddddddddddddddddddddddd",
+      "sha256": "ba5093f649b4dc39babcec4e23042c0610b434c3fff13b36acb5e661bad6f0c2"
+    },
+    "evidence_record_count": 1,
+    "evidence_digests": [
+      "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
+    ]
+  },
+  "summary": {
+    "repository_count": 1,
+    "assessed_repository_count": 1,
+    "unknown_repository_count": 0,
+    "conformance": {
+      "conformant": 1
+    },
+    "maturity_evidence": {
+      "unknown": 1
+    }
+  },
+  "repositories": [
+    {
+      "repository": {
+        "name": "example-tool",
+        "full_name": "egohygiene/example-tool",
+        "url": "https://github.com/egohygiene/example-tool",
+        "visibility": "public",
+        "plane": "developer-runtime"
+      },
+      "declared": {
+        "lifecycle": "active",
+        "maturity": "fixture",
+        "source": "Hygiene repository catalog"
+      },
+      "assessment": {
+        "maturity": {
+          "status": "unknown",
+          "reason": "no required evidence assesses this dimension",
+          "rule": "required checks only; no numeric score",
+          "counts": {
+            "pass": 0,
+            "fail": 0,
+            "partial": 0,
+            "unknown": 0,
+            "blocked": 0,
+            "not_applicable": 0,
+            "stale": 0
+          },
+          "check_ids": []
+        },
+        "conformance": {
+          "status": "conformant",
+          "reason": "all applicable required checks passed with current evidence",
+          "rule": "required checks only; no numeric score",
+          "counts": {
+            "pass": 1,
+            "fail": 0,
+            "partial": 0,
+            "unknown": 0,
+            "blocked": 0,
+            "not_applicable": 0,
+            "stale": 0
+          },
+          "check_ids": ["egohygiene/pace:lock-state"]
+        }
+      },
+      "evidence": {
+        "freshness": "current",
+        "record_count": 1,
+        "check_count": 1,
+        "earliest_observed_at": "2026-09-02T11:00:00Z",
+        "latest_observed_at": "2026-09-02T11:00:00Z",
+        "represented_commits": [
+          "ffffffffffffffffffffffffffffffffffffffff"
+        ],
+        "producers": [
+          "egohygiene/pace"
+        ],
+        "checks": [
+          {
+            "id": "lock-state",
+            "key": "egohygiene/pace:lock-state",
+            "title": "Pace lock state",
+            "domain": "dependencies",
+            "status": "pass",
+            "reported_status": "pass",
+            "freshness": "current",
+            "severity": "required",
+            "assesses": [
+              "conformance"
+            ],
+            "contract": {
+              "id": "egohygiene.pace.lock/v1",
+              "version": "1",
+              "url": "https://github.com/egohygiene/pace/blob/99d479445406/schemas/pace-lock-v1.schema.json"
+            },
+            "evidence": [
+              "https://example.invalid/evidence/example-tool/pace-lock.json"
+            ],
+            "observation": {
+              "observed_at": "2026-09-02T11:00:00Z",
+              "valid_until": "2026-09-09T11:00:00Z",
+              "represented_commit": "ffffffffffffffffffffffffffffffffffffffff",
+              "producer": {
+                "name": "egohygiene/pace",
+                "version": "0.1.0",
+                "url": "https://github.com/egohygiene/pace"
+              },
+              "source": {
+                "kind": "pace-lock-validation",
+                "url": "https://example.invalid/evidence/example-tool/pace-lock.json",
+                "sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+              }
+            }
+          }
+        ]
+      }
+    }
+  ]
+}
diff --git a/schemas/README.md b/schemas/README.md
index f62bb86..448e7f7 100644
--- a/schemas/README.md
+++ b/schemas/README.md
@@ -3,6 +3,10 @@
 | Contract | Schema identity | File |
 | --- | --- | --- |
 | Repository dependency lock | `egohygiene.pace.lock/v1` | [`pace-lock-v1.schema.json`](pace-lock-v1.schema.json) |
+| Fleet convergence manifest | `egohygiene.pace.fleet-convergence-manifest/v1` | [`fleet-convergence-manifest-v1.schema.json`](fleet-convergence-manifest-v1.schema.json) |
+| Fleet convergence plan | `egohygiene.pace.fleet-convergence-plan/v1` | [`fleet-convergence-plan-v1.schema.json`](fleet-convergence-plan-v1.schema.json) |
+| Fleet convergence review | `egohygiene.pace.fleet-convergence-review/v1` | [`fleet-convergence-review-v1.schema.json`](fleet-convergence-review-v1.schema.json) |
+| Bounded upgrade PR | `egohygiene.pace.upgrade-pull-request/v1` | [`upgrade-pull-request-v1.schema.json`](upgrade-pull-request-v1.schema.json) |
 | Presentation fleet inventory | `egohygiene.pace.repository-presentation-inventory/v1` | [`repository-presentation-inventory-v1.schema.json`](repository-presentation-inventory-v1.schema.json) |
 | Presentation dry-run plan | `egohygiene.pace.repository-presentation-plan/v1` | [`repository-presentation-plan-v1.schema.json`](repository-presentation-plan-v1.schema.json) |
 | Reviewed-plan authorization | `egohygiene.pace.repository-presentation-review/v1` | [`repository-presentation-review-v1.schema.json`](repository-presentation-review-v1.schema.json) |
diff --git a/schemas/fleet-convergence-manifest-v1.schema.json b/schemas/fleet-convergence-manifest-v1.schema.json
new file mode 100644
index 0000000..6cc0689
--- /dev/null
+++ b/schemas/fleet-convergence-manifest-v1.schema.json
@@ -0,0 +1,51 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://egohygiene.github.io/pace/contracts/fleet-convergence-manifest/v1/schema.json",
+  "title": "Pace fleet convergence manifest v1",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["$schema", "schema", "organization", "policy", "repositories"],
+  "properties": {
+    "$schema": { "type": "string", "pattern": "(^|/)fleet-convergence-manifest-v1.schema.json$" },
+    "schema": { "const": "egohygiene.pace.fleet-convergence-manifest/v1" },
+    "organization": { "const": "egohygiene" },
+    "policy": {
+      "const": {
+        "unknown_state": "block",
+        "update_mode": "reviewed-pull-request",
+        "unit": "repository",
+        "partial_adoption": true
+      }
+    },
+    "repositories": {
+      "type": "array",
+      "minItems": 1,
+      "items": {
+        "type": "object",
+        "additionalProperties": false,
+        "required": [
+          "repository", "foundation_manifest", "current_lock", "desired_lock",
+          "holon_plan", "lock_path", "default_branch", "depends_on", "state"
+        ],
+        "properties": {
+          "repository": { "$ref": "#/$defs/repository" },
+          "foundation_manifest": { "$ref": "#/$defs/path" },
+          "current_lock": { "oneOf": [{ "$ref": "#/$defs/path" }, { "type": "null" }] },
+          "desired_lock": { "$ref": "#/$defs/path" },
+          "holon_plan": { "oneOf": [{ "$ref": "#/$defs/path" }, { "type": "null" }] },
+          "lock_path": { "$ref": "#/$defs/path" },
+          "default_branch": { "type": "string", "minLength": 1 },
+          "depends_on": { "type": "array", "uniqueItems": true, "items": { "$ref": "#/$defs/repository" } },
+          "state": { "enum": ["active", "paused"] }
+        }
+      }
+    }
+  },
+  "$defs": {
+    "repository": { "type": "string", "pattern": "^[a-z0-9_.-]+/[a-z0-9_.-]+$" },
+    "path": {
+      "type": "string",
+      "pattern": "^(?!/)(?!.*//)(?!.*(?:^|/)\\.{1,2}(?:/|$))(?!.*\\\\)(?!.*\/$)[^\\u0000]+$"
+    }
+  }
+}
diff --git a/schemas/fleet-convergence-plan-v1.schema.json b/schemas/fleet-convergence-plan-v1.schema.json
new file mode 100644
index 0000000..c9c37d8
--- /dev/null
+++ b/schemas/fleet-convergence-plan-v1.schema.json
@@ -0,0 +1,39 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://egohygiene.github.io/pace/contracts/fleet-convergence-plan/v1/schema.json",
+  "title": "Pace fleet convergence plan v1",
+  "$comment": "Canonical digest, dependency order, external-contract semantics, and exact review state are enforced by scripts/plan_fleet_convergence.py.",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["schema", "sources", "policy", "review", "summary", "units", "plan_digest"],
+  "properties": {
+    "schema": { "const": "egohygiene.pace.fleet-convergence-plan/v1" },
+    "sources": { "type": "object" },
+    "policy": { "type": "object" },
+    "review": { "type": "object" },
+    "summary": { "type": "object" },
+    "units": {
+      "type": "array",
+      "items": {
+        "type": "object",
+        "required": [
+          "unit_id", "order", "repository", "disposition", "risk", "blockers",
+          "depends_on", "changes", "current_lock_present", "current_lock", "desired_lock"
+        ],
+        "properties": {
+          "unit_id": { "type": "string", "pattern": "^upgrade:[a-z0-9_.-]+/[a-z0-9_.-]+:[0-9a-f]{12}$" },
+          "order": { "type": "integer", "minimum": 1 },
+          "repository": { "type": "string" },
+          "disposition": { "enum": ["ready_for_review", "blocked", "paused", "no_change"] },
+          "risk": { "enum": ["low", "medium", "high"] },
+          "blockers": { "type": "array", "items": { "type": "string" } },
+          "depends_on": { "type": "array", "items": { "type": "string" } },
+          "changes": { "type": "array", "items": { "type": "object" } },
+          "current_lock_present": { "type": "boolean" }
+        }
+      }
+    },
+    "plan_digest": { "$ref": "#/$defs/sha256" }
+  },
+  "$defs": { "sha256": { "type": "string", "pattern": "^[0-9a-f]{64}$" } }
+}
diff --git a/schemas/fleet-convergence-review-v1.schema.json b/schemas/fleet-convergence-review-v1.schema.json
new file mode 100644
index 0000000..4ad4fbc
--- /dev/null
+++ b/schemas/fleet-convergence-review-v1.schema.json
@@ -0,0 +1,21 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://egohygiene.github.io/pace/contracts/fleet-convergence-review/v1/schema.json",
+  "title": "Pace fleet convergence review v1",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "schema", "plan_digest", "reviewer", "reviewed_at", "decision",
+    "approved_units", "completed_units"
+  ],
+  "properties": {
+    "schema": { "const": "egohygiene.pace.fleet-convergence-review/v1" },
+    "plan_digest": { "$ref": "#/$defs/sha256" },
+    "reviewer": { "type": "string", "minLength": 1 },
+    "reviewed_at": { "type": "string", "format": "date-time", "pattern": "Z$" },
+    "decision": { "const": "approved" },
+    "approved_units": { "type": "array", "uniqueItems": true, "items": { "type": "string" } },
+    "completed_units": { "type": "array", "uniqueItems": true, "items": { "type": "string" } }
+  },
+  "$defs": { "sha256": { "type": "string", "pattern": "^[0-9a-f]{64}$" } }
+}
diff --git a/schemas/upgrade-pull-request-v1.schema.json b/schemas/upgrade-pull-request-v1.schema.json
new file mode 100644
index 0000000..4a03565
--- /dev/null
+++ b/schemas/upgrade-pull-request-v1.schema.json
@@ -0,0 +1,38 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://egohygiene.github.io/pace/contracts/upgrade-pull-request/v1/schema.json",
+  "title": "Pace bounded upgrade pull-request proposal v1",
+  "$comment": "Exact proposal integrity and candidate-tree verification are enforced by scripts/plan_fleet_convergence.py.",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "schema", "proposal_id", "proposal_digest", "plan_digest", "review", "unit_id",
+    "repository", "base_branch", "base_commit", "head_branch", "title", "body",
+    "risk", "changes", "depends_on", "lock_path", "expected_candidate_changes", "desired_lock",
+    "holon_plan", "rollback", "controls"
+  ],
+  "properties": {
+    "schema": { "const": "egohygiene.pace.upgrade-pull-request/v1" },
+    "proposal_id": { "type": "string", "pattern": "^pull-request:[0-9a-f]{64}$" },
+    "proposal_digest": { "$ref": "#/$defs/sha256" },
+    "plan_digest": { "$ref": "#/$defs/sha256" },
+    "review": { "type": "object" },
+    "unit_id": { "type": "string" },
+    "repository": { "type": "string" },
+    "base_branch": { "type": "string" },
+    "base_commit": { "type": "string", "pattern": "^[0-9a-f]{40}$" },
+    "head_branch": { "type": "string" },
+    "title": { "type": "string" },
+    "body": { "type": "string" },
+    "risk": { "enum": ["low", "medium", "high"] },
+    "changes": { "type": "array" },
+    "depends_on": { "type": "array", "uniqueItems": true, "items": { "type": "string" } },
+    "lock_path": { "type": "string" },
+    "expected_candidate_changes": { "type": "array", "minItems": 1 },
+    "desired_lock": { "type": "object" },
+    "holon_plan": { "type": ["object", "null"] },
+    "rollback": { "type": "object" },
+    "controls": { "type": "object" }
+  },
+  "$defs": { "sha256": { "type": "string", "pattern": "^[0-9a-f]{64}$" } }
+}
diff --git a/scripts/plan_fleet_convergence.py b/scripts/plan_fleet_convergence.py
new file mode 100755
index 0000000..351db25
--- /dev/null
+++ b/scripts/plan_fleet_convergence.py
@@ -0,0 +1,1655 @@
+#!/usr/bin/env python3
+# Copyright 2026 Ego Hygiene
+# SPDX-License-Identifier: MIT
+
+"""Plan reviewed fleet convergence and open one bounded pull request at a time."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+from collections import Counter
+from collections.abc import Mapping, Sequence
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+from urllib.error import HTTPError
+from urllib.parse import urlencode, urlparse
+from urllib.request import Request, urlopen
+
+SCRIPT_DIRECTORY = Path(__file__).resolve().parent
+if str(SCRIPT_DIRECTORY) not in sys.path:
+    sys.path.insert(0, str(SCRIPT_DIRECTORY))
+
+from validate_lock import load_lock, validate_lock
+
+MANIFEST_SCHEMA = "egohygiene.pace.fleet-convergence-manifest/v1"
+PLAN_SCHEMA = "egohygiene.pace.fleet-convergence-plan/v1"
+REVIEW_SCHEMA = "egohygiene.pace.fleet-convergence-review/v1"
+PROPOSAL_SCHEMA = "egohygiene.pace.upgrade-pull-request/v1"
+CATALOG_SCHEMA_VERSION = "1.0.0"
+FOUNDATION_MANIFEST_VERSION = "1.0.0"
+OBSERVATORY_SCHEMA = "egohygiene.observatory.organization-health/v1"
+OBSERVATORY_VERSION = "1.0.0-alpha.1"
+HOLON_PLAN_SCHEMA = "holon.materialization-plan/v1"
+COMMIT_RE = re.compile(r"^[0-9a-f]{40}$")
+DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")
+REPOSITORY_RE = re.compile(r"^[a-z0-9][a-z0-9.-]*/(?:\.github|[a-z0-9][a-z0-9.-]*)$")
+UNIT_RE = re.compile(r"^upgrade:[a-z0-9_.-]+/[a-z0-9_.-]+:[0-9a-f]{12}$")
+RISK_ORDER = {"low": 0, "medium": 1, "high": 2}
+MUTATING_HOLON_ACTIONS = {"create", "update", "delete"}
+MANIFEST_POLICY = {
+    "unknown_state": "block",
+    "update_mode": "reviewed-pull-request",
+    "unit": "repository",
+    "partial_adoption": True,
+}
+
+
+class ContractError(ValueError):
+    """Raised when an input crosses a closed contract boundary."""
+
+
+class DuplicateKeyError(ValueError):
+    """Raised when strict JSON input repeats a key."""
+
+
+def _strict_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    value: dict[str, Any] = {}
+    for key, item in pairs:
+        if key in value:
+            raise DuplicateKeyError(f"duplicate JSON key: {key}")
+        value[key] = item
+    return value
+
+
+def load_json(path: Path) -> dict[str, Any]:
+    value = json.loads(
+        path.read_text(encoding="utf-8"), object_pairs_hook=_strict_object
+    )
+    if not isinstance(value, dict):
+        raise ContractError(f"{path} must contain a JSON object")
+    return value
+
+
+def canonical_json(value: object) -> str:
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
+
+
+def digest(value: object) -> str:
+    payload = value if isinstance(value, str) else canonical_json(value)
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def file_digest(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def write_json(path: Path, document: object) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+
+
+def _require_keys(value: object, keys: set[str], location: str) -> Mapping[str, Any]:
+    if not isinstance(value, dict):
+        raise ContractError(f"{location} must be an object")
+    missing = keys - set(value)
+    unknown = set(value) - keys
+    if missing:
+        raise ContractError(f"{location} is missing keys: {', '.join(sorted(missing))}")
+    if unknown:
+        raise ContractError(
+            f"{location} has unknown keys: {', '.join(sorted(unknown))}"
+        )
+    return value
+
+
+def _timestamp(value: object, location: str) -> datetime:
+    if not isinstance(value, str) or not value.endswith("Z"):
+        raise ContractError(f"{location} must be an RFC 3339 UTC timestamp")
+    try:
+        parsed = datetime.fromisoformat(value[:-1] + "+00:00")
+    except ValueError as error:
+        raise ContractError(f"{location} must be an RFC 3339 UTC timestamp") from error
+    return parsed
+
+
+def _unique_sorted_strings(value: object) -> bool:
+    return (
+        isinstance(value, list)
+        and all(isinstance(item, str) for item in value)
+        and value == sorted(set(value))
+    )
+
+
+def _relative_path(value: object, location: str) -> str:
+    if (
+        not isinstance(value, str)
+        or not value
+        or value.startswith("/")
+        or value.endswith("/")
+        or "\\" in value
+        or "//" in value
+        or any(part in {"", ".", ".."} for part in value.split("/"))
+        or ".git" in value.split("/")
+    ):
+        raise ContractError(f"{location} must be a normalized relative path")
+    return value
+
+
+def _branch(value: object, location: str) -> str:
+    if (
+        not isinstance(value, str)
+        or not value
+        or not re.fullmatch(r"[A-Za-z0-9._/-]+", value)
+        or value.startswith(("/", "."))
+        or value.endswith(("/", ".", ".lock"))
+        or "//" in value
+        or ".." in value
+        or "@{" in value
+    ):
+        raise ContractError(f"{location} is not a safe Git branch name")
+    return value
+
+
+def _resolve_input(directory: Path, value: object, location: str) -> Path:
+    relative = _relative_path(value, location)
+    resolved = (directory / relative).resolve()
+    try:
+        resolved.relative_to(directory.resolve())
+    except ValueError as error:
+        raise ContractError(f"{location} escapes the manifest directory") from error
+    if not resolved.is_file():
+        raise ContractError(f"{location} does not identify a readable file: {relative}")
+    return resolved
+
+
+def _validate_fleet_manifest(document: object) -> list[dict[str, Any]]:
+    root = _require_keys(
+        document,
+        {"$schema", "schema", "organization", "policy", "repositories"},
+        "fleet manifest",
+    )
+    if root["schema"] != MANIFEST_SCHEMA:
+        raise ContractError(f"fleet manifest.schema must be {MANIFEST_SCHEMA}")
+    if root["organization"] != "egohygiene":
+        raise ContractError("fleet manifest.organization must be egohygiene")
+    if root["policy"] != MANIFEST_POLICY:
+        raise ContractError(
+            "fleet manifest.policy does not preserve the review boundary"
+        )
+    repositories = root["repositories"]
+    if not isinstance(repositories, list) or not repositories:
+        raise ContractError("fleet manifest.repositories must be a non-empty array")
+    expected = {
+        "repository",
+        "foundation_manifest",
+        "current_lock",
+        "desired_lock",
+        "holon_plan",
+        "lock_path",
+        "default_branch",
+        "depends_on",
+        "state",
+    }
+    normalized: list[dict[str, Any]] = []
+    names: list[str] = []
+    for index, raw in enumerate(repositories):
+        item = dict(_require_keys(raw, expected, f"repositories[{index}]"))
+        repository = item["repository"]
+        if (
+            not isinstance(repository, str)
+            or REPOSITORY_RE.fullmatch(repository) is None
+        ):
+            raise ContractError(f"repositories[{index}].repository is invalid")
+        names.append(repository)
+        for field in ("foundation_manifest", "desired_lock", "lock_path"):
+            _relative_path(item[field], f"repositories[{index}].{field}")
+        for field in ("current_lock", "holon_plan"):
+            if item[field] is not None:
+                _relative_path(item[field], f"repositories[{index}].{field}")
+        _branch(item["default_branch"], f"repositories[{index}].default_branch")
+        dependencies = item["depends_on"]
+        if not isinstance(dependencies, list) or not all(
+            isinstance(dependency, str) and REPOSITORY_RE.fullmatch(dependency)
+            for dependency in dependencies
+        ):
+            raise ContractError(f"repositories[{index}].depends_on is invalid")
+        if dependencies != sorted(set(dependencies)):
+            raise ContractError(
+                f"repositories[{index}].depends_on must be unique and sorted"
+            )
+        if repository in dependencies:
+            raise ContractError(f"repositories[{index}] cannot depend on itself")
+        if item["state"] not in {"active", "paused"}:
+            raise ContractError(f"repositories[{index}].state must be active or paused")
+        normalized.append(item)
+    if names != sorted(set(names)):
+        raise ContractError("fleet manifest repositories must be unique and sorted")
+    known = set(names)
+    for item in normalized:
+        missing = set(item["depends_on"]) - known
+        if missing:
+            raise ContractError(
+                f"{item['repository']} depends on unmanaged repositories: {', '.join(sorted(missing))}"
+            )
+    _topological_repositories(normalized)
+    return normalized
+
+
+def _topological_repositories(records: Sequence[Mapping[str, Any]]) -> list[str]:
+    dependencies = {item["repository"]: set(item["depends_on"]) for item in records}
+    ordered: list[str] = []
+    ready = sorted(name for name, values in dependencies.items() if not values)
+    while ready:
+        name = ready.pop(0)
+        ordered.append(name)
+        for candidate in sorted(dependencies):
+            if name in dependencies[candidate]:
+                dependencies[candidate].remove(name)
+                if (
+                    not dependencies[candidate]
+                    and candidate not in ordered
+                    and candidate not in ready
+                ):
+                    ready.append(candidate)
+                    ready.sort()
+    if len(ordered) != len(records):
+        cycle = sorted(name for name, values in dependencies.items() if values)
+        raise ContractError(f"fleet manifest dependency cycle: {', '.join(cycle)}")
+    return ordered
+
+
+def _catalog_records(catalog: object) -> dict[str, Mapping[str, Any]]:
+    if not isinstance(catalog, dict):
+        raise ContractError("catalog must be an object")
+    if catalog.get("schema_version") != CATALOG_SCHEMA_VERSION:
+        raise ContractError("catalog schema_version is unsupported")
+    if catalog.get("organization") != "egohygiene":
+        raise ContractError("catalog organization must be egohygiene")
+    repositories = catalog.get("repositories")
+    if not isinstance(repositories, list) or not repositories:
+        raise ContractError("catalog.repositories must be a non-empty array")
+    result: dict[str, Mapping[str, Any]] = {}
+    for index, item in enumerate(repositories):
+        if not isinstance(item, dict):
+            raise ContractError(f"catalog.repositories[{index}] must be an object")
+        name = item.get("full_name")
+        if not isinstance(name, str) or REPOSITORY_RE.fullmatch(name) is None:
+            raise ContractError(f"catalog.repositories[{index}].full_name is invalid")
+        if name in result:
+            raise ContractError(f"catalog contains duplicate repository {name}")
+        result[name] = item
+    declared_count = catalog.get("repository_count")
+    if declared_count != len(repositories):
+        raise ContractError("catalog.repository_count does not match repositories")
+    return result
+
+
+def _observation_records(
+    observation: object, catalog_path: Path
+) -> tuple[Mapping[str, Any], dict[str, Mapping[str, Any]]]:
+    if not isinstance(observation, dict):
+        raise ContractError("Observatory snapshot must be an object")
+    if observation.get("schema") != OBSERVATORY_SCHEMA:
+        raise ContractError(f"Observatory snapshot schema must be {OBSERVATORY_SCHEMA}")
+    if observation.get("contract_version") != OBSERVATORY_VERSION:
+        raise ContractError(
+            f"Observatory contract must be pinned to {OBSERVATORY_VERSION}"
+        )
+    if observation.get("organization") != "egohygiene":
+        raise ContractError("Observatory organization must be egohygiene")
+    snapshot_id = observation.get("snapshot_id")
+    if not isinstance(snapshot_id, str) or not re.fullmatch(
+        r"organization-health:sha256:[0-9a-f]{64}", snapshot_id
+    ):
+        raise ContractError("Observatory snapshot_id is invalid")
+    _timestamp(observation.get("as_of"), "Observatory as_of")
+    sources = observation.get("sources")
+    if not isinstance(sources, dict) or not isinstance(sources.get("catalog"), dict):
+        raise ContractError("Observatory catalog provenance is missing")
+    if sources["catalog"].get("sha256") != file_digest(catalog_path):
+        raise ContractError(
+            "Observatory snapshot does not represent the supplied catalog bytes"
+        )
+    repositories = observation.get("repositories")
+    if not isinstance(repositories, list):
+        raise ContractError("Observatory repositories must be an array")
+    result: dict[str, Mapping[str, Any]] = {}
+    for index, item in enumerate(repositories):
+        if not isinstance(item, dict) or not isinstance(item.get("repository"), dict):
+            raise ContractError(f"Observatory repositories[{index}] is malformed")
+        name = item["repository"].get("full_name")
+        if not isinstance(name, str) or REPOSITORY_RE.fullmatch(name) is None:
+            raise ContractError(
+                f"Observatory repositories[{index}] has an invalid name"
+            )
+        if name in result:
+            raise ContractError(f"Observatory snapshot repeats repository {name}")
+        result[name] = item
+    return observation, result
+
+
+def _validate_foundation_manifest(document: object, repository: str) -> None:
+    if not isinstance(document, dict):
+        raise ContractError(f"{repository} foundation manifest must be an object")
+    if document.get("schema_version") != FOUNDATION_MANIFEST_VERSION:
+        raise ContractError(f"{repository} foundation manifest version is unsupported")
+    if document.get("repository") != repository:
+        raise ContractError(f"{repository} foundation manifest identity does not match")
+    pins = document.get("pins")
+    if not isinstance(pins, dict) or not all(
+        isinstance(pins.get(key), str) and pins[key].strip()
+        for key in ("architecture", "foundation")
+    ):
+        raise ContractError(
+            f"{repository} foundation manifest lacks immutable core pins"
+        )
+
+
+def _as_of(observation: Mapping[str, Any]) -> datetime:
+    return _timestamp(observation["as_of"], "Observatory as_of")
+
+
+def _lock_errors(document: object, observation: Mapping[str, Any]) -> list[str]:
+    return validate_lock(document, _as_of(observation))
+
+
+def _ownership_errors(document: Mapping[str, Any], repository: str) -> list[str]:
+    errors: list[str] = []
+    for item in document.get("locks", []):
+        if isinstance(item, dict) and item.get("target", {}).get("owner") != repository:
+            errors.append(f"lock {item.get('id')} target owner is not {repository}")
+    return errors
+
+
+def _change_risk(
+    before: object, after: object, change_type: str
+) -> tuple[str, list[str]]:
+    if change_type == "remove":
+        return "high", ["removes a dependency and its recorded rollback state"]
+    if change_type == "add":
+        return "medium", ["adopts a new dependency and target ownership record"]
+    if not isinstance(before, dict) or not isinstance(after, dict):
+        raise ContractError(
+            "updated lock entries must contain before and after objects"
+        )
+    reasons: list[str] = []
+    risk = "low"
+    if before.get("target") != after.get("target"):
+        risk = "high"
+        reasons.append("changes target path or ownership")
+    before_contract = before.get("compatibility", {}).get("contract")
+    after_contract = after.get("compatibility", {}).get("contract")
+    if before_contract != after_contract:
+        risk = "high"
+        reasons.append("changes the compatibility contract")
+    if before.get("exception") != after.get("exception"):
+        risk = max((risk, "medium"), key=RISK_ORDER.get)
+        reasons.append("changes a bounded exception")
+    if before.get("source") != after.get("source"):
+        risk = max((risk, "medium"), key=RISK_ORDER.get)
+        reasons.append("changes immutable source provenance")
+    if before.get("compatibility") != after.get("compatibility") and not reasons:
+        risk = "medium"
+        reasons.append("changes migration or rollback declarations")
+    if not reasons:
+        reasons.append("changes dependency metadata")
+    return risk, reasons
+
+
+def _lock_changes(
+    current: Mapping[str, Any], desired: Mapping[str, Any]
+) -> list[dict[str, Any]]:
+    before_by_id = {item["id"]: item for item in current.get("locks", [])}
+    after_by_id = {item["id"]: item for item in desired.get("locks", [])}
+    changes: list[dict[str, Any]] = []
+    if current.get("policy") != desired.get("policy"):
+        changes.append(
+            {
+                "id": "lock-policy",
+                "type": "update",
+                "kind": "policy",
+                "risk": "medium",
+                "reasons": ["changes fleet update, exception, or rollback policy"],
+                "before": current.get("policy"),
+                "after": desired.get("policy"),
+            }
+        )
+    for identifier in sorted(set(before_by_id) | set(after_by_id)):
+        before = before_by_id.get(identifier)
+        after = after_by_id.get(identifier)
+        if before == after:
+            continue
+        if before is None:
+            change_type = "add"
+        elif after is None:
+            change_type = "remove"
+        else:
+            change_type = "update"
+        risk, reasons = _change_risk(before, after, change_type)
+        changes.append(
+            {
+                "id": identifier,
+                "type": change_type,
+                "kind": (after or before)["kind"],
+                "risk": risk,
+                "reasons": reasons,
+                "before": before,
+                "after": after,
+            }
+        )
+    return changes
+
+
+def _plan_without_digest(plan: Mapping[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in plan.items() if key != "plan_digest"}
+
+
+def _holon_operations(
+    document: Mapping[str, Any], repository: str
+) -> tuple[list[dict[str, Any]], list[str]]:
+    errors: list[str] = []
+    if document.get("schema_version") != HOLON_PLAN_SCHEMA:
+        errors.append("Holon plan schema is unsupported")
+    if document.get("repository") != repository:
+        errors.append("Holon plan repository does not match")
+    operations = document.get("operations")
+    if not isinstance(operations, list):
+        return [], errors + ["Holon plan operations are missing"]
+    payload = {key: value for key, value in document.items() if key != "plan_id"}
+    if document.get("plan_id") != digest(payload):
+        errors.append("Holon plan_id does not match its canonical payload")
+    normalized: list[dict[str, Any]] = []
+    for index, operation in enumerate(operations):
+        if not isinstance(operation, dict):
+            errors.append(f"Holon operation {index} is malformed")
+            continue
+        action = operation.get("action")
+        path = operation.get("path")
+        if action == "conflict":
+            errors.append(f"Holon reports a conflict at {path}")
+        if action in MUTATING_HOLON_ACTIONS:
+            try:
+                _relative_path(path, f"Holon operation {index}.path")
+            except ContractError as error:
+                errors.append(str(error))
+                continue
+            normalized.append(dict(operation))
+    return normalized, errors
+
+
+def _generated_targets(changes: Sequence[Mapping[str, Any]]) -> set[str]:
+    targets: set[str] = set()
+    for change in changes:
+        for side in (change.get("before"), change.get("after")):
+            if (
+                isinstance(side, dict)
+                and side.get("target", {}).get("management") == "generated"
+            ):
+                targets.add(side["target"]["path"])
+    return targets
+
+
+def build_plan(
+    manifest_path: Path, catalog_path: Path, observation_path: Path
+) -> dict[str, Any]:
+    manifest = load_json(manifest_path)
+    fleet_records = _validate_fleet_manifest(manifest)
+    catalog = load_json(catalog_path)
+    catalog_records = _catalog_records(catalog)
+    observation, observation_records = _observation_records(
+        load_json(observation_path), catalog_path
+    )
+    if set(catalog_records) != set(observation_records):
+        raise ContractError("Observatory and Hygiene repository membership differ")
+
+    manifest_directory = manifest_path.resolve().parent
+    records_by_name = {item["repository"]: item for item in fleet_records}
+    order = _topological_repositories(fleet_records)
+    units_by_repository: dict[str, dict[str, Any]] = {}
+
+    for repository in order:
+        configured = records_by_name[repository]
+        if repository not in catalog_records:
+            raise ContractError(f"{repository} is absent from the Hygiene catalog")
+        foundation_path = _resolve_input(
+            manifest_directory,
+            configured["foundation_manifest"],
+            f"{repository}.foundation_manifest",
+        )
+        desired_path = _resolve_input(
+            manifest_directory, configured["desired_lock"], f"{repository}.desired_lock"
+        )
+        current_path = None
+        if configured["current_lock"] is not None:
+            current_path = _resolve_input(
+                manifest_directory,
+                configured["current_lock"],
+                f"{repository}.current_lock",
+            )
+        holon_path = None
+        if configured["holon_plan"] is not None:
+            holon_path = _resolve_input(
+                manifest_directory, configured["holon_plan"], f"{repository}.holon_plan"
+            )
+
+        foundation = load_json(foundation_path)
+        _validate_foundation_manifest(foundation, repository)
+        desired = load_lock(desired_path)
+        if desired.get("repository") != repository:
+            raise ContractError(f"{repository} desired lock identity does not match")
+        desired_errors = _lock_errors(desired, observation) + _ownership_errors(
+            desired, repository
+        )
+        if desired_errors:
+            raise ContractError(
+                f"{repository} desired lock is not adoptable: {'; '.join(desired_errors)}"
+            )
+        current: dict[str, Any] = {
+            "$schema": desired["$schema"],
+            "schema": desired["schema"],
+            "repository": repository,
+            "policy": desired["policy"],
+            "locks": [],
+        }
+        current_errors: list[str] = []
+        if current_path is not None:
+            current = load_lock(current_path)
+            if current.get("repository") != repository:
+                raise ContractError(
+                    f"{repository} current lock identity does not match"
+                )
+            current_errors = _lock_errors(current, observation) + _ownership_errors(
+                current, repository
+            )
+
+        changes = _lock_changes(current, desired)
+        blockers = list(current_errors)
+        observed = observation_records[repository]
+        evidence = observed.get("evidence")
+        if not isinstance(evidence, dict):
+            blockers.append("Observatory repository evidence is malformed")
+            represented_commit = None
+        else:
+            freshness = evidence.get("freshness")
+            commits = evidence.get("represented_commits")
+            if freshness != "current":
+                blockers.append(
+                    f"Observatory evidence freshness is {freshness or 'unknown'}"
+                )
+            if (
+                not isinstance(commits, list)
+                or len(commits) != 1
+                or not isinstance(commits[0], str)
+                or COMMIT_RE.fullmatch(commits[0]) is None
+            ):
+                blockers.append(
+                    "Observatory must identify exactly one represented commit"
+                )
+                represented_commit = None
+            else:
+                represented_commit = commits[0]
+
+        holon_plan = None
+        holon_operations: list[dict[str, Any]] = []
+        if holon_path is not None:
+            holon_plan = load_json(holon_path)
+            holon_operations, holon_errors = _holon_operations(holon_plan, repository)
+            blockers.extend(holon_errors)
+        generated = _generated_targets(changes)
+        operation_paths = {operation["path"] for operation in holon_operations}
+        coverage_paths = set(operation_paths)
+        if holon_plan is not None:
+            for operation in holon_plan.get("operations", []):
+                if not isinstance(operation, dict):
+                    continue
+                if operation.get("action") == "noop":
+                    coverage_paths.add(operation.get("path"))
+                if (
+                    operation.get("action") == "preserve"
+                    and operation.get("path") in generated
+                ):
+                    blockers.append(
+                        f"Holon preserves generated lock target: {operation.get('path')}"
+                    )
+        if generated and holon_plan is None:
+            blockers.append(
+                "generated-target drift requires an exact Holon materialization plan"
+            )
+        missing_operations = generated - coverage_paths
+        if missing_operations:
+            blockers.append(
+                "Holon plan does not cover generated targets: "
+                + ", ".join(sorted(missing_operations))
+            )
+        extra_operations = operation_paths - generated
+        if extra_operations:
+            blockers.append(
+                "Holon plan mutates paths outside lock drift: "
+                + ", ".join(sorted(extra_operations))
+            )
+
+        risk = max(
+            (change["risk"] for change in changes),
+            key=RISK_ORDER.get,
+            default="low",
+        )
+        unit_seed = {
+            "repository": repository,
+            "current_lock": digest(current),
+            "desired_lock": digest(desired),
+            "represented_commit": represented_commit,
+        }
+        unit_id = f"upgrade:{repository}:{digest(unit_seed)[:12]}"
+        if not changes:
+            disposition = "no_change"
+        elif configured["state"] == "paused":
+            disposition = "paused"
+        elif blockers:
+            disposition = "blocked"
+        else:
+            disposition = "ready_for_review"
+        units_by_repository[repository] = {
+            "unit_id": unit_id,
+            "order": len(units_by_repository) + 1,
+            "repository": repository,
+            "default_branch": configured["default_branch"],
+            "represented_commit": represented_commit,
+            "catalog": {
+                "lifecycle": catalog_records[repository].get("lifecycle"),
+                "maturity": catalog_records[repository].get("maturity"),
+                "visibility": catalog_records[repository].get("visibility"),
+            },
+            "disposition": disposition,
+            "risk": risk,
+            "blockers": sorted(set(blockers)),
+            "depends_on_repositories": configured["depends_on"],
+            "depends_on": [],
+            "lock_path": configured["lock_path"],
+            "changes": changes,
+            "sources": {
+                "foundation_manifest_sha256": digest(foundation),
+                "current_lock_sha256": digest(current),
+                "desired_lock_sha256": digest(desired),
+                "holon_plan_sha256": digest(holon_plan) if holon_plan else None,
+            },
+            "foundation_manifest": foundation,
+            "current_lock_present": current_path is not None,
+            "current_lock": current,
+            "desired_lock": desired,
+            "holon_plan": holon_plan,
+            "holon_operations": holon_operations,
+        }
+
+    units = [units_by_repository[name] for name in order]
+    for unit in units:
+        unit["depends_on"] = [
+            units_by_repository[name]["unit_id"]
+            for name in unit.pop("depends_on_repositories")
+            if units_by_repository[name]["disposition"] != "no_change"
+        ]
+    counts = Counter(unit["disposition"] for unit in units)
+    plan: dict[str, Any] = {
+        "schema": PLAN_SCHEMA,
+        "sources": {
+            "fleet_manifest_sha256": digest(manifest),
+            "catalog_sha256": file_digest(catalog_path),
+            "observatory_snapshot_id": observation["snapshot_id"],
+            "observatory_snapshot_sha256": digest(observation),
+            "observatory_as_of": observation["as_of"],
+        },
+        "policy": MANIFEST_POLICY,
+        "review": {
+            "required_before_pull_request": True,
+            "approved": False,
+            "review_record": None,
+        },
+        "summary": {
+            "catalog_repositories": len(catalog_records),
+            "managed_repositories": len(units),
+            "unmanaged_repositories": sorted(
+                set(catalog_records) - set(records_by_name)
+            ),
+            "dispositions": dict(sorted(counts.items())),
+            "risk": dict(sorted(Counter(unit["risk"] for unit in units).items())),
+        },
+        "units": units,
+    }
+    plan["plan_digest"] = digest(_plan_without_digest(plan))
+    return plan
+
+
+def validate_plan(plan: object) -> list[str]:
+    if not isinstance(plan, dict):
+        return ["plan must be an object"]
+    errors: list[str] = []
+    if plan.get("schema") != PLAN_SCHEMA:
+        errors.append(f"plan.schema must be {PLAN_SCHEMA}")
+    if plan.get("policy") != MANIFEST_POLICY:
+        errors.append("plan policy changed")
+    if plan.get("plan_digest") != digest(_plan_without_digest(plan)):
+        errors.append("plan_digest does not match the canonical plan")
+    review = plan.get("review")
+    if review != {
+        "required_before_pull_request": True,
+        "approved": False,
+        "review_record": None,
+    }:
+        errors.append("plan does not preserve the unreviewed boundary")
+    units = plan.get("units")
+    if not isinstance(units, list):
+        return errors + ["plan.units must be an array"]
+    ids: set[str] = set()
+    repositories: set[str] = set()
+    prior: set[str] = set()
+    for index, unit in enumerate(units):
+        if not isinstance(unit, dict):
+            errors.append(f"units[{index}] must be an object")
+            continue
+        identifier = unit.get("unit_id")
+        repository = unit.get("repository")
+        if not isinstance(identifier, str) or UNIT_RE.fullmatch(identifier) is None:
+            errors.append(f"units[{index}].unit_id is invalid")
+        elif identifier in ids:
+            errors.append(f"duplicate unit_id: {identifier}")
+        else:
+            ids.add(identifier)
+        if (
+            not isinstance(repository, str)
+            or REPOSITORY_RE.fullmatch(repository) is None
+        ):
+            errors.append(f"units[{index}].repository is invalid")
+        elif repository in repositories:
+            errors.append(f"duplicate unit repository: {repository}")
+        else:
+            repositories.add(repository)
+        dependencies = unit.get("depends_on")
+        if not isinstance(dependencies, list) or not all(
+            isinstance(item, str) for item in dependencies
+        ):
+            errors.append(f"units[{index}].depends_on must be an array of unit IDs")
+        elif any(item not in prior for item in dependencies):
+            errors.append(f"units[{index}] dependencies are not in plan order")
+        if unit.get("order") != index + 1:
+            errors.append(f"units[{index}].order is not deterministic")
+        sources = unit.get("sources")
+        if not isinstance(sources, dict):
+            errors.append(f"units[{index}].sources must be an object")
+        else:
+            source_bindings = (
+                ("foundation_manifest_sha256", "foundation_manifest"),
+                ("current_lock_sha256", "current_lock"),
+                ("desired_lock_sha256", "desired_lock"),
+            )
+            for source_key, document_key in source_bindings:
+                if sources.get(source_key) != digest(unit.get(document_key)):
+                    errors.append(
+                        f"units[{index}].{source_key} does not match embedded input"
+                    )
+            expected_holon_digest = (
+                digest(unit.get("holon_plan"))
+                if unit.get("holon_plan") is not None
+                else None
+            )
+            if sources.get("holon_plan_sha256") != expected_holon_digest:
+                errors.append(
+                    f"units[{index}].holon_plan_sha256 does not match embedded input"
+                )
+            seed = {
+                "repository": repository,
+                "current_lock": sources.get("current_lock_sha256"),
+                "desired_lock": sources.get("desired_lock_sha256"),
+                "represented_commit": unit.get("represented_commit"),
+            }
+            expected_id = f"upgrade:{repository}:{digest(seed)[:12]}"
+            if identifier != expected_id:
+                errors.append(f"units[{index}].unit_id does not bind its inputs")
+        current = unit.get("current_lock")
+        desired = unit.get("desired_lock")
+        if not isinstance(unit.get("current_lock_present"), bool):
+            errors.append(f"units[{index}].current_lock_present must be boolean")
+        if isinstance(current, dict) and isinstance(desired, dict):
+            try:
+                expected_changes = _lock_changes(current, desired)
+            except (AttributeError, KeyError, TypeError):
+                expected_changes = []
+                errors.append(f"units[{index}] embedded locks are malformed")
+            if unit.get("changes") != expected_changes:
+                errors.append(f"units[{index}].changes do not match lock drift")
+            expected_risk = max(
+                (change["risk"] for change in expected_changes),
+                key=RISK_ORDER.get,
+                default="low",
+            )
+            if unit.get("risk") != expected_risk:
+                errors.append(f"units[{index}].risk does not match explained drift")
+            disposition = unit.get("disposition")
+            if not expected_changes and disposition != "no_change":
+                errors.append(f"units[{index}] without drift must be no_change")
+            if expected_changes and disposition == "no_change":
+                errors.append(f"units[{index}] with drift cannot be no_change")
+        else:
+            errors.append(f"units[{index}] must embed current and desired locks")
+        holon_plan = unit.get("holon_plan")
+        if holon_plan is None:
+            expected_operations: list[dict[str, Any]] = []
+        elif isinstance(holon_plan, dict) and isinstance(repository, str):
+            expected_operations, holon_errors = _holon_operations(
+                holon_plan, repository
+            )
+            errors.extend(f"units[{index}]: {error}" for error in holon_errors)
+        else:
+            expected_operations = []
+            errors.append(f"units[{index}].holon_plan is malformed")
+        if unit.get("holon_operations") != expected_operations:
+            errors.append(
+                f"units[{index}].holon_operations do not match the Holon plan"
+            )
+        if isinstance(identifier, str):
+            prior.add(identifier)
+    return errors
+
+
+def validate_review(review: object, plan: Mapping[str, Any]) -> list[str]:
+    if not isinstance(review, dict):
+        return ["review must be an object"]
+    errors: list[str] = []
+    expected = {
+        "schema",
+        "plan_digest",
+        "reviewer",
+        "reviewed_at",
+        "decision",
+        "approved_units",
+        "completed_units",
+    }
+    if set(review) != expected:
+        errors.append("review contains unsupported or missing fields")
+    if review.get("schema") != REVIEW_SCHEMA:
+        errors.append(f"review.schema must be {REVIEW_SCHEMA}")
+    if review.get("plan_digest") != plan.get("plan_digest"):
+        errors.append("review does not authorize this exact plan digest")
+    if not isinstance(review.get("reviewer"), str) or not review["reviewer"].strip():
+        errors.append("reviewer is required")
+    try:
+        _timestamp(review.get("reviewed_at"), "review.reviewed_at")
+    except ContractError as error:
+        errors.append(str(error))
+    if review.get("decision") != "approved":
+        errors.append("review decision must be approved")
+    plan_units = plan.get("units")
+    if not isinstance(plan_units, list):
+        plan_units = []
+    unit_ids = {
+        unit.get("unit_id")
+        for unit in plan_units
+        if isinstance(unit, dict) and isinstance(unit.get("unit_id"), str)
+    }
+    for field in ("approved_units", "completed_units"):
+        values = review.get(field)
+        if not _unique_sorted_strings(values):
+            errors.append(f"review.{field} must be a unique sorted array")
+        elif not set(values) <= unit_ids:
+            errors.append(f"review.{field} contains an unknown unit")
+    return errors
+
+
+def _proposal_without_digest(proposal: Mapping[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in proposal.items() if key != "proposal_digest"}
+
+
+def _single_line(value: object) -> str:
+    return " ".join(str(value).split()).replace("`", "'")
+
+
+def build_proposal(
+    plan: dict[str, Any], review: dict[str, Any], unit_id: str
+) -> dict[str, Any]:
+    errors = validate_plan(plan) + validate_review(review, plan)
+    if errors:
+        raise ContractError("pull-request proposal refused: " + "; ".join(errors))
+    units = {
+        item.get("unit_id"): item
+        for item in plan.get("units", [])
+        if isinstance(item, dict)
+    }
+    unit = units.get(unit_id)
+    if unit is None:
+        errors.append(f"unit is absent from the reviewed plan: {unit_id}")
+    else:
+        if unit["disposition"] != "ready_for_review":
+            errors.append(f"unit is {unit['disposition']}, not ready_for_review")
+        if unit_id not in review.get("approved_units", []):
+            errors.append("review does not approve this unit")
+        missing_dependencies = set(unit["depends_on"]) - set(
+            review.get("completed_units", [])
+        )
+        if missing_dependencies:
+            errors.append(
+                "unit dependencies are incomplete: "
+                + ", ".join(sorted(missing_dependencies))
+            )
+    if errors:
+        raise ContractError("pull-request proposal refused: " + "; ".join(errors))
+    if unit is None:  # pragma: no cover - guarded by the refusal above
+        raise ContractError("reviewed unit disappeared during proposal construction")
+
+    expected_changes: list[dict[str, Any]] = [
+        {
+            "path": unit["lock_path"],
+            "action": "upsert",
+            "verification": "exact-json",
+            "desired_sha256": unit["sources"]["desired_lock_sha256"],
+            "previous_sha256": (
+                unit["sources"]["current_lock_sha256"]
+                if unit["current_lock_present"]
+                else None
+            ),
+        }
+    ]
+    for operation in unit["holon_operations"]:
+        expected_changes.append(
+            {
+                "path": operation["path"],
+                "action": "delete" if operation["action"] == "delete" else "upsert",
+                "verification": "sha256",
+                "desired_sha256": operation["desired_sha256"],
+                "previous_sha256": operation["previous_sha256"],
+            }
+        )
+    if unit["holon_operations"]:
+        expected_changes.append(
+            {
+                "path": ".holon/materialization-state.v1.json",
+                "action": "upsert",
+                "verification": "holon-state",
+                "desired_sha256": None,
+                "previous_sha256": None,
+            }
+        )
+    expected_changes.sort(key=lambda item: item["path"])
+    branch_slug = re.sub(
+        r"[^a-z0-9-]+", "-", unit["repository"].split("/", 1)[1]
+    ).strip("-")
+    branch = f"pace/{branch_slug}-{plan['plan_digest'][:12]}"
+    change_lines = [
+        f"- `{change['id']}`: {change['type']} {change['kind']} ({change['risk']} risk)"
+        for change in unit["changes"]
+    ]
+    body = "\n".join(
+        [
+            f"Converges `{unit['repository']}` through reviewed Pace unit `{unit_id}`.",
+            "",
+            "## Explainable drift",
+            "",
+            *change_lines,
+            "",
+            "## Safety boundary",
+            "",
+            f"- Plan: `{plan['plan_digest']}`",
+            f"- Reviewed by: `{_single_line(review['reviewer'])}` at `{review['reviewed_at']}`",
+            f"- Base commit: `{unit['represented_commit']}`",
+            "- Retry is idempotent on the proposal branch and candidate tree.",
+            "- Rollback restores the previous lock and uses Holon's recorded prior bytes for generated targets.",
+            "",
+            f"Closes no issue automatically. Planned from egohygiene/pace#{2}.",
+        ]
+    )
+    proposal: dict[str, Any] = {
+        "schema": PROPOSAL_SCHEMA,
+        "proposal_id": f"pull-request:{digest({'plan': plan['plan_digest'], 'unit': unit_id})}",
+        "plan_digest": plan["plan_digest"],
+        "review": review,
+        "unit_id": unit_id,
+        "repository": unit["repository"],
+        "base_branch": unit["default_branch"],
+        "base_commit": unit["represented_commit"],
+        "head_branch": branch,
+        "title": f"chore(pace): converge {unit['repository'].split('/', 1)[1]}",
+        "body": body,
+        "risk": unit["risk"],
+        "changes": unit["changes"],
+        "depends_on": unit["depends_on"],
+        "lock_path": unit["lock_path"],
+        "expected_candidate_changes": expected_changes,
+        "desired_lock": unit["desired_lock"],
+        "holon_plan": unit["holon_plan"],
+        "rollback": {
+            "mode": "new-reviewed-convergence-plan",
+            "previous_lock_present": unit["current_lock_present"],
+            "previous_lock": unit["current_lock"],
+            "holon_plan_id": unit["holon_plan"].get("plan_id")
+            if unit["holon_plan"]
+            else None,
+            "force": False,
+        },
+        "controls": {
+            "pause": "set the fleet manifest unit state to paused and regenerate the plan",
+            "retry": "reuse this exact proposal; a different candidate tree is refused",
+            "partial_adoption": "only units named by review.approved_units may be proposed",
+        },
+    }
+    proposal["proposal_digest"] = digest(_proposal_without_digest(proposal))
+    return proposal
+
+
+def validate_proposal(proposal: object) -> list[str]:
+    if not isinstance(proposal, dict):
+        return ["proposal must be an object"]
+    errors: list[str] = []
+    if proposal.get("schema") != PROPOSAL_SCHEMA:
+        errors.append(f"proposal.schema must be {PROPOSAL_SCHEMA}")
+    if proposal.get("proposal_digest") != digest(_proposal_without_digest(proposal)):
+        errors.append("proposal_digest does not match the canonical proposal")
+    if (
+        not isinstance(proposal.get("base_commit"), str)
+        or COMMIT_RE.fullmatch(proposal["base_commit"]) is None
+    ):
+        errors.append("proposal.base_commit is invalid")
+    try:
+        _branch(proposal.get("base_branch"), "proposal.base_branch")
+        _branch(proposal.get("head_branch"), "proposal.head_branch")
+    except ContractError as error:
+        errors.append(str(error))
+    review = proposal.get("review")
+    if not isinstance(review, dict):
+        errors.append("proposal.review must be an object")
+    else:
+        review_keys = {
+            "schema",
+            "plan_digest",
+            "reviewer",
+            "reviewed_at",
+            "decision",
+            "approved_units",
+            "completed_units",
+        }
+        if set(review) != review_keys:
+            errors.append("proposal review contains unsupported or missing fields")
+        if (
+            review.get("schema") != REVIEW_SCHEMA
+            or review.get("decision") != "approved"
+        ):
+            errors.append("proposal does not contain an approved convergence review")
+        if (
+            not isinstance(review.get("reviewer"), str)
+            or not review["reviewer"].strip()
+        ):
+            errors.append("proposal review requires a reviewer")
+        if review.get("plan_digest") != proposal.get("plan_digest"):
+            errors.append("proposal review does not bind the plan digest")
+        approved_units = review.get("approved_units")
+        completed_units = review.get("completed_units")
+        if not _unique_sorted_strings(approved_units):
+            errors.append(
+                "proposal review approved_units must be a unique sorted array"
+            )
+            approved_units = []
+        if not _unique_sorted_strings(completed_units):
+            errors.append(
+                "proposal review completed_units must be a unique sorted array"
+            )
+            completed_units = []
+        if proposal.get("unit_id") not in approved_units:
+            errors.append("proposal unit is not approved by the embedded review")
+        dependencies = proposal.get("depends_on")
+        if not _unique_sorted_strings(dependencies):
+            errors.append("proposal.depends_on must be a unique sorted array")
+        elif not set(dependencies) <= set(completed_units):
+            errors.append("proposal has incomplete dependencies")
+        try:
+            _timestamp(review.get("reviewed_at"), "proposal.review.reviewed_at")
+        except ContractError as error:
+            errors.append(str(error))
+    unit_id = proposal.get("unit_id")
+    repository = proposal.get("repository")
+    expected_proposal_id = (
+        f"pull-request:{digest({'plan': proposal.get('plan_digest'), 'unit': unit_id})}"
+    )
+    if proposal.get("proposal_id") != expected_proposal_id:
+        errors.append("proposal_id does not bind the plan and unit")
+    if isinstance(repository, str) and REPOSITORY_RE.fullmatch(repository):
+        seed = {
+            "repository": repository,
+            "current_lock": digest(proposal.get("rollback", {}).get("previous_lock")),
+            "desired_lock": digest(proposal.get("desired_lock")),
+            "represented_commit": proposal.get("base_commit"),
+        }
+        if unit_id != f"upgrade:{repository}:{digest(seed)[:12]}":
+            errors.append("proposal unit_id does not bind its lock and base inputs")
+        branch_slug = re.sub(r"[^a-z0-9-]+", "-", repository.split("/", 1)[1]).strip(
+            "-"
+        )
+        if (
+            proposal.get("head_branch")
+            != f"pace/{branch_slug}-{str(proposal.get('plan_digest'))[:12]}"
+        ):
+            errors.append("proposal head branch is not deterministic")
+    else:
+        errors.append("proposal.repository is invalid")
+    current_lock = proposal.get("rollback", {}).get("previous_lock")
+    if not isinstance(proposal.get("rollback", {}).get("previous_lock_present"), bool):
+        errors.append("proposal rollback must record whether the prior lock existed")
+    desired_lock = proposal.get("desired_lock")
+    if isinstance(current_lock, dict) and isinstance(desired_lock, dict):
+        try:
+            expected_drift = _lock_changes(current_lock, desired_lock)
+        except (AttributeError, KeyError, TypeError):
+            expected_drift = []
+            errors.append("proposal embedded locks are malformed")
+        if proposal.get("changes") != expected_drift:
+            errors.append("proposal changes do not match embedded lock drift")
+        expected_risk = max(
+            (change["risk"] for change in expected_drift),
+            key=RISK_ORDER.get,
+            default="low",
+        )
+        if proposal.get("risk") != expected_risk:
+            errors.append("proposal risk does not match embedded lock drift")
+    else:
+        errors.append("proposal must embed current and desired locks")
+    changes = proposal.get("expected_candidate_changes")
+    if not isinstance(changes, list) or not changes:
+        errors.append("proposal.expected_candidate_changes must be non-empty")
+    elif [item.get("path") for item in changes if isinstance(item, dict)] != sorted(
+        item.get("path") for item in changes if isinstance(item, dict)
+    ):
+        errors.append("proposal expected changes must be sorted")
+    elif isinstance(current_lock, dict) and isinstance(desired_lock, dict):
+        try:
+            lock_path = _relative_path(proposal.get("lock_path"), "proposal.lock_path")
+        except ContractError as error:
+            errors.append(str(error))
+            lock_path = proposal.get("lock_path")
+        expected_changes: list[dict[str, Any]] = [
+            {
+                "path": lock_path,
+                "action": "upsert",
+                "verification": "exact-json",
+                "desired_sha256": digest(desired_lock),
+                "previous_sha256": (
+                    digest(current_lock)
+                    if proposal.get("rollback", {}).get("previous_lock_present")
+                    else None
+                ),
+            }
+        ]
+        holon_plan = proposal.get("holon_plan")
+        holon_operations: list[dict[str, Any]] = []
+        if holon_plan is not None:
+            if isinstance(holon_plan, dict) and isinstance(repository, str):
+                holon_operations, holon_errors = _holon_operations(
+                    holon_plan, repository
+                )
+                errors.extend(holon_errors)
+            else:
+                errors.append("proposal.holon_plan is malformed")
+        for operation in holon_operations:
+            expected_changes.append(
+                {
+                    "path": operation["path"],
+                    "action": "delete" if operation["action"] == "delete" else "upsert",
+                    "verification": "sha256",
+                    "desired_sha256": operation["desired_sha256"],
+                    "previous_sha256": operation["previous_sha256"],
+                }
+            )
+        if holon_operations:
+            expected_changes.append(
+                {
+                    "path": ".holon/materialization-state.v1.json",
+                    "action": "upsert",
+                    "verification": "holon-state",
+                    "desired_sha256": None,
+                    "previous_sha256": None,
+                }
+            )
+        expected_changes.sort(key=lambda item: str(item["path"]))
+        if changes != expected_changes:
+            errors.append(
+                "proposal expected changes do not match its lock and Holon inputs"
+            )
+        for index, item in enumerate(changes):
+            if not isinstance(item, dict):
+                errors.append(f"proposal expected change {index} is malformed")
+                continue
+            try:
+                _relative_path(
+                    item.get("path"), f"proposal expected change {index}.path"
+                )
+            except ContractError as error:
+                errors.append(str(error))
+    return errors
+
+
+def _git_executable() -> str:
+    executable = shutil.which("git")
+    if executable is None:
+        raise ContractError("git is required to verify a candidate tree")
+    return executable
+
+
+def _git(arguments: Sequence[str], directory: Path) -> str:
+    result = subprocess.run(
+        [_git_executable(), *arguments],
+        cwd=directory,
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        raise ContractError(result.stderr.strip() or "git command failed")
+    return result.stdout.strip()
+
+
+def _candidate_changes(directory: Path) -> dict[str, str]:
+    tracked = subprocess.run(
+        [
+            _git_executable(),
+            "diff",
+            "HEAD",
+            "--name-status",
+            "--no-renames",
+            "-z",
+        ],
+        cwd=directory,
+        check=False,
+        capture_output=True,
+    )
+    if tracked.returncode != 0:
+        raise ContractError(tracked.stderr.decode("utf-8", errors="replace").strip())
+    fields = tracked.stdout.decode("utf-8").split("\0")
+    changes: dict[str, str] = {}
+    index = 0
+    while index + 1 < len(fields) and fields[index]:
+        status = fields[index]
+        path = fields[index + 1]
+        index += 2
+        if status not in {"A", "M", "D"}:
+            raise ContractError(f"unsupported candidate Git status {status}: {path}")
+        changes[path] = status
+    untracked = _git(["ls-files", "--others", "--exclude-standard", "-z"], directory)
+    for path in filter(None, untracked.split("\0")):
+        changes[path] = "A"
+    return changes
+
+
+def _base_blob(directory: Path, path: str) -> bytes | None:
+    result = subprocess.run(
+        [_git_executable(), "cat-file", "blob", f"HEAD:{path}"],
+        cwd=directory,
+        check=False,
+        capture_output=True,
+    )
+    if result.returncode == 0:
+        return result.stdout
+    return None
+
+
+def _verify_candidate(proposal: Mapping[str, Any], directory: Path) -> dict[str, str]:
+    if _git(["rev-parse", "--show-toplevel"], directory) != str(directory.resolve()):
+        raise ContractError("candidate tree must be the Git worktree root")
+    if _git(["rev-parse", "HEAD"], directory) != proposal["base_commit"]:
+        raise ContractError(
+            "candidate tree HEAD does not match the reviewed base commit"
+        )
+    actual = _candidate_changes(directory)
+    expected = {item["path"]: item for item in proposal["expected_candidate_changes"]}
+    if set(actual) != set(expected):
+        missing = sorted(set(expected) - set(actual))
+        extra = sorted(set(actual) - set(expected))
+        details = []
+        if missing:
+            details.append("missing " + ", ".join(missing))
+        if extra:
+            details.append("unexpected " + ", ".join(extra))
+        raise ContractError(
+            "candidate tree differs from the reviewed boundary: " + "; ".join(details)
+        )
+    for path, specification in expected.items():
+        target = directory / path
+        previous = _base_blob(directory, path)
+        previous_sha256 = specification["previous_sha256"]
+        if (
+            specification["verification"] != "holon-state"
+            and previous_sha256 is None
+            and previous is not None
+        ):
+            raise ContractError(f"reviewed base unexpectedly contains {path}")
+        if (
+            specification["verification"] != "holon-state"
+            and previous_sha256 is not None
+        ):
+            if previous is None:
+                raise ContractError(f"reviewed base is missing {path}")
+            if specification["verification"] == "exact-json":
+                try:
+                    previous_document = json.loads(
+                        previous.decode("utf-8"), object_pairs_hook=_strict_object
+                    )
+                except (UnicodeError, json.JSONDecodeError, DuplicateKeyError) as error:
+                    raise ContractError(
+                        f"reviewed base lock is not strict JSON: {path}"
+                    ) from error
+                actual_previous_sha256 = digest(previous_document)
+            else:
+                actual_previous_sha256 = hashlib.sha256(previous).hexdigest()
+            if actual_previous_sha256 != previous_sha256:
+                raise ContractError(f"reviewed base digest does not match {path}")
+        if specification["action"] == "delete":
+            if actual[path] != "D" or target.exists():
+                raise ContractError(f"candidate must delete {path}")
+            continue
+        if actual[path] == "D" or not target.is_file() or target.is_symlink():
+            raise ContractError(f"candidate must contain a regular file at {path}")
+        if specification["verification"] == "exact-json":
+            if load_json(target) != proposal["desired_lock"]:
+                raise ContractError(
+                    f"candidate lock does not match the reviewed desired lock: {path}"
+                )
+        elif specification["verification"] == "sha256":
+            if file_digest(target) != specification["desired_sha256"]:
+                raise ContractError(
+                    f"candidate digest does not match the Holon plan: {path}"
+                )
+        elif specification["verification"] == "holon-state":
+            state = load_json(target)
+            if state.get("schema_version") != "holon.materialization-state/v1":
+                raise ContractError("candidate Holon state schema is unsupported")
+            if state.get("plan_id") != proposal["rollback"]["holon_plan_id"]:
+                raise ContractError(
+                    "candidate Holon state does not match the reviewed plan"
+                )
+    return actual
+
+
+class GitHubClient:
+    """Small GitHub REST adapter used only after exact plan review."""
+
+    def __init__(self, token: str, api_url: str = "https://api.github.com") -> None:
+        parsed = urlparse(api_url)
+        if (
+            parsed.scheme != "https"
+            or not parsed.netloc
+            or parsed.username is not None
+            or parsed.password is not None
+            or parsed.path not in {"", "/"}
+            or parsed.query
+            or parsed.fragment
+        ):
+            raise ContractError(
+                "GitHub API URL must be an HTTPS origin without user info"
+            )
+        self.token = token
+        self.api_url = api_url.rstrip("/")
+
+    def request(
+        self, method: str, path: str, payload: object | None = None
+    ) -> tuple[int, Any]:
+        data = None
+        if payload is not None:
+            data = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+        request = Request(
+            self.api_url + path,
+            method=method,
+            data=data,
+            headers={
+                "Accept": "application/vnd.github+json",
+                "Authorization": f"Bearer {self.token}",
+                "Content-Type": "application/json",
+                "User-Agent": "egohygiene-pace",
+                "X-GitHub-Api-Version": "2022-11-28",
+            },
+        )
+        try:
+            with urlopen(request, timeout=30) as response:
+                body = response.read()
+                return response.status, json.loads(body) if body else None
+        except HTTPError as error:
+            body = error.read()
+            value = json.loads(body) if body else None
+            return error.code, value
+
+
+def _mode(directory: Path, path: str, status: str) -> str:
+    if status != "A":
+        entry = _git(["ls-files", "--stage", "--", path], directory)
+        if entry:
+            return entry.split()[0]
+    return "100755" if os.access(directory / path, os.X_OK) else "100644"
+
+
+def open_pull_request(
+    proposal: dict[str, Any], candidate: Path, client: GitHubClient
+) -> dict[str, Any]:
+    errors = validate_proposal(proposal)
+    if errors:
+        raise ContractError("invalid proposal: " + "; ".join(errors))
+    changes = _verify_candidate(proposal, candidate)
+    repository = proposal["repository"]
+    prefix = f"/repos/{repository}"
+    status, base_ref = client.request(
+        "GET", f"{prefix}/git/ref/heads/{proposal['base_branch']}"
+    )
+    if status != 200 or not isinstance(base_ref, dict):
+        raise ContractError("could not read the target base branch")
+    if base_ref.get("object", {}).get("sha") != proposal["base_commit"]:
+        raise ContractError(
+            "target base branch advanced; collect Observatory state and re-plan"
+        )
+    status, base_commit = client.request(
+        "GET", f"{prefix}/git/commits/{proposal['base_commit']}"
+    )
+    if status != 200 or not isinstance(base_commit, dict):
+        raise ContractError("could not read the target base commit")
+    tree_entries: list[dict[str, Any]] = []
+    for path in sorted(changes):
+        if changes[path] == "D":
+            tree_entries.append(
+                {"path": path, "mode": "100644", "type": "blob", "sha": None}
+            )
+            continue
+        content = base64.b64encode((candidate / path).read_bytes()).decode("ascii")
+        status, blob = client.request(
+            "POST", f"{prefix}/git/blobs", {"content": content, "encoding": "base64"}
+        )
+        if status != 201 or not isinstance(blob, dict) or not blob.get("sha"):
+            raise ContractError(f"GitHub refused candidate blob {path}")
+        tree_entries.append(
+            {
+                "path": path,
+                "mode": _mode(candidate, path, changes[path]),
+                "type": "blob",
+                "sha": blob["sha"],
+            }
+        )
+    status, tree = client.request(
+        "POST",
+        f"{prefix}/git/trees",
+        {"base_tree": base_commit["tree"]["sha"], "tree": tree_entries},
+    )
+    if status != 201 or not isinstance(tree, dict) or not tree.get("sha"):
+        raise ContractError("GitHub refused the bounded candidate tree")
+
+    status, existing_ref = client.request(
+        "GET", f"{prefix}/git/ref/heads/{proposal['head_branch']}"
+    )
+    commit_sha: str
+    if status == 200 and isinstance(existing_ref, dict):
+        commit_sha = existing_ref.get("object", {}).get("sha", "")
+        check_status, existing_commit = client.request(
+            "GET", f"{prefix}/git/commits/{commit_sha}"
+        )
+        if (
+            check_status != 200
+            or not isinstance(existing_commit, dict)
+            or existing_commit.get("tree", {}).get("sha") != tree["sha"]
+            or [parent.get("sha") for parent in existing_commit.get("parents", [])]
+            != [proposal["base_commit"]]
+        ):
+            raise ContractError(
+                "proposal branch exists with a different candidate tree"
+            )
+    elif status == 404:
+        identity = {
+            "name": "Ego Hygiene Pace",
+            "email": "pace@users.noreply.github.com",
+            "date": proposal["review"]["reviewed_at"],
+        }
+        commit_status, commit = client.request(
+            "POST",
+            f"{prefix}/git/commits",
+            {
+                "message": proposal["title"],
+                "tree": tree["sha"],
+                "parents": [proposal["base_commit"]],
+                "author": identity,
+                "committer": identity,
+            },
+        )
+        if (
+            commit_status != 201
+            or not isinstance(commit, dict)
+            or not commit.get("sha")
+        ):
+            raise ContractError("GitHub refused the reviewed candidate commit")
+        commit_sha = commit["sha"]
+        ref_status, _ = client.request(
+            "POST",
+            f"{prefix}/git/refs",
+            {"ref": f"refs/heads/{proposal['head_branch']}", "sha": commit_sha},
+        )
+        if ref_status != 201:
+            raise ContractError("GitHub refused the proposal branch")
+    else:
+        raise ContractError("could not inspect the proposal branch")
+
+    query = urlencode(
+        {
+            "state": "open",
+            "head": f"{repository.split('/', 1)[0]}:{proposal['head_branch']}",
+            "base": proposal["base_branch"],
+        }
+    )
+    status, pulls = client.request("GET", f"{prefix}/pulls?{query}")
+    if status != 200 or not isinstance(pulls, list):
+        raise ContractError("could not inspect existing pull requests")
+    if pulls:
+        pull = pulls[0]
+    else:
+        status, pull = client.request(
+            "POST",
+            f"{prefix}/pulls",
+            {
+                "title": proposal["title"],
+                "head": proposal["head_branch"],
+                "base": proposal["base_branch"],
+                "body": proposal["body"],
+                "maintainer_can_modify": True,
+            },
+        )
+        if status != 201 or not isinstance(pull, dict):
+            raise ContractError("GitHub refused the bounded pull request")
+    return {
+        "repository": repository,
+        "number": pull.get("number"),
+        "url": pull.get("html_url"),
+        "head_branch": proposal["head_branch"],
+        "commit": commit_sha,
+        "proposal_digest": proposal["proposal_digest"],
+    }
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    plan = subparsers.add_parser("plan", help="generate a no-write fleet plan")
+    plan.add_argument("--manifest", type=Path, required=True)
+    plan.add_argument("--catalog", type=Path, required=True)
+    plan.add_argument("--observatory", type=Path, required=True)
+    plan.add_argument("--output", type=Path, required=True)
+
+    verify = subparsers.add_parser("verify-plan", help="verify a generated plan")
+    verify.add_argument("--plan", type=Path, required=True)
+
+    propose = subparsers.add_parser("propose", help="prepare one reviewed PR request")
+    propose.add_argument("--plan", type=Path, required=True)
+    propose.add_argument("--review", type=Path, required=True)
+    propose.add_argument("--unit", required=True)
+    propose.add_argument("--output", type=Path, required=True)
+
+    open_pr = subparsers.add_parser("open-pr", help="open exactly one reviewed PR")
+    open_pr.add_argument("--proposal", type=Path, required=True)
+    open_pr.add_argument("--candidate", type=Path, required=True)
+    open_pr.add_argument("--token-env", default="GITHUB_TOKEN")
+    open_pr.add_argument("--api-url", default="https://api.github.com")
+    open_pr.add_argument("--output", type=Path)
+    return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    arguments = build_parser().parse_args(argv)
+    try:
+        if arguments.command == "plan":
+            plan = build_plan(
+                arguments.manifest, arguments.catalog, arguments.observatory
+            )
+            write_json(arguments.output, plan)
+            print(
+                f"WROTE no-write convergence plan {arguments.output} ({plan['plan_digest']})"
+            )
+            return 0
+        if arguments.command == "verify-plan":
+            errors = validate_plan(load_json(arguments.plan))
+            if errors:
+                raise ContractError("; ".join(errors))
+            print(f"VALID convergence plan {arguments.plan}")
+            return 0
+        if arguments.command == "propose":
+            proposal = build_proposal(
+                load_json(arguments.plan), load_json(arguments.review), arguments.unit
+            )
+            write_json(arguments.output, proposal)
+            print(f"WROTE one-unit pull-request proposal {arguments.output}")
+            return 0
+        if arguments.command == "open-pr":
+            token = os.environ.get(arguments.token_env)
+            if not token:
+                raise ContractError(
+                    f"token environment variable is unset: {arguments.token_env}"
+                )
+            proposal = load_json(arguments.proposal)
+            result = open_pull_request(
+                proposal,
+                arguments.candidate.resolve(),
+                GitHubClient(token, arguments.api_url),
+            )
+            if arguments.output:
+                write_json(arguments.output, result)
+            print(f"OPENED pull request {result['url']}")
+            return 0
+    except (
+        ContractError,
+        DuplicateKeyError,
+        OSError,
+        UnicodeError,
+        json.JSONDecodeError,
+    ) as error:
+        print(f"REFUSED: {error}", file=sys.stderr)
+        return 1
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_fleet_convergence.py b/tests/test_fleet_convergence.py
new file mode 100644
index 0000000..c178d29
--- /dev/null
+++ b/tests/test_fleet_convergence.py
@@ -0,0 +1,475 @@
+# Copyright 2026 Ego Hygiene
+# SPDX-License-Identifier: MIT
+
+"""Adversarial evidence for reviewed fleet convergence."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import importlib.util
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "scripts" / "plan_fleet_convergence.py"
+EXAMPLES = ROOT / "examples" / "convergence"
+SPEC = importlib.util.spec_from_file_location("plan_fleet_convergence", SCRIPT)
+assert SPEC is not None
+assert SPEC.loader is not None
+planner = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(planner)
+
+
+def fixture_plan() -> dict[str, object]:
+    return planner.build_plan(
+        EXAMPLES / "fleet.manifest.json",
+        EXAMPLES / "catalog.json",
+        EXAMPLES / "observatory.snapshot.json",
+    )
+
+
+def review(
+    plan: dict[str, object], *, approved: list[str] | None = None
+) -> dict[str, object]:
+    unit_id = plan["units"][0]["unit_id"]
+    return {
+        "schema": planner.REVIEW_SCHEMA,
+        "plan_digest": plan["plan_digest"],
+        "reviewer": "fixture-reviewer",
+        "reviewed_at": "2026-09-02T13:00:00Z",
+        "decision": "approved",
+        "approved_units": approved if approved is not None else [unit_id],
+        "completed_units": [],
+    }
+
+
+def run_git(directory: Path, *arguments: str) -> str:
+    result = subprocess.run(
+        ["git", *arguments],
+        cwd=directory,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return result.stdout.strip()
+
+
+class FakeGitHubClient:
+    """Record the exact GitHub write boundary without network access."""
+
+    def __init__(self, base_commit: str) -> None:
+        self.base_commit = base_commit
+        self.branch_commit: str | None = None
+        self.branch_tree: str | None = None
+        self.pull: dict[str, object] | None = None
+        self.calls: list[tuple[str, str, object | None]] = []
+
+    def request(self, method: str, path: str, payload: object | None = None):
+        self.calls.append((method, path, payload))
+        if method == "GET" and path.endswith("/git/ref/heads/main"):
+            return 200, {"object": {"sha": self.base_commit}}
+        if method == "GET" and f"/git/commits/{self.base_commit}" in path:
+            return 200, {"tree": {"sha": "base-tree"}}
+        if method == "POST" and path.endswith("/git/blobs"):
+            return 201, {"sha": f"blob-{len(self.calls)}"}
+        if method == "POST" and path.endswith("/git/trees"):
+            self.branch_tree = "candidate-tree"
+            return 201, {"sha": self.branch_tree}
+        if method == "GET" and "/git/ref/heads/pace/" in path:
+            if self.branch_commit is None:
+                return 404, {"message": "Not Found"}
+            return 200, {"object": {"sha": self.branch_commit}}
+        if (
+            method == "GET"
+            and self.branch_commit
+            and path.endswith(f"/git/commits/{self.branch_commit}")
+        ):
+            return 200, {
+                "tree": {"sha": self.branch_tree},
+                "parents": [{"sha": self.base_commit}],
+            }
+        if method == "POST" and path.endswith("/git/commits"):
+            self.branch_commit = "1" * 40
+            return 201, {"sha": self.branch_commit}
+        if method == "POST" and path.endswith("/git/refs"):
+            return 201, {"ref": payload["ref"]}
+        if method == "GET" and "/pulls?" in path:
+            return 200, [self.pull] if self.pull else []
+        if method == "POST" and path.endswith("/pulls"):
+            self.pull = {
+                "number": 42,
+                "html_url": "https://github.com/egohygiene/example-tool/pull/42",
+            }
+            return 201, self.pull
+        raise AssertionError(f"unexpected fake GitHub call: {method} {path}")
+
+
+class FleetConvergenceTests(unittest.TestCase):
+    """Protect determinism, review, bounded writes, and recovery controls."""
+
+    def test_fixture_plan_is_deterministic_and_explainable(self) -> None:
+        first = fixture_plan()
+        second = fixture_plan()
+
+        self.assertEqual(first, second)
+        self.assertEqual(planner.validate_plan(first), [])
+        self.assertFalse(first["review"]["approved"])
+        self.assertEqual(first["summary"]["dispositions"], {"ready_for_review": 1})
+        unit = first["units"][0]
+        self.assertEqual(unit["risk"], "medium")
+        self.assertEqual(unit["represented_commit"], "f" * 40)
+        self.assertEqual(unit["changes"][0]["type"], "update")
+        self.assertIn("immutable source provenance", unit["changes"][0]["reasons"][0])
+
+    def test_observatory_stale_or_ambiguous_state_blocks_instead_of_guessing(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory(prefix="pace-observation-") as temporary:
+            target = Path(temporary)
+            for source in EXAMPLES.iterdir():
+                if source.is_file():
+                    (target / source.name).write_bytes(source.read_bytes())
+            observation_path = target / "observatory.snapshot.json"
+            observation = json.loads(observation_path.read_text(encoding="utf-8"))
+            evidence = observation["repositories"][0]["evidence"]
+            evidence["freshness"] = "stale"
+            evidence["represented_commits"].append("0" * 40)
+            observation_path.write_text(json.dumps(observation), encoding="utf-8")
+
+            plan = planner.build_plan(
+                target / "fleet.manifest.json",
+                target / "catalog.json",
+                observation_path,
+            )
+
+        unit = plan["units"][0]
+        self.assertEqual(unit["disposition"], "blocked")
+        self.assertIsNone(unit["represented_commit"])
+        self.assertTrue(any("freshness is stale" in item for item in unit["blockers"]))
+        self.assertTrue(any("exactly one" in item for item in unit["blockers"]))
+
+    def test_pause_is_manifest_state_and_supersedes_the_plan(self) -> None:
+        original = fixture_plan()
+        with tempfile.TemporaryDirectory(prefix="pace-pause-") as temporary:
+            target = Path(temporary)
+            for source in EXAMPLES.iterdir():
+                if source.is_file():
+                    (target / source.name).write_bytes(source.read_bytes())
+            manifest_path = target / "fleet.manifest.json"
+            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+            manifest["repositories"][0]["state"] = "paused"
+            manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+            paused = planner.build_plan(
+                manifest_path,
+                target / "catalog.json",
+                target / "observatory.snapshot.json",
+            )
+
+        self.assertEqual(paused["units"][0]["disposition"], "paused")
+        self.assertNotEqual(paused["plan_digest"], original["plan_digest"])
+
+    def test_dependency_order_is_explicit_and_cycles_fail(self) -> None:
+        records = [
+            {"repository": "egohygiene/b", "depends_on": ["egohygiene/a"]},
+            {"repository": "egohygiene/a", "depends_on": []},
+        ]
+        self.assertEqual(
+            planner._topological_repositories(records),
+            ["egohygiene/a", "egohygiene/b"],
+        )
+        records[1]["depends_on"] = ["egohygiene/b"]
+        with self.assertRaisesRegex(planner.ContractError, "dependency cycle"):
+            planner._topological_repositories(records)
+
+    def test_generated_target_drift_requires_exact_holon_plan(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="pace-holon-") as temporary:
+            target = Path(temporary)
+            for source in EXAMPLES.iterdir():
+                if source.is_file():
+                    (target / source.name).write_bytes(source.read_bytes())
+            for name in (
+                "example-tool.current.lock.json",
+                "example-tool.desired.lock.json",
+            ):
+                path = target / name
+                document = json.loads(path.read_text(encoding="utf-8"))
+                lock = document["locks"][0]
+                lock["target"]["management"] = "generated"
+                lock["target"]["generator"] = "egohygiene/holon"
+                lock["compatibility"]["rollback"] = "restore-lock-and-generated-targets"
+                path.write_text(json.dumps(document), encoding="utf-8")
+            plan = planner.build_plan(
+                target / "fleet.manifest.json",
+                target / "catalog.json",
+                target / "observatory.snapshot.json",
+            )
+
+        unit = plan["units"][0]
+        self.assertEqual(unit["disposition"], "blocked")
+        self.assertTrue(any("exact Holon" in item for item in unit["blockers"]))
+
+    def test_exact_holon_plan_unblocks_only_the_generated_target(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="pace-holon-ready-") as temporary:
+            target = Path(temporary)
+            for source in EXAMPLES.iterdir():
+                if source.is_file():
+                    (target / source.name).write_bytes(source.read_bytes())
+            for name in (
+                "example-tool.current.lock.json",
+                "example-tool.desired.lock.json",
+            ):
+                path = target / name
+                document = json.loads(path.read_text(encoding="utf-8"))
+                lock = document["locks"][0]
+                lock["target"]["management"] = "generated"
+                lock["target"]["generator"] = "egohygiene/holon"
+                lock["compatibility"]["rollback"] = "restore-lock-and-generated-targets"
+                path.write_text(json.dumps(document), encoding="utf-8")
+            old_bytes = b"old schema\n"
+            new_bytes = b"new schema\n"
+            holon_plan = {
+                "schema_version": planner.HOLON_PLAN_SCHEMA,
+                "engine_version": "1.0.0",
+                "repository": "egohygiene/example-tool",
+                "resolved_manifest_sha256": "1" * 64,
+                "resolved_manifest": {},
+                "inputs": {"render_source": None, "aether": None},
+                "capability_adapters": [],
+                "operations": [
+                    {
+                        "action": "update",
+                        "path": "schemas/repository-report.schema.json",
+                        "owner": "egohygiene/holon",
+                        "source": "fixture",
+                        "desired_sha256": hashlib.sha256(new_bytes).hexdigest(),
+                        "previous_sha256": hashlib.sha256(old_bytes).hexdigest(),
+                        "reason": "rendered source changed",
+                    }
+                ],
+                "summary": {"update": 1},
+            }
+            holon_plan["plan_id"] = planner.digest(holon_plan)
+            (target / "example-tool.holon.plan.json").write_text(
+                json.dumps(holon_plan), encoding="utf-8"
+            )
+            manifest_path = target / "fleet.manifest.json"
+            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+            manifest["repositories"][0]["holon_plan"] = "example-tool.holon.plan.json"
+            manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+
+            plan = planner.build_plan(
+                manifest_path,
+                target / "catalog.json",
+                target / "observatory.snapshot.json",
+            )
+            unit = plan["units"][0]
+            proposal = planner.build_proposal(plan, review(plan), unit["unit_id"])
+
+        self.assertEqual(unit["disposition"], "ready_for_review")
+        self.assertEqual(
+            [item["path"] for item in proposal["expected_candidate_changes"]],
+            [
+                ".config/pace.lock.json",
+                ".holon/materialization-state.v1.json",
+                "schemas/repository-report.schema.json",
+            ],
+        )
+
+    def test_lock_policy_drift_is_not_mistaken_for_no_change(self) -> None:
+        current = json.loads(
+            (EXAMPLES / "example-tool.current.lock.json").read_text(encoding="utf-8")
+        )
+        desired = copy.deepcopy(current)
+        desired["policy"]["max_exception_days"] = 14
+
+        changes = planner._lock_changes(current, desired)
+
+        self.assertEqual(len(changes), 1)
+        self.assertEqual(changes[0]["id"], "lock-policy")
+        self.assertEqual(changes[0]["risk"], "medium")
+
+    def test_review_can_approve_only_a_partial_fleet_subset(self) -> None:
+        plan = fixture_plan()
+        unit_id = plan["units"][0]["unit_id"]
+        denied = review(plan, approved=[])
+        with self.assertRaisesRegex(planner.ContractError, "does not approve"):
+            planner.build_proposal(plan, denied, unit_id)
+
+        proposal = planner.build_proposal(plan, review(plan), unit_id)
+        self.assertEqual(planner.validate_proposal(proposal), [])
+        self.assertEqual(proposal["unit_id"], unit_id)
+        self.assertEqual(len(proposal["expected_candidate_changes"]), 1)
+        self.assertEqual(
+            proposal["rollback"]["previous_lock"], plan["units"][0]["current_lock"]
+        )
+        self.assertFalse(proposal["rollback"]["force"])
+
+    def test_tampered_plan_and_review_are_rejected(self) -> None:
+        plan = fixture_plan()
+        plan["units"][0]["risk"] = "low"
+        self.assertIn("plan_digest", " ".join(planner.validate_plan(plan)))
+
+        clean = fixture_plan()
+        bad_review = review(clean)
+        bad_review["plan_digest"] = "0" * 64
+        with self.assertRaisesRegex(planner.ContractError, "exact plan digest"):
+            planner.build_proposal(clean, bad_review, clean["units"][0]["unit_id"])
+
+    def _candidate(self, proposal: dict[str, object], directory: Path) -> None:
+        run_git(directory, "init", "--initial-branch=main")
+        run_git(directory, "config", "user.name", "Pace Tests")
+        run_git(directory, "config", "user.email", "pace-tests@example.invalid")
+        lock_path = directory / ".config" / "pace.lock.json"
+        lock_path.parent.mkdir(parents=True)
+        lock_path.write_text(
+            json.dumps(proposal["rollback"]["previous_lock"], indent=2, sort_keys=True)
+            + "\n",
+            encoding="utf-8",
+        )
+        run_git(directory, "add", ".config/pace.lock.json")
+        run_git(directory, "commit", "-m", "base")
+        proposal["base_commit"] = run_git(directory, "rev-parse", "HEAD")
+        unit_seed = {
+            "repository": proposal["repository"],
+            "current_lock": planner.digest(proposal["rollback"]["previous_lock"]),
+            "desired_lock": planner.digest(proposal["desired_lock"]),
+            "represented_commit": proposal["base_commit"],
+        }
+        proposal["unit_id"] = (
+            f"upgrade:{proposal['repository']}:{planner.digest(unit_seed)[:12]}"
+        )
+        proposal["review"]["approved_units"] = [proposal["unit_id"]]
+        proposal["proposal_id"] = "pull-request:" + planner.digest(
+            {"plan": proposal["plan_digest"], "unit": proposal["unit_id"]}
+        )
+        proposal["proposal_digest"] = planner.digest(
+            planner._proposal_without_digest(proposal)
+        )
+        lock_path.write_text(
+            json.dumps(proposal["desired_lock"], indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+
+    def test_candidate_tree_is_exactly_bounded(self) -> None:
+        plan = fixture_plan()
+        proposal = planner.build_proposal(
+            plan, review(plan), plan["units"][0]["unit_id"]
+        )
+        with tempfile.TemporaryDirectory(prefix="pace-candidate-") as temporary:
+            candidate = Path(temporary)
+            self._candidate(proposal, candidate)
+            changes = planner._verify_candidate(proposal, candidate)
+            self.assertEqual(changes, {".config/pace.lock.json": "M"})
+            (candidate / "README.md").write_text("unexpected\n", encoding="utf-8")
+            with self.assertRaisesRegex(planner.ContractError, "unexpected README.md"):
+                planner._verify_candidate(proposal, candidate)
+
+    def test_github_adapter_opens_one_pr_and_retry_is_idempotent(self) -> None:
+        plan = fixture_plan()
+        proposal = planner.build_proposal(
+            plan, review(plan), plan["units"][0]["unit_id"]
+        )
+        with tempfile.TemporaryDirectory(prefix="pace-open-pr-") as temporary:
+            candidate = Path(temporary)
+            self._candidate(proposal, candidate)
+            client = FakeGitHubClient(proposal["base_commit"])
+            first = planner.open_pull_request(proposal, candidate, client)
+            second = planner.open_pull_request(proposal, candidate, client)
+
+        self.assertEqual(first, second)
+        self.assertEqual(first["number"], 42)
+        pull_creates = [
+            call
+            for call in client.calls
+            if call[0] == "POST" and call[1].endswith("/pulls")
+        ]
+        ref_creates = [
+            call
+            for call in client.calls
+            if call[0] == "POST" and call[1].endswith("/git/refs")
+        ]
+        self.assertEqual(len(pull_creates), 1)
+        self.assertEqual(len(ref_creates), 1)
+        self.assertTrue(
+            all(
+                "git/refs/heads/main" not in call[1]
+                for call in client.calls
+                if call[0] == "POST"
+            )
+        )
+
+    def test_github_adapter_never_sends_tokens_to_an_unsafe_origin(self) -> None:
+        for url in (
+            "http://api.github.com",
+            "https://token@example.invalid",
+            "https://example.invalid/unexpected/path",
+        ):
+            with (
+                self.subTest(url=url),
+                self.assertRaisesRegex(planner.ContractError, "HTTPS origin"),
+            ):
+                planner.GitHubClient("secret", url)
+
+    def test_cli_plans_verifies_and_proposes_without_credentials(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="pace-cli-") as temporary:
+            target = Path(temporary)
+            plan_path = target / "plan.json"
+            create = subprocess.run(
+                [
+                    sys.executable,
+                    str(SCRIPT),
+                    "plan",
+                    "--manifest",
+                    str(EXAMPLES / "fleet.manifest.json"),
+                    "--catalog",
+                    str(EXAMPLES / "catalog.json"),
+                    "--observatory",
+                    str(EXAMPLES / "observatory.snapshot.json"),
+                    "--output",
+                    str(plan_path),
+                ],
+                check=False,
+                capture_output=True,
+                text=True,
+            )
+            self.assertEqual(create.returncode, 0, create.stderr)
+            verify = subprocess.run(
+                [sys.executable, str(SCRIPT), "verify-plan", "--plan", str(plan_path)],
+                check=False,
+                capture_output=True,
+                text=True,
+            )
+            self.assertEqual(verify.returncode, 0, verify.stderr)
+            plan = json.loads(plan_path.read_text(encoding="utf-8"))
+            review_path = target / "review.json"
+            review_path.write_text(json.dumps(review(plan)), encoding="utf-8")
+            proposal_path = target / "proposal.json"
+            propose = subprocess.run(
+                [
+                    sys.executable,
+                    str(SCRIPT),
+                    "propose",
+                    "--plan",
+                    str(plan_path),
+                    "--review",
+                    str(review_path),
+                    "--unit",
+                    plan["units"][0]["unit_id"],
+                    "--output",
+                    str(proposal_path),
+                ],
+                check=False,
+                capture_output=True,
+                text=True,
+            )
+            self.assertEqual(propose.returncode, 0, propose.stderr)
+            self.assertTrue(proposal_path.is_file())
+
+
+if __name__ == "__main__":
+    unittest.main()