diff --git a/api/services/work_record_identifier_migration.py b/api/services/work_record_identifier_migration.py index fcc9319..f0d2a4d 100644 --- a/api/services/work_record_identifier_migration.py +++ b/api/services/work_record_identifier_migration.py @@ -338,6 +338,62 @@ def _same_task(row: Task, expected: dict[str, Any], workplan_id: uuid.UUID) -> b ) +_TASK_DUPLICATE_FIELDS = ( + "workplan_id", + "title", + "description", + "status", + "priority", + "assignee", + "due_date", + "blocking_reason", + "needs_human", + "intervention_note", + "parent_task_id", +) + + +def _same_duplicate_task( + legacy: Task, + derived: Task, + record_id: str, +) -> bool: + """Accept only a deterministic duplicate of an otherwise identical legacy row.""" + return ( + legacy.record_id in {None, record_id} + and derived.record_id == record_id + and all( + getattr(legacy, field) == getattr(derived, field) + for field in _TASK_DUPLICATE_FIELDS + ) + ) + + +async def _task_reference_count(session: AsyncSession, task_id: uuid.UUID) -> int: + """Count every inbound task reference before duplicate coalescence. + + Coalescence deliberately refuses referenced legacy rows. Updating references + could collide with relationships already attached to the deterministic row; + a zero-reference proof keeps the repair lossless and reversible. + """ + return int( + await session.scalar( + text( + "SELECT " + "(SELECT count(*) FROM token_events WHERE task_id = :task_id) + " + "(SELECT count(*) FROM tasks WHERE parent_task_id = :task_id) + " + "(SELECT count(*) FROM review_contracts WHERE task_id = :task_id) + " + "(SELECT count(*) FROM progress_events WHERE task_id = :task_id) + " + "(SELECT count(*) FROM capability_requests WHERE blocking_task_id = :task_id) + " + "(SELECT count(*) FROM workplan_dependencies WHERE to_task_id = :task_id) + " + "(SELECT count(*) FROM suggestions WHERE promoted_task_id = :task_id)" + ), + {"task_id": task_id}, + ) + or 0 + ) + + async def repair_absent_prederivation_projection( session: AsyncSession, plan: dict[str, Any], @@ -511,9 +567,34 @@ async def _assert_projection_preconditions( continue if source is not None and target is not None: - raise IdentifierMigrationError( - f"both legacy and derived {mapping['kind']} rows exist for {mapping['record_id']}" + if mapping["kind"] != "task": + raise IdentifierMigrationError( + f"both legacy and derived {mapping['kind']} rows exist for {mapping['record_id']}" + ) + legacy_task = await session.scalar( + select(Task).where(Task.id == source_id).with_for_update() ) + derived_task = await session.scalar( + select(Task).where(Task.id == target_id).with_for_update() + ) + if ( + legacy_task is None + or derived_task is None + or not _same_duplicate_task( + legacy_task, derived_task, mapping["record_id"] + ) + ): + raise IdentifierMigrationError( + f"legacy and derived task rows differ for {mapping['record_id']}" + ) + references = await _task_reference_count(session, source_id) + if references: + raise IdentifierMigrationError( + f"legacy duplicate task has {references} inbound references for " + f"{mapping['record_id']}" + ) + states[mapping["record_id"]] = "duplicate_target" + continue if source is None and target is None: raise IdentifierMigrationError( f"neither legacy nor derived {mapping['kind']} row exists for {mapping['record_id']}" @@ -593,6 +674,20 @@ async def apply_repository_identifier_migration( ) await session.flush() + for mapping in ( + item + for item in replacements + if projection_states[item["record_id"]] == "duplicate_target" + ): + result = await session.execute( + text("DELETE FROM tasks WHERE id = :old_id"), + {"old_id": mapping["old_id"]}, + ) + if result.rowcount != 1: + raise IdentifierMigrationError( + f"failed to coalesce duplicate {mapping['record_id']}" + ) + for kind in ("workplan", "task"): table = "workplans" if kind == "workplan" else "tasks" for mapping in ( @@ -621,7 +716,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()), + migrated=sum( + state in {"legacy_source", "duplicate_target"} + for state in projection_states.values() + ), already_derived=sum( state == "derived_target" for state in projection_states.values() ), diff --git a/tests/test_work_record_identifier_migration.py b/tests/test_work_record_identifier_migration.py index 5b6631f..80c443b 100644 --- a/tests/test_work_record_identifier_migration.py +++ b/tests/test_work_record_identifier_migration.py @@ -318,6 +318,111 @@ async def test_repository_migration_accepts_mixed_legacy_and_derived_projection( assert retried.already_derived == 2 +@pytest.mark.asyncio +async def test_repository_migration_coalesces_identical_unreferenced_task_duplicate( + test_engine, +): + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + ids = await _seed_projection(factory, "duplicate-repo") + plan = _sealed_plan("duplicate-repo", ids["workplan"], ids["task"]) + task_new = _derived("TEST-WP-0001-T01") + workplan_new = _derived("TEST-WP-0001") + + async with factory() as session: + await session.execute( + text("DELETE FROM workplan_dependencies WHERE to_task_id = :task_id"), + {"task_id": ids["task"]}, + ) + await session.execute( + text("DELETE FROM progress_events WHERE task_id = :task_id"), + {"task_id": ids["task"]}, + ) + await session.execute( + text("DELETE FROM token_events WHERE task_id = :task_id"), + {"task_id": ids["task"]}, + ) + await session.execute( + text("DELETE FROM tasks WHERE parent_task_id = :task_id"), + {"task_id": ids["task"]}, + ) + legacy = await session.get(Task, ids["task"]) + session.add( + Task( + id=task_new, + workplan_id=legacy.workplan_id, + record_id="TEST-WP-0001-T01", + title=legacy.title, + description=legacy.description, + status=legacy.status, + priority=legacy.priority, + assignee=legacy.assignee, + due_date=legacy.due_date, + blocking_reason=legacy.blocking_reason, + needs_human=legacy.needs_human, + intervention_note=legacy.intervention_note, + parent_task_id=legacy.parent_task_id, + ) + ) + await session.commit() + + async with factory() as session: + result = await apply_repository_identifier_migration( + session, plan, "duplicate-repo" + ) + assert result.migrated == 2 + assert result.already_derived == 0 + + async with factory() as session: + assert await session.get(Task, ids["task"]) is None + task = await session.get(Task, task_new) + assert task is not None + assert task.workplan_id == workplan_new + assert task.record_id == "TEST-WP-0001-T01" + + async with factory() as session: + await reverse_repository_identifier_migration(session, plan, "duplicate-repo") + async with factory() as session: + task = await session.get(Task, ids["task"]) + assert task is not None + assert task.record_id == "TEST-WP-0001-T01" + assert await session.get(Task, task_new) is None + + +@pytest.mark.asyncio +async def test_repository_migration_rejects_referenced_task_duplicate(test_engine): + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + ids = await _seed_projection(factory, "referenced-duplicate-repo") + plan = _sealed_plan( + "referenced-duplicate-repo", ids["workplan"], ids["task"] + ) + async with factory() as session: + legacy = await session.get(Task, ids["task"]) + session.add( + Task( + id=_derived("TEST-WP-0001-T01"), + workplan_id=legacy.workplan_id, + record_id="TEST-WP-0001-T01", + title=legacy.title, + description=legacy.description, + status=legacy.status, + priority=legacy.priority, + assignee=legacy.assignee, + due_date=legacy.due_date, + blocking_reason=legacy.blocking_reason, + needs_human=legacy.needs_human, + intervention_note=legacy.intervention_note, + parent_task_id=legacy.parent_task_id, + ) + ) + await session.commit() + + async with factory() as session: + with pytest.raises(IdentifierMigrationError, match="inbound references"): + await apply_repository_identifier_migration( + session, plan, "referenced-duplicate-repo" + ) + + @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)