coalesce safe identifier task duplicates

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

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()
         ),