From 4452c7ada92b62c7586b44d23d6e2330e06595f9 Mon Sep 17 00:00:00 2001
From: Greg Shear <greg@estuary.dev>
Date: Wed, 5 Aug 2026 15:57:50 -0400
Subject: [PATCH 1/3] authz: add AuthScope, a grant-graph ceiling on a
 subject's authority

An AuthScope narrows every authorization answer derived from it by intersecting the subject's capabilities at a name with the capabilities the scope confers there. Intersection means a scope can only remove authority, never add it, which is what makes a scope safe to carry in a bearer token: a scope frozen at issuance cannot over-authorize, and the grant tables stay the sole source of what the subject may do.

A scope's reach follows the grant graph rather than the literal prefix. `AuthScope::resolve(role_grants, "acmeCo/")` covers `acmeCo/` at full capabilities plus every prefix `acmeCo/` reaches through role_grants, at whatever capabilities those edges carry. So a role grant `acmeCo/ -> sharedCo/` places `sharedCo/` inside a scope of `acmeCo/`, while two tenants with no edge between them can never appear in one another's scope. Resolving with Assume leaves edges out of the prefix unattenuated, making the ceiling the full authority footprint a name under that prefix would itself reach.

`UserGrant::is_authorized`, `get_user_capability`, and `reachable_prefixes` now take a scope. `AuthScope::resolve` and `AuthScope::unscoped` are the only constructors, so every deliberately unscoped authorization is visible at its call site rather than being the value a forgotten argument defaults to. `RoleGrant::is_authorized` takes none: it asks what one catalog role may do to another, which is a property of the graph alone, and no token is involved.

Intersection for `reachable_prefixes` requires splitting rather than filtering. Where a user prefix and a scope prefix cover one another, the narrower of the pair is the intersection of the two subtrees and is what gets emitted: a user holding `acmeCo/` within a scope reaching only `acmeCo/team/` is authorized at `acmeCo/team/`. Filtering would have dropped the grant and denied access the user legitimately holds.

`get_user_capability` now also skips nodes whose capabilities are entirely removed by the ceiling. This matters under a scope, where reporting an unnarrowed legacy `admin` would drive dashboard affordances the scope does not permit, and it corrects the same overstatement for fully attenuated nodes in the unscoped case.
---
 crates/tables/src/behaviors.rs | 393 ++++++++++++++++++++++++++++++++-
 crates/tables/src/lib.rs       |  31 +++
 2 files changed, 416 insertions(+), 8 deletions(-)

diff --git a/crates/tables/src/behaviors.rs b/crates/tables/src/behaviors.rs
index a041c92a1ca..e20edde3920 100644
--- a/crates/tables/src/behaviors.rs
+++ b/crates/tables/src/behaviors.rs
@@ -64,20 +64,119 @@ fn effective_bits(
     bits
 }
 
+impl<'a> super::AuthScope<'a> {
+    /// A scope that narrows nothing: answers reflect the subject's full
+    /// authority.
+    pub fn unscoped() -> Self {
+        Self {
+            reach: None,
+            prefix: None,
+        }
+    }
+
+    /// Resolve the scope of `prefix` against `role_grants`.
+    ///
+    /// `prefix` confers all capabilities at and under itself, because scoping to
+    /// a prefix is not meant to attenuate the subject's own grants there — the
+    /// intersection against those grants is what narrows them. Prefixes reached
+    /// through role grants confer only the capabilities their edges carry.
+    pub fn resolve(role_grants: &'a [super::RoleGrant], prefix: &'a str) -> Self {
+        let mut reach: std::collections::BTreeMap<
+            &'a str,
+            (authz::CapabilitySet, models::Capability),
+        > = Default::default();
+
+        reach.insert(
+            prefix,
+            (authz::CapabilitySet::all(), models::Capability::Admin),
+        );
+
+        // Seeded with Assume, so edges out of `prefix` are unattenuated: the
+        // ceiling is the full authority footprint of that role, which is what a
+        // name under `prefix` would itself be able to reach.
+        for node in super::RoleGrant::reachable_nodes(role_grants, prefix) {
+            // A role edge can be visited yet confer nothing when delegation
+            // attenuation removes all of its capabilities. Such a destination
+            // is not in scope.
+            if node.capabilities.is_empty() {
+                continue;
+            }
+            let entry = reach
+                .entry(node.object_role)
+                .or_insert((authz::CapabilitySet::empty(), models::Capability::None));
+            entry.0 |= node.capabilities;
+            entry.1 = std::cmp::max(entry.1, node.legacy);
+        }
+
+        Self {
+            reach: Some(reach),
+            prefix: Some(prefix),
+        }
+    }
+
+    /// The prefix this scope was resolved from, or `None` when unscoped.
+    pub fn prefix(&self) -> Option<&'a str> {
+        self.prefix
+    }
+
+    /// The ceiling this scope places at `object_role_or_name`: the union of
+    /// capabilities over reach entries whose prefix covers the name, paired with
+    /// the max legacy capability among them.
+    ///
+    /// Bits union (and legacy maxes) across covering entries for the same reason
+    /// grant paths compose additively in [`any_path_satisfies`] — two role-grant
+    /// paths into the same subtree each contribute their own capabilities. The
+    /// unscoped ceiling is everything, making it the identity of both the
+    /// intersection applied to bits and the `min` applied to legacy.
+    fn ceiling_at(&self, object_role_or_name: &str) -> (authz::CapabilitySet, models::Capability) {
+        let Some(reach) = &self.reach else {
+            return (authz::CapabilitySet::all(), models::Capability::Admin);
+        };
+
+        let mut bits = authz::CapabilitySet::empty();
+        let mut legacy = models::Capability::None;
+        for (prefix, (prefix_bits, prefix_legacy)) in reach {
+            if object_role_or_name.starts_with(prefix) {
+                bits |= *prefix_bits;
+                legacy = std::cmp::max(legacy, *prefix_legacy);
+            }
+        }
+        (bits, legacy)
+    }
+
+    /// Iterate the scope's reach, or `None` when unscoped.
+    fn reach(
+        &self,
+    ) -> Option<impl Iterator<Item = (&'a str, (authz::CapabilitySet, models::Capability))> + '_>
+    {
+        self.reach
+            .as_ref()
+            .map(|reach| reach.iter().map(|(prefix, value)| (*prefix, *value)))
+    }
+}
+
 /// True when bits accumulated across `nodes` at prefixes covering
 /// `object_role_or_name` satisfy `required`. Bits compose additively
 /// across paths: distinct grant paths that each contribute partial bits
 /// at covering prefixes can jointly authorize a request that no single
 /// path would on its own.
+///
+/// Each path's contribution is first intersected with `scope`'s ceiling at the
+/// name. Intersection distributes over the union of paths, so masking per node
+/// gives the same answer as masking their union, and a name the scope does not
+/// reach at all has an empty ceiling and can satisfy nothing.
 fn any_path_satisfies<'a>(
     nodes: impl IntoIterator<Item = super::NodeRef<'a>>,
     object_role_or_name: &str,
     required: impl Into<authz::CapabilitySet>,
+    scope: &super::AuthScope<'_>,
 ) -> bool {
+    let (ceiling, _legacy) = scope.ceiling_at(object_role_or_name);
+
     let mut remaining = required.into();
     for node in nodes {
         if object_role_or_name.starts_with(node.object_role) {
-            remaining -= node.capabilities;
+            remaining -= node.capabilities & ceiling;
             if remaining.is_empty() {
                 return true;
             }
@@ -102,6 +201,13 @@ impl super::RoleGrant {
         .skip(1)
     }
 
+    /// Whether the role `subject_role_or_name` may act on
+    /// `object_role_or_name`.
+    ///
+    /// Takes no [`super::AuthScope`]: this asks what one catalog role may do to
+    /// another, which is a property of the grant graph alone. Scopes narrow the
+    /// authority of a *subject holding a token*, and no token is involved here —
+    /// task authorization and role-to-role checks are the callers.
     pub fn is_authorized<'a>(
         role_grants: &'a [super::RoleGrant],
         subject_role_or_name: &'a str,
@@ -112,6 +218,7 @@ impl super::RoleGrant {
             Self::reachable_nodes(role_grants, subject_role_or_name),
             object_role_or_name,
             capability,
+            &super::AuthScope::unscoped(),
         )
     }
 
@@ -150,10 +257,19 @@ impl super::UserGrant {
     /// a literal pass-through from storage, max'd across same-prefix
     /// arrivals. Applying a min-capability filter to the bit set agrees
     /// with `is_authorized` on the same inputs.
+    ///
+    /// Under a scope the result is the intersection of the user's prefixes with
+    /// the scope's reach, which requires splitting rather than filtering: where
+    /// one prefix covers the other, the *narrower* of the pair is the
+    /// intersection of the two subtrees and is what gets emitted. A user holding
+    /// `acmeCo/` within a scope reaching only `acmeCo/team/` is therefore
+    /// authorized at `acmeCo/team/`, not at `acmeCo/`. Pairs that don't overlap,
+    /// and pairs whose capabilities intersect to nothing, are dropped.
     pub fn reachable_prefixes<'a>(
         role_grants: &'a [super::RoleGrant],
         user_grants: &'a [super::UserGrant],
         user_id: uuid::Uuid,
+        scope: &super::AuthScope<'a>,
     ) -> std::collections::BTreeMap<&'a str, (authz::CapabilitySet, models::Capability)> {
         let mut out: std::collections::BTreeMap<
             &'a str,
@@ -168,18 +284,61 @@ impl super::UserGrant {
                 entry.1 = node.legacy;
             }
         }
-        out
+
+        let Some(reach) = scope.reach() else {
+            return out;
+        };
+        let reach: Vec<_> = reach.collect();
+
+        let mut scoped: std::collections::BTreeMap<
+            &'a str,
+            (authz::CapabilitySet, models::Capability),
+        > = Default::default();
+        for (user_prefix, (user_bits, user_legacy)) in out {
+            for (scope_prefix, (scope_bits, scope_legacy)) in reach.iter().copied() {
+                let narrower = if user_prefix.starts_with(scope_prefix) {
+                    user_prefix
+                } else if scope_prefix.starts_with(user_prefix) {
+                    scope_prefix
+                } else {
+                    continue; // Disjoint subtrees.
+                };
+
+                let bits = user_bits & scope_bits;
+                if bits.is_empty() {
+                    continue; // The scope removes everything the user holds here.
+                }
+
+                let entry = scoped
+                    .entry(narrower)
+                    .or_insert((authz::CapabilitySet::empty(), models::Capability::None));
+                entry.0 |= bits;
+                entry.1 = std::cmp::max(entry.1, std::cmp::min(user_legacy, scope_legacy));
+            }
+        }
+        scoped
     }
 
+    /// The max legacy `capability` column value the user holds at
+    /// `object_role_or_name`, or None if they hold none.
+    ///
+    /// Under a scope, each node's legacy value is clamped by the scope's own
+    /// legacy ceiling at the name, and nodes whose capabilities the scope removes
+    /// entirely are skipped: a node that confers no capabilities must not report
+    /// a capability level either.
     pub fn get_user_capability<'a>(
         role_grants: &'a [super::RoleGrant],
         user_grants: &'a [super::UserGrant],
         user_id: uuid::Uuid,
         object_role_or_name: &str,
+        scope: &super::AuthScope<'_>,
     ) -> Option<models::Capability> {
+        let (ceiling, ceiling_legacy) = scope.ceiling_at(object_role_or_name);
+
         Self::reachable_nodes(role_grants, user_grants, user_id)
             .filter(|n| object_role_or_name.starts_with(n.object_role))
-            .map(|n| n.legacy)
+            .filter(|n| !(n.capabilities & ceiling).is_empty())
+            .map(|n| std::cmp::min(n.legacy, ceiling_legacy))
             .filter(|c| *c != models::Capability::None)
             .max()
     }
@@ -190,11 +349,13 @@ impl super::UserGrant {
         subject_user_id: uuid::Uuid,
         object_role_or_name: &'a str,
         capability: impl Into<authz::CapabilitySet>,
+        scope: &super::AuthScope<'_>,
     ) -> bool {
         any_path_satisfies(
             Self::reachable_nodes(role_grants, user_grants, subject_user_id),
             object_role_or_name,
             capability,
+            scope,
         )
     }
 
@@ -447,6 +608,7 @@ mod test {
             uuid::Uuid::nil(),
             "bobCo/thing",
             models::Capability::Read,
+            &crate::AuthScope::unscoped(),
         ));
         assert!(!UserGrant::is_authorized(
             &role_grants,
@@ -454,6 +616,7 @@ mod test {
             uuid::Uuid::nil(),
             "bobCo/thing",
             models::Capability::Write,
+            &crate::AuthScope::unscoped(),
         ));
         assert!(UserGrant::is_authorized(
             &role_grants,
@@ -461,6 +624,7 @@ mod test {
             uuid::Uuid::nil(),
             "carolCo/hidden/thing",
             models::Capability::Read,
+            &crate::AuthScope::unscoped(),
         ));
 
         // User max: admin on aliceCo/widgets/ (propagates to bobCo/burgers/).
@@ -470,6 +634,7 @@ mod test {
             uuid::Uuid::max(),
             "bobCo/burgers/thing",
             models::Capability::Admin,
+            &crate::AuthScope::unscoped(),
         ));
     }
 
@@ -582,7 +747,8 @@ mod test {
                 &role_grants,
                 &user_grants,
                 user1,
-                "ops/private/dp/acmeCo/foooo"
+                "ops/private/dp/acmeCo/foooo",
+                &crate::AuthScope::unscoped(),
             )
         );
         assert_eq!(
@@ -591,7 +757,8 @@ mod test {
                 &role_grants,
                 &user_grants,
                 user2,
-                "ops/private/dp/acmeCo/foooo"
+                "ops/private/dp/acmeCo/foooo",
+                &crate::AuthScope::unscoped(),
             )
         );
         assert_eq!(
@@ -600,7 +767,8 @@ mod test {
                 &role_grants,
                 &user_grants,
                 user1,
-                "different/co/altogether"
+                "different/co/altogether",
+                &crate::AuthScope::unscoped(),
             )
         );
     }
@@ -643,6 +811,7 @@ mod test {
             uuid::Uuid::from_bytes([1; 16]),
             "ops/private/dp/acmeCo/foo",
             models::Capability::Read,
+            &crate::AuthScope::unscoped(),
         ));
         // User 2 has admin on acmeCo/nested/, which also picks up the
         // acmeCo/ role grants (parent prefix matching).
@@ -652,6 +821,7 @@ mod test {
             uuid::Uuid::from_bytes([2; 16]),
             "ops/private/dp/acmeCo/foo",
             models::Capability::Read,
+            &crate::AuthScope::unscoped(),
         ));
     }
 
@@ -705,7 +875,14 @@ mod test {
         required: EnumSet<Capability>,
     ) {
         assert!(
-            UserGrant::is_authorized(role_grants, user_grants, user_id, name, required),
+            UserGrant::is_authorized(
+                role_grants,
+                user_grants,
+                user_id,
+                name,
+                required,
+                &crate::AuthScope::unscoped()
+            ),
             "expected {user_id} to have {required:?} on {name}",
         );
     }
@@ -718,7 +895,14 @@ mod test {
         required: EnumSet<Capability>,
     ) {
         assert!(
-            !UserGrant::is_authorized(role_grants, user_grants, user_id, name, required),
+            !UserGrant::is_authorized(
+                role_grants,
+                user_grants,
+                user_id,
+                name,
+                required,
+                &crate::AuthScope::unscoped()
+            ),
             "expected {user_id} NOT to have {required:?} on {name}",
         );
     }
@@ -1700,4 +1884,197 @@ mod test {
             CapabilityBundle::Viewer.capabilities(),
         );
     }
+
+    #[test]
+    fn test_scope_drops_prefixes_the_scope_does_not_reach() {
+        // Alice administers two tenants with no role grant between them. A scope
+        // of one confines her to it entirely — this is the containment the scope
+        // exists to provide.
+        let (role_grants, user_grants, user_id) = build_scenario(
+            vec![
+                ("acmeCo/", vec![CapabilityBundle::Admin]),
+                ("otherCo/", vec![CapabilityBundle::Admin]),
+            ],
+            vec![],
+        );
+        let scope = crate::AuthScope::resolve(&role_grants, "acmeCo/");
+
+        let reachable = UserGrant::reachable_prefixes(&role_grants, &user_grants, user_id, &scope);
+        assert_eq!(
+            reachable.keys().copied().collect::<Vec<_>>(),
+            vec!["acmeCo/"]
+        );
+
+        assert!(UserGrant::is_authorized(
+            &role_grants,
+            &user_grants,
+            user_id,
+            "acmeCo/thing",
+            CapabilityBundle::Admin.capabilities(),
+            &scope,
+        ));
+        assert!(!UserGrant::is_authorized(
+            &role_grants,
+            &user_grants,
+            user_id,
+            "otherCo/thing",
+            Capability::CatalogRead,
+            &scope,
+        ));
+        // Unscoped, the very same grants reach both tenants.
+        assert!(UserGrant::is_authorized(
+            &role_grants,
+            &user_grants,
+            user_id,
+            "otherCo/thing",
+            CapabilityBundle::Admin.capabilities(),
+            &crate::AuthScope::unscoped(),
+        ));
+    }
+
+    #[test]
+    fn test_scope_follows_role_grants_and_clamps_to_the_edge() {
+        // acmeCo/ reaches sharedCo/ as a Viewer. Alice independently administers
+        // sharedCo/, but a scope of acmeCo/ confines her there to what the edge
+        // carries: a scope reaches through the grant graph, and only as far as
+        // the graph's own capabilities go.
+        let (role_grants, user_grants, user_id) = build_scenario(
+            vec![
+                ("acmeCo/", vec![CapabilityBundle::Admin]),
+                ("sharedCo/", vec![CapabilityBundle::Admin]),
+            ],
+            vec![("acmeCo/", "sharedCo/", vec![CapabilityBundle::Viewer])],
+        );
+        let scope = crate::AuthScope::resolve(&role_grants, "acmeCo/");
+
+        let reachable = UserGrant::reachable_prefixes(&role_grants, &user_grants, user_id, &scope);
+        assert_eq!(
+            reachable.keys().copied().collect::<Vec<_>>(),
+            vec!["acmeCo/", "sharedCo/"]
+        );
+        assert_eq!(
+            reachable["acmeCo/"].0,
+            CapabilityBundle::Admin.capabilities()
+        );
+        assert_eq!(
+            reachable["sharedCo/"].0,
+            CapabilityBundle::Viewer.capabilities()
+        );
+
+        // Her admin authority at sharedCo/ does not survive the scope.
+        assert!(UserGrant::is_authorized(
+            &role_grants,
+            &user_grants,
+            user_id,
+            "sharedCo/thing",
+            CapabilityBundle::Viewer.capabilities(),
+            &scope,
+        ));
+        assert!(!UserGrant::is_authorized(
+            &role_grants,
+            &user_grants,
+            user_id,
+            "sharedCo/thing",
+            Capability::SpecEdit,
+            &scope,
+        ));
+    }
+
+    #[test]
+    fn test_scope_narrower_than_a_grant_splits_the_prefix() {
+        // Alice administers acmeCo/, but the scope reaches only acmeCo/team/.
+        // The intersection of the two subtrees is the narrower prefix, so that
+        // is what she is authorized at — filtering alone would have dropped her
+        // grant entirely and denied access she legitimately holds.
+        let (role_grants, user_grants, user_id) =
+            build_scenario(vec![("acmeCo/", vec![CapabilityBundle::Admin])], vec![]);
+        let scope = crate::AuthScope::resolve(&role_grants, "acmeCo/team/");
+
+        let reachable = UserGrant::reachable_prefixes(&role_grants, &user_grants, user_id, &scope);
+        assert_eq!(
+            reachable.keys().copied().collect::<Vec<_>>(),
+            vec!["acmeCo/team/"]
+        );
+        assert_eq!(
+            reachable["acmeCo/team/"].0,
+            CapabilityBundle::Admin.capabilities()
+        );
+
+        assert!(UserGrant::is_authorized(
+            &role_grants,
+            &user_grants,
+            user_id,
+            "acmeCo/team/thing",
+            CapabilityBundle::Admin.capabilities(),
+            &scope,
+        ));
+        assert!(!UserGrant::is_authorized(
+            &role_grants,
+            &user_grants,
+            user_id,
+            "acmeCo/other/thing",
+            Capability::CatalogRead,
+            &scope,
+        ));
+    }
+
+    #[test]
+    fn test_scope_clamps_the_legacy_capability() {
+        // The legacy `capability` column drives dashboard affordances, so it has
+        // to narrow with the bits. Alice is an admin of sharedCo/ directly, but
+        // acmeCo/ only reaches it with legacy `read`.
+        let user_id = uuid::Uuid::from_bytes([1; 16]);
+        let user_grants = UserGrants::from_iter(
+            [
+                ("acmeCo/", models::Capability::Admin),
+                ("sharedCo/", models::Capability::Admin),
+            ]
+            .into_iter()
+            .map(|(obj, capability)| UserGrant {
+                user_id,
+                object_role: models::Prefix::new(obj),
+                capability,
+                bundles: vec![],
+            }),
+        );
+        let role_grants = RoleGrants::from_iter([RoleGrant {
+            subject_role: models::Prefix::new("acmeCo/"),
+            object_role: models::Prefix::new("sharedCo/"),
+            capability: models::Capability::Read,
+            bundles: vec![],
+        }]);
+        let scope = crate::AuthScope::resolve(&role_grants, "acmeCo/");
+
+        assert_eq!(
+            Some(models::Capability::Read),
+            UserGrant::get_user_capability(
+                &role_grants,
+                &user_grants,
+                user_id,
+                "sharedCo/thing",
+                &scope,
+            )
+        );
+        assert_eq!(
+            Some(models::Capability::Admin),
+            UserGrant::get_user_capability(
+                &role_grants,
+                &user_grants,
+                user_id,
+                "sharedCo/thing",
+                &crate::AuthScope::unscoped(),
+            )
+        );
+        // At and under the scope prefix the user's own grant is untouched.
+        assert_eq!(
+            Some(models::Capability::Admin),
+            UserGrant::get_user_capability(
+                &role_grants,
+                &user_grants,
+                user_id,
+                "acmeCo/thing",
+                &scope,
+            )
+        );
+    }
 }
diff --git a/crates/tables/src/lib.rs b/crates/tables/src/lib.rs
index 5e8623cafe8..a24793eb5b6 100644
--- a/crates/tables/src/lib.rs
+++ b/crates/tables/src/lib.rs
@@ -421,6 +421,37 @@ pub struct NodeRef<'a> {
     pub legacy: models::Capability,
 }
 
+/// AuthScope is a ceiling placed on a subject's authority.
+///
+/// Every authorization answer derived with a scope intersects the subject's own
+/// capabilities at a name with the capabilities the scope confers there, so a
+/// scope can only remove authority and never add it. This makes a scope safe to
+/// carry in a bearer token: a stale scope cannot over-authorize, and the grant
+/// tables remain the sole source of what the subject may do.
+///
+/// A scope's reach follows the grant graph rather than the literal prefix. It
+/// covers the prefix itself at full capabilities, plus every prefix that prefix
+/// reaches through `role_grants` — so a role grant `acmeCo/ -> sharedCo/` puts
+/// `sharedCo/` inside a scope of `acmeCo/`, at whatever capabilities that edge
+/// carries. Two tenants with no role grant between them can never appear in one
+/// another's scope.
+///
+/// [`AuthScope::resolve`] and [`AuthScope::unscoped`] are the only
+/// constructors, which makes every deliberately unscoped authorization visible
+/// at its call site.
+pub struct AuthScope<'a> {
+    /// Maps each prefix the scope reaches to the capabilities it confers there.
+    /// `None` is the unscoped case and confers everything everywhere.
+    ///
+    /// Set together with `prefix` by the two constructors: both are `None` when
+    /// unscoped and both are `Some` when scoped.
+    reach: Option<
+        std::collections::BTreeMap<&'a str, (models::authz::CapabilitySet, models::Capability)>,
+    >,
+    /// The prefix this scope was resolved from, for error messages.
+    prefix: Option<&'a str>,
+}
+
 /// Attempts to parse a catalog type and name from a URL in the form of:
 /// `flow://<catalog-type>/<catalog-name>`. Returns None if the URL doesn't
 /// have a valid `CatalogType`, or if the scheme doesn't match.

From 8259e50beb8686bb7fea8de35e22466ac619ea77 Mon Sep 17 00:00:00 2001
From: Greg Shear <greg@estuary.dev>
Date: Wed, 5 Aug 2026 16:14:01 -0400
Subject: [PATCH 2/3] control-plane-api: route all authorization through a
 request-scoped Authority
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds `crate::Authority`: the caller's resolved authorization inputs for one request — the Snapshot's grant tables, the authenticated user, and the scope their token confines them to. Handlers get one via `Envelope::authority()` and ask it their authorization questions instead of calling `tables::UserGrant` directly.

This is what makes a scope enforceable rather than remembered. Threading an `Option<&str>` scope through the dozen-plus call sites would leave a dozen chances to omit it, and the value that compiles by default (`None`) is also the unscoped one. An Authority already carries the scope, so there is no argument to forget. Answering a question without a scope now requires building an `AuthScope::unscoped()` in place, which keeps the complete set of unscoped authorizations greppable.

Four aggregating helpers become Authority methods, and every direct caller of the grant tables now goes through one of them:

- `evaluate_names_authorization` -> `Authority::evaluate` (verify_authorization, live_spec_refs, alerts, alert_configs, open_metrics, status)
- `attach_user_capabilities` -> `Authority::attach_capabilities` (live_spec_refs, live_specs, data_planes)
- `may_access` and `authorized_prefixes`/`filtered_authorized_prefixes` now take an Authority (alert_configs, invite_links, service_accounts, storage_mappings)
- direct `tables::` calls in prefixes, data_planes, storage_mappings, status, and the three `/authorize/user/*` endpoints

The `/authorize/user/*` endpoints mint data-plane tokens for journal and shard access, so routing them through Authority is what keeps a scoped control-plane token from reading collection data outside its scope. `authorize_task` keeps calling `RoleGrant::is_authorized`: task authorization asks what one catalog role may do to another and involves no user token.

`ControlClaims` grows `scope_prefix`, the claim an Authority resolves its scope from. Named to avoid colliding with the OAuth 2.0 `scope` claim, which is a space-delimited list and means something else. Issuance comes next; today nothing sets it, so every token resolves to an unscoped Authority and behavior is unchanged.

Two tests in `authorized_prefixes` cover what the chokepoint buys: a scope narrows every prefix-scoped list query without per-query work, and a caller-supplied filter naming an out-of-scope prefix returns nothing rather than restoring access to it.
---
 crates/agent/src/integration_tests/harness.rs |   1 +
 crates/control-plane-api/src/authority.rs     | 189 ++++++++++++++++++
 crates/control-plane-api/src/envelope.rs      |  10 +
 crates/control-plane-api/src/lib.rs           |   7 +-
 .../src/server/authorize_dekaf.rs             |   4 +
 .../src/server/authorize_user_collection.rs   |  39 ++--
 .../src/server/authorize_user_prefix.rs       |  48 ++---
 .../src/server/authorize_user_task.rs         |  40 ++--
 crates/control-plane-api/src/server/mod.rs    |  69 -------
 .../server/public/graphql/alert_configs.rs    |  20 +-
 .../src/server/public/graphql/alerts.rs       |   9 +-
 .../public/graphql/authorized_prefixes.rs     | 170 ++++++++++++----
 .../src/server/public/graphql/data_planes.rs  |  22 +-
 .../src/server/public/graphql/invite_links.rs |   5 +-
 .../server/public/graphql/live_spec_refs.rs   |  26 +--
 .../src/server/public/graphql/live_specs.rs   |   4 +-
 .../src/server/public/graphql/mod.rs          |  18 +-
 .../src/server/public/graphql/prefixes.rs     |   9 +-
 .../server/public/graphql/service_accounts.rs |   5 +-
 .../server/public/graphql/storage_mappings.rs |  75 +++----
 .../src/server/public/open_metrics.rs         |   9 +-
 .../src/server/public/status.rs               |  19 +-
 crates/control-plane-api/src/test_server.rs   |  12 ++
 crates/models/src/authorizations.rs           |  15 ++
 24 files changed, 475 insertions(+), 350 deletions(-)
 create mode 100644 crates/control-plane-api/src/authority.rs

diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs
index a78cc34140f..90fcc130140 100644
--- a/crates/agent/src/integration_tests/harness.rs
+++ b/crates/agent/src/integration_tests/harness.rs
@@ -1578,6 +1578,7 @@ impl TestHarness {
             exp: (req_start + chrono::Duration::hours(1)).timestamp() as u64,
             role: "authenticated".to_string(),
             email: Some("user@example.com".to_string()),
+            scope_prefix: None,
         };
 
         let token = tokens::jwt::sign(&claims, &app.control_plane_jwt_encode_key)
diff --git a/crates/control-plane-api/src/authority.rs b/crates/control-plane-api/src/authority.rs
new file mode 100644
index 00000000000..d207721dc05
--- /dev/null
+++ b/crates/control-plane-api/src/authority.rs
@@ -0,0 +1,189 @@
+/// Authority is the caller's resolved authorization inputs for one request: the
+/// Snapshot's grant tables, the authenticated user, and the scope their token
+/// confines them to.
+///
+/// It exists so that a scope cannot be forgotten. Handlers and resolvers ask an
+/// Authority their authorization questions rather than calling
+/// `tables::UserGrant` directly, and because the Authority already carries the
+/// scope there is no per-call-site argument to omit — the failure mode of
+/// threading an `Option` through a dozen call sites, where the value that
+/// compiles by default is also the unscoped one, does not arise.
+///
+/// A call site that must answer a question without a scope has to build its own
+/// `tables::AuthScope::unscoped()` and say why, which makes the complete set of
+/// unscoped authorizations greppable.
+pub struct Authority<'a> {
+    role_grants: &'a tables::RoleGrants,
+    user_grants: &'a tables::UserGrants,
+    user_id: uuid::Uuid,
+    /// The caller's email, or "user" when the token carries none. Only for
+    /// operator-facing error text.
+    user_email: &'a str,
+    scope: tables::AuthScope<'a>,
+}
+
+impl<'a> Authority<'a> {
+    /// Resolve authority from a Snapshot and verified control-plane claims.
+    ///
+    /// Resolving the token's scope walks the role-grant graph out from its
+    /// prefix. That is the same order of work as a single authorization check
+    /// (which walks the graph out from the user), so an Authority is resolved per
+    /// request rather than memoized: memoizing would require the scope to own its
+    /// prefixes instead of borrowing them from the Snapshot.
+    pub fn resolve(snapshot: &'a crate::Snapshot, claims: &'a crate::ControlClaims) -> Self {
+        let scope = match &claims.scope_prefix {
+            Some(prefix) => tables::AuthScope::resolve(&snapshot.role_grants, prefix),
+            None => tables::AuthScope::unscoped(),
+        };
+
+        Self {
+            role_grants: &snapshot.role_grants,
+            user_grants: &snapshot.user_grants,
+            user_id: claims.sub,
+            user_email: claims.email.as_deref().unwrap_or("user"),
+            scope,
+        }
+    }
+
+    /// Assemble authority from its parts.
+    ///
+    /// Prefer [`Self::resolve`], which derives them from a request's Snapshot and
+    /// claims. This exists for tests and for callers holding grant tables that
+    /// did not come from a Snapshot.
+    pub fn new(
+        role_grants: &'a tables::RoleGrants,
+        user_grants: &'a tables::UserGrants,
+        user_id: uuid::Uuid,
+        user_email: &'a str,
+        scope: tables::AuthScope<'a>,
+    ) -> Self {
+        Self {
+            role_grants,
+            user_grants,
+            user_id,
+            user_email,
+            scope,
+        }
+    }
+
+    /// The authenticated user.
+    pub fn user_id(&self) -> uuid::Uuid {
+        self.user_id
+    }
+
+    /// The scope confining this caller.
+    pub fn scope(&self) -> &tables::AuthScope<'a> {
+        &self.scope
+    }
+
+    /// The caller's email, or "user" when their token carries none. For
+    /// operator-facing error text only.
+    pub fn user_email(&self) -> &'a str {
+        self.user_email
+    }
+
+    /// Whether the caller holds `capability` on `prefix_or_name`, as a pure check
+    /// against the request's Snapshot and scope.
+    ///
+    /// This is the visibility gate: use it to hide a field or filter a list,
+    /// failing closed to an empty or default value. Unlike [`Self::evaluate`] it
+    /// neither errors nor asks for a Snapshot refresh on a negative result,
+    /// because momentarily hiding a field against a slightly-stale Snapshot is
+    /// the correct, low-cost behavior.
+    pub fn is_authorized(
+        &self,
+        prefix_or_name: &str,
+        capability: impl Into<models::authz::CapabilitySet>,
+    ) -> bool {
+        tables::UserGrant::is_authorized(
+            self.role_grants,
+            self.user_grants,
+            self.user_id,
+            prefix_or_name,
+            capability,
+            &self.scope,
+        )
+    }
+
+    /// The caller's legacy `capability` column value at `prefix_or_name`, or None
+    /// if they hold none there.
+    pub fn capability_at(&self, prefix_or_name: &str) -> Option<models::Capability> {
+        tables::UserGrant::get_user_capability(
+            self.role_grants,
+            self.user_grants,
+            self.user_id,
+            prefix_or_name,
+            &self.scope,
+        )
+    }
+
+    /// Every prefix the caller is authorized to, mapped to the capabilities they
+    /// hold there. Already narrowed by the scope.
+    pub fn reachable_prefixes(
+        &self,
+    ) -> std::collections::BTreeMap<&'a str, (models::authz::CapabilitySet, models::Capability)>
+    {
+        tables::UserGrant::reachable_prefixes(
+            self.role_grants,
+            self.user_grants,
+            self.user_id,
+            &self.scope,
+        )
+    }
+
+    /// Evaluate whether the caller holds at least `min_capability` on every one
+    /// of `prefixes_or_names`, returning a policy result shaped for
+    /// [`crate::Envelope::authorization_outcome`].
+    ///
+    /// This is the hard gate for mutations and access-controlled queries: a
+    /// denial becomes `permission_denied`, and a provisional denial against a
+    /// stale Snapshot follows the standard refresh-and-retry path.
+    pub fn evaluate<Iter, S, C>(
+        &self,
+        min_capability: C,
+        prefixes_or_names: Iter,
+    ) -> crate::AuthZResult<()>
+    where
+        Iter: IntoIterator<Item = S>,
+        S: AsRef<str> + std::fmt::Display,
+        C: Into<models::authz::CapabilitySet> + std::fmt::Display + Copy,
+    {
+        for prefix_or_name in prefixes_or_names {
+            if !self.is_authorized(prefix_or_name.as_ref(), min_capability) {
+                return Err(tonic::Status::permission_denied(format!(
+                    "{} is not authorized to access prefix or name '{prefix_or_name}' with required capability {min_capability}{}",
+                    self.user_email,
+                    self.scope_suffix(),
+                )));
+            }
+        }
+        Ok((None, ()))
+    }
+
+    /// Looks up the caller's capability at each item of `prefixes_or_names` and
+    /// calls `attach` with the item and that capability, collecting the `Some`
+    /// results.
+    pub fn attach_capabilities<Iter, F, T>(&self, prefixes_or_names: Iter, mut attach: F) -> Vec<T>
+    where
+        Iter: IntoIterator<Item = String>,
+        F: FnMut(String, Option<models::Capability>) -> Option<T>,
+    {
+        prefixes_or_names
+            .into_iter()
+            .flat_map(|prefix| {
+                let capability = self.capability_at(&prefix);
+                attach(prefix, capability)
+            })
+            .collect()
+    }
+
+    /// Names the scope in a denial message, so an operator can tell "you were
+    /// never granted this" apart from "your token is confined elsewhere" — the
+    /// two look identical to a caller who does in fact hold the grant.
+    fn scope_suffix(&self) -> String {
+        match self.scope.prefix() {
+            Some(prefix) => format!(" within the token's scope of '{prefix}'"),
+            None => String::new(),
+        }
+    }
+}
diff --git a/crates/control-plane-api/src/envelope.rs b/crates/control-plane-api/src/envelope.rs
index 3ca7d720771..1f79833eb45 100644
--- a/crates/control-plane-api/src/envelope.rs
+++ b/crates/control-plane-api/src/envelope.rs
@@ -81,6 +81,16 @@ impl Envelope {
         self.refresh.result().expect("Snapshot refresh never fails")
     }
 
+    /// Returns the caller's [`crate::Authority`]: the grant tables, the
+    /// authenticated user, and the scope their token is confined to.
+    ///
+    /// This is how request handlers ask authorization questions. Reaching into
+    /// `snapshot().role_grants` directly bypasses the token's scope and must be
+    /// justified in place.
+    pub fn authority(&self) -> tonic::Result<crate::Authority<'_>> {
+        Ok(crate::Authority::resolve(self.snapshot(), self.claims()?))
+    }
+
     /// Evaluate an authorization policy result and return its outcome.
     ///
     /// This method handles the complexity of Snapshot refresh, retry logic,
diff --git a/crates/control-plane-api/src/lib.rs b/crates/control-plane-api/src/lib.rs
index 856766a693e..0016c1aed63 100644
--- a/crates/control-plane-api/src/lib.rs
+++ b/crates/control-plane-api/src/lib.rs
@@ -4,6 +4,7 @@ use sqlx::types::Uuid;
 
 pub mod alert_subscriptions;
 pub mod alerts;
+mod authority;
 pub mod billing;
 pub mod connector_tags;
 pub mod controllers;
@@ -48,9 +49,13 @@ pub type AuthZResult<Ok> = tonic::Result<(Option<tokens::DateTime>, Ok)>;
 /// Envelope is common fields and parameters of every API request.
 pub use envelope::{Envelope, Locale, MaybeControlClaims};
 
+/// Authority is a caller's resolved authorization inputs, including the scope
+/// their token is confined to. It is the only route to the Snapshot's grant
+/// tables within this crate.
+pub use authority::Authority;
+
 // TODO(johnny): These types are all fundamental to this crate, and should be
 // hoisted from the `server` module. For now, just re-export to minimize churn.
-pub(crate) use server::evaluate_names_authorization;
 pub use server::{
     ApiError, App, AuthZRetry, build_router,
     snapshot::{self, Snapshot},
diff --git a/crates/control-plane-api/src/server/authorize_dekaf.rs b/crates/control-plane-api/src/server/authorize_dekaf.rs
index c043fced266..d23cf6e979c 100644
--- a/crates/control-plane-api/src/server/authorize_dekaf.rs
+++ b/crates/control-plane-api/src/server/authorize_dekaf.rs
@@ -94,6 +94,10 @@ pub async fn authorize_dekaf(
         sub: uuid::Uuid::nil(),
         role: DEKAF_ROLE.to_string(),
         email: None,
+        // Dekaf's token authenticates as the nil user and derives its authority
+        // from the `dekaf` Postgres role rather than from catalog grants, so a
+        // catalog-prefix scope has nothing to narrow.
+        scope_prefix: None,
     };
 
     // Only return a token if we are not redirecting
diff --git a/crates/control-plane-api/src/server/authorize_user_collection.rs b/crates/control-plane-api/src/server/authorize_user_collection.rs
index a02267bab29..9222e48b9cd 100644
--- a/crates/control-plane-api/src/server/authorize_user_collection.rs
+++ b/crates/control-plane-api/src/server/authorize_user_collection.rs
@@ -19,7 +19,7 @@ pub async fn authorize_user_collection(
     }
 
     let policy_result =
-        evaluate_authorization(env.snapshot(), env.claims()?, &collection, capability);
+        evaluate_authorization(env.snapshot(), &env.authority()?, &collection, capability);
 
     // Legacy: if `started_unix` was set then use a custom 200 response for client-side retries.
     let (expiry, (encoding_key, mut claims, broker_address, journal_name_prefix)) =
@@ -49,7 +49,7 @@ pub async fn authorize_user_collection(
 
 fn evaluate_authorization(
     snapshot: &crate::Snapshot,
-    claims: &crate::ControlClaims,
+    authority: &crate::Authority<'_>,
     collection_name: &models::Collection,
     capability: models::Capability,
 ) -> crate::AuthZResult<(
@@ -58,20 +58,10 @@ fn evaluate_authorization(
     String,
     String,
 )> {
-    let models::authorizations::ControlClaims {
-        sub: user_id,
-        email: user_email,
-        ..
-    } = claims;
-    let user_email = user_email.as_ref().map(String::as_str).unwrap_or("user");
-
-    if !tables::UserGrant::is_authorized(
-        &snapshot.role_grants,
-        &snapshot.user_grants,
-        *user_id,
-        collection_name,
-        capability,
-    ) {
+    let user_email = authority.user_email();
+    let user_id = authority.user_id();
+
+    if !authority.is_authorized(collection_name, capability) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not authorized to {collection_name} for {capability:?}",
         )));
@@ -79,13 +69,8 @@ fn evaluate_authorization(
 
     // For admin capability, require that the user has a transitive role grant to estuary_support/
     if capability == models::Capability::Admin {
-        let has_support_access = tables::UserGrant::is_authorized(
-            &snapshot.role_grants,
-            &snapshot.user_grants,
-            *user_id,
-            "estuary_support/",
-            models::Capability::Admin,
-        );
+        let has_support_access =
+            authority.is_authorized("estuary_support/", models::Capability::Admin);
 
         if !has_support_access {
             return Err(tonic::Status::permission_denied(format!(
@@ -371,9 +356,15 @@ mod tests {
             sub: user_id,
             role: "authenticated".to_string(),
             email,
+            scope_prefix: None,
         };
 
-        match evaluate_authorization(&snapshot, &claims, &collection, capability) {
+        match evaluate_authorization(
+            &snapshot,
+            &crate::Authority::resolve(&snapshot, &claims),
+            &collection,
+            capability,
+        ) {
             Ok((cordon_at, (_key, mut data_claims, broker_address, journal_name_prefix))) => {
                 // Zero out timestamps for stable snapshots.
                 data_claims.iat = 0;
diff --git a/crates/control-plane-api/src/server/authorize_user_prefix.rs b/crates/control-plane-api/src/server/authorize_user_prefix.rs
index dec72c990a2..c7a3ff4369b 100644
--- a/crates/control-plane-api/src/server/authorize_user_prefix.rs
+++ b/crates/control-plane-api/src/server/authorize_user_prefix.rs
@@ -21,7 +21,7 @@ pub async fn authorize_user_prefix(
 
     let policy_result = evaluate_authorization(
         env.snapshot(),
-        env.claims()?,
+        &env.authority()?,
         &prefix,
         &data_plane,
         capability,
@@ -61,7 +61,7 @@ pub async fn authorize_user_prefix(
 
 fn evaluate_authorization(
     snapshot: &crate::Snapshot,
-    claims: &crate::ControlClaims,
+    authority: &crate::Authority<'_>,
     prefix: &models::Prefix,
     data_plane_name: &models::Name,
     capability: models::Capability,
@@ -75,20 +75,10 @@ fn evaluate_authorization(
         String,                // Reactor address.
     ),
 )> {
-    let models::authorizations::ControlClaims {
-        sub: user_id,
-        email: user_email,
-        ..
-    } = claims;
-    let user_email = user_email.as_ref().map(String::as_str).unwrap_or("user");
-
-    if !tables::UserGrant::is_authorized(
-        &snapshot.role_grants,
-        &snapshot.user_grants,
-        *user_id,
-        prefix,
-        capability,
-    ) {
+    let user_email = authority.user_email();
+    let user_id = authority.user_id();
+
+    if !authority.is_authorized(prefix, capability) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not authorized to {prefix} for {capability:?}",
         )));
@@ -96,13 +86,8 @@ fn evaluate_authorization(
 
     // For admin capability, require that the user has a transitive role grant to estuary_support/
     if capability == models::Capability::Admin {
-        let has_support_access = tables::UserGrant::is_authorized(
-            &snapshot.role_grants,
-            &snapshot.user_grants,
-            *user_id,
-            "estuary_support/",
-            models::Capability::Admin,
-        );
+        let has_support_access =
+            authority.is_authorized("estuary_support/", models::Capability::Admin);
 
         if !has_support_access {
             return Err(tonic::Status::permission_denied(format!(
@@ -111,13 +96,7 @@ fn evaluate_authorization(
         }
     }
 
-    if !tables::UserGrant::is_authorized(
-        &snapshot.role_grants,
-        &snapshot.user_grants,
-        *user_id,
-        data_plane_name,
-        models::Capability::Read,
-    ) {
+    if !authority.is_authorized(data_plane_name, models::Capability::Read) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not authorized to {data_plane_name}",
         )));
@@ -551,9 +530,16 @@ mod tests {
             sub: user_id,
             role: "authenticated".to_string(),
             email,
+            scope_prefix: None,
         };
 
-        match evaluate_authorization(&snapshot, &claims, &prefix, &data_plane, capability) {
+        match evaluate_authorization(
+            &snapshot,
+            &crate::Authority::resolve(&snapshot, &claims),
+            &prefix,
+            &data_plane,
+            capability,
+        ) {
             Ok((
                 _cordon_at,
                 (_key, mut broker_claims, broker_address, mut reactor_claims, reactor_address),
diff --git a/crates/control-plane-api/src/server/authorize_user_task.rs b/crates/control-plane-api/src/server/authorize_user_task.rs
index 493c6060ed6..17d09602ce2 100644
--- a/crates/control-plane-api/src/server/authorize_user_task.rs
+++ b/crates/control-plane-api/src/server/authorize_user_task.rs
@@ -18,7 +18,8 @@ pub async fn authorize_user_task(
             tokens::DateTime::from_timestamp_secs(1 + started_unix as i64).unwrap_or_default();
     }
 
-    let policy_result = evaluate_authorization(env.snapshot(), env.claims()?, &task, capability);
+    let policy_result =
+        evaluate_authorization(env.snapshot(), &env.authority()?, &task, capability);
 
     // Legacy: if `started_unix` was set then use a custom 200 response for client-side retries.
     let (
@@ -66,7 +67,7 @@ pub async fn authorize_user_task(
 
 fn evaluate_authorization(
     snapshot: &crate::Snapshot,
-    claims: &crate::ControlClaims,
+    authority: &crate::Authority<'_>,
     task_name: &models::Name,
     capability: models::Capability,
 ) -> tonic::Result<(
@@ -82,20 +83,10 @@ fn evaluate_authorization(
         String,                // Shard ID prefix.
     ),
 )> {
-    let models::authorizations::ControlClaims {
-        sub: user_id,
-        email: user_email,
-        ..
-    } = claims;
-    let user_email = user_email.as_ref().map(String::as_str).unwrap_or("user");
-
-    if !tables::UserGrant::is_authorized(
-        &snapshot.role_grants,
-        &snapshot.user_grants,
-        *user_id,
-        task_name,
-        capability,
-    ) {
+    let user_email = authority.user_email();
+    let user_id = authority.user_id();
+
+    if !authority.is_authorized(task_name, capability) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not authorized to {task_name} for {capability:?}",
         )));
@@ -103,13 +94,8 @@ fn evaluate_authorization(
 
     // For admin capability, require that the user has a transitive role grant to estuary_support/
     if capability == models::Capability::Admin {
-        let has_support_access = tables::UserGrant::is_authorized(
-            &snapshot.role_grants,
-            &snapshot.user_grants,
-            *user_id,
-            "estuary_support/",
-            models::Capability::Admin,
-        );
+        let has_support_access =
+            authority.is_authorized("estuary_support/", models::Capability::Admin);
 
         if !has_support_access {
             return Err(tonic::Status::permission_denied(format!(
@@ -498,9 +484,15 @@ mod tests {
             sub: user_id,
             role: "authenticated".to_string(),
             email,
+            scope_prefix: None,
         };
 
-        match evaluate_authorization(&snapshot, &claims, &task, capability) {
+        match evaluate_authorization(
+            &snapshot,
+            &crate::Authority::resolve(&snapshot, &claims),
+            &task,
+            capability,
+        ) {
             Ok((
                 cordon_at,
                 (
diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs
index e3eaef0fe71..0e95c831216 100644
--- a/crates/control-plane-api/src/server/mod.rs
+++ b/crates/control-plane-api/src/server/mod.rs
@@ -1,4 +1,3 @@
-use crate::AuthZResult;
 use anyhow::Context;
 use axum::{http::StatusCode, response::IntoResponse};
 use std::sync::Arc;
@@ -93,74 +92,6 @@ pub(crate) async fn wake_tenant_controller(
     Ok(res.rows_affected() > 0u64)
 }
 
-/// Evaluate whether the user identified by `claims` is authorized to access all
-/// of the enumerated `prefixes_or_names` with at least `min_capability`.
-/// Return a policy_result shape which fits Envelope::authorization_outcome.
-///
-/// `min_capability` accepts any value that converts into a `CapabilitySet`:
-/// legacy `models::Capability` (mapped via `bits_for_legacy`), a single
-/// `models::authz::Capability` bit, or an explicit `CapabilitySet`.
-pub fn evaluate_names_authorization<'r, Iter, S, C>(
-    snapshot: &Snapshot,
-    claims: &crate::ControlClaims,
-    min_capability: C,
-    prefixes_or_names: Iter,
-) -> AuthZResult<()>
-where
-    Iter: IntoIterator<Item = S>,
-    S: AsRef<str> + std::fmt::Display,
-    C: Into<models::authz::CapabilitySet> + std::fmt::Display + Copy,
-{
-    let models::authorizations::ControlClaims {
-        sub: user_id,
-        email: user_email,
-        ..
-    } = claims;
-    let user_email = user_email.as_ref().map(String::as_str).unwrap_or("user");
-
-    for prefix_or_name in prefixes_or_names.into_iter() {
-        if !tables::UserGrant::is_authorized(
-            &snapshot.role_grants,
-            &snapshot.user_grants,
-            *user_id,
-            prefix_or_name.as_ref(),
-            min_capability,
-        ) {
-            return Err(tonic::Status::permission_denied(format!(
-                "{user_email} is not authorized to access prefix or name '{prefix_or_name}' with required capability {min_capability}",
-            )));
-        }
-    }
-    Ok((None, ()))
-}
-
-/// 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.
-pub fn attach_user_capabilities<I, F, T>(
-    snapshot: &Snapshot,
-    claims: &crate::ControlClaims,
-    prefixes_or_names: I,
-    mut attach: F,
-) -> Vec<T>
-where
-    I: IntoIterator<Item = String>,
-    F: FnMut(String, Option<models::Capability>) -> Option<T>,
-{
-    prefixes_or_names
-        .into_iter()
-        .flat_map(|prefix| {
-            let capability = tables::UserGrant::get_user_capability(
-                &snapshot.role_grants,
-                &snapshot.user_grants,
-                claims.sub,
-                &prefix,
-            );
-            attach(prefix, capability)
-        })
-        .collect()
-}
-
 /// Build the agent's API router.
 pub fn build_router(
     app: Arc<App>,
diff --git a/crates/control-plane-api/src/server/public/graphql/alert_configs.rs b/crates/control-plane-api/src/server/public/graphql/alert_configs.rs
index 463b64c591b..9c0d03f9068 100644
--- a/crates/control-plane-api/src/server/public/graphql/alert_configs.rs
+++ b/crates/control-plane-api/src/server/public/graphql/alert_configs.rs
@@ -121,14 +121,10 @@ impl AlertConfigsQuery {
         first: Option<i32>,
     ) -> async_graphql::Result<PaginatedAlertConfigs> {
         let env = ctx.data::<crate::Envelope>()?;
-        let claims = env.claims()?;
 
-        let snapshot = env.snapshot();
         let (read_prefixes, prefix_starts_with, prefix_in) =
             super::authorized_prefixes::filtered_authorized_prefixes(
-                &snapshot.role_grants,
-                &snapshot.user_grants,
-                claims.sub,
+                &env.authority()?,
                 models::Capability::Read,
                 filter.and_then(|f| f.catalog_prefix_or_name),
                 "filter.catalogPrefixOrName",
@@ -215,7 +211,6 @@ impl AlertConfigsQuery {
         catalog_prefix_or_name: String,
     ) -> async_graphql::Result<EffectiveAlertConfig> {
         let env = ctx.data::<crate::Envelope>()?;
-        let claims = env.claims()?;
 
         validate_prefix_or_name(&catalog_prefix_or_name)?;
 
@@ -223,9 +218,7 @@ impl AlertConfigsQuery {
         // that scope. Ancestor layers merged into the result are visible to
         // anyone who can read the scope, matching the `effective` field on
         // AlertConfigEntry and `effectiveAlertConfig` on liveSpec.
-        let policy_result = crate::server::evaluate_names_authorization(
-            env.snapshot(),
-            claims,
+        let policy_result = env.authority()?.evaluate(
             models::authz::Capability::CatalogRead,
             [catalog_prefix_or_name.as_str()],
         );
@@ -267,12 +260,9 @@ impl AlertConfigsMutation {
         validate_prefix_or_name(&catalog_prefix_or_name)?;
 
         let gov = governing_prefix(&catalog_prefix_or_name)?;
-        let policy_result = crate::server::evaluate_names_authorization(
-            env.snapshot(),
-            claims,
-            models::Capability::Admin,
-            [gov.as_str()],
-        );
+        let policy_result = env
+            .authority()?
+            .evaluate(models::Capability::Admin, [gov.as_str()]);
         env.authorization_outcome(policy_result).await?;
 
         if !catalog_prefix_or_name.ends_with('/') {
diff --git a/crates/control-plane-api/src/server/public/graphql/alerts.rs b/crates/control-plane-api/src/server/public/graphql/alerts.rs
index 8c3a784e8fd..1513a68b77e 100644
--- a/crates/control-plane-api/src/server/public/graphql/alerts.rs
+++ b/crates/control-plane-api/src/server/public/graphql/alerts.rs
@@ -139,12 +139,9 @@ async fn fetch_alert_history_by_prefix(
     let env = ctx.data::<crate::Envelope>()?;
 
     // Verify user authorization to read alerts for the given prefix.
-    let policy_result = crate::server::evaluate_names_authorization(
-        env.snapshot(),
-        env.claims()?,
-        models::Capability::Read,
-        [&by.prefix],
-    );
+    let policy_result = env
+        .authority()?
+        .evaluate(models::Capability::Read, [&by.prefix]);
     let (_expiry, ()) = env.authorization_outcome(policy_result).await?;
 
     connection::query_with::<PaginatedAlertsCursor, _, _, _, _>(
diff --git a/crates/control-plane-api/src/server/public/graphql/authorized_prefixes.rs b/crates/control-plane-api/src/server/public/graphql/authorized_prefixes.rs
index d3186d18170..e1ef2862806 100644
--- a/crates/control-plane-api/src/server/public/graphql/authorized_prefixes.rs
+++ b/crates/control-plane-api/src/server/public/graphql/authorized_prefixes.rs
@@ -1,17 +1,19 @@
-/// Returns catalog prefixes where the authenticated user has at least
-/// `min_capability`, optionally narrowed to those overlapping `prefix_filter`.
+/// Returns catalog prefixes where the caller has at least `min_capability`,
+/// optionally narrowed to those overlapping `prefix_filter`.
 ///
 /// Intended for use by GraphQL queries that list resources scoped to the
 /// caller's authorized prefixes, with an optional prefix filter.
 ///
+/// The prefixes come from the caller's [`crate::Authority`], so a scoped token's
+/// narrowing is already applied before `prefix_filter` is considered — a filter
+/// can only remove prefixes from an already-narrowed set, never restore one.
+///
 /// When `prefix_filter` is provided, a prefix is included if the filter is a
 /// sub-prefix of the grant OR the grant is a sub-prefix of the filter. This
 /// bidirectional check lets callers query with a filter that is either broader
 /// or narrower than their grants.
 pub(super) fn authorized_prefixes(
-    role_grants: &tables::RoleGrants,
-    user_grants: &tables::UserGrants,
-    user_id: uuid::Uuid,
+    authority: &crate::Authority<'_>,
     min_capability: impl Into<models::authz::CapabilitySet>,
     prefix_filter: Option<&str>,
 ) -> Vec<String> {
@@ -19,7 +21,8 @@ pub(super) fn authorized_prefixes(
 
     // BTreeMap iteration from reachable_prefixes is already prefix-sorted,
     // so the parent-prune step below can run directly on it.
-    let prefixes = tables::UserGrant::reachable_prefixes(role_grants, user_grants, user_id)
+    let prefixes = authority
+        .reachable_prefixes()
         .into_iter()
         .filter(|(prefix, _)| {
             prefix_filter.is_none_or(|pf| prefix.starts_with(pf) || pf.starts_with(*prefix))
@@ -50,9 +53,7 @@ pub(super) fn authorized_prefixes(
 /// `PrefixFilter::narrow_to_exact_set` — so the narrow-only invariant (a filter
 /// can only remove authorized prefixes, never add them) has a single owner.
 pub(super) fn filtered_authorized_prefixes(
-    role_grants: &tables::RoleGrants,
-    user_grants: &tables::UserGrants,
-    user_id: uuid::Uuid,
+    authority: &crate::Authority<'_>,
     min_capability: impl Into<models::authz::CapabilitySet>,
     filter: Option<super::filters::PrefixFilter>,
     field: &str,
@@ -61,13 +62,7 @@ pub(super) fn filtered_authorized_prefixes(
         Some(cp) => cp.into_parts(field)?,
         None => (None, None),
     };
-    let mut prefixes = authorized_prefixes(
-        role_grants,
-        user_grants,
-        user_id,
-        min_capability,
-        starts_with.as_deref(),
-    );
+    let mut prefixes = authorized_prefixes(authority, min_capability, starts_with.as_deref());
     if let Some(exact) = r#in.as_deref() {
         super::filters::PrefixFilter::narrow_to_exact_set(&mut prefixes, exact);
     }
@@ -105,6 +100,38 @@ mod tests {
 
     const ALICE: uuid::Uuid = uuid::Uuid::from_bytes([0x11; 16]);
 
+    /// An unscoped Authority over `rg`/`ug`, as a token carrying no
+    /// `scope_prefix` claim resolves to.
+    fn authority<'a>(
+        rg: &'a tables::RoleGrants,
+        ug: &'a tables::UserGrants,
+        user_id: uuid::Uuid,
+    ) -> crate::Authority<'a> {
+        crate::Authority::new(
+            rg,
+            ug,
+            user_id,
+            "alice@example.com",
+            tables::AuthScope::unscoped(),
+        )
+    }
+
+    /// An Authority over `rg`/`ug` confined to `scope`.
+    fn scoped_authority<'a>(
+        rg: &'a tables::RoleGrants,
+        ug: &'a tables::UserGrants,
+        user_id: uuid::Uuid,
+        scope: &'a str,
+    ) -> crate::Authority<'a> {
+        crate::Authority::new(
+            rg,
+            ug,
+            user_id,
+            "alice@example.com",
+            tables::AuthScope::resolve(rg, scope),
+        )
+    }
+
     #[test]
     fn no_filter_returns_all_at_or_above_capability() {
         let (ug, rg) = make_grants(
@@ -116,13 +143,13 @@ mod tests {
             &[],
         );
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Admin, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, None);
         assert_eq!(result, vec!["acmeCo/"]);
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Write, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Write, None);
         assert_eq!(result, vec!["acmeCo/", "widgets/"]);
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Read, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Read, None);
         assert_eq!(result, vec!["acmeCo/", "readonly/", "widgets/"]);
     }
 
@@ -132,7 +159,7 @@ mod tests {
         // the filter, so "acmeCo/" is included.
         let (ug, rg) = make_grants(&[(ALICE, "acmeCo/", Admin)], &[]);
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Admin, Some("acmeCo/data/"));
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, Some("acmeCo/data/"));
         assert_eq!(result, vec!["acmeCo/"]);
     }
 
@@ -142,7 +169,7 @@ mod tests {
         // with the filter, so "acmeCo/data/" is included.
         let (ug, rg) = make_grants(&[(ALICE, "acmeCo/data/", Admin)], &[]);
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Admin, Some("acmeCo/"));
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, Some("acmeCo/"));
         assert_eq!(result, vec!["acmeCo/data/"]);
     }
 
@@ -150,7 +177,7 @@ mod tests {
     fn filter_excludes_non_overlapping() {
         let (ug, rg) = make_grants(&[(ALICE, "acmeCo/", Admin), (ALICE, "other/", Admin)], &[]);
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Admin, Some("acmeCo/"));
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, Some("acmeCo/"));
         assert_eq!(result, vec!["acmeCo/"]);
     }
 
@@ -158,7 +185,7 @@ mod tests {
     fn no_grants_returns_empty() {
         let (ug, rg) = make_grants(&[], &[]);
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Admin, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, None);
         assert!(result.is_empty());
     }
 
@@ -170,11 +197,11 @@ mod tests {
             &[("acmeCo/", "shared/", Write)],
         );
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Write, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Write, None);
         assert_eq!(result, vec!["acmeCo/", "shared/"]);
 
         // Admin threshold excludes the transitive Write grant.
-        let result = authorized_prefixes(&rg, &ug, ALICE, Admin, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, None);
         assert_eq!(result, vec!["acmeCo/"]);
     }
 
@@ -187,7 +214,7 @@ mod tests {
             &[],
         );
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Admin, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, None);
         assert_eq!(result, vec!["acmeCo/"]);
     }
 
@@ -200,7 +227,7 @@ mod tests {
             &[("acmeCo/", "acmeCo/team/", Write)],
         );
 
-        let result = authorized_prefixes(&rg, &ug, ALICE, Write, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Write, None);
         assert_eq!(result, vec!["acmeCo/"]);
     }
 
@@ -209,7 +236,7 @@ mod tests {
         let bob = uuid::Uuid::from_bytes([0x22; 16]);
         let (ug, rg) = make_grants(&[(ALICE, "acmeCo/", Admin)], &[]);
 
-        let result = authorized_prefixes(&rg, &ug, bob, Read, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, bob), Read, None);
         assert!(result.is_empty());
     }
 
@@ -237,7 +264,8 @@ mod tests {
         ]);
         let rg = tables::RoleGrants::new();
 
-        let reachable = tables::UserGrant::reachable_prefixes(&rg, &ug, ALICE);
+        let reachable =
+            tables::UserGrant::reachable_prefixes(&rg, &ug, ALICE, &tables::AuthScope::unscoped());
         assert_eq!(
             reachable["acmeCo/"].0,
             CapabilityBundle::Editor.capabilities() | CapabilityBundle::TeamAdmin.capabilities(),
@@ -274,7 +302,8 @@ mod tests {
             },
         ]);
 
-        let reachable = tables::UserGrant::reachable_prefixes(&rg, &ug, ALICE);
+        let reachable =
+            tables::UserGrant::reachable_prefixes(&rg, &ug, ALICE, &tables::AuthScope::unscoped());
         assert_eq!(
             reachable["sharedCo/"].0,
             CapabilityBundle::Editor.capabilities() | CapabilityBundle::TeamAdmin.capabilities(),
@@ -308,11 +337,11 @@ mod tests {
         // child of the qualifying parent. If the union were across
         // ancestors, acmeCo/data/ would qualify on its own (Writer +
         // inherited Admin bits) — it does not.
-        let result = authorized_prefixes(&rg, &ug, ALICE, Admin, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, None);
         assert_eq!(result, vec!["acmeCo/"]);
 
         // min=Write: both qualify on their own bits; parent prunes child.
-        let result = authorized_prefixes(&rg, &ug, ALICE, Write, None);
+        let result = authorized_prefixes(&authority(&rg, &ug, ALICE), Write, None);
         assert_eq!(result, vec!["acmeCo/"]);
     }
 
@@ -320,9 +349,13 @@ mod tests {
     fn filtered_no_filter_returns_all_prefixes_and_no_parts() {
         let (ug, rg) = make_grants(&[(ALICE, "acmeCo/", Admin), (ALICE, "beta/", Admin)], &[]);
 
-        let (prefixes, starts_with, r#in) =
-            filtered_authorized_prefixes(&rg, &ug, ALICE, Admin, None, "filter.catalogPrefix")
-                .unwrap();
+        let (prefixes, starts_with, r#in) = filtered_authorized_prefixes(
+            &authority(&rg, &ug, ALICE),
+            Admin,
+            None,
+            "filter.catalogPrefix",
+        )
+        .unwrap();
         assert_eq!(prefixes, vec!["acmeCo/", "beta/"]);
         assert_eq!(starts_with, None);
         assert_eq!(r#in, None);
@@ -337,9 +370,7 @@ mod tests {
             r#in: None,
         };
         let (prefixes, starts_with, r#in) = filtered_authorized_prefixes(
-            &rg,
-            &ug,
-            ALICE,
+            &authority(&rg, &ug, ALICE),
             Admin,
             Some(filter),
             "filter.catalogPrefix",
@@ -362,9 +393,7 @@ mod tests {
             r#in: Some(vec!["acmeCo/".to_string()]),
         };
         let (prefixes, starts_with, r#in) = filtered_authorized_prefixes(
-            &rg,
-            &ug,
-            ALICE,
+            &authority(&rg, &ug, ALICE),
             Admin,
             Some(filter),
             "filter.catalogPrefix",
@@ -384,9 +413,7 @@ mod tests {
             r#in: Some(vec!["acmeCo/".to_string()]),
         };
         let err = filtered_authorized_prefixes(
-            &rg,
-            &ug,
-            ALICE,
+            &authority(&rg, &ug, ALICE),
             Admin,
             Some(filter),
             "filter.catalogPrefix",
@@ -397,4 +424,59 @@ mod tests {
             "`filter.catalogPrefix.startsWith` and `.in` are mutually exclusive; provide only one"
         );
     }
+
+    #[test]
+    fn scope_narrows_every_prefix_scoped_query() {
+        // Every list query in the API resolves its prefixes here, so a scoped
+        // token narrows all of them without any per-query work: the caller
+        // administers two tenants and a scope of one hides the other.
+        let (ug, rg) = make_grants(
+            &[(ALICE, "acmeCo/", Admin), (ALICE, "otherCo/", Admin)],
+            &[],
+        );
+
+        let unscoped = authorized_prefixes(&authority(&rg, &ug, ALICE), Admin, None);
+        assert_eq!(unscoped, vec!["acmeCo/", "otherCo/"]);
+
+        let scoped =
+            authorized_prefixes(&scoped_authority(&rg, &ug, ALICE, "acmeCo/"), Admin, None);
+        assert_eq!(scoped, vec!["acmeCo/"]);
+    }
+
+    #[test]
+    fn a_filter_cannot_widen_past_the_scope() {
+        // A caller-supplied filter is applied after the scope has already
+        // narrowed the set, so naming an out-of-scope prefix yields nothing
+        // rather than restoring access to it. Both filter modes are checked
+        // because each narrows at a different step.
+        let (ug, rg) = make_grants(
+            &[(ALICE, "acmeCo/", Admin), (ALICE, "otherCo/", Admin)],
+            &[],
+        );
+        let authority = scoped_authority(&rg, &ug, ALICE, "acmeCo/");
+
+        let (prefixes, _, _) = filtered_authorized_prefixes(
+            &authority,
+            Admin,
+            Some(PrefixFilter {
+                starts_with: Some("otherCo/".to_string()),
+                r#in: None,
+            }),
+            "filter.catalogPrefix",
+        )
+        .unwrap();
+        assert!(prefixes.is_empty());
+
+        let (prefixes, _, _) = filtered_authorized_prefixes(
+            &authority,
+            Admin,
+            Some(PrefixFilter {
+                starts_with: None,
+                r#in: Some(vec!["otherCo/".to_string()]),
+            }),
+            "filter.catalogPrefix",
+        )
+        .unwrap();
+        assert!(prefixes.is_empty());
+    }
 }
diff --git a/crates/control-plane-api/src/server/public/graphql/data_planes.rs b/crates/control-plane-api/src/server/public/graphql/data_planes.rs
index 276c344af35..3e047a42fdd 100644
--- a/crates/control-plane-api/src/server/public/graphql/data_planes.rs
+++ b/crates/control-plane-api/src/server/public/graphql/data_planes.rs
@@ -478,7 +478,7 @@ impl DataPlanesQuery {
         last: Option<i32>,
     ) -> async_graphql::Result<PaginatedDataPlanes> {
         let env = ctx.data::<crate::Envelope>()?;
-        let claims = env.claims()?;
+        let authority = env.authority()?;
         let snapshot = env.snapshot();
 
         let closed_eq = filter
@@ -496,13 +496,7 @@ impl DataPlanesQuery {
                     tracing::warn!(data_plane_name = %dp.data_plane_name, "skipping data plane with unparseable name");
                     return false;
                 }
-                tables::UserGrant::is_authorized(
-                        &snapshot.role_grants,
-                        &snapshot.user_grants,
-                        claims.sub,
-                        &dp.data_plane_name,
-                        models::Capability::Read,
-                    )
+                authority.is_authorized(&dp.data_plane_name, models::Capability::Read)
             })
             .collect();
         accessible_data_planes.sort_by(|a, b| a.data_plane_name.cmp(&b.data_plane_name));
@@ -537,7 +531,7 @@ impl DataPlanesQuery {
         // Preserve sorted order from pagination before moving into HashMap.
         let names: Vec<String> = rows.iter().map(|dp| dp.data_plane_name.clone()).collect();
 
-        // Build row data map for attach_user_capabilities.
+        // Build row data map for attach_capabilities.
         let row_data: HashMap<String, &tables::DataPlane> = rows
             .into_iter()
             .map(|dp| (dp.data_plane_name.clone(), dp))
@@ -546,11 +540,8 @@ impl DataPlanesQuery {
         // Fetch detail fields from the database for all data planes in this page.
         let details_map = fetch_data_plane_details(&env.pg_pool, &names).await?;
 
-        let edges = crate::server::attach_user_capabilities(
-            env.snapshot(),
-            env.claims()?,
-            names.into_iter(),
-            |data_plane_name, user_capability| {
+        let edges =
+            authority.attach_capabilities(names.into_iter(), |data_plane_name, user_capability| {
                 let dp = row_data.get(&data_plane_name)?;
                 let details = details_map.get(&data_plane_name);
                 let (cloud_provider, region, tag, is_public) =
@@ -584,8 +575,7 @@ impl DataPlanesQuery {
                         .unwrap_or_default(),
                 };
                 Some(connection::Edge::new(data_plane_name, node))
-            },
-        );
+            });
 
         let mut conn = PaginatedDataPlanes::new(has_prev, has_next);
         conn.edges = edges;
diff --git a/crates/control-plane-api/src/server/public/graphql/invite_links.rs b/crates/control-plane-api/src/server/public/graphql/invite_links.rs
index d11643742a1..b80b8547094 100644
--- a/crates/control-plane-api/src/server/public/graphql/invite_links.rs
+++ b/crates/control-plane-api/src/server/public/graphql/invite_links.rs
@@ -76,12 +76,9 @@ impl InviteLinksQuery {
             .as_ref()
             .and_then(|f| f.single_use.as_ref())
             .and_then(|f| f.eq);
-        let snapshot = env.snapshot();
         let (admin_prefixes, prefix_starts_with, prefix_in) =
             super::authorized_prefixes::filtered_authorized_prefixes(
-                &snapshot.role_grants,
-                &snapshot.user_grants,
-                env.claims()?.sub,
+                &env.authority()?,
                 models::Capability::Admin,
                 filter.and_then(|f| f.catalog_prefix),
                 "filter.catalogPrefix",
diff --git a/crates/control-plane-api/src/server/public/graphql/live_spec_refs.rs b/crates/control-plane-api/src/server/public/graphql/live_spec_refs.rs
index b594f96901f..08e40bcac8a 100644
--- a/crates/control-plane-api/src/server/public/graphql/live_spec_refs.rs
+++ b/crates/control-plane-api/src/server/public/graphql/live_spec_refs.rs
@@ -183,11 +183,9 @@ pub async fn paginate_live_specs_refs(
     if all_names.is_empty() {
         return Ok(connection::Connection::new(false, false));
     }
-    let all_refs = crate::server::attach_user_capabilities(
-        env.snapshot(),
-        env.claims()?,
-        all_names,
-        |name, maybe_capability| {
+    let all_refs = env
+        .authority()?
+        .attach_capabilities(all_names, |name, maybe_capability| {
             if require_min_capability.is_some_and(|min_cap| maybe_capability < Some(min_cap)) {
                 return None;
             }
@@ -195,8 +193,7 @@ pub async fn paginate_live_specs_refs(
                 catalog_name: models::Name::new(name),
                 user_capability: maybe_capability,
             })
-        },
-    );
+        });
     apply_pagination(all_refs, after, before, first, last).await
 }
 
@@ -288,9 +285,7 @@ impl LiveSpecsQuery {
         let names = names.unwrap_or_default();
 
         // Fail the entire request if it passed a name or prefix that the user is unauthorized to.
-        let policy_result = crate::server::evaluate_names_authorization(
-            env.snapshot(),
-            env.claims()?,
+        let policy_result = env.authority()?.evaluate(
             models::Capability::Read,
             names
                 .iter()
@@ -358,11 +353,9 @@ impl LiveSpecsQuery {
         // We already know that the user at least has read capability to the prefix,
         // but it's possible that they may have a greater capability to specific
         // sub-prefixes, so resolve those here.
-        let edges = crate::server::attach_user_capabilities(
-            env.snapshot(),
-            env.claims()?,
-            names,
-            |name, user_capability| {
+        let edges = env
+            .authority()?
+            .attach_capabilities(names, |name, user_capability| {
                 Some(connection::Edge::new(
                     name.clone(),
                     LiveSpecRef {
@@ -370,8 +363,7 @@ impl LiveSpecsQuery {
                         user_capability,
                     },
                 ))
-            },
-        );
+            });
 
         let mut conn = PaginatedLiveSpecsRefs::new(has_prev, has_next);
         conn.edges = edges;
diff --git a/crates/control-plane-api/src/server/public/graphql/live_specs.rs b/crates/control-plane-api/src/server/public/graphql/live_specs.rs
index 32948a0248b..145fd6d6b22 100644
--- a/crates/control-plane-api/src/server/public/graphql/live_specs.rs
+++ b/crates/control-plane-api/src/server/public/graphql/live_specs.rs
@@ -96,9 +96,7 @@ impl LiveSpec {
         let Some(source_capture_name) = &self.source_capture else {
             return Ok(None);
         };
-        let attached = crate::server::attach_user_capabilities(
-            env.snapshot(),
-            env.claims()?,
+        let attached = env.authority()?.attach_capabilities(
             [source_capture_name.clone()],
             |name, user_capability| {
                 Some(LiveSpecRef {
diff --git a/crates/control-plane-api/src/server/public/graphql/mod.rs b/crates/control-plane-api/src/server/public/graphql/mod.rs
index 9b03bb2db70..6992aa97e70 100644
--- a/crates/control-plane-api/src/server/public/graphql/mod.rs
+++ b/crates/control-plane-api/src/server/public/graphql/mod.rs
@@ -47,7 +47,7 @@ mod tenant;
 pub(crate) use scalars::Sensitive;
 
 /// Whether the current user holds `capability` on `name`, as a pure check
-/// against the request's authorization Snapshot.
+/// against the request's [`crate::Authority`].
 ///
 /// This is the visibility gate: use it to hide a field or filter a list,
 /// failing closed to an empty or default value when it returns `false`. Unlike
@@ -63,14 +63,7 @@ fn may_access(
     capability: impl Into<models::authz::CapabilitySet>,
 ) -> async_graphql::Result<bool> {
     let env = ctx.data::<crate::Envelope>()?;
-    let snapshot = env.snapshot();
-    Ok(tables::UserGrant::is_authorized(
-        &snapshot.role_grants,
-        &snapshot.user_grants,
-        env.claims()?.sub,
-        name,
-        capability,
-    ))
+    Ok(env.authority()?.is_authorized(name, capability))
 }
 
 /// Errors unless the current user holds `capability` on `prefix`.
@@ -87,12 +80,7 @@ async fn verify_authorization(
     prefix: &str,
     capability: impl Into<models::authz::CapabilitySet> + std::fmt::Display + Copy,
 ) -> async_graphql::Result<()> {
-    let policy_result = crate::server::evaluate_names_authorization(
-        env.snapshot(),
-        env.claims()?,
-        capability,
-        [prefix],
-    );
+    let policy_result = env.authority()?.evaluate(capability, [prefix]);
     let (_expiry, ()) = env.authorization_outcome(policy_result).await?;
     Ok(())
 }
diff --git a/crates/control-plane-api/src/server/public/graphql/prefixes.rs b/crates/control-plane-api/src/server/public/graphql/prefixes.rs
index 1b2e36dbd61..b6ba146d484 100644
--- a/crates/control-plane-api/src/server/public/graphql/prefixes.rs
+++ b/crates/control-plane-api/src/server/public/graphql/prefixes.rs
@@ -50,16 +50,9 @@ impl PrefixesQuery {
         let env = ctx.data::<crate::Envelope>()?;
 
         connection::query(after, None, first, None, |after, _, first, _| async move {
-            let snapshot = env.snapshot();
-            let user_id = env.claims()?.sub;
-
             let min_bits: models::authz::CapabilitySet = by.min_capability.into();
 
-            let reachable = tables::UserGrant::reachable_prefixes(
-                &snapshot.role_grants,
-                &snapshot.user_grants,
-                user_id,
-            );
+            let reachable = env.authority()?.reachable_prefixes();
             // Cursor pagination: BTreeMap::range jumps directly to the
             // first key strictly greater than the previous page's last
             // prefix, rather than iterating from the start and filtering
diff --git a/crates/control-plane-api/src/server/public/graphql/service_accounts.rs b/crates/control-plane-api/src/server/public/graphql/service_accounts.rs
index 6dc27050caa..72fb731b5bf 100644
--- a/crates/control-plane-api/src/server/public/graphql/service_accounts.rs
+++ b/crates/control-plane-api/src/server/public/graphql/service_accounts.rs
@@ -84,13 +84,10 @@ impl ServiceAccountsQuery {
     ) -> async_graphql::Result<PaginatedServiceAccounts> {
         let env = ctx.data::<crate::Envelope>()?;
 
-        let snapshot = env.snapshot();
         // Service accounts are visible to callers holding QueryServiceAccounts
         // on a prefix covering the account's catalog_name.
         let user_accessible_prefixes = super::authorized_prefixes::authorized_prefixes(
-            &snapshot.role_grants,
-            &snapshot.user_grants,
-            env.claims()?.sub,
+            &env.authority()?,
             models::authz::Capability::QueryServiceAccounts,
             None,
         );
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..95c156125c0 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
@@ -206,7 +206,6 @@ impl StorageMappingsMutation {
         spec: async_graphql::Json<models::StorageDef>,
     ) -> async_graphql::Result<CreateStorageMappingResult> {
         let env = ctx.data::<crate::Envelope>()?;
-        let claims = env.claims()?;
         let snapshot = env.snapshot();
         let async_graphql::Json(spec) = spec;
 
@@ -214,7 +213,7 @@ impl StorageMappingsMutation {
         validate_inputs(&catalog_prefix, &spec)?;
 
         // Verify user has admin capability to the catalog prefix and read capability to named data planes.
-        evaluate_authorization(env, claims, &catalog_prefix, &spec.data_planes).await?;
+        evaluate_authorization(env, &catalog_prefix, &spec.data_planes).await?;
 
         let data_planes = resolve_data_planes(&snapshot, &spec.data_planes)?;
 
@@ -346,7 +345,7 @@ impl StorageMappingsMutation {
         validate_inputs(&catalog_prefix, &spec)?;
 
         // Verify user has admin capability to the catalog prefix and read capability to named data planes.
-        evaluate_authorization(env, claims, &catalog_prefix, &spec.data_planes).await?;
+        evaluate_authorization(env, &catalog_prefix, &spec.data_planes).await?;
 
         let data_planes = resolve_data_planes(&snapshot, &spec.data_planes)?;
 
@@ -483,7 +482,6 @@ impl StorageMappingsMutation {
         spec: async_graphql::Json<models::StorageDef>,
     ) -> async_graphql::Result<ConnectionHealthTestResult> {
         let env = ctx.data::<crate::Envelope>()?;
-        let claims = env.claims()?;
         let snapshot = env.snapshot();
         let async_graphql::Json(spec) = spec;
 
@@ -491,7 +489,7 @@ impl StorageMappingsMutation {
         validate_inputs(&catalog_prefix, &spec)?;
 
         // Verify user has admin capability to the catalog prefix and read capability to named data planes.
-        evaluate_authorization(env, claims, &catalog_prefix, &spec.data_planes).await?;
+        evaluate_authorization(env, &catalog_prefix, &spec.data_planes).await?;
 
         let data_planes = resolve_data_planes(&snapshot, &spec.data_planes)?;
 
@@ -507,46 +505,36 @@ impl StorageMappingsMutation {
 
 async fn evaluate_authorization(
     env: &crate::Envelope,
-    claims: &crate::ControlClaims,
     catalog_prefix: &models::Prefix,
     data_plane_names: &[String],
 ) -> Result<(), crate::ApiError> {
-    let policy_result =
-        check_authorization(&env.snapshot(), claims, catalog_prefix, data_plane_names);
+    let policy_result = check_authorization(
+        &env.authority()?,
+        &env.snapshot().role_grants,
+        catalog_prefix,
+        data_plane_names,
+    );
     env.authorization_outcome(policy_result).await?;
     Ok(())
 }
 
 fn check_authorization(
-    snapshot: &crate::Snapshot,
-    claims: &crate::ControlClaims,
+    authority: &crate::Authority<'_>,
+    role_grants: &tables::RoleGrants,
     catalog_prefix: &models::Prefix,
     data_plane_names: &[String],
 ) -> crate::AuthZResult<()> {
-    let models::authorizations::ControlClaims {
-        sub: user_id,
-        email: user_email,
-        ..
-    } = claims;
-    let user_email = user_email.as_ref().map(String::as_str).unwrap_or("user");
-
     // Verify the User admins `catalog_prefix`.
-    if !tables::UserGrant::is_authorized(
-        &snapshot.role_grants,
-        &snapshot.user_grants,
-        *user_id,
-        catalog_prefix,
-        models::Capability::Admin,
-    ) {
-        return Err(tonic::Status::permission_denied(format!(
-            "{user_email} is not an authorized as an Admin of catalog prefix '{catalog_prefix}'",
-        )));
-    }
+    let (_cordon, ()) = authority.evaluate(models::Capability::Admin, [catalog_prefix])?;
 
     for data_plane_name in data_plane_names {
         // Verify `catalog_prefix` is authorized to access the data-plane for Read.
+        // This is a role-to-role question — may this catalog prefix use this data
+        // plane — so it is answered from the grant graph and takes no scope. The
+        // caller's own authority over `catalog_prefix` was already established
+        // above, and that check is scoped.
         if !tables::RoleGrant::is_authorized(
-            &snapshot.role_grants,
+            role_grants,
             catalog_prefix,
             data_plane_name,
             models::Capability::Read,
@@ -701,12 +689,9 @@ impl StorageMappingsQuery {
             (None, filter_catalog_prefix) => filter_catalog_prefix,
         };
 
-        let snapshot = env.snapshot();
         let (read_prefixes, under_prefix, exact_prefixes) =
             super::authorized_prefixes::filtered_authorized_prefixes(
-                &snapshot.role_grants,
-                &snapshot.user_grants,
-                env.claims()?.sub,
+                &env.authority()?,
                 models::authz::Capability::CatalogRead,
                 prefix_filter,
                 "filter.catalogPrefix",
@@ -769,23 +754,19 @@ impl StorageMappingsQuery {
             )
             .await?;
 
-        let snapshot = env.snapshot();
-        let claims = env.claims()?;
+        let authority = env.authority()?;
         let edges = rows
             .into_iter()
             .map(|row| {
-                let user_capability = tables::UserGrant::get_user_capability(
-                    &snapshot.role_grants,
-                    &snapshot.user_grants,
-                    claims.sub,
-                    &row.catalog_prefix,
-                )
-                .ok_or_else(|| {
-                    async_graphql::Error::new(format!(
-                        "missing capability for catalog prefix '{}'",
-                        row.catalog_prefix
-                    ))
-                })?;
+                let user_capability =
+                    authority
+                        .capability_at(&row.catalog_prefix)
+                        .ok_or_else(|| {
+                            async_graphql::Error::new(format!(
+                                "missing capability for catalog prefix '{}'",
+                                row.catalog_prefix
+                            ))
+                        })?;
 
                 // Strip "collection-data/" suffix from store prefixes before returning to user.
                 let user_facing_spec = strip_collection_data_suffix(row.spec);
diff --git a/crates/control-plane-api/src/server/public/open_metrics.rs b/crates/control-plane-api/src/server/public/open_metrics.rs
index 942b73812a6..ede167c7225 100644
--- a/crates/control-plane-api/src/server/public/open_metrics.rs
+++ b/crates/control-plane-api/src/server/public/open_metrics.rs
@@ -16,12 +16,9 @@ pub async fn handle_get_metrics(
         .into());
     }
 
-    let policy_result = crate::evaluate_names_authorization(
-        env.snapshot(),
-        env.claims()?,
-        models::Capability::Read,
-        [&prefix],
-    );
+    let policy_result = env
+        .authority()?
+        .evaluate(models::Capability::Read, [&prefix]);
     let (_expiry, ()) = env.authorization_outcome(policy_result).await?;
 
     // Map `started` to midnight at the open of the current month.
diff --git a/crates/control-plane-api/src/server/public/status.rs b/crates/control-plane-api/src/server/public/status.rs
index bf41fc71e87..5f11685e9fc 100644
--- a/crates/control-plane-api/src/server/public/status.rs
+++ b/crates/control-plane-api/src/server/public/status.rs
@@ -26,12 +26,7 @@ pub(crate) async fn handle_get_status(
         connected,
     }): axum_extra::extract::Query<StatusQuery>,
 ) -> Result<axum::Json<Vec<StatusResponse>>, crate::ApiError> {
-    let policy_result = crate::evaluate_names_authorization(
-        env.snapshot(),
-        env.claims()?,
-        models::Capability::Read,
-        &name,
-    );
+    let policy_result = env.authority()?.evaluate(models::Capability::Read, &name);
     let (_expiry, ()) = env.authorization_outcome(policy_result).await?;
 
     let mut require_names = name.iter().map(|s| s.as_str()).collect::<BTreeSet<_>>();
@@ -42,19 +37,11 @@ pub(crate) async fn handle_get_status(
     let status = if connected {
         // Filter out any names that the user cannot read before fetching the statuses
         let unfiltered_names = add_connected_names(&name, &env.pg_pool).await?;
-        let (snapshot, claims) = (env.snapshot(), env.claims()?);
+        let authority = env.authority()?;
 
         let filtered = unfiltered_names
             .into_iter()
-            .filter(|name| {
-                tables::UserGrant::is_authorized(
-                    &snapshot.role_grants,
-                    &snapshot.user_grants,
-                    claims.sub,
-                    name,
-                    models::Capability::Read,
-                )
-            })
+            .filter(|name| authority.is_authorized(name, models::Capability::Read))
             .collect::<Vec<_>>();
 
         fetch_status(&env.pg_pool, &filtered, short).await?
diff --git a/crates/control-plane-api/src/test_server.rs b/crates/control-plane-api/src/test_server.rs
index 7a3ca8f7cc6..8c6605f27dc 100644
--- a/crates/control-plane-api/src/test_server.rs
+++ b/crates/control-plane-api/src/test_server.rs
@@ -162,6 +162,17 @@ impl TestServer {
     /// Create a valid access token for a test user.
     /// The token includes all required claims for the server's JWT validation.
     pub fn make_access_token(&self, user_id: uuid::Uuid, email: Option<&str>) -> String {
+        self.make_scoped_access_token(user_id, email, None)
+    }
+
+    /// Create a valid access token confined to `scope_prefix`, as
+    /// `createScopedApiKey` mints for a real caller.
+    pub fn make_scoped_access_token(
+        &self,
+        user_id: uuid::Uuid,
+        email: Option<&str>,
+        scope_prefix: Option<&str>,
+    ) -> String {
         let now = tokens::now();
         let claims = models::authorizations::ControlClaims {
             iat: now.timestamp() as u64,
@@ -170,6 +181,7 @@ impl TestServer {
             role: "authenticated".to_string(),
             aud: "authenticated".to_string(),
             email: email.map(String::from),
+            scope_prefix: scope_prefix.map(String::from),
         };
 
         jsonwebtoken::encode(
diff --git a/crates/models/src/authorizations.rs b/crates/models/src/authorizations.rs
index 432dc089823..0014f05d5ec 100644
--- a/crates/models/src/authorizations.rs
+++ b/crates/models/src/authorizations.rs
@@ -20,6 +20,21 @@ pub struct ControlClaims {
     // Authorized user email, if known.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub email: Option<String>,
+    // Catalog prefix confining this token's authority, if any.
+    //
+    // Every authorization decision made with the token is intersected with the
+    // authority reachable from this prefix through role grants, so the token can
+    // only ever do less than the user could do unscoped. Named `scope_prefix`
+    // rather than `scope` to avoid colliding with the OAuth 2.0 `scope` claim,
+    // which is a space-delimited list of scope strings and means something else.
+    //
+    // The claim carries only the prefix, never a materialized list of authorized
+    // prefixes. Authority is still derived from the grant tables at request time,
+    // so revoking a grant takes effect on the next Snapshot regardless of how
+    // long the token lives; freezing the scope for a token's lifetime is safe
+    // because a scope can only narrow.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub scope_prefix: Option<String>,
 }
 
 impl ControlClaims {

From f94cd7e4e03e8d6c14db0f74bfa5efdec8d54f43 Mon Sep 17 00:00:00 2001
From: Greg Shear <greg@estuary.dev>
Date: Wed, 5 Aug 2026 16:31:18 -0400
Subject: [PATCH 3/3] authz: mint scope-confined credentials, and close the
 ways out of a scope
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds the issuance half of token scoping. `refresh_tokens` grows a `scope_prefix` column which `generate_access_token` stamps into the access token's `scope_prefix` claim, where `Authority::resolve` picks it up. Unscoped tokens emit exactly the claims they did before.

The scope lives on the token row rather than being chosen at exchange time, so whoever holds a credential cannot re-scope it. That is the property the feature exists for: a credential handed to an agent should be confined by whoever minted it, not by whoever presents it.

`createRefreshToken` takes an optional `scopePrefix`. Requesting one requires being able to read the prefix, checked against the caller's own Authority, which makes scoping non-escalating and also means a scoped caller can only mint equally or more narrowly scoped tokens — their Authority is already confined, so an out-of-scope prefix fails the same check with no separate rule. `refreshTokens` reports `scopePrefix` back so the dashboard can show what a credential is confined to.

Two paths would otherwise have let a scoped token escape completely, both because they are keyed on identity rather than on a catalog prefix:

- `createRefreshToken` with no `scopePrefix` would have minted an unscoped token. It now inherits the caller's scope: omitting the argument means "as confined as I am", not "unconfined".
- `createApiKey` gates on the service account's `catalog_name`, which is a management anchor — the account's own user_grants may reach prefixes outside the caller's scope, and the minted key would have carried that reach. A scoped caller now stamps its scope onto the key, intersecting the account's grants with the same ceiling.

The three surfaces a scope deliberately does not narrow are now marked where they live rather than left implicit: `connectors` and `alertTypes` are global reference data with no prefix to intersect, and `fetch_spec_history_no_authz` is scope-correct only through its callers. That last one is the shape worth watching for — a prefix that arrives from a row rather than being checked reads as authorized code while enforcing nothing.

Tests cover both halves of what a scope means, using the grant shape every real tenant has (`beta_onboard` gives each one `read` on `ops/dp/public/`): an unconnected tenant disappears, while the role-granted public data plane stays visible. A literal prefix match would have returned no data planes at all and broken plane selection for every scoped caller. The SQL claim-stamping is covered by pgTAP, since `sign()` needs pgjwt and the vault-held secret that `sqlx::test` databases lack.
---
 ...70d99ab231f8463061a1e0773a7eae725b349.json |  32 +++
 ...b419fcef567f2d85ca5d1d60867bd4daf79c.json} |  12 +-
 ...ba27da79d900f848c3d2195a17891b3fcb48d.json |  31 ---
 ...d431240effe02185b95683342cda765506f4b.json |  31 ---
 ...4fe082c3354488dd4d7f3212b5bd340451388.json |  32 +++
 .../src/server/public/graphql/alert_types.rs  |   4 +
 .../src/server/public/graphql/connectors.rs   |  10 +
 .../public/graphql/publication_history.rs     |   7 +
 .../server/public/graphql/refresh_tokens.rs   | 238 +++++++++++++++++-
 .../server/public/graphql/service_accounts.rs |  15 +-
 crates/flow-client/control-plane-api.graphql  |  24 +-
 .../20260805120000_scoped_refresh_tokens.sql  | 103 ++++++++
 supabase/tests/scoped_refresh_tokens.test.sql | 109 ++++++++
 13 files changed, 577 insertions(+), 71 deletions(-)
 create mode 100644 .sqlx/query-1e45c4757c3dd1dc736cbeb028570d99ab231f8463061a1e0773a7eae725b349.json
 rename .sqlx/{query-13f41829e10f47a4d8c299c2096eb76aa2c4a5fd60979a451b210833e718a4f4.json => query-4e4ad7988eada196ea1a97c4cc9bb419fcef567f2d85ca5d1d60867bd4daf79c.json} (70%)
 delete mode 100644 .sqlx/query-5b15d04217b7c6777cbcc51534eba27da79d900f848c3d2195a17891b3fcb48d.json
 delete mode 100644 .sqlx/query-b53236eb4a09b985887de632ed0d431240effe02185b95683342cda765506f4b.json
 create mode 100644 .sqlx/query-c34a42bb63d8b14a757338810b34fe082c3354488dd4d7f3212b5bd340451388.json
 create mode 100644 supabase/migrations/20260805120000_scoped_refresh_tokens.sql
 create mode 100644 supabase/tests/scoped_refresh_tokens.test.sql

diff --git a/.sqlx/query-1e45c4757c3dd1dc736cbeb028570d99ab231f8463061a1e0773a7eae725b349.json b/.sqlx/query-1e45c4757c3dd1dc736cbeb028570d99ab231f8463061a1e0773a7eae725b349.json
new file mode 100644
index 00000000000..9dd8991e393
--- /dev/null
+++ b/.sqlx/query-1e45c4757c3dd1dc736cbeb028570d99ab231f8463061a1e0773a7eae725b349.json
@@ -0,0 +1,32 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            WITH new_token AS (\n                SELECT gen_random_uuid()::text AS secret\n            )\n            INSERT INTO public.refresh_tokens\n                (user_id, multi_use, valid_for, hash, detail, created_by, scope_prefix)\n            SELECT\n                $1,\n                true,\n                v.valid_for,\n                crypt(nt.secret, gen_salt('bf')),\n                $3,\n                $4,\n                $5::text::catalog_prefix\n            FROM new_token nt, (SELECT $2::text::interval AS valid_for) v\n            WHERE v.valid_for > interval '0' AND v.valid_for <= interval '366 days'\n            RETURNING\n                id AS \"id!: models::Id\",\n                (SELECT secret FROM new_token) AS \"secret!: String\"\n            ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id!: models::Id",
+        "type_info": "Macaddr8"
+      },
+      {
+        "ordinal": 1,
+        "name": "secret!: String",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text",
+        "Text",
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "1e45c4757c3dd1dc736cbeb028570d99ab231f8463061a1e0773a7eae725b349"
+}
diff --git a/.sqlx/query-13f41829e10f47a4d8c299c2096eb76aa2c4a5fd60979a451b210833e718a4f4.json b/.sqlx/query-4e4ad7988eada196ea1a97c4cc9bb419fcef567f2d85ca5d1d60867bd4daf79c.json
similarity index 70%
rename from .sqlx/query-13f41829e10f47a4d8c299c2096eb76aa2c4a5fd60979a451b210833e718a4f4.json
rename to .sqlx/query-4e4ad7988eada196ea1a97c4cc9bb419fcef567f2d85ca5d1d60867bd4daf79c.json
index ca48eb88cdd..ae443a4fdea 100644
--- a/.sqlx/query-13f41829e10f47a4d8c299c2096eb76aa2c4a5fd60979a451b210833e718a4f4.json
+++ b/.sqlx/query-4e4ad7988eada196ea1a97c4cc9bb419fcef567f2d85ca5d1d60867bd4daf79c.json
@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n                    SELECT\n                        id AS \"id!: models::Id\",\n                        detail,\n                        created_at AS \"created_at!: chrono::DateTime<chrono::Utc>\",\n                        updated_at AS \"updated_at!: chrono::DateTime<chrono::Utc>\",\n                        multi_use AS \"multi_use!: bool\",\n                        uses AS \"uses!: i32\",\n                        (now() > updated_at + valid_for) AS \"expired!: bool\"\n                    FROM refresh_tokens\n                    WHERE user_id = $1\n                      AND valid_for <> interval '0'\n                      AND ($2::timestamptz IS NULL OR created_at < $2)\n                    ORDER BY created_at DESC\n                    LIMIT $3 + 1\n                    ",
+  "query": "\n                    SELECT\n                        id AS \"id!: models::Id\",\n                        detail,\n                        created_at AS \"created_at!: chrono::DateTime<chrono::Utc>\",\n                        updated_at AS \"updated_at!: chrono::DateTime<chrono::Utc>\",\n                        multi_use AS \"multi_use!: bool\",\n                        uses AS \"uses!: i32\",\n                        (now() > updated_at + valid_for) AS \"expired!: bool\",\n                        scope_prefix AS \"scope_prefix: String\"\n                    FROM refresh_tokens\n                    WHERE user_id = $1\n                      AND valid_for <> interval '0'\n                      AND ($2::timestamptz IS NULL OR created_at < $2)\n                    ORDER BY created_at DESC\n                    LIMIT $3 + 1\n                    ",
   "describe": {
     "columns": [
       {
@@ -37,6 +37,11 @@
         "ordinal": 6,
         "name": "expired!: bool",
         "type_info": "Bool"
+      },
+      {
+        "ordinal": 7,
+        "name": "scope_prefix: String",
+        "type_info": "Text"
       }
     ],
     "parameters": {
@@ -53,8 +58,9 @@
       false,
       true,
       true,
-      null
+      null,
+      true
     ]
   },
-  "hash": "13f41829e10f47a4d8c299c2096eb76aa2c4a5fd60979a451b210833e718a4f4"
+  "hash": "4e4ad7988eada196ea1a97c4cc9bb419fcef567f2d85ca5d1d60867bd4daf79c"
 }
diff --git a/.sqlx/query-5b15d04217b7c6777cbcc51534eba27da79d900f848c3d2195a17891b3fcb48d.json b/.sqlx/query-5b15d04217b7c6777cbcc51534eba27da79d900f848c3d2195a17891b3fcb48d.json
deleted file mode 100644
index bd7f3d7c074..00000000000
--- a/.sqlx/query-5b15d04217b7c6777cbcc51534eba27da79d900f848c3d2195a17891b3fcb48d.json
+++ /dev/null
@@ -1,31 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n            WITH new_token AS (\n                SELECT gen_random_uuid()::text AS secret\n            )\n            INSERT INTO refresh_tokens (user_id, multi_use, valid_for, hash, detail)\n            SELECT\n                $1,\n                $2,\n                v.valid_for,\n                crypt(nt.secret, gen_salt('bf')),\n                $4\n            FROM new_token nt, (SELECT $3::text::interval AS valid_for) v\n            WHERE v.valid_for > interval '0' AND v.valid_for <= interval '366 days'\n            RETURNING\n                id AS \"id!: models::Id\",\n                (SELECT secret FROM new_token) AS \"secret!: String\"\n            ",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id!: models::Id",
-        "type_info": "Macaddr8"
-      },
-      {
-        "ordinal": 1,
-        "name": "secret!: String",
-        "type_info": "Text"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Bool",
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": [
-      false,
-      null
-    ]
-  },
-  "hash": "5b15d04217b7c6777cbcc51534eba27da79d900f848c3d2195a17891b3fcb48d"
-}
diff --git a/.sqlx/query-b53236eb4a09b985887de632ed0d431240effe02185b95683342cda765506f4b.json b/.sqlx/query-b53236eb4a09b985887de632ed0d431240effe02185b95683342cda765506f4b.json
deleted file mode 100644
index b54ca3c81fc..00000000000
--- a/.sqlx/query-b53236eb4a09b985887de632ed0d431240effe02185b95683342cda765506f4b.json
+++ /dev/null
@@ -1,31 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n            WITH new_token AS (\n                SELECT gen_random_uuid()::text AS secret\n            )\n            INSERT INTO public.refresh_tokens\n                (user_id, multi_use, valid_for, hash, detail, created_by)\n            SELECT\n                $1,\n                true,\n                v.valid_for,\n                crypt(nt.secret, gen_salt('bf')),\n                $3,\n                $4\n            FROM new_token nt, (SELECT $2::text::interval AS valid_for) v\n            WHERE v.valid_for > interval '0' AND v.valid_for <= interval '366 days'\n            RETURNING\n                id AS \"id!: models::Id\",\n                (SELECT secret FROM new_token) AS \"secret!: String\"\n            ",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id!: models::Id",
-        "type_info": "Macaddr8"
-      },
-      {
-        "ordinal": 1,
-        "name": "secret!: String",
-        "type_info": "Text"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Text",
-        "Text",
-        "Uuid"
-      ]
-    },
-    "nullable": [
-      false,
-      null
-    ]
-  },
-  "hash": "b53236eb4a09b985887de632ed0d431240effe02185b95683342cda765506f4b"
-}
diff --git a/.sqlx/query-c34a42bb63d8b14a757338810b34fe082c3354488dd4d7f3212b5bd340451388.json b/.sqlx/query-c34a42bb63d8b14a757338810b34fe082c3354488dd4d7f3212b5bd340451388.json
new file mode 100644
index 00000000000..63bb0a01ce2
--- /dev/null
+++ b/.sqlx/query-c34a42bb63d8b14a757338810b34fe082c3354488dd4d7f3212b5bd340451388.json
@@ -0,0 +1,32 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            WITH new_token AS (\n                SELECT gen_random_uuid()::text AS secret\n            )\n            INSERT INTO refresh_tokens (user_id, multi_use, valid_for, hash, detail, scope_prefix)\n            SELECT\n                $1,\n                $2,\n                v.valid_for,\n                crypt(nt.secret, gen_salt('bf')),\n                $4,\n                $5::text::catalog_prefix\n            FROM new_token nt, (SELECT $3::text::interval AS valid_for) v\n            WHERE v.valid_for > interval '0' AND v.valid_for <= interval '366 days'\n            RETURNING\n                id AS \"id!: models::Id\",\n                (SELECT secret FROM new_token) AS \"secret!: String\"\n            ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id!: models::Id",
+        "type_info": "Macaddr8"
+      },
+      {
+        "ordinal": 1,
+        "name": "secret!: String",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Bool",
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "c34a42bb63d8b14a757338810b34fe082c3354488dd4d7f3212b5bd340451388"
+}
diff --git a/crates/control-plane-api/src/server/public/graphql/alert_types.rs b/crates/control-plane-api/src/server/public/graphql/alert_types.rs
index ea10fe9754c..a3a9c384b60 100644
--- a/crates/control-plane-api/src/server/public/graphql/alert_types.rs
+++ b/crates/control-plane-api/src/server/public/graphql/alert_types.rs
@@ -21,6 +21,10 @@ pub struct AlertTypeInfo {
 #[async_graphql::Object]
 impl AlertTypesQuery {
     /// Returns all possible alert types with their user-facing metadata.
+    ///
+    /// A static enumeration with no catalog prefix, so a token's scope has
+    /// nothing to narrow here. See `connectors` for the other global
+    /// reference-data surface a scope deliberately does not reach.
     async fn alert_types(&self) -> Vec<AlertTypeInfo> {
         AlertType::all()
             .iter()
diff --git a/crates/control-plane-api/src/server/public/graphql/connectors.rs b/crates/control-plane-api/src/server/public/graphql/connectors.rs
index 01437e79111..a4dff0fa04f 100644
--- a/crates/control-plane-api/src/server/public/graphql/connectors.rs
+++ b/crates/control-plane-api/src/server/public/graphql/connectors.rs
@@ -231,6 +231,11 @@ impl ConnectorsQuery {
         // Require an authenticated user, just to avoid getting spammed by
         // randos. There's no authorization checks to perform, though, as our
         // ACLs don't currently cover connectors.
+        //
+        // A token's scope therefore has no effect here: the connector catalog is
+        // global reference data with no catalog prefix to intersect against. This
+        // is one of the surfaces a scope deliberately does not narrow, along with
+        // `alertTypes` and `publicDataPlanes`.
         let env = ctx.data::<Envelope>()?;
         let _claims = env.claims()?;
         let locale: &str = env.locale.as_ref();
@@ -288,6 +293,11 @@ impl ConnectorsQuery {
         // Require an authenticated user, just to avoid getting spammed by
         // randos. There's no authorization checks to perform, though, as our
         // ACLs don't currently cover connectors.
+        //
+        // A token's scope therefore has no effect here: the connector catalog is
+        // global reference data with no catalog prefix to intersect against. This
+        // is one of the surfaces a scope deliberately does not narrow, along with
+        // `alertTypes` and `publicDataPlanes`.
         let env = ctx.data::<Envelope>()?;
         let _claims = env.claims()?;
         let locale = env.locale;
diff --git a/crates/control-plane-api/src/server/public/graphql/publication_history.rs b/crates/control-plane-api/src/server/public/graphql/publication_history.rs
index 7de64bd36cf..e6f7667e7d4 100644
--- a/crates/control-plane-api/src/server/public/graphql/publication_history.rs
+++ b/crates/control-plane-api/src/server/public/graphql/publication_history.rs
@@ -112,6 +112,13 @@ pub type SpecHistoryConnection = async_graphql::connection::Connection<
 
 /// Fetches the publication history for a given live spec, **without performing
 /// any authorization checks**.
+///
+/// The caller must have authorized `catalog_name` first. That check is where a
+/// token's scope is applied, so this function is scope-correct only by virtue of
+/// its callers — nothing here would notice an unauthorized name. This is the
+/// shape to watch for when adding resolvers: the prefix exists but arrives from a
+/// row rather than being checked, which reads as authorized code while enforcing
+/// nothing.
 pub async fn fetch_spec_history_no_authz(
     ctx: &Context<'_>,
     catalog_name: models::Name,
diff --git a/crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs b/crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs
index 420009cab6e..e2c24ee315d 100644
--- a/crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs
+++ b/crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs
@@ -18,6 +18,12 @@ pub struct RefreshTokenInfo {
     /// True once the token's validity window has elapsed
     /// (now is past `updated_at + valid_for`).
     pub expired: bool,
+    /// Catalog prefix this token is confined to, or null if it carries the
+    /// owner's full authority.
+    ///
+    /// A scoped token's authority is intersected with what this prefix reaches
+    /// through role grants, so it can only ever do less than its owner could.
+    pub scope_prefix: Option<models::Prefix>,
 }
 
 pub type PaginatedRefreshTokens = connection::Connection<
@@ -65,7 +71,8 @@ impl RefreshTokensQuery {
                         updated_at AS "updated_at!: chrono::DateTime<chrono::Utc>",
                         multi_use AS "multi_use!: bool",
                         uses AS "uses!: i32",
-                        (now() > updated_at + valid_for) AS "expired!: bool"
+                        (now() > updated_at + valid_for) AS "expired!: bool",
+                        scope_prefix AS "scope_prefix: String"
                     FROM refresh_tokens
                     WHERE user_id = $1
                       AND valid_for <> interval '0'
@@ -96,6 +103,7 @@ impl RefreshTokensQuery {
                                 multi_use: r.multi_use,
                                 uses: r.uses,
                                 expired: r.expired,
+                                scope_prefix: r.scope_prefix.map(models::Prefix::new),
                             },
                         )
                     })
@@ -110,6 +118,36 @@ impl RefreshTokensQuery {
     }
 }
 
+/// Validates a requested token scope and returns it as a `models::Prefix`.
+///
+/// A caller may only scope a token to a prefix they can read. Two things follow
+/// from checking this against the caller's own [`crate::Authority`] rather than
+/// against their raw grants:
+///
+/// - Scoping is not an escalation path. A prefix the caller cannot read is
+///   rejected, so minting a scoped token never produces authority the caller
+///   lacks. (The scope is a ceiling, not a grant — the token still derives its
+///   authority from its owner's grants — but a caller should not be able to
+///   point a credential at a namespace they cannot see.)
+/// - A scoped caller can only mint equally or more narrowly scoped tokens. Their
+///   Authority is already confined, so a prefix outside their own scope fails
+///   this check with no separate rule needed.
+async fn validate_scope_prefix(
+    env: &crate::Envelope,
+    scope_prefix: &str,
+) -> async_graphql::Result<models::Prefix> {
+    let prefix = models::Prefix::new(scope_prefix);
+    if let Err(err) = validator::Validate::validate(&prefix) {
+        return Err(async_graphql::Error::new(format!(
+            "invalid scopePrefix: {err}"
+        )));
+    }
+
+    super::verify_authorization(env, &prefix, models::authz::Capability::CatalogRead).await?;
+
+    Ok(prefix)
+}
+
 #[derive(Debug, Default)]
 pub struct RefreshTokensMutation;
 
@@ -117,6 +155,12 @@ pub struct RefreshTokensMutation;
 impl RefreshTokensMutation {
     /// Create a refresh token for the authenticated user.
     ///
+    /// Pass `scopePrefix` to confine the token to a catalog prefix. A scoped
+    /// token's authority is intersected with what that prefix reaches through
+    /// role grants, so it can only ever do less than the caller could. This is
+    /// how a credential is handed to something that should see one tenant's data
+    /// and nothing else.
+    ///
     /// Service-account callers are rejected: their API keys are administered
     /// via createApiKey and revokeApiKey.
     async fn create_refresh_token(
@@ -129,6 +173,10 @@ impl RefreshTokensMutation {
         valid_for: String,
         #[graphql(default = true)] multi_use: bool,
         #[graphql(default)] detail: Option<String>,
+        #[graphql(
+            desc = "Catalog prefix to confine the token to. The caller must be able to read it. Omit for a token carrying the caller's full authority."
+        )]
+        scope_prefix: Option<String>,
     ) -> async_graphql::Result<RefreshTokenResult> {
         let env = ctx.data::<crate::Envelope>()?;
         let claims = env.claims()?;
@@ -142,18 +190,29 @@ impl RefreshTokensMutation {
             ));
         }
 
+        let scope_prefix = match scope_prefix {
+            Some(prefix) => Some(validate_scope_prefix(env, &prefix).await?),
+            // Omitting `scopePrefix` inherits the caller's own scope rather than
+            // dropping it. Otherwise minting a token would be a complete escape
+            // from a scope: this mutation is keyed on the caller's user_id and
+            // touches no catalog prefix, so nothing else here would confine the
+            // credential it hands back.
+            None => env.authority()?.scope().prefix().map(models::Prefix::new),
+        };
+
         let row = sqlx::query!(
             r#"
             WITH new_token AS (
                 SELECT gen_random_uuid()::text AS secret
             )
-            INSERT INTO refresh_tokens (user_id, multi_use, valid_for, hash, detail)
+            INSERT INTO refresh_tokens (user_id, multi_use, valid_for, hash, detail, scope_prefix)
             SELECT
                 $1,
                 $2,
                 v.valid_for,
                 crypt(nt.secret, gen_salt('bf')),
-                $4
+                $4,
+                $5::text::catalog_prefix
             FROM new_token nt, (SELECT $3::text::interval AS valid_for) v
             WHERE v.valid_for > interval '0' AND v.valid_for <= interval '366 days'
             RETURNING
@@ -164,6 +223,7 @@ impl RefreshTokensMutation {
             multi_use,
             valid_for,
             detail.as_deref(),
+            scope_prefix.as_ref().map(models::Prefix::as_str),
         )
         .fetch_optional(&env.pg_pool)
         .await
@@ -186,6 +246,7 @@ impl RefreshTokensMutation {
         tracing::info!(
             refresh_token_id = %row.id,
             %claims.sub,
+            scope_prefix = ?scope_prefix,
             "created refresh token"
         );
 
@@ -483,4 +544,175 @@ mod test {
             .await;
         assert!(revoke_again["errors"].is_array());
     }
+
+    const ALICE: uuid::Uuid = uuid::Uuid::from_bytes([0x11; 16]);
+
+    /// Ask for a refresh token confined to `scope`, returning the raw response so
+    /// callers can assert on either data or errors.
+    async fn mint_scoped(
+        server: &test_server::TestServer,
+        token: &str,
+        scope: &str,
+    ) -> serde_json::Value {
+        server
+            .graphql(
+                &serde_json::json!({
+                    "query": r#"mutation($s: String!) {
+                        createRefreshToken(validFor: "P30D", scopePrefix: $s) { id }
+                    }"#,
+                    "variables": { "s": scope }
+                }),
+                Some(token),
+            )
+            .await
+    }
+
+    /// Covers what a `scope_prefix` claim does to a live request, and the gate on
+    /// requesting one.
+    ///
+    /// Alice administers `aliceCo/` and (added below) `otherCo/`, with a role
+    /// grant `aliceCo/ -> ops/dp/public/` for read — the same grant
+    /// `beta_onboard` gives every real tenant. That combination exercises both
+    /// halves of what a scope means: `otherCo/` disappears because nothing
+    /// connects it to `aliceCo/`, while the public data plane stays visible
+    /// because a role grant does connect it.
+    ///
+    /// Scoped tokens here are minted directly rather than by exchanging a scoped
+    /// refresh token: `generate_access_token` needs pgjwt's `sign()` and the
+    /// vault-held JWT secret, neither of which exists in the `sqlx::test` DB. The
+    /// SQL that stamps the claim is covered by `scoped_refresh_tokens.test.sql`;
+    /// what is covered here is the claim's effect once presented.
+    #[sqlx::test(
+        migrations = "../../supabase/migrations",
+        fixtures(path = "../../../fixtures", scripts("data_planes", "alice"))
+    )]
+    async fn test_scoped_token_narrows_requests(pool: sqlx::PgPool) {
+        let _guard = test_server::init();
+
+        sqlx::query!(
+            "INSERT INTO user_grants (user_id, object_role, capability) \
+             VALUES ($1, 'otherCo/', 'admin')",
+            ALICE,
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+
+        // Ungated: these are query assertions, and a gated source serves an empty
+        // first snapshot in which nobody holds any grant.
+        let server = test_server::TestServer::start(
+            pool.clone(),
+            test_server::snapshot(pool.clone(), false).await,
+        )
+        .await;
+
+        let unscoped = server.make_access_token(ALICE, Some("alice@example.com"));
+        let scoped =
+            server.make_scoped_access_token(ALICE, Some("alice@example.com"), Some("aliceCo/"));
+
+        // === A scope narrows prefix-scoped queries ===
+        let prefixes = |resp: serde_json::Value| -> Vec<String> {
+            resp["data"]["prefixes"]["edges"]
+                .as_array()
+                .expect("edges")
+                .iter()
+                .map(|e| e["node"]["prefix"].as_str().unwrap().to_string())
+                .collect()
+        };
+        let query = serde_json::json!({
+            "query": r#"query { prefixes(by: {minCapability: admin}) { edges { node { prefix } } } }"#
+        });
+
+        let all = prefixes(server.graphql(&query, Some(&unscoped)).await);
+        assert_eq!(all, vec!["aliceCo/", "otherCo/"]);
+
+        let confined = prefixes(server.graphql(&query, Some(&scoped)).await);
+        assert_eq!(
+            confined,
+            vec!["aliceCo/"],
+            "a scope of aliceCo/ hides the unconnected tenant"
+        );
+
+        // === A scope follows role grants ===
+        // `aliceCo/ -> ops/dp/public/` keeps the public data plane in scope. A
+        // scope implemented as a literal prefix match would return nothing here
+        // and break data-plane selection for every scoped caller.
+        let data_planes = serde_json::json!({
+            "query": r#"query { dataPlanes { edges { node { name } } } }"#
+        });
+        let planes: serde_json::Value = server.graphql(&data_planes, Some(&scoped)).await;
+        let names: Vec<String> = planes["data"]["dataPlanes"]["edges"]
+            .as_array()
+            .expect("edges")
+            .iter()
+            .map(|e| e["node"]["name"].as_str().unwrap().to_string())
+            .collect();
+        assert!(
+            names.iter().any(|n| n == "ops/dp/public/aws-us-west-2-c1"),
+            "the role-granted data plane stays in scope: {names:?}"
+        );
+
+        // === Requesting a scope requires being able to read it ===
+        let denied = mint_scoped(&server, &unscoped, "nobodyCo/").await;
+        assert!(
+            denied["errors"].is_array(),
+            "a prefix Alice cannot read is refused: {denied}"
+        );
+
+        // A scoped caller can only mint equally or more narrowly scoped tokens.
+        // Their own Authority is already confined, so this needs no separate rule.
+        let escalation = mint_scoped(&server, &scoped, "otherCo/").await;
+        assert!(
+            escalation["errors"].is_array(),
+            "a scoped caller cannot mint a token outside its own scope: {escalation}"
+        );
+        let narrower = mint_scoped(&server, &scoped, "aliceCo/data/").await;
+        assert!(
+            narrower["errors"].is_null(),
+            "a scoped caller can mint within its own scope: {narrower}"
+        );
+
+        // === A scoped caller that omits scopePrefix inherits its own scope ===
+        // Without this a scoped token could mint an unscoped one and escape
+        // completely, since this mutation is keyed on user_id and touches no
+        // catalog prefix.
+        let inherited: serde_json::Value = server
+            .graphql(
+                &serde_json::json!({
+                    "query": r#"mutation { createRefreshToken(validFor: "P30D", detail: "inherits") { id } }"#
+                }),
+                Some(&scoped),
+            )
+            .await;
+        assert!(
+            inherited["errors"].is_null(),
+            "minting without a scope should succeed: {inherited}"
+        );
+
+        // === Scopes are reported back on the token listing ===
+        let listed: serde_json::Value = server
+            .graphql(
+                &serde_json::json!({
+                    "query": r#"query { refreshTokens { edges { node { detail scopePrefix } } } }"#
+                }),
+                Some(&unscoped),
+            )
+            .await;
+        let scopes: Vec<(&str, Option<&str>)> = listed["data"]["refreshTokens"]["edges"]
+            .as_array()
+            .expect("edges")
+            .iter()
+            .map(|e| {
+                (
+                    e["node"]["detail"].as_str().unwrap_or(""),
+                    e["node"]["scopePrefix"].as_str(),
+                )
+            })
+            .collect();
+        assert_eq!(
+            scopes,
+            vec![("inherits", Some("aliceCo/")), ("", Some("aliceCo/data/"))],
+            "the inherited token carries the caller's scope, not no scope"
+        );
+    }
 }
diff --git a/crates/control-plane-api/src/server/public/graphql/service_accounts.rs b/crates/control-plane-api/src/server/public/graphql/service_accounts.rs
index 72fb731b5bf..10e476cfa57 100644
--- a/crates/control-plane-api/src/server/public/graphql/service_accounts.rs
+++ b/crates/control-plane-api/src/server/public/graphql/service_accounts.rs
@@ -529,6 +529,15 @@ impl ServiceAccountsMutation {
             ));
         }
 
+        // A scoped caller mints a scope-confined key. The check above only
+        // establishes authority over the account's `catalog_name`, which is a
+        // management anchor: the account's own user_grants may reach prefixes
+        // outside the caller's scope, and without this the key would carry that
+        // reach and become an escape hatch. Stamping the caller's scope onto the
+        // key intersects the account's grants with the same ceiling the caller
+        // is under.
+        let scope_prefix = env.authority()?.scope().prefix().map(String::from);
+
         // Mint the credential as a multi_use refresh token owned by the service
         // account. The lifetime is bounded to at most one year and required to
         // be positive; Postgres does the calendar-aware interval math (the WHERE
@@ -540,14 +549,15 @@ impl ServiceAccountsMutation {
                 SELECT gen_random_uuid()::text AS secret
             )
             INSERT INTO public.refresh_tokens
-                (user_id, multi_use, valid_for, hash, detail, created_by)
+                (user_id, multi_use, valid_for, hash, detail, created_by, scope_prefix)
             SELECT
                 $1,
                 true,
                 v.valid_for,
                 crypt(nt.secret, gen_salt('bf')),
                 $3,
-                $4
+                $4,
+                $5::text::catalog_prefix
             FROM new_token nt, (SELECT $2::text::interval AS valid_for) v
             WHERE v.valid_for > interval '0' AND v.valid_for <= interval '366 days'
             RETURNING
@@ -558,6 +568,7 @@ impl ServiceAccountsMutation {
             valid_for,
             detail,
             claims.sub,
+            scope_prefix.as_deref(),
         )
         .fetch_optional(&env.pg_pool)
         .await
diff --git a/crates/flow-client/control-plane-api.graphql b/crates/flow-client/control-plane-api.graphql
index fa741a9d818..eaf7aa48de7 100644
--- a/crates/flow-client/control-plane-api.graphql
+++ b/crates/flow-client/control-plane-api.graphql
@@ -1351,6 +1351,12 @@ type MutationRoot {
 	"""
 	Create a refresh token for the authenticated user.
 	
+	Pass `scopePrefix` to confine the token to a catalog prefix. A scoped
+	token's authority is intersected with what that prefix reaches through
+	role grants, so it can only ever do less than the caller could. This is
+	how a credential is handed to something that should see one tenant's data
+	and nothing else.
+	
 	Service-account callers are rejected: their API keys are administered
 	via createApiKey and revokeApiKey.
 	"""
@@ -1358,7 +1364,11 @@ type MutationRoot {
 		"""
 		ISO 8601 duration for token validity (e.g. P90D); must be greater than zero and at most one year
 		"""
-		validFor: String! = "P90D",		multiUse: Boolean! = true,		detail: String = null
+		validFor: String! = "P90D",		multiUse: Boolean! = true,		detail: String = null,
+		"""
+		Catalog prefix to confine the token to. The caller must be able to read it. Omit for a token carrying the caller's full authority.
+		"""
+		scopePrefix: String
 	): RefreshTokenResult!
 	"""
 	Revoke a refresh token owned by the authenticated user.
@@ -1832,6 +1842,10 @@ type QueryRoot {
 	effectiveAlertConfig(catalogPrefixOrName: String!): EffectiveAlertConfig!
 	"""
 	Returns all possible alert types with their user-facing metadata.
+	
+	A static enumeration with no catalog prefix, so a token's scope has
+	nothing to narrow here. See `connectors` for the other global
+	reference-data surface a scope deliberately does not reach.
 	"""
 	alertTypes: [AlertTypeInfo!]!
 	prefixes(by: PrefixesBy!, after: String, first: Int): PrefixRefConnection!
@@ -1942,6 +1956,14 @@ type RefreshTokenInfo {
 	(now is past `updated_at + valid_for`).
 	"""
 	expired: Boolean!
+	"""
+	Catalog prefix this token is confined to, or null if it carries the
+	owner's full authority.
+	
+	A scoped token's authority is intersected with what this prefix reaches
+	through role grants, so it can only ever do less than its owner could.
+	"""
+	scopePrefix: Prefix
 }
 
 type RefreshTokenInfoConnection {
diff --git a/supabase/migrations/20260805120000_scoped_refresh_tokens.sql b/supabase/migrations/20260805120000_scoped_refresh_tokens.sql
new file mode 100644
index 00000000000..3641b682d47
--- /dev/null
+++ b/supabase/migrations/20260805120000_scoped_refresh_tokens.sql
@@ -0,0 +1,103 @@
+begin;
+
+-- Lets a refresh token be confined to a catalog prefix. generate_access_token stamps
+-- the prefix into the access token's `scope_prefix` claim, and control-plane-api
+-- resolves that claim into a `tables::AuthScope` which intersects every authorization
+-- answer with the authority reachable from the prefix through role_grants.
+--
+-- The scope lives on the token row rather than being requested at exchange time so
+-- that whoever holds the credential cannot re-scope it. This matters for the case the
+-- feature exists for: a credential handed to an agent should be confined by whoever
+-- minted it, not by whoever presents it.
+--
+-- The claim carries only the prefix, never a materialized list of authorized prefixes.
+-- Authority is still derived from the grant tables per request, so revoking a grant
+-- takes effect on the next authorization Snapshot regardless of the token's remaining
+-- lifetime. Freezing the scope for the token's lifetime is safe because a scope can
+-- only narrow: a stale scope cannot authorize anything the user could not do unscoped.
+alter table public.refresh_tokens
+  add column scope_prefix public.catalog_prefix;
+
+comment on column public.refresh_tokens.scope_prefix is
+  'Optional catalog prefix stamped into the access token `scope_prefix` claim by '
+  'generate_access_token. Null yields an unscoped token. When set, every authorization '
+  'decision made with the token is intersected with the authority reachable from this '
+  'prefix through role_grants, so the token can only ever do less than its owner could.';
+
+-- Identical to the prior definition except that a non-null scope_prefix is added to the
+-- claims. Unscoped tokens (scope_prefix null) emit exactly the claims they did before.
+create or replace function public.generate_access_token(refresh_token_id public.flowid, secret text) returns json
+    language plpgsql security definer
+    as $$
+declare
+  rt refresh_tokens;
+  rt_new_secret text;
+  claims jsonb;
+  access_token text;
+begin
+
+  select * into rt from refresh_tokens where
+    refresh_tokens.id = refresh_token_id;
+
+  if not found then
+    raise 'could not find refresh_token with the given `refresh_token_id`';
+  end if;
+
+  if rt.hash <> crypt(secret, rt.hash) then
+    raise 'invalid secret provided';
+  end if;
+
+  if (rt.updated_at + rt.valid_for) < now() then
+    raise 'refresh_token has expired.';
+  end if;
+
+  claims = jsonb_build_object(
+    'exp', trunc(extract(epoch from (now() + interval '1 hour'))),
+    'iat', trunc(extract(epoch from (now()))),
+    'sub', rt.user_id,
+    'aud', 'authenticated',
+    'role', coalesce(rt.pg_role, 'authenticated')
+  );
+
+  if rt.scope_prefix is not null then
+    claims = claims || jsonb_build_object('scope_prefix', rt.scope_prefix);
+  end if;
+
+  select sign(claims::json, internal.access_token_jwt_secret()) into access_token
+  limit 1;
+
+  if rt.multi_use = false then
+    rt_new_secret = gen_random_uuid();
+    update refresh_tokens
+      set
+        hash = crypt(rt_new_secret, gen_salt('bf')),
+        uses = (uses + 1),
+        updated_at = clock_timestamp()
+      where refresh_tokens.id = rt.id;
+  else
+    -- re-set the updated_at timer so the token's validity is refreshed
+    update refresh_tokens
+      set
+        uses = (uses + 1),
+        updated_at = clock_timestamp()
+      where refresh_tokens.id = rt.id;
+  end if;
+
+  if rt_new_secret is null then
+    return json_build_object(
+      'access_token', access_token
+    );
+  else
+    return json_build_object(
+      'access_token', access_token,
+      'refresh_token', json_build_object(
+        'id', rt.id,
+        'secret', rt_new_secret
+        )
+    );
+  end if;
+commit;
+end
+$$;
+
+commit;
diff --git a/supabase/tests/scoped_refresh_tokens.test.sql b/supabase/tests/scoped_refresh_tokens.test.sql
new file mode 100644
index 00000000000..a458e43e7bc
--- /dev/null
+++ b/supabase/tests/scoped_refresh_tokens.test.sql
@@ -0,0 +1,109 @@
+-- Tests for scope-confined refresh tokens (20260805120000_scoped_refresh_tokens.sql).
+-- Covers that generate_access_token stamps a non-null scope_prefix into the
+-- `scope_prefix` claim, and that a null scope_prefix leaves the claims as they were.
+--
+-- The Rust side resolves this claim into a `tables::AuthScope`; the narrowing it
+-- performs is covered by unit tests in `crates/tables` and `control-plane-api`.
+-- What can only be checked here is that the SQL actually emits the claim, since
+-- `sign()` needs pgjwt and the vault-held JWT secret, neither of which exists in
+-- the `sqlx::test` databases the Rust tests run against.
+
+-- Decode the (unverified) claims of a JWT's payload segment. The payload is
+-- base64url, so map back to standard base64 and pad before decoding.
+create function tests.scope_jwt_claims(token text) returns jsonb as $$
+  select convert_from(
+    decode(
+      rpad(
+        translate(split_part(token, '.', 2), '-_', '+/'),
+        ((length(split_part(token, '.', 2)) + 3) / 4) * 4,
+        '='
+      ),
+      'base64'
+    ),
+    'utf8'
+  )::jsonb;
+$$ language sql;
+
+
+create function tests.test_access_token_omits_scope_when_unscoped()
+returns setof text as $$
+declare
+  rt_response jsonb;
+  response json;
+  claims jsonb;
+begin
+  delete from refresh_tokens;
+
+  -- A refresh token created the normal way has a null scope_prefix, so the
+  -- emitted claims must carry no `scope_prefix` at all: an absent claim is what
+  -- the Rust side reads as "unscoped", and an explicit null would be a distinct
+  -- (and unhandled) shape.
+  perform set_authenticated_context('11111111-1111-1111-1111-111111111111');
+  select create_refresh_token(true, '1 day', 'unscoped') into rt_response;
+  select generate_access_token((rt_response->>'id')::flowid, rt_response->>'secret') into response;
+
+  -- scope_jwt_claims lives in the `tests` schema, which `authenticated` cannot reach.
+  set role postgres;
+  claims := tests.scope_jwt_claims(response->>'access_token');
+
+  return query select ok(not (claims ? 'scope_prefix'), 'unscoped token carries no scope_prefix claim');
+  return query select is(claims->>'role', 'authenticated', 'role claim unchanged');
+  return query select is(claims->>'aud', 'authenticated', 'aud claim unchanged');
+  return query select is(claims->>'sub', '11111111-1111-1111-1111-111111111111', 'sub is the token user');
+end;
+$$ language plpgsql;
+
+
+create function tests.test_access_token_carries_scope_prefix()
+returns setof text as $$
+declare
+  rt refresh_tokens;
+  rt_response jsonb;
+  response json;
+  claims jsonb;
+begin
+  delete from refresh_tokens;
+
+  perform set_authenticated_context('11111111-1111-1111-1111-111111111111');
+  select create_refresh_token(true, '1 day', 'scoped') into rt_response;
+
+  -- The GraphQL mutation sets this column at insert time, after checking that the
+  -- caller can read the prefix. Set it directly here: this test covers the claim
+  -- stamping, not the mutation's authorization gate.
+  set role postgres;
+  update refresh_tokens set scope_prefix = 'aliceCo/' where id = (rt_response->>'id')::flowid;
+
+  perform set_authenticated_context('11111111-1111-1111-1111-111111111111');
+  select generate_access_token((rt_response->>'id')::flowid, rt_response->>'secret') into response;
+
+  set role postgres;
+  claims := tests.scope_jwt_claims(response->>'access_token');
+
+  return query select is(claims->>'scope_prefix', 'aliceCo/', 'access token carries the scope_prefix claim');
+  return query select is(claims->>'sub', '11111111-1111-1111-1111-111111111111', 'sub is still the token user');
+  return query select is(claims->>'role', 'authenticated', 'a scope does not change the Postgres role');
+
+  -- The scope rides on the row, so every exchange of this credential is scoped:
+  -- whoever holds the secret cannot mint an unscoped token from it.
+  select * into rt from refresh_tokens where id = (rt_response->>'id')::flowid;
+  return query select is(rt.scope_prefix::text, 'aliceCo/', 'scope_prefix persists across exchange');
+end;
+$$ language plpgsql;
+
+
+create function tests.test_scope_prefix_rejects_malformed_prefix()
+returns setof text as $$
+begin
+  delete from refresh_tokens;
+
+  -- The column is a catalog_prefix, so the domain rejects a name that is not a
+  -- prefix (no trailing slash) even if the application layer were to miss it.
+  set role postgres;
+  prepare bad_prefix as
+    insert into refresh_tokens (user_id, multi_use, valid_for, hash, scope_prefix)
+    values ('11111111-1111-1111-1111-111111111111', true, '1 day', 'x', 'aliceCo');
+  return query select throws_ok('bad_prefix', '23514',
+    null, 'catalog_prefix domain rejects a non-prefix scope');
+  deallocate bad_prefix;
+end;
+$$ language plpgsql;