fix(identity): enforce binding lifecycle invariants

Fail closed across database and relay authority paths when identity state is ambiguous, stale, revoked, or unreadable.

Signed-off-by: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com>

Cargo.lock

@@ -1003,6 +1003,7 @@ dependencies = [
 name = "buzz-db"
 version = "0.1.0"
 dependencies = [
+ "buzz-auth",
  "buzz-core",
  "chrono",
  "hex",

crates/buzz-db/Cargo.toml

@@ -8,6 +8,7 @@ repository.workspace = true
 description = "Postgres event store and data access layer for Buzz"
 
 [dependencies]
+buzz-auth = { workspace = true }
 buzz-core = { workspace = true }
 sqlx        = { workspace = true }
 tokio       = { workspace = true }

crates/buzz-db/src/channel.rs

@@ -67,7 +67,7 @@ pub struct ChannelRecord {
 }
 
 /// A channel membership row as returned from the database.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct MemberRecord {
     /// The channel this membership belongs to.
     pub channel_id: Uuid,
@@ -400,7 +400,7 @@ pub async fn add_member(
 }
 
 /// Outcome of atomically adding a channel member and binding corporate identity.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ChannelAdmissionOutcome {
     /// Membership and any staged identity binding committed together.
     Joined {
@@ -413,6 +413,9 @@ pub enum ChannelAdmissionOutcome {
     IdentityConflict(IdentityBindingConflict),
     /// The staged identity principal or key is revoked.
     IdentityRevoked,
+    /// The staged identity has no active binding and lacks sealed enrollment
+    /// evidence.
+    IdentityBindingRequired,
 }
 
 /// Add a channel member and optional corporate identity binding in one transaction.
@@ -459,6 +462,10 @@ pub async fn add_member_with_identity(
                 tx.rollback().await?;
                 return Ok(ChannelAdmissionOutcome::IdentityRevoked);
             }
+            Ok(BindIdentityResult::BindingRequired) => {
+                tx.rollback().await?;
+                return Ok(ChannelAdmissionOutcome::IdentityBindingRequired);
+            }
             Err(error) => {
                 tx.rollback().await?;
                 return Err(error);
@@ -1806,14 +1813,20 @@ mod tests {
         )
         .await
         .expect("create private huddle");
-        crate::identity_binding::bind_or_validate_identity(
+        crate::identity_binding::resolve_identity_binding(
             &pool,
-            community,
-            "https://idp.example",
-            "conflicting-principal",
-            &bound_key,
-            Some("bound@example.com"),
-            crate::identity_binding::SOURCE_JWT_NPUB,
+            &crate::identity_binding::ResolveBindingInput {
+                authorization_domain: community,
+                issuer: "https://idp.example",
+                subject: "conflicting-principal",
+                pubkey: &bound_key,
+                display_name: Some("bound@example.com"),
+                enrollment_mode: crate::identity_binding::EnrollmentMode::AttestedKey,
+                key_attested: true,
+                policy_version: "channel-test-policy-v1",
+                evidence_valid_from: 0,
+                evidence_valid_until: i64::MAX as u64,
+            },
         )
         .await
         .expect("seed conflicting binding");
@@ -1843,7 +1856,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore = "requires Postgres"]
-    async fn atomic_huddle_admission_identity_storage_failure_rolls_back_membership() {
+    async fn atomic_huddle_admission_raw_identity_cannot_enroll_and_rolls_back_membership() {
         let pool = setup_pool().await;
         let community_id = make_test_community(&pool).await;
         let community = CommunityId::from_uuid(community_id);
@@ -1852,7 +1865,7 @@ mod tests {
         let channel = create_test_channel(
             &pool,
             community_id,
-            "atomic-identity-storage-failure",
+            "atomic-identity-binding-required",
             ChannelType::Stream,
             ChannelVisibility::Private,
             None,
@@ -1861,28 +1874,9 @@ mod tests {
         )
         .await
         .expect("create private huddle");
-        let suffix = community_id.simple();
-        let function_name = format!("buzz_test_fail_identity_{suffix}");
-        let trigger_name = format!("buzz_test_fail_identity_insert_{suffix}");
-        // Identifiers and the literal UUID below are derived only from a generated UUID.
-        sqlx::query(sqlx::AssertSqlSafe(format!(
-            "CREATE FUNCTION {function_name}() RETURNS trigger LANGUAGE plpgsql AS $$ \
-             BEGIN RAISE EXCEPTION 'injected identity storage failure'; END $$"
-        )))
-        .execute(&pool)
-        .await
-        .expect("create failure function");
-        sqlx::query(sqlx::AssertSqlSafe(format!(
-            "CREATE TRIGGER {trigger_name} BEFORE INSERT ON identity_bindings \
-             FOR EACH ROW WHEN (NEW.community_id = '{community_id}'::uuid) \
-             EXECUTE FUNCTION {function_name}()"
-        )))
-        .execute(&pool)
-        .await
-        .expect("create failure trigger");
-        let identity = identity_for(&joiner, "storage-failure");
+        let identity = identity_for(&joiner, "binding-required");
 
-        let result = add_member_with_identity(
+        let outcome = add_member_with_identity(
             &pool,
             community,
             channel.id,
@@ -1891,22 +1885,10 @@ mod tests {
             Some(&owner),
             Some(&identity),
         )
-        .await;
-
-        sqlx::query(sqlx::AssertSqlSafe(format!(
-            "DROP TRIGGER {trigger_name} ON identity_bindings"
-        )))
-        .execute(&pool)
         .await
-        .expect("drop failure trigger");
-        sqlx::query(sqlx::AssertSqlSafe(format!(
-            "DROP FUNCTION {function_name}()"
-        )))
-        .execute(&pool)
-        .await
-        .expect("drop failure function");
+        .expect("typed binding-required outcome");
 
-        assert!(matches!(result, Err(DbError::Sqlx(_))), "{result:?}");
+        assert_eq!(outcome, ChannelAdmissionOutcome::IdentityBindingRequired);
         assert_eq!(
             active_membership_count(&pool, community, channel.id, &joiner).await,
             0
@@ -1943,6 +1925,23 @@ mod tests {
         .await
         .expect("create private huddle");
         let identity = identity_for(&joiner, "successful-principal");
+        crate::identity_binding::resolve_identity_binding(
+            &pool,
+            &crate::identity_binding::ResolveBindingInput {
+                authorization_domain: community,
+                issuer: identity.issuer,
+                subject: identity.uid,
+                pubkey: identity.pubkey,
+                display_name: identity.display_name,
+                enrollment_mode: crate::identity_binding::EnrollmentMode::AttestedKey,
+                key_attested: true,
+                policy_version: "channel-test-policy-v1",
+                evidence_valid_from: 0,
+                evidence_valid_until: i64::MAX as u64,
+            },
+        )
+        .await
+        .expect("seed verified binding");
 
         let first = add_member_with_identity(
             &pool,
@@ -1958,7 +1957,7 @@ mod tests {
         assert!(matches!(
             first,
             ChannelAdmissionOutcome::Joined {
-                identity_binding: Some(BindIdentityResult::Created),
+                identity_binding: Some(BindIdentityResult::Matched),
                 ..
             }
         ));
@@ -1987,7 +1986,8 @@ mod tests {
         );
         let binding_count: i64 = sqlx::query_scalar(
             "SELECT COUNT(*) FROM identity_bindings \
-             WHERE community_id = $1 AND pubkey = $2 AND revoked_at IS NULL",
+             WHERE community_id = $1 AND pubkey = $2 \
+               AND binding_state = 'active' AND revoked_at IS NULL",
         )
         .bind(community.as_uuid())
         .bind(&joiner)

crates/buzz-db/src/identity_lifecycle_deterministic_tests.rs

@@ -27,6 +27,7 @@ enum Action {
     Disable,
     Revoke,
     Enable,
+    Archive,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -42,6 +43,7 @@ struct Fixture {
     community_id: CommunityId,
     expected_pending: Option<PendingLineage>,
     enrollment_key: [u8; 32],
+    archive_binding_id: Option<Uuid>,
 }
 
 async fn setup_pool() -> PgPool {
@@ -76,9 +78,10 @@ fn principal() -> IdentityPrincipal<'static> {
 }
 
 fn context(operation_id: Uuid, reason: &'static str) -> LifecycleContext<'static> {
+    const ACTOR: [u8; 32] = [0xA1; 32];
     LifecycleContext {
-        operation_id,
-        actor: None,
+        operation_id: LifecycleOperationId::from_uuid_for_test(operation_id),
+        actor: &ACTOR,
         reason,
     }
 }
@@ -88,27 +91,37 @@ fn replacement(pubkey: &'static [u8; 32]) -> VerifiedReplacementKey<'static> {
         pubkey,
         None,
         BindingProvenance::AttestedKey,
-        Some("deterministic-policy-v1"),
+        "deterministic-policy-v1",
     )
     .expect("construct deterministic replacement")
 }
 
-async fn enroll_key(pool: &PgPool, community_id: CommunityId, pubkey: &[u8]) {
+async fn enroll_key(
+    pool: &PgPool,
+    community_id: CommunityId,
+    pubkey: &[u8],
+) -> crate::identity_binding::BindingEvidence {
     let result = resolve_identity_binding(
         pool,
-        community_id,
         &ResolveBindingInput {
+            authorization_domain: community_id,
             issuer: ISSUER,
             subject: SUBJECT,
             pubkey,
             display_name: None,
             enrollment_mode: EnrollmentMode::AttestedKey,
             key_attested: true,
+            policy_version: "deterministic-policy-v1",
+            evidence_valid_from: 0,
+            evidence_valid_until: i64::MAX as u64,
         },
     )
     .await
     .expect("seed enrollment");
-    assert!(matches!(result, ResolveBindingResult::Enrolled(_)));
+    match result {
+        ResolveBindingResult::Enrolled(evidence) => evidence,
+        other => panic!("expected deterministic enrollment, got {other:?}"),
+    }
 }
 
 fn pair_contains(pair: (Action, Action), action: Action) -> bool {
@@ -119,8 +132,24 @@ async fn setup_fixture(pool: &PgPool, pair: (Action, Action), label: &str) -> Fi
     let community_id = make_community(pool, label).await;
     let mut expected_pending = None;
     let mut enrollment_key = ENROLL_KEY;
+    let mut archive_binding_id = None;
 
-    if pair_contains(pair, Action::Enroll) && pair_contains(pair, Action::Rotate) {
+    if pair_contains(pair, Action::Archive) {
+        let evidence = enroll_key(pool, community_id, &OLD_KEY).await;
+        retire_identity_pair(
+            pool,
+            community_id,
+            context(Uuid::from_u128(9), "prepare archive recovery"),
+            principal(),
+            &OLD_KEY,
+        )
+        .await
+        .expect("prepare archivable pending recovery");
+        expected_pending = get_pending_lineage(pool, community_id, principal())
+            .await
+            .expect("read archive recovery selector");
+        archive_binding_id = Some(evidence.binding_id());
+    } else if pair_contains(pair, Action::Enroll) && pair_contains(pair, Action::Rotate) {
         enrollment_key = OLD_KEY;
     } else if pair_contains(pair, Action::Enroll) && pair_contains(pair, Action::Recover) {
         enroll_key(pool, community_id, &OLD_KEY).await;
@@ -171,6 +200,7 @@ async fn setup_fixture(pool: &PgPool, pair: (Action, Action), label: &str) -> Fi
         community_id,
         expected_pending,
         enrollment_key,
+        archive_binding_id,
     }
 }
 
@@ -183,14 +213,17 @@ async fn run_action(
     match action {
         Action::Enroll => match resolve_identity_binding(
             pool,
-            fixture.community_id,
             &ResolveBindingInput {
+                authorization_domain: fixture.community_id,
                 issuer: ISSUER,
                 subject: SUBJECT,
                 pubkey: &fixture.enrollment_key,
                 display_name: None,
                 enrollment_mode: EnrollmentMode::AttestedKey,
                 key_attested: true,
+                policy_version: "deterministic-policy-v1",
+                evidence_valid_from: 0,
+                evidence_valid_until: i64::MAX as u64,
             },
         )
         .await
@@ -201,7 +234,9 @@ async fn run_action(
                 BindingDenial::Conflict
                 | BindingDenial::Revoked
                 | BindingDenial::BindingRequired
-                | BindingDenial::KeyAttestationRequired,
+                | BindingDenial::KeyAttestationRequired
+                | BindingDenial::BindingExpired
+                | BindingDenial::StaleEvidence,
             )) => Outcome::Denied,
             Err(_) => Outcome::Error,
         },
@@ -265,6 +300,19 @@ async fn run_action(
         )
         .await
         .map_or(Outcome::Error, |_| Outcome::Applied),
+        Action::Archive => {
+            let Some(binding_id) = fixture.archive_binding_id else {
+                return Outcome::Error;
+            };
+            archive_identity_binding(
+                pool,
+                fixture.community_id,
+                context(operation_id, "deterministic archive"),
+                binding_id,
+            )
+            .await
+            .map_or(Outcome::Error, |_| Outcome::Applied)
+        }
     }
 }
 
@@ -287,7 +335,7 @@ async fn normalized_rows(
 
 async fn logical_snapshot(pool: &PgPool, community_id: CommunityId) -> Vec<String> {
     let queries = [
-        "SELECT jsonb_build_object('t','binding','v',to_jsonb(b)-ARRAY['community_id','binding_id','replacement_binding_id','created_at','updated_at','last_seen_at','revoked_at','revoked_by','rotation_completed_at','rotation_by']::text[]) FROM identity_bindings b WHERE community_id=$1",
+        "SELECT jsonb_build_object('t','binding','v',to_jsonb(b)-ARRAY['community_id','binding_id','replacement_binding_id','created_at','updated_at','last_seen_at','revoked_at','revoked_by','rotation_completed_at','rotation_by','archived_at']::text[]) FROM identity_bindings b WHERE community_id=$1",
         "SELECT jsonb_build_object('t','principal','v',jsonb_build_object('issuer',issuer,'subject',uid,'disabled',disabled_at IS NOT NULL,'reason',disabled_reason)) FROM identity_principals WHERE community_id=$1",
         "SELECT jsonb_build_object('t','revoked_key','v',jsonb_build_object('pubkey',encode(pubkey,'hex'),'reason',reason)) FROM identity_revoked_keys WHERE community_id=$1",
         "SELECT jsonb_build_object('t','retired','v',to_jsonb(r)-ARRAY['community_id','retired_binding_id','retired_at','retired_by']::text[]) FROM identity_retired_pairs r WHERE community_id=$1",
@@ -664,71 +712,308 @@ async fn enrollment_rotate_and_recover_have_forced_orders_and_references() {
     }
 }
 
+#[tokio::test]
+#[ignore = "requires a dedicated disposable Postgres database"]
+async fn archive_and_recovery_have_forced_orders_and_references() {
+    let pool = setup_pool().await;
+    exercise_ordered_case(&pool, Action::Archive, Action::Recover).await;
+    exercise_ordered_case(&pool, Action::Recover, Action::Archive).await;
+}
+
 #[tokio::test]
 #[ignore = "requires a dedicated disposable Postgres database"]
 async fn compare_and_clear_rejects_aba_recreation_and_preserves_domain_b() {
     let pool = setup_pool().await;
     let fixture = setup_fixture(&pool, (Action::Recover, Action::Disable), "identity-aba").await;
     let expected = fixture.expected_pending.expect("pending selector");
+    let independently_observed = get_pending_lineage(&pool, fixture.community_id, principal())
+        .await
+        .expect("read second stale G1 observation")
+        .expect("G1 remains active before either actor");
+    assert!(expected == independently_observed);
     let domain_b = make_community(&pool, "identity-aba-domain-b").await;
     enroll_key(&pool, domain_b, &DOMAIN_B_KEY).await;
     let domain_b_bytes = raw_domain_snapshot(&pool, domain_b).await;
     let domain_b_auth = authorization_sentinel(&pool, domain_b).await;
+    let history_before: i64 =
+        sqlx::query_scalar("SELECT COUNT(*) FROM identity_binding_history WHERE community_id=$1")
+            .bind(fixture.community_id.as_uuid())
+            .fetch_one(&pool)
+            .await
+            .expect("count pre-ABA history");
+    let operations_before: i64 = sqlx::query_scalar(
+        "SELECT COUNT(*) FROM identity_lifecycle_operations WHERE community_id=$1",
+    )
+    .bind(fixture.community_id.as_uuid())
+    .fetch_one(&pool)
+    .await
+    .expect("count pre-ABA operations");
 
-    let mut tx = pool.begin().await.expect("begin ABA transaction");
-    sqlx::query(
-        "UPDATE identity_pending_replacements SET cleared_at=NOW(), cleared_operation_id=$4 \
-         WHERE community_id=$1 AND issuer=$2 AND subject=$3 AND cleared_at IS NULL",
-    )
-    .bind(fixture.community_id.as_uuid())
-    .bind(ISSUER)
-    .bind(SUBJECT)
-    .bind(Uuid::from_u128(200))
-    .execute(&mut *tx)
-    .await
-    .expect("clear G1");
-    sqlx::query(
-        "INSERT INTO identity_pending_replacements \
-         (community_id,issuer,subject,selector_version,retired_pubkey,retired_binding_id,retired_binding_version,created_operation_id) \
-         VALUES ($1,$2,$3,$4,$5,$6,$7,$8)",
-    )
-    .bind(fixture.community_id.as_uuid())
-    .bind(ISSUER)
-    .bind(SUBJECT)
-    .bind(i64::try_from(expected.selector_version + 1).expect("selector fits i64"))
-    .bind(&expected.retired_pubkey)
-    .bind(expected.retired_binding_id)
-    .bind(i64::try_from(expected.retired_binding_version).expect("version fits i64"))
-    .bind(Uuid::from_u128(201))
-    .execute(&mut *tx)
-    .await
-    .expect("recreate G2");
-    assert!(compare_and_clear_pending_tx(
-        &mut tx,
-        fixture.community_id,
-        context(Uuid::from_u128(202), "stale ABA compare"),
-        principal(),
-        &expected,
-    )
-    .await
-    .is_err());
-    let active_selector: i64 = sqlx::query_scalar(
-        "SELECT selector_version FROM identity_pending_replacements \
-         WHERE community_id=$1 AND issuer=$2 AND subject=$3 AND cleared_at IS NULL",
-    )
-    .bind(fixture.community_id.as_uuid())
-    .bind(ISSUER)
-    .bind(SUBJECT)
-    .fetch_one(&mut *tx)
-    .await
-    .expect("G2 remains active");
-    assert_eq!(
-        active_selector,
-        i64::try_from(expected.selector_version + 1).unwrap()
-    );
-    tx.rollback()
+    let (mut events, _controller) = test_lock_schedule::install();
+    let winner_pool = pool.clone();
+    let winner_expected = expected.clone();
+    let winner_community = fixture.community_id;
+    let winner_task = tokio::spawn(test_lock_schedule::actor_scope("aba-winner", async move {
+        let winner_context = context(Uuid::from_u128(200), "committed ABA recreation");
+        let coordinates = lifecycle_coordinates(
+            winner_community,
+            winner_context,
+            Some(principal()),
+            &[],
+            None,
+        );
+        let mut tx = begin_locked(&winner_pool, coordinates)
+            .await
+            .expect("begin locked ABA winner");
+        let cleared = sqlx::query(
+            "UPDATE identity_pending_replacements \
+             SET cleared_at=NOW(),cleared_operation_id=$8 \
+             WHERE community_id=$1 AND issuer=$2 AND subject=$3 \
+               AND retired_pubkey=$4 AND retired_binding_id=$5 \
+               AND retired_binding_version=$6 AND selector_version=$7 \
+               AND cleared_at IS NULL",
+        )
+        .bind(winner_community.as_uuid())
+        .bind(ISSUER)
+        .bind(SUBJECT)
+        .bind(&winner_expected.retired_pubkey)
+        .bind(winner_expected.retired_binding_id)
+        .bind(i64::try_from(winner_expected.retired_binding_version).unwrap())
+        .bind(i64::try_from(winner_expected.selector_version).unwrap())
+        .bind(winner_context.operation_id.as_uuid())
+        .execute(&mut *tx)
         .await
-        .expect("rollback synthetic ABA recreation");
+        .expect("clear committed G1");
+        assert_eq!(cleared.rows_affected(), 1);
+        sqlx::query(
+            "INSERT INTO identity_pending_replacements \
+             (community_id,issuer,subject,selector_version,retired_pubkey, \
+              retired_binding_id,retired_binding_version,created_operation_id) \
+             VALUES ($1,$2,$3,$4,$5,$6,$7,$8)",
+        )
+        .bind(winner_community.as_uuid())
+        .bind(ISSUER)
+        .bind(SUBJECT)
+        .bind(i64::try_from(winner_expected.selector_version + 1).unwrap())
+        .bind(&winner_expected.retired_pubkey)
+        .bind(winner_expected.retired_binding_id)
+        .bind(i64::try_from(winner_expected.retired_binding_version).unwrap())
+        .bind(Uuid::from_u128(201))
+        .execute(&mut *tx)
+        .await
+        .expect("create semantically equal G2");
+        tx.commit().await.expect("durably commit G1-to-G2 ABA");
+    }));
+    let winner_request = events.recv().await.expect("winner lock request");
+    assert_eq!(
+        (winner_request.actor(), winner_request.phase()),
+        ("aba-winner", test_lock_schedule::LockPhase::Request)
+    );
+    let winner_pid = winner_request.backend_pid();
+    let database_oid = winner_request.database_oid();
+    let winner_lock_keys = winner_request.lock_keys().to_vec();
+    winner_request.resume();
+    let winner_acquired = events.recv().await.expect("winner lock acquired");
+    assert_eq!(
+        (winner_acquired.actor(), winner_acquired.phase()),
+        ("aba-winner", test_lock_schedule::LockPhase::Acquired)
+    );
+    let winner_transaction_id = winner_acquired
+        .transaction_id()
+        .expect("winner transaction assigned");
+
+    let loser_pool = pool.clone();
+    let loser_expected = independently_observed.clone();
+    let loser_community = fixture.community_id;
+    let loser_task = tokio::spawn(test_lock_schedule::actor_scope("aba-loser", async move {
+        let loser_context = context(Uuid::from_u128(202), "stale committed ABA compare");
+        let coordinates =
+            lifecycle_coordinates(loser_community, loser_context, Some(principal()), &[], None);
+        let mut tx = begin_locked(&loser_pool, coordinates)
+            .await
+            .expect("begin locked ABA loser");
+        let g2_before: String = sqlx::query_scalar(
+            "SELECT to_jsonb(row_value)::TEXT FROM identity_pending_replacements row_value \
+             WHERE community_id=$1 AND issuer=$2 AND subject=$3 AND cleared_at IS NULL",
+        )
+        .bind(loser_community.as_uuid())
+        .bind(ISSUER)
+        .bind(SUBJECT)
+        .fetch_one(&mut *tx)
+        .await
+        .expect("read committed G2 before stale compare");
+        let exact_stale_match: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM identity_pending_replacements \
+             WHERE community_id=$1 AND issuer=$2 AND subject=$3 \
+               AND retired_pubkey=$4 AND retired_binding_id=$5 \
+               AND retired_binding_version=$6 AND selector_version=$7 \
+               AND cleared_at IS NULL",
+        )
+        .bind(loser_community.as_uuid())
+        .bind(ISSUER)
+        .bind(SUBJECT)
+        .bind(&loser_expected.retired_pubkey)
+        .bind(loser_expected.retired_binding_id)
+        .bind(i64::try_from(loser_expected.retired_binding_version).unwrap())
+        .bind(i64::try_from(loser_expected.selector_version).unwrap())
+        .fetch_one(&mut *tx)
+        .await
+        .expect("count stale G1 tuple at compare point");
+        assert_eq!(exact_stale_match, 0, "stale compare must affect zero rows");
+        let error = compare_and_clear_pending_tx(
+            &mut tx,
+            loser_community,
+            loser_context,
+            principal(),
+            &loser_expected,
+        )
+        .await
+        .expect_err("committed G2 must reject stale G1 compare");
+        assert!(error
+            .to_string()
+            .contains("pending identity lineage changed concurrently"));
+        let g2_after: String = sqlx::query_scalar(
+            "SELECT to_jsonb(row_value)::TEXT FROM identity_pending_replacements row_value \
+             WHERE community_id=$1 AND issuer=$2 AND subject=$3 AND cleared_at IS NULL",
+        )
+        .bind(loser_community.as_uuid())
+        .bind(ISSUER)
+        .bind(SUBJECT)
+        .fetch_one(&mut *tx)
+        .await
+        .expect("read G2 after stale compare");
+        assert_eq!(g2_after, g2_before);
+        tx.rollback().await.expect("rollback stale ABA actor");
+        g2_after
+    }));
+    let loser_request = events.recv().await.expect("loser lock request");
+    assert_eq!(
+        (loser_request.actor(), loser_request.phase()),
+        ("aba-loser", test_lock_schedule::LockPhase::Request)
+    );
+    let loser_pid = loser_request.backend_pid();
+    let shared_keys = winner_lock_keys
+        .iter()
+        .copied()
+        .filter(|key| loser_request.lock_keys().contains(key))
+        .collect::<Vec<_>>();
+    assert!(!shared_keys.is_empty());
+    loser_request.resume();
+    wait_for_advisory_waiter(&pool, database_oid, winner_pid, loser_pid, &shared_keys).await;
+    assert_eq!(raw_domain_snapshot(&pool, domain_b).await, domain_b_bytes);
+    assert_eq!(authorization_sentinel(&pool, domain_b).await, domain_b_auth);
+
+    winner_acquired.resume();
+    winner_task.await.expect("join committed ABA winner");
+    let loser_acquired = events
+        .recv()
+        .await
+        .expect("loser lock acquired after commit");
+    assert_eq!(
+        (loser_acquired.actor(), loser_acquired.phase()),
+        ("aba-loser", test_lock_schedule::LockPhase::Acquired)
+    );
+    assert_ne!(
+        loser_acquired.transaction_id(),
+        Some(winner_transaction_id),
+        "ABA actors must use distinct transactions"
+    );
+    loser_acquired.resume();
+    let g2_from_loser = loser_task.await.expect("join stale ABA loser");
+
+    let final_g2: String = sqlx::query_scalar(
+        "SELECT to_jsonb(row_value)::TEXT FROM identity_pending_replacements row_value \
+         WHERE community_id=$1 AND issuer=$2 AND subject=$3 AND cleared_at IS NULL",
+    )
+    .bind(fixture.community_id.as_uuid())
+    .bind(ISSUER)
+    .bind(SUBJECT)
+    .fetch_one(&pool)
+    .await
+    .expect("read final committed G2");
+    assert_eq!(final_g2, g2_from_loser);
+    let final_lineage = get_pending_lineage(&pool, fixture.community_id, principal())
+        .await
+        .expect("read final G2")
+        .expect("G2 remains active");
+    assert_eq!(
+        final_lineage.selector_version,
+        expected.selector_version + 1
+    );
+    assert_eq!(final_lineage.retired_pubkey, expected.retired_pubkey);
+    assert_eq!(
+        final_lineage.retired_binding_id,
+        expected.retired_binding_id
+    );
+    assert_eq!(
+        final_lineage.retired_binding_version,
+        expected.retired_binding_version
+    );
+    let active_pending: i64 = sqlx::query_scalar(
+        "SELECT COUNT(*) FROM identity_pending_replacements \
+         WHERE community_id=$1 AND issuer=$2 AND subject=$3 AND cleared_at IS NULL",
+    )
+    .bind(fixture.community_id.as_uuid())
+    .bind(ISSUER)
+    .bind(SUBJECT)
+    .fetch_one(&pool)
+    .await
+    .expect("count final active G2");
+    assert_eq!(active_pending, 1);
+    let cleared_g1: i64 = sqlx::query_scalar(
+        "SELECT COUNT(*) FROM identity_pending_replacements \
+         WHERE community_id=$1 AND issuer=$2 AND subject=$3 \
+           AND selector_version=$4 AND cleared_at IS NOT NULL",
+    )
+    .bind(fixture.community_id.as_uuid())
+    .bind(ISSUER)
+    .bind(SUBJECT)
+    .bind(i64::try_from(expected.selector_version).unwrap())
+    .fetch_one(&pool)
+    .await
+    .expect("count durably cleared G1");
+    assert_eq!(cleared_g1, 1);
+    assert_eq!(
+        sqlx::query_scalar::<_, i64>(
+            "SELECT COUNT(*) FROM identity_binding_history WHERE community_id=$1",
+        )
+        .bind(fixture.community_id.as_uuid())
+        .fetch_one(&pool)
+        .await
+        .expect("count post-ABA history"),
+        history_before
+    );
+    assert_eq!(
+        sqlx::query_scalar::<_, i64>(
+            "SELECT COUNT(*) FROM identity_lifecycle_operations WHERE community_id=$1",
+        )
+        .bind(fixture.community_id.as_uuid())
+        .fetch_one(&pool)
+        .await
+        .expect("count post-ABA operations"),
+        operations_before
+    );
+    let fresh_attempt = resolve_identity_binding(
+        &pool,
+        &ResolveBindingInput {
+            authorization_domain: fixture.community_id,
+            issuer: ISSUER,
+            subject: SUBJECT,
+            pubkey: &ENABLE_KEY,
+            display_name: None,
+            enrollment_mode: EnrollmentMode::AttestedKey,
+            key_attested: true,
+            policy_version: "deterministic-policy-v1",
+            evidence_valid_from: 0,
+            evidence_valid_until: i64::MAX as u64,
+        },
+    )
+    .await
+    .expect("pending G2 denies routine enrollment");
+    assert_eq!(
+        fresh_attempt,
+        ResolveBindingResult::Denied(BindingDenial::Revoked)
+    );
 
     assert_eq!(raw_domain_snapshot(&pool, domain_b).await, domain_b_bytes);
     assert_eq!(authorization_sentinel(&pool, domain_b).await, domain_b_auth);

crates/buzz-db/src/lib.rs

@@ -643,6 +643,14 @@ pub struct TokenSummary {
 }
 
 impl Db {
+    /// Build the single PostgreSQL authority adapter for application-root
+    /// authorization runtime composition.
+    pub fn federated_authority_adapter(
+        &self,
+    ) -> identity_binding::PostgresFederatedAuthorityAdapter {
+        identity_binding::PostgresFederatedAuthorityAdapter::new(self.pool.clone())
+    }
+
     /// Creates a new `Db` by connecting a Postgres pool with the given config.
     ///
     /// When `config.read_database_url` is set, a second pool with the same
@@ -2597,14 +2605,16 @@ impl Db {
     pub async fn revoke_identity_principal(
         &self,
         community_id: CommunityId,
+        operation_id: identity_lifecycle::LifecycleOperationId,
         issuer: &str,
         uid: &str,
-        revoked_by: Option<&[u8]>,
+        revoked_by: &[u8],
         reason: &str,
     ) -> Result<bool> {
         identity_binding::revoke_identity_principal(
             &self.pool,
             community_id,
+            operation_id,
             issuer,
             uid,
             revoked_by,
@@ -2617,12 +2627,20 @@ impl Db {
     pub async fn revoke_identity_key(
         &self,
         community_id: CommunityId,
+        operation_id: identity_lifecycle::LifecycleOperationId,
         pubkey: &[u8],
-        revoked_by: Option<&[u8]>,
+        revoked_by: &[u8],
         reason: &str,
     ) -> Result<bool> {
-        identity_binding::revoke_identity_key(&self.pool, community_id, pubkey, revoked_by, reason)
-            .await
+        identity_binding::revoke_identity_key(
+            &self.pool,
+            community_id,
+            operation_id,
+            pubkey,
+            revoked_by,
+            reason,
+        )
+        .await
     }
 
     /// Atomically rotate a corporate principal to a replacement key.
@@ -2630,24 +2648,22 @@ impl Db {
     pub async fn rotate_identity_binding(
         &self,
         community_id: CommunityId,
+        operation_id: identity_lifecycle::LifecycleOperationId,
         issuer: &str,
         uid: &str,
         old_pubkey: &[u8],
-        new_pubkey: &[u8],
-        display_name: Option<&str>,
-        source: &str,
-        rotated_by: Option<&[u8]>,
+        replacement: identity_lifecycle::VerifiedReplacementKey<'_>,
+        rotated_by: &[u8],
         reason: &str,
     ) -> Result<()> {
         identity_binding::rotate_identity_binding(
             &self.pool,
             community_id,
+            operation_id,
             issuer,
             uid,
             old_pubkey,
-            new_pubkey,
-            display_name,
-            source,
+            replacement,
             rotated_by,
             reason,
         )

crates/buzz-db/src/migration.rs

@@ -959,15 +959,19 @@ mod tests {
             .as_str()
             .contains("CREATE TABLE identity_revoked_keys"));
 
-        // O3 is a brownfield-safe additive projection over the frozen 0027/0028
-        // identity tables. It must not rewrite or discard legacy authority.
+        // Migration 0029 is a brownfield-safe additive projection over the
+        // frozen 0027/0028 identity tables. It must not rewrite or discard
+        // legacy authority.
         assert_eq!(migrations[28].version, 29);
-        let o3 = migrations[28].sql.as_str();
+        let projection = migrations[28].sql.as_str();
         for required in [
             "ADD COLUMN binding_id",
             "ADD COLUMN binding_version",
             "ADD COLUMN binding_state",
             "ADD COLUMN binding_provenance",
+            "ADD COLUMN expires_at",
+            "CREATE TABLE identity_enrollment_policies",
+            "CREATE TRIGGER identity_enrollment_policy_lineage_guard",
             "CREATE TABLE identity_retired_pairs",
             "CREATE TABLE identity_pending_replacements",
             "CREATE TABLE identity_binding_history",
@@ -975,13 +979,13 @@ mod tests {
             "CREATE TABLE identity_migration_denials",
         ] {
             assert!(
-                o3.contains(required),
+                projection.contains(required),
                 "migration 0029 is missing {required}"
             );
         }
     }
 
-    fn o3_executable_sql(sql: &str) -> String {
+    fn additive_identity_executable_sql(sql: &str) -> String {
         let mut output = String::with_capacity(sql.len());
         let mut chars = sql.chars().peekable();
         let mut in_line_comment = false;
@@ -1019,7 +1023,7 @@ mod tests {
             .iter()
             .find(|migration| migration.version == 29)
             .expect("migration 0029");
-        let executable = o3_executable_sql(migration.sql.as_str());
+        let executable = additive_identity_executable_sql(migration.sql.as_str());
         let normalized = normalize_sql(&executable);
 
         for forbidden in [" rename ", " drop ", " truncate ", " delete "] {
@@ -1038,6 +1042,7 @@ mod tests {
             "binding_state",
             "binding_provenance",
             "replacement_binding_id",
+            "creation_attribution_kind",
         ];
         for statement in split_sql_statements(&executable) {
             let statement = normalize_sql(&statement);
@@ -1061,6 +1066,51 @@ mod tests {
         }
     }
 
+    #[test]
+    fn migration_0029_enforces_authoritative_binding_state_and_attribution() {
+        let migration = MIGRATOR
+            .iter()
+            .find(|migration| migration.version == 29)
+            .expect("migration 0029");
+        // This gate intentionally checks persisted enum literals; unlike the
+        // additive-mutation scanner above, it must retain SQL string contents.
+        let normalized = normalize_sql(migration.sql.as_str());
+
+        assert!(
+            normalized.contains("binding_state = 'active' and revoked_at is null"),
+            "every authoritative partial index and lookup contract must require active state"
+        );
+        assert!(
+            normalized.contains("'archived'"),
+            "migration 0029 must represent archived bindings explicitly"
+        );
+        assert!(
+            normalized.contains("creation_attribution_kind"),
+            "migration 0029 must distinguish verified creation attribution from legacy unknowns"
+        );
+        assert!(
+            !normalized.contains("set binding_provenance = 'provisioned'"),
+            "legacy db_binding rows must remain tofu, including imported successors"
+        );
+
+        let desired = normalize_sql(include_str!("../../../schema/schema.sql"));
+        for required in [
+            "binding_state = 'active' and revoked_at is null",
+            "creation_attribution_kind",
+            "archived_at",
+            "archived_by",
+            "archived_reason",
+            "identity_enrollment_policies",
+            "identity_enrollment_policy_lineage_guard",
+            "expires_at",
+        ] {
+            assert!(
+                desired.contains(required),
+                "desired schema is missing identity-binding authority invariant: {required}"
+            );
+        }
+    }
+
     #[test]
     fn migration_lint_detects_tables_missing_community_id_by_default() {
         let sql = r#"
@@ -1260,7 +1310,8 @@ mod tests {
             "SELECT to_jsonb(binding) - ARRAY[\
                 'binding_id', 'binding_version', 'binding_state',\
                 'binding_provenance', 'replacement_binding_id', 'created_by',\
-                'created_policy_version']::text[] \
+                'created_policy_version', 'expires_at', 'creation_attribution_kind',\
+                'archived_at', 'archived_by', 'archived_reason']::text[] \
              FROM identity_bindings binding",
         )
         .await;
@@ -1635,7 +1686,7 @@ mod tests {
                ON replacement.community_id=binding.community_id \
               AND replacement.binding_id=binding.replacement_binding_id \
              WHERE binding.community_id=$1 AND binding.issuer='https://idp.example' AND binding.uid='chain' \
-             ORDER BY binding.binding_version",
+             ORDER BY binding.pubkey",
         )
         .bind(domain_a)
         .fetch_all(&pool)
@@ -1655,14 +1706,14 @@ mod tests {
             chain[1],
             (
                 chain_keys[1].clone(),
-                2,
+                1,
                 Some(chain_keys[2].clone()),
-                "provisioned".to_owned()
+                "tofu".to_owned()
             )
         );
         assert_eq!(
             chain[2],
-            (chain_keys[2].clone(), 3, None, "provisioned".to_owned())
+            (chain_keys[2].clone(), 1, None, "tofu".to_owned())
         );
 
         let pending: (Vec<u8>, i64, i64) = sqlx::query_as(
@@ -1714,15 +1765,15 @@ mod tests {
                 &clean_replacement_key,
                 None,
                 crate::identity_binding::BindingProvenance::Provisioned,
-                Some("migration-test-policy"),
+                "migration-test-policy",
             )
             .expect("construct clean rotation replacement");
         assert!(crate::identity_lifecycle::rotate_identity_binding(
             &pool,
             buzz_core::CommunityId::from_uuid(domain_a),
             crate::identity_lifecycle::LifecycleContext {
-                operation_id: uuid::Uuid::new_v4(),
-                actor: None,
+                operation_id: crate::identity_lifecycle::LifecycleOperationId::issue(),
+                actor: &missing_target,
                 reason: "migrated key quarantine must not be laundered",
             },
             crate::identity_lifecycle::IdentityPrincipal {
@@ -1751,15 +1802,15 @@ mod tests {
                 &tombstone_rotation_key,
                 None,
                 crate::identity_binding::BindingProvenance::Provisioned,
-                Some("migration-test-policy"),
+                "migration-test-policy",
             )
             .expect("construct tombstone rotation replacement");
         assert!(crate::identity_lifecycle::rotate_identity_binding(
             &pool,
             buzz_core::CommunityId::from_uuid(domain_a),
             crate::identity_lifecycle::LifecycleContext {
-                operation_id: uuid::Uuid::new_v4(),
-                actor: None,
+                operation_id: crate::identity_lifecycle::LifecycleOperationId::issue(),
+                actor: &active_tombstoned_key,
                 reason: "legacy key tombstone must not be laundered",
             },
             crate::identity_lifecycle::IdentityPrincipal {
@@ -1790,14 +1841,17 @@ mod tests {
         ] {
             let result = crate::identity_binding::resolve_identity_binding(
                 &pool,
-                domain_a_id,
                 &crate::identity_binding::ResolveBindingInput {
+                    authorization_domain: domain_a_id,
                     issuer: "https://idp.example",
                     subject,
                     pubkey: key,
                     display_name: None,
                     enrollment_mode: crate::identity_binding::EnrollmentMode::AttestedKey,
                     key_attested: true,
+                    policy_version: "migration-test-policy-v1",
+                    evidence_valid_from: 0,
+                    evidence_valid_until: i64::MAX as u64,
                 },
             )
             .await
@@ -1814,15 +1868,15 @@ mod tests {
                 &missing_target,
                 None,
                 crate::identity_binding::BindingProvenance::Provisioned,
-                Some("migration-test-policy"),
+                "migration-test-policy",
             )
             .expect("construct denied migrated replacement");
         assert!(crate::identity_lifecycle::provision_identity_binding(
             &pool,
             domain_a_id,
             crate::identity_lifecycle::LifecycleContext {
-                operation_id: uuid::Uuid::new_v4(),
-                actor: None,
+                operation_id: crate::identity_lifecycle::LifecycleOperationId::issue(),
+                actor: &missing_target,
                 reason: "migrated ambiguity must block lifecycle",
             },
             crate::identity_lifecycle::IdentityPrincipal {
@@ -1848,14 +1902,17 @@ mod tests {
 
         let domain_b_result = crate::identity_binding::resolve_identity_binding(
             &pool,
-            buzz_core::CommunityId::from_uuid(domain_b),
             &crate::identity_binding::ResolveBindingInput {
+                authorization_domain: buzz_core::CommunityId::from_uuid(domain_b),
                 issuer: "https://idp.example",
                 subject: "cross-domain-allowed",
                 pubkey: &missing_target,
                 display_name: None,
                 enrollment_mode: crate::identity_binding::EnrollmentMode::AttestedKey,
                 key_attested: true,
+                policy_version: "migration-test-policy-v1",
+                evidence_valid_from: 0,
+                evidence_valid_until: i64::MAX as u64,
             },
         )
         .await
@@ -1866,6 +1923,302 @@ mod tests {
         ));
     }
 
+    const IDENTITY_HISTORY_LENGTHS: [usize; 6] = [0, 1, 2, 3, 8, 32];
+
+    fn identity_history_keys(length: usize, namespace: u8) -> Vec<Vec<u8>> {
+        (0..length)
+            .map(|index| {
+                let mut key = vec![0_u8; 32];
+                key[0] = length as u8;
+                key[1] = index as u8;
+                key[30] = namespace;
+                key[31] = 0xa5;
+                key
+            })
+            .collect()
+    }
+
+    async fn insert_legacy_identity_history(
+        pool: &PgPool,
+        domain: uuid::Uuid,
+        issuer: &str,
+        subject: &str,
+        keys: &[Vec<u8>],
+    ) {
+        // Deliberately reverse insertion order. Equal timestamps model the
+        // transaction-stable NOW() values emitted by the legacy helper.
+        for index in (0..keys.len()).rev() {
+            if index + 1 == keys.len() {
+                sqlx::query(
+                    "INSERT INTO identity_bindings \
+                     (community_id,issuer,uid,pubkey,source,created_at,updated_at,last_seen_at) \
+                     VALUES ($1,$2,$3,$4,'db_binding', \
+                             TIMESTAMPTZ '2025-04-01 00:00:00Z', \
+                             TIMESTAMPTZ '2025-04-01 00:00:00Z', \
+                             TIMESTAMPTZ '2025-04-01 00:00:00Z')",
+                )
+                .bind(domain)
+                .bind(issuer)
+                .bind(subject)
+                .bind(&keys[index])
+                .execute(pool)
+                .await
+                .expect("insert terminal history node");
+            } else {
+                sqlx::query(
+                    "INSERT INTO identity_bindings \
+                     (community_id,issuer,uid,pubkey,source,created_at,updated_at,last_seen_at, \
+                      revoked_at,revoked_reason,revocation_scope,rotation_completed_at, \
+                      rotated_to_pubkey,rotation_reason) \
+                     VALUES ($1,$2,$3,$4,'db_binding', \
+                             TIMESTAMPTZ '2025-04-01 00:00:00Z', \
+                             TIMESTAMPTZ '2025-04-01 00:00:00Z', \
+                             TIMESTAMPTZ '2025-04-01 00:00:00Z', \
+                             TIMESTAMPTZ '2025-04-01 00:00:00Z','legacy rotation','rotation', \
+                             TIMESTAMPTZ '2025-04-01 00:00:00Z',$5,'legacy rotation')",
+                )
+                .bind(domain)
+                .bind(issuer)
+                .bind(subject)
+                .bind(&keys[index])
+                .bind(&keys[index + 1])
+                .execute(pool)
+                .await
+                .expect("insert retired history node");
+            }
+        }
+    }
+
+    async fn assert_legacy_domain_sentinels(
+        pool: &PgPool,
+        domain: uuid::Uuid,
+        keys: &[Vec<u8>],
+        expected_facts: &[String],
+        expected_authorization: &[bool],
+        expected_audit: &[String],
+        expected_history: &[String],
+    ) {
+        assert_eq!(
+            super::deterministic_tests::legacy_identity_facts(pool, domain).await,
+            expected_facts
+        );
+        let mut authorization = Vec::with_capacity(keys.len());
+        for key in keys {
+            authorization
+                .push(super::deterministic_tests::raw_domain_authorized(pool, domain, key).await);
+        }
+        assert_eq!(authorization, expected_authorization);
+        assert_eq!(
+            super::deterministic_tests::domain_audit_snapshot(pool, domain).await,
+            expected_audit
+        );
+        assert_eq!(
+            super::deterministic_tests::domain_legacy_history_snapshot(pool, domain).await,
+            expected_history
+        );
+    }
+
+    async fn assert_imported_identity_history(
+        pool: &PgPool,
+        domain: uuid::Uuid,
+        issuer: &str,
+        subject: &str,
+        keys: &[Vec<u8>],
+    ) {
+        type BindingRow = (uuid::Uuid, i64, Vec<u8>, String, String, Option<uuid::Uuid>);
+        let rows: Vec<BindingRow> = sqlx::query_as(
+            "SELECT binding_id,binding_version,pubkey,binding_state, \
+                    binding_provenance,replacement_binding_id \
+             FROM identity_bindings \
+             WHERE community_id=$1 AND issuer=$2 AND uid=$3 \
+             ORDER BY pubkey",
+        )
+        .bind(domain)
+        .bind(issuer)
+        .bind(subject)
+        .fetch_all(pool)
+        .await
+        .expect("read imported history");
+        assert_eq!(rows.len(), keys.len());
+        for (index, row) in rows.iter().enumerate() {
+            assert_eq!(row.1, 1);
+            assert_eq!(row.2, keys[index]);
+            assert_eq!(
+                row.3,
+                if index + 1 == keys.len() {
+                    "active"
+                } else {
+                    "rotated"
+                }
+            );
+            assert_eq!(row.4, "tofu");
+            assert_eq!(row.5, rows.get(index + 1).map(|successor| successor.0));
+        }
+
+        let edges: Vec<(uuid::Uuid, uuid::Uuid)> = sqlx::query_as(
+            "SELECT predecessor.binding_id,successor.binding_id \
+             FROM identity_binding_lineage lineage \
+             JOIN identity_bindings predecessor \
+               ON predecessor.community_id=lineage.community_id \
+              AND predecessor.binding_id=lineage.predecessor_binding_id \
+             JOIN identity_bindings successor \
+               ON successor.community_id=lineage.community_id \
+              AND successor.binding_id=lineage.successor_binding_id \
+             WHERE predecessor.community_id=$1 \
+               AND predecessor.issuer=$2 AND predecessor.uid=$3 \
+             ORDER BY predecessor.pubkey",
+        )
+        .bind(domain)
+        .bind(issuer)
+        .bind(subject)
+        .fetch_all(pool)
+        .await
+        .expect("read imported lineage");
+        let expected_edges = rows
+            .windows(2)
+            .map(|pair| (pair[0].0, pair[1].0))
+            .collect::<Vec<_>>();
+        assert_eq!(edges, expected_edges);
+
+        let retired: Vec<(Vec<u8>, Option<uuid::Uuid>, Option<i64>)> = sqlx::query_as(
+            "SELECT pubkey,retired_binding_id,retired_binding_version \
+             FROM identity_retired_pairs \
+             WHERE community_id=$1 AND issuer=$2 AND subject=$3 \
+             ORDER BY pubkey",
+        )
+        .bind(domain)
+        .bind(issuer)
+        .bind(subject)
+        .fetch_all(pool)
+        .await
+        .expect("read imported retired pairs");
+        let expected_retired = rows
+            .iter()
+            .take(rows.len().saturating_sub(1))
+            .map(|row| (row.2.clone(), Some(row.0), Some(row.1)))
+            .collect::<Vec<_>>();
+        assert_eq!(retired, expected_retired);
+
+        type HistoryRow = (
+            uuid::Uuid,
+            i64,
+            Vec<u8>,
+            String,
+            String,
+            String,
+            Option<uuid::Uuid>,
+        );
+        let history: Vec<HistoryRow> = sqlx::query_as(
+            "SELECT binding_id,binding_version,pubkey,binding_state, \
+                    binding_provenance,transition_kind,replacement_binding_id \
+             FROM identity_binding_history \
+             WHERE community_id=$1 AND issuer=$2 AND subject=$3 \
+             ORDER BY pubkey",
+        )
+        .bind(domain)
+        .bind(issuer)
+        .bind(subject)
+        .fetch_all(pool)
+        .await
+        .expect("read imported binding history");
+        assert_eq!(history.len(), rows.len());
+        for (binding, history_row) in rows.iter().zip(&history) {
+            assert_eq!(history_row.0, binding.0);
+            assert_eq!(history_row.1, binding.1);
+            assert_eq!(history_row.2, binding.2);
+            assert_eq!(history_row.3, binding.3);
+            assert_eq!(history_row.4, binding.4);
+            assert_eq!(history_row.5, "legacy_import");
+            assert_eq!(history_row.6, binding.5);
+        }
+
+        let denied: bool = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM identity_migration_denials \
+             WHERE community_id=$1 AND issuer=$2 AND subject=$3)",
+        )
+        .bind(domain)
+        .bind(issuer)
+        .bind(subject)
+        .fetch_one(pool)
+        .await
+        .expect("read imported principal denial");
+        assert!(!denied);
+        let pending: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM identity_pending_replacements \
+             WHERE community_id=$1 AND issuer=$2 AND subject=$3",
+        )
+        .bind(domain)
+        .bind(issuer)
+        .bind(subject)
+        .fetch_one(pool)
+        .await
+        .expect("read imported pending selectors");
+        assert_eq!(pending, 0);
+        for (index, key) in keys.iter().enumerate() {
+            let denied_key: bool = sqlx::query_scalar(
+                "SELECT EXISTS(SELECT 1 FROM identity_migration_denied_keys \
+                 WHERE community_id=$1 AND pubkey=$2)",
+            )
+            .bind(domain)
+            .bind(key)
+            .fetch_one(pool)
+            .await
+            .expect("read imported key denial");
+            assert!(!denied_key);
+            let active = crate::identity_binding::get_active_identity_binding_by_pubkey(
+                pool,
+                buzz_core::CommunityId::from_uuid(domain),
+                key,
+            )
+            .await
+            .expect("read imported authorization");
+            if index + 1 == keys.len() {
+                let active = active.expect("history head remains authoritative");
+                assert_eq!(active.binding_id, rows[index].0);
+                assert_eq!(active.binding_version, rows[index].1 as u64);
+                assert_eq!(active.issuer, issuer);
+                assert_eq!(active.uid, subject);
+                assert_eq!(active.pubkey, *key);
+                assert_eq!(
+                    active.binding_state,
+                    crate::identity_binding::BindingState::Active
+                );
+            } else {
+                assert!(active.is_none(), "retired history key regained authority");
+            }
+        }
+        if keys.is_empty() {
+            let absent = crate::identity_binding::get_active_identity_binding_by_pubkey(
+                pool,
+                buzz_core::CommunityId::from_uuid(domain),
+                &[0xe0_u8; 32],
+            )
+            .await
+            .expect("read zero-history authorization sentinel");
+            assert!(absent.is_none(), "zero history invented authority");
+        }
+    }
+
+    async fn assert_temporal_inversion_quarantined(
+        pool: &PgPool,
+        domain: uuid::Uuid,
+        older_target: &[u8],
+    ) {
+        let inversion_denied: (bool, bool) = sqlx::query_as(
+            "SELECT \
+               EXISTS(SELECT 1 FROM identity_migration_denials \
+                      WHERE community_id=$1 AND issuer='https://idp.example' AND subject='temporal-inversion'), \
+               EXISTS(SELECT 1 FROM identity_migration_denied_keys \
+                      WHERE community_id=$1 AND pubkey=$2)",
+        )
+        .bind(domain)
+        .bind(older_target)
+        .fetch_one(pool)
+        .await
+        .expect("read temporal inversion quarantine");
+        assert_eq!(inversion_denied, (true, true));
+    }
+
     #[tokio::test]
     #[ignore = "requires a dedicated disposable Postgres database"]
     async fn identity_0029_imports_arbitrary_histories_and_quarantines_temporal_inversion() {
@@ -1876,66 +2229,73 @@ mod tests {
             .await
             .expect("apply migrations through legacy identity lifecycle");
         let domain = uuid::Uuid::new_v4();
-        sqlx::query("INSERT INTO communities (id, host) VALUES ($1,$2)")
-            .bind(domain)
-            .bind(format!("identity-0029-history-{}.example", domain.simple()))
-            .execute(&pool)
-            .await
-            .expect("insert history community");
-
-        for length in [1_usize, 2, 3, 8, 32] {
-            let subject = format!("history-{length}");
-            let keys = (0..length)
-                .map(|index| {
-                    let mut key = vec![0_u8; 32];
-                    key[0] = length as u8;
-                    key[1] = index as u8;
-                    key[31] = 0xa5;
-                    key
-                })
-                .collect::<Vec<_>>();
-            // Deliberately reverse insertion order. Equal timestamps model the
-            // transaction-stable NOW() values emitted by the legacy helper.
-            for index in (0..length).rev() {
-                if index + 1 == length {
-                    sqlx::query(
-                        "INSERT INTO identity_bindings \
-                         (community_id,issuer,uid,pubkey,source,created_at,updated_at,last_seen_at) \
-                         VALUES ($1,'https://idp.example',$2,$3,'db_binding', \
-                                 TIMESTAMPTZ '2025-04-01 00:00:00Z', \
-                                 TIMESTAMPTZ '2025-04-01 00:00:00Z', \
-                                 TIMESTAMPTZ '2025-04-01 00:00:00Z')",
-                    )
-                    .bind(domain)
-                    .bind(&subject)
-                    .bind(&keys[index])
-                    .execute(&pool)
-                    .await
-                    .expect("insert terminal history node");
-                } else {
-                    sqlx::query(
-                        "INSERT INTO identity_bindings \
-                         (community_id,issuer,uid,pubkey,source,created_at,updated_at,last_seen_at, \
-                          revoked_at,revoked_reason,revocation_scope,rotation_completed_at, \
-                          rotated_to_pubkey,rotation_reason) \
-                         VALUES ($1,'https://idp.example',$2,$3,'db_binding', \
-                                 TIMESTAMPTZ '2025-04-01 00:00:00Z', \
-                                 TIMESTAMPTZ '2025-04-01 00:00:00Z', \
-                                 TIMESTAMPTZ '2025-04-01 00:00:00Z', \
-                                 TIMESTAMPTZ '2025-04-01 00:00:00Z','legacy rotation','rotation', \
-                                 TIMESTAMPTZ '2025-04-01 00:00:00Z',$4,'legacy rotation')",
-                    )
-                    .bind(domain)
-                    .bind(&subject)
-                    .bind(&keys[index])
-                    .bind(&keys[index + 1])
-                    .execute(&pool)
-                    .await
-                    .expect("insert retired history node");
-                }
-            }
+        let domain_b = uuid::Uuid::new_v4();
+        for (community, suffix) in [(domain, "a"), (domain_b, "b")] {
+            sqlx::query("INSERT INTO communities (id, host) VALUES ($1,$2)")
+                .bind(community)
+                .bind(format!(
+                    "identity-0029-history-{suffix}-{}.example",
+                    community.simple()
+                ))
+                .execute(&pool)
+                .await
+                .expect("insert history community");
         }
 
+        let mut seeded_lengths = BTreeSet::new();
+        for length in IDENTITY_HISTORY_LENGTHS {
+            let subject = format!("history-{length}");
+            let keys = identity_history_keys(length, 0);
+            insert_legacy_identity_history(&pool, domain, "https://idp.example", &subject, &keys)
+                .await;
+            assert!(seeded_lengths.insert(length));
+        }
+        assert_eq!(
+            seeded_lengths,
+            IDENTITY_HISTORY_LENGTHS.into_iter().collect()
+        );
+
+        let domain_b_keys = identity_history_keys(3, 0xb7);
+        insert_legacy_identity_history(
+            &pool,
+            domain_b,
+            "https://domain-b.example",
+            "domain-b-history-sentinel",
+            &domain_b_keys,
+        )
+        .await;
+        sqlx::query(
+            "INSERT INTO audit_log \
+             (community_id,seq,hash,action,object_id,detail,created_at) \
+             VALUES ($1,1,$2,'preexisting_domain_b_history', \
+                     'domain-b-history-sentinel', \
+                     '{\"sentinel\":\"before-domain-a-operation\"}'::jsonb, \
+                     TIMESTAMPTZ '2025-04-01 00:00:00Z')",
+        )
+        .bind(domain_b)
+        .bind(vec![0xb7_u8; 32])
+        .execute(&pool)
+        .await
+        .expect("insert substantive domain-B audit sentinel");
+
+        let domain_b_facts_pre =
+            super::deterministic_tests::legacy_identity_facts(&pool, domain_b).await;
+        let domain_b_audit_pre =
+            super::deterministic_tests::domain_audit_snapshot(&pool, domain_b).await;
+        let domain_b_history_pre =
+            super::deterministic_tests::domain_legacy_history_snapshot(&pool, domain_b).await;
+        let domain_b_authorization_pre = vec![false, false, true];
+        assert_legacy_domain_sentinels(
+            &pool,
+            domain_b,
+            &domain_b_keys,
+            &domain_b_facts_pre,
+            &domain_b_authorization_pre,
+            &domain_b_audit_pre,
+            &domain_b_history_pre,
+        )
+        .await;
+
         let older_target = vec![240_u8; 32];
         let newer_predecessor = vec![241_u8; 32];
         sqlx::query(
@@ -1973,50 +2333,129 @@ mod tests {
         run_migrations(&pool)
             .await
             .expect("import arbitrary valid histories without aborting");
-        for length in [1_usize, 2, 3, 8, 32] {
-            let subject = format!("history-{length}");
-            let rows: Vec<(i64, String)> = sqlx::query_as(
-                "SELECT binding_version,binding_provenance FROM identity_bindings \
-                 WHERE community_id=$1 AND issuer='https://idp.example' AND uid=$2 \
-                 ORDER BY binding_version",
-            )
-            .bind(domain)
-            .bind(&subject)
-            .fetch_all(&pool)
-            .await
-            .expect("read imported history");
-            assert_eq!(rows.len(), length);
-            for (index, (version, provenance)) in rows.iter().enumerate() {
-                assert_eq!(*version, (index + 1) as i64);
-                assert_eq!(provenance, if index == 0 { "tofu" } else { "provisioned" });
-            }
-            let edges: i64 = sqlx::query_scalar(
-                "SELECT COUNT(*) FROM identity_binding_lineage lineage \
-                 JOIN identity_bindings binding \
-                   ON binding.community_id=lineage.community_id \
-                  AND binding.binding_id=lineage.predecessor_binding_id \
-                 WHERE binding.community_id=$1 AND binding.issuer='https://idp.example' AND binding.uid=$2",
-            )
-            .bind(domain)
-            .bind(&subject)
-            .fetch_one(&pool)
-            .await
-            .expect("count imported lineage");
-            assert_eq!(edges, length.saturating_sub(1) as i64);
-        }
-        let inversion_denied: (bool, bool) = sqlx::query_as(
-            "SELECT \
-               EXISTS(SELECT 1 FROM identity_migration_denials \
-                      WHERE community_id=$1 AND issuer='https://idp.example' AND subject='temporal-inversion'), \
-               EXISTS(SELECT 1 FROM identity_migration_denied_keys \
-                      WHERE community_id=$1 AND pubkey=$2)",
+        // Prove domain B before the first domain-A read after the operation.
+        assert_legacy_domain_sentinels(
+            &pool,
+            domain_b,
+            &domain_b_keys,
+            &domain_b_facts_pre,
+            &domain_b_authorization_pre,
+            &domain_b_audit_pre,
+            &domain_b_history_pre,
         )
-        .bind(domain)
-        .bind(&older_target)
-        .fetch_one(&pool)
-        .await
-        .expect("read temporal inversion quarantine");
-        assert_eq!(inversion_denied, (true, true));
+        .await;
+        assert_imported_identity_history(
+            &pool,
+            domain_b,
+            "https://domain-b.example",
+            "domain-b-history-sentinel",
+            &domain_b_keys,
+        )
+        .await;
+        let domain_b_state_post =
+            super::deterministic_tests::domain_identity_snapshot(&pool, domain_b).await;
+        let domain_b_history_post =
+            super::deterministic_tests::domain_binding_history_snapshot(&pool, domain_b).await;
+        assert_eq!(domain_b_history_post.len(), domain_b_keys.len());
+
+        let mut verified_lengths = BTreeSet::new();
+        for length in IDENTITY_HISTORY_LENGTHS {
+            let subject = format!("history-{length}");
+            let keys = identity_history_keys(length, 0);
+            assert_imported_identity_history(&pool, domain, "https://idp.example", &subject, &keys)
+                .await;
+            assert!(verified_lengths.insert(length));
+        }
+        assert_eq!(
+            verified_lengths,
+            IDENTITY_HISTORY_LENGTHS.into_iter().collect()
+        );
+        assert_temporal_inversion_quarantined(&pool, domain, &older_target).await;
+        let domain_a_state_post =
+            super::deterministic_tests::domain_identity_snapshot(&pool, domain).await;
+        assert_legacy_domain_sentinels(
+            &pool,
+            domain_b,
+            &domain_b_keys,
+            &domain_b_facts_pre,
+            &domain_b_authorization_pre,
+            &domain_b_audit_pre,
+            &domain_b_history_pre,
+        )
+        .await;
+        assert_eq!(
+            super::deterministic_tests::domain_identity_snapshot(&pool, domain_b).await,
+            domain_b_state_post
+        );
+        assert_eq!(
+            super::deterministic_tests::domain_binding_history_snapshot(&pool, domain_b).await,
+            domain_b_history_post
+        );
+
+        pool.close().await;
+        let pool = connect_test_pool().await;
+        for length in IDENTITY_HISTORY_LENGTHS {
+            let subject = format!("history-{length}");
+            let keys = identity_history_keys(length, 0);
+            assert_imported_identity_history(&pool, domain, "https://idp.example", &subject, &keys)
+                .await;
+        }
+        assert_temporal_inversion_quarantined(&pool, domain, &older_target).await;
+        assert_eq!(
+            super::deterministic_tests::domain_identity_snapshot(&pool, domain).await,
+            domain_a_state_post
+        );
+        assert_legacy_domain_sentinels(
+            &pool,
+            domain_b,
+            &domain_b_keys,
+            &domain_b_facts_pre,
+            &domain_b_authorization_pre,
+            &domain_b_audit_pre,
+            &domain_b_history_pre,
+        )
+        .await;
+        assert_eq!(
+            super::deterministic_tests::domain_identity_snapshot(&pool, domain_b).await,
+            domain_b_state_post
+        );
+        assert_eq!(
+            super::deterministic_tests::domain_binding_history_snapshot(&pool, domain_b).await,
+            domain_b_history_post
+        );
+
+        run_migrations(&pool)
+            .await
+            .expect("retry arbitrary-history migration idempotently");
+        for length in IDENTITY_HISTORY_LENGTHS {
+            let subject = format!("history-{length}");
+            let keys = identity_history_keys(length, 0);
+            assert_imported_identity_history(&pool, domain, "https://idp.example", &subject, &keys)
+                .await;
+        }
+        assert_temporal_inversion_quarantined(&pool, domain, &older_target).await;
+        assert_eq!(
+            super::deterministic_tests::domain_identity_snapshot(&pool, domain).await,
+            domain_a_state_post
+        );
+        assert_legacy_domain_sentinels(
+            &pool,
+            domain_b,
+            &domain_b_keys,
+            &domain_b_facts_pre,
+            &domain_b_authorization_pre,
+            &domain_b_audit_pre,
+            &domain_b_history_pre,
+        )
+        .await;
+        assert_eq!(
+            super::deterministic_tests::domain_identity_snapshot(&pool, domain_b).await,
+            domain_b_state_post
+        );
+        assert_eq!(
+            super::deterministic_tests::domain_binding_history_snapshot(&pool, domain_b).await,
+            domain_b_history_post
+        );
     }
 
     #[tokio::test]

crates/buzz-db/src/relay_invite.rs

@@ -62,6 +62,9 @@ pub enum ClaimOutcome {
     IdentityConflict(IdentityBindingConflict),
     /// The staged identity principal or key is revoked.
     IdentityRevoked,
+    /// The staged identity has no active binding and lacks sealed enrollment
+    /// evidence.
+    IdentityBindingRequired,
 }
 
 /// A freshly minted v2 invite, including the plaintext code and metadata.
@@ -287,6 +290,17 @@ pub async fn claim_relay_invite_with_identity(
                 );
                 return Ok(ClaimOutcome::IdentityRevoked);
             }
+            BindIdentityResult::BindingRequired => {
+                tx.rollback().await?;
+                log_claim_outcome(
+                    community,
+                    Some(invite_id),
+                    "identity_binding_required",
+                    max_uses,
+                    Some(use_count),
+                );
+                return Ok(ClaimOutcome::IdentityBindingRequired);
+            }
         }
     } else {
         None
@@ -662,7 +676,7 @@ mod tests {
 
     #[tokio::test]
     #[ignore = "requires Postgres"]
-    async fn invite_claim_commits_identity_and_membership_atomically() {
+    async fn invite_claim_requires_verified_binding_and_rolls_back_membership_atomically() {
         let pool = setup_pool().await;
         let community = make_test_community(&pool).await;
         let claimer = test_pubkey();
@@ -702,7 +716,7 @@ mod tests {
             .await
             .expect("mint invite");
         let hash = hash_v2_code(&invite.code);
-        assert!(matches!(
+        assert_eq!(
             claim_relay_invite_with_identity(
                 &pool,
                 community,
@@ -712,22 +726,21 @@ mod tests {
                 Some(&identity),
             )
             .await
-            .expect("valid atomic claim"),
-            ClaimOutcome::Joined { .. }
-        ));
-        assert!(is_relay_member(&pool, community, &claimer)
+            .expect("binding-required atomic claim"),
+            ClaimOutcome::IdentityBindingRequired
+        );
+        assert!(!is_relay_member(&pool, community, &claimer)
             .await
-            .expect("membership committed"));
-        assert_eq!(
+            .expect("membership rolled back"));
+        assert!(
             crate::identity_binding::get_active_identity_binding_by_pubkey(
                 &pool, community, &pubkey,
             )
             .await
             .expect("binding lookup")
-            .expect("binding committed")
-            .uid,
-            "atomic-user"
+            .is_none()
         );
+        assert_eq!(use_count(&pool, community, invite.invite_id).await, 0);
         delete_test_community(&pool, community).await;
     }
 

crates/buzz-db/src/relay_members.rs

@@ -158,11 +158,11 @@ pub async fn claim_relay_membership(
         .await?
     {
         MembershipClaimOutcome::Joined { inserted, .. } => Ok(inserted),
-        MembershipClaimOutcome::IdentityConflict(_) | MembershipClaimOutcome::IdentityRevoked => {
-            Err(crate::DbError::InvalidData(
-                "unexpected corporate identity result without staged identity".to_string(),
-            ))
-        }
+        MembershipClaimOutcome::IdentityConflict(_)
+        | MembershipClaimOutcome::IdentityRevoked
+        | MembershipClaimOutcome::IdentityBindingRequired => Err(crate::DbError::InvalidData(
+            "unexpected corporate identity result without staged identity".to_string(),
+        )),
     }
 }
 
@@ -180,6 +180,9 @@ pub enum MembershipClaimOutcome {
     IdentityConflict(IdentityBindingConflict),
     /// The staged identity is revoked.
     IdentityRevoked,
+    /// The staged identity has no active binding and lacks sealed enrollment
+    /// evidence.
+    IdentityBindingRequired,
 }
 
 /// Claims relay membership and an optional corporate identity in one transaction.
@@ -206,6 +209,10 @@ pub async fn claim_relay_membership_with_identity(
                 tx.rollback().await?;
                 return Ok(MembershipClaimOutcome::IdentityRevoked);
             }
+            BindIdentityResult::BindingRequired => {
+                tx.rollback().await?;
+                return Ok(MembershipClaimOutcome::IdentityBindingRequired);
+            }
         }
     } else {
         None

crates/buzz-relay/src/api/invites.rs

@@ -548,6 +548,16 @@ pub async fn claim_invite(
                 )
                 .await)
             }
+            buzz_db::relay_invite::ClaimOutcome::IdentityBindingRequired => {
+                Err(record_atomic_identity_rejection(
+                    &state,
+                    tenant.community(),
+                    pubkey,
+                    identity_proof,
+                    buzz_db::identity_binding::BindIdentityResult::BindingRequired,
+                )
+                .await)
+            }
         };
     }
 
@@ -614,6 +624,16 @@ pub async fn claim_invite(
             )
             .await);
         }
+        buzz_db::relay_members::MembershipClaimOutcome::IdentityBindingRequired => {
+            return Err(record_atomic_identity_rejection(
+                &state,
+                tenant.community(),
+                pubkey,
+                identity_proof,
+                buzz_db::identity_binding::BindIdentityResult::BindingRequired,
+            )
+            .await);
+        }
     };
     crate::corporate_identity::finalize_atomic_corporate_identity_result(
         &state,

crates/buzz-relay/src/audio/handler.rs

@@ -695,6 +695,9 @@ async fn handle_active_audio_connection(
             Ok(buzz_db::channel::ChannelAdmissionOutcome::IdentityRevoked) => {
                 Some(buzz_db::identity_binding::BindIdentityResult::Revoked)
             }
+            Ok(buzz_db::channel::ChannelAdmissionOutcome::IdentityBindingRequired) => {
+                Some(buzz_db::identity_binding::BindIdentityResult::BindingRequired)
+            }
             Err(e) => {
                 warn!(channel_id = %channel_id, pubkey = %pubkey_hex, "audio membership auto-add failed: {e}");
                 let _ = ws_send

crates/buzz-relay/src/corporate_identity.rs

@@ -496,6 +496,9 @@ pub enum CorporateIdentityError {
     /// The requested uid/pubkey binding was previously revoked.
     #[error("corporate identity binding revoked")]
     BindingRevoked,
+    /// No active binding exists and this compatibility path cannot enroll one.
+    #[error("corporate identity binding requires authorized enrollment")]
+    BindingRequired,
     /// NIP-OA delegation was present but did not satisfy corporate identity.
     #[error("corporate identity delegation denied")]
     DelegationDenied,
@@ -515,6 +518,7 @@ impl CorporateIdentityError {
             | Self::NpubMismatch
             | Self::BindingConflict
             | Self::BindingRevoked
+            | Self::BindingRequired
             | Self::DelegationDenied => StatusCode::FORBIDDEN,
             Self::Db(_) => StatusCode::INTERNAL_SERVER_ERROR,
         }
@@ -531,6 +535,7 @@ impl CorporateIdentityError {
             Self::NpubMismatch => "relay identity pubkey mismatch",
             Self::BindingConflict => "relay identity binding conflict",
             Self::BindingRevoked => "relay identity binding revoked",
+            Self::BindingRequired => "relay identity binding required",
             Self::DelegationDenied => "relay identity delegation denied",
             Self::Db(_) => "relay identity unavailable",
         }
@@ -725,7 +730,7 @@ async fn complete_direct_corporate_identity(
     binding: BindIdentityResult,
 ) -> Result<CorporateIdentityDecision, CorporateIdentityError> {
     let binding = match binding {
-        BindIdentityResult::Conflict(conflict) => {
+        BindIdentityResult::Conflict(_) => {
             metrics::counter!("buzz_corporate_identity_bindings_total", "result" => "conflict")
                 .increment(1);
             record_identity_binding_audit(
@@ -737,19 +742,10 @@ async fn complete_direct_corporate_identity(
                 &claims.uid,
                 serde_json::json!({
                     "source": source,
-                    "issuer": claims.issuer,
-                    "existing_uid": conflict.uid,
-                    "existing_issuer": conflict.issuer,
-                    "existing_pubkey": hex::encode(conflict.pubkey),
-                    "existing_source": conflict.source,
                 }),
             )
             .await;
-            warn!(
-                uid = %claims.uid,
-                signer = %signer.to_hex(),
-                "corporate identity binding conflict"
-            );
+            warn!("corporate identity binding conflict");
             return Err(CorporateIdentityError::BindingConflict);
         }
         BindIdentityResult::Revoked => {
@@ -765,13 +761,18 @@ async fn complete_direct_corporate_identity(
                 serde_json::json!({ "source": source, "issuer": claims.issuer }),
             )
             .await;
-            warn!(
-                uid = %claims.uid,
-                signer = %signer.to_hex(),
-                "corporate identity binding was previously revoked"
-            );
+            warn!("corporate identity binding was previously revoked");
             return Err(CorporateIdentityError::BindingRevoked);
         }
+        BindIdentityResult::BindingRequired => {
+            metrics::counter!(
+                "buzz_corporate_identity_bindings_total",
+                "result" => "binding_required"
+            )
+            .increment(1);
+            warn!("corporate identity binding requires sealed enrollment evidence");
+            return Err(CorporateIdentityError::BindingRequired);
+        }
         binding => binding,
     };
     record_identity_binding_metric(&binding);
@@ -1194,6 +1195,7 @@ fn record_identity_binding_metric(binding: &BindIdentityResult) {
         BindIdentityResult::Matched => "matched",
         BindIdentityResult::Conflict(_) => "conflict",
         BindIdentityResult::Revoked => "revoked",
+        BindIdentityResult::BindingRequired => "binding_required",
     };
     metrics::counter!("buzz_corporate_identity_bindings_total", "result" => result).increment(1);
 }
@@ -1208,6 +1210,7 @@ fn record_corporate_identity_denial(error: &CorporateIdentityError) {
         CorporateIdentityError::NpubMismatch => "npub_mismatch",
         CorporateIdentityError::BindingConflict => "binding_conflict",
         CorporateIdentityError::BindingRevoked => "binding_revoked",
+        CorporateIdentityError::BindingRequired => "binding_required",
         CorporateIdentityError::DelegationDenied => "delegation_denied",
         CorporateIdentityError::Db(_) => "db",
     };
@@ -1303,6 +1306,11 @@ mod tests {
             binding_version: 1,
             binding_state: buzz_db::identity_binding::BindingState::Active,
             binding_provenance: buzz_db::identity_binding::BindingProvenance::Tofu,
+            creation_attribution:
+                buzz_db::identity_binding::CreationAttributionKind::AuthenticatedKey,
+            created_by: Some(pubkey.to_bytes().to_vec()),
+            created_policy_version: Some("relay-test-policy-v1".to_owned()),
+            expires_at: None,
             display_name: None,
             source: SOURCE_DB_BINDING.to_string(),
             created_at: now,
@@ -2120,16 +2128,24 @@ mod tests {
         .expect_err("owner without binding should be denied");
         assert!(matches!(err, CorporateIdentityError::DelegationDenied));
 
-        db.bind_or_validate_identity(
-            community,
-            &config.issuer,
-            "owner-uid",
-            owner_keys.public_key().as_bytes(),
-            Some("owner@example.com"),
-            SOURCE_DB_BINDING,
+        sqlx::query(
+            "INSERT INTO identity_bindings \
+             (community_id, issuer, uid, pubkey, display_name, source, binding_id, \
+              binding_version, binding_state, binding_provenance, created_by, \
+              created_policy_version, creation_attribution_kind) \
+             VALUES ($1,$2,$3,$4,$5,$6,$7,1,'active','attested_key',$4,$8,'authenticated_key')",
         )
+        .bind(community.as_uuid())
+        .bind(&config.issuer)
+        .bind("owner-uid")
+        .bind(owner_keys.public_key().as_bytes())
+        .bind("owner@example.com")
+        .bind(SOURCE_DB_BINDING)
+        .bind(Uuid::new_v4())
+        .bind("relay-test-policy-v1")
+        .execute(&pool)
         .await
-        .expect("create owner binding");
+        .expect("seed verified owner binding fixture");
 
         let decision = verify_delegated_corporate_identity(
             &db,

crates/buzz-relay/src/mesh_boot.rs

@@ -315,25 +315,39 @@ pub(crate) async fn run_demo_echo(
     tracing::info!(%session_id, %peer, "mesh demo echo: session open");
     let mut drain_tick = tokio::time::interval(std::time::Duration::from_millis(100));
     loop {
-        let frame = tokio::select! {
-            _ = drain_tick.tick() => {
-                if shutting_down.load(Ordering::Relaxed) {
-                    if let Some(community_id) = stream.community_id() {
-                        if let Err(e) = stream.send_goodbye(community_id, GoodbyeReason::Draining).await {
-                            tracing::warn!(%session_id, "mesh demo echo: draining goodbye failed: {e}");
-                        } else {
-                            tracing::info!(%session_id, "mesh demo echo: sent draining goodbye");
+        // recv_validated reads the frame before asynchronously checking its
+        // Redis fence. Keep that future alive across drain polls: cancelling
+        // and recreating it after a tick would discard an already-read frame.
+        let frame = {
+            let recv = stream.recv_validated(&directory);
+            tokio::pin!(recv);
+            loop {
+                tokio::select! {
+                    frame = &mut recv => break frame,
+                    _ = drain_tick.tick() => {
+                        if shutting_down.load(Ordering::Relaxed) {
+                            break Ok(None);
                         }
-                    } else {
-                        let _ = stream.finish();
-                        tracing::info!(%session_id, "mesh demo echo: drain before community latch — closing");
                     }
-                    return;
                 }
-                continue;
             }
-            frame = stream.recv_validated(&directory) => frame,
         };
+        if shutting_down.load(Ordering::Relaxed) {
+            if let Some(community_id) = stream.community_id() {
+                if let Err(e) = stream
+                    .send_goodbye(community_id, GoodbyeReason::Draining)
+                    .await
+                {
+                    tracing::warn!(%session_id, "mesh demo echo: draining goodbye failed: {e}");
+                } else {
+                    tracing::info!(%session_id, "mesh demo echo: sent draining goodbye");
+                }
+            } else {
+                let _ = stream.finish();
+                tracing::info!(%session_id, "mesh demo echo: drain before community latch — closing");
+            }
+            return;
+        }
         match frame {
             Ok(Some(ReliableFrame::Data(payload))) => {
                 // recv_validated latched the community from the frame it just

crates/git-sign-nostr/src/lib.rs

@@ -84,7 +84,7 @@ use chrono::DateTime;
 use nostr::hashes::sha256::Hash as Sha256Hash;
 use nostr::hashes::{Hash, HashEngine};
 use nostr::secp256k1::schnorr::Signature;
-use nostr::secp256k1::{Keypair, Message};
+use nostr::secp256k1::{Keypair, Message, XOnlyPublicKey};
 use nostr::{FromBech32, PublicKey, SecretKey, SECP256K1};
 use zeroize::Zeroize;
 
@@ -1017,7 +1017,7 @@ fn do_sign(key_id: &str, status: &mut StatusWriter) -> Result<(), Error> {
     let oa = load_auth_tag()?;
     if let Some(ref oa_val) = oa {
         // Owner pubkey must be a valid BIP-340 key
-        if PublicKey::from_hex(&oa_val.0).is_err() {
+        if parse_bip340_xonly_public_key(&oa_val.0).is_err() {
             return Err(Error::Fatal(
                 "auth tag owner (oa[0]) is not a valid BIP-340 public key".to_string(),
             ));
@@ -1188,7 +1188,7 @@ fn do_verify(sig_file: &str, status: &mut StatusWriter) -> Result<(), Error> {
     }
 
     // Validate pk is a valid BIP-340 x-only public key
-    let pk = PublicKey::from_hex(&envelope.pk).map_err(|e| {
+    let xonly = parse_bip340_xonly_public_key(&envelope.pk).map_err(|e| {
         write_errsig(status, Some(&envelope.pk));
         Error::VerifyFailed {
             pk: Some(envelope.pk.clone()),
@@ -1223,13 +1223,6 @@ fn do_verify(sig_file: &str, status: &mut StatusWriter) -> Result<(), Error> {
     })?;
 
     // Verify BIP-340 signature
-    let xonly = pk.xonly().map_err(|_| {
-        write_errsig(status, Some(&envelope.pk));
-        Error::VerifyFailed {
-            pk: Some(envelope.pk.clone()),
-            msg: "invalid public key xonly conversion".to_string(),
-        }
-    })?;
     if SECP256K1.verify_schnorr(&sig, &message, &xonly).is_err() {
         status.write_line("NEWSIG");
         status.write_line(&format!("BADSIG {} {}", envelope.pk, envelope.pk));
@@ -1243,7 +1236,7 @@ fn do_verify(sig_file: &str, status: &mut StatusWriter) -> Result<(), Error> {
     let oa_result = if let Some(ref oa) = envelope.oa {
         // Validate oa[0] is a valid BIP-340 public key. Per NIP-GS spec,
         // an invalid owner pubkey is a structural error → ERRSIG.
-        if PublicKey::from_hex(&oa.0).is_err() {
+        if parse_bip340_xonly_public_key(&oa.0).is_err() {
             write_errsig(status, Some(&envelope.pk));
             return Err(Error::VerifyFailed {
                 pk: Some(envelope.pk),
@@ -1420,7 +1413,7 @@ fn parse_envelope(json_str: &str) -> Result<Envelope, String> {
         }
 
         // Validate oa[0] is a valid BIP-340 x-only public key (not just hex)
-        PublicKey::from_hex(owner)
+        parse_bip340_xonly_public_key(owner)
             .map_err(|e| format!("oa[0] is not a valid BIP-340 public key: {e}"))?;
 
         // Self-attestation is meaningless — owner must differ from signer
@@ -1461,6 +1454,17 @@ fn validate_hex_field(val: &str, expected_len: usize, name: &str) -> Result<(),
     Ok(())
 }
 
+/// Decode a 32-byte public-key value and require a valid secp256k1 x-only
+/// point. `nostr::PublicKey::from_hex` checks only the byte encoding; BIP-340
+/// validity is established by the x-only conversion.
+fn parse_bip340_xonly_public_key(public_key: &str) -> Result<XOnlyPublicKey, String> {
+    let public_key =
+        PublicKey::from_hex(public_key).map_err(|e| format!("invalid hex encoding: {e}"))?;
+    public_key
+        .xonly()
+        .map_err(|e| format!("invalid x-only point: {e}"))
+}
+
 fn parse_armor(content: &str) -> Result<&str, String> {
     // NIP-GS spec requires armor to end with a newline after the END marker.
     let content = content
@@ -1500,9 +1504,9 @@ fn parse_armor(content: &str) -> Result<&str, String> {
 fn verify_oa(agent_pk_hex: &str, oa: &(String, String, String)) -> bool {
     let (owner_pk_hex, conditions, owner_sig_hex) = oa;
 
-    // Parse owner pubkey
-    let owner_pk = match PublicKey::from_hex(owner_pk_hex) {
-        Ok(p) => p,
+    // Parse owner pubkey and require a valid x-only point.
+    let xonly = match parse_bip340_xonly_public_key(owner_pk_hex) {
+        Ok(xonly) => xonly,
         Err(_) => {
             eprintln!("warning: oa owner pubkey is not a valid BIP-340 key");
             return false;
@@ -1530,13 +1534,6 @@ fn verify_oa(agent_pk_hex: &str, oa: &(String, String, String)) -> bool {
         }
     };
 
-    let xonly = match owner_pk.xonly() {
-        Ok(x) => x,
-        Err(_) => {
-            eprintln!("warning: oa owner pubkey conversion to xonly failed");
-            return false;
-        }
-    };
     if SECP256K1.verify_schnorr(&sig, &message, &xonly).is_err() {
         eprintln!("warning: NIP-OA owner attestation signature verification failed");
         return false;
@@ -2261,7 +2258,7 @@ Initial commit"
         if !is_lower_hex(&owner, 64) {
             return Err("auth tag owner must be 64 lowercase hex chars".to_string());
         }
-        PublicKey::from_hex(&owner)
+        parse_bip340_xonly_public_key(&owner)
             .map_err(|e| format!("auth tag owner is not a valid BIP-340 key: {e}"))?;
         if !is_lower_hex(&sig, 128) {
             return Err("auth tag sig must be 128 lowercase hex chars".to_string());

migrations/0029_additive_identity_binding_state.sql

@@ -1,4 +1,4 @@
--- Additive O3 identity-binding projection.
+-- Additive identity-binding state projection.
 --
 -- Migrations 0027/0028 are a frozen compatibility boundary. This migration
 -- never renames or removes their columns, constraints, indexes, rows, or
@@ -6,6 +6,57 @@
 -- remains authoritative because rotation-created and explicitly strengthened
 -- tombstones cannot be distinguished after the fact.
 
+-- Materialize every retained legacy identity row before projecting any new
+-- state. A storage/decoding/read-policy failure must abort this migration;
+-- treating an unreadable binding, principal denial, or key tombstone as
+-- absent could otherwise invent authority. This block is read-only and runs
+-- inside the migration transaction before the first persisted write.
+WITH legacy_rows AS MATERIALIZED (
+    SELECT to_jsonb(row_value) AS payload FROM identity_bindings row_value
+    UNION ALL
+    SELECT to_jsonb(row_value) AS payload FROM identity_principals row_value
+    UNION ALL
+    SELECT to_jsonb(row_value) AS payload FROM identity_revoked_keys row_value
+)
+SELECT COUNT(payload) FROM legacy_rows;
+
+-- Current verifier-owned enrollment policy. The table is intentionally empty
+-- after migration: installing a policy is a separately authorized server
+-- configuration action, so the disabled candidate fails closed by default.
+CREATE TABLE identity_enrollment_policies (
+    community_id   UUID NOT NULL PRIMARY KEY REFERENCES communities(id),
+    policy_id      UUID NOT NULL,
+    policy_epoch   BIGINT NOT NULL,
+    requirement    TEXT NOT NULL,
+    effective_from BIGINT NOT NULL,
+    effective_until BIGINT NOT NULL,
+    updated_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    CHECK (policy_id <> '00000000-0000-0000-0000-000000000000'::UUID),
+    CHECK (policy_epoch > 0),
+    CHECK (requirement IN ('not_required', 'attested_key', 'provisioned', 'tofu')),
+    CHECK (effective_from >= 0),
+    CHECK (effective_from < effective_until)
+);
+
+-- The policy UUID is a stable namespace and every semantic replacement must
+-- advance its positive epoch. This makes an ID/epoch comparison inside the
+-- binding transaction sufficient to detect requirement or interval drift.
+CREATE FUNCTION enforce_identity_enrollment_policy_lineage()
+RETURNS TRIGGER LANGUAGE plpgsql AS $$
+BEGIN
+    IF NEW.community_id <> OLD.community_id
+       OR NEW.policy_id <> OLD.policy_id
+       OR NEW.policy_epoch <= OLD.policy_epoch THEN
+        RAISE EXCEPTION 'identity enrollment policy lineage must advance monotonically';
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+CREATE TRIGGER identity_enrollment_policy_lineage_guard
+BEFORE UPDATE ON identity_enrollment_policies
+FOR EACH ROW EXECUTE FUNCTION enforce_identity_enrollment_policy_lineage();
+
 ALTER TABLE identity_bindings
     ADD COLUMN binding_id UUID,
     ADD COLUMN binding_version BIGINT,
@@ -13,7 +64,12 @@ ALTER TABLE identity_bindings
     ADD COLUMN binding_provenance TEXT,
     ADD COLUMN replacement_binding_id UUID,
     ADD COLUMN created_by BYTEA,
-    ADD COLUMN created_policy_version TEXT;
+    ADD COLUMN created_policy_version TEXT,
+    ADD COLUMN expires_at TIMESTAMPTZ,
+    ADD COLUMN creation_attribution_kind TEXT,
+    ADD COLUMN archived_at TIMESTAMPTZ,
+    ADD COLUMN archived_by BYTEA,
+    ADD COLUMN archived_reason TEXT;
 
 -- Every row receives a stable persisted identifier. Unique legacy exact pairs
 -- use a reproducible length-prefixed hash. Byte-identical duplicate rows lack
@@ -77,7 +133,8 @@ SET binding_state = CASE
     binding_provenance = CASE source
         WHEN 'jwt_npub' THEN 'attested_key'
         ELSE 'tofu'
-    END;
+    END,
+    creation_attribution_kind = 'legacy_unknown';
 
 ALTER TABLE identity_bindings
     ALTER COLUMN binding_id SET NOT NULL,
@@ -86,21 +143,56 @@ ALTER TABLE identity_bindings
     ALTER COLUMN binding_state SET DEFAULT 'active',
     ALTER COLUMN binding_provenance SET NOT NULL,
     ALTER COLUMN binding_provenance SET DEFAULT 'tofu',
-    ADD CONSTRAINT identity_bindings_o3_id_unique
+    ALTER COLUMN creation_attribution_kind SET NOT NULL,
+    ADD CONSTRAINT identity_bindings_binding_id_unique
         UNIQUE (community_id, binding_id),
-    ADD CONSTRAINT chk_identity_bindings_o3_id_not_nil
+    ADD CONSTRAINT chk_identity_bindings_id_not_nil
         CHECK (binding_id <> '00000000-0000-0000-0000-000000000000'::UUID),
-    ADD CONSTRAINT chk_identity_bindings_o3_state
-        CHECK (binding_state IN ('active', 'revoked', 'rotated')),
-    ADD CONSTRAINT chk_identity_bindings_o3_provenance
+    ADD CONSTRAINT chk_identity_bindings_state
+        CHECK (binding_state IN ('active', 'revoked', 'rotated', 'archived')),
+    ADD CONSTRAINT chk_identity_bindings_provenance
         CHECK (binding_provenance IN ('attested_key', 'provisioned', 'tofu')),
-    ADD CONSTRAINT chk_identity_bindings_o3_created_by_len
+    ADD CONSTRAINT chk_identity_bindings_created_by_len
         CHECK (created_by IS NULL OR length(created_by) = 32),
-    ADD CONSTRAINT chk_identity_bindings_o3_policy_version
-        CHECK (created_policy_version IS NULL OR length(created_policy_version) > 0);
+    ADD CONSTRAINT chk_identity_bindings_policy_version
+        CHECK (created_policy_version IS NULL OR length(created_policy_version) > 0),
+    ADD CONSTRAINT chk_identity_bindings_expiry
+        CHECK (expires_at IS NULL OR expires_at > TIMESTAMPTZ 'epoch'),
+    ADD CONSTRAINT chk_identity_bindings_creation_attribution
+        CHECK (
+            (creation_attribution_kind = 'legacy_unknown'
+                AND created_by IS NULL AND created_policy_version IS NULL)
+            OR
+            (creation_attribution_kind IN ('authenticated_key', 'operator')
+                AND created_by IS NOT NULL AND length(created_by) = 32
+                AND created_policy_version IS NOT NULL
+                AND length(created_policy_version) > 0)
+        ),
+    ADD CONSTRAINT chk_identity_bindings_authority_state
+        CHECK ((binding_state = 'active') = (revoked_at IS NULL)),
+    ADD CONSTRAINT chk_identity_bindings_archive_attribution
+        CHECK (
+            (binding_state <> 'archived'
+                AND archived_at IS NULL AND archived_by IS NULL AND archived_reason IS NULL)
+            OR
+            (binding_state = 'archived'
+                AND archived_at IS NOT NULL
+                AND archived_by IS NOT NULL AND length(archived_by) = 32
+                AND archived_reason IS NOT NULL AND length(archived_reason) > 0)
+        );
+
+-- Preserve the checksum-frozen legacy indexes while making the authoritative
+-- predicate explicit in the current catalog as well as in every authority read.
+CREATE UNIQUE INDEX idx_identity_bindings_authoritative_principal
+    ON identity_bindings (community_id, issuer, uid)
+    WHERE binding_state = 'active' AND revoked_at IS NULL;
+CREATE UNIQUE INDEX idx_identity_bindings_authoritative_pubkey
+    ON identity_bindings (community_id, pubkey)
+    WHERE binding_state = 'active' AND revoked_at IS NULL;
 
 -- Invalid legacy graphs remain stored verbatim but are not usable as binding
--- authority. O3 exposes no operation that clears these migration denials.
+-- authority. The binding subsystem exposes no operation that clears these
+-- migration denials.
 CREATE TABLE identity_migration_denials (
     community_id UUID NOT NULL REFERENCES communities(id),
     issuer       TEXT NOT NULL,
@@ -266,92 +358,16 @@ JOIN identity_migration_denials denial
  AND denial.subject = binding.uid
 WHERE binding.rotated_to_pubkey IS NOT NULL;
 
--- Topological versions are deterministic for valid histories. Quarantined
--- rows receive stable positive versions solely for attribution, never auth.
-WITH RECURSIVE
-candidate_edges AS (
-    SELECT
-        predecessor.community_id,
-        predecessor.issuer,
-        predecessor.uid,
-        predecessor.binding_id AS predecessor_id,
-        successor.binding_id AS successor_id,
-        COUNT(*) OVER (
-            PARTITION BY predecessor.community_id, predecessor.binding_id
-        ) AS outgoing_candidates
-    FROM identity_bindings predecessor
-    JOIN identity_bindings successor
-      ON successor.community_id = predecessor.community_id
-     AND successor.issuer = predecessor.issuer
-     AND successor.uid = predecessor.uid
-     AND successor.pubkey = predecessor.rotated_to_pubkey
-     AND successor.binding_id <> predecessor.binding_id
-     AND successor.created_at >= predecessor.rotation_completed_at
-    WHERE predecessor.rotation_completed_at IS NOT NULL
-),
-resolved_edges AS (
-    SELECT community_id, issuer, uid, predecessor_id, successor_id
-    FROM candidate_edges
-    WHERE outgoing_candidates = 1
-),
-roots AS (
-    SELECT node.community_id, node.issuer, node.uid, node.binding_id
-    FROM identity_bindings node
-    WHERE NOT EXISTS (
-        SELECT 1 FROM resolved_edges edge
-        WHERE edge.community_id = node.community_id
-          AND edge.successor_id = node.binding_id
-    )
-),
-walk AS (
-    SELECT root.community_id, root.issuer, root.uid, root.binding_id, 1::BIGINT AS binding_version
-    FROM roots root
-    WHERE NOT EXISTS (
-        SELECT 1 FROM identity_migration_denials denial
-        WHERE denial.community_id = root.community_id
-          AND denial.issuer = root.issuer
-          AND denial.subject = root.uid
-    )
-    UNION ALL
-    SELECT edge.community_id, edge.issuer, edge.uid, edge.successor_id, walk.binding_version + 1
-    FROM walk
-    JOIN resolved_edges edge
-      ON edge.community_id = walk.community_id
-     AND edge.predecessor_id = walk.binding_id
-),
-quarantined AS (
-    SELECT
-        binding.community_id,
-        binding.binding_id,
-        ROW_NUMBER() OVER (
-            PARTITION BY binding.community_id, binding.issuer, binding.uid
-            ORDER BY binding.created_at, binding.updated_at,
-                     encode(binding.pubkey, 'hex'), binding.binding_id
-        )::BIGINT AS binding_version
-    FROM identity_bindings binding
-    JOIN identity_migration_denials denial
-      ON denial.community_id = binding.community_id
-     AND denial.issuer = binding.issuer
-     AND denial.subject = binding.uid
-),
-versions AS (
-    SELECT community_id, binding_id, binding_version FROM walk
-    UNION ALL
-    SELECT community_id, binding_id, binding_version FROM quarantined
-)
-UPDATE identity_bindings binding
-SET binding_version = versions.binding_version
-FROM versions
-WHERE binding.community_id = versions.community_id
-  AND binding.binding_id = versions.binding_id;
+-- A version belongs to one stable binding ID. Imported rows are snapshots of
+-- distinct legacy binding IDs, so each starts at version 1; later transitions
+-- increment only the binding ID whose authorization state changes.
+UPDATE identity_bindings SET binding_version = 1;
 
 ALTER TABLE identity_bindings
     ALTER COLUMN binding_version SET NOT NULL,
     ALTER COLUMN binding_version SET DEFAULT 1,
-    ADD CONSTRAINT chk_identity_bindings_o3_version_positive
-        CHECK (binding_version > 0),
-    ADD CONSTRAINT identity_bindings_o3_principal_version_unique
-        UNIQUE (community_id, issuer, uid, binding_version);
+    ADD CONSTRAINT chk_identity_bindings_version_positive
+        CHECK (binding_version > 0);
 
 CREATE TABLE identity_binding_lineage (
     community_id          UUID NOT NULL REFERENCES communities(id),
@@ -399,27 +415,26 @@ WHERE edge.outgoing_candidates = 1
         AND denial.subject = edge.uid
   );
 
--- The legacy rotation helper admits `db_binding` only after privileged
--- replacement proof. Roots remain TOFU; unique imported successors retain the
--- stronger provisioned provenance implied by that helper.
-UPDATE identity_bindings successor
-SET binding_provenance = 'provisioned'
-FROM identity_binding_lineage lineage
-WHERE lineage.community_id = successor.community_id
-  AND lineage.successor_binding_id = successor.binding_id
-  AND successor.source = 'db_binding';
-
 UPDATE identity_bindings predecessor
 SET replacement_binding_id = lineage.successor_binding_id
 FROM identity_binding_lineage lineage
 WHERE lineage.community_id = predecessor.community_id
   AND lineage.predecessor_binding_id = predecessor.binding_id;
 
+-- A legacy row whose successor could not be proven is inactive but not a
+-- completed rotation. Keep its legacy fields and quarantine intact while
+-- projecting the truthful binding state as revoked.
+UPDATE identity_bindings
+SET binding_state = 'revoked'
+WHERE binding_state = 'rotated' AND replacement_binding_id IS NULL;
+
 ALTER TABLE identity_bindings
-    ADD CONSTRAINT identity_bindings_o3_replacement_fk
+    ADD CONSTRAINT identity_bindings_replacement_fk
         FOREIGN KEY (community_id, replacement_binding_id)
         REFERENCES identity_bindings (community_id, binding_id)
-        DEFERRABLE INITIALLY DEFERRED;
+        DEFERRABLE INITIALLY DEFERRED,
+    ADD CONSTRAINT chk_identity_bindings_rotated_lineage
+        CHECK (binding_state <> 'rotated' OR replacement_binding_id IS NOT NULL);
 
 CREATE TABLE identity_retired_pairs (
     community_id           UUID NOT NULL REFERENCES communities(id),
@@ -524,6 +539,7 @@ WHERE terminal.revoked_at IS NOT NULL
       WHERE active.community_id = terminal.community_id
         AND active.issuer = terminal.issuer
         AND active.uid = terminal.uid
+        AND active.binding_state = 'active'
         AND active.revoked_at IS NULL
   )
   AND NOT EXISTS (
@@ -566,12 +582,12 @@ CREATE TABLE identity_binding_history (
     CHECK (length(issuer) > 0),
     CHECK (length(subject) > 0),
     CHECK (length(pubkey) = 32),
-    CHECK (binding_state IN ('active', 'revoked', 'rotated')),
+    CHECK (binding_state IN ('active', 'revoked', 'rotated', 'archived')),
     CHECK (binding_provenance IN ('attested_key', 'provisioned', 'tofu')),
     CHECK (transition_kind IN (
         'legacy_import', 'enroll', 'provision', 'provenance_strengthened',
         'retire_pair', 'disable_identity', 'revoke_key', 'rotate',
-        'recover', 'enable_identity'
+        'recover', 'enable_identity', 'archive'
     )),
     CHECK (actor IS NULL OR length(actor) = 32),
     CHECK (length(reason) > 0)
@@ -601,7 +617,7 @@ SELECT
 FROM identity_bindings;
 
 -- Idempotency and local state history only. Authorization and complete
--- operator audit authority remain outside O3.
+-- operator audit authority remain outside the binding persistence layer.
 CREATE TABLE identity_lifecycle_operations (
     community_id          UUID NOT NULL REFERENCES communities(id),
     operation_id          UUID NOT NULL,
@@ -614,8 +630,9 @@ CREATE TABLE identity_lifecycle_operations (
     binding_id            UUID,
     replacement_binding_id UUID,
     binding_version       BIGINT,
+    replacement_binding_version BIGINT,
     selector_version      BIGINT,
-    actor                 BYTEA,
+    actor                 BYTEA NOT NULL,
     reason                TEXT NOT NULL,
     created_at            TIMESTAMPTZ NOT NULL DEFAULT NOW(),
     PRIMARY KEY (community_id, operation_id),
@@ -626,7 +643,7 @@ CREATE TABLE identity_lifecycle_operations (
     CHECK (operation_id <> '00000000-0000-0000-0000-000000000000'::UUID),
     CHECK (operation_kind IN (
         'provision', 'retire_pair', 'disable_identity', 'revoke_key',
-        'rotate', 'recover', 'enable_identity'
+        'rotate', 'recover', 'enable_identity', 'archive'
     )),
     CHECK (length(request_fingerprint) = 32),
     CHECK (issuer IS NULL OR length(issuer) > 0),
@@ -634,8 +651,9 @@ CREATE TABLE identity_lifecycle_operations (
     CHECK (pubkey IS NULL OR length(pubkey) = 32),
     CHECK (replacement_pubkey IS NULL OR length(replacement_pubkey) = 32),
     CHECK (binding_version IS NULL OR binding_version > 0),
+    CHECK (replacement_binding_version IS NULL OR replacement_binding_version > 0),
     CHECK (selector_version IS NULL OR selector_version > 0),
-    CHECK (actor IS NULL OR length(actor) = 32),
+    CHECK (length(actor) = 32),
     CHECK (length(reason) > 0)
 );
 

schema/schema.sql

@@ -187,9 +187,43 @@ CREATE UNIQUE INDEX idx_users_okta ON users (community_id, okta_user_id)
 -- ── Relay-verified identity bindings ─────────────────────────────────────────
 -- Conformance: verified identity is community-scoped. An issuer-qualified uid
 -- is the stable product/user-management identity; a Nostr pubkey is the
--- protocol credential currently bound to it. This table is intentionally a
--- binding and lifecycle authority. Revocation scope distinguishes principal
--- disablement, a single-key revocation, and an operator-authorized rotation.
+-- protocol credential currently bound to it. The binding table below is the
+-- lifecycle authority. Revocation scope distinguishes principal disablement,
+-- single-key revocation, and operator-authorized rotation.
+--
+-- Current verifier-owned enrollment policy. A fresh database intentionally
+-- contains no row, so the disabled candidate cannot enroll until a separately
+-- authorized server-configuration action installs one.
+CREATE TABLE identity_enrollment_policies (
+    community_id    UUID NOT NULL PRIMARY KEY REFERENCES communities(id),
+    policy_id       UUID NOT NULL,
+    policy_epoch    BIGINT NOT NULL,
+    requirement     TEXT NOT NULL,
+    effective_from  BIGINT NOT NULL,
+    effective_until BIGINT NOT NULL,
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    CHECK (policy_id <> '00000000-0000-0000-0000-000000000000'::UUID),
+    CHECK (policy_epoch > 0),
+    CHECK (requirement IN ('not_required', 'attested_key', 'provisioned', 'tofu')),
+    CHECK (effective_from >= 0),
+    CHECK (effective_from < effective_until)
+);
+
+CREATE FUNCTION enforce_identity_enrollment_policy_lineage()
+RETURNS TRIGGER LANGUAGE plpgsql AS $$
+BEGIN
+    IF NEW.community_id <> OLD.community_id
+       OR NEW.policy_id <> OLD.policy_id
+       OR NEW.policy_epoch <= OLD.policy_epoch THEN
+        RAISE EXCEPTION 'identity enrollment policy lineage must advance monotonically';
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+CREATE TRIGGER identity_enrollment_policy_lineage_guard
+BEFORE UPDATE ON identity_enrollment_policies
+FOR EACH ROW EXECUTE FUNCTION enforce_identity_enrollment_policy_lineage();
 
 CREATE TABLE identity_bindings (
     community_id    UUID NOT NULL REFERENCES communities(id),
@@ -217,6 +251,11 @@ CREATE TABLE identity_bindings (
     replacement_binding_id UUID,
     created_by       BYTEA,
     created_policy_version TEXT,
+    expires_at       TIMESTAMPTZ,
+    creation_attribution_kind TEXT NOT NULL,
+    archived_at      TIMESTAMPTZ,
+    archived_by      BYTEA,
+    archived_reason  TEXT,
     CONSTRAINT chk_identity_bindings_issuer_not_empty CHECK (length(issuer) > 0),
     CONSTRAINT chk_identity_bindings_uid_not_empty CHECK (length(uid) > 0),
     CONSTRAINT chk_identity_bindings_pubkey_len CHECK (length(pubkey) = 32),
@@ -234,32 +273,63 @@ CREATE TABLE identity_bindings (
             AND rotation_reason IS NOT NULL
             AND length(rotation_reason) > 0)
     ),
-    CONSTRAINT identity_bindings_o3_id_unique UNIQUE (community_id, binding_id),
-    CONSTRAINT identity_bindings_o3_principal_version_unique
-        UNIQUE (community_id, issuer, uid, binding_version),
-    CONSTRAINT identity_bindings_o3_replacement_fk
+    CONSTRAINT identity_bindings_binding_id_unique UNIQUE (community_id, binding_id),
+    CONSTRAINT identity_bindings_replacement_fk
         FOREIGN KEY (community_id, replacement_binding_id)
         REFERENCES identity_bindings (community_id, binding_id)
         DEFERRABLE INITIALLY DEFERRED,
-    CONSTRAINT chk_identity_bindings_o3_id_not_nil
+    CONSTRAINT chk_identity_bindings_id_not_nil
         CHECK (binding_id <> '00000000-0000-0000-0000-000000000000'::UUID),
-    CONSTRAINT chk_identity_bindings_o3_version_positive CHECK (binding_version > 0),
-    CONSTRAINT chk_identity_bindings_o3_state
-        CHECK (binding_state IN ('active', 'revoked', 'rotated')),
-    CONSTRAINT chk_identity_bindings_o3_provenance
+    CONSTRAINT chk_identity_bindings_version_positive CHECK (binding_version > 0),
+    CONSTRAINT chk_identity_bindings_state
+        CHECK (binding_state IN ('active', 'revoked', 'rotated', 'archived')),
+    CONSTRAINT chk_identity_bindings_provenance
         CHECK (binding_provenance IN ('attested_key', 'provisioned', 'tofu')),
-    CONSTRAINT chk_identity_bindings_o3_created_by_len
+    CONSTRAINT chk_identity_bindings_created_by_len
         CHECK (created_by IS NULL OR length(created_by) = 32),
-    CONSTRAINT chk_identity_bindings_o3_policy_version
-        CHECK (created_policy_version IS NULL OR length(created_policy_version) > 0)
+    CONSTRAINT chk_identity_bindings_policy_version
+        CHECK (created_policy_version IS NULL OR length(created_policy_version) > 0),
+    CONSTRAINT chk_identity_bindings_expiry
+        CHECK (expires_at IS NULL OR expires_at > TIMESTAMPTZ 'epoch'),
+    CONSTRAINT chk_identity_bindings_creation_attribution CHECK (
+        (creation_attribution_kind = 'legacy_unknown'
+            AND created_by IS NULL AND created_policy_version IS NULL)
+        OR
+        (creation_attribution_kind IN ('authenticated_key', 'operator')
+            AND created_by IS NOT NULL AND length(created_by) = 32
+            AND created_policy_version IS NOT NULL
+            AND length(created_policy_version) > 0)
+    ),
+    CONSTRAINT chk_identity_bindings_authority_state
+        CHECK ((binding_state = 'active') = (revoked_at IS NULL)),
+    CONSTRAINT chk_identity_bindings_archive_attribution CHECK (
+        (binding_state <> 'archived'
+            AND archived_at IS NULL AND archived_by IS NULL AND archived_reason IS NULL)
+        OR
+        (binding_state = 'archived'
+            AND archived_at IS NOT NULL
+            AND archived_by IS NOT NULL AND length(archived_by) = 32
+            AND archived_reason IS NOT NULL AND length(archived_reason) > 0)
+    ),
+    CONSTRAINT chk_identity_bindings_rotated_lineage
+        CHECK (binding_state <> 'rotated' OR replacement_binding_id IS NOT NULL)
 );
 
+-- Frozen 0027 compatibility indexes. The authority-state CHECK above makes
+-- `revoked_at IS NULL` equivalent to `binding_state = 'active'`; authoritative
+-- readers and the current indexes below still spell out both predicates.
 CREATE UNIQUE INDEX idx_identity_bindings_active_principal
     ON identity_bindings (community_id, issuer, uid)
     WHERE revoked_at IS NULL;
 CREATE UNIQUE INDEX idx_identity_bindings_active_pubkey
     ON identity_bindings (community_id, pubkey)
     WHERE revoked_at IS NULL;
+CREATE UNIQUE INDEX idx_identity_bindings_authoritative_principal
+    ON identity_bindings (community_id, issuer, uid)
+    WHERE binding_state = 'active' AND revoked_at IS NULL;
+CREATE UNIQUE INDEX idx_identity_bindings_authoritative_pubkey
+    ON identity_bindings (community_id, pubkey)
+    WHERE binding_state = 'active' AND revoked_at IS NULL;
 CREATE INDEX idx_identity_bindings_pubkey
     ON identity_bindings (community_id, pubkey);
 CREATE INDEX idx_identity_bindings_revoked_principal
@@ -422,12 +492,12 @@ CREATE TABLE identity_binding_history (
     CHECK (length(issuer) > 0),
     CHECK (length(subject) > 0),
     CHECK (length(pubkey) = 32),
-    CHECK (binding_state IN ('active', 'revoked', 'rotated')),
+    CHECK (binding_state IN ('active', 'revoked', 'rotated', 'archived')),
     CHECK (binding_provenance IN ('attested_key', 'provisioned', 'tofu')),
     CHECK (transition_kind IN (
         'legacy_import', 'enroll', 'provision', 'provenance_strengthened',
         'retire_pair', 'disable_identity', 'revoke_key', 'rotate',
-        'recover', 'enable_identity'
+        'recover', 'enable_identity', 'archive'
     )),
     CHECK (actor IS NULL OR length(actor) = 32),
     CHECK (length(reason) > 0)
@@ -448,8 +518,9 @@ CREATE TABLE identity_lifecycle_operations (
     binding_id             UUID,
     replacement_binding_id UUID,
     binding_version        BIGINT,
+    replacement_binding_version BIGINT,
     selector_version       BIGINT,
-    actor                  BYTEA,
+    actor                  BYTEA NOT NULL,
     reason                 TEXT NOT NULL,
     created_at             TIMESTAMPTZ NOT NULL DEFAULT NOW(),
     PRIMARY KEY (community_id, operation_id),
@@ -460,7 +531,7 @@ CREATE TABLE identity_lifecycle_operations (
     CHECK (operation_id <> '00000000-0000-0000-0000-000000000000'::UUID),
     CHECK (operation_kind IN (
         'provision', 'retire_pair', 'disable_identity', 'revoke_key',
-        'rotate', 'recover', 'enable_identity'
+        'rotate', 'recover', 'enable_identity', 'archive'
     )),
     CHECK (length(request_fingerprint) = 32),
     CHECK (issuer IS NULL OR length(issuer) > 0),
@@ -468,8 +539,9 @@ CREATE TABLE identity_lifecycle_operations (
     CHECK (pubkey IS NULL OR length(pubkey) = 32),
     CHECK (replacement_pubkey IS NULL OR length(replacement_pubkey) = 32),
     CHECK (binding_version IS NULL OR binding_version > 0),
+    CHECK (replacement_binding_version IS NULL OR replacement_binding_version > 0),
     CHECK (selector_version IS NULL OR selector_version > 0),
-    CHECK (actor IS NULL OR length(actor) = 32),
+    CHECK (length(actor) = 32),
     CHECK (length(reason) > 0)
 );