feat: converge mixed identifier projections

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

api/routers/identifier_migrations.py

@@ -4,18 +4,92 @@ from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.database import get_session
+from api.config import settings
 from api.schemas.identifier_migration import (
+    IdentifierMigrationApplySubmit,
     SealedProjectionRepairReceiptRead,
     SealedProjectionRepairSubmit,
 )
 from api.services.work_record_identifier_migration import (
     IdentifierMigrationError,
+    apply_repository_identifier_migration,
     repair_absent_prederivation_projection,
+    reverse_repository_identifier_migration,
 )
 
 router = APIRouter(prefix="/identifier-migrations", tags=["identifier-migrations"])
 
 
+@router.post("/repositories/{repo_slug}/apply")
+async def apply_identifier_migration(
+    repo_slug: str,
+    body: IdentifierMigrationApplySubmit,
+    session: AsyncSession = Depends(get_session),
+) -> dict:
+    """Apply one sealed repository-atomic identifier convergence transaction."""
+    if settings.state_hub_instance_role != "primary":
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "message": "identifier migration writes require the primary State Hub",
+                "instance_role": settings.state_hub_instance_role,
+                "instance_label": settings.state_hub_instance_label,
+            },
+        )
+    if not body.primary_confirmed:
+        raise HTTPException(status_code=409, detail="explicit primary confirmation is required")
+    if body.plan.get("plan_sha256") != body.expected_plan_sha256:
+        raise HTTPException(status_code=409, detail="explicit plan SHA-256 does not match sealed plan")
+    try:
+        result = await apply_repository_identifier_migration(session, body.plan, repo_slug)
+    except IdentifierMigrationError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    from api.routers.workstreams import _invalidate_workplan_index_cache
+
+    _invalidate_workplan_index_cache()
+    return {
+        "schema": "state-hub.identifier-migration-apply.v1",
+        "instance_role": settings.state_hub_instance_role,
+        "instance_label": settings.state_hub_instance_label,
+        "result": asdict(result),
+    }
+
+
+@router.post("/repositories/{repo_slug}/reverse")
+async def reverse_identifier_migration(
+    repo_slug: str,
+    body: IdentifierMigrationApplySubmit,
+    session: AsyncSession = Depends(get_session),
+) -> dict:
+    """Reverse one sealed repository migration after a failed file phase."""
+    if settings.state_hub_instance_role != "primary":
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "message": "identifier migration writes require the primary State Hub",
+                "instance_role": settings.state_hub_instance_role,
+                "instance_label": settings.state_hub_instance_label,
+            },
+        )
+    if not body.primary_confirmed:
+        raise HTTPException(status_code=409, detail="explicit primary confirmation is required")
+    if body.plan.get("plan_sha256") != body.expected_plan_sha256:
+        raise HTTPException(status_code=409, detail="explicit plan SHA-256 does not match sealed plan")
+    try:
+        result = await reverse_repository_identifier_migration(session, body.plan, repo_slug)
+    except IdentifierMigrationError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    from api.routers.workstreams import _invalidate_workplan_index_cache
+
+    _invalidate_workplan_index_cache()
+    return {
+        "schema": "state-hub.identifier-migration-reverse.v1",
+        "instance_role": settings.state_hub_instance_role,
+        "instance_label": settings.state_hub_instance_label,
+        "result": asdict(result),
+    }
+
+
 @router.post(
     "/sealed-projection-repairs",
     response_model=SealedProjectionRepairReceiptRead,

api/schemas/identifier_migration.py

@@ -31,3 +31,9 @@ class SealedProjectionRepairReceiptRead(BaseModel):
     source_fingerprint: str
     projection_identity: str
     observed_at: str
+
+
+class IdentifierMigrationApplySubmit(BaseModel):
+    plan: dict[str, Any]
+    expected_plan_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
+    primary_confirmed: bool

api/services/work_record_identifier_migration.py

@@ -35,6 +35,8 @@ class IdentifierMigrationResult:
     plan_sha256: str
     direction: str
     replacements: int
+    migrated: int
+    already_derived: int
     assignments_deferred: int
 
 
@@ -471,38 +473,66 @@ async def _assert_projection_preconditions(
     replacements: list[dict[str, Any]],
     *,
     reverse: bool,
-) -> None:
+) -> dict[str, str]:
+    states: dict[str, str] = {}
     for mapping in replacements:
         source_id = mapping["new_id"] if reverse else mapping["old_id"]
         target_id = mapping["old_id"] if reverse else mapping["new_id"]
         if mapping["kind"] == "workplan":
-            source_query = text(
-                "SELECT workplans.id FROM workplans "
-                "WHERE workplans.repo_id = :repository_id AND workplans.id = :record_id "
-                "FOR UPDATE"
+            query = text(
+                "SELECT workplans.id, workplans.repo_id, workplans.slug "
+                "FROM workplans WHERE workplans.id = :record_id FOR UPDATE"
             )
-            target_query = text("SELECT id FROM workplans WHERE id = :record_id")
         else:
-            source_query = text(
-                "SELECT tasks.id FROM tasks "
+            query = text(
+                "SELECT tasks.id, workplans.repo_id, tasks.record_id "
+                "FROM tasks "
                 "JOIN workplans ON workplans.id = tasks.workplan_id "
-                "WHERE workplans.repo_id = :repository_id AND tasks.id = :record_id "
-                "FOR UPDATE"
+                "WHERE tasks.id = :record_id FOR UPDATE"
             )
-            target_query = text("SELECT id FROM tasks WHERE id = :record_id")
-        source = await session.execute(
-            source_query,
-            {"repository_id": repository_id, "record_id": source_id},
-        )
-        if source.scalar_one_or_none() is None:
+        source = (await session.execute(query, {"record_id": source_id})).one_or_none()
+        target = (await session.execute(query, {"record_id": target_id})).one_or_none()
+        for label, row in (("source", source), ("target", target)):
+            if row is not None and row.repo_id != repository_id:
+                raise IdentifierMigrationError(
+                    f"{label} {mapping['kind']} {row.id} belongs to another repository"
+                )
+
+        if reverse:
+            if source is None:
+                raise IdentifierMigrationError(
+                    f"source {mapping['kind']} {source_id} is absent from repository {repo_slug}"
+                )
+            if target is not None:
+                raise IdentifierMigrationError(
+                    f"target {mapping['kind']} {target_id} already exists"
+                )
+            states[mapping["record_id"]] = "derived_source"
+            continue
+
+        if source is not None and target is not None:
             raise IdentifierMigrationError(
-                f"source {mapping['kind']} {source_id} is absent from repository {repo_slug}"
+                f"both legacy and derived {mapping['kind']} rows exist for {mapping['record_id']}"
             )
-        target = await session.execute(target_query, {"record_id": target_id})
-        if target.scalar_one_or_none() is not None:
+        if source is None and target is None:
             raise IdentifierMigrationError(
-                f"target {mapping['kind']} {target_id} already exists"
+                f"neither legacy nor derived {mapping['kind']} row exists for {mapping['record_id']}"
             )
+        if source is not None:
+            states[mapping["record_id"]] = "legacy_source"
+            continue
+
+        # A pre-existing derived target is safe only when it already represents
+        # the canonical record from this repository. This is the state produced
+        # when forge reconciliation reached a partial projection before the
+        # sealed file migration did.
+        identity = target.slug if mapping["kind"] == "workplan" else target.record_id
+        if not isinstance(identity, str) or identity.strip().lower() != mapping["record_id"].lower():
+            raise IdentifierMigrationError(
+                f"derived target identity mismatch for {mapping['record_id']}"
+            )
+        states[mapping["record_id"]] = "derived_target"
+    return states
 
 
 async def apply_repository_identifier_migration(
@@ -520,7 +550,7 @@ async def apply_repository_identifier_migration(
         resolution = await resolve_repository_slug(session, repo_slug, required=False)
         if resolution is None:
             raise IdentifierMigrationError(f"repository projection is absent: {repo_slug}")
-        await _assert_projection_preconditions(
+        projection_states = await _assert_projection_preconditions(
             session, resolution.repo.id, repo_slug, replacements, reverse=False
         )
         aliases = {
@@ -556,7 +586,7 @@ async def apply_repository_identifier_migration(
                 or alias.repo_slug != repo_slug
                 or alias.namespace != FLEET_NAMESPACE
                 or alias.plan_sha256 != plan_sha256
-                or alias.migration_status != "reversed"
+                or alias.migration_status not in {"reversed", "applied"}
             ):
                 raise IdentifierMigrationError(
                     f"conflicting durable alias for {mapping['old_id']}"
@@ -565,7 +595,12 @@ async def apply_repository_identifier_migration(
 
         for kind in ("workplan", "task"):
             table = "workplans" if kind == "workplan" else "tasks"
-            for mapping in (item for item in replacements if item["kind"] == kind):
+            for mapping in (
+                item
+                for item in replacements
+                if item["kind"] == kind
+                and projection_states[item["record_id"]] == "legacy_source"
+            ):
                 result = await session.execute(
                     text(f"UPDATE {table} SET id = :new_id WHERE id = :old_id"),
                     {"old_id": mapping["old_id"], "new_id": mapping["new_id"]},
@@ -586,6 +621,10 @@ async def apply_repository_identifier_migration(
         plan_sha256=plan_sha256,
         direction="forward",
         replacements=len(replacements),
+        migrated=sum(state == "legacy_source" for state in projection_states.values()),
+        already_derived=sum(
+            state == "derived_target" for state in projection_states.values()
+        ),
         assignments_deferred=assignments,
     )
 
@@ -656,5 +695,7 @@ async def reverse_repository_identifier_migration(
         plan_sha256=plan_sha256,
         direction="reverse",
         replacements=len(replacements),
+        migrated=len(replacements),
+        already_derived=0,
         assignments_deferred=assignments,
     )