@@ -168,6 +168,9 @@
     from adcp_reporting_storage.migration import (
         ReportingOwnershipMigrationError as ReportingOwnershipMigrationError,
     )
+    from adcp_reporting_storage.migration import (
+        ReportingSchemaAheadError as ReportingSchemaAheadError,
+    )
     from adcp_reporting_storage.migration import (
         backfill_legacy_reporting as backfill_legacy_reporting,
     )
@@ -252,6 +255,7 @@
     # Caller-ownership migration
     "ReportingOwnershipBackfill": ("migration", "ReportingOwnershipBackfill"),
     "ReportingOwnershipMigrationError": ("migration", "ReportingOwnershipMigrationError"),
+    "ReportingSchemaAheadError": ("migration", "ReportingSchemaAheadError"),
     "backfill_legacy_reporting": ("migration", "backfill_legacy_reporting"),
     "legacy_generation_digest": ("migration", "legacy_generation_digest"),
     "migrate_legacy_reporting": ("migration", "migrate_legacy_reporting"),
 
@@ -159,6 +159,7 @@
     ReportingRowStorageV1,
     ReportingRowStoreError,
 )
+from adcp_reporting_storage.migration import schema_ahead_error
 
 if TYPE_CHECKING:
     from psycopg_pool import AsyncConnectionPool
@@ -1028,40 +1029,46 @@ async def put_configuration(self, configuration: ReportingConfiguration) -> None
         digest = _fingerprint(payload)
         async with self._connection() as connection, connection.transaction():
             await self._lock_account(connection, key.account_id)
-            inserted_cursor = await connection.execute(
-                "INSERT INTO reporting_configurations"
-                " (delivery_config_id, delivery_config_version, account_id,"
-                "  report_definition_id, reporting_profile, feed_purpose, required_finality,"
-                "  account_timezone, schedule, media_buy_ids, activated_at, deactivated_at,"
-                "  automated_recovery_seconds, status_retention_days, definition,"
-                "  authoritative_party, content_sha256, consumer_id, quarantined)"
-                " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s, %s, %s, %s,"
-                "         %s::jsonb, %s, %s, %s, %s)"
-                " ON CONFLICT (account_id, consumer_id, delivery_config_id, delive"
-                "ry_config_version) DO NOTHING"
-                " RETURNING delivery_config_id",
-                (
-                    key.delivery_config_id,
-                    key.delivery_config_version,
-                    key.account_id,
-                    configuration.report_definition_id,
-                    configuration.reporting_profile,
-                    configuration.feed_purpose,
-                    configuration.required_finality,
-                    configuration.account_timezone,
-                    _json(payload["schedule"]),
-                    _json(sorted(configuration.media_buy_ids)),
-                    configuration.activated_at,
-                    configuration.deactivated_at,
-                    configuration.automated_recovery_window.total_seconds(),
-                    configuration.status_retention_days,
-                    _json(payload["definition"]) if payload["definition"] else None,
-                    configuration.authoritative_party,
-                    digest,
-                    configuration.consumer_id,
-                    configuration.quarantined,
-                ),
-            )
+            try:
+                inserted_cursor = await connection.execute(
+                    "INSERT INTO reporting_configurations"
+                    " (delivery_config_id, delivery_config_version, account_id,"
+                    "  report_definition_id, reporting_profile, feed_purpose, required_finality,"
+                    "  account_timezone, schedule, media_buy_ids, activated_at, deactivated_at,"
+                    "  automated_recovery_seconds, status_retention_days, definition,"
+                    "  authoritative_party, content_sha256, consumer_id, quarantined)"
+                    " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s, %s, %s, %s,"
+                    "         %s::jsonb, %s, %s, %s, %s)"
+                    " ON CONFLICT (account_id, consumer_id, delivery_config_id, delive"
+                    "ry_config_version) DO NOTHING"
+                    " RETURNING delivery_config_id",
+                    (
+                        key.delivery_config_id,
+                        key.delivery_config_version,
+                        key.account_id,
+                        configuration.report_definition_id,
+                        configuration.reporting_profile,
+                        configuration.feed_purpose,
+                        configuration.required_finality,
+                        configuration.account_timezone,
+                        _json(payload["schedule"]),
+                        _json(sorted(configuration.media_buy_ids)),
+                        configuration.activated_at,
+                        configuration.deactivated_at,
+                        configuration.automated_recovery_window.total_seconds(),
+                        configuration.status_retention_days,
+                        _json(payload["definition"]) if payload["definition"] else None,
+                        configuration.authoritative_party,
+                        digest,
+                        configuration.consumer_id,
+                        configuration.quarantined,
+                    ),
+                )
+            except Exception as error:
+                ahead = schema_ahead_error(error)
+                if ahead is not None:
+                    raise ahead from error
+                raise
             inserted = await inserted_cursor.fetchone()
             # Check *after* the insert. ON CONFLICT waits for a concurrent
             # winner; a fresh READ COMMITTED statement sees its retained
@@ -3493,6 +3500,9 @@ def _translate_integrity_error(error: Exception) -> LedgerConflictError:
     Matching the index name rather than the driver's prose: the message text
     varies by Postgres version and locale, the index name does not.
     """
+    ahead = schema_ahead_error(error)
+    if ahead is not None:
+        return LedgerConflictError("REPORTING_SCHEMA_AHEAD", str(ahead))
     text = str(error)
     if "reporting_revisions_one_official" in text:
         return LedgerConflictError(
 
@@ -29,6 +29,13 @@ BEGIN
     ] THEN
         RETURN;
     END IF;
+    -- adcp-reporting-storage 0.2+ widens the key with offering_id (storage
+    -- contract 2). This release cannot run on it, and the caller-ownership
+    -- remedy below would be wrong advice.
+    IF 'offering_id' = ANY(primary_key_columns) THEN
+        RAISE EXCEPTION 'adcp-reporting-storage 0.2 or later has upgraded this reporting ledger to storage contract 2; this release implements storage contract 1'
+            USING HINT = 'Run adcp>=10.0.0b7 with adcp-reporting-storage>=0.2.0b1 on every reporting process. Downgrading is unsupported: restore the backup taken before the upgrade. Do not run migrate_legacy_reporting.';
+    END IF;
     RAISE EXCEPTION 'Reporting caller ownership requires a stopped-worker maintenance migration'
         USING HINT = 'Use migrate_legacy_reporting; retain legacy state in quarantine and backfill authoritative ownership.';
 
 
@@ -26,6 +26,7 @@
 __all__ = [
     "ReportingOwnershipBackfill",
     "ReportingOwnershipMigrationError",
+    "ReportingSchemaAheadError",
     "backfill_legacy_reporting",
     "legacy_generation_digest",
     "migrate_legacy_reporting",
@@ -43,6 +44,41 @@ class ReportingOwnershipMigrationError(ReportingOperatorActionError):
     """
 
 
+_CONTRACT_2_REMEDY = (
+    "Run adcp>=10.0.0b7 with adcp-reporting-storage>=0.2.0b1 on every reporting process."
+    " Downgrading is unsupported: to go back, restore the backup taken before the upgrade."
+    " This is not a caller-ownership migration; do not run migrate_legacy_reporting."
+)
+# PostgreSQL: no unique index or constraint matches an ON CONFLICT target.
+_NO_MATCHING_CONFLICT_KEY = "42P10"
+
+
+class ReportingSchemaAheadError(ReportingOperatorActionError):
+    """adcp-reporting-storage 0.2 or later upgraded this ledger (storage contract 2).
+
+    This release implements storage contract 1: its generation, obligation
+    and status keys name the columns before ``offering_id`` was added, so it
+    can neither start on the upgraded ledger nor write to it. Service startup
+    re-raises it unwrapped, so the operator sees the remedy.
+    """
+
+
+def schema_ahead_error(error: BaseException) -> ReportingSchemaAheadError | None:
+    """The contract-2 refusal for a write whose conflict key the upgrade widened.
+
+    A running process of this release meets an upgraded ledger as PostgreSQL
+    42P10 on its ``ON CONFLICT`` writes, since the unique keys they name now
+    include ``offering_id``. ``None`` for any other error.
+    """
+    if getattr(error, "sqlstate", None) != _NO_MATCHING_CONFLICT_KEY:
+        return None
+    return ReportingSchemaAheadError(
+        "this reporting write names a key that a newer adcp-reporting-storage (storage contract"
+        " 2, with offering_id) has widened, so the ledger was upgraded while this process ran. "
+        + _CONTRACT_2_REMEDY
+    )
+
+
 @dataclass(frozen=True)
 class ReportingOwnershipBackfill:
     """Authoritative mapping with an explicit retained-only recovery decision.
@@ -83,9 +119,17 @@ async def require_owned_schema_or_empty(connection: Any) -> None:
             "SELECT EXISTS (SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace"
             " WHERE n.nspname=current_schema() AND c.relname='reporting_configurations'),"
             " EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=current_schema()"
-            " AND table_name='reporting_configurations' AND column_name='consumer_id')"
+            " AND table_name='reporting_configurations' AND column_name='consumer_id'),"
+            " EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=current_schema()"
+            " AND table_name='reporting_configurations' AND column_name='offering_id')"
         )
     ).fetchone()
+    if row[2]:
+        raise ReportingSchemaAheadError(
+            "adcp-reporting-storage 0.2 or later has upgraded this reporting ledger to storage"
+            " contract 2 (reporting_configurations has offering_id). This release implements"
+            " storage contract 1 and cannot run against it. " + _CONTRACT_2_REMEDY
+        )
     if row[0] and not row[1]:
         raise ReportingOwnershipMigrationError(
             "If all legacy tables are empty, call replace_empty_legacy_reporting. "
 
@@ -49,6 +49,7 @@
     persist_replay_lifecycles_on,
     settle_snapshot_on,
 )
+from adcp_reporting_storage.migration import schema_ahead_error
 from adcp_reporting_storage.outbox.pg import PgReportingOutbox, database_now
 
 _KEY = (
@@ -288,13 +289,20 @@ async def _lock_scopes_on(self, connection: Any, snapshot: ReportingStatusSnapsh
         # Insert placeholders before allocating random IDs. New and existing
         # typed scopes have exactly the same row-lock/notification boundary.
         for scope in projection_scopes(snapshot):
-            await connection.execute(
-                f"INSERT INTO reporting_status_scope_checkpoints ({_KEY}, scope, fingerprint,"  # nosec B608
-                " snapshot, source_sequence, baseline, publishable, selector_writer_floor)"
-                ' VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s,\'{"health":"waiting"}\',0,FALSE,FALSE,2)'
-                f" ON CONFLICT ({_KEY}) DO NOTHING",  # nosec B608
-                (*scope.checkpoint_key, json.dumps(asdict(scope)), "0" * 64),
-            )
+            try:
+                await connection.execute(
+                    f"INSERT INTO reporting_status_scope_checkpoints ({_KEY}, scope, fingerprint,"  # nosec B608
+                    " snapshot, source_sequence, baseline, publishable, selector_writer_floor)"
+                    " VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s,"
+                    ' \'{"health":"waiting"}\',0,FALSE,FALSE,2)'
+                    f" ON CONFLICT ({_KEY}) DO NOTHING",  # nosec B608
+                    (*scope.checkpoint_key, json.dumps(asdict(scope)), "0" * 64),
+                )
+            except Exception as error:
+                ahead = schema_ahead_error(error)
+                if ahead is not None:
+                    raise ahead from error
+                raise
         await (
             await connection.execute(
                 "SELECT 1 FROM reporting_status_scope_checkpoints WHERE account_id=%s"
 
@@ -0,0 +1,130 @@
+"""A ledger upgraded to storage contract 2 is refused clearly by this release (#1831).
+
+adcp-reporting-storage 0.2.0b1+ upgrades a 0.1.x ledger in place: migration
+0002 adds ``offering_id`` and widens the configuration key and the obligation
+period key with it. If a process of this release keeps running during a
+rolling deploy, or restarts afterwards, it must name the actual problem and
+remedy rather than fail with a raw driver error or point at the unrelated
+caller-ownership migration.
+
+``_widen`` applies the same key changes as 0002 on ``next``
+(``reporting_ledger_account_generations.sql``).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from adcp.reporting.ledger import LedgerConflictError
+
+from ._generation_support import configuration, isolated_reporting_pool, obligation_for
+
+_WIDEN = """
+ALTER TABLE reporting_configurations ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '';
+ALTER TABLE reporting_configurations DROP CONSTRAINT reporting_configurations_pkey CASCADE;
+ALTER TABLE reporting_configurations ADD CONSTRAINT reporting_configurations_pkey
+    PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id);
+ALTER TABLE reporting_obligations ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '';
+DROP INDEX reporting_obligations_period_key;
+CREATE UNIQUE INDEX reporting_obligations_period_key
+    ON reporting_obligations
+       (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id,
+        period_start, period_end);
+ALTER TABLE reporting_consumer_statuses ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '';
+"""
+
+
+async def _upgraded_store(pool):
+    from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore
+
+    store = PgReportingLedgerStore(pool=pool)
+    await store.create_schema()
+    await store.put_configuration(configuration("acct_a"))
+    async with pool.connection() as connection:
+        await connection.execute(_WIDEN)
+    return store
+
+
+def _assert_remedy(message: str) -> None:
+    assert "storage contract 2" in message
+    assert "adcp>=10.0.0b7 with adcp-reporting-storage>=0.2.0b1" in message
+    assert "restore the backup" in message
+    assert "do not run migrate_legacy_reporting" in message
+
+
+def test_only_a_missing_conflict_key_is_a_contract_2_refusal() -> None:
+    from adcp_reporting_storage.migration import schema_ahead_error
+
+    class _Error(Exception):
+        def __init__(self, sqlstate: str) -> None:
+            super().__init__(sqlstate)
+            self.sqlstate = sqlstate
+
+    assert schema_ahead_error(_Error("42P10")) is not None
+    assert schema_ahead_error(_Error("23505")) is None  # unique violation
+    assert schema_ahead_error(ValueError("no sqlstate")) is None
+
+
+@pytest.mark.asyncio
+async def test_create_schema_refuses_an_upgraded_ledger_and_changes_nothing() -> None:
+    from adcp_reporting_storage import ReportingSchemaAheadError
+
+    async with isolated_reporting_pool(autocommit=True) as pool:
+        store = await _upgraded_store(pool)
+        async with pool.connection() as connection:
+            before = await (
+                await connection.execute(
+                    "SELECT pg_get_constraintdef(oid) FROM pg_constraint"
+                    " WHERE conname = 'reporting_configurations_pkey'"
+                    " AND connamespace = current_schema()::regnamespace"
+                )
+            ).fetchall()
+        with pytest.raises(ReportingSchemaAheadError) as refused:
+            await store.create_schema()
+        _assert_remedy(str(refused.value))
+        async with pool.connection() as connection:
+            after = await (
+                await connection.execute(
+                    "SELECT pg_get_constraintdef(oid) FROM pg_constraint"
+                    " WHERE conname = 'reporting_configurations_pkey'"
+                    " AND connamespace = current_schema()::regnamespace"
+                )
+            ).fetchall()
+        assert after == before and "offering_id" in before[0][0]
+
+
+@pytest.mark.asyncio
+async def test_the_ddl_alone_names_contract_2_not_the_ownership_migration() -> None:
+    """For operators who apply the packaged SQL themselves."""
+    from importlib.resources import files
+
+    ddl = (
+        files("adcp_reporting_storage.ledger")
+        .joinpath("reporting_ledger_account_generations.sql")
+        .read_text()
+    )
+    async with isolated_reporting_pool(autocommit=True) as pool:  # skips without psycopg
+        from psycopg import errors
+
+        await _upgraded_store(pool)
+        async with pool.connection() as connection:
+            with pytest.raises(errors.RaiseException) as raised:
+                await connection.execute(ddl)
+    diagnostic = raised.value.diag
+    assert "storage contract 2" in (diagnostic.message_primary or "")
+    assert "Do not run migrate_legacy_reporting" in (diagnostic.message_hint or "")
+
+
+@pytest.mark.asyncio
+async def test_a_running_process_names_the_upgrade_on_every_widened_write() -> None:
+    from adcp_reporting_storage import ReportingSchemaAheadError
+
+    async with isolated_reporting_pool(autocommit=True) as pool:
+        store = await _upgraded_store(pool)
+        with pytest.raises(ReportingSchemaAheadError) as refused:
+            await store.put_configuration(configuration("acct_b"))
+        _assert_remedy(str(refused.value))
+        with pytest.raises(LedgerConflictError) as conflict:
+            await store.commit_obligation(obligation_for(configuration("acct_a")))
+        assert conflict.value.code == "REPORTING_SCHEMA_AHEAD"
+        _assert_remedy(str(conflict.value))
@@ -8225,6 +8225,7 @@
       "ReportingRowQuarantineReason": "union[value:'BINDING_MISMATCH',value:'ROWS_NOT_PORTABLE',value:'ROW_SEQUENCE_INVALID']",
       "ReportingRowStorageOptionsV1": "adcp_reporting_storage.ledger.row_storage",
       "ReportingRowStorageV1": "adcp_reporting_storage.ledger.row_storage",
+      "ReportingSchemaAheadError": "adcp_reporting_storage.migration",
       "assert_gcs_reporting_row_bucket_policy_v1": "adcp_reporting_storage.ledger.row_storage_gcs",
       "backfill_legacy_reporting": "adcp_reporting_storage.migration",
       "create_azure_blob_reporting_row_object_provider_v1": "adcp_reporting_storage.ledger.row_storage_azure",