Add Stage 12 observability identifier schema

changelog.d/3851.changed.md

@@ -0,0 +1 @@
+Added a nullable Stage 12 observability identifier column before the simulation observability rollout.

migrations/v2/versions/91af4fc8d8c1_add_stage12_observability_id.py

@@ -0,0 +1,38 @@
+"""add Stage 12 observability identifier
+
+Revision ID: 91af4fc8d8c1
+Revises: 60d6518b6a98
+Create Date: 2026-09-23 18:30:00.000000
+Generation: uv run alembic -c alembic-v2.ini revision --autogenerate
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+revision: str = "91af4fc8d8c1"
+down_revision: Union[str, None] = "60d6518b6a98"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "stage12_evaluation_reports",
+        sa.Column(
+            "observability_id",
+            sqlmodel.sql.sqltypes.AutoString(length=36),
+            nullable=True,
+        ),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("stage12_evaluation_reports", "observability_id")
+    # ### end Alembic commands ###

policyengine_api/data/v2/catalog/publication.py

@@ -24,6 +24,7 @@
     PublicationEvidence,
 )
 from policyengine_api.data.v2.catalog.records import NormalizedCatalog
+from policyengine_api.data.v2.migration_target import V2_ALEMBIC_HEAD_REVISION
 from policyengine_api.data.v2.models import (
     DatasetVersion,
     Report,
@@ -32,7 +33,6 @@
 )
 
 
-EXPECTED_ALEMBIC_REVISION = "60d6518b6a98"
 # Stable application-defined PostgreSQL lock ID shared by all v2 catalog publishers.
 PUBLICATION_ADVISORY_LOCK_KEY = 8_629_020_026_090_001
 
@@ -64,7 +64,7 @@ def _verify_expected_revision(connection: Connection) -> None:
     revisions = set(
         connection.execute(sa.select(ALEMBIC_VERSION.c.version_num)).scalars()
     )
-    if revisions != {EXPECTED_ALEMBIC_REVISION}:
+    if revisions != {V2_ALEMBIC_HEAD_REVISION}:
         raise CatalogPublicationError(
             "the v2 database is not at the expected Alembic revision"
         )

policyengine_api/data/v2/migration_target.py

@@ -21,6 +21,7 @@
 )
 
 V2_ALEMBIC_DISPOSABLE_TEST = "V2_ALEMBIC_DISPOSABLE_TEST"
+V2_ALEMBIC_HEAD_REVISION = "91af4fc8d8c1"
 DISPOSABLE_DATABASE_NAME = "policyengine_v2_alembic_test"
 DISPOSABLE_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "postgres"})
 MIGRATION_ROLE = "policyengine_v2_migrator"

policyengine_api/data/v2/models/comparison_runs.py

@@ -140,6 +140,7 @@ class Stage12ComparisonReport(Stage12RunTimestamps, table=True):
     environment: str = Field(max_length=255)
     calculation_flow: str = Field(max_length=255)
     originating_request_id: str = Field(max_length=255)
+    observability_id: str | None = Field(default=None, max_length=36)
     production_identity: str = Field(max_length=255)
     incumbent_execution_id: str | None = Field(default=None, max_length=255)
     worker_version: str = Field(max_length=255)

tests/integration/test_alembic_v2_lifecycle.py

@@ -15,6 +15,7 @@
 from policyengine_api.constants import REPO
 from policyengine_api.data.v2.migration_target import (
     V2_ALEMBIC_DISPOSABLE_TEST,
+    V2_ALEMBIC_HEAD_REVISION,
     V2MigrationTargetError,
     load_v2_alembic_settings,
 )
@@ -26,7 +27,6 @@
 STAGE_11_PREVIOUS_REVISION = "af34023a728f"
 STAGE_12_PREVIOUS_REVISION = "724b1b11a33e"
 STAGE_12_COMPARISON_PREVIOUS_REVISION = "439303be14fe"
-HEAD_REVISION = "60d6518b6a98"
 V2_TABLE_NAMES = frozenset(table.name for table in V2_METADATA.tables.values())
 
 
@@ -55,7 +55,7 @@ def _assert_head(engine) -> None:
     )
     with engine.connect() as connection:
         context = MigrationContext.configure(connection)
-        assert context.get_current_revision() == HEAD_REVISION
+        assert context.get_current_revision() == V2_ALEMBIC_HEAD_REVISION
         assert compare_metadata(context, V2_METADATA) == []
         model_count = connection.execute(
             text(

tests/unit/test_asgi_factory.py

@@ -516,7 +516,8 @@ def test_cors_preflight_is_handled_before_v2_route_resolution(path, method):
         "authorization, content-type, x-policyengine-request-id"
     )
     assert response.headers["access-control-max-age"] == "600"
-    assert response.headers["vary"] == "Origin"
+    vary_values = {value.strip() for value in response.headers["vary"].split(",")}
+    assert "Origin" in vary_values
 
 
 def test_cors_preflight_rejects_a_method_outside_the_public_http_contract():

tests/unit/v2/test_alembic_v2.py

@@ -21,6 +21,7 @@
     DISPOSABLE_DATABASE_NAME,
     MIGRATION_ROLE,
     V2_ALEMBIC_DISPOSABLE_TEST,
+    V2_ALEMBIC_HEAD_REVISION,
     V2AlembicSettings,
     V2MigrationTargetError,
     load_v2_alembic_settings,
@@ -230,8 +231,9 @@ def test_v2_files_are_mechanically_separate_from_v1() -> None:
 def test_v2_revision_chain_has_generated_policy_and_user_identity_changes() -> None:
     config = Config(str(REPO / "alembic-v2.ini"))
     script = ScriptDirectory.from_config(config)
-    assert script.get_heads() == ["60d6518b6a98"]
+    assert script.get_heads() == [V2_ALEMBIC_HEAD_REVISION]
     assert [revision.revision for revision in script.walk_revisions()] == [
+        V2_ALEMBIC_HEAD_REVISION,
         "60d6518b6a98",
         "439303be14fe",
         "724b1b11a33e",
@@ -466,6 +468,25 @@ def test_stage_12_result_comparison_revision_is_generated_and_reversible() -> No
     assert "op.bulk_insert(" not in revision
 
 
+def test_stage_12_observability_revision_is_generated_and_reversible() -> None:
+    revision = (
+        REPO / "migrations/v2/versions/91af4fc8d8c1_add_stage12_observability_id.py"
+    ).read_text(encoding="utf-8")
+
+    assert (
+        "Generation: uv run alembic -c alembic-v2.ini revision --autogenerate"
+        in revision
+    )
+    assert 'down_revision: Union[str, None] = "60d6518b6a98"' in revision
+    assert '"observability_id"' in revision
+    assert "AutoString(length=36)" in revision
+    assert (
+        'op.drop_column("stage12_evaluation_reports", "observability_id")' in revision
+    )
+    assert "op.execute(" not in revision
+    assert "op.bulk_insert(" not in revision
+
+
 def test_alembic_rejects_unknown_missing_and_divergent_history(tmp_path: Path) -> None:
     original = REPO / "migrations/v2"
     missing = tmp_path / "missing"

tests/unit/v2/test_catalog_publication.py

@@ -19,6 +19,7 @@
     publication_reconciliation,
     publication_staging,
 )
+from policyengine_api.data.v2.migration_target import V2_ALEMBIC_HEAD_REVISION
 from policyengine_api.data.v2.models import (
     Dataset,
     Parameter,
@@ -86,7 +87,7 @@ def test_expected_publication_revision_is_the_alembic_head() -> None:
     config = Config(str(REPO / "alembic-v2.ini"), output_buffer=StringIO())
     script = ScriptDirectory.from_config(config)
 
-    assert script.get_current_head() == publication.EXPECTED_ALEMBIC_REVISION
+    assert script.get_current_head() == V2_ALEMBIC_HEAD_REVISION
 
 
 class _ScalarResult: