fix: bound task identity ambiguity

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

api/services/forge_projection.py

@@ -446,13 +446,16 @@ class ProjectionDiff:
     missing: list[dict[str, Any]] = field(default_factory=list)   # forge has, hub lacks
     stale: list[dict[str, Any]] = field(default_factory=list)     # hub has, forge lacks
     differing: list[dict[str, Any]] = field(default_factory=list) # both, fields differ
+    # More than one row/file claims one canonical identity. Choosing a winner
+    # would turn a diagnostic into an implicit identity decision.
+    ambiguous: list[dict[str, Any]] = field(default_factory=list)
     # Set when the source could not support a claim of absence, so `stale` was
     # deliberately left empty rather than computed (`STATE-WP-0084-T01`).
     stale_withheld: str | None = None
 
     @property
     def clean(self) -> bool:
-        return not (self.missing or self.stale or self.differing)
+        return not (self.missing or self.stale or self.differing or self.ambiguous)
 
     @property
     def would_remove(self) -> int:
@@ -468,14 +471,49 @@ class ProjectionDiff:
                 "missing": len(self.missing),
                 "stale": len(self.stale),
                 "differing": len(self.differing),
+                "ambiguous": len(self.ambiguous),
             },
             "stale_withheld": self.stale_withheld,
             "missing": self.missing,
             "stale": self.stale,
             "differing": self.differing,
+            "ambiguous": self.ambiguous,
         }
 
 
+def _derived_identity_ambiguities(derived: DerivedProjection) -> list[dict[str, Any]]:
+    """Return duplicate canonical identities declared by one forge projection."""
+    workplans: dict[str, list[DerivedWorkplan]] = {}
+    tasks: dict[str, list[tuple[DerivedWorkplan, DerivedTask]]] = {}
+    for workplan in derived.workplans:
+        workplans.setdefault(workplan.record_id.strip().lower(), []).append(workplan)
+        for task in workplan.tasks:
+            tasks.setdefault(task.record_id.strip().lower(), []).append((workplan, task))
+
+    out: list[dict[str, Any]] = []
+    for rows in workplans.values():
+        if len(rows) > 1:
+            out.append(
+                {
+                    "kind": "workplan",
+                    "record_id": rows[0].record_id,
+                    "source": "forge",
+                    "paths": [row.relative_path for row in rows],
+                }
+            )
+    for rows in tasks.values():
+        if len(rows) > 1:
+            out.append(
+                {
+                    "kind": "task",
+                    "record_id": rows[0][1].record_id,
+                    "source": "forge",
+                    "workplans": [row[0].record_id for row in rows],
+                }
+            )
+    return out
+
+
 def diff_against_hub(
     derived: DerivedProjection,
     hub_workplans: list[dict[str, Any]],
@@ -487,6 +525,7 @@ def diff_against_hub(
     is testable without a database and cannot accidentally mutate anything.
     """
     d = ProjectionDiff(repo_slug=derived.repo_slug, commit=derived.commit)
+    d.ambiguous.extend(_derived_identity_ambiguities(derived))
     if not derived.retirement_eligible:
         # Compute what is missing and what differs as usual — those only ever
         # add or correct. Absence is the one conclusion this source cannot
@@ -578,9 +617,27 @@ def diff_against_hub(
                 return record_id.strip().lower()
             return "title:" + (title or "").strip().lower()
 
-        have = {
-            _task_key(t.get("record_id"), t.get("title")): t for t in hub_rows
-        }
+        have_groups: dict[str, list[dict[str, Any]]] = {}
+        for task in hub_rows:
+            have_groups.setdefault(
+                _task_key(task.get("record_id"), task.get("title")), []
+            ).append(task)
+        for key, rows in have_groups.items():
+            if len(rows) > 1:
+                d.ambiguous.append(
+                    {
+                        "kind": "task",
+                        "record_id": rows[0].get("record_id"),
+                        "source": "hub",
+                        "workplan": w.record_id,
+                        "uuids": [str(row["id"]) for row in rows],
+                        "match_key": key,
+                    }
+                )
+        # Keep one representative so the rest of the diff remains useful, but
+        # `ambiguous` makes the result non-clean and forbids treating it as an
+        # actionable reset plan.
+        have = {key: rows[0] for key, rows in have_groups.items()}
         want = {_task_key(t.record_id, t.title): t for t in w.tasks}
         for key, t in want.items():
             if key not in have:
@@ -827,7 +884,7 @@ async def reset_repository_projection(
     """
     from datetime import datetime, timezone
 
-    from sqlalchemy import select
+    from sqlalchemy import func, or_, select
 
     from api.models.managed_repo import ManagedRepo
     from api.models.task import Task
@@ -860,6 +917,22 @@ async def reset_repository_projection(
         outcome.refused.append({"reason": "repository is not registered", "slug": repo_slug})
         return outcome
 
+    source_ambiguities = _derived_identity_ambiguities(derived)
+    if source_ambiguities:
+        outcome.status = "refused"
+        for ambiguity in source_ambiguities:
+            outcome.refused.append(
+                {
+                    "reason": "forge projection contains duplicate canonical identity",
+                    **ambiguity,
+                }
+            )
+        outcome.notes.append(
+            "Duplicate identities in the authoritative projection require a file-level "
+            "decision; the reset will not choose a winner."
+        )
+        return outcome
+
     rows = list(
         (
             await session.execute(select(Workplan).where(Workplan.repo_id == repo.id))
@@ -1117,6 +1190,118 @@ async def reset_repository_projection(
         for task_row in loaded:
             tasks_by_wp.setdefault(task_row.workplan_id, []).append(task_row)
 
+    # Canonical task identity is scoped to the matched current workplan while
+    # reconciling. Duplicate rows attached to a retired/displaced workplan are
+    # historical evidence and are deliberately not loaded above. Two rows with
+    # one identity inside the matched workplan, however, are genuinely
+    # ambiguous: the reset cannot know which row owns progress or dependencies.
+    current_task_ambiguities: list[dict[str, Any]] = []
+    for key, workplan in want.items():
+        row = matched.get(key)
+        if row is None:
+            continue
+        groups: dict[str, list[Any]] = {}
+        for task_row in tasks_by_wp.get(row.id, []):
+            record_id = (task_row.record_id or "").strip().lower()
+            if record_id:
+                groups.setdefault(record_id, []).append(task_row)
+        for task_rows in groups.values():
+            if len(task_rows) > 1:
+                current_task_ambiguities.append(
+                    {
+                        "reason": "current workplan contains duplicate canonical task identity",
+                        "kind": "task",
+                        "record_id": task_rows[0].record_id,
+                        "workplan": workplan.record_id,
+                        "uuids": [str(task_row.id) for task_row in task_rows],
+                    }
+                )
+    if current_task_ambiguities:
+        outcome.status = "refused"
+        outcome.refused.extend(current_task_ambiguities)
+        outcome.notes.append(
+            "Retired workplan copies do not block reset, but duplicate identities "
+            "inside the matched current workplan require manual disposition."
+        )
+        return outcome
+
+    # Check every task the reset would create in one query. A derived UUID held
+    # by another workplan is a global identity collision, even when the two
+    # human-readable rows happen to have different titles. Letting the INSERT
+    # discover this would turn a deterministic refusal into an IntegrityError.
+    creating_tasks: dict[str, tuple[DerivedWorkplan, DerivedTask]] = {}
+    for key, workplan in want.items():
+        row = matched.get(key)
+        local_rows = tasks_by_wp.get(row.id, []) if row is not None else []
+        local_record_ids = {
+            (task_row.record_id or "").strip().lower()
+            for task_row in local_rows
+            if (task_row.record_id or "").strip()
+        }
+        local_uuids = {str(task_row.id) for task_row in local_rows}
+        for task in workplan.tasks:
+            if (
+                task.record_id.strip().lower() not in local_record_ids
+                and task.uuid not in local_uuids
+            ):
+                creating_tasks[task.uuid] = (workplan, task)
+
+    if creating_tasks:
+        creating_tasks_by_record_id = {
+            task.record_id.strip().lower(): (workplan, task)
+            for workplan, task in creating_tasks.values()
+        }
+        current_holder_filters = [Workplan.projection_retired_at.is_(None)]
+        if stale:
+            # The caller has already acknowledged these retirements (the
+            # unacknowledged path returned above). Their tasks are preserved as
+            # history, but must not prevent the replacement projection from
+            # claiming its deterministic identities in the same transaction.
+            current_holder_filters.append(
+                Workplan.id.not_in([workplan.id for workplan in stale])
+            )
+        task_holders = list(
+            (
+                await session.execute(
+                    select(Task)
+                    .join(Workplan, Workplan.id == Task.workplan_id)
+                    .where(
+                        *current_holder_filters,
+                        or_(
+                            Task.id.in_(
+                                [uuid.UUID(task_id) for task_id in creating_tasks]
+                            ),
+                            func.lower(Task.record_id).in_(creating_tasks_by_record_id),
+                        ),
+                    )
+                )
+            ).scalars()
+        )
+        if task_holders:
+            outcome.status = "refused"
+            for holder in task_holders:
+                candidate = creating_tasks.get(str(holder.id))
+                if candidate is None:
+                    candidate = creating_tasks_by_record_id[
+                        (holder.record_id or "").strip().lower()
+                    ]
+                workplan, task = candidate
+                outcome.refused.append(
+                    {
+                        "reason": "task identity already belongs to another current workplan",
+                        "kind": "task",
+                        "record_id": task.record_id,
+                        "uuid": task.uuid,
+                        "workplan": workplan.record_id,
+                        "held_by_workplan_id": str(holder.workplan_id),
+                    }
+                )
+            outcome.notes.append(
+                "Task identity collisions require an identity decision; the reset "
+                "will not overwrite or move the existing row."
+            )
+            return outcome
+
     for key, w in want.items():
         row = matched.get(key)
         if row is None: