diff --git a/Cargo.lock b/Cargo.lock
index ca91a240a1..12ce6399d1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4298,6 +4298,7 @@ dependencies = [
  "tempfile",
  "thiserror 2.0.21",
  "tokio",
+ "toml",
  "utoipa",
  "uuid",
 ]
diff --git a/benchmarks/duckshop/edge_cases/orchestration/custom_role_publication.json b/benchmarks/duckshop/edge_cases/orchestration/custom_role_publication.json
index 482290cc7d..3e24af5e8a 100644
--- a/benchmarks/duckshop/edge_cases/orchestration/custom_role_publication.json
+++ b/benchmarks/duckshop/edge_cases/orchestration/custom_role_publication.json
@@ -1,8 +1,8 @@
 {
-  "description": "A support role is published, widened in a different action order, narrowed, then removed. Every publication takes effect at once.",
+  "description": "A support role is published, widened in a different permission order over the tag:orders selector, narrowed, then removed. Every publication takes effect at once.",
   "publications": [
-    {"actions": ["view"], "expected": {"view": "orders_support", "job.run": null}},
-    {"actions": ["job.run", "view"], "expected": {"view": "orders_support", "job.run": "orders_support"}},
-    {"actions": ["job.run"], "expected": {"view": null, "job.run": "orders_support"}}
+    {"permissions": ["view"], "expected": {"view": "orders_support", "job.run": null}},
+    {"permissions": ["job.run", "view"], "expected": {"view": "orders_support", "job.run": "orders_support"}},
+    {"permissions": ["job.run"], "expected": {"view": null, "job.run": "orders_support"}}
   ]
 }
diff --git a/benchmarks/duckshop/edge_cases/orchestration/project_roles.toml b/benchmarks/duckshop/edge_cases/orchestration/project_roles.toml
new file mode 100644
index 0000000000..be64548f96
--- /dev/null
+++ b/benchmarks/duckshop/edge_cases/orchestration/project_roles.toml
@@ -0,0 +1,11 @@
+# Project roles: one grant follows tag coverage across publications; one intersection selects nothing yet.
+[roles.orders_support]
+description = "Answer order questions"
+
+[[roles.orders_support.grants]]
+nodes = ["tag:support"]
+permissions = ["view.sql"]
+
+[[roles.orders_support.grants]]
+nodes = ["tag:support,tag:pii"]
+permissions = ["view.evidence"]
diff --git a/crates/sqlbuild-orchestration/Cargo.toml b/crates/sqlbuild-orchestration/Cargo.toml
index 94170f1655..a2cbd1322e 100644
--- a/crates/sqlbuild-orchestration/Cargo.toml
+++ b/crates/sqlbuild-orchestration/Cargo.toml
@@ -29,6 +29,7 @@ sqlx-postgres.workspace = true
 subtle.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
+toml.workspace = true
 utoipa.workspace = true
 uuid.workspace = true
 
diff --git a/crates/sqlbuild-orchestration/src/authorization/_helpers/custom_roles.rs b/crates/sqlbuild-orchestration/src/authorization/_helpers/custom_roles.rs
index e15fb43cdc..bfb34a400f 100644
--- a/crates/sqlbuild-orchestration/src/authorization/_helpers/custom_roles.rs
+++ b/crates/sqlbuild-orchestration/src/authorization/_helpers/custom_roles.rs
@@ -1,11 +1,16 @@
-//! Published custom roles grant only non-administrative actions.
+//! Published project roles take effect at once; their node coverage follows each snapshot.
 
 use std::collections::{BTreeMap, BTreeSet};
 
 use serde_json::Value;
+use sqlbuild_recall::records::main::sha256_hex::sha256_hex;
+use uuid::Uuid;
 
-use crate::credentials::models::{Action, RoleDefinition};
+use crate::credentials::models::{Action, PermissionScope, RoleDefinition, RoleGrant};
 use crate::store::errors::ControlError;
+use crate::store::main::query::query;
+use crate::store::models::{ControlTransaction, Parameter};
+use crate::store::types::Row;
 
 const ADMINISTRATIVE: [Action; 3] = [Action::UserManage, Action::RoleAssign, Action::AuditView];
 
@@ -18,9 +23,21 @@ fn action_name(action: Action) -> Result<String, ControlError> {
     }
 }
 
+fn valid_grant(grant: &RoleGrant) -> bool {
+    let blank_selector = grant
+        .nodes
+        .iter()
+        .any(|selector| selector.trim().is_empty());
+    let administrative = grant
+        .permissions
+        .iter()
+        .any(|action| ADMINISTRATIVE.contains(action));
+    !grant.nodes.is_empty() && !blank_selector && !grant.permissions.is_empty() && !administrative
+}
+
 pub(crate) fn decode_roles(
     document: Option<&Value>,
-) -> Result<BTreeMap<String, BTreeSet<String>>, ControlError> {
+) -> Result<BTreeMap<String, Vec<RoleGrant>>, ControlError> {
     let roles: BTreeMap<String, RoleDefinition> = document.map_or_else(
         || Ok(BTreeMap::new()),
         |document| {
@@ -28,37 +45,135 @@ pub(crate) fn decode_roles(
                 .map_err(|error| ControlError::Invalid(format!("roles: {error}")))
         },
     )?;
-    let mut published: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
+    let mut published: BTreeMap<String, Vec<RoleGrant>> = BTreeMap::new();
     for (key, definition) in roles {
         if key.trim().is_empty()
             || BUILTIN.contains(&key.to_lowercase().as_str())
             || definition.description.trim().is_empty()
-            || definition
-                .actions
-                .iter()
-                .any(|action| ADMINISTRATIVE.contains(action))
+            || definition.grants.is_empty()
+            || !definition.grants.iter().all(valid_grant)
         {
             return Err(ControlError::Invalid(format!(
-                "role {key} needs a description, a key that is not a built-in role, and only non-administrative actions"
+                "role {key} needs a description, a key that is not a built-in role, and grants of node selectors with only non-administrative permissions"
             )));
         }
-        let mut grants: BTreeSet<String> = BTreeSet::new();
-        for action in definition.actions {
-            grants.insert(action_name(action)?);
-        }
+        let mut grants: Vec<RoleGrant> = definition.grants;
+        grants.sort();
+        grants.dedup();
         published.insert(key, grants);
     }
     Ok(published)
 }
 
-pub(crate) fn decode_grants(value: &Value) -> Result<BTreeSet<String>, ControlError> {
+pub(crate) fn decode_grants(value: &Value) -> Result<Vec<RoleGrant>, ControlError> {
     serde_json::from_str(value.as_str().unwrap_or("[]"))
         .map_err(|error| ControlError::Schema(format!("effective role grants: {error}")))
 }
 
-pub(crate) fn grants_allow(value: &Value, action: Action) -> bool {
-    match (decode_grants(value), action_name(action)) {
-        (Ok(grants), Ok(name)) => grants.contains(&name),
-        _ => false,
+/// Nodes on which these effective grants allow the action; a grant that resolved to none gives nothing.
+pub(crate) fn granted_nodes(value: &Value, action: Action) -> BTreeSet<String> {
+    let mut nodes: BTreeSet<String> = BTreeSet::new();
+    for grant in decode_grants(value).unwrap_or_default() {
+        if !grant.permissions.contains(&action) {
+            continue;
+        }
+        for selector in &grant.nodes {
+            nodes.extend(
+                grant
+                    .resolved_nodes
+                    .get(selector)
+                    .cloned()
+                    .unwrap_or_default(),
+            );
+        }
     }
+    nodes
+}
+
+/// Custom roles whose effective grants add coverage, and requested nodes none covers; built-ins have no grants.
+pub(crate) fn custom_coverage(
+    rows: &[&Row],
+    scope: &PermissionScope,
+) -> (Vec<String>, Vec<String>) {
+    let mut covered: BTreeSet<String> = BTreeSet::new();
+    let mut granting: Vec<String> = Vec::new();
+    for row in rows {
+        let role_key = row["role_key"].as_str().unwrap_or_default();
+        let nodes = granted_nodes(&row["resolved_grants_json"], scope.action);
+        let adds = if scope.nodes.is_empty() {
+            !nodes.is_empty() && granting.is_empty()
+        } else {
+            scope
+                .nodes
+                .iter()
+                .any(|node| nodes.contains(node) && !covered.contains(node))
+        };
+        if !adds {
+            continue;
+        }
+        if !granting.iter().any(|key| key == role_key) {
+            granting.push(role_key.into());
+        }
+        covered.extend(nodes.into_iter().filter(|node| scope.nodes.contains(node)));
+    }
+    let missing: Vec<String> = scope.nodes.difference(&covered).cloned().collect();
+    (granting, missing)
+}
+
+/// A denial names at most twenty uncovered nodes so the audited reason stays bounded.
+pub(crate) fn denial_reason(action: Action, missing: &[String]) -> Result<String, ControlError> {
+    if missing.is_empty() {
+        return Ok("missing permission in this scope; all requested nodes denied".into());
+    }
+    let listed: Vec<&str> = missing.iter().take(20).map(String::as_str).collect();
+    let more = missing.len() - listed.len();
+    let suffix = if more > 0 {
+        format!(" and {more} more")
+    } else {
+        String::new()
+    };
+    Ok(format!(
+        "missing {} on nodes {}{suffix}; all requested nodes denied",
+        action_name(action)?,
+        listed.join(", ")
+    ))
+}
+
+fn grants_digest(grants: Option<&[RoleGrant]>) -> Result<Option<String>, ControlError> {
+    let Some(grants) = grants else {
+        return Ok(None);
+    };
+    let text =
+        serde_json::to_string(grants).map_err(|error| ControlError::Invalid(error.to_string()))?;
+    Ok(Some(sha256_hex(text.as_bytes())))
+}
+
+/// Append a role's coverage change to the snapshot's change feed, after its node changes.
+pub(crate) async fn record_role_change(
+    transaction: ControlTransaction,
+    (snapshot_id, role_key): (Uuid, &str),
+    (before, after): (Option<&[RoleGrant]>, Option<&[RoleGrant]>),
+) -> Result<ControlTransaction, ControlError> {
+    let kind = match (before, after) {
+        (None, _) => "added",
+        (Some(_), Some(_)) => "changed",
+        (Some(_), None) => "removed",
+    };
+    let (transaction, rows) = query(
+        transaction,
+        "SELECT COALESCE(MAX(change_seq) + 1, 0) AS next FROM sqb_snapshot_changes WHERE snapshot_id=$1",
+        vec![Parameter::Id(snapshot_id)],
+    )
+    .await?;
+    let sequence = rows
+        .first()
+        .and_then(|row| row["next"].as_i64())
+        .unwrap_or_default();
+    let (transaction, _) = query(transaction,
+        "INSERT INTO sqb_snapshot_changes (snapshot_id, change_seq, change_kind, subject_key, before_hash, after_hash) \
+         VALUES ($1,$2,$3,$4,$5,$6)",
+        vec![Parameter::Id(snapshot_id), Parameter::Integer(sequence), Parameter::Text(kind.into()),
+            Parameter::Text(format!("role:{role_key}")), Parameter::OptionalText(grants_digest(before)?),
+            Parameter::OptionalText(grants_digest(after)?)]).await?;
+    Ok(transaction)
 }
diff --git a/crates/sqlbuild-orchestration/src/authorization/main/authorize_listing.rs b/crates/sqlbuild-orchestration/src/authorization/main/authorize_listing.rs
new file mode 100644
index 0000000000..1801ec9f17
--- /dev/null
+++ b/crates/sqlbuild-orchestration/src/authorization/main/authorize_listing.rs
@@ -0,0 +1,34 @@
+//! A location-wide listing needs view coverage of every node in the current snapshot; it is never filtered.
+
+use std::collections::BTreeSet;
+
+use uuid::Uuid;
+
+use crate::authorization::main::authorize_person::authorize_person;
+use crate::credentials::models::{Action, AuditContext, DefaultRole, PermissionScope};
+use crate::store::errors::ControlError;
+use crate::store::main::query::query;
+use crate::store::models::{ControlTransaction, Parameter};
+
+pub(crate) async fn authorize_listing(
+    transaction: ControlTransaction,
+    credentials: (&AuditContext, Uuid),
+    (location_id, target): (Uuid, &str),
+    default_role: DefaultRole,
+) -> Result<(ControlTransaction, AuditContext), ControlError> {
+    let (transaction, rows) = query(transaction,
+        "SELECT n.node_key FROM sqb_snapshot_nodes n JOIN sqb_current_snapshots c ON c.snapshot_id=n.snapshot_id \
+         WHERE c.location_id=$1 AND c.target=$2",
+        vec![Parameter::Id(location_id), Parameter::Text(target.into())]).await?;
+    let mut nodes: BTreeSet<String> = BTreeSet::new();
+    for row in &rows {
+        nodes.insert(row["node_key"].as_str().unwrap_or_default().into());
+    }
+    let scope = PermissionScope {
+        action: Action::View,
+        location_id: Some(location_id),
+        target: Some(target.into()),
+        nodes,
+    };
+    authorize_person(transaction, credentials, &scope, default_role).await
+}
diff --git a/crates/sqlbuild-orchestration/src/authorization/main/authorize_person.rs b/crates/sqlbuild-orchestration/src/authorization/main/authorize_person.rs
index 1eec4aadf5..9489453bbf 100644
--- a/crates/sqlbuild-orchestration/src/authorization/main/authorize_person.rs
+++ b/crates/sqlbuild-orchestration/src/authorization/main/authorize_person.rs
@@ -1,14 +1,15 @@
-//! A live person's assignments and live group roles are evaluated under the operation's lock.
+//! Live identity and the union of assigned and group role grants decide every node under lock.
 
 use serde_json::json;
 use uuid::Uuid;
 
-use crate::authorization::_helpers::custom_roles::grants_allow;
+use crate::authorization::_helpers::custom_roles::{custom_coverage, denial_reason};
 use crate::authorization::main::deny_authorization::deny;
 use crate::credentials::models::{Action, AuditContext, BuiltinRole, DefaultRole, PermissionScope};
 use crate::store::errors::ControlError;
 use crate::store::main::query::query;
 use crate::store::models::{ControlTransaction, Parameter};
+use crate::store::types::Row;
 
 fn role_allows(role: BuiltinRole, action: Action) -> bool {
     use Action::{
@@ -67,32 +68,34 @@ pub(crate) async fn authorize_person(
          LEFT JOIN sqb_effective_roles e ON e.location_id=r.location_id AND e.role_key=r.role_key AND e.target=$2 \
          ORDER BY r.role_key,r.group_name",
         vec![Parameter::Id(principal_id), Parameter::OptionalText(scope.target.clone())]).await?;
-    for row in rows {
-        let allowed = serde_json::from_value::<BuiltinRole>(row["role_key"].clone()).map_or_else(
-            |_| grants_allow(&row["resolved_grants_json"], scope.action),
-            |role| role_allows(role, scope.action),
-        );
-        if allowed
-            && (row["location_id"].is_null() || row["location_id"] == json!(scope.location_id))
-            && (row["target"].is_null() || row["target"] == json!(scope.target))
-        {
-            let role = row["role_key"].as_str().unwrap_or_default();
-            context.granting_role = match row["group_name"].as_str() {
-                Some(group) => format!("group:{group}:{role}"),
-                None => role.into(),
-            };
-            return Ok((transaction, context));
-        }
+    let rows: Vec<&Row> = rows
+        .iter()
+        .filter(|row| {
+            (row["location_id"].is_null() || row["location_id"] == json!(scope.location_id))
+                && (row["target"].is_null() || row["target"] == json!(scope.target))
+        })
+        .collect();
+    let builtin = rows.iter().find(|row| {
+        serde_json::from_value::<BuiltinRole>(row["role_key"].clone())
+            .is_ok_and(|role| role_allows(role, scope.action))
+    });
+    if let Some(row) = builtin {
+        let role = row["role_key"].as_str().unwrap_or_default();
+        context.granting_role = match row["group_name"].as_str() {
+            Some(group) => format!("group:{group}:{role}"),
+            None => role.into(),
+        };
+        return Ok((transaction, context));
+    }
+    let (granting, missing) = custom_coverage(&rows, scope);
+    if !granting.is_empty() && missing.is_empty() {
+        context.granting_role = granting.join("+");
+        return Ok((transaction, context));
     }
     if default_role == DefaultRole::Viewer && role_allows(BuiltinRole::Viewer, scope.action) {
         context.granting_role = "default:viewer".into();
         return Ok((transaction, context));
     }
-    deny(
-        transaction,
-        &context,
-        json!(scope),
-        "missing permission in this scope; all requested nodes denied",
-    )
-    .await
+    let reason = denial_reason(scope.action, &missing)?;
+    deny(transaction, &context, json!(scope), &reason).await
 }
diff --git a/crates/sqlbuild-orchestration/src/authorization/main/compile_roles.rs b/crates/sqlbuild-orchestration/src/authorization/main/compile_roles.rs
new file mode 100644
index 0000000000..b42306f392
--- /dev/null
+++ b/crates/sqlbuild-orchestration/src/authorization/main/compile_roles.rs
@@ -0,0 +1,40 @@
+//! A project's `roles.toml` becomes the snapshot `roles` document, its selectors resolved per snapshot.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use serde_json::Value;
+
+use crate::authorization::_helpers::custom_roles::decode_roles;
+use crate::credentials::models::RolesFile;
+
+/// The `roles` document and one warning per grant whose selectors match no node in this snapshot.
+pub fn compile_roles(
+    text: &str,
+    resolve: impl Fn(&str) -> Result<BTreeSet<String>, String>,
+) -> Result<(Value, Vec<String>), String> {
+    let file: RolesFile = toml::from_str(text).map_err(|error| format!("roles.toml: {error}"))?;
+    let mut roles = file.roles;
+    let mut warnings: Vec<String> = Vec::new();
+    for (key, role) in &mut roles {
+        for (index, grant) in role.grants.iter_mut().enumerate() {
+            let mut resolved: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
+            for selector in &grant.nodes {
+                let nodes = resolve(selector).map_err(|error| {
+                    format!("roles.toml: role {key} selector '{selector}': {error}")
+                })?;
+                resolved.insert(selector.clone(), nodes);
+            }
+            if resolved.values().all(BTreeSet::is_empty) {
+                warnings.push(format!(
+                    "role {key} grant {} selects no nodes ({}); it grants nothing in this snapshot",
+                    index + 1,
+                    grant.nodes.iter().cloned().collect::<Vec<_>>().join(" ")
+                ));
+            }
+            grant.resolved_nodes = resolved;
+        }
+    }
+    let document = serde_json::to_value(&roles).map_err(|error| error.to_string())?;
+    decode_roles(Some(&document)).map_err(|error| format!("roles.toml: {error}"))?;
+    Ok((document, warnings))
+}
diff --git a/crates/sqlbuild-orchestration/src/authorization/main/mod.rs b/crates/sqlbuild-orchestration/src/authorization/main/mod.rs
index cfccfc5065..6ae84eb147 100644
--- a/crates/sqlbuild-orchestration/src/authorization/main/mod.rs
+++ b/crates/sqlbuild-orchestration/src/authorization/main/mod.rs
@@ -1,8 +1,10 @@
 pub mod assign_builtin_role;
+pub(crate) mod authorize_listing;
 pub(crate) mod authorize_person;
 pub(crate) mod begin_authorized;
 pub mod bootstrap_user;
 pub mod change_administrator;
+pub mod compile_roles;
 pub(crate) mod deny_authorization;
 pub mod grant_admin;
 pub(crate) mod publish_custom_roles;
diff --git a/crates/sqlbuild-orchestration/src/authorization/main/publish_custom_roles.rs b/crates/sqlbuild-orchestration/src/authorization/main/publish_custom_roles.rs
index 7e0fe23386..31af84635f 100644
--- a/crates/sqlbuild-orchestration/src/authorization/main/publish_custom_roles.rs
+++ b/crates/sqlbuild-orchestration/src/authorization/main/publish_custom_roles.rs
@@ -1,8 +1,13 @@
-//! Each published `roles` document replaces the target's custom roles, which take effect at once.
+//! Each published `roles` document replaces the target's custom roles at once; coverage changes enter the change feed.
+
+use std::collections::{BTreeMap, BTreeSet};
 
 use serde_json::Value;
 
-use crate::authorization::_helpers::custom_roles::decode_roles;
+use crate::authorization::_helpers::custom_roles::{
+    decode_grants, decode_roles, record_role_change,
+};
+use crate::credentials::models::RoleGrant;
 use crate::snapshots::models::SnapshotRecord;
 use crate::store::errors::ControlError;
 use crate::store::main::query::query;
@@ -14,13 +19,40 @@ pub(crate) async fn publish_custom_roles(
     document: Option<&Value>,
 ) -> Result<ControlTransaction, ControlError> {
     let roles = decode_roles(document)?;
+    let scope = vec![
+        Parameter::Id(record.location_id),
+        Parameter::Text(record.target.clone()),
+    ];
+    let rows;
+    (transaction, rows) = query(
+        transaction,
+        "SELECT role_key,resolved_grants_json FROM sqb_effective_roles WHERE location_id=$1 AND target=$2",
+        scope.clone(),
+    )
+    .await?;
+    let mut previous: BTreeMap<String, Vec<RoleGrant>> = BTreeMap::new();
+    for row in &rows {
+        previous.insert(
+            row["role_key"].as_str().unwrap_or_default().into(),
+            decode_grants(&row["resolved_grants_json"])?,
+        );
+    }
+    let keys: BTreeSet<&String> = previous.keys().chain(roles.keys()).collect();
+    for key in keys {
+        let (before, after) = (previous.get(key), roles.get(key));
+        if before != after {
+            transaction = record_role_change(
+                transaction,
+                (record.snapshot_id, key),
+                (before.map(Vec::as_slice), after.map(Vec::as_slice)),
+            )
+            .await?;
+        }
+    }
     (transaction, _) = query(
         transaction,
         "DELETE FROM sqb_effective_roles WHERE location_id=$1 AND target=$2",
-        vec![
-            Parameter::Id(record.location_id),
-            Parameter::Text(record.target.clone()),
-        ],
+        scope,
     )
     .await?;
     for (key, grants) in roles {
diff --git a/crates/sqlbuild-orchestration/src/credentials/models.rs b/crates/sqlbuild-orchestration/src/credentials/models.rs
index c1c25b23c8..fc0a74344e 100644
--- a/crates/sqlbuild-orchestration/src/credentials/models.rs
+++ b/crates/sqlbuild-orchestration/src/credentials/models.rs
@@ -71,7 +71,9 @@ impl Default for AuditRetention {
 }
 
 /// Closed permission vocabulary; unknown actions fail deserialization.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
+#[derive(
+    Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, utoipa::ToSchema,
+)]
 pub enum Action {
     #[serde(rename = "view")]
     View,
@@ -126,12 +128,30 @@ impl From<BuiltinRole> for RoleKey {
     }
 }
 
-/// One custom role entry of a published `roles` document; only non-administrative actions.
-#[derive(Clone, Debug, Deserialize)]
+/// One grant of a project role: node selectors, their permissions, and each selector's nodes.
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct RoleGrant {
+    pub nodes: std::collections::BTreeSet<String>,
+    pub permissions: std::collections::BTreeSet<Action>,
+    #[serde(default)]
+    pub resolved_nodes: std::collections::BTreeMap<String, std::collections::BTreeSet<String>>,
+}
+
+/// One project role of a `roles.toml` file or a published `roles` document.
+#[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
-pub(crate) struct RoleDefinition {
-    pub(crate) description: String,
-    pub(crate) actions: Vec<Action>,
+pub struct RoleDefinition {
+    pub description: String,
+    pub grants: Vec<RoleGrant>,
+}
+
+/// The reviewed project-root `roles.toml` file.
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct RolesFile {
+    #[serde(default)]
+    pub(crate) roles: std::collections::BTreeMap<String, RoleDefinition>,
 }
 
 /// Installation-owned ambient access; never permits an administrative or execution default.
diff --git a/crates/sqlbuild-orchestration/src/declarations/_helpers/mod.rs b/crates/sqlbuild-orchestration/src/declarations/_helpers/mod.rs
index 1e3bd2943f..90ce175a05 100644
--- a/crates/sqlbuild-orchestration/src/declarations/_helpers/mod.rs
+++ b/crates/sqlbuild-orchestration/src/declarations/_helpers/mod.rs
@@ -3,4 +3,5 @@ pub(crate) mod documents;
 pub(crate) mod operator_change;
 pub(crate) mod operator_state;
 pub(crate) mod publication;
+pub(crate) mod trigger_scope;
 pub(crate) mod trigger_state;
diff --git a/crates/sqlbuild-orchestration/src/declarations/_helpers/trigger_scope.rs b/crates/sqlbuild-orchestration/src/declarations/_helpers/trigger_scope.rs
new file mode 100644
index 0000000000..00b9691698
--- /dev/null
+++ b/crates/sqlbuild-orchestration/src/declarations/_helpers/trigger_scope.rs
@@ -0,0 +1,54 @@
+//! A trigger override's permission scope is every node the trigger's current job or group writes.
+
+use uuid::Uuid;
+
+use crate::credentials::models::{Action, AuditContext, PermissionScope};
+use crate::declarations::models::{TriggerChange, TriggerChangeRequest};
+use crate::runs::main::frozen_job_request::frozen_job_request;
+use crate::runs::models::JobRunRequest;
+use crate::snapshots::main::job_write_set::job_write_set;
+use crate::store::errors::ControlError;
+use crate::store::main::query::query;
+use crate::store::models::{ControlTransaction, Parameter};
+
+/// A group's scope is its frozen selection's write set, which covers every child's write set.
+pub(crate) async fn trigger_scope(
+    transaction: ControlTransaction,
+    context: &AuditContext,
+    request: &TriggerChangeRequest,
+) -> Result<(ControlTransaction, PermissionScope), ControlError> {
+    let (transaction, rows) = query(transaction,
+        "SELECT location_id,target,job_key FROM sqb_triggers WHERE trigger_id=$1 AND removed_in_snapshot_id IS NULL",
+        vec![Parameter::Id(request.trigger_id)]).await?;
+    let row = rows
+        .first()
+        .ok_or_else(|| ControlError::Invalid("current trigger not found".into()))?;
+    let job = JobRunRequest {
+        run_id: Uuid::nil(),
+        location_id: serde_json::from_value(row["location_id"].clone())
+            .map_err(|error| ControlError::Invalid(error.to_string()))?,
+        target: row["target"].as_str().unwrap_or_default().into(),
+        job_key: row["job_key"].as_str().unwrap_or_default().into(),
+        reason: request.reason.clone(),
+        priority: 0,
+        window: None,
+    };
+    let (transaction, frozen) = frozen_job_request(transaction, context, &job).await?;
+    let (transaction, nodes) =
+        job_write_set(transaction, frozen.snapshot_id, &frozen.selection).await?;
+    let action = match request.change {
+        TriggerChange::Enable | TriggerChange::Pause | TriggerChange::ResetDefault => {
+            Action::TriggerToggle
+        }
+        TriggerChange::SetCursor { .. } | TriggerChange::ResetCursor => Action::SensorResetCursor,
+    };
+    Ok((
+        transaction,
+        PermissionScope {
+            action,
+            location_id: Some(job.location_id),
+            target: Some(job.target),
+            nodes,
+        },
+    ))
+}
diff --git a/crates/sqlbuild-orchestration/src/declarations/main/change_person_trigger.rs b/crates/sqlbuild-orchestration/src/declarations/main/change_person_trigger.rs
index 9eea084384..bab23514ed 100644
--- a/crates/sqlbuild-orchestration/src/declarations/main/change_person_trigger.rs
+++ b/crates/sqlbuild-orchestration/src/declarations/main/change_person_trigger.rs
@@ -1,43 +1,29 @@
-//! Resolve a trigger's real location before authorizing an authenticated runtime override.
+//! A trigger override needs its permission on every node the trigger's current job or group writes.
 
 use serde_json::json;
 use uuid::Uuid;
 
 use crate::authorization::main::authorize_person::authorize_person;
 use crate::authorization::main::deny_authorization::deny;
-use crate::credentials::models::{Action, AuditContext, DefaultRole, PermissionScope};
+use crate::credentials::models::{AuditContext, DefaultRole};
 use crate::declarations::_helpers::operator_change::change_trigger_in;
-use crate::declarations::models::{TriggerChange, TriggerChangeRequest};
+use crate::declarations::_helpers::trigger_scope::trigger_scope;
+use crate::declarations::models::TriggerChangeRequest;
 use crate::store::errors::ControlError;
-use crate::store::main::{begin::begin, commit::commit, query::query};
-use crate::store::models::{ControlStore, Parameter, TransactionMode};
+use crate::store::main::{begin::begin, commit::commit};
+use crate::store::models::{ControlStore, TransactionMode};
 
 pub async fn change_person_trigger(
     store: &ControlStore,
     (context, credentials): (&AuditContext, Uuid),
     request: &TriggerChangeRequest,
 ) -> Result<bool, ControlError> {
-    let (transaction, rows) = query(begin(store, TransactionMode::Write).await?,
-        "SELECT location_id,target FROM sqb_triggers WHERE trigger_id=$1 AND removed_in_snapshot_id IS NULL",
-        vec![Parameter::Id(request.trigger_id)]).await?;
-    let row = rows
-        .first()
-        .ok_or_else(|| ControlError::Invalid("current trigger not found".into()))?;
-    let scope = PermissionScope {
-        action: match request.change {
-            TriggerChange::Enable | TriggerChange::Pause | TriggerChange::ResetDefault => {
-                Action::TriggerToggle
-            }
-            TriggerChange::SetCursor { .. } | TriggerChange::ResetCursor => {
-                Action::SensorResetCursor
-            }
-        },
-        location_id: serde_json::from_value(row["location_id"].clone())
-            .map_err(|error| ControlError::Invalid(error.to_string()))?,
-        target: serde_json::from_value(row["target"].clone())
-            .map_err(|error| ControlError::Invalid(error.to_string()))?,
-        nodes: Default::default(),
-    };
+    let (transaction, scope) = trigger_scope(
+        begin(store, TransactionMode::Write).await?,
+        context,
+        request,
+    )
+    .await?;
     let (transaction, context) = authorize_person(
         transaction,
         (context, credentials),
diff --git a/crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs b/crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs
index 1e8043429f..c340fcc693 100644
--- a/crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs
+++ b/crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs
@@ -2,13 +2,13 @@
 
 use uuid::Uuid;
 
-use crate::authorization::main::begin_authorized::begin_authorized;
-use crate::credentials::models::{Action, AuditContext, DefaultRole, PermissionScope};
+use crate::authorization::main::authorize_listing::authorize_listing;
+use crate::credentials::models::{AuditContext, DefaultRole};
 use crate::runs::_helpers::lifecycle::reads::list_runs_in;
 use crate::runs::models::RunRecord;
 use crate::store::errors::ControlError;
-use crate::store::main::commit::commit;
-use crate::store::models::ControlStore;
+use crate::store::main::{begin::begin, commit::commit};
+use crate::store::models::{ControlStore, TransactionMode};
 
 pub async fn list_person_runs(
     store: &ControlStore,
@@ -16,10 +16,10 @@ pub async fn list_person_runs(
     (location_id, target, limit): (Uuid, &str, u32),
     default_role: DefaultRole,
 ) -> Result<Vec<RunRecord>, ControlError> {
-    let (transaction, _) = begin_authorized(
-        store,
+    let (transaction, _) = authorize_listing(
+        begin(store, TransactionMode::Write).await?,
         (context, credentials),
-        &PermissionScope::action(Action::View, Some((location_id, target))),
+        (location_id, target),
         default_role,
     )
     .await?;
diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs
index a8c7a7c7dc..8324b44276 100644
--- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs
+++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs
@@ -2,11 +2,11 @@
 
 use uuid::Uuid;
 
-use crate::authorization::main::begin_authorized::begin_authorized;
-use crate::credentials::models::{Action, AuditContext, DefaultRole, PermissionScope};
+use crate::authorization::main::authorize_listing::authorize_listing;
+use crate::credentials::models::{AuditContext, DefaultRole};
 use crate::store::errors::ControlError;
-use crate::store::main::commit::commit;
-use crate::store::models::ControlStore;
+use crate::store::main::{begin::begin, commit::commit};
+use crate::store::models::{ControlStore, TransactionMode};
 use crate::warehouse_metadata::_helpers::catalog::read_catalog_page;
 use crate::warehouse_metadata::models::{CatalogPage, CatalogQuery};
 
@@ -16,10 +16,10 @@ pub async fn read_person_catalog(
     (location_id, target, options): (Uuid, &str, &CatalogQuery),
     default_role: DefaultRole,
 ) -> Result<CatalogPage, ControlError> {
-    let (transaction, _) = begin_authorized(
-        store,
+    let (transaction, _) = authorize_listing(
+        begin(store, TransactionMode::Write).await?,
         (context, credentials),
-        &PermissionScope::action(Action::View, Some((location_id, target))),
+        (location_id, target),
         default_role,
     )
     .await?;
diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs
index 7be98d7c96..f34acda4cd 100644
--- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs
+++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs
@@ -2,11 +2,11 @@
 
 use uuid::Uuid;
 
-use crate::authorization::main::begin_authorized::begin_authorized;
-use crate::credentials::models::{Action, AuditContext, DefaultRole, PermissionScope};
+use crate::authorization::main::authorize_listing::authorize_listing;
+use crate::credentials::models::{AuditContext, DefaultRole};
 use crate::store::errors::ControlError;
-use crate::store::main::commit::commit;
-use crate::store::models::ControlStore;
+use crate::store::main::{begin::begin, commit::commit};
+use crate::store::models::{ControlStore, TransactionMode};
 use crate::warehouse_metadata::_helpers::statistics::read_statistics_page;
 use crate::warehouse_metadata::models::{StatisticsPage, StatisticsQuery, StatisticsRetention};
 
@@ -16,10 +16,10 @@ pub async fn read_person_statistics(
     (location_id, target, options): (Uuid, &str, &StatisticsQuery),
     (default_role, retention): (DefaultRole, StatisticsRetention),
 ) -> Result<StatisticsPage, ControlError> {
-    let (transaction, _) = begin_authorized(
-        store,
+    let (transaction, _) = authorize_listing(
+        begin(store, TransactionMode::Write).await?,
         (context, credentials),
-        &PermissionScope::action(Action::View, Some((location_id, target))),
+        (location_id, target),
         default_role,
     )
     .await?;
diff --git a/crates/sqlbuild-orchestration/tests/credentials/roles/helpers.rs b/crates/sqlbuild-orchestration/tests/credentials/roles/helpers.rs
index fad141b126..31c25cb2c3 100644
--- a/crates/sqlbuild-orchestration/tests/credentials/roles/helpers.rs
+++ b/crates/sqlbuild-orchestration/tests/credentials/roles/helpers.rs
@@ -17,11 +17,19 @@ use crate::credentials::models::{
 };
 use crate::credentials::tests::location_helpers::create_location;
 use crate::credentials::tests::test_types::RoleFixture;
+use crate::declarations::main::change_person_trigger::change_person_trigger;
+use crate::declarations::main::list_triggers::list_triggers;
+use crate::declarations::models::{TriggerChange, TriggerChangeRequest};
+use crate::runs::main::list_person_runs::list_person_runs;
 use crate::runs::tests::helpers::snapshot;
+use crate::scheduler::tests::contracts::ticks::helpers::calendar_text;
 use crate::snapshots::main::publish_snapshot::publish_snapshot;
 use crate::store::errors::ControlError;
-use crate::store::main::begin::begin;
+use crate::store::main::{begin::begin, commit::commit, query::query};
 use crate::store::models::{ControlStore, TransactionMode};
+use crate::warehouse_metadata::main::read_person_catalog::read_person_catalog;
+use crate::warehouse_metadata::main::read_person_statistics::read_person_statistics;
+use crate::warehouse_metadata::models::{CatalogQuery, StatisticsQuery, StatisticsRetention};
 
 pub(crate) const SUPPORT_ROLE: &str = "orders_support";
 
@@ -52,7 +60,7 @@ pub(crate) fn credentials(session: &IssuedSession) -> Uuid {
     session.identity.principal_id
 }
 
-async fn assign(
+pub(crate) async fn assign(
     store: &ControlStore,
     fixture: (&AuditContext, &IssuedSession),
     request: RoleAssignmentRequest,
@@ -64,17 +72,27 @@ async fn assign(
     );
 }
 
-/// Publish one generation whose `roles` document grants the support role these actions.
+/// One grant over the `tag:orders` selector, resolved to these snapshot nodes.
+pub(crate) fn orders_grant(permissions: Value, resolved_nodes: &[&str]) -> Value {
+    json!({"nodes": ["tag:orders"], "permissions": permissions, "resolved_nodes": {"tag:orders": resolved_nodes}})
+}
+
+/// Ad-hoc and SQL permissions over one selector, resolved to these snapshot nodes.
+pub(crate) fn developer_grants(selector: &str, resolved_nodes: &[&str]) -> Value {
+    json!([{"nodes": [selector], "permissions": ["run.adhoc", "view.sql"], "resolved_nodes": {selector: resolved_nodes}}])
+}
+
+/// Publish one generation whose `roles` document gives the support role these grants.
 pub(crate) async fn publish_support_role(
     store: &ControlStore,
     location_id: Uuid,
-    (generation, actions): (i64, Value),
+    (generation, grants): (i64, Value),
 ) {
     let mut input = snapshot(location_id);
     input.generation = generation;
     input.documents.insert(
         "roles".into(),
-        json!({SUPPORT_ROLE: {"description": "Answer order questions", "actions": actions}}),
+        json!({SUPPORT_ROLE: {"description": "Answer order questions", "grants": grants}}),
     );
     publish_snapshot(store, &input)
         .await
@@ -100,7 +118,12 @@ pub(crate) async fn role_fixture(store: &ControlStore) -> RoleFixture {
     let location_id = create_location(store, "orders", "orders-admin")
         .await
         .expect("location");
-    publish_support_role(store, location_id, (1, json!(["view"]))).await;
+    publish_support_role(
+        store,
+        location_id,
+        (1, json!([orders_grant(json!(["view"]), &["orders"])])),
+    )
+    .await;
     assign(
         store,
         (&context, &administrator),
@@ -122,17 +145,17 @@ pub(crate) async fn role_fixture(store: &ControlStore) -> RoleFixture {
     }
 }
 
-/// The role that grants the member this action in the fixture location, or `None` when denied.
-pub(crate) async fn granting_role(
+/// The role that grants the member this action on these nodes, or the denial reason.
+pub(crate) async fn authorize_member(
     store: &ControlStore,
     fixture: &RoleFixture,
-    action: Action,
-) -> Option<String> {
+    (action, nodes): (Action, &[&str]),
+) -> Result<String, String> {
     let scope = PermissionScope {
         action,
         location_id: Some(fixture.location_id),
         target: Some("development".into()),
-        nodes: Default::default(),
+        nodes: nodes.iter().map(|node| (*node).into()).collect(),
     };
     let decision = authorize_person(
         begin(store, TransactionMode::Write).await.expect("begin"),
@@ -148,5 +171,147 @@ pub(crate) async fn granting_role(
             .is_none_or(|error| matches!(error, ControlError::Denied { .. })),
         "authorization either grants or denies"
     );
-    decision.ok().map(|(_, context)| context.granting_role)
+    decision
+        .map(|(_, context)| context.granting_role)
+        .map_err(denial)
+}
+
+/// A denial's reason without its decision ID, or the whole message for any other error.
+fn denial(error: ControlError) -> String {
+    let reason = error.to_string();
+    let reason = reason.trim_start_matches("control-plane operation denied: ");
+    reason
+        .split("; decision ")
+        .next()
+        .unwrap_or_default()
+        .to_owned()
+}
+
+/// Publish support role grants with a job that writes only customers and its daily trigger.
+pub(crate) async fn publish_customers_trigger(
+    store: &ControlStore,
+    location_id: Uuid,
+    (generation, grants): (i64, Value),
+) {
+    let mut input = snapshot(location_id);
+    input.generation = generation;
+    input.documents.insert(
+        "roles".into(),
+        json!({SUPPORT_ROLE: {"description": "Answer order questions", "grants": grants}}),
+    );
+    input.documents.insert(
+        "jobs".into(),
+        json!({"customers_job": {"description": "Build customers", "selection": ["customers"]}}),
+    );
+    input.documents.insert("triggers".into(), json!({"customers_daily": {"description": "Refresh customers",
+        "job_key": "customers_job", "starts_at": calendar_text(0),
+        "schedule": {"kind": "calendar", "times": [calendar_text(0)], "refresh_at": "2030-01-01T00:00:00Z"}}}));
+    publish_snapshot(store, &input)
+        .await
+        .expect("publish trigger");
+}
+
+/// The member's override of the location's only trigger, or the denial reason.
+pub(crate) async fn member_trigger_change(
+    store: &ControlStore,
+    fixture: &RoleFixture,
+    change: TriggerChange,
+) -> Result<bool, String> {
+    let trigger_id = list_triggers(store, fixture.location_id, "development")
+        .await
+        .expect("triggers")[0]
+        .trigger_id;
+    change_person_trigger(
+        store,
+        (&fixture.context, credentials(&fixture.member)),
+        &TriggerChangeRequest {
+            event_id: Uuid::now_v7(),
+            trigger_id,
+            actor: String::new(),
+            reason: "Hold customer refreshes".into(),
+            change,
+        },
+    )
+    .await
+    .map_err(denial)
+}
+
+/// The member's catalog, statistics and run listings of the fixture location: allowed or the denial.
+pub(crate) async fn member_listings(
+    store: &ControlStore,
+    fixture: &RoleFixture,
+) -> Vec<Result<(), String>> {
+    let member = (&fixture.context, credentials(&fixture.member));
+    let scope = (fixture.location_id, "development");
+    let catalog = CatalogQuery {
+        database: None,
+        schema: "main".into(),
+        after_observed_at: None,
+        after_event_id: None,
+        limit: None,
+    };
+    let statistics = StatisticsQuery {
+        database: None,
+        schema: "main".into(),
+        name: "orders".into(),
+        after_observed_at: None,
+        after_event_id: None,
+        limit: None,
+    };
+    let retention = StatisticsRetention::days(30).expect("retention");
+    vec![
+        read_person_catalog(
+            store,
+            member,
+            (scope.0, scope.1, &catalog),
+            DefaultRole::None,
+        )
+        .await
+        .map(|_| ())
+        .map_err(denial),
+        read_person_statistics(
+            store,
+            member,
+            (scope.0, scope.1, &statistics),
+            (DefaultRole::None, retention),
+        )
+        .await
+        .map(|_| ())
+        .map_err(denial),
+        list_person_runs(store, member, (scope.0, scope.1, 10), DefaultRole::None)
+            .await
+            .map(|_| ())
+            .map_err(denial),
+    ]
+}
+
+/// The role that grants the member this location-scoped action, or `None` when denied.
+pub(crate) async fn granting_role(
+    store: &ControlStore,
+    fixture: &RoleFixture,
+    action: Action,
+) -> Option<String> {
+    authorize_member(store, fixture, (action, &[])).await.ok()
+}
+
+/// Custom role entries of the snapshot change feed, by publication generation.
+pub(crate) async fn role_changes(store: &ControlStore) -> Vec<(i64, String)> {
+    let transaction = begin(store, TransactionMode::Read).await.expect("read");
+    let (transaction, rows) = query(
+        transaction,
+        "SELECT s.generation, c.change_kind FROM sqb_snapshot_changes c JOIN sqb_snapshots s ON s.snapshot_id=c.snapshot_id \
+         WHERE c.subject_key='role:orders_support' ORDER BY s.generation, c.change_seq",
+        vec![],
+    )
+    .await
+    .expect("role changes");
+    commit(transaction).await.expect("commit");
+    rows.iter()
+        .map(|row| {
+            (
+                row["generation"].as_i64().unwrap_or_default(),
+                row["change_kind"].as_str().unwrap_or_default().into(),
+            )
+        })
+        .collect()
 }
diff --git a/crates/sqlbuild-orchestration/tests/credentials/test_contracts.rs b/crates/sqlbuild-orchestration/tests/credentials/test_contracts.rs
index d8c431a139..982ae42330 100644
--- a/crates/sqlbuild-orchestration/tests/credentials/test_contracts.rs
+++ b/crates/sqlbuild-orchestration/tests/credentials/test_contracts.rs
@@ -18,6 +18,8 @@ mod test_contract;
 mod test_custom_roles;
 #[path = "passwords/test_contract.rs"]
 mod test_passwords;
+#[path = "test_project_roles.rs"]
+mod test_project_roles;
 #[path = "test_recovery.rs"]
 mod test_recovery;
 #[path = "sessions/test_contract.rs"]
diff --git a/crates/sqlbuild-orchestration/tests/credentials/test_custom_roles.rs b/crates/sqlbuild-orchestration/tests/credentials/test_custom_roles.rs
index 59c3915a79..48679facee 100644
--- a/crates/sqlbuild-orchestration/tests/credentials/test_custom_roles.rs
+++ b/crates/sqlbuild-orchestration/tests/credentials/test_custom_roles.rs
@@ -3,7 +3,9 @@
 use serde_json::{Value, json};
 
 use crate::credentials::models::Action;
-use crate::credentials::tests::role_helpers::{granting_role, publish_support_role, role_fixture};
+use crate::credentials::tests::role_helpers::{
+    granting_role, orders_grant, publish_support_role, role_fixture,
+};
 use crate::credentials::tests::test_types::CustomRoleTestCase;
 use crate::runs::tests::helpers::snapshot;
 use crate::snapshots::main::publish_snapshot::publish_snapshot;
@@ -50,7 +52,13 @@ fn given_published_custom_roles_when_republished_then_each_definition_applies_at
                 publish_support_role(
                     store,
                     roles.location_id,
-                    (generation, publication["actions"].clone()),
+                    (
+                        generation,
+                        json!([orders_grant(
+                            publication["permissions"].clone(),
+                            &["orders"]
+                        )]),
+                    ),
                 )
                 .await;
                 assert_eq!(
@@ -58,7 +66,7 @@ fn given_published_custom_roles_when_republished_then_each_definition_applies_at
                     publication["expected"],
                     "{}: {}",
                     test_case.description,
-                    publication["actions"]
+                    publication["permissions"]
                 );
             }
             let mut removed = snapshot(roles.location_id);
diff --git a/crates/sqlbuild-orchestration/tests/credentials/test_project_roles.rs b/crates/sqlbuild-orchestration/tests/credentials/test_project_roles.rs
new file mode 100644
index 0000000000..2182207428
--- /dev/null
+++ b/crates/sqlbuild-orchestration/tests/credentials/test_project_roles.rs
@@ -0,0 +1,514 @@
+//! Project role grants take effect on publish and follow selector coverage per snapshot.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use serde_json::json;
+
+use crate::authorization::main::compile_roles::compile_roles;
+use crate::credentials::models::{Action, RoleAssignmentRequest, RoleKey};
+use crate::credentials::tests::role_helpers::{
+    SUPPORT_ROLE, assign, authorize_member, developer_grants, member_listings,
+    member_trigger_change, orders_grant, publish_customers_trigger, publish_support_role,
+    role_changes, role_fixture,
+};
+use crate::credentials::tests::test_types::{ProjectRoleTestCase, RolesFileTestCase};
+use crate::declarations::models::TriggerChange;
+use crate::runs::tests::helpers::snapshot;
+use crate::snapshots::main::publish_snapshot::publish_snapshot;
+use crate::store::main::migrate_store::migrate_store;
+use crate::store::tests::store_helpers::{close_test_store, open_test_store, run};
+use crate::store::tests::store_test_types::Backend;
+
+const ADHOC_DENIED: &str = "missing run.adhoc on nodes customers; all requested nodes denied";
+
+#[test]
+fn given_published_selector_when_it_matches_more_nodes_then_coverage_grows_on_publish() {
+    let test_cases = [
+        ProjectRoleTestCase {
+            description: "SQLite widens coverage of an unchanged selector on publish",
+            backend: Backend::Sqlite,
+            expected_before: Err(ADHOC_DENIED),
+            expected_after: Ok(SUPPORT_ROLE),
+        },
+        ProjectRoleTestCase {
+            description: "PostgreSQL widens coverage of an unchanged selector on publish",
+            backend: Backend::Postgres,
+            expected_before: Err(ADHOC_DENIED),
+            expected_after: Ok(SUPPORT_ROLE),
+        },
+    ];
+    for test_case in test_cases {
+        run(async {
+            let Some(fixture) = open_test_store(test_case.backend).await else {
+                return;
+            };
+            let store = &fixture.store;
+            migrate_store(store).await.expect("migrate");
+            let roles = role_fixture(store).await;
+            let grants = |nodes: &[&str]| developer_grants("tag:orders", nodes);
+            publish_support_role(store, roles.location_id, (2, grants(&["orders"]))).await;
+            let adhoc = (Action::RunAdhoc, ["customers", "orders"].as_slice());
+            assert_eq!(
+                authorize_member(store, &roles, adhoc)
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_before,
+                "{}: an ad-hoc run needs every written node",
+                test_case.description
+            );
+            publish_support_role(
+                store,
+                roles.location_id,
+                (3, grants(&["customers", "orders"])),
+            )
+            .await;
+            assert_eq!(
+                authorize_member(store, &roles, adhoc)
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_after,
+                "{}",
+                test_case.description
+            );
+            assert_eq!(
+                authorize_member(store, &roles, (Action::ViewSql, &["customers"]))
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_after,
+                "{}: view.sql follows the same coverage",
+                test_case.description
+            );
+            assert_eq!(
+                role_changes(store).await,
+                [
+                    (1, "added".to_owned()),
+                    (2, "changed".to_owned()),
+                    (3, "changed".to_owned())
+                ],
+                "{}: the new role, the dropped view grant and the grown coverage enter the change feed",
+                test_case.description
+            );
+            close_test_store(fixture).await;
+        });
+    }
+}
+
+#[test]
+fn given_selector_resolving_to_no_nodes_when_published_then_it_grants_nothing() {
+    let test_cases = [
+        ProjectRoleTestCase {
+            description: "SQLite grants nothing for an empty resolution",
+            backend: Backend::Sqlite,
+            expected_before: Ok(SUPPORT_ROLE),
+            expected_after: Err("missing permission in this scope; all requested nodes denied"),
+        },
+        ProjectRoleTestCase {
+            description: "PostgreSQL grants nothing for an empty resolution",
+            backend: Backend::Postgres,
+            expected_before: Ok(SUPPORT_ROLE),
+            expected_after: Err("missing permission in this scope; all requested nodes denied"),
+        },
+    ];
+    for test_case in test_cases {
+        run(async {
+            let Some(fixture) = open_test_store(test_case.backend).await else {
+                return;
+            };
+            let store = &fixture.store;
+            migrate_store(store).await.expect("migrate");
+            let roles = role_fixture(store).await;
+            assert_eq!(
+                authorize_member(store, &roles, (Action::View, &[]))
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_before,
+                "{}: the resolved grant allows view",
+                test_case.description
+            );
+            publish_support_role(
+                store,
+                roles.location_id,
+                (2, json!([orders_grant(json!(["view", "run.adhoc"]), &[])])),
+            )
+            .await;
+            assert_eq!(
+                authorize_member(store, &roles, (Action::View, &[]))
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_after,
+                "{}: a location-scoped action needs a grant that resolved to a node",
+                test_case.description
+            );
+            assert_eq!(
+                authorize_member(store, &roles, (Action::RunAdhoc, &["orders"]))
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                Err("missing run.adhoc on nodes orders; all requested nodes denied"),
+                "{}: the denial names the uncovered node",
+                test_case.description
+            );
+            close_test_store(fixture).await;
+        });
+    }
+}
+
+#[test]
+fn given_grant_outside_a_triggers_job_when_the_member_overrides_it_then_both_trigger_actions_are_denied()
+ {
+    let test_cases = [
+        ProjectRoleTestCase {
+            description: "SQLite authorizes trigger overrides on the job's write set",
+            backend: Backend::Sqlite,
+            expected_before: Err(
+                "missing trigger.toggle on nodes customers; all requested nodes denied",
+            ),
+            expected_after: Err(
+                "missing sensor.reset_cursor on nodes customers; all requested nodes denied",
+            ),
+        },
+        ProjectRoleTestCase {
+            description: "PostgreSQL authorizes trigger overrides on the job's write set",
+            backend: Backend::Postgres,
+            expected_before: Err(
+                "missing trigger.toggle on nodes customers; all requested nodes denied",
+            ),
+            expected_after: Err(
+                "missing sensor.reset_cursor on nodes customers; all requested nodes denied",
+            ),
+        },
+    ];
+    for test_case in test_cases {
+        run(async {
+            let Some(fixture) = open_test_store(test_case.backend).await else {
+                return;
+            };
+            let store = &fixture.store;
+            migrate_store(store).await.expect("migrate");
+            let roles = role_fixture(store).await;
+            let grants = |nodes: &[&str]| {
+                json!([orders_grant(
+                    json!(["view", "trigger.toggle", "sensor.reset_cursor"]),
+                    nodes
+                )])
+            };
+            publish_customers_trigger(store, roles.location_id, (2, grants(&["orders"]))).await;
+            assert_eq!(
+                member_trigger_change(store, &roles, TriggerChange::Pause)
+                    .await
+                    .map(|_| SUPPORT_ROLE)
+                    .as_ref()
+                    .map(|role| *role)
+                    .map_err(String::as_str),
+                test_case.expected_before,
+                "{}: an orders-only grant cannot pause a customers-only trigger",
+                test_case.description
+            );
+            assert_eq!(
+                member_trigger_change(store, &roles, TriggerChange::ResetCursor)
+                    .await
+                    .map(|_| SUPPORT_ROLE)
+                    .as_ref()
+                    .map(|role| *role)
+                    .map_err(String::as_str),
+                test_case.expected_after,
+                "{}: nor reset its cursor",
+                test_case.description
+            );
+            publish_customers_trigger(
+                store,
+                roles.location_id,
+                (3, grants(&["customers", "orders"])),
+            )
+            .await;
+            assert_eq!(
+                member_trigger_change(store, &roles, TriggerChange::Pause).await,
+                Ok(true),
+                "{}: coverage of the job's write set allows the pause",
+                test_case.description
+            );
+            close_test_store(fixture).await;
+        });
+    }
+}
+
+#[test]
+fn given_view_grant_on_part_of_the_location_when_listing_then_each_listing_is_denied_until_full_coverage()
+ {
+    let test_cases = [
+        ProjectRoleTestCase {
+            description: "SQLite location listings need view on every node",
+            backend: Backend::Sqlite,
+            expected_before: Err(
+                "missing view on nodes customers, inventory, products; all requested nodes denied",
+            ),
+            expected_after: Ok(SUPPORT_ROLE),
+        },
+        ProjectRoleTestCase {
+            description: "PostgreSQL location listings need view on every node",
+            backend: Backend::Postgres,
+            expected_before: Err(
+                "missing view on nodes customers, inventory, products; all requested nodes denied",
+            ),
+            expected_after: Ok(SUPPORT_ROLE),
+        },
+    ];
+    for test_case in test_cases {
+        run(async {
+            let Some(fixture) = open_test_store(test_case.backend).await else {
+                return;
+            };
+            let store = &fixture.store;
+            migrate_store(store).await.expect("migrate");
+            let roles = role_fixture(store).await;
+            assert_eq!(
+                member_listings(store, &roles).await,
+                vec![test_case.expected_before.map(|_| ()).map_err(str::to_owned); 3],
+                "{}: catalog, statistics and runs are denied, not filtered",
+                test_case.description
+            );
+            publish_support_role(
+                store,
+                roles.location_id,
+                (
+                    2,
+                    json!([orders_grant(
+                        json!(["view"]),
+                        &["customers", "inventory", "orders", "products"]
+                    )]),
+                ),
+            )
+            .await;
+            assert_eq!(
+                member_listings(store, &roles).await,
+                vec![test_case.expected_after.map(|_| ()).map_err(str::to_owned); 3],
+                "{}: full coverage allows every listing",
+                test_case.description
+            );
+            close_test_store(fixture).await;
+        });
+    }
+}
+
+#[test]
+fn given_role_removed_from_code_when_its_assignment_remains_then_it_grants_nothing() {
+    let test_cases = [
+        ProjectRoleTestCase {
+            description: "SQLite stale assignment grants nothing",
+            backend: Backend::Sqlite,
+            expected_before: Ok(SUPPORT_ROLE),
+            expected_after: Err("missing run.adhoc on nodes orders; all requested nodes denied"),
+        },
+        ProjectRoleTestCase {
+            description: "PostgreSQL stale assignment grants nothing",
+            backend: Backend::Postgres,
+            expected_before: Ok(SUPPORT_ROLE),
+            expected_after: Err("missing run.adhoc on nodes orders; all requested nodes denied"),
+        },
+    ];
+    for test_case in test_cases {
+        run(async {
+            let Some(fixture) = open_test_store(test_case.backend).await else {
+                return;
+            };
+            let store = &fixture.store;
+            migrate_store(store).await.expect("migrate");
+            let roles = role_fixture(store).await;
+            publish_support_role(
+                store,
+                roles.location_id,
+                (2, developer_grants("tag:orders", &["orders"])),
+            )
+            .await;
+            let adhoc = (Action::RunAdhoc, ["orders"].as_slice());
+            assert_eq!(
+                authorize_member(store, &roles, adhoc)
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_before,
+                "{}: the published role grants ad-hoc runs",
+                test_case.description
+            );
+            let mut input = snapshot(roles.location_id);
+            input.generation = 3;
+            input.documents.insert("roles".into(), json!({}));
+            publish_snapshot(store, &input)
+                .await
+                .expect("publish without roles");
+            assert_eq!(
+                authorize_member(store, &roles, adhoc)
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_after,
+                "{}: the removed role's assignment grants nothing on nodes",
+                test_case.description
+            );
+            assert_eq!(
+                authorize_member(store, &roles, (Action::View, &[]))
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                Err("missing permission in this scope; all requested nodes denied"),
+                "{}: nor location-wide",
+                test_case.description
+            );
+            assert_eq!(
+                role_changes(store).await.last(),
+                Some(&(3, "removed".to_owned())),
+                "{}: the removal enters the change feed",
+                test_case.description
+            );
+            close_test_store(fixture).await;
+        });
+    }
+}
+
+#[test]
+fn given_two_custom_roles_when_each_covers_part_of_a_run_then_their_union_authorizes_it() {
+    let test_cases = [
+        ProjectRoleTestCase {
+            description: "SQLite authorizes the union of two custom roles",
+            backend: Backend::Sqlite,
+            expected_before: Err(ADHOC_DENIED),
+            expected_after: Ok("orders_returns+orders_support"),
+        },
+        ProjectRoleTestCase {
+            description: "PostgreSQL authorizes the union of two custom roles",
+            backend: Backend::Postgres,
+            expected_before: Err(ADHOC_DENIED),
+            expected_after: Ok("orders_returns+orders_support"),
+        },
+    ];
+    for test_case in test_cases {
+        run(async {
+            let Some(fixture) = open_test_store(test_case.backend).await else {
+                return;
+            };
+            let store = &fixture.store;
+            migrate_store(store).await.expect("migrate");
+            let roles = role_fixture(store).await;
+            let mut input = snapshot(roles.location_id);
+            input.generation = 2;
+            input.documents.insert(
+                "roles".into(),
+                json!({
+                    SUPPORT_ROLE: {"description": "Answer order questions", "grants": developer_grants("tag:orders", &["orders"])},
+                    "orders_returns": {"description": "Handle returns", "grants": developer_grants("tag:returns", &["customers"])},
+                }),
+            );
+            publish_snapshot(store, &input)
+                .await
+                .expect("publish roles");
+            let adhoc = (Action::RunAdhoc, ["customers", "orders"].as_slice());
+            assert_eq!(
+                authorize_member(store, &roles, adhoc)
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_before,
+                "{}: one role alone does not cover the run",
+                test_case.description
+            );
+            assign(
+                store,
+                (&roles.context, &roles.administrator),
+                RoleAssignmentRequest {
+                    principal_id: roles.member_id,
+                    role: RoleKey::Custom("orders_returns".into()),
+                    location_id: Some(roles.location_id),
+                    target: None,
+                    revoke: false,
+                },
+            )
+            .await;
+            assert_eq!(
+                authorize_member(store, &roles, adhoc)
+                    .await
+                    .as_deref()
+                    .map_err(String::as_str),
+                test_case.expected_after,
+                "{}",
+                test_case.description
+            );
+            close_test_store(fixture).await;
+        });
+    }
+}
+
+#[test]
+fn given_roles_file_when_compiled_then_selectors_resolve_and_empty_grants_warn() {
+    let roles = r#"
+[roles.orders_support]
+description = "Answer order questions"
+[[roles.orders_support.grants]]
+nodes = ["tag:orders"]
+permissions = ["view.sql", "run.adhoc"]
+[[roles.orders_support.grants]]
+nodes = ["tag:orders,tag:refunds"]
+permissions = ["view.evidence"]
+"#;
+    let test_cases = [
+        RolesFileTestCase {
+            description: "resolved grants and a warning for the empty intersection",
+            text: roles,
+            expected_compiled: Ok((
+                json!({"orders_support": {"description": "Answer order questions", "grants": [
+                    {"nodes": ["tag:orders"], "permissions": ["view.sql", "run.adhoc"],
+                        "resolved_nodes": {"tag:orders": ["model:orders", "model:returns"]}},
+                    {"nodes": ["tag:orders,tag:refunds"], "permissions": ["view.evidence"],
+                        "resolved_nodes": {"tag:orders,tag:refunds": []}},
+                ]}}),
+                vec![
+                    "role orders_support grant 2 selects no nodes (tag:orders,tag:refunds); it grants nothing in this snapshot",
+                ],
+            )),
+        },
+        RolesFileTestCase {
+            description: "an administrative permission is never granted from code",
+            text: "[roles.orders_support]\ndescription = \"Answer order questions\"\n[[roles.orders_support.grants]]\nnodes = [\"tag:orders\"]\npermissions = [\"role.assign\"]\n",
+            expected_compiled: Err(
+                "roles.toml: invalid control-plane data: role orders_support needs a description, a key that is not a built-in role, and grants of node selectors with only non-administrative permissions",
+            ),
+        },
+        RolesFileTestCase {
+            description: "a selector the project rejects fails publication",
+            text: "[roles.orders_support]\ndescription = \"Answer order questions\"\n[[roles.orders_support.grants]]\nnodes = [\"tag:unknown\"]\npermissions = [\"view.sql\"]\n",
+            expected_compiled: Err(
+                "roles.toml: role orders_support selector 'tag:unknown': no models found with tag 'unknown'",
+            ),
+        },
+    ];
+    let catalog: BTreeMap<&str, Result<Vec<&str>, &str>> = BTreeMap::from([
+        ("tag:orders", Ok(vec!["model:orders", "model:returns"])),
+        ("tag:orders,tag:refunds", Ok(vec![])),
+        ("tag:unknown", Err("no models found with tag 'unknown'")),
+    ]);
+    let resolve = |selector: &str| {
+        catalog[selector]
+            .clone()
+            .map(|keys| keys.into_iter().map(str::to_owned).collect::<BTreeSet<_>>())
+            .map_err(str::to_owned)
+    };
+    for test_case in test_cases {
+        let compiled = compile_roles(test_case.text, resolve);
+        assert_eq!(
+            compiled
+                .as_ref()
+                .map(|(document, warnings)| (
+                    document.clone(),
+                    warnings.iter().map(String::as_str).collect::<Vec<_>>()
+                ))
+                .map_err(String::as_str),
+            test_case.expected_compiled,
+            "{}",
+            test_case.description
+        );
+    }
+}
diff --git a/crates/sqlbuild-orchestration/tests/credentials/test_types.rs b/crates/sqlbuild-orchestration/tests/credentials/test_types.rs
index d4dc5ef2e3..c309a0a9b5 100644
--- a/crates/sqlbuild-orchestration/tests/credentials/test_types.rs
+++ b/crates/sqlbuild-orchestration/tests/credentials/test_types.rs
@@ -55,3 +55,16 @@ pub(crate) struct RoleFixture {
     pub(crate) member: crate::credentials::models::IssuedSession,
     pub(crate) member_id: uuid::Uuid,
 }
+
+pub(crate) struct ProjectRoleTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) backend: Backend,
+    pub(crate) expected_before: Result<&'static str, &'static str>,
+    pub(crate) expected_after: Result<&'static str, &'static str>,
+}
+
+pub(crate) struct RolesFileTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) text: &'static str,
+    pub(crate) expected_compiled: Result<(Value, Vec<&'static str>), &'static str>,
+}
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/mod.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/mod.rs
index 2101d9ab7d..5949e9aa31 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/mod.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/mod.rs
@@ -10,5 +10,6 @@ pub(crate) mod snapshot_declarations;
 pub(crate) mod snapshot_execution;
 pub(crate) mod snapshot_models;
 pub(crate) mod snapshot_resources;
+pub(crate) mod snapshot_roles;
 pub(crate) mod snapshot_worker;
 pub(crate) mod values;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_roles.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_roles.rs
new file mode 100644
index 0000000000..4867f861ab
--- /dev/null
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_roles.rs
@@ -0,0 +1,52 @@
+//! Project roles from the reviewed `roles.toml` file, their selectors resolved against this snapshot.
+
+use std::collections::BTreeSet;
+use std::path::Path;
+
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::{PyResult, Python};
+use sqlbuild_analysis::graph::main::resolve_selectors::resolve_selectors;
+use sqlbuild_analysis::graph::models::ProjectGraph;
+use sqlbuild_orchestration::authorization::main::compile_roles::compile_roles;
+use sqlbuild_orchestration::snapshots::models::SnapshotInput;
+
+use crate::bindings::_helpers::native_host::host::write_console;
+
+/// The project-root file that defines project roles.
+pub(crate) const ROLES_FILE: &str = "roles.toml";
+
+/// Node keys of one grant selector, in the job selection grammar.
+fn selector_nodes(graph: &ProjectGraph, selector: &str) -> Result<BTreeSet<String>, String> {
+    let tokens: Vec<String> = selector.split_whitespace().map(str::to_owned).collect();
+    let keys = resolve_selectors(graph, &tokens, &[]).map_err(|error| error.message)?;
+    let mut nodes: BTreeSet<String> = BTreeSet::new();
+    for (kind, name) in keys {
+        nodes.insert(format!("{kind}:{name}"));
+    }
+    Ok(nodes)
+}
+
+pub(crate) fn compile_project_roles(
+    py: Python<'_>,
+    mut input: SnapshotInput,
+    (project_dir, graph): (&Path, &ProjectGraph),
+) -> PyResult<SnapshotInput> {
+    let path = project_dir.join(ROLES_FILE);
+    if !path.is_file() {
+        return Ok(input);
+    }
+    if input.documents.contains_key("roles") {
+        return Err(PyValueError::new_err(
+            "roles.toml cannot overwrite a snapshot roles document",
+        ));
+    }
+    let text = std::fs::read_to_string(&path)
+        .map_err(|error| PyValueError::new_err(format!("roles.toml: {error}")))?;
+    let (document, warnings) = compile_roles(&text, |selector| selector_nodes(graph, selector))
+        .map_err(PyValueError::new_err)?;
+    for warning in warnings {
+        write_console(py, "stderr", &format!("Warning: {warning}\n"))?;
+    }
+    input.documents.insert("roles".into(), document);
+    Ok(input)
+}
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/location_command/snapshot.rs b/crates/sqlbuild-python/src/bindings/_helpers/location_command/snapshot.rs
index f975aa24ed..e5e7028962 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/location_command/snapshot.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/location_command/snapshot.rs
@@ -17,6 +17,7 @@ use crate::bindings::_helpers::compiled_project::snapshot_declarations::{
 };
 use crate::bindings::_helpers::compiled_project::snapshot_models::model_nodes;
 use crate::bindings::_helpers::compiled_project::snapshot_resources::resource_nodes;
+use crate::bindings::_helpers::compiled_project::snapshot_roles::compile_project_roles;
 use crate::bindings::_helpers::compiled_views::project_view::NativeProjectView;
 use crate::bindings::_helpers::native_host::orchestration_declarations::collect_declarations;
 use crate::bindings::_helpers::native_host::orchestration_environment::environment_hash;
@@ -67,9 +68,9 @@ pub(crate) fn build_orchestration_snapshot(
         Some(declarations) => declarations,
         None => discover_declarations(py, &project_dir)?,
     };
+    let graph = project.lineage_graph(py)?;
     if !declarations.is_empty() {
         let authored = collect_declarations(py, declarations)?;
-        let graph = project.lineage_graph(py)?;
         input = compile_declarations(
             py,
             input,
@@ -77,6 +78,7 @@ pub(crate) fn build_orchestration_snapshot(
             authored,
         )?;
     }
+    input = compile_project_roles(py, input, (&project_dir, graph.bind(py).get().graph()))?;
     input.nodes = identity_nodes(project.into_pyobject(py)?.as_any(), input.nodes)?;
     Ok(NativeOrchestrationSnapshot { input })
 }
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md
index 9a352beddb..dfa7f047ff 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md
@@ -11,6 +11,7 @@ Online: https://sqlbuild.com/docs/cli/server/
 - Set retention
 - Grant roles to OIDC groups (beta)
 - Inspect published SQL
+- Define project roles
 
 Run the control API and scheduler without loading project Python:
 
@@ -85,8 +86,8 @@ assignment. Add `groups_claim` and `group_roles` to the JSON file passed to `--o
 | --- | --- |
 | `groups_claim` | ID-token claim that lists the person's groups. Required when `group_roles` is set. |
 | `group_roles[].group` | Group name exactly as the claim lists it. |
-| `group_roles[].role` | `viewer`, `operator`, `developer` or a custom role the location publishes. Not `admin`. |
-| `group_roles[].location_id` | Limits the role to one location. Required for a custom role or a target. |
+| `group_roles[].role` | `viewer`, `operator`, `developer` or a project role the location publishes. Not `admin`. |
+| `group_roles[].location_id` | Limits the role to one location. Required for a project role or a target. |
 | `group_roles[].target` | Limits the role to one target in that location. |
 
 Groups grant roles only to people who already have an entry in `identities`; a group never
@@ -102,7 +103,7 @@ person's groups with the ones in that token:
 - A claim that is missing, is not a list of non-empty names, or is replaced by an overage pointer
   gives no groups.
 
-The server refuses to start when a mapping grants `admin`, scopes a custom role or target without
+The server refuses to start when a mapping grants `admin`, scopes a project role or target without
 a location, has a blank group, or when `group_roles` is set without `groups_claim`.
 
 Servers that share one control database must use identical OIDC and `group_roles` configuration.
@@ -129,7 +130,8 @@ stderr; a failed request prints no SQL to stdout. Permission denials include the
 The equivalent person-authenticated API is a `GET` request to
 `/api/control/locations/{location_id}/targets/{target}/nodes/{node_key}/sql`.
 
-Developer or Admin access must cover the requested location and target. Viewer, Operator and
+Developer or Admin access must cover the requested location and target, or a
+[project role](#define-project-roles) must grant `view.sql` on the node. Viewer, Operator and
 the configurable default Viewer role do not grant SQL inspection. The response contains the
 snapshot ID, exact node key, raw SQL and compiled SQL. Missing SQL fields are `null`; an absent
 current publication or node returns JSON `null` after authorization.
@@ -138,3 +140,37 @@ SQL inspection does not execute SQL or return Python source, execution artifacts
 data. Every successful inspection records an audit entry without copying SQL into it. If the
 audit cannot be saved, the request fails without returning SQL. Revocation takes effect on the
 next request, including sessions created by [`sqb login`](login.md).
+
+## Define project roles
+
+Put project roles in `roles.toml` at the project root. `sqb location publish` resolves each
+grant's node selectors against the published project:
+
+```toml
+[roles.orders_support]
+description = "Answer order questions"
+
+[[roles.orders_support.grants]]
+nodes = ["tag:orders"]
+permissions = ["view.sql", "run.adhoc"]
+
+[[roles.orders_support.grants]]
+nodes = ["tag:orders,tag:pii"]
+permissions = ["view.evidence"]
+```
+
+| Rule | Behaviour |
+| --- | --- |
+| Selectors | Same grammar as a job's `select`; an unknown tag or name fails publication |
+| No matching nodes | Publication succeeds with a warning; the grant gives nothing |
+| Permissions | Any non-administrative action; `user.manage`, `role.assign` and `audit.view` are rejected |
+| Runs | `job.run`, `run.adhoc`, `run.rerun` and `run.rewrite_history` need the permission on every node the run writes |
+| Triggers | `trigger.toggle` and `sensor.reset_cursor` need the permission on every node the trigger's job writes |
+| Listings | The catalog, statistics and run list need `view` on every node of the current publication; partial coverage is denied, not filtered |
+| Several roles | Grants add up across a person's roles |
+
+Assign a project role with `POST /api/admin/roles`, using the role key as `role`. Each
+publication replaces the location's project roles: new and changed grants apply at once, and
+removed roles stop applying. A selector covers newly matching nodes as soon as they are
+published. Each change appears in the publication's change list as `role:<key>`. A denial lists
+the nodes the request was missing.
diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py
index 0b5cd7e114..2e0d493e9a 100644
--- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py
+++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py
@@ -311,3 +311,11 @@ class SchedulerCrashTestCase:
     fixture: str
     expected_anomalies: dict[str, int]
     expected_minimum_kills: int
+
+
+@dataclass(frozen=True)
+class ProjectRolesTestCase:
+    description: str
+    roles_file: str
+    expected_warning: str
+    expected_denial: str
diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py
index 9cd952273c..ed3043b631 100644
--- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py
+++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py
@@ -1218,3 +1218,55 @@ def scheduler_crash_history(directory: Path, fixture: dict[str, Any]) -> dict[st
         "ticks_before_survivor": interrupted,
         "ticks": len(ticks),
     }
+
+
+def publish_generation(
+    project: Path, api: SnapshotApi, generation: int
+) -> subprocess.CompletedProcess[str]:
+    """Publish one generation of a project through the installed location command."""
+    return subprocess.run(
+        [
+            str(Path(sys.executable).with_name("sqb")),
+            "location",
+            "publish",
+            "--project-dir",
+            str(project),
+            "--location-id",
+            api.location_id,
+            "--target",
+            "development",
+            "--server",
+            api.url,
+            "--generation",
+            str(generation),
+        ],
+        env=os.environ | {"SQLBUILD_LOCATION_TOKEN": api.bearer},
+        capture_output=True,
+        text=True,
+        timeout=60,
+        check=False,
+    )
+
+
+def read_person_sql(api: SnapshotApi, session: tuple[str, str], node: str) -> tuple[int, str]:
+    """A person's view.sql read of one node: the node key on success, else the denial."""
+    path: str = f"/api/control/locations/{api.location_id}/targets/development/nodes/{node}/sql"
+    headers: dict[str, str] = {
+        "Host": "control.example.com",
+        "Origin": "https://control.example.com",
+        "Cookie": session[0],
+        "x-sqb-csrf": session[1],
+    }
+    try:
+        with urlopen(Request(api.url + path, headers=headers), timeout=5) as response:
+            return 200, json.load(response)["node_key"]
+    except HTTPError as denied:
+        return denied.code, json.load(denied)["error"].split("; decision ")[0]
+
+
+def tagged_model(name: str, tags: str) -> str:
+    """A table model with these tags."""
+    return (
+        f"MODEL (description 'Synthetic {name}.', materialized table, tags [{tags}]);\n"
+        "select 1 as id\n"
+    )
diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_project_roles.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_project_roles.py
new file mode 100644
index 0000000000..61968b6f40
--- /dev/null
+++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_project_roles.py
@@ -0,0 +1,104 @@
+"""Installed publication compiles roles.toml; person grants follow selector coverage per snapshot."""
+
+import subprocess
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tests.integration.src.sqlbuild.compiler.compiled_project.publication._test_types import (
+    ProjectRolesTestCase,
+)
+from tests.integration.src.sqlbuild.compiler.compiled_project.publication.helpers import (
+    publish_generation,
+    read_person_sql,
+    snapshot_api,
+    tagged_model,
+)
+from tests.integration.src.sqlbuild.compiler.compiled_project.publication.person.helpers import (
+    person_login,
+    person_request,
+    provision_person,
+)
+from tests.integration.src.sqlbuild.compiler.graph.helpers import write_files
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ProjectRolesTestCase(
+            description="roles.toml grants follow tags and every publication applies at once",
+            roles_file="benchmarks/duckshop/edge_cases/orchestration/project_roles.toml",
+            expected_warning=(
+                "Warning: role orders_support grant 2 selects no nodes (tag:support,tag:pii); "
+                "it grants nothing in this snapshot"
+            ),
+            expected_denial=(
+                "missing view.sql on nodes model:customers; all requested nodes denied"
+            ),
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_project_roles_when_published_then_person_sql_follows_selector_coverage_at_once(
+    test_case: ProjectRolesTestCase, tmp_path: Path
+) -> None:
+    project: Path = tmp_path / "project"
+    roles: str = Path(test_case.roles_file).read_text()
+    write_files(
+        project_dir=project,
+        files={
+            "sqlbuild_project.toml": (
+                'name = "orders"\nadapter = "duckdb"\n'
+                '[connections.local]\ndatabase = "warehouse.duckdb"\n'
+                '[targets.development]\nschema = "main"\n'
+            ),
+            "models/orders.sql": tagged_model("orders", "support"),
+            "models/customers.sql": tagged_model("customers", "pii"),
+            "roles.toml": roles,
+        },
+    )
+    with snapshot_api(
+        tmp_path / "control",
+        server_arguments=("--password-origin", "https://control.example.com"),
+    ) as api:
+        first: subprocess.CompletedProcess[str] = publish_generation(project, api, 1)
+        assert first.returncode == 0, first.stderr
+        assert test_case.expected_warning in first.stderr
+        admin: dict[str, Any] = provision_person(
+            tmp_path / "control", "create-user", "orders-admin"
+        )
+        person: dict[str, Any] = provision_person(
+            tmp_path / "control", "create-principal", "orders-support"
+        )
+        admin_session: tuple[str, str] = person_login(api, admin["name"])
+        session: tuple[str, str] = person_login(api, person["name"])
+        assert person_request(
+            api,
+            "/api/admin/roles",
+            body={
+                "principal_id": person["principal_id"],
+                "role": "orders_support",
+                "location_id": api.location_id,
+                "target": "development",
+            },
+            session=admin_session,
+        ) == {"applied": True}
+        assert read_person_sql(api, session, "model:orders") == (200, "model:orders")
+        assert read_person_sql(api, session, "model:customers") == (403, test_case.expected_denial)
+        write_files(
+            project_dir=project,
+            files={"models/customers.sql": tagged_model("customers", "pii, support")},
+        )
+        grown: subprocess.CompletedProcess[str] = publish_generation(project, api, 2)
+        assert grown.returncode == 0, grown.stderr
+        assert "Warning:" not in grown.stderr
+        assert read_person_sql(api, session, "model:customers") == (200, "model:customers")
+        write_files(
+            project_dir=project,
+            files={"roles.toml": roles.replace('nodes = ["tag:support"]', 'nodes = ["tag:pii"]')},
+        )
+        rewritten: subprocess.CompletedProcess[str] = publish_generation(project, api, 3)
+        assert rewritten.returncode == 0, rewritten.stderr
+        assert read_person_sql(api, session, "model:customers") == (200, "model:customers")
+        assert read_person_sql(api, session, "model:orders")[0] == 403
diff --git a/website/src/content/docs/docs/cli/server.mdx b/website/src/content/docs/docs/cli/server.mdx
index cb76ab83ef..91be26c0d9 100644
--- a/website/src/content/docs/docs/cli/server.mdx
+++ b/website/src/content/docs/docs/cli/server.mdx
@@ -78,8 +78,8 @@ assignment. Add `groups_claim` and `group_roles` to the JSON file passed to `--o
 | --- | --- |
 | `groups_claim` | ID-token claim that lists the person's groups. Required when `group_roles` is set. |
 | `group_roles[].group` | Group name exactly as the claim lists it. |
-| `group_roles[].role` | `viewer`, `operator`, `developer` or a custom role the location publishes. Not `admin`. |
-| `group_roles[].location_id` | Limits the role to one location. Required for a custom role or a target. |
+| `group_roles[].role` | `viewer`, `operator`, `developer` or a project role the location publishes. Not `admin`. |
+| `group_roles[].location_id` | Limits the role to one location. Required for a project role or a target. |
 | `group_roles[].target` | Limits the role to one target in that location. |
 
 Groups grant roles only to people who already have an entry in `identities`; a group never
@@ -95,7 +95,7 @@ person's groups with the ones in that token:
 - A claim that is missing, is not a list of non-empty names, or is replaced by an overage pointer
   gives no groups.
 
-The server refuses to start when a mapping grants `admin`, scopes a custom role or target without
+The server refuses to start when a mapping grants `admin`, scopes a project role or target without
 a location, has a blank group, or when `group_roles` is set without `groups_claim`.
 
 Servers that share one control database must use identical OIDC and `group_roles` configuration.
@@ -122,7 +122,8 @@ stderr; a failed request prints no SQL to stdout. Permission denials include the
 The equivalent person-authenticated API is a `GET` request to
 `/api/control/locations/{location_id}/targets/{target}/nodes/{node_key}/sql`.
 
-Developer or Admin access must cover the requested location and target. Viewer, Operator and
+Developer or Admin access must cover the requested location and target, or a
+[project role](#define-project-roles) must grant `view.sql` on the node. Viewer, Operator and
 the configurable default Viewer role do not grant SQL inspection. The response contains the
 snapshot ID, exact node key, raw SQL and compiled SQL. Missing SQL fields are `null`; an absent
 current publication or node returns JSON `null` after authorization.
@@ -131,3 +132,37 @@ SQL inspection does not execute SQL or return Python source, execution artifacts
 data. Every successful inspection records an audit entry without copying SQL into it. If the
 audit cannot be saved, the request fails without returning SQL. Revocation takes effect on the
 next request, including sessions created by [`sqb login`](/docs/cli/login/).
+
+## Define project roles
+
+Put project roles in `roles.toml` at the project root. `sqb location publish` resolves each
+grant's node selectors against the published project:
+
+```toml
+[roles.orders_support]
+description = "Answer order questions"
+
+[[roles.orders_support.grants]]
+nodes = ["tag:orders"]
+permissions = ["view.sql", "run.adhoc"]
+
+[[roles.orders_support.grants]]
+nodes = ["tag:orders,tag:pii"]
+permissions = ["view.evidence"]
+```
+
+| Rule | Behaviour |
+| --- | --- |
+| Selectors | Same grammar as a job's `select`; an unknown tag or name fails publication |
+| No matching nodes | Publication succeeds with a warning; the grant gives nothing |
+| Permissions | Any non-administrative action; `user.manage`, `role.assign` and `audit.view` are rejected |
+| Runs | `job.run`, `run.adhoc`, `run.rerun` and `run.rewrite_history` need the permission on every node the run writes |
+| Triggers | `trigger.toggle` and `sensor.reset_cursor` need the permission on every node the trigger's job writes |
+| Listings | The catalog, statistics and run list need `view` on every node of the current publication; partial coverage is denied, not filtered |
+| Several roles | Grants add up across a person's roles |
+
+Assign a project role with `POST /api/admin/roles`, using the role key as `role`. Each
+publication replaces the location's project roles: new and changed grants apply at once, and
+removed roles stop applying. A selector covers newly matching nodes as soon as they are
+published. Each change appears in the publication's change list as `role:<key>`. A denial lists
+the nodes the request was missing.