fix: preserve BReg permission boundaries and explanations

crates/registry-breg/Cargo.toml

@@ -400,6 +400,16 @@ name = "postgres_proposal"
 path = "tests/postgres_proposal.rs"
 required-features = ["postgres-test"]
 
+[[test]]
+name = "postgres_proposal_task_authority"
+path = "tests/postgres_proposal_task_authority.rs"
+required-features = ["postgres-test"]
+
+[[test]]
+name = "postgres_proposal_recovery_permissions"
+path = "tests/postgres_proposal_recovery_permissions.rs"
+required-features = ["postgres-test", "tooling"]
+
 [[test]]
 name = "postgres_proposal_human_recovery"
 path = "tests/postgres_proposal_human_recovery.rs"

crates/registry-breg/src/access.rs

@@ -335,7 +335,6 @@ pub(crate) fn compiled_access_findings(
         // A `self` consent issuer's subject and record targets are bound
         // through the caller's principal link, not by a row boundary.
         if reach.rows == "all"
-            && !reach.owner_only
             && !link_bound.contains(&reach.source_path)
             && entities
                 .get(&reach.entity)

crates/registry-breg/src/auth.rs

@@ -373,9 +373,11 @@ impl RegistryAuthenticator {
 fn admit_refusal_warning(last_warning: &Mutex<Option<Instant>>, now: Instant) -> bool {
     // One timestamp per authenticator bounds both memory and warning volume,
     // including concurrent requests with distinct caller-controlled key IDs.
-    let Ok(mut last) = last_warning.lock() else {
-        return false;
-    };
+    // A timestamp contains no partial authorization state. Recover it after a
+    // panic so poisoning cannot permanently silence refusal diagnostics.
+    let mut last = last_warning
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     if last
         .is_some_and(|previous| now.saturating_duration_since(previous) < REFUSAL_WARNING_INTERVAL)
     {
@@ -870,6 +872,29 @@ pub(crate) fn authentication_refused() -> Response {
 mod tests {
     use super::*;
 
+    #[test]
+    fn refusal_warnings_recover_poisoned_timestamps_and_preserve_rate_limits() {
+        let last_warning = Mutex::new(None);
+        let now = Instant::now();
+        std::thread::scope(|scope| {
+            let poisoned = scope.spawn(|| {
+                let _guard = last_warning.lock().expect("initial warning state");
+                panic!("canary warning-state poison");
+            });
+            assert!(poisoned.join().is_err());
+        });
+        assert!(last_warning.is_poisoned());
+        assert!(admit_refusal_warning(&last_warning, now));
+        assert!(!admit_refusal_warning(
+            &last_warning,
+            now + REFUSAL_WARNING_INTERVAL - Duration::from_nanos(1)
+        ));
+        assert!(admit_refusal_warning(
+            &last_warning,
+            now + REFUSAL_WARNING_INTERVAL
+        ));
+    }
+
     #[test]
     fn key_refusal_warnings_are_bounded_across_concurrent_misses_and_resume_after_interval() {
         let last_warning = Mutex::new(None);

crates/registry-breg/src/mutation.rs

@@ -7,7 +7,9 @@ mod action;
 pub(crate) use action::erase_expired_action_evidence;
 pub(crate) use action::{HookChangeApplication, HookChangeOutcome};
 mod proposal;
-pub(crate) use proposal::{proposal_operation_name, ProposalApplyPreflight};
+pub(crate) use proposal::{
+    proposal_operation_name, CheckedProposalTaskAuthority, ProposalApplyPreflight,
+};
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;

crates/registry-breg/src/mutation/proposal.rs

@@ -27,6 +27,7 @@ use crate::proposal_workflow::{
 pub(crate) enum ProposalApplyPreflight {
     Receipt(MutationOutcome),
     Continue {
+        task_authority: Option<crate::task_grant::TaskGrantBinding>,
         review: Option<(
             String,
             registry_review_client::ReviewRequestAccepted,
@@ -39,6 +40,42 @@ pub(crate) enum ProposalApplyPreflight {
     },
 }
 
+/// The retained submission authority observed outside the effect transaction.
+///
+/// Apply reloads this exact binding under PostgreSQL before releasing effects.
+/// The remote authority cannot share that transaction, so the local deadline
+/// is also checked at effect admission and again immediately before commit.
+pub(crate) struct CheckedProposalTaskAuthority {
+    retained: Option<crate::task_grant::TaskGrantBinding>,
+}
+
+impl CheckedProposalTaskAuthority {
+    pub(crate) fn new(retained: Option<crate::task_grant::TaskGrantBinding>) -> Self {
+        Self { retained }
+    }
+
+    fn verify_binding(
+        &self,
+        retained: Option<&crate::task_grant::TaskGrantBinding>,
+    ) -> Result<(), MutationError> {
+        if retained != self.retained.as_ref() {
+            return Err(MutationError::PreconditionFailed);
+        }
+        self.verify_current()
+    }
+
+    pub(crate) fn verify_current(&self) -> Result<(), MutationError> {
+        if self
+            .retained
+            .as_ref()
+            .is_some_and(|binding| !binding.is_current())
+        {
+            return Err(MutationError::PreconditionFailed);
+        }
+        Ok(())
+    }
+}
+
 fn check_current_proposal_grant(authority: &ProposalAuthority) -> Result<(), MutationError> {
     if authority
         .task_grant
@@ -943,6 +980,13 @@ impl MutationTransaction {
             return Err(MutationError::PreconditionFailed);
         }
         verify_apply_binding(input, proposal)?;
+        let task_authority = proposal_store::load_task_authority(
+            tx,
+            &document.action_id,
+            id,
+            document.header.submission_version,
+        )
+        .await?;
         if !input.authority.is_system()
             && !crate::api::proposals::external_recovery_actor_allowed(
                 proposal.application_policy(),
@@ -1002,6 +1046,7 @@ impl MutationTransaction {
             vec![]
         };
         Ok(ProposalApplyPreflight::Continue {
+            task_authority,
             review: accepted,
             evidence,
         })
@@ -1015,6 +1060,7 @@ impl MutationTransaction {
         input: ProposalMutationInput<'_>,
         review_evidence: Option<&crate::review_integration::AcceptedReviewEvidence>,
         commit_evidence: &[crate::action_evidence_client::VerifiedAcquisition],
+        checked_task_authority: Option<&CheckedProposalTaskAuthority>,
         deadline: tokio::time::Instant,
     ) -> Result<MutationOutcome, MutationError> {
         let action = proposal_action(registry, input.action_id)?;
@@ -1579,6 +1625,16 @@ impl MutationTransaction {
                             .current_proposal()
                             .ok_or(MutationError::PreconditionFailed)?;
                         verify_apply_binding(&input, frozen)?;
+                        let checked_task_authority =
+                            checked_task_authority.ok_or(MutationError::Unavailable)?;
+                        let retained_task_authority = proposal_store::load_task_authority(
+                            tx,
+                            &action.id,
+                            id,
+                            document.header.submission_version,
+                        )
+                        .await?;
+                        checked_task_authority.verify_binding(retained_task_authority.as_ref())?;
                         if !input.authority.is_system()
                             && !crate::api::proposals::external_recovery_actor_allowed(
                                 frozen.application_policy(),
@@ -1740,6 +1796,11 @@ impl MutationTransaction {
                 return Err(MutationError::Unavailable);
             }
             check_current_proposal_grant(input.authority)?;
+            if input.operation == ProposalAuthorityOperation::Lifecycle(ProposalOperation::Apply) {
+                checked_task_authority
+                    .ok_or(MutationError::Unavailable)?
+                    .verify_current()?;
+            }
             guarded
                 .commit()
                 .await
@@ -1846,7 +1907,6 @@ impl MutationTransaction {
         if tokio::time::Instant::now() >= deadline {
             return Err(MutationError::Unavailable);
         }
-        check_current_proposal_grant(input.authority)?;
         if input.operation == ProposalAuthorityOperation::Lifecycle(ProposalOperation::Apply) {
             let frozen = workflow
                 .current_proposal()
@@ -1891,6 +1951,12 @@ impl MutationTransaction {
         if tokio::time::Instant::now() >= deadline {
             return Err(MutationError::Unavailable);
         }
+        check_current_proposal_grant(input.authority)?;
+        if input.operation == ProposalAuthorityOperation::Lifecycle(ProposalOperation::Apply) {
+            checked_task_authority
+                .ok_or(MutationError::Unavailable)?
+                .verify_current()?;
+        }
         guarded
             .commit()
             .await

crates/registry-breg/src/postgres/proposal.rs

@@ -6,7 +6,7 @@ use super::*;
 use crate::api::proposals::{ProposalListQuery, ProposalMutationInput};
 use crate::api::VerifiedRequestClaims;
 use crate::model::ProposalOperation;
-use crate::mutation::ProposalApplyPreflight;
+use crate::mutation::{CheckedProposalTaskAuthority, ProposalApplyPreflight};
 use crate::postgres::{begin_proposal_transaction, ProposalAuthority, ProposalAuthorityOperation};
 use crate::proposal_store;
 use std::collections::{BTreeMap, BTreeSet};
@@ -515,63 +515,72 @@ impl PostgresRecordMutationService {
                 .await
                 .map_err(|_| MutationError::Unavailable)??;
         }
-        let (review_evidence, commit_requests) =
-            if input.operation == ProposalAuthorityOperation::Lifecycle(ProposalOperation::Apply) {
-                let preflight = {
-                    let client = tokio::time::timeout_at(deadline, self.pool.get())
-                        .await
-                        .map_err(|_| MutationError::Unavailable)?
-                        .map_err(|_| MutationError::Unavailable)?;
-                    let mut guard = QueryCancellationGuard::new(self.pool.clone(), client);
-                    let preflight = tokio::time::timeout_at(
+        let (review_evidence, commit_requests, checked_task_authority) = if input.operation
+            == ProposalAuthorityOperation::Lifecycle(ProposalOperation::Apply)
+        {
+            let preflight = {
+                let client = tokio::time::timeout_at(deadline, self.pool.get())
+                    .await
+                    .map_err(|_| MutationError::Unavailable)?
+                    .map_err(|_| MutationError::Unavailable)?;
+                let mut guard = QueryCancellationGuard::new(self.pool.clone(), client);
+                let preflight = tokio::time::timeout_at(
+                    deadline,
+                    self.coordinator.preflight_proposal_apply(
+                        guard.client(),
+                        &self.registry,
+                        &input,
+                    ),
+                )
+                .await;
+                match preflight {
+                    Ok(result) => {
+                        guard.disarm();
+                        result?
+                    }
+                    Err(_) => {
+                        guard.cancel_and_discard().await;
+                        return Err(MutationError::Unavailable);
+                    }
+                }
+            };
+            let (task_authority, review, evidence) = match preflight {
+                ProposalApplyPreflight::Receipt(receipt) => return Ok(receipt),
+                ProposalApplyPreflight::Continue {
+                    task_authority,
+                    review,
+                    evidence,
+                } => (task_authority, review, evidence),
+            };
+            if let Some(binding) = &task_authority {
+                tokio::time::timeout_at(deadline, self.coordinator.check_task_authority(binding))
+                    .await
+                    .map_err(|_| MutationError::Unavailable)??;
+            }
+            let checked_task_authority = CheckedProposalTaskAuthority::new(task_authority);
+            let review_evidence = match review {
+                None => None,
+                Some((authority, accepted, result_id)) => {
+                    let source = self
+                        .review_result_source
+                        .as_ref()
+                        .ok_or(MutationError::Unavailable)?;
+                    let fresh = tokio::time::timeout_at(
                         deadline,
-                        self.coordinator.preflight_proposal_apply(
-                            guard.client(),
-                            &self.registry,
-                            &input,
-                        ),
+                        source.approved_evidence(&authority, &accepted),
                     )
-                    .await;
-                    match preflight {
-                        Ok(result) => {
-                            guard.disarm();
-                            result?
-                        }
-                        Err(_) => {
-                            guard.cancel_and_discard().await;
-                            return Err(MutationError::Unavailable);
-                        }
-                    }
-                };
-                match preflight {
-                    ProposalApplyPreflight::Receipt(receipt) => return Ok(receipt),
-                    ProposalApplyPreflight::Continue {
-                        review: None,
-                        evidence,
-                    } => (None, evidence),
-                    ProposalApplyPreflight::Continue {
-                        review: Some((authority, accepted, result_id)),
-                        evidence,
-                    } => {
-                        let source = self
-                            .review_result_source
-                            .as_ref()
-                            .ok_or(MutationError::Unavailable)?;
-                        let fresh = tokio::time::timeout_at(
-                            deadline,
-                            source.approved_evidence(&authority, &accepted),
-                        )
-                        .await
-                        .map_err(|_| MutationError::Unavailable)??;
-                        if result_id.is_some_and(|result_id| fresh.result_id() != result_id) {
-                            return Err(MutationError::ProposalReviewResultMismatch);
-                        }
-                        (Some(fresh), evidence)
+                    .await
+                    .map_err(|_| MutationError::Unavailable)??;
+                    if result_id.is_some_and(|result_id| fresh.result_id() != result_id) {
+                        return Err(MutationError::ProposalReviewResultMismatch);
                     }
+                    Some(fresh)
                 }
-            } else {
-                (None, vec![])
             };
+            (review_evidence, evidence, Some(checked_task_authority))
+        } else {
+            (None, vec![], None)
+        };
         // Preflight admitted and projected only the frozen guard selectors.
         // Reuse the configured, pinned provider adapter with no checked-out
         // database connection; the coordinator rechecks those same selectors.
@@ -618,6 +627,7 @@ impl PostgresRecordMutationService {
                 input,
                 review_evidence.as_ref(),
                 &commit_evidence,
+                checked_task_authority.as_ref(),
                 deadline,
             ),
         )

crates/registry-breg/src/proposal_retention.rs

@@ -913,15 +913,24 @@ pub async fn guard_successor_activation(
         let CompiledActionExecution::Proposal { proposal } = &action.execution else {
             return Err(ProposalRetentionError::ActiveProposalRecoveryUnavailable);
         };
-        if !proposal.permissions.iter().any(|grant| {
+        let inspection_available = proposal.permissions.iter().any(|grant| {
             grant
                 .operations
                 .contains(&crate::model::ProposalOperation::Get)
-        }) || !proposal.permissions.iter().any(|grant| {
+        });
+        let cancellation_recovery_available = proposal.permissions.iter().any(|grant| {
             grant
                 .operations
                 .contains(&crate::model::ProposalOperation::Cancel)
-        }) {
+                && grant.operations.iter().any(|operation| {
+                    matches!(
+                        operation,
+                        crate::model::ProposalOperation::Get
+                            | crate::model::ProposalOperation::List
+                    )
+                })
+        });
+        if !inspection_available || !cancellation_recovery_available {
             return Err(ProposalRetentionError::ActiveProposalRecoveryUnavailable);
         }
     }