diff --git a/api/routers/identifier_migrations.py b/api/routers/identifier_migrations.py index 25c0aac..95cd647 100644 --- a/api/routers/identifier_migrations.py +++ b/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, diff --git a/api/schemas/identifier_migration.py b/api/schemas/identifier_migration.py index a0556b6..8b06629 100644 --- a/api/schemas/identifier_migration.py +++ b/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 diff --git a/api/services/work_record_identifier_migration.py b/api/services/work_record_identifier_migration.py index a49c6da..fcc9319 100644 --- a/api/services/work_record_identifier_migration.py +++ b/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, ) diff --git a/tests/test_work_record_identifier_migration.py b/tests/test_work_record_identifier_migration.py index 2189fbc..5b6631f 100644 --- a/tests/test_work_record_identifier_migration.py +++ b/tests/test_work_record_identifier_migration.py @@ -204,6 +204,8 @@ async def test_repository_migration_cascades_and_reverses(test_engine): result = await apply_repository_identifier_migration(session, plan, "test-repo") assert result.direction == "forward" assert result.replacements == 2 + assert result.migrated == 2 + assert result.already_derived == 0 assert result.assignments_deferred == 1 async with factory() as session: @@ -277,6 +279,103 @@ async def test_repository_migration_cascades_and_reverses(test_engine): assert all(alias.reversed_at is not None for alias in aliases) +@pytest.mark.asyncio +async def test_repository_migration_accepts_mixed_legacy_and_derived_projection(test_engine): + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + ids = await _seed_projection(factory, "mixed-repo") + plan = _sealed_plan("mixed-repo", ids["workplan"], ids["task"]) + task_new = _derived("TEST-WP-0001-T01") + + # Forge reconciliation reached the task before the sealed repository + # migration. The parent remains legacy, so the repository is genuinely + # mixed rather than already migrated as one unit. + async with factory() as session: + await session.execute( + text("UPDATE tasks SET id = :new_id, record_id = :record_id WHERE id = :old_id"), + { + "old_id": ids["task"], + "new_id": task_new, + "record_id": "TEST-WP-0001-T01", + }, + ) + await session.commit() + + async with factory() as session: + result = await apply_repository_identifier_migration(session, plan, "mixed-repo") + assert result.replacements == 2 + assert result.migrated == 1 + assert result.already_derived == 1 + + async with factory() as session: + aliases = list((await session.execute(select(WorkRecordIdentifierAlias))).scalars()) + assert len(aliases) == 2 + assert all(alias.migration_status == "applied" for alias in aliases) + + # Retry is a verified idempotent convergence, not a conflicting target. + async with factory() as session: + retried = await apply_repository_identifier_migration(session, plan, "mixed-repo") + assert retried.migrated == 0 + assert retried.already_derived == 2 + + +@pytest.mark.asyncio +async def test_repository_migration_rejects_neither_legacy_nor_derived(test_engine): + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + ids = await _seed_projection(factory, "missing-repo") + plan = _sealed_plan( + "missing-repo", ids["workplan"], ids["task"], task_source_override=uuid.uuid4() + ) + + async with factory() as session: + with pytest.raises(IdentifierMigrationError, match="neither legacy nor derived task"): + await apply_repository_identifier_migration(session, plan, "missing-repo") + + async with factory() as session: + assert await session.scalar( + text("SELECT count(*) FROM work_record_identifier_aliases") + ) == 0 + + +@pytest.mark.asyncio +async def test_identifier_migration_http_apply_and_reverse(client, test_engine, monkeypatch): + from api.routers import identifier_migrations + + monkeypatch.setattr( + identifier_migrations.settings, "state_hub_instance_role", "primary" + ) + monkeypatch.setattr( + identifier_migrations.settings, "state_hub_instance_label", "railliance01" + ) + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + ids = await _seed_projection(factory, "http-migration-repo") + plan = _sealed_plan("http-migration-repo", ids["workplan"], ids["task"]) + body = { + "plan": plan, + "expected_plan_sha256": plan["plan_sha256"], + "primary_confirmed": True, + } + + response = await client.post( + "/identifier-migrations/repositories/http-migration-repo/apply", json=body + ) + assert response.status_code == 200, response.text + assert response.json()["result"] == { + "repo_slug": "http-migration-repo", + "plan_sha256": plan["plan_sha256"], + "direction": "forward", + "replacements": 2, + "migrated": 2, + "already_derived": 0, + "assignments_deferred": 1, + } + + response = await client.post( + "/identifier-migrations/repositories/http-migration-repo/reverse", json=body + ) + assert response.status_code == 200, response.text + assert response.json()["result"]["direction"] == "reverse" + + @pytest.mark.asyncio async def test_repository_migration_accepts_protected_prior_slug(test_engine): factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) @@ -340,7 +439,7 @@ async def test_repository_migration_is_atomic_when_a_source_is_missing(test_engi ) async with factory() as session: - with pytest.raises(IdentifierMigrationError, match="source task"): + with pytest.raises(IdentifierMigrationError, match="neither legacy nor derived task"): await apply_repository_identifier_migration(session, plan, "atomic-repo") async with factory() as session: diff --git a/workplans/STATE-WP-0083-forge-derived-projection-reset.md b/workplans/STATE-WP-0083-forge-derived-projection-reset.md index 66dfc44..bafa8cd 100644 --- a/workplans/STATE-WP-0083-forge-derived-projection-reset.md +++ b/workplans/STATE-WP-0083-forge-derived-projection-reset.md @@ -365,6 +365,17 @@ in the meantime. **Task-level reset remains blocked**, now on a narrower and better-understood problem: 35 identities, not 5974 unidentified rows. +**Mixed identifier convergence support (2026-08-31).** The fast forge path +exposed a second, separate consistency shape: one repository can contain legacy +UUID rows and already-derived UUID rows at the same time. State Hub's sealed +identifier transaction now classifies every mapping, migrates only verified +legacy sources, accepts already-derived targets only when their canonical +record identity and repository match, and records durable aliases for both. +Both-present and neither-present mappings fail the whole repository transaction. +The primary-only HTTP surface has explicit plan-hash confirmation and a reverse +operation for a failed file phase. This does not resolve the 35 duplicate +canonical task identities, so T02 remains `progress`. + ## Restore a migration mechanism for central @@ -724,4 +735,4 @@ displaced files moved to 0025-0027. The source numbering was checked before renaming; the target was not. **Remaining, and not solvable by retirement:** 9 private repositories central -cannot clone, and 9 identifier collisions that are identity decisions. \ No newline at end of file +cannot clone, and 9 identifier collisions that are identity decisions.