fix(compat): reject unfingerprinted continuation writes

docs/legacy-purchase-continuations.md

@@ -134,7 +134,10 @@ makes older workers fail closed during a rolling upgrade instead of silently
 bypassing the new replay fence. Migrated pre-fingerprint rows cannot be
 represented safely by a compact issuance tombstone and are therefore never
 purged automatically; retain them until an operator completes migration and
-replay-risk resolution.
+replay-risk resolution. After migration, database guards reject every new
+NULL-fingerprint continuation and any attempt to remove a modern row's
+fingerprint. Existing pre-fingerprint rows remain untouched and readable, but
+pre-#1061 writers can no longer create fresh authorizations in this ledger.
 
 Configure global and per-principal record/logical-byte limits plus a per-payload
 limit for the deployment. Both byte limits count the same persisted fields,

src/adcp/compat/sqlite_continuation_store.py

@@ -101,6 +101,23 @@
     WHERE issuance_fingerprint IS NOT NULL
     """,
     """
+    CREATE TRIGGER IF NOT EXISTS adcp_compat_continuations_fingerprint_insert_guard
+    BEFORE INSERT ON adcp_compat_continuations
+    WHEN NEW.issuance_fingerprint IS NULL
+    BEGIN
+        SELECT RAISE(ABORT, 'new continuations require an issuance fingerprint');
+    END
+    """,
+    """
+    CREATE TRIGGER IF NOT EXISTS adcp_compat_continuations_fingerprint_downgrade_guard
+    BEFORE UPDATE OF issuance_fingerprint ON adcp_compat_continuations
+    WHEN OLD.issuance_fingerprint IS NOT NULL
+     AND NEW.issuance_fingerprint IS NULL
+    BEGIN
+        SELECT RAISE(ABORT, 'continuation issuance fingerprint cannot be removed');
+    END
+    """,
+    """
     CREATE TRIGGER IF NOT EXISTS adcp_compat_continuations_retired_insert_guard
     BEFORE INSERT ON adcp_compat_continuations
     WHEN EXISTS (
@@ -161,6 +178,8 @@
 )
 
 _REPLAY_FENCE_TRIGGER_NAMES = (
+    "adcp_compat_continuations_fingerprint_insert_guard",
+    "adcp_compat_continuations_fingerprint_downgrade_guard",
     "adcp_compat_continuations_retired_insert_guard",
     "adcp_compat_continuations_retired_update_guard",
     "adcp_compat_continuations_delete_guard",

tests/test_purchase_continuation.py

@@ -1185,6 +1185,8 @@ async def test_pre_fingerprint_authorization_blocks_duplicate_reissuance(tmp_pat
     first = _coordinator(store, lambda _ctx: {})
     await _issue(first, case)
     with closing(sqlite3.connect(database)) as conn, conn:
+        # Model a row written before the fingerprint guards were installed.
+        conn.execute("DROP TRIGGER adcp_compat_continuations_fingerprint_downgrade_guard")
         conn.execute(
             "UPDATE adcp_compat_continuations "
             "SET token_hash = ?, issuance_fingerprint = NULL, issuance_binding_hash = NULL",
@@ -1758,13 +1760,16 @@ def test_sqlite_store_forces_private_wal_sidecars_under_permissive_umask(
         with closing(store._connect()) as conn, conn:
             conn.execute(
                 "INSERT INTO adcp_compat_continuations ("
-                "token_hash, principal_id, account_identity, source_adcp_version, "
+                "token_hash, issuance_fingerprint, issuance_binding_hash, "
+                "principal_id, account_identity, source_adcp_version, "
                 "expires_at, observed_request_json, observed_response_json, "
                 "observed_payload_hash, product_ids_json, losses_json, target_binding, "
                 "created_at, updated_at"
-                ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                 (
                     "hash",
+                    "fingerprint",
+                    "binding-hash",
                     "principal",
                     "account",
                     "2.5",
@@ -2105,6 +2110,10 @@ async def test_sqlite_replay_fence_triggers_fail_closed_for_older_workers(
         conn.row_factory = sqlite3.Row
         continuation = dict(conn.execute("SELECT * FROM adcp_compat_continuations").fetchone())
 
+        with pytest.raises(sqlite3.IntegrityError, match="fingerprint cannot be removed"):
+            conn.execute("UPDATE adcp_compat_continuations SET issuance_fingerprint = NULL")
+        conn.rollback()
+
         # Simulate the cleanup SQL from a process that predates tombstones.
         conn.execute("BEGIN")
         conn.execute("DELETE FROM adcp_compat_operations")
@@ -2126,6 +2135,19 @@ async def test_sqlite_replay_fence_triggers_fail_closed_for_older_workers(
                 f"VALUES ({placeholders})",
                 tuple(continuation[column] for column in columns),
             )
+        legacy_continuation = {
+            **continuation,
+            "token_hash": "f" * 64,
+            "issuance_fingerprint": None,
+            "issuance_binding_hash": None,
+            "claimed_operation_id": None,
+        }
+        with pytest.raises(sqlite3.IntegrityError, match="require an issuance fingerprint"):
+            conn.execute(
+                f"INSERT INTO adcp_compat_continuations ({', '.join(columns)}) "
+                f"VALUES ({placeholders})",
+                tuple(legacy_continuation[column] for column in columns),
+            )
         with pytest.raises(sqlite3.IntegrityError, match="immutable"):
             conn.execute("UPDATE adcp_compat_issuance_tombstones SET retired_at = retired_at")
         with pytest.raises(sqlite3.IntegrityError, match="permanent"):