diff --git a/api/services/forge_projection.py b/api/services/forge_projection.py index 939bc56..144376c 100644 --- a/api/services/forge_projection.py +++ b/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: diff --git a/docs/evidence/STATE-WP-0083-task-identity-audit-2026-08-31.json b/docs/evidence/STATE-WP-0083-task-identity-audit-2026-08-31.json new file mode 100644 index 0000000..4828e60 --- /dev/null +++ b/docs/evidence/STATE-WP-0083-task-identity-audit-2026-08-31.json @@ -0,0 +1,48 @@ +{ + "schema": "state-hub.task-identity-audit.v1", + "observed_at": "2026-08-31", + "source": "primary State Hub task and workplan APIs", + "counts": { + "tasks": 6613, + "identified_tasks": 5831, + "duplicate_record_id_groups": 131, + "rows_in_duplicate_groups": 274, + "cross_workplan_groups": 131, + "same_workplan_groups": 0, + "cross_repository_groups": 31 + }, + "projection_classification": { + "one_current_claimant_with_retired_history": 104, + "retired_history_only": 15, + "multiple_non_retired_claimants": 12 + }, + "multiple_non_retired_claimants": [ + { + "record_id_prefix": "CUST-WP-0010b-T", + "task_identity_groups": 2, + "workplan_slugs": [ + "cust-wp-0010b", + "cust-wp-0010" + ], + "workplan_status": "finished", + "cause": "duplicate workplan projection over one backing file" + }, + { + "record_id_prefix": "SECRETS-WP-0002-T", + "task_identity_groups": 10, + "workplan_slugs": [ + "secrets-wp-0001", + "secrets-wp-0002" + ], + "workplan_status": "finished", + "cause": "duplicate workplan projection over one backing file" + } + ], + "decision": { + "retired_rows": "preserve as historical evidence; exclude from current projection ambiguity", + "duplicate_inside_matched_workplan": "refuse reset and report every claimant UUID", + "duplicate_inside_forge_projection": "refuse reset; require file-level disposition", + "new_task_identity_held_by_other_current_workplan": "refuse before insert", + "duplicate_across_retired_or_displaced_workplans": "does not block current projection reset" + } +} diff --git a/tests/test_forge_projection.py b/tests/test_forge_projection.py index c1b1259..d3fda39 100644 --- a/tests/test_forge_projection.py +++ b/tests/test_forge_projection.py @@ -218,12 +218,21 @@ class _Row: class _FakeSession: """Stands in for AsyncSession: enough to prove intent without a database.""" - def __init__(self, repo, rows, foreign=None, slug_clash=None, task_rows=None): + def __init__( + self, + repo, + rows, + foreign=None, + slug_clash=None, + task_rows=None, + task_collision_rows=None, + ): self._repo = repo self.rows = list(rows) self._foreign = list(foreign or []) self._slug_clash = list(slug_clash or []) self._task_rows = task_rows + self._task_collision_rows = list(task_collision_rows or []) self.added = [] self.deleted = [] self.committed = False @@ -244,12 +253,14 @@ class _FakeSession: payload = rows elif update_only and self._calls == 3: payload = self._task_rows + elif update_only and self._calls == 4: + payload = self._task_collision_rows elif self._calls == 3: payload = self._foreign elif self._calls == 4: payload = self._slug_clash elif self._calls == 5: - payload = self._task_rows or [] + payload = self._task_collision_rows else: payload = [] @@ -1129,3 +1140,124 @@ class TestExistingWorkplanTasks: derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")), ) assert orphan.status == "todo" + + @pytest.mark.asyncio + async def test_duplicate_identity_inside_current_workplan_is_refused(self): + row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md") + first = _TaskRow("DEMO-WP-0001-T01", workplan_id=row.id) + second = _TaskRow("demo-wp-0001-t01", workplan_id=row.id) + session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[first, second]) + out = await fp.reset_repository_projection( + session, + "demo", + derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")), + ) + assert out.status == "refused" + assert out.refused[0]["reason"].startswith("current workplan contains duplicate") + assert session.added == [] + + @pytest.mark.asyncio + async def test_derived_task_uuid_held_elsewhere_is_refused_before_insert(self): + row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md") + derived = self._derived(("DEMO-WP-0001-T01", "Do it", "todo")) + holder = _TaskRow("DEMO-WP-0001-T01", workplan_id=__import__("uuid").uuid4()) + holder.id = __import__("uuid").UUID(derived.workplans[0].tasks[0].uuid) + session = _FakeSession( + repo=_Repo(), + rows=[row], + task_rows=[], + task_collision_rows=[holder], + ) + out = await fp.reset_repository_projection(session, "demo", derived=derived) + assert out.status == "refused" + assert out.refused[0]["reason"].startswith("task identity already belongs") + assert session.added == [] + + @pytest.mark.asyncio + async def test_tasks_on_retired_copy_do_not_block_current_projection(self): + from datetime import datetime, timezone + + current = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md") + current.id = __import__("uuid").UUID(fp.derived_record_uuid("DEMO-WP-0001")) + retired = _Row( + slug="demo-wp-0001@retired-20260831", + status="finished", + path="workplans/a.md", + ) + retired.projection_retired_at = datetime.now(tz=timezone.utc) + task = _TaskRow("DEMO-WP-0001-T01", status="todo", workplan_id=current.id) + session = _FakeSession(repo=_Repo(), rows=[current, retired], task_rows=[task]) + out = await fp.reset_repository_projection( + session, + "demo", + derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")), + ) + assert out.status in {"applied", "noop"} + assert out.refused == [] + + +def test_diff_reports_duplicate_current_task_identity_as_ambiguous(): + derived = TestExistingWorkplanTasks()._derived( + ("DEMO-WP-0001-T01", "Do it", "todo") + ) + workplan_id = derived.workplans[0].uuid + hub_workplans = [ + { + "id": workplan_id, + "slug": "demo-wp-0001", + "status": "active", + "backing_relative_path": "workplans/a.md", + } + ] + hub_tasks = { + workplan_id: [ + { + "id": "11111111-1111-4111-8111-111111111111", + "record_id": "DEMO-WP-0001-T01", + "title": "a", + "status": "todo", + }, + { + "id": "22222222-2222-4222-8222-222222222222", + "record_id": "demo-wp-0001-t01", + "title": "b", + "status": "todo", + }, + ] + } + diff = fp.diff_against_hub(derived, hub_workplans, hub_tasks) + assert diff.clean is False + assert diff.to_dict()["counts"]["ambiguous"] == 1 + assert diff.ambiguous[0]["source"] == "hub" + + +@pytest.mark.asyncio +async def test_reset_refuses_duplicate_task_identity_in_forge_projection(): + first = TestExistingWorkplanTasks()._derived( + ("SHARED-WP-0001-T01", "First", "todo") + ).workplans[0] + second = fp.DerivedWorkplan( + record_id="DEMO-WP-0002", + uuid=fp.derived_record_uuid("DEMO-WP-0002"), + title="Second", + status="active", + relative_path="workplans/b.md", + archived=False, + tasks=[ + fp.DerivedTask( + record_id="SHARED-WP-0001-T01", + uuid=fp.derived_record_uuid("SHARED-WP-0001-T01"), + title="Second claim", + status="todo", + priority="medium", + ) + ], + ) + derived = fp.DerivedProjection( + repo_slug="demo", commit="c0ffee", workplans=[first, second] + ) + session = _FakeSession(repo=_Repo(), rows=[]) + out = await fp.reset_repository_projection(session, "demo", derived=derived) + assert out.status == "refused" + assert any(item.get("source") == "forge" for item in out.refused) + assert session.added == [] diff --git a/workplans/STATE-WP-0083-forge-derived-projection-reset.md b/workplans/STATE-WP-0083-forge-derived-projection-reset.md index bafa8cd..edb0fa7 100644 --- a/workplans/STATE-WP-0083-forge-derived-projection-reset.md +++ b/workplans/STATE-WP-0083-forge-derived-projection-reset.md @@ -8,7 +8,7 @@ status: active owner: codex topic_slug: infotech created: "2026-08-25" -updated: "2026-08-25" +updated: "2026-08-31" related: - CUST-ADR-012 - CUST-WP-0068 @@ -376,6 +376,27 @@ 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`. +**Duplicate identity disposition (2026-08-31).** A fresh primary audit found +131 duplicate task `record_id` groups across 274 rows. The global count is not +the reset safety boundary: 104 groups have exactly one current projection +claimant plus retired history, and 15 exist only under retired workplans. Those +rows are retained as history and do not compete with the matched current +workplan. + +The remaining 12 groups are two duplicated finished workplan projections: +`CUST-WP-0010b-T01..T02` and `SECRETS-WP-0002-T01..T10`. They must be resolved +by retiring the displaced workplan projection, not by deleting or re-keying its +tasks. Evidence: +`docs/evidence/STATE-WP-0083-task-identity-audit-2026-08-31.json`. + +The diff now reports an explicit `ambiguous` class instead of silently choosing +the last row in a dictionary. Reset refuses duplicate identities declared by +the forge, duplicate identified rows inside the matched current workplan, and +any task creation whose canonical id or derived UUID is held by another current +workplan. These checks are batched; retired/displaced workplan tasks are outside +the current matching set and remain preserved. Live deployment and a clean +repository diff remain before T02 can close. + ## Restore a migration mechanism for central