fix(projection): refuse a colliding identifier instead of failing on a constraint

The reset raised IntegrityError on net-kingdom: its ADHOC-2026-08-23 derives to
an identifier another repository already holds — the collision CUST-WP-0066
documents, where two repositories created the same daily identifier on the same
day. Derivation is deterministic, so the clash is real rather than incidental.

It now checks, before creating anything, whether a derived identifier belongs to
another repository, and refuses naming both the record and the holder. A refusal
is something the caller can rule on; a constraint violation is a stack trace.

Acknowledging retirements deliberately does not authorise a collision. Those are
different decisions — one says the work is gone, the other says take an
identifier another repository owns — and conflating them would let a routine
acknowledgement smuggle an identity change through.

Refs STATE-WP-0083-T03

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 2583210@bnt-lap001
Assistant-Session: f2bff2d5-e9b2-4338-92ca-10282a927006

api/services/forge_projection.py

@@ -492,6 +492,45 @@ async def reset_repository_projection(
                 key = cand[0] if len(cand) == 1 else key
         matched[key] = row
 
+    # An identifier this repository would create may already belong to another
+    # repository. Two repositories creating the same daily identifier on the
+    # same day is a documented case (CUST-WP-0066), and the derivation is
+    # deliberately deterministic, so the collision is real rather than
+    # incidental. Refuse and say so: a constraint violation is a stack trace,
+    # a refusal is something the caller can rule on.
+    creating = [w for k, w in want.items() if k not in matched]
+    if creating:
+        foreign = list(
+            (
+                await session.execute(
+                    select(Workplan).where(
+                        Workplan.id.in_([uuid.UUID(w.uuid) for w in creating]),
+                        Workplan.repo_id != repo.id,
+                    )
+                )
+            ).scalars()
+        )
+        if foreign:
+            owned = {str(r.id): r for r in foreign}
+            outcome.status = "refused"
+            for w in creating:
+                held = owned.get(w.uuid)
+                if held is None:
+                    continue
+                outcome.refused.append(
+                    {
+                        "reason": "derived identifier already belongs to another repository",
+                        "record_id": w.record_id,
+                        "uuid": w.uuid,
+                        "held_by_slug": held.slug,
+                    }
+                )
+            outcome.notes.append(
+                "Identifier collision is an identity decision, not a projection "
+                "one; acknowledging retirements does not authorise it."
+            )
+            return outcome
+
     stale = [
         r for k, r in matched.items()
         if k not in want and r.projection_retired_at is None

tests/test_forge_projection.py

@@ -212,9 +212,10 @@ class _Row:
 class _FakeSession:
     """Stands in for AsyncSession: enough to prove intent without a database."""
 
-    def __init__(self, repo, rows):
+    def __init__(self, repo, rows, foreign=None):
         self._repo = repo
         self.rows = list(rows)
+        self._foreign = list(foreign or [])
         self.added = []
         self.deleted = []
         self.committed = False
@@ -223,13 +224,16 @@ class _FakeSession:
     async def execute(self, *_a, **_k):
         self._calls += 1
         repo, rows = self._repo, self.rows
+        # 1st call resolves the repo, 2nd loads its workplans, 3rd is the
+        # foreign-identifier lookup.
+        payload = rows if self._calls == 2 else self._foreign
 
         class R:
             def scalar_one_or_none(self_inner):
                 return repo
 
             def scalars(self_inner):
-                return iter(rows)
+                return iter(payload)
 
         return R()
 
@@ -244,3 +248,54 @@ class _FakeSession:
 
     async def commit(self):
         self.committed = True
+
+
+class TestCollisionRefusal:
+    """A colliding identifier must refuse, not raise (STATE-WP-0083-T03).
+
+    net-kingdom's ADHOC-2026-08-23 derives to an identifier another repository
+    already holds — the case CUST-WP-0066 documents. The reset previously failed
+    on a database constraint, which tells the caller nothing they can act on.
+    """
+
+    @staticmethod
+    def _derived(rid="DEMO-WP-0001"):
+        return fp.DerivedProjection(
+            repo_slug="demo", commit="c0ffee",
+            workplans=[fp.DerivedWorkplan(
+                record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
+                status="active", relative_path="workplans/a.md",
+                archived=False, tasks=[])],
+        )
+
+    @pytest.mark.asyncio
+    async def test_refuses_when_the_identifier_belongs_elsewhere(self):
+        derived = self._derived()
+        foreign = _Row(slug="held-by-someone-else", status="finished", path="workplans/x.md")
+        foreign.id = __import__("uuid").UUID(derived.workplans[0].uuid)
+        session = _FakeSession(repo=_Repo(), rows=[], foreign=[foreign])
+        out = await fp.reset_repository_projection(session, "demo", derived=derived)
+        assert out.status == "refused"
+        assert out.refused[0]["reason"].startswith("derived identifier already belongs")
+        assert out.refused[0]["held_by_slug"] == "held-by-someone-else"
+        assert out.created == [] and session.added == []
+
+    @pytest.mark.asyncio
+    async def test_acknowledging_retirements_does_not_authorise_a_collision(self):
+        """Different decision, different authorisation."""
+        derived = self._derived()
+        foreign = _Row(slug="held-by-someone-else", status="finished", path="workplans/x.md")
+        foreign.id = __import__("uuid").UUID(derived.workplans[0].uuid)
+        session = _FakeSession(repo=_Repo(), rows=[], foreign=[foreign])
+        out = await fp.reset_repository_projection(
+            session, "demo", derived=derived, acknowledge_retirements=True
+        )
+        assert out.status == "refused"
+        assert session.added == []
+
+    @pytest.mark.asyncio
+    async def test_no_collision_still_creates(self):
+        derived = self._derived()
+        session = _FakeSession(repo=_Repo(), rows=[], foreign=[])
+        out = await fp.reset_repository_projection(session, "demo", derived=derived)
+        assert out.status == "applied" and out.created == ["DEMO-WP-0001"]