diff --git a/.sqlx/query-6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce.json b/.sqlx/query-6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce.json
new file mode 100644
index 00000000000..9246b22b475
--- /dev/null
+++ b/.sqlx/query-6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce.json
@@ -0,0 +1,71 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        with drafted as (\n            select\n                ds.catalog_name,\n                ds.id as draft_spec_id,\n                ls.id as live_spec_id,\n                ds.expect_pub_id,\n                ls.last_pub_id as last_pub_id,\n                ds.spec as spec,\n                ds.spec_type as spec_type\n            from draft_specs ds\n            left join live_specs ls\n                on ds.catalog_name = ls.catalog_name\n            where ds.draft_id = $1\n        ),\n        not_drafted as (\n            select catalog_name from unnest($2::text[]) as names(catalog_name)\n            except\n            select catalog_name from drafted\n        ),\n        live as (\n            select\n                ls.catalog_name,\n                ls.spec,\n                ls.spec_type,\n                ls.last_pub_id,\n                ls.id\n            from not_drafted\n            join live_specs ls on not_drafted.catalog_name = ls.catalog_name\n        )\n        select\n            catalog_name as \"catalog_name!: String\",\n            draft_spec_id as \"draft_spec_id: Id\",\n            live_spec_id as \"live_spec_id: Id\",\n            expect_pub_id as \"expect_pub_id: Id\",\n            last_pub_id as \"last_pub_id: Id\",\n            spec as \"spec: Json<Box<RawValue>>\",\n            spec_type as \"spec_type: CatalogType\"\n        from drafted\n        union all\n        select\n            catalog_name as \"catalog_name!: String\",\n            null as \"draft_spec_id: Id\",\n            id as \"live_spec_id: Id\",\n            null as \"expect_pub_id: Id\",\n            last_pub_id as \"last_pub_id: Id\",\n            spec as \"spec: Json<Box<RawValue>>\",\n            spec_type as \"spec_type: CatalogType\"\n        from live\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "catalog_name!: String",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "draft_spec_id: Id",
+        "type_info": "Macaddr8"
+      },
+      {
+        "ordinal": 2,
+        "name": "live_spec_id: Id",
+        "type_info": "Macaddr8"
+      },
+      {
+        "ordinal": 3,
+        "name": "expect_pub_id: Id",
+        "type_info": "Macaddr8"
+      },
+      {
+        "ordinal": 4,
+        "name": "last_pub_id: Id",
+        "type_info": "Macaddr8"
+      },
+      {
+        "ordinal": 5,
+        "name": "spec: Json<Box<RawValue>>",
+        "type_info": "Json"
+      },
+      {
+        "ordinal": 6,
+        "name": "spec_type: CatalogType",
+        "type_info": {
+          "Custom": {
+            "name": "catalog_spec_type",
+            "kind": {
+              "Enum": [
+                "capture",
+                "collection",
+                "materialization",
+                "test"
+              ]
+            }
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Macaddr8",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      null,
+      null,
+      null,
+      null,
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce"
+}
diff --git a/.sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json b/.sqlx/query-74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1.json
similarity index 60%
rename from .sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json
rename to .sqlx/query-74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1.json
index c1ba203be75..88031d307dd 100644
--- a/.sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json
+++ b/.sqlx/query-74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1.json
@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n        with collections(id) as (\n            select ls.id\n            from unnest($2::text[]) as names(catalog_name)\n            join live_specs ls on ls.catalog_name = names.catalog_name\n        ),\n        exp(id) as (\n            select lsf.source_id as id\n            from collections c\n            join live_spec_flows lsf on c.id = lsf.target_id\n            union\n            select lsf.target_id as id\n            from collections c\n            join live_spec_flows lsf on c.id = lsf.source_id\n        )\n        select\n            ls.id as \"id: Id\",\n            ls.last_pub_id as \"last_pub_id: Id\",\n            ls.last_build_id as \"last_build_id: Id\",\n            ls.data_plane_id as \"data_plane_id: Id\",\n            ls.catalog_name,\n            ls.spec_type as \"spec_type?: CatalogType\",\n            ls.spec as \"spec: TextJson<Box<RawValue>>\",\n            ls.built_spec as \"built_spec: TextJson<Box<RawValue>>\",\n            ls.inferred_schema_md5,\n            (\n                select max(capability) from internal.user_roles($1) r\n                where starts_with(ls.catalog_name, r.role_prefix)\n            ) as \"user_capability: Capability\",\n            coalesce(\n                (select json_agg(row_to_json(role_grants))\n                from role_grants\n                where starts_with(ls.catalog_name, subject_role)),\n                '[]'\n            ) as \"spec_capabilities!: Json<Vec<RoleGrant>>\",\n            ls.dependency_hash\n        from exp\n        join live_specs ls on ls.id = exp.id\n        where ls.spec is not null and not ls.catalog_name = any($3);\n        ",
+  "query": "\n        with collections(id) as (\n            select ls.id\n            from unnest($1::text[]) as names(catalog_name)\n            join live_specs ls on ls.catalog_name = names.catalog_name\n        ),\n        exp(id) as (\n            select lsf.source_id as id\n            from collections c\n            join live_spec_flows lsf on c.id = lsf.target_id\n            union\n            select lsf.target_id as id\n            from collections c\n            join live_spec_flows lsf on c.id = lsf.source_id\n        )\n        select\n            ls.id as \"id: Id\",\n            ls.last_pub_id as \"last_pub_id: Id\",\n            ls.last_build_id as \"last_build_id: Id\",\n            ls.data_plane_id as \"data_plane_id: Id\",\n            ls.catalog_name,\n            ls.spec_type as \"spec_type?: CatalogType\",\n            ls.spec as \"spec: TextJson<Box<RawValue>>\",\n            ls.built_spec as \"built_spec: TextJson<Box<RawValue>>\",\n            ls.inferred_schema_md5,\n            -- `user_capability` and `spec_capabilities` are synthesized from the\n            -- authorization Snapshot below rather than queried here. The `null`\n            -- (→ None) and empty-array placeholders are overwritten afterwards.\n            null as \"user_capability: Capability\",\n            '[]' as \"spec_capabilities!: Json<Vec<RoleGrant>>\",\n            ls.dependency_hash\n        from exp\n        join live_specs ls on ls.id = exp.id\n        where ls.spec is not null and not ls.catalog_name = any($2);\n        ",
   "describe": {
     "columns": [
       {
@@ -63,51 +63,12 @@
       {
         "ordinal": 9,
         "name": "user_capability: Capability",
-        "type_info": {
-          "Custom": {
-            "name": "grant_capability",
-            "kind": {
-              "Enum": [
-                "none",
-                "x_01",
-                "x_02",
-                "x_03",
-                "x_04",
-                "x_05",
-                "x_06",
-                "x_07",
-                "x_08",
-                "x_09",
-                "read",
-                "x_11",
-                "x_12",
-                "x_13",
-                "x_14",
-                "x_15",
-                "x_16",
-                "x_17",
-                "x_18",
-                "x_19",
-                "write",
-                "x_21",
-                "x_22",
-                "x_23",
-                "x_24",
-                "x_25",
-                "x_26",
-                "x_27",
-                "x_28",
-                "x_29",
-                "admin"
-              ]
-            }
-          }
-        }
+        "type_info": "Text"
       },
       {
         "ordinal": 10,
         "name": "spec_capabilities!: Json<Vec<RoleGrant>>",
-        "type_info": "Json"
+        "type_info": "Text"
       },
       {
         "ordinal": 11,
@@ -117,7 +78,6 @@
     ],
     "parameters": {
       "Left": [
-        "Uuid",
         "TextArray",
         "TextArray"
       ]
@@ -137,5 +97,5 @@
       true
     ]
   },
-  "hash": "6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343"
+  "hash": "74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1"
 }
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-bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671.json b/.sqlx/query-bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671.json
deleted file mode 100644
index 461e13dbfe5..00000000000
--- a/.sqlx/query-bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671.json
+++ /dev/null
@@ -1,72 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n        with drafted as (\n            select\n                ds.catalog_name,\n                ds.id as draft_spec_id,\n                ls.id as live_spec_id,\n                ds.expect_pub_id,\n                ls.last_pub_id as last_pub_id,\n                ds.spec as spec,\n                ds.spec_type as spec_type\n            from draft_specs ds\n            left join live_specs ls\n                on ds.catalog_name = ls.catalog_name\n                -- filter out live_specs rows that the user does not have admin access to\n                and exists (select 1 from internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix)\n            where ds.draft_id = $1\n        ),\n        not_drafted as (\n            select catalog_name from unnest($3::text[]) as names(catalog_name)\n            except\n            select catalog_name from drafted\n        ),\n        live as (\n            select\n                ls.catalog_name,\n                ls.spec,\n                ls.spec_type,\n                ls.last_pub_id,\n                ls.id\n            from not_drafted\n            join live_specs ls on not_drafted.catalog_name = ls.catalog_name\n            where\n                -- filter out live_specs rows that the user does not have admin access to\n                exists (select 1 from internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix)\n        )\n        select\n            catalog_name as \"catalog_name!: String\",\n            draft_spec_id as \"draft_spec_id: Id\",\n            live_spec_id as \"live_spec_id: Id\",\n            expect_pub_id as \"expect_pub_id: Id\",\n            last_pub_id as \"last_pub_id: Id\",\n            spec as \"spec: Json<Box<RawValue>>\",\n            spec_type as \"spec_type: CatalogType\"\n        from drafted\n        union all\n        select\n            catalog_name as \"catalog_name!: String\",\n            null as \"draft_spec_id: Id\",\n            id as \"live_spec_id: Id\",\n            null as \"expect_pub_id: Id\",\n            last_pub_id as \"last_pub_id: Id\",\n            spec as \"spec: Json<Box<RawValue>>\",\n            spec_type as \"spec_type: CatalogType\"\n        from live\n        ",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "catalog_name!: String",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 1,
-        "name": "draft_spec_id: Id",
-        "type_info": "Macaddr8"
-      },
-      {
-        "ordinal": 2,
-        "name": "live_spec_id: Id",
-        "type_info": "Macaddr8"
-      },
-      {
-        "ordinal": 3,
-        "name": "expect_pub_id: Id",
-        "type_info": "Macaddr8"
-      },
-      {
-        "ordinal": 4,
-        "name": "last_pub_id: Id",
-        "type_info": "Macaddr8"
-      },
-      {
-        "ordinal": 5,
-        "name": "spec: Json<Box<RawValue>>",
-        "type_info": "Json"
-      },
-      {
-        "ordinal": 6,
-        "name": "spec_type: CatalogType",
-        "type_info": {
-          "Custom": {
-            "name": "catalog_spec_type",
-            "kind": {
-              "Enum": [
-                "capture",
-                "collection",
-                "materialization",
-                "test"
-              ]
-            }
-          }
-        }
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Macaddr8",
-        "Uuid",
-        "TextArray"
-      ]
-    },
-    "nullable": [
-      null,
-      null,
-      null,
-      null,
-      null,
-      null,
-      null
-    ]
-  },
-  "hash": "bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671"
-}
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..1120a3d86cc 100644
--- a/crates/agent/src/directives/mod.rs
+++ b/crates/agent/src/directives/mod.rs
@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use control_plane_api::{
     directives::{Row, fetch_directive, resolve},
     logs,
@@ -49,13 +51,21 @@ pub enum Directive {
 pub struct DirectiveHandler {
     accounts_user_email: String,
     logs_tx: logs::Tx,
+    /// Live authorization Snapshot watch, retained so tests can force it to
+    /// re-fetch from Postgres after mutating grants. See `refresh_snapshot`.
+    snapshot_watch: Arc<dyn tokens::Watch<control_plane_api::Snapshot>>,
 }
 
 impl DirectiveHandler {
-    pub fn new(accounts_user_email: String, logs_tx: &logs::Tx) -> Self {
+    pub fn new(
+        accounts_user_email: String,
+        logs_tx: &logs::Tx,
+        snapshot_watch: Arc<dyn tokens::Watch<control_plane_api::Snapshot>>,
+    ) -> Self {
         Self {
             accounts_user_email,
             logs_tx: logs_tx.clone(),
+            snapshot_watch,
         }
     }
 }
@@ -133,7 +143,7 @@ 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?
+                storage_mappings::apply(d, row, &self.logs_tx, txn, &self.snapshot_watch).await?
             }
         };
         Ok(status)
diff --git a/crates/agent/src/directives/storage_mappings.rs b/crates/agent/src/directives/storage_mappings.rs
index aece82ea866..32d52479c0f 100644
--- a/crates/agent/src/directives/storage_mappings.rs
+++ b/crates/agent/src/directives/storage_mappings.rs
@@ -1,14 +1,11 @@
-use std::io::Write;
+use std::{io::Write, sync::Arc};
 
 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,
-        },
+        storage_mappings::{StorageMapping, fetch_storage_mappings, upsert_storage_mapping},
     },
     jobs, logs,
 };
@@ -33,12 +30,13 @@ pub async fn apply(
     row: Row,
     logs_tx: &logs::Tx,
     txn: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    snapshot_watch: &Arc<dyn tokens::Watch<control_plane_api::Snapshot>>,
 ) -> 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 {
+    let (collection_data, recovery) = match validate(txn, logs_tx, row, &snapshot_watch).await {
         Ok(c) => c,
         Err(err) => {
             return Ok(JobStatus::invalid_claims(err));
@@ -75,6 +73,7 @@ async fn validate(
     txn: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     logs_tx: &logs::Tx,
     row: Row,
+    snapshot_watch: &Arc<dyn tokens::Watch<control_plane_api::Snapshot>>,
 ) -> anyhow::Result<(ProposedMapping, ProposedMapping)> {
     let claims: Claims =
         serde_json::from_str(row.user_claims.get()).context("parsing user_claims")?;
@@ -94,8 +93,11 @@ async fn validate(
     // 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 refresh = snapshot_watch.token();
+    let snapshot = refresh.result().unwrap();
+
     let user_has_admin =
-        user_has_admin_capability(row.user_id, &claims.catalog_prefix, txn).await?;
+        user_can_admin_tenant(&snapshot, row.user_id, claims.catalog_prefix.as_str());
     anyhow::ensure!(
         user_has_admin,
         "user does not have required 'admin' capability to '{}'",
@@ -128,6 +130,28 @@ async fn validate(
     Ok((collection_data, recovery_data))
 }
 
+/// Returns whether `user_id` holds `admin` capability on `catalog_prefix`
+/// itself, or on any prefix nested beneath it, according to `snapshot`.
+///
+/// This deliberately mirrors the legacy
+/// `internal.user_roles($user, 'admin') where starts_with(role_prefix, catalog_prefix)`
+/// SQL check that it replaced: a user who administers only a *sub-prefix* of
+/// the tenant (e.g. `acmeCo/team/` when altering `acmeCo/`) remains authorized.
+/// Note this is the reverse of `tables::UserGrant::is_authorized`, which
+/// considers only grants at or *above* `catalog_prefix`.
+fn user_can_admin_tenant(
+    snapshot: &control_plane_api::Snapshot,
+    user_id: Uuid,
+    catalog_prefix: &str,
+) -> bool {
+    snapshot
+        .prefix_and_capabilities_per_user(user_id)
+        .iter()
+        .any(|(role_prefix, capabilities)| {
+            role_prefix.starts_with(catalog_prefix) && capabilities.1 >= models::Capability::Admin
+        })
+}
+
 // 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 {
@@ -323,3 +347,127 @@ This file is written to your storage bucket in order to test that we have the ne
 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."#;
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use control_plane_api::snapshot::SnapshotData;
+
+    // Tenant prefix under test. Storage mappings may only be altered for a
+    // top-level tenant, so this is always a single-segment prefix.
+    const TENANT: &str = "acmeCo/";
+
+    fn user_grant(
+        user_id: Uuid,
+        object_role: &str,
+        capability: models::Capability,
+    ) -> tables::UserGrant {
+        tables::UserGrant {
+            user_id,
+            object_role: models::Prefix::new(object_role),
+            capability,
+            bundles: Vec::new(),
+        }
+    }
+
+    fn role_grant(
+        subject_role: &str,
+        object_role: &str,
+        capability: models::Capability,
+    ) -> tables::RoleGrant {
+        tables::RoleGrant {
+            subject_role: models::Prefix::new(subject_role),
+            object_role: models::Prefix::new(object_role),
+            capability,
+            bundles: Vec::new(),
+        }
+    }
+
+    fn snapshot(
+        user_grants: Vec<tables::UserGrant>,
+        role_grants: Vec<tables::RoleGrant>,
+    ) -> control_plane_api::Snapshot {
+        let data = SnapshotData {
+            collections: Vec::new(),
+            data_planes: Vec::new(),
+            migrations: Vec::new(),
+            role_grants,
+            user_grants,
+            tasks: Vec::new(),
+        };
+        control_plane_api::Snapshot::new(chrono::DateTime::UNIX_EPOCH, data)
+    }
+
+    // Case 1: a user granted `admin` directly on the tenant root is authorized.
+    // This is the case where the old SQL check and `is_authorized` agreed.
+    #[test]
+    fn admin_on_tenant_root_is_authorized() {
+        let user_id = Uuid::from_bytes([1; 16]);
+        let snapshot = snapshot(
+            vec![user_grant(user_id, TENANT, models::Capability::Admin)],
+            Vec::new(),
+        );
+        assert!(user_can_admin_tenant(&snapshot, user_id, TENANT));
+    }
+
+    // Case 2: a user granted `admin` only on a prefix *beneath* the tenant root
+    // is still authorized. This is the behavior that would have been lost with a
+    // plain `is_authorized` check, and which the sub-prefix scan preserves.
+    #[test]
+    fn admin_on_sub_prefix_is_authorized() {
+        let user_id = Uuid::from_bytes([2; 16]);
+        let snapshot = snapshot(
+            vec![user_grant(
+                user_id,
+                "acmeCo/team/",
+                models::Capability::Admin,
+            )],
+            Vec::new(),
+        );
+        assert!(user_can_admin_tenant(&snapshot, user_id, TENANT));
+    }
+
+    // Case 3: a user reaches `admin` on the tenant transitively, through a role
+    // grant (e.g. an `estuary_support/` role that is itself granted admin over
+    // the tenant). The grant-graph projection must be honored.
+    #[test]
+    fn admin_via_role_grant_projection_is_authorized() {
+        let user_id = Uuid::from_bytes([3; 16]);
+        let snapshot = snapshot(
+            vec![user_grant(
+                user_id,
+                "estuary_support/",
+                models::Capability::Admin,
+            )],
+            vec![role_grant(
+                "estuary_support/",
+                TENANT,
+                models::Capability::Admin,
+            )],
+        );
+        assert!(user_can_admin_tenant(&snapshot, user_id, TENANT));
+    }
+
+    // Negative control: a capability below `admin` on the tenant is not enough.
+    #[test]
+    fn write_capability_is_denied() {
+        let user_id = Uuid::from_bytes([4; 16]);
+        let snapshot = snapshot(
+            vec![user_grant(user_id, TENANT, models::Capability::Write)],
+            Vec::new(),
+        );
+        assert!(!user_can_admin_tenant(&snapshot, user_id, TENANT));
+    }
+
+    // Negative control: admin over a sibling tenant confers no authority here,
+    // proving the sub-prefix scan does not match unrelated or ancestor prefixes.
+    #[test]
+    fn admin_on_unrelated_tenant_is_denied() {
+        let user_id = Uuid::from_bytes([5; 16]);
+        let snapshot = snapshot(
+            vec![user_grant(user_id, "bobCo/", models::Capability::Admin)],
+            Vec::new(),
+        );
+        assert!(!user_can_admin_tenant(&snapshot, user_id, TENANT));
+    }
+}
diff --git a/crates/agent/src/integration_tests/.tenant_alerts.rs.pending-snap b/crates/agent/src/integration_tests/.tenant_alerts.rs.pending-snap
index 26d6e544f1f..566be0fd902 100644
--- a/crates/agent/src/integration_tests/.tenant_alerts.rs.pending-snap
+++ b/crates/agent/src/integration_tests/.tenant_alerts.rs.pending-snap
@@ -14,3 +14,13 @@
 {"run_id":"1784726685-655096928","line":91,"new":null,"old":null}
 {"run_id":"1784726685-655096928","line":137,"new":null,"old":null}
 {"run_id":"1784726685-655096928","line":181,"new":null,"old":null}
+{"run_id":"1784732770-555533510","line":19,"new":null,"old":null}
+{"run_id":"1784732770-555533510","line":51,"new":null,"old":null}
+{"run_id":"1784732770-555533510","line":91,"new":null,"old":null}
+{"run_id":"1784732770-555533510","line":137,"new":null,"old":null}
+{"run_id":"1784732770-555533510","line":181,"new":null,"old":null}
+{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":19,"new":null,"old":null}
+{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":51,"new":null,"old":null}
+{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":91,"new":null,"old":null}
+{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":137,"new":null,"old":null}
+{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":181,"new":null,"old":null}
diff --git a/crates/agent/src/integration_tests/.user_publications.rs.pending-snap b/crates/agent/src/integration_tests/.user_publications.rs.pending-snap
index adc0e6ab600..18e519929ed 100644
--- a/crates/agent/src/integration_tests/.user_publications.rs.pending-snap
+++ b/crates/agent/src/integration_tests/.user_publications.rs.pending-snap
@@ -5,3 +5,11 @@
 {"run_id":"1784726664-40620475","line":167,"new":null,"old":null}
 {"run_id":"1784726685-655096928","line":142,"new":null,"old":null}
 {"run_id":"1784726685-655096928","line":167,"new":null,"old":null}
+{"run_id":"1784732711-231719153","line":142,"new":null,"old":null}
+{"run_id":"1784732711-231719153","line":167,"new":null,"old":null}
+{"run_id":"1784732736-313546420","line":142,"new":null,"old":null}
+{"run_id":"1784732736-313546420","line":167,"new":null,"old":null}
+{"run_id":"1784732770-555533510","line":142,"new":null,"old":null}
+{"run_id":"1784732770-555533510","line":167,"new":null,"old":null}
+{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":142,"new":null,"old":null}
+{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":167,"new":null,"old":null}
diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs
index 850f18fdac0..5a1530aefef 100644
--- a/crates/agent/src/integration_tests/harness.rs
+++ b/crates/agent/src/integration_tests/harness.rs
@@ -292,8 +292,11 @@ 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);
+        let directive_exec = crate::directives::DirectiveHandler::new(
+            "support@estuary.test".to_string(),
+            &logs_tx,
+            snapshot_watch.clone(),
+        );
 
         let mut harness = TestHarness {
             test_name,
diff --git a/crates/agent/src/main.rs b/crates/agent/src/main.rs
index b7731758498..5c3fef8d8c0 100644
--- a/crates/agent/src/main.rs
+++ b/crates/agent/src/main.rs
@@ -395,7 +395,8 @@ 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, &logs_tx, snapshot_watch.clone());
         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/storage_mappings.rs b/crates/control-plane-api/src/directives/storage_mappings.rs
index 74c1fa72f8f..7106e4ce21a 100644
--- a/crates/control-plane-api/src/directives/storage_mappings.rs
+++ b/crates/control-plane-api/src/directives/storage_mappings.rs
@@ -1,21 +1,5 @@
 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/evolutions/db.rs b/crates/control-plane-api/src/evolutions/db.rs
index 5a641d6f5b9..dc8b447161b 100644
--- a/crates/control-plane-api/src/evolutions/db.rs
+++ b/crates/control-plane-api/src/evolutions/db.rs
@@ -1,6 +1,6 @@
 use crate::TextJson as Json;
 use chrono::{DateTime, Utc};
-use models::{CatalogType, Id};
+use models::{Capability, CatalogType, Id};
 use serde::Serialize;
 use serde_json::value::RawValue;
 use sqlx::types::Uuid;
@@ -79,12 +79,13 @@ pub struct SpecRow {
 /// bindings. Technically, we could implement that filtering as part of the sql
 /// query, but the extra complexity doesn't seem warranted at this time.
 pub async fn resolve_specs(
-    user_id: Uuid,
     draft_id: Id,
     collection_names: Vec<String>,
+    user_id: uuid::Uuid,
+    snapshot: &crate::Snapshot,
     txn: &mut sqlx::Transaction<'_, sqlx::Postgres>,
 ) -> sqlx::Result<Vec<SpecRow>> {
-    sqlx::query_as!(
+    let mut rows = sqlx::query_as!(
         SpecRow,
         r#"
         with drafted as (
@@ -99,12 +100,10 @@ pub async fn resolve_specs(
             from draft_specs ds
             left join live_specs ls
                 on ds.catalog_name = ls.catalog_name
-                -- filter out live_specs rows that the user does not have admin access to
-                and exists (select 1 from internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix)
             where ds.draft_id = $1
         ),
         not_drafted as (
-            select catalog_name from unnest($3::text[]) as names(catalog_name)
+            select catalog_name from unnest($2::text[]) as names(catalog_name)
             except
             select catalog_name from drafted
         ),
@@ -117,9 +116,6 @@ pub async fn resolve_specs(
                 ls.id
             from not_drafted
             join live_specs ls on not_drafted.catalog_name = ls.catalog_name
-            where
-                -- filter out live_specs rows that the user does not have admin access to
-                exists (select 1 from internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix)
         )
         select
             catalog_name as "catalog_name!: String",
@@ -142,10 +138,38 @@ pub async fn resolve_specs(
         from live
         "#,
         draft_id as Id,
-        user_id as Uuid,
         collection_names as Vec<String>,
-    ).fetch_all(&mut **txn)
-    .await
+    )
+    .fetch_all(&mut **txn)
+    .await?;
+
+    // The admin filter previously applied in SQL (an `^@` prefix match against
+    // the user's `>= Admin` role prefixes) is now applied here from the
+    // Snapshot. It has two distinct effects, both preserved:
+    //   * a drafted spec is always retained, but its live-spec linkage
+    //     (`live_spec_id`/`last_pub_id`) is cleared when the user lacks admin —
+    //     mirroring the previous conditional LEFT JOIN; while
+    //   * a live-only (not-drafted) spec is dropped entirely without admin.
+    let reachable = snapshot.prefix_and_capabilities_per_user(user_id);
+    let user_can_admin = |catalog_name: &str| {
+        reachable.iter().any(|(prefix, (_, legacy))| {
+            *legacy >= Capability::Admin && catalog_name.starts_with(*prefix)
+        })
+    };
+
+    rows.retain_mut(|row| {
+        if row.draft_spec_id.is_some() {
+            if !user_can_admin(&row.catalog_name) {
+                row.live_spec_id = None;
+                row.last_pub_id = None;
+            }
+            true
+        } else {
+            user_can_admin(&row.catalog_name)
+        }
+    });
+
+    Ok(rows)
 }
 
 pub async fn fetch_resource_spec_schema(
diff --git a/crates/control-plane-api/src/evolutions/mod.rs b/crates/control-plane-api/src/evolutions/mod.rs
index 42a35c2d075..2b4e1bc9a17 100644
--- a/crates/control-plane-api/src/evolutions/mod.rs
+++ b/crates/control-plane-api/src/evolutions/mod.rs
@@ -190,6 +190,7 @@ pub async fn evolve(
         &exclude_names,
         capability_filter,
         db,
+        &snapshot,
     )
     .await?;
     draft.add_live(expanded_live);
diff --git a/crates/control-plane-api/src/live_specs/db.rs b/crates/control-plane-api/src/live_specs/db.rs
index 2858bb683b7..1d46f910afb 100644
--- a/crates/control-plane-api/src/live_specs/db.rs
+++ b/crates/control-plane-api/src/live_specs/db.rs
@@ -1,7 +1,7 @@
 use crate::TextJson;
 use models::{Capability, CatalogType, Id};
 use serde_json::value::RawValue;
-use sqlx::types::{Json, Uuid};
+use sqlx::types::Json;
 use tables::RoleGrant;
 
 /// Deletes the given live spec row, along with the corresponding `controller_jobs` row.
@@ -134,17 +134,18 @@ pub async fn fetch_inferred_schemas(
 /// Queries for all non-deleted `live_specs` that are connected to the given `collection_names` via
 /// `live_spec_flows`.
 pub async fn fetch_expanded_live_specs(
-    user_id: Uuid,
+    user_id: uuid::Uuid,
     collection_names: &[&str],
     exclude_names: &[&str],
     db: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
+    snapshot: &crate::Snapshot,
 ) -> sqlx::Result<Vec<LiveSpec>> {
-    sqlx::query_as!(
+    let mut expanded = sqlx::query_as!(
         LiveSpec,
         r#"
         with collections(id) as (
             select ls.id
-            from unnest($2::text[]) as names(catalog_name)
+            from unnest($1::text[]) as names(catalog_name)
             join live_specs ls on ls.catalog_name = names.catalog_name
         ),
         exp(id) as (
@@ -166,27 +167,39 @@ pub async fn fetch_expanded_live_specs(
             ls.spec as "spec: TextJson<Box<RawValue>>",
             ls.built_spec as "built_spec: TextJson<Box<RawValue>>",
             ls.inferred_schema_md5,
-            (
-                select max(capability) from internal.user_roles($1) r
-                where starts_with(ls.catalog_name, r.role_prefix)
-            ) as "user_capability: Capability",
-            coalesce(
-                (select json_agg(row_to_json(role_grants))
-                from role_grants
-                where starts_with(ls.catalog_name, subject_role)),
-                '[]'
-            ) as "spec_capabilities!: Json<Vec<RoleGrant>>",
+            -- `user_capability` and `spec_capabilities` are synthesized from the
+            -- authorization Snapshot below rather than queried here. The `null`
+            -- (→ None) and empty-array placeholders are overwritten afterwards.
+            null as "user_capability: Capability",
+            '[]' as "spec_capabilities!: Json<Vec<RoleGrant>>",
             ls.dependency_hash
         from exp
         join live_specs ls on ls.id = exp.id
-        where ls.spec is not null and not ls.catalog_name = any($3);
+        where ls.spec is not null and not ls.catalog_name = any($2);
         "#,
-        user_id,
         collection_names as &[&str],
         exclude_names as &[&str],
     )
     .fetch_all(db)
-    .await
+    .await?;
+
+    // Compute each spec's `user_capability` (the user's greatest capability among
+    // the prefixes that `catalog_name` falls under) and `spec_capabilities` (the
+    // role grants the spec holds by virtue of its own name/role) from the
+    // Snapshot, replacing the previous `user_roles` and `role_grants` subqueries.
+    let reachable = snapshot.prefix_and_capabilities_per_user(user_id);
+    for spec in expanded.iter_mut() {
+        let mut max_capability: Option<Capability> = None;
+        for (prefix, (_, capability)) in reachable.iter() {
+            if spec.catalog_name.starts_with(*prefix) {
+                max_capability = max_capability.max(Some(*capability));
+            }
+        }
+        spec.user_capability = max_capability;
+        spec.spec_capabilities = Json(snapshot.spec_capabilities(&spec.catalog_name));
+    }
+
+    Ok(expanded)
 }
 
 /// Returns all live spec names under the given prefix.
diff --git a/crates/control-plane-api/src/live_specs/mod.rs b/crates/control-plane-api/src/live_specs/mod.rs
index 75cb724775b..ce65a86a4d7 100644
--- a/crates/control-plane-api/src/live_specs/mod.rs
+++ b/crates/control-plane-api/src/live_specs/mod.rs
@@ -7,7 +7,6 @@ pub use db::{
 };
 use models::Capability;
 use std::ops::Deref;
-use uuid::Uuid;
 
 /// Fetches live specs, returning them as a `tables::LiveCatalog`. Optionally
 /// filters the specs based on user capability. If `filter_capability` is
@@ -76,14 +75,16 @@ pub async fn get_live_specs(
 }
 
 pub async fn get_connected_live_specs(
-    user_id: Uuid,
+    user_id: uuid::Uuid,
     collection_names: &[&str],
     exclude_names: &[&str],
     filter_capability: Option<Capability>,
     db: &sqlx::PgPool,
+    snapshot: &crate::Snapshot,
 ) -> anyhow::Result<tables::LiveCatalog> {
     let expanded_rows =
-        db::fetch_expanded_live_specs(user_id, collection_names, exclude_names, db).await?;
+        db::fetch_expanded_live_specs(user_id, collection_names, exclude_names, db, snapshot)
+            .await?;
     let mut live = tables::LiveCatalog::default();
     for exp in expanded_rows {
         if let Some(minimum_capability) = filter_capability {
diff --git a/crates/control-plane-api/src/publications/initialize.rs b/crates/control-plane-api/src/publications/initialize.rs
index dd142d50dac..e971358e675 100644
--- a/crates/control-plane-api/src/publications/initialize.rs
+++ b/crates/control-plane-api/src/publications/initialize.rs
@@ -1,7 +1,8 @@
+use crate::Snapshot;
 use anyhow::Context;
 use itertools::Itertools;
 use models::Capability;
-use std::future::Future;
+use std::{future::Future, sync::Arc};
 use uuid::Uuid;
 
 /// Initialize a draft prior to build/validation. This may add additional specs to the draft.
@@ -11,6 +12,7 @@ pub trait Initialize: Send + Sync {
         db: &sqlx::PgPool,
         user_id: Uuid,
         draft: &mut tables::DraftCatalog,
+        snapshot_watch: &Arc<dyn tokens::Watch<Snapshot>>,
     ) -> impl Future<Output = anyhow::Result<()>> + Send;
 }
 
@@ -22,6 +24,7 @@ impl Initialize for NoopInitialize {
         _db: &sqlx::PgPool,
         _user_id: Uuid,
         _draft: &mut tables::DraftCatalog,
+        _snapshot_watch: &Arc<dyn tokens::Watch<Snapshot>>,
     ) -> anyhow::Result<()> {
         Ok(())
     }
@@ -37,9 +40,14 @@ where
         db: &sqlx::PgPool,
         user_id: Uuid,
         draft: &mut tables::DraftCatalog,
+        snapshot_watch: &Arc<dyn tokens::Watch<Snapshot>>,
     ) -> anyhow::Result<()> {
-        self.0.initialize(db, user_id, draft).await?;
-        self.1.initialize(db, user_id, draft).await?;
+        self.0
+            .initialize(db, user_id, draft, snapshot_watch)
+            .await?;
+        self.1
+            .initialize(db, user_id, draft, snapshot_watch)
+            .await?;
         Ok(())
     }
 }
@@ -65,6 +73,7 @@ impl Initialize for ExpandDraft {
         db: &sqlx::PgPool,
         user_id: Uuid,
         draft: &mut tables::DraftCatalog,
+        snapshot_watch: &Arc<dyn tokens::Watch<Snapshot>>,
     ) -> anyhow::Result<()> {
         // Expand the set of drafted specs to include any tasks that read from or write to any of
         // the published collections. We do this so that validation can catch any inconsistencies
@@ -81,12 +90,16 @@ impl Initialize for ExpandDraft {
         } else {
             None
         };
+        let snapshot = snapshot_watch.token();
+        let snapshot = snapshot.result().unwrap();
+
         let expanded_catalog = crate::live_specs::get_connected_live_specs(
             user_id,
             &drafted_collections,
             &all_drafted_specs,
             capability_filter,
             db,
+            snapshot,
         )
         .await?;
         tracing::debug!(
@@ -112,6 +125,7 @@ impl Initialize for RuntimeV2Rollout {
         db: &sqlx::PgPool,
         _user_id: Uuid,
         draft: &mut tables::DraftCatalog,
+        _snapshot_watch: &Arc<dyn tokens::Watch<Snapshot>>,
     ) -> anyhow::Result<()> {
         if !self.new_captures {
             return Ok(());
diff --git a/crates/control-plane-api/src/publications/mod.rs b/crates/control-plane-api/src/publications/mod.rs
index 5017f9103a7..689f064c6b2 100644
--- a/crates/control-plane-api/src/publications/mod.rs
+++ b/crates/control-plane-api/src/publications/mod.rs
@@ -143,7 +143,6 @@ impl PublicationResult {
 
 /// A PublishHandler is a Handler which publishes catalog specifications.
 #[derive(Clone)]
-#[allow(dead_code)]
 pub struct Publisher {
     flowctl_go: std::path::PathBuf,
     builds_root: url::Url,
@@ -317,7 +316,7 @@ impl Publisher {
     ) -> anyhow::Result<PublicationResult> {
         let mut draft = raw_draft.clone_specs();
         initialize
-            .initialize(&self.db, *user_id, &mut draft)
+            .initialize(&self.db, *user_id, &mut draft, &self.snapshot)
             .await
             .context("initializing draft")?;
         // It's important that we generate the pub id inside the retry loop so that we can
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 15ca92f9802..0615d63b5a4 100644
--- a/crates/control-plane-api/src/server/create_data_plane.rs
+++ b/crates/control-plane-api/src/server/create_data_plane.rs
@@ -63,20 +63,15 @@ pub async fn create_data_plane(
         category,
     }): super::Request<Request>,
 ) -> Result<axum::Json<Response>, crate::server::error::ApiError> {
-    let models::authorizations::ControlClaims { sub: user_id, .. } = env.claims()?;
+    let claims = env.claims()?;
+    let user_id = &claims.sub;
+    let user_email = claims.email.as_deref().unwrap_or("user");
 
-    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());
-    }
+    // Authorize against the request's Snapshot (not a fresh watch token) so that
+    // `authorization_outcome` can refresh-and-retry a denial that was decided
+    // from a Snapshot older than the request.
+    let policy_result = evaluate_ops_admin_authorization(env.snapshot(), *user_id, user_email);
+    let (_expiry, ()) = env.authorization_outcome(policy_result).await?;
 
     let (data_plane_fqdn, base_name, pulumi_stack) = match &private {
         None => (
@@ -317,6 +312,39 @@ pub async fn create_data_plane(
     Ok(axum::Json(Response {}))
 }
 
+/// Builds a policy result (for `Envelope::authorization_outcome`) asserting that
+/// `user_id` holds `admin` capability resolving to exactly the `ops/` tenant,
+/// which is required to create a data-plane.
+///
+/// This replaces the prior
+/// `internal.user_roles($user, 'admin') where role_prefix = 'ops/'` SQL check.
+/// Because that check was an *exact* `ops/` match — a data-plane always lives
+/// under `ops/` — this looks for `admin` at precisely the `ops/` prefix, and
+/// deliberately does not accept a grant at an ancestor or a descendant of it.
+///
+/// It evaluates against a caller-provided `Snapshot` (the request's Snapshot via
+/// `Envelope::snapshot`) rather than re-reading the watch, so that
+/// `authorization_outcome` can refresh-and-retry a denial decided from a stale
+/// Snapshot.
+pub(super) fn evaluate_ops_admin_authorization(
+    snapshot: &crate::Snapshot,
+    user_id: uuid::Uuid,
+    user_email: &str,
+) -> crate::AuthZResult<()> {
+    let is_ops_admin = snapshot
+        .prefix_and_capabilities_per_user(user_id)
+        .get("ops/")
+        .is_some_and(|(_bits, legacy)| *legacy >= models::Capability::Admin);
+
+    if is_ops_admin {
+        Ok((None, ()))
+    } else {
+        Err(tonic::Status::permission_denied(format!(
+            "{user_email} is not an admin of the 'ops/' tenant",
+        )))
+    }
+}
+
 impl Validate for Category {
     fn validate(&self) -> Result<(), validator::ValidationErrors> {
         if let Self::Manual(manual) = &self {
@@ -326,3 +354,117 @@ impl Validate for Category {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::evaluate_ops_admin_authorization;
+
+    // Inserts `user_id` along with the given `user_grants`
+    // (as `(object_role, capability)`) and `role_grants` (as
+    // `(subject_role, object_role, capability)`), builds an authorization
+    // Snapshot from that database state, then evaluates the `ops/` admin check
+    // for that user. Running through a real Snapshot built from real grants
+    // proves the snapshot-based check preserves the prior SQL behavior across
+    // every scenario below.
+    async fn eval(
+        pool: &sqlx::PgPool,
+        user_id: uuid::Uuid,
+        user_grants: &[(&str, &str)],
+        role_grants: &[(&str, &str, &str)],
+    ) -> bool {
+        sqlx::query("insert into auth.users (id, email) values ($1, $2)")
+            .bind(user_id)
+            .bind(format!("{user_id}@example.com"))
+            .execute(pool)
+            .await
+            .unwrap();
+
+        for (object_role, capability) in user_grants {
+            sqlx::query(
+                "insert into user_grants (user_id, object_role, capability)
+                 values ($1, $2, $3::grant_capability)",
+            )
+            .bind(user_id)
+            .bind(object_role)
+            .bind(capability)
+            .execute(pool)
+            .await
+            .unwrap();
+        }
+        for (subject_role, object_role, capability) in role_grants {
+            sqlx::query(
+                "insert into role_grants (subject_role, object_role, capability)
+                 values ($1, $2, $3::grant_capability)",
+            )
+            .bind(subject_role)
+            .bind(object_role)
+            .bind(capability)
+            .execute(pool)
+            .await
+            .unwrap();
+        }
+
+        // `gate: false` so the first (and only) refresh serves the real
+        // snapshot built from the grants inserted above, rather than the empty
+        // snapshot used to exercise the server's retry flow.
+        let snapshot_watch = crate::test_server::snapshot(pool.clone(), false).await;
+        let refresh = snapshot_watch.token();
+        let snapshot = refresh.result().unwrap();
+        evaluate_ops_admin_authorization(snapshot, user_id, "test@example.com").is_ok()
+    }
+
+    // A user granted `admin` directly on `ops/` is authorized.
+    #[sqlx::test(migrations = "../../supabase/migrations")]
+    async fn admin_directly_on_ops_is_authorized(pool: sqlx::PgPool) {
+        let user_id = uuid::Uuid::from_bytes([1; 16]);
+        assert!(eval(&pool, user_id, &[("ops/", "admin")], &[]).await);
+    }
+
+    // A user reaching `admin` on `ops/` transitively through a role grant
+    // (e.g. an `estuary_support/` role that is itself granted admin over `ops/`)
+    // is authorized. This exercises the recursive expansion in `user_roles`.
+    #[sqlx::test(migrations = "../../supabase/migrations")]
+    async fn admin_on_ops_via_role_grant_is_authorized(pool: sqlx::PgPool) {
+        let user_id = uuid::Uuid::from_bytes([2; 16]);
+        assert!(
+            eval(
+                &pool,
+                user_id,
+                &[("estuary_support/", "admin")],
+                &[("estuary_support/", "ops/", "admin")],
+            )
+            .await
+        );
+    }
+
+    // Admin over a *sub-prefix* of `ops/` (e.g. `ops/dp/`) does NOT authorize:
+    // the check is an exact `role_prefix = 'ops/'` match, not a prefix match.
+    // This deliberately differs from the storage-mappings sub-prefix behavior,
+    // and pins it so a later snapshot refactor cannot silently loosen it.
+    #[sqlx::test(migrations = "../../supabase/migrations")]
+    async fn admin_on_sub_prefix_of_ops_is_denied(pool: sqlx::PgPool) {
+        let user_id = uuid::Uuid::from_bytes([3; 16]);
+        assert!(!eval(&pool, user_id, &[("ops/dp/", "admin")], &[]).await);
+    }
+
+    // A capability below `admin` on `ops/` is not enough.
+    #[sqlx::test(migrations = "../../supabase/migrations")]
+    async fn write_on_ops_is_denied(pool: sqlx::PgPool) {
+        let user_id = uuid::Uuid::from_bytes([4; 16]);
+        assert!(!eval(&pool, user_id, &[("ops/", "write")], &[]).await);
+    }
+
+    // Admin over an unrelated tenant confers no authority over `ops/`.
+    #[sqlx::test(migrations = "../../supabase/migrations")]
+    async fn admin_on_unrelated_tenant_is_denied(pool: sqlx::PgPool) {
+        let user_id = uuid::Uuid::from_bytes([5; 16]);
+        assert!(!eval(&pool, user_id, &[("aliceCo/", "admin")], &[]).await);
+    }
+
+    // A user with no grants at all is denied.
+    #[sqlx::test(migrations = "../../supabase/migrations")]
+    async fn no_grants_is_denied(pool: sqlx::PgPool) {
+        let user_id = uuid::Uuid::from_bytes([6; 16]);
+        assert!(!eval(&pool, user_id, &[], &[]).await);
+    }
+}
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 9bf0e42d082..b69a70a01c8 100644
--- a/crates/control-plane-api/src/server/update_l2_reporting.rs
+++ b/crates/control-plane-api/src/server/update_l2_reporting.rs
@@ -32,20 +32,19 @@ pub async fn update_l2_reporting(
         dry_run,
     }): 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());
-    }
+    let claims = env.claims()?;
+    let user_id = &claims.sub;
+    let user_email = claims.email.as_deref().unwrap_or("user");
+
+    // Authorize against the request's Snapshot (not a fresh watch token) so that
+    // `authorization_outcome` can refresh-and-retry a denial that was decided
+    // from a Snapshot older than the request.
+    let policy_result = super::create_data_plane::evaluate_ops_admin_authorization(
+        env.snapshot(),
+        *user_id,
+        user_email,
+    );
+    let (_expiry, ()) = env.authorization_outcome(policy_result).await?;
 
     let template = include_str!("../../../../ops-catalog/reporting-L2-template.bundle.json");
     let tables::DraftCatalog { collections, .. } =