feat: support mixed identifier convergence

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c

docs/work-record-uuid-derivation_v1.md

@@ -79,12 +79,28 @@ clean worktree, exactly match its configured upstream, and use a non-retired
 origin. The batch records these facts and its own SHA-256 seal, but always emits
 `apply_authorized: false`. An explicit decision must cite that batch hash before
 any database or file mutation. Repeat `--projection-api-base` for every hub in
-the cutover: each replacement must resolve its current UUID with HTTP 200 and
+the cutover. Each replacement is classified as `legacy_source`,
-its derived target with HTTP 404 on every named projection, or the manifest is
+`derived_target`, `both_present`, or `neither_present`. The first two may coexist
-not ready for approval. Saved projection endpoints are rechecked by
+inside one repository-atomic migration: State Hub migrates legacy rows, verifies
-`migration-batch-verify`; omitting the option retains the offline source/Git-only
+already-derived rows against their canonical record identity, and records the
-planning mode. A projection-aware batch containing UUID assignments fails
+same durable aliases for both. The latter two states are refused. Saved
-closed until an assignment-specific projection identity check is implemented.
+projection endpoints are rechecked by `migration-batch-verify`; omitting the
+option retains the offline source/Git-only planning mode. A projection-aware
+batch containing UUID assignments fails closed until an assignment-specific
+projection identity check is implemented.
+
+After approval, apply the sealed database phase and then the atomic file phase:
+
+```bash
+rmgr identifier migration-projection --plan PLAN.json --repo REPO \
+  --confirm-plan-sha256 SHA256 --api-base http://127.0.0.1:8000
+rmgr identifier migration-files --plan PLAN.json --repo REPO \
+  --confirm-plan-sha256 SHA256 --execute
+```
+
+If the file phase fails, run `migration-projection` with `--direction reverse`
+before changing the sealed plan or retrying. Routine `rmgr sync` never performs
+this migration implicitly.
 
 Activation and applying a bulk migration remain separate governed steps.
 Publishing or planning this function does not retroactively rewrite existing

src/repo_manager/cli.py

@@ -420,7 +420,10 @@ def main(argv: list[str] | None = None) -> int:
         action="append",
         default=[],
         dest="projection_api_bases",
-        help="Require current UUID=200 and derived UUID=404 on this projection",
+        help=(
+            "Classify legacy/derived UUID presence on this projection; both-present "
+            "and neither-present mappings are refused"
+        ),
     )
     p_id_batch.add_argument("--output", default=None)
     p_id_batch.add_argument("--force", action="store_true")
@@ -445,6 +448,18 @@ def main(argv: list[str] | None = None) -> int:
         action="store_true",
         help="Write files; without this flag only validate and report",
     )
+    p_id_projection = identifier_sub.add_parser(
+        "migration-projection",
+        help="Apply or reverse one sealed repository migration on the primary hub",
+    )
+    p_id_projection.add_argument("--plan", required=True)
+    p_id_projection.add_argument("--repo", required=True)
+    p_id_projection.add_argument("--confirm-plan-sha256", required=True)
+    p_id_projection.add_argument("--api-base", required=True)
+    p_id_projection.add_argument("--expected-instance-label", default="railliance01")
+    p_id_projection.add_argument(
+        "--direction", choices=["forward", "reverse"], default="forward"
+    )
 
     p_sbom = sub.add_parser(
         "sbom",
@@ -925,6 +940,7 @@ def main(argv: list[str] | None = None) -> int:
             derive_work_record_uuid,
             load_fleet_namespace,
             migrate_repository_identifier_files,
+            migrate_repository_projection,
             plan_identifier_migration,
             plan_identifier_migration_batch,
             scan_live_identifier_collisions,
@@ -972,6 +988,22 @@ def main(argv: list[str] | None = None) -> int:
             except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc:
                 print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
                 return 1
+        elif args.identifier_command == "migration-projection":
+            try:
+                plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))
+                if not isinstance(plan, dict):
+                    raise TypeError("migration plan must be a JSON object")
+                result = migrate_repository_projection(
+                    plan,
+                    repo_slug=args.repo,
+                    confirm_plan_sha256=args.confirm_plan_sha256,
+                    api_base=args.api_base,
+                    direction=args.direction,
+                    expected_instance_label=args.expected_instance_label,
+                )
+            except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc:
+                print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
+                return 1
         elif args.identifier_command == "migration-batch-plan":
             try:
                 plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))

src/repo_manager/identifiers.py

@@ -371,7 +371,13 @@ def _git_cutover_preflight(repo: Path, *, expected_head_sha: str | None) -> dict
 def _projection_migration_preflight(
     mappings: list[dict[str, Any]], api_bases: list[str]
 ) -> dict[str, Any]:
-    """Require every replacement source and no replacement target per projection."""
+    """Classify every replacement without assuming which side currently exists.
+
+    A projection may legitimately be mixed when forge reconciliation created
+    some deterministic targets before the sealed migration ran. Legacy-source
+    and derived-target are both convergent states; both-present and neither-
+    present are ambiguous and remain hard refusals.
+    """
     projections: list[dict[str, Any]] = []
     errors: list[dict[str, str]] = [
         {
@@ -409,12 +415,20 @@ def _projection_migration_preflight(
                     route = "workplans" if mapping.get("kind") == "workplan" else "tasks"
                     current = client.get(f"{api_base}/{route}/{mapping.get('current_uuid')}")
                     derived = client.get(f"{api_base}/{route}/{mapping.get('derived_uuid')}")
-                    check_ok = current.status_code == 200 and derived.status_code == 404
+                    status_pair = (current.status_code, derived.status_code)
+                    state = {
+                        (200, 404): "legacy_source",
+                        (404, 200): "derived_target",
+                        (200, 200): "both_present",
+                        (404, 404): "neither_present",
+                    }.get(status_pair, "unexpected_status")
+                    check_ok = state in {"legacy_source", "derived_target"}
                     check = {
                         "record_id": mapping.get("record_id"),
                         "kind": mapping.get("kind"),
                         "current_status": current.status_code,
                         "derived_status": derived.status_code,
+                        "state": state,
                         "ok": check_ok,
                     }
                     checks.append(check)
@@ -422,15 +436,43 @@ def _projection_migration_preflight(
                         errors.append(
                             {
                                 "scope": f"{api_base}:{mapping.get('record_id')}",
-                                "reason": "projection requires current UUID=200 and derived UUID=404",
+                                "reason": (
+                                    "projection is ambiguous: both legacy and derived UUIDs exist"
+                                    if state == "both_present"
+                                    else "projection is incomplete: neither legacy nor derived UUID exists"
+                                    if state == "neither_present"
+                                    else "projection returned an unexpected HTTP status"
+                                ),
                             }
                         )
         except httpx.HTTPError:
             errors.append({"scope": api_base, "reason": "projection API is unavailable"})
+        state_counts = {
+            state: sum(check["state"] == state for check in checks)
+            for state in (
+                "legacy_source",
+                "derived_target",
+                "both_present",
+                "neither_present",
+                "unexpected_status",
+            )
+        }
+        convergent_states = {check["state"] for check in checks if check["ok"]}
+        mode = (
+            "legacy_migration"
+            if convergent_states == {"legacy_source"}
+            else "file_convergence"
+            if convergent_states == {"derived_target"}
+            else "mixed_convergence"
+            if convergent_states == {"legacy_source", "derived_target"}
+            else "blocked"
+        )
         projections.append(
             {
                 "api_base": api_base,
                 "ok": all(check["ok"] for check in checks),
+                "mode": mode,
+                "state_counts": state_counts,
                 "replacement_checks": checks,
             }
         )
@@ -939,3 +981,103 @@ def migrate_repository_identifier_files(
         "assignments": assignments,
         "database_coordinated": False,
     }
+
+
+def migrate_repository_projection(
+    plan: dict[str, Any],
+    *,
+    repo_slug: str,
+    confirm_plan_sha256: str,
+    api_base: str,
+    direction: str = "forward",
+    expected_instance_label: str | None = "railliance01",
+    transport: httpx.BaseTransport | None = None,
+) -> dict[str, Any]:
+    """Apply or reverse the sealed projection half on the authoritative hub."""
+    plan_sha256 = _verified_plan_seal(plan)
+    if confirm_plan_sha256 != plan_sha256:
+        raise ValueError("--confirm-plan-sha256 does not match the sealed plan")
+    if direction not in {"forward", "reverse"}:
+        raise ValueError("direction must be 'forward' or 'reverse'")
+    verification = verify_identifier_migration_plan(plan, repo_slug=repo_slug)
+    if not verification["ok"]:
+        reasons = "; ".join(error["reason"] for error in verification["errors"])
+        raise ValueError(f"repository source preconditions failed: {reasons}")
+    repositories = [
+        item for item in plan.get("repositories", []) if item.get("repo") == repo_slug
+    ]
+    repository = repositories[0]
+    git_preflight = _git_cutover_preflight(
+        Path(str(repository.get("path") or "")).resolve(),
+        expected_head_sha=repository.get("planned_head_sha"),
+    )
+    if not git_preflight["ok"]:
+        raise ValueError(
+            "repository Git preconditions failed: " + "; ".join(git_preflight["errors"])
+        )
+    api_base = api_base.strip().rstrip("/")
+    try:
+        url = httpx.URL(api_base)
+        if (
+            url.scheme not in {"http", "https"}
+            or not url.host
+            or url.username
+            or url.password
+            or url.query
+            or url.fragment
+        ):
+            raise ValueError
+    except (TypeError, ValueError) as exc:
+        raise ValueError("projection API base must be a plain HTTP(S) origin") from exc
+
+    headers = {
+        "Idempotency-Key": f"rmgr-identifier-migration:{direction}:{repo_slug}:{plan_sha256}",
+        "X-StateHub-Source-Agent": "repo-manager",
+    }
+    try:
+        with httpx.Client(
+            base_url=api_base,
+            timeout=httpx.Timeout(120.0, connect=5.0),
+            follow_redirects=False,
+            transport=transport,
+        ) as client:
+            health = client.get("/state/health")
+            health.raise_for_status()
+            identity = health.json()
+            if identity.get("instance_role") != "primary" or (
+                expected_instance_label is not None
+                and identity.get("instance_label") != expected_instance_label
+            ):
+                raise ValueError("identifier migration requires the expected primary State Hub")
+            response = client.post(
+                f"/identifier-migrations/repositories/{repo_slug}/"
+                f"{'apply' if direction == 'forward' else 'reverse'}",
+                json={
+                    "plan": plan,
+                    "expected_plan_sha256": plan_sha256,
+                    "primary_confirmed": True,
+                },
+                headers=headers,
+            )
+            response.raise_for_status()
+    except httpx.HTTPStatusError as exc:
+        try:
+            detail = exc.response.json()
+        except ValueError:
+            detail = exc.response.text[:500]
+        raise ValueError(f"projection migration rejected: {detail}") from exc
+    except httpx.HTTPError as exc:
+        raise ValueError(f"projection migration unavailable: {exc}") from exc
+    receipt = response.json()
+    return {
+        "schema": "repo-manager.identifier-projection-migration.v1",
+        "ok": True,
+        "direction": direction,
+        "repo": repo_slug,
+        "plan_sha256": plan_sha256,
+        "api_base": api_base,
+        "instance": identity,
+        "source_verification": verification,
+        "git_preflight": git_preflight,
+        "receipt": receipt,
+    }

tests/test_identifiers.py

@@ -11,6 +11,7 @@ from repo_manager.identifiers import (
     derive_work_record_uuid,
     load_fleet_namespace,
     migrate_repository_identifier_files,
+    migrate_repository_projection,
     plan_identifier_migration,
     plan_identifier_migration_batch,
     scan_live_identifier_collisions,
@@ -38,6 +39,101 @@ def test_projection_preflight_rejects_unproven_assignment() -> None:
     ]
 
 
+def test_projection_preflight_accepts_mixed_convergent_states(monkeypatch) -> None:
+    mappings = [
+        {
+            "action": "replace",
+            "kind": "workplan",
+            "record_id": "ONE-WP-0001",
+            "current_uuid": "11111111-1111-4111-8111-111111111111",
+            "derived_uuid": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
+        },
+        {
+            "action": "replace",
+            "kind": "task",
+            "record_id": "ONE-WP-0001-T01",
+            "current_uuid": "22222222-2222-4222-8222-222222222222",
+            "derived_uuid": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
+        },
+    ]
+
+    def projection_get(self, url):
+        value = str(url)
+        present = value.endswith(
+            (mappings[0]["current_uuid"], mappings[1]["derived_uuid"])
+        )
+        return httpx.Response(200 if present else 404, request=httpx.Request("GET", url))
+
+    monkeypatch.setattr(httpx.Client, "get", projection_get)
+    result = _projection_migration_preflight(mappings, ["http://hub.test"])
+
+    assert result["ok"] is True
+    projection = result["projections"][0]
+    assert projection["mode"] == "mixed_convergence"
+    assert projection["state_counts"]["legacy_source"] == 1
+    assert projection["state_counts"]["derived_target"] == 1
+
+
+def test_projection_migration_client_requires_primary_and_exact_seal(tmp_path: Path) -> None:
+    repo = tmp_path / "one"
+    path = repo / "workplans" / "one.md"
+    _workplan(path, "ONE-WP-0001", "active")
+    subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
+    subprocess.run(["git", "add", "."], cwd=repo, check=True, capture_output=True)
+    subprocess.run(
+        [
+            "git",
+            "-c",
+            "user.name=Test",
+            "-c",
+            "user.email=test@example.com",
+            "commit",
+            "-m",
+            "seed",
+        ],
+        cwd=repo,
+        check=True,
+        capture_output=True,
+    )
+    _push_fixture_to_upstream(repo, tmp_path / "migration-remote.git")
+    plan = plan_identifier_migration(tmp_path, "helixforge")
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/state/health":
+            return httpx.Response(
+                200,
+                json={
+                    "status": "ok",
+                    "instance_role": "primary",
+                    "instance_label": "railliance01",
+                },
+            )
+        assert request.url.path == "/identifier-migrations/repositories/one/apply"
+        assert request.headers["idempotency-key"].endswith(plan["plan_sha256"])
+        return httpx.Response(
+            200, json={"schema": "state-hub.identifier-migration-apply.v1"}
+        )
+
+    result = migrate_repository_projection(
+        plan,
+        repo_slug="one",
+        confirm_plan_sha256=plan["plan_sha256"],
+        api_base="http://hub.test",
+        transport=httpx.MockTransport(handler),
+    )
+    assert result["ok"] is True
+    assert result["direction"] == "forward"
+
+    with pytest.raises(ValueError, match="confirm-plan-sha256"):
+        migrate_repository_projection(
+            plan,
+            repo_slug="one",
+            confirm_plan_sha256="0" * 64,
+            api_base="http://hub.test",
+            transport=httpx.MockTransport(handler),
+        )
+
+
 def _workplan(path: Path, identifier: str, status: str, task_status: str = "todo") -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     path.write_text(
@@ -335,7 +431,7 @@ def test_migration_batch_plan_rejects_projection_gap(tmp_path: Path, monkeypatch
     assert batch["ok"] is False
     assert batch["ready_for_approval"] is False
     assert batch["repositories"][0]["projection_preflight"]["ok"] is False
-    assert any("current UUID=200" in error["reason"] for error in batch["errors"])
+    assert any("neither legacy nor derived" in error["reason"] for error in batch["errors"])
 
 
 def test_migration_batch_plan_rejects_dirty_or_duplicate_scope(tmp_path: Path) -> None:

workplans/RMGR-WP-0005-registrar-consolidation-deterministic-ids.md

@@ -571,6 +571,22 @@ note are in
 `docs/evidence/RMGR-WP-0005-batch-0004-railiance-cluster-cutover-2026-08-22.md`.
 T04 remains in progress for separately sealed and approved fleet batches.
 
+**Mixed projection convergence implemented (2026-08-31).** Fast forge-derived
+reconciliation can legitimately reach a deterministic target before the sealed
+identifier cutover reaches its authoritative file. Projection preflight now
+classifies every replacement as `legacy_source`, `derived_target`,
+`both_present`, or `neither_present`. The first two may coexist within the same
+repository-atomic batch; the latter two remain refusals. State Hub migrates only
+legacy rows, verifies already-derived rows against repository and canonical
+record identity, and writes the same durable alias provenance for both. Repo
+Manager exposes the explicit primary-only database phase as `rmgr identifier
+migration-projection`; the existing `migration-files` command remains the
+atomic file phase, and the projection phase is reversible if file writeback
+fails. Focused State Hub and Repo Manager suites cover mixed convergence,
+idempotent retry, ambiguity refusal, HTTP apply/reverse, primary identity, and
+exact plan-seal checks. A fresh sealed State Hub pilot remains before this slice
+is operationally complete.
+
 **Batch 0005 preflight blocked safely (2026-08-22):** the refreshed zero-collision
 fleet plan covers 39 repositories and 215 live records. A projection-aware
 `adaptive-pricing` batch pinned its clean synchronized source and five UUID