diff --git a/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json b/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json
deleted file mode 100644
index 9dcadd077f7..00000000000
--- a/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json
+++ /dev/null
@@ -1,22 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "role_prefix",
-        "type_info": "Text"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d"
-}
diff --git a/.sqlx/query-d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa.json b/.sqlx/query-d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa.json
deleted file mode 100644
index ad62491448e..00000000000
--- a/.sqlx/query-d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa.json
+++ /dev/null
@@ -1,23 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "select true as whatever_column from internal.user_roles($1, 'admin') where starts_with(role_prefix, $2)",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "whatever_column",
-        "type_info": "Bool"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Text"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa"
-}
diff --git a/crates/agent/src/directives/mod.rs b/crates/agent/src/directives/mod.rs
index 40ff93ffa2f..cd4a1c47f08 100644
--- a/crates/agent/src/directives/mod.rs
+++ b/crates/agent/src/directives/mod.rs
@@ -1,7 +1,4 @@
-use control_plane_api::{
-    directives::{Row, fetch_directive, resolve},
-    logs,
-};
+use control_plane_api::directives::{Row, fetch_directive, resolve};
 
 use anyhow::Context;
 use serde::{Deserialize, Serialize};
@@ -11,7 +8,6 @@ use validator::Validate;
 pub mod accept_demo_tenant;
 pub mod beta_onboard;
 pub mod click_to_accept;
-pub mod storage_mappings;
 
 /// JobStatus is the possible outcomes of a handled directive operation.
 #[derive(Debug, Deserialize, Serialize)]
@@ -36,26 +32,27 @@ impl JobStatus {
     }
 }
 
+// The `storageMappings` directive type was removed after the GraphQL
+// storage-mapping mutations superseded it (2026-03-16). A queued row of that
+// type now resolves as `invalidDirective` (unknown variant) instead of being
+// applied.
 #[derive(Debug, Deserialize, Serialize, schemars::JsonSchema)]
 #[serde(rename_all = "camelCase", tag = "type")]
 pub enum Directive {
     BetaOnboard(beta_onboard::Directive),
     ClickToAccept(click_to_accept::Directive),
     AcceptDemoTenant(accept_demo_tenant::Directive),
-    StorageMappings(storage_mappings::Directive),
 }
 
 #[derive(Clone)]
 pub struct DirectiveHandler {
     accounts_user_email: String,
-    logs_tx: logs::Tx,
 }
 
 impl DirectiveHandler {
-    pub fn new(accounts_user_email: String, logs_tx: &logs::Tx) -> Self {
+    pub fn new(accounts_user_email: String) -> Self {
         Self {
             accounts_user_email,
-            logs_tx: logs_tx.clone(),
         }
     }
 }
@@ -132,9 +129,6 @@ impl DirectiveHandler {
             }
             Ok(Directive::ClickToAccept(d)) => click_to_accept::apply(d, row, txn).await?,
             Ok(Directive::AcceptDemoTenant(d)) => accept_demo_tenant::apply(d, row, txn).await?,
-            Ok(Directive::StorageMappings(d)) => {
-                storage_mappings::apply(d, row, &self.logs_tx, txn).await?
-            }
         };
         Ok(status)
     }
diff --git a/crates/agent/src/directives/storage_mappings.rs b/crates/agent/src/directives/storage_mappings.rs
deleted file mode 100644
index aece82ea866..00000000000
--- a/crates/agent/src/directives/storage_mappings.rs
+++ /dev/null
@@ -1,325 +0,0 @@
-use std::io::Write;
-
-use crate::directives::JobStatus;
-use anyhow::Context;
-use control_plane_api::{
-    directives::{
-        Row,
-        storage_mappings::{
-            StorageMapping, fetch_storage_mappings, upsert_storage_mapping,
-            user_has_admin_capability,
-        },
-    },
-    jobs, logs,
-};
-use serde::{Deserialize, Serialize};
-use sqlx::types::Uuid;
-use validator::Validate;
-
-#[derive(Debug, Deserialize, Serialize, Validate, schemars::JsonSchema)]
-#[serde(rename_all = "camelCase", deny_unknown_fields)]
-pub struct Directive {}
-
-#[derive(Debug, Deserialize, Serialize, Validate, schemars::JsonSchema)]
-#[serde(rename_all = "camelCase", deny_unknown_fields)]
-pub struct Claims {
-    add_store: models::Store,
-    catalog_prefix: models::Prefix,
-}
-
-#[tracing::instrument(skip_all, ret, err, fields(row.user_claims))]
-pub async fn apply(
-    _: Directive,
-    row: Row,
-    logs_tx: &logs::Tx,
-    txn: &mut sqlx::Transaction<'_, sqlx::Postgres>,
-) -> anyhow::Result<JobStatus> {
-    let detail = format!(
-        "updated by user {} via applied directive {}",
-        row.user_id, row.apply_id
-    );
-    let (collection_data, recovery) = match validate(txn, logs_tx, row).await {
-        Ok(c) => c,
-        Err(err) => {
-            return Ok(JobStatus::invalid_claims(err));
-        }
-    };
-
-    let ProposedMapping {
-        catalog_prefix,
-        spec,
-    } = collection_data;
-    upsert_storage_mapping(Some(&detail), &catalog_prefix, spec, txn).await?;
-    let ProposedMapping {
-        catalog_prefix,
-        spec,
-    } = recovery;
-    upsert_storage_mapping(Some(&detail), &catalog_prefix, spec, txn).await?;
-
-    Ok(JobStatus::Success)
-}
-
-pub struct ProposedMapping {
-    catalog_prefix: String,
-    spec: models::StorageDef,
-}
-
-fn add_store(stores: &mut models::StorageDef, store: models::Store) {
-    // If there's already an equivalent store, then remove it so that we don't end up with
-    // duplicates. This could happen if someone added store A, then store B, then store A again.
-    stores.stores.retain(|s| s != &store);
-    stores.stores.insert(0, store);
-}
-
-async fn validate(
-    txn: &mut sqlx::Transaction<'_, sqlx::Postgres>,
-    logs_tx: &logs::Tx,
-    row: Row,
-) -> anyhow::Result<(ProposedMapping, ProposedMapping)> {
-    let claims: Claims =
-        serde_json::from_str(row.user_claims.get()).context("parsing user_claims")?;
-
-    if !claims.catalog_prefix.ends_with('/') {
-        anyhow::bail!("invalid prefix, must end with '/'");
-    }
-
-    // Storage mappings can only be updated for an entire tenant. We may one day wish to support
-    // updates to narrower prefixes, but are trying to keep it simple for now.
-    if claims.catalog_prefix.matches('/').count() > 1 {
-        anyhow::bail!(
-            "catalog prefix contains too many path segments. Only top-level tenant prefixes can have storage mappings altered"
-        );
-    }
-
-    // Note: we must assert that user has admin capability for the _entire tenant_, even if in the
-    // future we allow for updating mappings of narrower prefixes. This is required because a new
-    // storage mapping for `a/b/` may implicitly override the existing mapping for `a/`.
-    let user_has_admin =
-        user_has_admin_capability(row.user_id, &claims.catalog_prefix, txn).await?;
-    anyhow::ensure!(
-        user_has_admin,
-        "user does not have required 'admin' capability to '{}'",
-        claims.catalog_prefix
-    );
-
-    // Check that we can actually access the storage bucket before fetching (and locking) the
-    // existing `storage_mappings` rows, since this check requires multiple network round trips.
-    check_bucket_access(row.logs_token, logs_tx, &claims.add_store).await?;
-
-    let recovery_prefix = format!("recovery/{}", claims.catalog_prefix);
-    let existing_mappings =
-        fetch_storage_mappings(&claims.catalog_prefix, &recovery_prefix, txn).await?;
-    let (mut collection_data, mut recovery_data) = parse_existing(existing_mappings)?;
-
-    let mut collection_store = claims.add_store.clone();
-    // The storage mapping for collection data should always have a `collection-data/` prefix in
-    // order to segregate it from recovery log data. This makes it easier to apply bucket lifecycle
-    // policies, since you can target just the `collection-data/` prefix. If the store already
-    // specifies a prefix, then we'll add `collection-data/` to the end.
-    collection_store
-        .prefix_mut()
-        .as_mut_string()
-        .push_str("collection-data/");
-    add_store(&mut collection_data.spec, collection_store);
-    // The recovery log store doesn't need a separate prefix because all recovery log journals
-    // already begin with `recovery/`.
-    add_store(&mut recovery_data.spec, claims.add_store.clone());
-
-    Ok((collection_data, recovery_data))
-}
-
-// This is a macro instead of a function to work around the fact that file paths are `OsStr`s
-// instead of regular `&str`s.
-macro_rules! check_command {
-    ($prog:expr, $($arg:expr),* $(; region $region:expr)?) => {{
-        let mut cmd = std::process::Command::new($prog);
-        $(
-            cmd.arg($arg);
-        )*
-        $(
-          if let Some(region_name) = $region {
-              cmd.arg("--region");
-              cmd.arg(region_name);
-          }
-        )?
-        cmd
-    } };
-}
-
-async fn check_bucket_access(
-    logs_token: Uuid,
-    logs_tx: &logs::Tx,
-    store: &models::Store,
-) -> anyhow::Result<()> {
-    let mut test_file = tempfile::NamedTempFile::new().context("creating temp file")?;
-    test_file
-        .write_all(TEST_FILE_CONTENT.as_bytes())
-        .context("writing test file content")?;
-    let test_file_path = test_file.path();
-
-    let commands = match store {
-        models::Store::S3(conf) => vec![
-            (
-                "put object",
-                check_command!(
-                    "aws",
-                    "s3",
-                    "cp",
-                    test_file_path,
-                    without_query(conf.as_url()).join(TEST_FILENAME)?.to_string()
-                    ; region conf.region.as_ref()),
-            ),
-            // List comes after put, so the prefix, if configured, is guaranteed to exist.
-            (
-                "list bucket",
-                check_command!("aws", "s3", "ls", without_query(conf.as_url()).to_string()
-                    ; region conf.region.as_ref()),
-            ),
-            (
-                "get object",
-                // Copy to stdout to avoid needing to cleanup a temp file.
-                // Don't use /dev/null because the cli will exit non-zero even when it gets the file successfully.
-                check_command!(
-                    "aws",
-                    "s3",
-                    "cp",
-                    without_query(conf.as_url())
-                        .join(TEST_FILENAME)?
-                        .to_string(),
-                    "/dev/stdout"
-                    ; region conf.region.as_ref()
-                ),
-            ),
-            (
-                "delete object",
-                check_command!(
-                    "aws",
-                    "s3",
-                    "rm",
-                    without_query(conf.as_url()).join(TEST_FILENAME)?.to_string()
-                    ; region conf.region.as_ref()
-                ),
-            ),
-        ],
-        models::Store::Gcs(conf) => vec![
-            (
-                "put object",
-                check_command!(
-                    "gcloud",
-                    "storage",
-                    "cp",
-                    test_file_path,
-                    without_query(conf.as_url())
-                        .join(TEST_FILENAME)?
-                        .to_string()
-                ),
-            ),
-            // List comes after put, so the prefix, if configured, is guaranteed to exist.
-            (
-                "list bucket",
-                check_command!(
-                    "gcloud",
-                    "storage",
-                    "ls",
-                    without_query(conf.as_url()).to_string()
-                ),
-            ),
-            (
-                "get object",
-                check_command!(
-                    "gcloud",
-                    "storage",
-                    "cat",
-                    without_query(conf.as_url())
-                        .join(TEST_FILENAME)?
-                        .to_string()
-                ),
-            ),
-            (
-                "delete object",
-                check_command!(
-                    "gcloud",
-                    "storage",
-                    "rm",
-                    without_query(conf.as_url())
-                        .join(TEST_FILENAME)?
-                        .to_string()
-                ),
-            ),
-        ],
-        models::Store::Azure(_) => {
-            anyhow::bail!("checking access for azure cloud storage is not yet implemented")
-        }
-        models::Store::Custom(_) => {
-            anyhow::bail!("checking access for custom cloud storage is not supported")
-        }
-    };
-
-    for (desc, mut cmd) in commands {
-        tracing::info!(
-            %desc,
-            program = ?cmd.get_program(),
-            args = ?cmd.get_args(),
-            "running storage check"
-        );
-        let exit_status = jobs::run_without_removing_env(&desc, logs_tx, logs_token, &mut cmd)
-            .await
-            .with_context(|| {
-                format!(
-                    "failed to execute {desc} command: {:?} with: {:?}",
-                    cmd.get_program(),
-                    cmd.get_args()
-                )
-            })?;
-        if !exit_status.success() {
-            anyhow::bail!("failed to {desc}, please check that permissions are set appropriately");
-        }
-    }
-
-    Ok(())
-}
-
-fn without_query(mut uri: url::Url) -> url::Url {
-    uri.set_query(None);
-    uri
-}
-
-fn parse_existing(
-    mut existing: Vec<StorageMapping>,
-) -> anyhow::Result<(ProposedMapping, ProposedMapping)> {
-    if existing.len() != 2 {
-        anyhow::bail!("expected 2 existing storage mappings, found: {existing:?}");
-    }
-    let Some(recovery_idx) = existing
-        .iter()
-        .position(|m| m.catalog_prefix.starts_with("recovery/"))
-    else {
-        anyhow::bail!("missing recovery/ storage mapping in {existing:?}");
-    };
-    let recovery = existing.remove(recovery_idx);
-    let recovery_storage: models::StorageDef = serde_json::from_str(recovery.spec.get())
-        .context("deserializing existing recovery/ storage mapping")?;
-
-    let collection_data = existing.remove(0);
-    let collection_store: models::StorageDef = serde_json::from_str(collection_data.spec.get())
-        .context("deserializing existing storage mapping")?;
-
-    Ok((
-        ProposedMapping {
-            catalog_prefix: collection_data.catalog_prefix,
-            spec: collection_store,
-        },
-        ProposedMapping {
-            catalog_prefix: recovery.catalog_prefix,
-            spec: recovery_storage,
-        },
-    ))
-}
-
-const TEST_FILENAME: &str = "estuary_test.txt";
-
-const TEST_FILE_CONTENT: &str = r#"Estuary storage test
-This file is written to your storage bucket in order to test that we have the necessary access
-permissions to create and delete objects. If you're seeing this file stick around, then it's
-likely because we lacked the necessary permissions to delete it. You may remove this file at
-any time, and doing so will not impact the function of Estuary."#;
diff --git a/crates/agent/src/integration_tests/graphql/mutations/storage_mappings.rs b/crates/agent/src/integration_tests/graphql/mutations/storage_mappings.rs
index 6cb6a624d99..f9311b3c24c 100644
--- a/crates/agent/src/integration_tests/graphql/mutations/storage_mappings.rs
+++ b/crates/agent/src/integration_tests/graphql/mutations/storage_mappings.rs
@@ -9,6 +9,69 @@ mutation CreateStorageMapping($catalogPrefix: Prefix!, $spec: JSON!) {
 }
 "#;
 
+/// Storage-mapping mutations require effective (attenuation-aware) admin over
+/// the *entire* claimed prefix, because a mapping at any prefix shadows the
+/// tenant mapping for everything under it (longest-match wins). This pins the
+/// authorization boundary formerly guarded by the removed `storageMappings`
+/// applied-directive, whose legacy `internal.user_roles()` SQL accepted the
+/// sub-prefix shape (it compared prefixes in the loose direction) and was
+/// blind to bundle attenuation.
+#[tokio::test]
+async fn test_create_storage_mapping_authorization_denials() {
+    let mut harness = TestHarness::init("storage_mapping_authz").await;
+    let _alice = harness.setup_tenant("aliceCo").await;
+
+    // bob administers only a *sub*-prefix: authority flows downward, so it
+    // must not reach the tenant root.
+    let bob = uuid::uuid!("bbbbbbbb-0000-0000-0000-000000000000");
+    // erin reaches `aliceCo/` through a raw-`admin` role grant, but her own
+    // grant delegates only the `editor` bundle: the walk attenuates the
+    // second hop's bits below Admin.
+    let erin = uuid::uuid!("eeeeeeee-0000-0000-0000-000000000000");
+    sqlx::query(
+        r#"with users as (
+            insert into auth.users (id, email) values
+                ($1, 'subprefix-admin@example.test'), ($2, 'attenuated-admin@example.test')
+        ),
+        user_grants as (
+            insert into user_grants (user_id, object_role, capability, bundles) values
+                ($1, 'aliceCo/sub/', 'admin', '{}'),
+                ($2, 'sharedCo/', 'none', '{editor}')
+        )
+        insert into role_grants (subject_role, object_role, capability) values
+            ('sharedCo/', 'aliceCo/', 'admin')"#,
+    )
+    .bind(bob)
+    .bind(erin)
+    .execute(&harness.pool)
+    .await
+    .unwrap();
+    // An authoritative Snapshot (taken after the request starts) makes the
+    // denials terminal rather than retryable.
+    harness.refresh_snapshot_authoritative().await;
+
+    for user_id in [bob, erin] {
+        let result: Result<serde_json::Value, _> = harness
+            .execute_graphql_query(
+                user_id,
+                CREATE_STORAGE_MAPPING_MUTATION,
+                &json!({
+                    "catalogPrefix": "aliceCo/",
+                    "spec": {
+                        "stores": [{"provider": "GCS", "bucket": "test-bucket"}],
+                        "data_planes": ["ops/dp/public/test"]
+                    },
+                }),
+            )
+            .await;
+        let err = result.unwrap_err().to_string();
+        assert!(
+            err.contains("is not an authorized as an Admin of catalog prefix 'aliceCo/'"),
+            "expected an admin denial for {user_id}, got: {err}"
+        );
+    }
+}
+
 #[tokio::test]
 async fn test_create_storage_mapping_validation_errors() {
     let mut harness = TestHarness::init("storage_mapping_validation").await;
diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs
index 7343ec1c819..d81ed0f9dfd 100644
--- a/crates/agent/src/integration_tests/harness.rs
+++ b/crates/agent/src/integration_tests/harness.rs
@@ -304,7 +304,7 @@ impl HarnessBuilder {
         let controller_exec =
             crate::controllers::executor::LiveSpecControllerExecutor::new(control_plane.clone());
         let directive_exec =
-            crate::directives::DirectiveHandler::new("support@estuary.test".to_string(), &logs_tx);
+            crate::directives::DirectiveHandler::new("support@estuary.test".to_string());
 
         let mut harness = TestHarness {
             test_name,
diff --git a/crates/agent/src/lib.rs b/crates/agent/src/lib.rs
index 7d1c8127209..65955377628 100644
--- a/crates/agent/src/lib.rs
+++ b/crates/agent/src/lib.rs
@@ -1,3 +1,12 @@
+// Computing the layout of `PublicationsExecutor::poll`'s async body recurses
+// through every future of the nested publication pipeline — a ~130-deep query
+// descent, over rustc's default limit of 128. Earlier trees compiled only
+// because since-deleted code (the storage-mappings directive chain) happened
+// to pre-compute shared sub-layouts, splitting the descent below the limit.
+// Raise the limit per rustc's own guidance rather than depend on that
+// accident.
+#![recursion_limit = "256"]
+
 pub mod alerts;
 pub(crate) mod connector_tags;
 pub mod controllers;
diff --git a/crates/agent/src/main.rs b/crates/agent/src/main.rs
index 668d6f2eb8b..65a61140ee5 100644
--- a/crates/agent/src/main.rs
+++ b/crates/agent/src/main.rs
@@ -408,7 +408,7 @@ async fn async_main(args: Args) -> Result<(), anyhow::Error> {
     let api_server = async move { anyhow::Result::Ok(api_server.await?) };
 
     let automations_fut = if args.max_automations > 0 {
-        let directive_executor = agent::DirectiveHandler::new(args.accounts_email, &logs_tx);
+        let directive_executor = agent::DirectiveHandler::new(args.accounts_email);
         let connector_tags_executor = agent::TagExecutor::new(&args.connector_network, &logs_tx);
         let mut automations_server = automations::Server::new()
             .register(agent::controllers::LiveSpecControllerExecutor::new(
diff --git a/crates/control-plane-api/src/directives/mod.rs b/crates/control-plane-api/src/directives/mod.rs
index 1ed74a587b9..bb1a272a11e 100644
--- a/crates/control-plane-api/src/directives/mod.rs
+++ b/crates/control-plane-api/src/directives/mod.rs
@@ -1,6 +1,5 @@
 pub mod accept_demo_tenant;
 pub mod beta_onboard;
-pub mod storage_mappings;
 
 use crate::TextJson;
 use chrono::{DateTime, Utc};
diff --git a/crates/control-plane-api/src/lib.rs b/crates/control-plane-api/src/lib.rs
index ca16775c095..f55f6693268 100644
--- a/crates/control-plane-api/src/lib.rs
+++ b/crates/control-plane-api/src/lib.rs
@@ -20,10 +20,13 @@ pub mod logs;
 pub mod proxy_connectors;
 pub mod publications;
 pub mod server;
+pub mod storage_mappings;
 mod text_json;
 
 #[cfg(test)]
 pub(crate) mod test_server;
+#[cfg(test)]
+pub(crate) mod test_support;
 
 /// TextJson encodes JSON for Postgres while preserving property ordering.
 pub use text_json::TextJson;
diff --git a/crates/control-plane-api/src/live_specs/mod.rs b/crates/control-plane-api/src/live_specs/mod.rs
index a7863e2d366..1576cc99ac3 100644
--- a/crates/control-plane-api/src/live_specs/mod.rs
+++ b/crates/control-plane-api/src/live_specs/mod.rs
@@ -48,15 +48,13 @@ pub async fn get_live_specs(
                 continue;
             };
             if let Some(min_capability) = filter_capability {
-                // For discovers, anchor to the discover request time (started_at).
-                // For other callers, anchor to the spec's publication time.
-                // An authoritative denial is today's silent drop; a provisional
-                // one surfaces as a retryable stale error.
-                let anchor = started_at.unwrap_or_else(|| row.last_pub_id.timestamp());
-                if !snapshot
-                    .user_authorization(user_id, &row.catalog_name, min_capability, Some(anchor))
-                    .ok_or_stale(&row.catalog_name)?
-                {
+                if !snapshot.spec_fetch_authorization(
+                    user_id,
+                    &row.catalog_name,
+                    min_capability,
+                    started_at,
+                    row.last_pub_id,
+                )? {
                     continue;
                 }
             }
@@ -161,6 +159,7 @@ pub async fn get_connected_live_specs(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_support::{assert_stale_for, authoritative, stale};
 
     // From `fixtures/authz_specs.sql`. Carol is admin of `carolCo/`; Dan holds no
     // grants at all and so models an unauthorized caller.
@@ -169,52 +168,6 @@ mod tests {
     const COLLECTION: &str = "carolCo/data/foo";
     const CAPTURE: &str = "carolCo/in/capture-foo";
 
-    /// Staleness compares the Snapshot's `taken` against the timestamp embedded
-    /// in a spec's `last_pub_id`, so read that back rather than recomputing it —
-    /// `flowid` is `macaddr8`, which silently widens short literals.
-    async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime {
-        sqlx::query_scalar!(
-            r#"select last_pub_id as "last_pub_id: models::Id"
-            from live_specs where catalog_name = $1"#,
-            COLLECTION,
-        )
-        .fetch_one(pool)
-        .await
-        .expect("fixture collection should exist")
-        .timestamp()
-    }
-
-    /// A Snapshot holding the fixture's real grants, stamped `offset` away from
-    /// the instant the fixture's specs were published.
-    async fn snapshot_offset(pool: &sqlx::PgPool, offset: chrono::TimeDelta) -> crate::Snapshot {
-        let mut decrypted_hmac_keys = std::collections::HashMap::new();
-        let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys)
-            .await
-            .expect("failed to fetch snapshot");
-        crate::Snapshot::new(published_at(pool).await + offset, data)
-    }
-
-    /// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive.
-    async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot {
-        snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await
-    }
-
-    /// Taken before the publication it would judge: denials are retryable.
-    async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot {
-        snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await
-    }
-
-    fn assert_stale_for(err: anyhow::Error, catalog_name: &str) {
-        assert!(
-            validation::is_authz_snapshot_stale(&err),
-            "expected a retryable stale-snapshot error, got: {err:#}"
-        );
-        assert!(
-            err.to_string().contains(catalog_name),
-            "stale error should name the offending spec, got: {err:#}"
-        );
-    }
-
     /// With no capability filter the Snapshot is never consulted, so even a
     /// wholly unauthorized caller reading against a stale Snapshot gets the spec.
     /// This is the path controllers and other system callers take.
diff --git a/crates/control-plane-api/src/publications/db_complete.rs b/crates/control-plane-api/src/publications/db_complete.rs
index a00f19c0d09..e9f92e549c4 100644
--- a/crates/control-plane-api/src/publications/db_complete.rs
+++ b/crates/control-plane-api/src/publications/db_complete.rs
@@ -1,6 +1,6 @@
 use crate::FlowType;
 
-use super::{Capability, CatalogType, Id, TextJson as Json};
+use super::{CatalogType, Id, TextJson as Json};
 
 use chrono::prelude::*;
 use serde::Serialize;
@@ -449,25 +449,6 @@ pub async fn find_tenant_quotas(
     .await
 }
 
-#[derive(Debug)]
-pub struct ExpandedRow {
-    // Name of the specification.
-    pub catalog_name: String,
-    // Last build ID of the live spec.
-    pub last_build_id: Id,
-    // Last publication ID of the live spec.
-    pub last_pub_id: Id,
-    // Current live specification of this expansion.
-    // It won't be changed by this publication.
-    pub live_spec: Json<Box<RawValue>>,
-    // ID of the expanded live specification.
-    pub live_spec_id: Id,
-    // Spec type of the live specification.
-    pub live_type: CatalogType,
-    // User's capability to the specification `catalog_name`.
-    pub user_capability: Option<Capability>,
-}
-
 pub async fn delete_stale_flow(
     live_spec_id: Id,
     catalog_type: CatalogType,
diff --git a/crates/control-plane-api/src/publications/specs.rs b/crates/control-plane-api/src/publications/specs.rs
index 8252cfecf2f..84e932c8529 100644
--- a/crates/control-plane-api/src/publications/specs.rs
+++ b/crates/control-plane-api/src/publications/specs.rs
@@ -1224,6 +1224,7 @@ mod test {
 #[cfg(test)]
 mod resolve_tests {
     use super::*;
+    use crate::test_support::{assert_stale_for, authoritative, published_at, stale};
 
     // From `fixtures/authz_specs.sql`.
     const CAROL: uuid::Uuid = uuid::uuid!("33333333-3333-3333-3333-333333333333");
@@ -1274,39 +1275,6 @@ mod resolve_tests {
         }))
     }
 
-    /// Staleness compares the Snapshot's `taken` against the timestamp embedded
-    /// in a spec's `last_pub_id`, so read that back rather than recomputing it —
-    /// `flowid` is `macaddr8`, which silently widens short literals.
-    async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime {
-        sqlx::query_scalar!(
-            r#"select last_pub_id as "last_pub_id: models::Id"
-            from live_specs where catalog_name = $1"#,
-            COLLECTION,
-        )
-        .fetch_one(pool)
-        .await
-        .expect("fixture collection should exist")
-        .timestamp()
-    }
-
-    async fn snapshot_offset(pool: &sqlx::PgPool, offset: chrono::TimeDelta) -> crate::Snapshot {
-        let mut decrypted_hmac_keys = std::collections::HashMap::new();
-        let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys)
-            .await
-            .expect("failed to fetch snapshot");
-        crate::Snapshot::new(published_at(pool).await + offset, data)
-    }
-
-    /// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive.
-    async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot {
-        snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await
-    }
-
-    /// Taken before the publication it would judge: denials are retryable.
-    async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot {
-        snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await
-    }
-
     /// Renders `live.errors` as `(scope, message)` pairs for snapshot assertions.
     fn error_pairs(live: &tables::LiveCatalog) -> Vec<(String, String)> {
         live.errors
@@ -1315,17 +1283,6 @@ mod resolve_tests {
             .collect()
     }
 
-    fn assert_stale_for(err: anyhow::Error, catalog_name: &str) {
-        assert!(
-            validation::is_authz_snapshot_stale(&err),
-            "expected a retryable stale-snapshot error, got: {err:#}"
-        );
-        assert!(
-            err.to_string().contains(catalog_name),
-            "stale error should name the offending spec, got: {err:#}"
-        );
-    }
-
     /// Branch 1: a user drafting an existing spec must hold `SpecEdit` to it.
     /// Dan does not, but the denial is only definitive once the Snapshot
     /// outlives the spec.
diff --git a/crates/control-plane-api/src/server/create_data_plane.rs b/crates/control-plane-api/src/server/create_data_plane.rs
index cacdd507887..2d5e8c798ba 100644
--- a/crates/control-plane-api/src/server/create_data_plane.rs
+++ b/crates/control-plane-api/src/server/create_data_plane.rs
@@ -1,7 +1,7 @@
-use crate::directives::storage_mappings::{fetch_storage_mappings, upsert_storage_mapping};
 use crate::publications::{
     DoNotRetry, DraftPublication, NoopInitialize, NoopWithCommit, PruneUnboundCollections,
 };
+use crate::storage_mappings::{fetch_storage_mappings, upsert_storage_mapping};
 use anyhow::Context;
 use validator::Validate;
 
@@ -64,19 +64,7 @@ pub async fn create_data_plane(
     }): super::Request<Request>,
 ) -> Result<axum::Json<Response>, crate::server::error::ApiError> {
     let models::authorizations::ControlClaims { sub: user_id, .. } = env.claims()?;
-
-    if let None = sqlx::query!(
-        "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'",
-        user_id,
-    )
-    .fetch_optional(&env.pg_pool)
-    .await?
-    {
-        return Err(tonic::Status::permission_denied(
-            "authenticated user is not an admin of the 'ops/' tenant",
-        )
-        .into());
-    }
+    super::authorize_ops_admin(&env).await?;
 
     let (data_plane_fqdn, base_name, pulumi_stack) = match &private {
         None => (
@@ -232,7 +220,6 @@ pub async fn create_data_plane(
         .unwrap()
         .into();
 
-    let snapshot = app.snapshot_watch.token();
     let publication = DraftPublication {
         user_id: *user_id,
         logs_token: insert.logs_token,
@@ -241,9 +228,7 @@ pub async fn create_data_plane(
         detail: Some(format!("publication for data-plane {base_name}")),
         // A one-shot handler invocation, with no queued row to anchor on.
         started_at: None,
-        snapshot: snapshot
-            .result()
-            .expect("authorization snapshot is not ready"),
+        snapshot: env.snapshot(),
         // We've already validated that the user can admin `ops/`,
         // so further authZ checks are unnecessary.
         verify_user_authz: false,
@@ -332,3 +317,176 @@ impl Validate for Category {
         }
     }
 }
+
+/// The `ops/`-admin pre-check, evaluated against the request's pinned
+/// Snapshot: a caller without `ops/` admin is rejected before any data-plane
+/// state is touched — terminally (403) when the Snapshot postdates the
+/// request, and with the platform-standard 307 retry when it doesn't.
+#[cfg(test)]
+mod test {
+    use crate::test_server;
+
+    // From `fixtures/alice.sql`: admin of `aliceCo/` and nothing else.
+    const ALICE: uuid::Uuid = uuid::uuid!("11111111-1111-1111-1111-111111111111");
+
+    #[sqlx::test(
+        migrations = "../../supabase/migrations",
+        fixtures(path = "../fixtures", scripts("data_planes", "alice"))
+    )]
+    async fn test_create_data_plane_denied_for_non_ops_admin(pool: sqlx::PgPool) {
+        let _guard = test_server::init();
+        let server = test_server::TestServer::start(
+            pool.clone(),
+            test_server::snapshot(pool.clone(), false).await,
+        )
+        .await;
+        let token = server.make_access_token(ALICE, Some("alice@example.com"));
+
+        let response = server
+            .rest_client()
+            .post(
+                "/admin/create-data-plane",
+                &serde_json::json!({"name": "test-plane-c1", "category": "managed"}),
+                Some(&token),
+            )
+            .send()
+            .await
+            .unwrap();
+
+        assert_eq!(reqwest::StatusCode::FORBIDDEN, response.status());
+    }
+
+    /// A denial evaluated against a Snapshot which predates the request is
+    /// provisional: the endpoint answers with the platform-standard 307
+    /// `AuthZRetry` (Retry-After + `started`/`retryAfter` params) rather than
+    /// a terminal 403, and a retry against the refreshed (authoritative)
+    /// Snapshot then resolves the denial terminally.
+    #[sqlx::test(
+        migrations = "../../supabase/migrations",
+        fixtures(path = "../fixtures", scripts("data_planes", "alice"))
+    )]
+    async fn test_create_data_plane_stale_snapshot_retries_then_denies(pool: sqlx::PgPool) {
+        let _guard = test_server::init();
+        // gate=true serves an empty epoch-taken Snapshot first: every denial
+        // under it is provisional. The revoke cancelled by the first request
+        // refreshes the watch to the real (+1h) Snapshot.
+        let server = test_server::TestServer::start(
+            pool.clone(),
+            test_server::snapshot(pool.clone(), true).await,
+        )
+        .await;
+        let token = server.make_access_token(ALICE, Some("alice@example.com"));
+
+        let client = flow_client_next::rest::Client {
+            base_url: server.base_url(),
+            http_client: reqwest::Client::builder()
+                .redirect(reqwest::redirect::Policy::none())
+                .build()
+                .unwrap(),
+        };
+        let body = serde_json::json!({"name": "test-plane-c1", "category": "managed"});
+
+        let response = client
+            .post("/admin/create-data-plane", &body, Some(&token))
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(
+            reqwest::StatusCode::TEMPORARY_REDIRECT,
+            response.status(),
+            "a stale denial must be provisional"
+        );
+        assert!(
+            response
+                .headers()
+                .contains_key(reqwest::header::RETRY_AFTER)
+        );
+        let location = response
+            .headers()
+            .get(reqwest::header::LOCATION)
+            .and_then(|l| l.to_str().ok())
+            .expect("redirect carries a Location");
+        assert!(
+            location.contains("started=") && location.contains("retryAfter="),
+            "Location must carry retry bookkeeping: {location}"
+        );
+
+        // The cancelled revoke triggers a refresh to the authoritative
+        // Snapshot; denials then become terminal. Bound the wait, since the
+        // refresh races this retry loop.
+        for attempt in 0..50 {
+            let response = client
+                .post("/admin/create-data-plane", &body, Some(&token))
+                .send()
+                .await
+                .unwrap();
+            match response.status() {
+                reqwest::StatusCode::FORBIDDEN => return,
+                reqwest::StatusCode::TEMPORARY_REDIRECT => {
+                    tokio::time::sleep(std::time::Duration::from_millis(20 * attempt)).await;
+                }
+                other => panic!("unexpected interim status {other}"),
+            }
+        }
+        panic!("denial never became terminal under the refreshed snapshot");
+    }
+
+    /// The seeded system user — these endpoints' routine caller — holds a
+    /// direct `('ops/', 'admin')` row in `user_grants` (seed.sql); this pins
+    /// that exactly that grant shape resolves through the Snapshot's grant
+    /// walk, and that it doesn't leak into unrelated tenants.
+    #[sqlx::test(
+        migrations = "../../supabase/migrations",
+        fixtures(path = "../fixtures", scripts("data_planes", "alice"))
+    )]
+    async fn test_ops_admin_authorizes_via_snapshot(pool: sqlx::PgPool) {
+        let ops_admin = uuid::uuid!("99999999-9999-9999-9999-999999999999");
+        sqlx::query("insert into auth.users (id, email) values ($1, 'ops-admin@example.com')")
+            .bind(ops_admin)
+            .execute(&pool)
+            .await
+            .unwrap();
+        sqlx::query(
+            "insert into user_grants (user_id, object_role, capability) values ($1, 'ops/', 'admin')",
+        )
+        .bind(ops_admin)
+        .execute(&pool)
+        .await
+        .unwrap();
+
+        let mut decrypted_hmac_keys = std::collections::HashMap::new();
+        let data = crate::snapshot::try_fetch(&pool, &mut decrypted_hmac_keys)
+            .await
+            .expect("failed to fetch snapshot");
+        let snapshot = crate::Snapshot::new(tokens::now(), data);
+
+        let claims = models::authorizations::ControlClaims {
+            iat: 0,
+            exp: u64::MAX,
+            sub: ops_admin,
+            role: "authenticated".to_string(),
+            aud: "authenticated".to_string(),
+            email: Some("ops-admin@example.com".to_string()),
+        };
+        assert!(
+            crate::evaluate_names_authorization(
+                &snapshot,
+                &claims,
+                models::Capability::Admin,
+                ["ops/"],
+            )
+            .is_ok(),
+            "an ops/ admin user_grant must satisfy the snapshot walk"
+        );
+        assert!(
+            crate::evaluate_names_authorization(
+                &snapshot,
+                &claims,
+                models::Capability::Admin,
+                ["aliceCo/"],
+            )
+            .is_err(),
+            "ops/ admin must not leak into unrelated tenants"
+        );
+    }
+}
diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs
index 406393688d2..19e36b9c045 100644
--- a/crates/control-plane-api/src/server/mod.rs
+++ b/crates/control-plane-api/src/server/mod.rs
@@ -134,6 +134,22 @@ where
     Ok((None, ()))
 }
 
+/// Require that the requesting user holds `admin` over the `ops/` tenant,
+/// evaluated against the request's pinned Snapshot — the same Snapshot the
+/// rest of the operation then uses: one request, one authorization view.
+/// A denial is terminal only once the Snapshot postdates the request's start;
+/// otherwise `authorization_outcome` yields the standard retry response.
+pub(crate) async fn authorize_ops_admin(env: &crate::Envelope) -> Result<(), crate::ApiError> {
+    let policy_result = crate::evaluate_names_authorization(
+        env.snapshot(),
+        env.claims()?,
+        models::Capability::Admin,
+        ["ops/"],
+    );
+    let (_expiry, ()) = env.authorization_outcome(policy_result).await?;
+    Ok(())
+}
+
 /// Looks up the user's authorization grants for each item in
 /// `prefixes_or_names`, and calls the provided `attach` function with each
 /// item and its capability. The `Some` results are returned in a vec.
diff --git a/crates/control-plane-api/src/server/public/graphql/storage_mappings.rs b/crates/control-plane-api/src/server/public/graphql/storage_mappings.rs
index 31ecd538405..6ebb7182f63 100644
--- a/crates/control-plane-api/src/server/public/graphql/storage_mappings.rs
+++ b/crates/control-plane-api/src/server/public/graphql/storage_mappings.rs
@@ -1,5 +1,5 @@
 use super::filters;
-use crate::directives::storage_mappings::{
+use crate::storage_mappings::{
     collection_and_recovery_spec_from, insert_storage_mapping, strip_collection_data_suffix,
     update_storage_mapping, upsert_storage_mapping,
 };
diff --git a/crates/control-plane-api/src/server/snapshot.rs b/crates/control-plane-api/src/server/snapshot.rs
index 2281a60ae3e..8d210e69e88 100644
--- a/crates/control-plane-api/src/server/snapshot.rs
+++ b/crates/control-plane-api/src/server/snapshot.rs
@@ -258,6 +258,30 @@ impl Snapshot {
         )
     }
 
+    /// Evaluate whether `user_id` may fetch the live spec `catalog_name` with
+    /// `capability`: the policy applied when fetching specs the caller
+    /// explicitly named (`live_specs::get_live_specs`). Staleness anchors on
+    /// `started` — the fetching operation's durable queued instant — when the
+    /// caller has one, and otherwise on the spec's own last publication, which
+    /// bounds the window in which grants could have been committed alongside
+    /// the spec.
+    ///
+    /// `Ok(false)` is an authoritative denial: callers drop or suppress the
+    /// spec, the pre-existing behavior. A provisional denial surfaces as the
+    /// retryable `AuthorizationSnapshotStale` error instead.
+    pub fn spec_fetch_authorization(
+        &self,
+        user_id: uuid::Uuid,
+        catalog_name: &str,
+        capability: impl Into<models::authz::CapabilitySet>,
+        started: Option<tokens::DateTime>,
+        last_pub_id: models::Id,
+    ) -> Result<bool, validation::Error> {
+        let anchor = started.unwrap_or_else(|| last_pub_id.timestamp());
+        self.user_authorization(user_id, catalog_name, capability, Some(anchor))
+            .ok_or_stale(catalog_name)
+    }
+
     /// Evaluate whether `subject` (a catalog spec acting as a role) holds
     /// `capability` to `object` under this Snapshot's role grants, classified
     /// against `anchor` freshness (see `resolve_authorization`).
diff --git a/crates/control-plane-api/src/server/update_l2_reporting.rs b/crates/control-plane-api/src/server/update_l2_reporting.rs
index f7fc33ec5a0..253455fa6a7 100644
--- a/crates/control-plane-api/src/server/update_l2_reporting.rs
+++ b/crates/control-plane-api/src/server/update_l2_reporting.rs
@@ -33,19 +33,7 @@ pub async fn update_l2_reporting(
     }): super::Request<Request>,
 ) -> Result<axum::Json<Response>, crate::ApiError> {
     let crate::ControlClaims { sub: user_id, .. } = env.claims()?;
-
-    if let None = sqlx::query!(
-        "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'",
-        user_id,
-    )
-    .fetch_optional(&env.pg_pool)
-    .await?
-    {
-        return Err(tonic::Status::permission_denied(
-            "authenticated user is not an admin of the 'ops/' tenant",
-        )
-        .into());
-    }
+    super::authorize_ops_admin(&env).await?;
 
     let template = include_str!("../../../../ops-catalog/reporting-L2-template.bundle.json");
     let tables::DraftCatalog { collections, .. } =
@@ -295,7 +283,6 @@ export class Derivation extends Types.IDerivation {"#
     };
 
     let logs_token = uuid::Uuid::new_v4();
-    let snapshot = app.snapshot_watch.token();
     let publication = DraftPublication {
         user_id: *user_id,
         logs_token,
@@ -304,9 +291,7 @@ export class Derivation extends Types.IDerivation {"#
         detail: Some(format!("publication for updating L2 reporting")),
         // A one-shot handler invocation, with no queued row to anchor on.
         started_at: None,
-        snapshot: snapshot
-            .result()
-            .expect("authorization snapshot is not ready"),
+        snapshot: env.snapshot(),
         default_data_plane_name: if default_data_plane.trim().is_empty() {
             None
         } else {
@@ -352,6 +337,42 @@ export class Derivation extends Types.IDerivation {"#
     }))
 }
 
+/// See `create_data_plane::test` — same pre-check, same contract: a caller
+/// without `ops/` admin is rejected with 403 before any template work.
+#[cfg(test)]
+mod test {
+    use crate::test_server;
+
+    const ALICE: uuid::Uuid = uuid::uuid!("11111111-1111-1111-1111-111111111111");
+
+    #[sqlx::test(
+        migrations = "../../supabase/migrations",
+        fixtures(path = "../fixtures", scripts("data_planes", "alice"))
+    )]
+    async fn test_update_l2_reporting_denied_for_non_ops_admin(pool: sqlx::PgPool) {
+        let _guard = test_server::init();
+        let server = test_server::TestServer::start(
+            pool.clone(),
+            test_server::snapshot(pool.clone(), false).await,
+        )
+        .await;
+        let token = server.make_access_token(ALICE, Some("alice@example.com"));
+
+        let response = server
+            .rest_client()
+            .post(
+                "/admin/update-l2-reporting",
+                &serde_json::json!({"defaultDataPlane": "", "dryRun": true}),
+                Some(&token),
+            )
+            .send()
+            .await
+            .unwrap();
+
+        assert_eq!(reqwest::StatusCode::FORBIDDEN, response.status());
+    }
+}
+
 // Copied from crates/derive-typescript/src/codegen/mod.rs
 fn camel_case(name: &str, mut upper: bool) -> String {
     let mut w = String::new();
diff --git a/crates/control-plane-api/src/directives/storage_mappings.rs b/crates/control-plane-api/src/storage_mappings.rs
similarity index 96%
rename from crates/control-plane-api/src/directives/storage_mappings.rs
rename to crates/control-plane-api/src/storage_mappings.rs
index 05ea027fdd2..f4887965cf7 100644
--- a/crates/control-plane-api/src/directives/storage_mappings.rs
+++ b/crates/control-plane-api/src/storage_mappings.rs
@@ -1,21 +1,12 @@
+//! Storage-mapping persistence helpers and the `collection-data/` prefix
+//! conventions: the shared write path for `storage_mappings` rows, used by
+//! the GraphQL storage-mapping resolvers and `create_data_plane`. Formerly
+//! under `directives/`, until the GraphQL mutations superseded the
+//! `storageMappings` applied-directive (2026-03-16) and the directive was
+//! removed.
+
 use crate::TextJson;
 use serde_json::value::RawValue;
-use sqlx::types::Uuid;
-
-pub async fn user_has_admin_capability(
-    user_id: Uuid,
-    catalog_prefix: &str,
-    txn: &mut sqlx::Transaction<'_, sqlx::Postgres>,
-) -> sqlx::Result<bool> {
-    let row = sqlx::query!(
-        r#"select true as whatever_column from internal.user_roles($1, 'admin') where starts_with(role_prefix, $2)"#,
-        user_id,
-        catalog_prefix,
-    )
-    .fetch_optional(&mut **txn)
-    .await?;
-    Ok(row.is_some())
-}
 
 pub async fn upsert_storage_mapping<T: serde::Serialize + Send + Sync>(
     detail: Option<&str>,
diff --git a/crates/control-plane-api/src/test_support.rs b/crates/control-plane-api/src/test_support.rs
new file mode 100644
index 00000000000..6358df779ad
--- /dev/null
+++ b/crates/control-plane-api/src/test_support.rs
@@ -0,0 +1,56 @@
+//! Shared helpers for authorization tests over the `authz_specs` fixture:
+//! Snapshots of the database's real grants, stamped relative to the fixture's
+//! publication instant so each test chooses whether a denial is authoritative
+//! or provisional.
+
+/// The `authz_specs.sql` collection whose publication anchors staleness.
+const ANCHOR_COLLECTION: &str = "carolCo/data/foo";
+
+/// Staleness compares the Snapshot's `taken` against the timestamp embedded
+/// in a spec's `last_pub_id`, so read that back rather than recomputing it —
+/// `flowid` is `macaddr8`, which silently widens short literals.
+pub(crate) async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime {
+    sqlx::query_scalar::<_, models::Id>(
+        "select last_pub_id from live_specs where catalog_name = $1",
+    )
+    .bind(ANCHOR_COLLECTION)
+    .fetch_one(pool)
+    .await
+    .expect("fixture collection should exist")
+    .timestamp()
+}
+
+/// A Snapshot holding the fixture's real grants, stamped `offset` away from
+/// the instant the fixture's specs were published.
+pub(crate) async fn snapshot_offset(
+    pool: &sqlx::PgPool,
+    offset: chrono::TimeDelta,
+) -> crate::Snapshot {
+    let mut decrypted_hmac_keys = std::collections::HashMap::new();
+    let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys)
+        .await
+        .expect("failed to fetch snapshot");
+    crate::Snapshot::new(published_at(pool).await + offset, data)
+}
+
+/// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive.
+pub(crate) async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot {
+    snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await
+}
+
+/// Taken before the publication it would judge: denials are retryable.
+pub(crate) async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot {
+    snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await
+}
+
+/// Asserts `err` is the retryable stale-snapshot error and names the spec.
+pub(crate) fn assert_stale_for(err: anyhow::Error, catalog_name: &str) {
+    assert!(
+        validation::is_authz_snapshot_stale(&err),
+        "expected a retryable stale-snapshot error, got: {err:#}"
+    );
+    assert!(
+        err.to_string().contains(catalog_name),
+        "stale error should name the offending spec, got: {err:#}"
+    );
+}
diff --git a/local/systemd/flow-plane-link@.service b/local/systemd/flow-plane-link@.service
index 8cc19f12821..802d70b6a3d 100644
--- a/local/systemd/flow-plane-link@.service
+++ b/local/systemd/flow-plane-link@.service
@@ -20,6 +20,11 @@ EnvironmentFile=%h/flow-local/env/plane-link-%i.env
 # during warm-up, bounded to 120s of 2s-spaced attempts. The agent binds its API
 # port only once it can serve (crates/agent/src/main.rs), so a connection being
 # accepted means ready; a non-transient 4xx still fails fast with its body.
+#
+# The -L flag follows the 307 AuthZRetry the /admin endpoints answer with when
+# an authorization denial is evaluated against a not-yet-authoritative
+# snapshot (307 preserves the POST and body; the server paces early retries
+# itself). Without it curl treats the redirect as success and silently no-ops.
 ExecStart=/bin/bash -c ' \
     if [ -n "${DEKAF_ADDRESS:-}" ] && [ -n "${DEKAF_REGISTRY_ADDRESS:-}" ]; then \
         printf "{\\"name\\":\\"%%s\\",\\"category\\":{\\"manual\\":{\\"brokerAddress\\":\\"%%s\\",\\"reactorAddress\\":\\"%%s\\",\\"hmacKeys\\":[\\"%%s\\"],\\"dekafAddress\\":\\"%%s\\",\\"dekafRegistryAddress\\":\\"%%s\\"}}}" \
@@ -32,6 +37,8 @@ ExecStart=/bin/bash -c ' \
         -X POST \
         -H "content-type: application/json" \
         -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \
+        -L \
+        --max-redirs 10 \
         --retry 60 \
         --retry-connrefused \
         --retry-delay 2 \
@@ -49,6 +56,8 @@ ExecStart=/bin/bash -c ' \
         -X POST \
         -H "content-type: application/json" \
         -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \
+        -L \
+        --max-redirs 10 \
         --retry 60 \
         --retry-connrefused \
         --retry-delay 2 \
@@ -76,6 +85,8 @@ ExecStop=/bin/bash -c ' \
         -X POST \
         -H "content-type: application/json" \
         -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \
+        -L \
+        --max-redirs 10 \
         --fail-with-body \
         --data-binary @- \
         ${AGENT_API}/admin/update-l2-reporting \