"""replace disposable Registry R1 state with the R2 publication catalog Revision ID: f62a9b7c4d11 Revises: d3f62e716ad4 Create Date: 2026-09-01 17:20:00.000000 """ from collections.abc import Sequence import sqlalchemy as sa from alembic import op revision: str = "f62a9b7c4d11" down_revision: str | None = "d3f62e716ad4" branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None def upgrade() -> None: _drop_r1_schema() op.create_table( "catalog_agent", sa.Column("id", sa.Uuid(), nullable=False), sa.Column("name", sa.String(length=63), nullable=False), sa.Column("display_name", sa.String(length=256), nullable=True), sa.Column("description", sa.Text(), nullable=True), sa.Column("owner_type", sa.String(length=16), nullable=True), sa.Column("owner_issuer", sa.String(length=1024), nullable=True), sa.Column("owner_subject", sa.String(length=512), nullable=True), sa.Column("lifecycle_status", sa.String(length=32), nullable=False), sa.Column("projection_health", sa.String(length=32), nullable=False), sa.Column("selected_version_id", sa.Uuid(), nullable=True), sa.Column("publication_revision", sa.BigInteger(), nullable=False), sa.Column("source_revision", sa.BigInteger(), nullable=False), sa.Column("metadata_revision", sa.BigInteger(), nullable=False), sa.Column("aggregate_revision", sa.BigInteger(), nullable=False), sa.Column("coverage_at", sa.DateTime(timezone=True), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), sa.CheckConstraint("publication_revision >= 0", name="ck_catalog_publication_revision"), sa.CheckConstraint("source_revision >= 0", name="ck_catalog_source_revision"), sa.CheckConstraint("metadata_revision >= 0", name="ck_catalog_metadata_revision"), sa.CheckConstraint("aggregate_revision >= 0", name="ck_catalog_aggregate_revision"), sa.CheckConstraint( "(owner_type IS NULL AND owner_issuer IS NULL AND owner_subject IS NULL) OR " "(owner_type IN ('principal', 'service') AND owner_issuer IS NOT NULL " "AND owner_subject IS NOT NULL)", name="ck_catalog_owner_complete", ), sa.PrimaryKeyConstraint("id"), ) op.create_index("ix_catalog_agent_name", "catalog_agent", ["name"]) op.create_index( "ix_catalog_agent_owner", "catalog_agent", ["owner_type", "owner_issuer", "owner_subject"], ) op.create_index( "ix_catalog_agent_serving", "catalog_agent", ["lifecycle_status", "projection_health", "updated_at"], ) op.create_index( "ix_catalog_agent_selected_version_id", "catalog_agent", ["selected_version_id"] ) op.create_table( "catalog_agent_version", sa.Column("id", sa.Uuid(), nullable=False), sa.Column("agent_id", sa.Uuid(), nullable=False), sa.Column("semantic_version", sa.String(length=128), nullable=False), sa.Column("protocol", sa.String(length=16), nullable=False), sa.Column("protocol_version", sa.String(length=64), nullable=False), sa.Column("implementation_type", sa.String(length=32), nullable=False), sa.Column("artifact_digest", sa.String(length=71), nullable=False), sa.Column("publication_revision", sa.BigInteger(), nullable=False), sa.Column("display_name", sa.String(length=256), nullable=False), sa.Column("description", sa.Text(), nullable=False), sa.Column("capabilities", sa.JSON(), nullable=False), sa.Column("descriptor_digest", sa.String(length=71), nullable=False), sa.Column("descriptor", sa.JSON(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.CheckConstraint("publication_revision >= 1", name="ck_catalog_version_revision"), sa.ForeignKeyConstraint(["agent_id"], ["catalog_agent.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("agent_id", "semantic_version", name="uq_catalog_agent_semver"), sa.UniqueConstraint("agent_id", "artifact_digest", name="uq_catalog_agent_artifact"), ) op.create_index("ix_catalog_agent_version_agent_id", "catalog_agent_version", ["agent_id"]) op.create_index( "ix_catalog_version_filters", "catalog_agent_version", ["protocol", "implementation_type"], ) op.create_index( "ix_catalog_agent_version_artifact_digest", "catalog_agent_version", ["artifact_digest"], ) op.create_foreign_key( "fk_catalog_agent_selected_version", "catalog_agent", "catalog_agent_version", ["selected_version_id"], ["id"], ondelete="SET NULL", use_alter=True, ) op.create_table( "catalog_agent_skill", sa.Column("id", sa.Uuid(), nullable=False), sa.Column("agent_version_id", sa.Uuid(), nullable=False), sa.Column("skill_id", sa.String(length=256), nullable=False), sa.Column("name", sa.String(length=256), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column("tags", sa.JSON(), nullable=False), sa.ForeignKeyConstraint( ["agent_version_id"], ["catalog_agent_version.id"], ondelete="CASCADE" ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("agent_version_id", "skill_id", name="uq_catalog_version_skill"), ) op.create_index( "ix_catalog_agent_skill_agent_version_id", "catalog_agent_skill", ["agent_version_id"], ) op.create_table( "catalog_agent_interface", sa.Column("agent_version_id", sa.Uuid(), nullable=False), sa.Column("service_uid", sa.Uuid(), nullable=False), sa.Column("route_uid", sa.Uuid(), nullable=False), sa.Column("route_generation", sa.BigInteger(), nullable=False), sa.Column("route_resource_version", sa.String(length=128), nullable=False), sa.Column("url", sa.String(length=2048), nullable=False), sa.Column("protocol_binding", sa.String(length=64), nullable=False), sa.Column("exposure", sa.String(length=64), nullable=False), sa.Column("authentication", sa.String(length=64), nullable=False), sa.Column("authorization", sa.String(length=64), nullable=False), sa.Column("backend_authentication", sa.String(length=64), nullable=False), sa.Column("backend_credential", sa.String(length=64), nullable=False), sa.Column("original_bearer_forwarded", sa.Boolean(), nullable=False), sa.Column("caller_identity_forwarded", sa.Boolean(), nullable=False), sa.Column("observed_at", sa.DateTime(timezone=True), nullable=False), sa.CheckConstraint("route_generation >= 1", name="ck_catalog_route_generation"), sa.ForeignKeyConstraint( ["agent_version_id"], ["catalog_agent_version.id"], ondelete="CASCADE" ), sa.PrimaryKeyConstraint("agent_version_id"), ) op.create_index( "ix_catalog_agent_interface_service_uid", "catalog_agent_interface", ["service_uid"] ) op.create_index( "ix_catalog_agent_interface_route_uid", "catalog_agent_interface", ["route_uid"] ) op.create_table( "publication_observation_state", sa.Column("id", sa.Uuid(), nullable=False), sa.Column("agent_id", sa.Uuid(), nullable=False), sa.Column("agent_version_id", sa.Uuid(), nullable=False), sa.Column("publication_revision", sa.BigInteger(), nullable=False), sa.Column("cluster_id", sa.String(length=128), nullable=False), sa.Column("namespace", sa.String(length=253), nullable=False), sa.Column("service_name", sa.String(length=253), nullable=False), sa.Column("service_uid", sa.Uuid(), nullable=False), sa.Column("service_resource_version", sa.String(length=128), nullable=False), sa.Column("artifact_digest", sa.String(length=71), nullable=False), sa.Column("metadata_revision", sa.BigInteger(), nullable=False), sa.Column("metadata_owner_digest", sa.String(length=71), nullable=False), sa.Column("status", sa.String(length=48), nullable=False), sa.Column("endpoint_ready", sa.Boolean(), nullable=False), sa.Column("descriptor_digest", sa.String(length=71), nullable=True), sa.Column("candidate_route_uid", sa.Uuid(), nullable=True), sa.Column("candidate_route_generation", sa.BigInteger(), nullable=True), sa.Column("public_route_uid", sa.Uuid(), nullable=True), sa.Column("public_route_generation", sa.BigInteger(), nullable=True), sa.Column("failure_code", sa.String(length=128), nullable=True), sa.Column("scope_revision", sa.String(length=128), nullable=False), sa.Column("coverage_at", sa.DateTime(timezone=True), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), sa.CheckConstraint("publication_revision >= 1", name="ck_observation_revision"), sa.CheckConstraint("metadata_revision >= 1", name="ck_observation_metadata_revision"), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint( "cluster_id", "namespace", "service_uid", name="uq_publication_service_uid" ), ) op.create_index( "ix_publication_agent_revision", "publication_observation_state", ["agent_id", "publication_revision"], ) op.create_index( "ix_publication_coverage", "publication_observation_state", ["cluster_id", "namespace", "coverage_at"], ) for column in ("agent_id", "agent_version_id", "service_uid", "status"): op.create_index( f"ix_publication_observation_state_{column}", "publication_observation_state", [column], ) op.create_table( "metadata_owner_projection_state", sa.Column("agent_id", sa.Uuid(), nullable=False), sa.Column("metadata_revision", sa.BigInteger(), nullable=False), sa.Column("metadata_owner_digest", sa.String(length=71), nullable=False), sa.Column("owner_type", sa.String(length=16), nullable=True), sa.Column("owner_issuer", sa.String(length=1024), nullable=True), sa.Column("owner_subject", sa.String(length=512), nullable=True), sa.Column("consistency_revision", sa.String(length=2048), nullable=False), sa.Column("state", sa.String(length=32), nullable=False), sa.Column("confirmed_at", sa.DateTime(timezone=True), nullable=False), sa.CheckConstraint("metadata_revision >= 1", name="ck_owner_projection_revision"), sa.ForeignKeyConstraint(["agent_id"], ["catalog_agent.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("agent_id"), ) op.create_table( "observation_coverage_state", sa.Column("cluster_id", sa.String(length=128), nullable=False), sa.Column("namespace", sa.String(length=253), nullable=False), sa.Column("scope_revision", sa.String(length=128), nullable=False), sa.Column("status", sa.String(length=32), nullable=False), sa.Column("last_successful_full_resync_at", sa.DateTime(timezone=True), nullable=True), sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), sa.PrimaryKeyConstraint("cluster_id", "namespace"), ) op.create_table( "drift_finding_state", sa.Column("id", sa.Uuid(), nullable=False), sa.Column("direction", sa.String(length=32), nullable=False), sa.Column("cluster_id", sa.String(length=128), nullable=False), sa.Column("namespace", sa.String(length=253), nullable=False), sa.Column("service_name", sa.String(length=253), nullable=False), sa.Column("service_uid", sa.Uuid(), nullable=False), sa.Column("agent_id", sa.Uuid(), nullable=True), sa.Column("agent_version_id", sa.Uuid(), nullable=True), sa.Column("reason_code", sa.String(length=128), nullable=False), sa.Column("first_detected_at", sa.DateTime(timezone=True), nullable=False), sa.Column("last_confirmed_at", sa.DateTime(timezone=True), nullable=False), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint( "cluster_id", "namespace", "service_uid", "direction", name="uq_drift_workload_direction", ), ) op.create_index( "ix_drift_scope", "drift_finding_state", ["cluster_id", "namespace", "last_confirmed_at"], ) op.create_table( "registry_event_outbox", sa.Column("id", sa.Uuid(), nullable=False), sa.Column("stream", sa.String(length=128), nullable=False), sa.Column("aggregate_id", sa.Uuid(), nullable=False), sa.Column("aggregate_revision", sa.BigInteger(), nullable=False), sa.Column("event_type", sa.String(length=160), nullable=False), sa.Column("payload", sa.JSON(), nullable=False), sa.Column("available_at", sa.DateTime(timezone=True), nullable=False), sa.Column("claimed_by", sa.String(length=256), nullable=True), sa.Column("claim_expires_at", sa.DateTime(timezone=True), nullable=True), sa.Column("publish_attempts", sa.Integer(), nullable=False), sa.Column("last_error_code", sa.String(length=128), nullable=True), sa.Column("published_at", sa.DateTime(timezone=True), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.CheckConstraint("aggregate_revision >= 1", name="ck_registry_event_revision"), sa.CheckConstraint("publish_attempts >= 0", name="ck_registry_event_attempts"), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint( "stream", "aggregate_id", "aggregate_revision", "event_type", name="uq_registry_event_revision", ), ) op.create_index( "ix_registry_event_queue", "registry_event_outbox", ["published_at", "available_at", "created_at"], ) op.create_index( "ix_registry_event_outbox_aggregate_id", "registry_event_outbox", ["aggregate_id"] ) op.create_table( "registry_audit_record", sa.Column("id", sa.Uuid(), nullable=False), sa.Column("agent_id", sa.Uuid(), nullable=True), sa.Column("agent_version_id", sa.Uuid(), nullable=True), sa.Column("service_uid", sa.Uuid(), nullable=True), sa.Column("action", sa.String(length=128), nullable=False), sa.Column("decision", sa.String(length=32), nullable=False), sa.Column("error_code", sa.String(length=128), nullable=True), sa.Column("trace_id", sa.String(length=32), nullable=False), sa.Column("safe_metadata", sa.JSON(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.PrimaryKeyConstraint("id"), ) op.create_index( "ix_registry_audit_agent_created", "registry_audit_record", ["agent_id", "created_at"], ) for column in ("agent_id", "agent_version_id", "trace_id", "created_at"): op.create_index(f"ix_registry_audit_record_{column}", "registry_audit_record", [column]) def downgrade() -> None: raise RuntimeError( "R2 intentionally discards disposable R1 catalog state; restore the pre-R2 " "PostgreSQL backup before deploying the R1 application" ) def _drop_r1_schema() -> None: # R1 has no production data and is replaced in one R2 maintenance boundary. # CASCADE removes only constraints between the exact owned tables below. for table in ( "registration_attempt", "workload_observation", "search_projection_state", "reindex_partition_offset", "protocol_snapshot", "permission_sync_state", "mcp_server_profile", "invocation_route_projection_state", "a2a_skill", "a2a_profile", "resource_version", "reindex_job", "registry_resource", "projection_event", "outbox_record", "audit_record", ): op.execute(sa.text(f'DROP TABLE IF EXISTS "{table}" CASCADE'))