diff --git a/api/models/__init__.py b/api/models/__init__.py index f4d81ba..f285fc4 100644 --- a/api/models/__init__.py +++ b/api/models/__init__.py @@ -41,6 +41,7 @@ from api.models.workplan_launch_request import WorkplanLaunchRequest from api.models.fabric_graph import FabricGraphImport, FabricGraphNode, FabricGraphEdge from api.models.legacy_meter import LegacyInterface, LegacyInterfaceUsageBucket from api.models.write_idempotency_key import WriteIdempotencyKey +from api.models.work_record_identifier_alias import WorkRecordIdentifierAlias from api.models.suggestion import ( Suggestion, SuggestionNote, @@ -81,5 +82,6 @@ __all__ = [ "FabricGraphImport", "FabricGraphNode", "FabricGraphEdge", "LegacyInterface", "LegacyInterfaceUsageBucket", "WriteIdempotencyKey", + "WorkRecordIdentifierAlias", "Suggestion", "SuggestionNote", "SuggestionRelevanceBump", "SuggestionStage", -] \ No newline at end of file +] diff --git a/api/models/capability_request.py b/api/models/capability_request.py index 8117375..7fdaba1 100644 --- a/api/models/capability_request.py +++ b/api/models/capability_request.py @@ -33,7 +33,7 @@ class CapabilityRequest(Base, TimestampMixin): ) requesting_workplan_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), - ForeignKey("workplans.id", ondelete="SET NULL"), + ForeignKey("workplans.id", ondelete="SET NULL", onupdate="CASCADE"), nullable=True, ) requesting_agent: Mapped[str] = mapped_column(String(100), nullable=False) @@ -47,7 +47,7 @@ class CapabilityRequest(Base, TimestampMixin): ) fulfilling_workplan_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), - ForeignKey("workplans.id", ondelete="SET NULL"), + ForeignKey("workplans.id", ondelete="SET NULL", onupdate="CASCADE"), nullable=True, ) fulfilling_agent: Mapped[str | None] = mapped_column(String(100), nullable=True) @@ -55,7 +55,7 @@ class CapabilityRequest(Base, TimestampMixin): # Links blocking_task_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), - ForeignKey("tasks.id", ondelete="SET NULL"), + ForeignKey("tasks.id", ondelete="SET NULL", onupdate="CASCADE"), nullable=True, ) catalog_entry_id: Mapped[uuid.UUID | None] = mapped_column( diff --git a/api/models/contribution.py b/api/models/contribution.py index 46b934e..0596784 100644 --- a/api/models/contribution.py +++ b/api/models/contribution.py @@ -2,7 +2,7 @@ import enum import uuid from datetime import datetime -from sqlalchemy import Boolean, DateTime, Enum, ForeignKey, String, Text +from sqlalchemy import DateTime, Enum, ForeignKey, String, Text from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -48,7 +48,9 @@ class Contribution(Base, TimestampMixin): UUID(as_uuid=True), ForeignKey("topics.id", ondelete="SET NULL"), nullable=True ) related_workplan_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="SET NULL"), nullable=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, ) repo_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey("managed_repos.id", ondelete="SET NULL"), nullable=True diff --git a/api/models/decision.py b/api/models/decision.py index 08ae0ed..1e9a2c6 100644 --- a/api/models/decision.py +++ b/api/models/decision.py @@ -37,7 +37,10 @@ class Decision(Base, TimestampMixin): UUID(as_uuid=True), ForeignKey("topics.id", ondelete="RESTRICT"), nullable=True, index=True ) workplan_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="RESTRICT"), nullable=True, index=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="RESTRICT", onupdate="CASCADE"), + nullable=True, + index=True, ) title: Mapped[str] = mapped_column(String(255), nullable=False) description: Mapped[str | None] = mapped_column(Text, nullable=True) diff --git a/api/models/extension_point.py b/api/models/extension_point.py index 84f6c2a..532ade4 100644 --- a/api/models/extension_point.py +++ b/api/models/extension_point.py @@ -45,7 +45,9 @@ class ExtensionPoint(Base, TimestampMixin): UUID(as_uuid=True), ForeignKey("topics.id", ondelete="SET NULL"), nullable=True ) workplan_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="SET NULL"), nullable=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, ) domain: Mapped["Domain"] = relationship("Domain", lazy="selectin") # noqa: F821 diff --git a/api/models/intake.py b/api/models/intake.py index 717cb7e..a9cc84f 100644 --- a/api/models/intake.py +++ b/api/models/intake.py @@ -67,7 +67,10 @@ class Intake(Base, TimestampMixin): UUID(as_uuid=True), ForeignKey("topics.id", ondelete="SET NULL"), nullable=True, index=True ) workplan_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="SET NULL"), nullable=True, index=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, + index=True, ) repo_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey("managed_repos.id", ondelete="SET NULL"), nullable=True, index=True diff --git a/api/models/progress_event.py b/api/models/progress_event.py index ca11ffd..b73fe60 100644 --- a/api/models/progress_event.py +++ b/api/models/progress_event.py @@ -20,10 +20,16 @@ class ProgressEvent(Base): UUID(as_uuid=True), ForeignKey("topics.id", ondelete="RESTRICT"), nullable=True, index=True ) workplan_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="RESTRICT"), nullable=True, index=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="RESTRICT", onupdate="CASCADE"), + nullable=True, + index=True, ) task_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="RESTRICT"), nullable=True, index=True + UUID(as_uuid=True), + ForeignKey("tasks.id", ondelete="RESTRICT", onupdate="CASCADE"), + nullable=True, + index=True, ) decision_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey("decisions.id", ondelete="RESTRICT"), nullable=True, index=True diff --git a/api/models/suggestion.py b/api/models/suggestion.py index df5a791..7b46abc 100644 --- a/api/models/suggestion.py +++ b/api/models/suggestion.py @@ -36,7 +36,9 @@ class Suggestion(Base, TimestampMixin): UUID(as_uuid=True), ForeignKey("topics.id", ondelete="SET NULL"), nullable=True ) workplan_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="SET NULL"), nullable=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, ) title: Mapped[str] = mapped_column(String(500), nullable=False) description: Mapped[str | None] = mapped_column(Text, nullable=True) @@ -61,7 +63,9 @@ class Suggestion(Base, TimestampMixin): Float, nullable=False, default=1.0, server_default="1" ) promoted_task_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="SET NULL"), nullable=True + UUID(as_uuid=True), + ForeignKey("tasks.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, ) domain: Mapped["Domain"] = relationship("Domain", lazy="selectin") # noqa: F821 @@ -117,4 +121,4 @@ class SuggestionRelevanceBump(Base): reason: Mapped[str | None] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False - ) \ No newline at end of file + ) diff --git a/api/models/task.py b/api/models/task.py index ecdb35b..f119c06 100644 --- a/api/models/task.py +++ b/api/models/task.py @@ -31,7 +31,10 @@ class Task(Base, TimestampMixin): UUID(as_uuid=True), primary_key=True, default=new_uuid ) workplan_id: Mapped[uuid.UUID] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="RESTRICT"), nullable=False, index=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="RESTRICT", onupdate="CASCADE"), + nullable=False, + index=True, ) title: Mapped[str] = mapped_column(String(255), nullable=False) description: Mapped[str | None] = mapped_column(Text, nullable=True) @@ -47,7 +50,9 @@ class Task(Base, TimestampMixin): needs_human: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False, index=True) intervention_note: Mapped[str | None] = mapped_column(Text, nullable=True) parent_task_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="SET NULL"), nullable=True + UUID(as_uuid=True), + ForeignKey("tasks.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, ) workplan: Mapped["Workplan"] = relationship("Workplan", back_populates="tasks") # noqa: F821 diff --git a/api/models/technical_debt.py b/api/models/technical_debt.py index d501f60..334bf6a 100644 --- a/api/models/technical_debt.py +++ b/api/models/technical_debt.py @@ -77,7 +77,9 @@ class TechnicalDebt(Base, TimestampMixin): UUID(as_uuid=True), ForeignKey("topics.id", ondelete="SET NULL"), nullable=True ) workplan_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="SET NULL"), nullable=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, ) domain: Mapped["Domain"] = relationship("Domain", lazy="selectin") # noqa: F821 diff --git a/api/models/token_event.py b/api/models/token_event.py index cc85825..f1dfe79 100644 --- a/api/models/token_event.py +++ b/api/models/token_event.py @@ -1,9 +1,16 @@ import uuid from datetime import datetime - from typing import Any -from sqlalchemy import DateTime, Float, ForeignKey, Integer, Text, UniqueConstraint, func +from sqlalchemy import ( + DateTime, + Float, + ForeignKey, + Integer, + Text, + UniqueConstraint, + func, +) from sqlalchemy.dialects.postgresql import JSONB, UUID from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -25,10 +32,16 @@ class TokenEvent(Base): UUID(as_uuid=True), primary_key=True, default=new_uuid ) task_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="SET NULL"), nullable=True, index=True + UUID(as_uuid=True), + ForeignKey("tasks.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, + index=True, ) workplan_id: Mapped[uuid.UUID | None] = mapped_column( - UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="SET NULL"), nullable=True, index=True + UUID(as_uuid=True), + ForeignKey("workplans.id", ondelete="SET NULL", onupdate="CASCADE"), + nullable=True, + index=True, ) repo_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey("managed_repos.id", ondelete="SET NULL"), nullable=True, index=True diff --git a/api/models/work_record_identifier_alias.py b/api/models/work_record_identifier_alias.py new file mode 100644 index 0000000..23f389c --- /dev/null +++ b/api/models/work_record_identifier_alias.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import uuid +from datetime import datetime + +from sqlalchemy import CheckConstraint, DateTime, String, UniqueConstraint +from sqlalchemy.dialects.postgresql import UUID +from sqlalchemy.orm import Mapped, mapped_column + +from api.models.base import Base, TimestampMixin, new_uuid + + +class WorkRecordIdentifierAlias(Base, TimestampMixin): + """Durable provenance for a canonical work-record UUID replacement.""" + + __tablename__ = "work_record_identifier_aliases" + __table_args__ = ( + CheckConstraint( + "record_kind IN ('workplan', 'task')", + name="ck_work_record_identifier_aliases_kind", + ), + CheckConstraint( + "migration_status IN ('prepared', 'applied', 'reversed')", + name="ck_work_record_identifier_aliases_status", + ), + UniqueConstraint("old_id", name="uq_work_record_identifier_aliases_old_id"), + UniqueConstraint("new_id", name="uq_work_record_identifier_aliases_new_id"), + ) + + id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, default=new_uuid + ) + old_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False) + new_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False) + record_kind: Mapped[str] = mapped_column(String(20), nullable=False) + record_id: Mapped[str] = mapped_column(String(160), nullable=False, index=True) + repo_slug: Mapped[str] = mapped_column(String(100), nullable=False, index=True) + namespace: Mapped[str] = mapped_column(String(64), nullable=False) + plan_sha256: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + migration_status: Mapped[str] = mapped_column( + String(20), nullable=False, default="prepared", server_default="prepared" + ) + applied_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + reversed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) diff --git a/api/models/workplan_dependency.py b/api/models/workplan_dependency.py index 4ec5391..d74b6e0 100644 --- a/api/models/workplan_dependency.py +++ b/api/models/workplan_dependency.py @@ -45,19 +45,19 @@ class WorkplanDependency(Base, TimestampMixin): ) from_workplan_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), - ForeignKey("workplans.id", ondelete="CASCADE"), + ForeignKey("workplans.id", ondelete="CASCADE", onupdate="CASCADE"), nullable=False, index=True, ) to_workplan_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), - ForeignKey("workplans.id", ondelete="CASCADE"), + ForeignKey("workplans.id", ondelete="CASCADE", onupdate="CASCADE"), nullable=True, index=True, ) to_task_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), - ForeignKey("tasks.id", ondelete="CASCADE"), + ForeignKey("tasks.id", ondelete="CASCADE", onupdate="CASCADE"), nullable=True, index=True, ) @@ -72,4 +72,4 @@ class WorkplanDependency(Base, TimestampMixin): to_workplan: Mapped["Workplan | None"] = relationship( # noqa: F821 "Workplan", foreign_keys=[to_workplan_id] ) - to_task: Mapped["Task | None"] = relationship("Task", foreign_keys=[to_task_id]) # noqa: F821 \ No newline at end of file + to_task: Mapped["Task | None"] = relationship("Task", foreign_keys=[to_task_id]) # noqa: F821 diff --git a/api/models/workplan_launch_request.py b/api/models/workplan_launch_request.py index dec994d..284bd37 100644 --- a/api/models/workplan_launch_request.py +++ b/api/models/workplan_launch_request.py @@ -15,7 +15,7 @@ class WorkplanLaunchRequest(Base, TimestampMixin): ) workplan_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), - ForeignKey("workplans.id", ondelete="CASCADE"), + ForeignKey("workplans.id", ondelete="CASCADE", onupdate="CASCADE"), nullable=False, index=True, ) diff --git a/api/services/work_record_identifier_migration.py b/api/services/work_record_identifier_migration.py new file mode 100644 index 0000000..38ba776 --- /dev/null +++ b/api/services/work_record_identifier_migration.py @@ -0,0 +1,308 @@ +"""Transactional State Hub projection migration for deterministic work-record UUIDs.""" + +from __future__ import annotations + +import hashlib +import json +import re +import uuid +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from api.models.work_record_identifier_alias import WorkRecordIdentifierAlias + +PLAN_SCHEMA = "repo-manager.identifier-migration-plan.v1" +FLEET_NAMESPACE = "helixforge" +DERIVATION_NAMESPACE_UUID = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009") +_RECORD_ID_RE = re.compile(r"^[A-Z][A-Z0-9-]*-WP-[0-9]{4}(?:-T[0-9]{2,})?$") + + +class IdentifierMigrationError(ValueError): + """The signed plan or current projection is unsafe to migrate.""" + + +@dataclass(frozen=True) +class IdentifierMigrationResult: + repo_slug: str + plan_sha256: str + direction: str + replacements: int + assignments_deferred: int + + +def verify_plan(plan: dict[str, Any]) -> str: + """Validate the immutable plan envelope and deterministic UUID mappings.""" + if plan.get("schema") != PLAN_SCHEMA: + raise IdentifierMigrationError("unsupported identifier migration plan schema") + if plan.get("namespace") != FLEET_NAMESPACE: + raise IdentifierMigrationError( + f"plan namespace must be {FLEET_NAMESPACE!r}" + ) + if plan.get("apply_policy") != "all-or-nothing per repository": + raise IdentifierMigrationError("plan does not declare repository-atomic apply") + if plan.get("ready_to_apply") is not True: + raise IdentifierMigrationError("plan is not ready_to_apply") + + expected = plan.get("plan_sha256") + if not isinstance(expected, str) or not re.fullmatch(r"[0-9a-f]{64}", expected): + raise IdentifierMigrationError("plan has no valid SHA-256 seal") + unsealed = {key: value for key, value in plan.items() if key != "plan_sha256"} + canonical = json.dumps(unsealed, sort_keys=True, separators=(",", ":")).encode() + actual = hashlib.sha256(canonical).hexdigest() + if actual != expected: + raise IdentifierMigrationError("plan SHA-256 mismatch") + return expected + + +def _repository_mappings( + plan: dict[str, Any], repo_slug: str +) -> tuple[list[dict[str, Any]], int]: + matches = [item for item in plan.get("repositories", []) if item.get("repo") == repo_slug] + if len(matches) != 1: + raise IdentifierMigrationError( + f"repository {repo_slug!r} must occur exactly once in plan" + ) + repository = matches[0] + if repository.get("eligible") is not True or repository.get("atomic_unit") is not True: + raise IdentifierMigrationError(f"repository {repo_slug!r} is not eligible and atomic") + + replacements: list[dict[str, Any]] = [] + assignments = 0 + seen_old: set[uuid.UUID] = set() + seen_new: set[uuid.UUID] = set() + for raw in repository.get("mappings", []): + action = raw.get("action") + if action == "assign": + assignments += 1 + continue + if action == "unchanged": + continue + if action != "replace": + raise IdentifierMigrationError(f"unsupported mapping action {action!r}") + if raw.get("repo") != repo_slug: + raise IdentifierMigrationError("mapping repository does not match atomic unit") + kind = raw.get("kind") + record_id = raw.get("record_id") + if kind not in {"workplan", "task"}: + raise IdentifierMigrationError(f"unsupported record kind {kind!r}") + if not isinstance(record_id, str) or not _RECORD_ID_RE.fullmatch(record_id): + raise IdentifierMigrationError(f"noncanonical record id {record_id!r}") + try: + old_id = uuid.UUID(str(raw.get("current_uuid"))) + new_id = uuid.UUID(str(raw.get("derived_uuid"))) + except (TypeError, ValueError) as exc: + raise IdentifierMigrationError(f"invalid UUID mapping for {record_id}") from exc + expected_new = uuid.uuid5( + DERIVATION_NAMESPACE_UUID, + f"{FLEET_NAMESPACE}\n{record_id}", + ) + if new_id != expected_new: + raise IdentifierMigrationError(f"derived UUID mismatch for {record_id}") + if old_id == new_id or old_id in seen_old or new_id in seen_new: + raise IdentifierMigrationError(f"non-unique replacement for {record_id}") + seen_old.add(old_id) + seen_new.add(new_id) + replacements.append( + { + "kind": kind, + "record_id": record_id, + "old_id": old_id, + "new_id": new_id, + } + ) + return replacements, assignments + + +async def _assert_projection_preconditions( + session: AsyncSession, + repo_slug: str, + replacements: list[dict[str, Any]], + *, + reverse: bool, +) -> None: + for mapping in replacements: + source_id = mapping["new_id"] if reverse else mapping["old_id"] + target_id = mapping["old_id"] if reverse else mapping["new_id"] + if mapping["kind"] == "workplan": + source_query = text( + "SELECT workplans.id FROM workplans " + "JOIN managed_repos ON managed_repos.id = workplans.repo_id " + "WHERE managed_repos.slug = :repo_slug AND workplans.id = :record_id " + "FOR UPDATE" + ) + target_query = text("SELECT id FROM workplans WHERE id = :record_id") + else: + source_query = text( + "SELECT tasks.id FROM tasks " + "JOIN workplans ON workplans.id = tasks.workplan_id " + "JOIN managed_repos ON managed_repos.id = workplans.repo_id " + "WHERE managed_repos.slug = :repo_slug AND tasks.id = :record_id " + "FOR UPDATE" + ) + target_query = text("SELECT id FROM tasks WHERE id = :record_id") + source = await session.execute( + source_query, + {"repo_slug": repo_slug, "record_id": source_id}, + ) + if source.scalar_one_or_none() is None: + raise IdentifierMigrationError( + f"source {mapping['kind']} {source_id} is absent from repository {repo_slug}" + ) + target = await session.execute(target_query, {"record_id": target_id}) + if target.scalar_one_or_none() is not None: + raise IdentifierMigrationError( + f"target {mapping['kind']} {target_id} already exists" + ) + + +async def apply_repository_identifier_migration( + session: AsyncSession, + plan: dict[str, Any], + repo_slug: str, +) -> IdentifierMigrationResult: + """Replace one repository's projected UUIDs in one database transaction.""" + plan_sha256 = verify_plan(plan) + replacements, assignments = _repository_mappings(plan, repo_slug) + if session.in_transaction(): + raise IdentifierMigrationError("migration requires a fresh database session") + + async with session.begin(): + await _assert_projection_preconditions( + session, repo_slug, replacements, reverse=False + ) + aliases = { + alias.old_id: alias + for alias in ( + await session.execute( + select(WorkRecordIdentifierAlias).where( + WorkRecordIdentifierAlias.old_id.in_( + [mapping["old_id"] for mapping in replacements] + ) + ) + ) + ).scalars() + } + for mapping in replacements: + alias = aliases.get(mapping["old_id"]) + if alias is None: + alias = WorkRecordIdentifierAlias( + old_id=mapping["old_id"], + new_id=mapping["new_id"], + record_kind=mapping["kind"], + record_id=mapping["record_id"], + repo_slug=repo_slug, + namespace=FLEET_NAMESPACE, + plan_sha256=plan_sha256, + ) + session.add(alias) + aliases[mapping["old_id"]] = alias + elif ( + alias.new_id != mapping["new_id"] + or alias.record_kind != mapping["kind"] + or alias.record_id != mapping["record_id"] + or alias.repo_slug != repo_slug + or alias.namespace != FLEET_NAMESPACE + or alias.plan_sha256 != plan_sha256 + or alias.migration_status != "reversed" + ): + raise IdentifierMigrationError( + f"conflicting durable alias for {mapping['old_id']}" + ) + await session.flush() + + for kind in ("workplan", "task"): + table = "workplans" if kind == "workplan" else "tasks" + for mapping in (item for item in replacements if item["kind"] == kind): + result = await session.execute( + text(f"UPDATE {table} SET id = :new_id WHERE id = :old_id"), + {"old_id": mapping["old_id"], "new_id": mapping["new_id"]}, + ) + if result.rowcount != 1: + raise IdentifierMigrationError( + f"failed to replace {mapping['record_id']}" + ) + + now = datetime.now(UTC) + for alias in aliases.values(): + alias.migration_status = "applied" + alias.applied_at = now + alias.reversed_at = None + + return IdentifierMigrationResult( + repo_slug=repo_slug, + plan_sha256=plan_sha256, + direction="forward", + replacements=len(replacements), + assignments_deferred=assignments, + ) + + +async def reverse_repository_identifier_migration( + session: AsyncSession, + plan: dict[str, Any], + repo_slug: str, +) -> IdentifierMigrationResult: + """Restore one repository's prior projected UUIDs and retain aliases.""" + plan_sha256 = verify_plan(plan) + replacements, assignments = _repository_mappings(plan, repo_slug) + if session.in_transaction(): + raise IdentifierMigrationError("migration requires a fresh database session") + + async with session.begin(): + aliases = list( + ( + await session.execute( + select(WorkRecordIdentifierAlias).where( + WorkRecordIdentifierAlias.plan_sha256 == plan_sha256, + WorkRecordIdentifierAlias.repo_slug == repo_slug, + ) + ) + ).scalars() + ) + aliases_by_old = {alias.old_id: alias for alias in aliases} + if len(aliases_by_old) != len(replacements): + raise IdentifierMigrationError("durable alias set is incomplete") + for mapping in replacements: + alias = aliases_by_old.get(mapping["old_id"]) + if ( + alias is None + or alias.new_id != mapping["new_id"] + or alias.record_kind != mapping["kind"] + or alias.record_id != mapping["record_id"] + or alias.migration_status != "applied" + ): + raise IdentifierMigrationError( + f"durable alias is not applied for {mapping['record_id']}" + ) + await _assert_projection_preconditions( + session, repo_slug, replacements, reverse=True + ) + + for kind in ("task", "workplan"): + table = "tasks" if kind == "task" else "workplans" + for mapping in (item for item in replacements if item["kind"] == kind): + result = await session.execute( + text(f"UPDATE {table} SET id = :old_id WHERE id = :new_id"), + {"old_id": mapping["old_id"], "new_id": mapping["new_id"]}, + ) + if result.rowcount != 1: + raise IdentifierMigrationError( + f"failed to reverse {mapping['record_id']}" + ) + + now = datetime.now(UTC) + for alias in aliases: + alias.migration_status = "reversed" + alias.reversed_at = now + + return IdentifierMigrationResult( + repo_slug=repo_slug, + plan_sha256=plan_sha256, + direction="reverse", + replacements=len(replacements), + assignments_deferred=assignments, + ) diff --git a/migrations/versions/b8d4f0a2c6e1_work_record_identifier_aliases.py b/migrations/versions/b8d4f0a2c6e1_work_record_identifier_aliases.py new file mode 100644 index 0000000..730156c --- /dev/null +++ b/migrations/versions/b8d4f0a2c6e1_work_record_identifier_aliases.py @@ -0,0 +1,182 @@ +"""add reversible work-record identifier migration support + +Revision ID: b8d4f0a2c6e1 +Revises: a7c3e9f1b4d2 +Create Date: 2026-08-22 + +""" +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import UUID + +revision = "b8d4f0a2c6e1" +down_revision = "a7c3e9f1b4d2" +branch_labels = None +depends_on = None + + +_CASCADE_REFERENCES = """ +DO $$ +DECLARE + fk RECORD; + updated_definition TEXT; +BEGIN + FOR fk IN + SELECT + constraint_row.conrelid::regclass AS relation_name, + constraint_row.conname AS constraint_name, + pg_get_constraintdef(constraint_row.oid) AS definition + FROM pg_constraint AS constraint_row + WHERE constraint_row.contype = 'f' + AND constraint_row.confrelid IN ( + 'workplans'::regclass, + 'tasks'::regclass + ) + LOOP + updated_definition := regexp_replace( + fk.definition, + ' ON UPDATE (NO ACTION|RESTRICT|CASCADE|SET NULL|SET DEFAULT)', + '', + 'i' + ); + IF position(' ON DELETE ' IN updated_definition) > 0 THEN + updated_definition := regexp_replace( + updated_definition, + ' ON DELETE ', + ' ON UPDATE CASCADE ON DELETE ', + 'i' + ); + ELSE + updated_definition := updated_definition || ' ON UPDATE CASCADE'; + END IF; + + EXECUTE format( + 'ALTER TABLE %s DROP CONSTRAINT %I', + fk.relation_name, + fk.constraint_name + ); + EXECUTE format( + 'ALTER TABLE %s ADD CONSTRAINT %I %s', + fk.relation_name, + fk.constraint_name, + updated_definition + ); + END LOOP; +END $$; +""" + +_REMOVE_UPDATE_CASCADE = """ +DO $$ +DECLARE + fk RECORD; + updated_definition TEXT; +BEGIN + FOR fk IN + SELECT + constraint_row.conrelid::regclass AS relation_name, + constraint_row.conname AS constraint_name, + pg_get_constraintdef(constraint_row.oid) AS definition + FROM pg_constraint AS constraint_row + WHERE constraint_row.contype = 'f' + AND constraint_row.confrelid IN ( + 'workplans'::regclass, + 'tasks'::regclass + ) + LOOP + updated_definition := regexp_replace( + fk.definition, + ' ON UPDATE CASCADE', + '', + 'i' + ); + EXECUTE format( + 'ALTER TABLE %s DROP CONSTRAINT %I', + fk.relation_name, + fk.constraint_name + ); + EXECUTE format( + 'ALTER TABLE %s ADD CONSTRAINT %I %s', + fk.relation_name, + fk.constraint_name, + updated_definition + ); + END LOOP; +END $$; +""" + + +def upgrade() -> None: + op.create_table( + "work_record_identifier_aliases", + sa.Column("id", UUID(as_uuid=True), primary_key=True), + sa.Column("old_id", UUID(as_uuid=True), nullable=False), + sa.Column("new_id", UUID(as_uuid=True), nullable=False), + sa.Column("record_kind", sa.String(length=20), nullable=False), + sa.Column("record_id", sa.String(length=160), nullable=False), + sa.Column("repo_slug", sa.String(length=100), nullable=False), + sa.Column("namespace", sa.String(length=64), nullable=False), + sa.Column("plan_sha256", sa.String(length=64), nullable=False), + sa.Column( + "migration_status", + sa.String(length=20), + nullable=False, + server_default="prepared", + ), + sa.Column("applied_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("reversed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.CheckConstraint( + "record_kind IN ('workplan', 'task')", + name="ck_work_record_identifier_aliases_kind", + ), + sa.CheckConstraint( + "migration_status IN ('prepared', 'applied', 'reversed')", + name="ck_work_record_identifier_aliases_status", + ), + sa.UniqueConstraint("old_id", name="uq_work_record_identifier_aliases_old_id"), + sa.UniqueConstraint("new_id", name="uq_work_record_identifier_aliases_new_id"), + ) + op.create_index( + "ix_work_record_identifier_aliases_record_id", + "work_record_identifier_aliases", + ["record_id"], + ) + op.create_index( + "ix_work_record_identifier_aliases_repo_slug", + "work_record_identifier_aliases", + ["repo_slug"], + ) + op.create_index( + "ix_work_record_identifier_aliases_plan_sha256", + "work_record_identifier_aliases", + ["plan_sha256"], + ) + op.execute(_CASCADE_REFERENCES) + + +def downgrade() -> None: + op.execute(_REMOVE_UPDATE_CASCADE) + op.drop_index( + "ix_work_record_identifier_aliases_plan_sha256", + table_name="work_record_identifier_aliases", + ) + op.drop_index( + "ix_work_record_identifier_aliases_repo_slug", + table_name="work_record_identifier_aliases", + ) + op.drop_index( + "ix_work_record_identifier_aliases_record_id", + table_name="work_record_identifier_aliases", + ) + op.drop_table("work_record_identifier_aliases") diff --git a/tests/test_work_record_identifier_migration.py b/tests/test_work_record_identifier_migration.py new file mode 100644 index 0000000..3179b21 --- /dev/null +++ b/tests/test_work_record_identifier_migration.py @@ -0,0 +1,298 @@ +from __future__ import annotations + +import hashlib +import json +import uuid + +import pytest +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from api.models import Base +from api.models.domain import Domain +from api.models.managed_repo import ManagedRepo +from api.models.progress_event import ProgressEvent +from api.models.task import Task +from api.models.token_event import TokenEvent +from api.models.work_record_identifier_alias import WorkRecordIdentifierAlias +from api.models.workplan import Workplan +from api.models.workplan_dependency import WorkplanDependency +from api.services.work_record_identifier_migration import ( + DERIVATION_NAMESPACE_UUID, + IdentifierMigrationError, + apply_repository_identifier_migration, + reverse_repository_identifier_migration, +) + + +def _derived(record_id: str) -> uuid.UUID: + return uuid.uuid5(DERIVATION_NAMESPACE_UUID, f"helixforge\n{record_id}") + + +def _sealed_plan( + repo_slug: str, + workplan_old: uuid.UUID, + task_old: uuid.UUID, + *, + task_source_override: uuid.UUID | None = None, +) -> dict: + mappings = [ + { + "repo": repo_slug, + "path": "workplans/TEST-WP-0001-migration.md", + "kind": "workplan", + "record_id": "TEST-WP-0001", + "current_uuid": str(workplan_old), + "derived_uuid": str(_derived("TEST-WP-0001")), + "action": "replace", + }, + { + "repo": repo_slug, + "path": "workplans/TEST-WP-0001-migration.md", + "kind": "task", + "record_id": "TEST-WP-0001-T01", + "current_uuid": str(task_source_override or task_old), + "derived_uuid": str(_derived("TEST-WP-0001-T01")), + "action": "replace", + }, + { + "repo": repo_slug, + "path": "workplans/TEST-WP-0001-migration.md", + "kind": "task", + "record_id": "TEST-WP-0001-T02", + "current_uuid": None, + "derived_uuid": str(_derived("TEST-WP-0001-T02")), + "action": "assign", + }, + ] + plan = { + "schema": "repo-manager.identifier-migration-plan.v1", + "ok": True, + "ready_to_apply": True, + "root": "/test", + "namespace": "helixforge", + "derivation_version": "repo-manager.work-record-uuid.v1", + "namespace_uuid": str(DERIVATION_NAMESPACE_UUID), + "scope": "live workplans and unfinished tasks", + "apply_policy": "all-or-nothing per repository", + "generated_at": "2026-08-21T00:00:00Z", + "totals": { + "repositories": 1, + "eligible": 1, + "skipped": 0, + "records": 3, + "replace": 2, + "assign": 1, + "unchanged": 0, + }, + "repositories": [ + { + "repo": repo_slug, + "path": "/test/repo", + "planned_head_sha": "0" * 40, + "source_fingerprint": "1" * 64, + "eligible": True, + "atomic_unit": True, + "blockers": [], + "mappings": mappings, + } + ], + } + canonical = json.dumps(plan, sort_keys=True, separators=(",", ":")).encode() + plan["plan_sha256"] = hashlib.sha256(canonical).hexdigest() + return plan + + +async def _seed_projection(factory, repo_slug: str): + domain_id = uuid.uuid4() + repo_id = uuid.uuid4() + workplan_old = uuid.uuid4() + task_old = uuid.uuid4() + child_task_id = uuid.uuid4() + progress_id = uuid.uuid4() + token_id = uuid.uuid4() + dependency_id = uuid.uuid4() + async with factory() as session: + session.add_all( + [ + Domain(id=domain_id, slug="infotech", name="Infotech", status="active"), + ManagedRepo( + id=repo_id, + domain_id=domain_id, + slug=repo_slug, + name="Test Repository", + status="active", + ), + Workplan( + id=workplan_old, + repo_id=repo_id, + slug="test-wp-0001", + title="Migration test", + status="active", + ), + Task( + id=task_old, + workplan_id=workplan_old, + title="Mapped task", + status="progress", + priority="high", + ), + Task( + id=child_task_id, + workplan_id=workplan_old, + parent_task_id=task_old, + title="Unmapped child", + status="todo", + priority="medium", + ), + ProgressEvent( + id=progress_id, + workplan_id=workplan_old, + task_id=task_old, + event_type="note", + summary="migration evidence", + ), + TokenEvent( + id=token_id, + workplan_id=workplan_old, + task_id=task_old, + tokens_in=1, + tokens_out=1, + ), + WorkplanDependency( + id=dependency_id, + from_workplan_id=workplan_old, + to_task_id=task_old, + relationship_type="blocks", + ), + ] + ) + await session.commit() + return { + "workplan": workplan_old, + "task": task_old, + "child": child_task_id, + "progress": progress_id, + "token": token_id, + "dependency": dependency_id, + } + + +def test_every_work_record_foreign_key_cascades_on_update(): + references = [ + foreign_key + for table in Base.metadata.sorted_tables + for foreign_key in table.foreign_keys + if foreign_key.target_fullname in {"workplans.id", "tasks.id"} + ] + assert len(references) == 20 + assert all(foreign_key.onupdate == "CASCADE" for foreign_key in references) + + +@pytest.mark.asyncio +async def test_repository_migration_cascades_and_reverses(test_engine): + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + ids = await _seed_projection(factory, "test-repo") + plan = _sealed_plan("test-repo", ids["workplan"], ids["task"]) + workplan_new = _derived("TEST-WP-0001") + task_new = _derived("TEST-WP-0001-T01") + + async with factory() as session: + result = await apply_repository_identifier_migration(session, plan, "test-repo") + assert result.direction == "forward" + assert result.replacements == 2 + assert result.assignments_deferred == 1 + + async with factory() as session: + assert await session.scalar( + text("SELECT id FROM workplans WHERE id = :id"), {"id": workplan_new} + ) == workplan_new + assert await session.scalar( + text("SELECT workplan_id FROM tasks WHERE id = :id"), {"id": task_new} + ) == workplan_new + assert await session.scalar( + text("SELECT parent_task_id FROM tasks WHERE id = :id"), {"id": ids["child"]} + ) == task_new + assert tuple( + ( + await session.execute( + text( + "SELECT workplan_id, task_id FROM progress_events WHERE id = :id" + ), + {"id": ids["progress"]}, + ) + ).one() + ) == (workplan_new, task_new) + assert tuple( + ( + await session.execute( + text("SELECT workplan_id, task_id FROM token_events WHERE id = :id"), + {"id": ids["token"]}, + ) + ).one() + ) == (workplan_new, task_new) + assert tuple( + ( + await session.execute( + text( + "SELECT from_workplan_id, to_task_id " + "FROM workplan_dependencies WHERE id = :id" + ), + {"id": ids["dependency"]}, + ) + ).one() + ) == (workplan_new, task_new) + aliases = list( + ( + await session.execute( + select(WorkRecordIdentifierAlias).order_by( + WorkRecordIdentifierAlias.record_kind + ) + ) + ).scalars() + ) + assert len(aliases) == 2 + assert all(alias.migration_status == "applied" for alias in aliases) + + async with factory() as session: + result = await reverse_repository_identifier_migration(session, plan, "test-repo") + assert result.direction == "reverse" + + async with factory() as session: + assert await session.scalar( + text("SELECT id FROM workplans WHERE id = :id"), {"id": ids["workplan"]} + ) == ids["workplan"] + assert await session.scalar( + text("SELECT workplan_id FROM tasks WHERE id = :id"), {"id": ids["task"]} + ) == ids["workplan"] + assert await session.scalar( + text("SELECT parent_task_id FROM tasks WHERE id = :id"), {"id": ids["child"]} + ) == ids["task"] + aliases = list((await session.execute(select(WorkRecordIdentifierAlias))).scalars()) + assert len(aliases) == 2 + assert all(alias.migration_status == "reversed" for alias in aliases) + assert all(alias.reversed_at is not None for alias in aliases) + + +@pytest.mark.asyncio +async def test_repository_migration_is_atomic_when_a_source_is_missing(test_engine): + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + ids = await _seed_projection(factory, "atomic-repo") + plan = _sealed_plan( + "atomic-repo", + ids["workplan"], + ids["task"], + task_source_override=uuid.uuid4(), + ) + + async with factory() as session: + with pytest.raises(IdentifierMigrationError, match="source task"): + await apply_repository_identifier_migration(session, plan, "atomic-repo") + + async with factory() as session: + assert await session.scalar( + text("SELECT id FROM workplans WHERE id = :id"), {"id": ids["workplan"]} + ) == ids["workplan"] + assert await session.scalar( + text("SELECT count(*) FROM work_record_identifier_aliases") + ) == 0 diff --git a/workplans/STATE-WP-0079-retirement-strangler.md b/workplans/STATE-WP-0079-retirement-strangler.md index 1cecf62..f3fe8e0 100644 --- a/workplans/STATE-WP-0079-retirement-strangler.md +++ b/workplans/STATE-WP-0079-retirement-strangler.md @@ -8,7 +8,7 @@ status: active owner: codex topic_slug: infotech created: "2026-08-09" -updated: "2026-08-19" +updated: "2026-08-22" parent_project: prj-state-hub-retirement parent_workplan: SHR-WP-0001 related: @@ -176,6 +176,26 @@ transactional projection migration, durable old-id aliases, rollback rehearsal, and compatibility evidence needed by RMGR-WP-0005-T04. Do not rewrite repository UUID fields until that sub-slice passes in isolated PostgreSQL. +**Identifier projection gate implemented and rehearsed (2026-08-22):** Alembic +revision `b8d4f0a2c6e1` adds durable old→new alias provenance and converts all 20 +foreign keys into `workplans.id` / `tasks.id` to `ON UPDATE CASCADE`, while +retaining their existing delete behavior. The internal executor verifies the +Repo Manager plan seal, `helixforge` namespace, UUIDv5 derivation, repository +membership, source presence, target absence, and alias consistency before a +repository-atomic apply. Reverse runs task-first/workplan-second and retains the +aliases as `reversed`; plan entries with `action: assign` are deferred to normal +file reconciliation because there is no old projection row to rewrite. + +Isolated PostgreSQL evidence: migration upgrade produced 20/20 update-cascade +constraints plus the alias table; downgrade restored 20/20 no-action update +constraints and removed the table. Service tests cover cascades through task +parentage, progress, token accounting and dependency edges, forward/reverse +alias state, and all-or-nothing failure (`3 passed`). Full State Hub regression: +`622 passed`. No live database migration or repository UUID rewrite was run. +The remaining gate is an operator-approved pilot ordering the central database +transaction and authoritative file rewrite/reconciliation as one recoverable +cutover unit. + **A2a executed (2026-08-20) — first live cutover slice.** Dual-run is on for the pilot repo: