policy: chassis core — omnigraph-policy crate + Omnigraph::enforce() (MR-722) (#102)

PR #2 of the policy chassis series (PR #1 = MR-731, merged in #101).
The structural fix that moves Cedar enforcement from HTTP-only to
engine-wide. apply_schema is the proof-of-concept writer; PR #3 fans
the enforce() call out to the remaining six (mutate_as, load,
ingest_as, branch_create_from, branch_delete, branch_merge).

## What lands

### New crate: omnigraph-policy

The 844-line policy.rs moves from `omnigraph-server` into a new
`omnigraph-policy` workspace crate so both engine and server can
depend on it. Cedar dependency moves with it. The server's policy.rs
becomes a re-export shim (`pub use omnigraph_policy::*`) so existing
`omnigraph_server::PolicyAction` etc. paths keep working — CLI and
test consumers don't have to migrate in one go.

### New trait: PolicyChecker

```rust
pub trait PolicyChecker: Send + Sync {
    fn check(&self, action: PolicyAction, scope: &ResourceScope,
             actor: &str) -> Result<(), PolicyError>;
}
```

`PolicyEngine` (Cedar-backed) implements it. `Omnigraph::with_policy()`
takes `Arc<dyn PolicyChecker>`. Engine tests mock the trait without
spinning up Cedar. MR-725 will extend the trait with `predicate_for()`
for query-layer pushdown — additive, no call-site changes.

### New enum: ResourceScope

Four variants — Graph, Branch, TargetBranch, BranchTransition —
mapping cleanly to today's `(branch, target_branch)` shape on
PolicyRequest via `to_branch_pair()`. Each engine writer picks the
variant that matches the existing HTTP-layer convention so engine
and HTTP evaluate the same Cedar decision.

**Invariant**: ResourceScope stays at branch granularity. Per-type
and per-row scope are MR-725's territory, not engine-layer's.
Adding Type/Row variants here creates two places per-type policy
can be evaluated, which can drift. See chassis design refinements
comment on MR-722 (2026-05-17).

### Omnigraph::with_policy() + enforce()

* New `policy: Option<Arc<dyn PolicyChecker>>` field on Omnigraph,
  None by default (preserves embedded/dev no-enforcement mode).
* `with_policy(self, checker)` setter — builder-style, consumes self.
* `enforce(action, scope, actor)` — the gate. When policy is None,
  no-op. When policy is Some AND actor is None, hard error — silent
  bypass via "I forgot the actor" is exactly the footgun this gate
  is here to prevent.

### apply_schema_as: first writer wired

* New public method `apply_schema_as(source, options, actor)` that
  calls `enforce(SchemaApply, TargetBranch("main"), actor)` before
  acquiring the schema-apply lock or doing any other work.
* Existing `apply_schema(source)` and `apply_schema_with_options(...)`
  delegate to it with actor=None (no-actor variants).
* HTTP handler `server_schema_apply` updated to call apply_schema_as
  with the resolved actor. AppState construction injects the
  PolicyEngine into Omnigraph via `with_policy`. HTTP-layer
  authorize_request still fires first; the engine gate is the
  redundant-but-correct backstop and the only path that protects SDK
  / embedded callers. PR #3 removes the HTTP redundancy.

### OmniError::Policy

New error variant for engine-layer policy denial / evaluation
failure. ApiError::from_omni maps it to 403.

### MR-724 Admin action — Option A reservation

PolicyAction::Admin kept in the enum with a load-bearing doc
comment naming its future consumers (hot reload, audit log query,
approvals list per MR-726 / MR-732 / MR-734). No enforce(Admin, ...)
call site exists yet — the variant is reserved so the action
vocabulary is complete from chassis day one. MR-724 closes when
the first consumer surface ships.

### New SDK-side integration test

`crates/omnigraph/tests/policy_engine_chassis.rs` — four tests
covering:
* Policy denies for unauthorized actor → OmniError::Policy
* Policy permits for authorized actor → apply succeeds
* Policy installed + no actor → hard error (forget-the-actor footgun)
* No policy → no-op (embedded/dev default still works)

These exercise the engine path directly — no HTTP layer involved.

## Test results

- cargo test --workspace --locked --no-fail-fast: 851 passed, 0 failed
  * 45 server tests (existing) pass
  * 14 schema_apply tests (existing) pass
  * 4 new chassis tests pass
  * 60 OpenAPI tests pass (no HTTP API surface changes)
  * No regressions across the workspace

## Architectural decisions baked in

Per MR-722 chassis design refinements comment (2026-05-17):

1. PolicyChecker is a trait, not just a concrete. Engine and server
   consume the trait. MR-725 adds predicate_for() additively.
2. ResourceScope stays at branch granularity. No Type/Row variants.
3. Coarse-vs-fine framing pinned: engine-layer is action gate;
   query-layer (MR-725) is predicate gate. Both backed by same Cedar
   engine; non-overlapping responsibilities.
4. Admin action reserved for policy-management surfaces (MR-724
   Option A).

## Pending follow-ups (PR #3+)

- Fan-out enforce() to mutate_as, load, ingest_as, branch_create_from,
  branch_delete, branch_merge (PR #3).
- Remove HTTP-layer authorize_request redundancy once engine gate
  covers all writers (PR #3).
- CLI policy injection into Omnigraph for non-`policy validate|test|explain`
  subcommands (PR #3 or follow-up).
- MR-723 default-deny 3-state matrix (PR #4).
- MR-736 severity warn/deny (PR #5).
- AGENTS.md scope-of-enforcement rewrite once chassis fully lands.
- Coarse-vs-fine framing in docs/user/policy.md.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Cargo.lock

@@ -4661,6 +4661,7 @@ dependencies = [
  "lance-table",
  "object_store",
  "omnigraph-compiler",
+ "omnigraph-policy",
  "regex",
  "reqwest",
  "serde",
@@ -4675,6 +4676,19 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "omnigraph-policy"
+version = "0.4.2"
+dependencies = [
+ "cedar-policy",
+ "clap",
+ "color-eyre",
+ "serde",
+ "serde_json",
+ "serde_yaml",
+ "tempfile",
+]
+
 [[package]]
 name = "omnigraph-server"
 version = "0.4.2"
@@ -4683,7 +4697,6 @@ dependencies = [
  "aws-config",
  "aws-sdk-secretsmanager",
  "axum",
- "cedar-policy",
  "clap",
  "color-eyre",
  "dashmap",
@@ -4691,6 +4704,7 @@ dependencies = [
  "lance-index",
  "omnigraph-compiler",
  "omnigraph-engine",
+ "omnigraph-policy",
  "serde",
  "serde_json",
  "serde_yaml",

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "crates/omnigraph-compiler",
     "crates/omnigraph",
     "crates/omnigraph-cli",
+    "crates/omnigraph-policy",
     "crates/omnigraph-server",
 ]
 default-members = [

crates/omnigraph-policy/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "omnigraph-policy"
+version = "0.4.2"
+edition = "2024"
+description = "Policy / authorization layer for Omnigraph — Cedar-backed PolicyEngine, PolicyChecker trait, ResourceScope enum."
+license = "MIT"
+repository = "https://github.com/ModernRelay/omnigraph"
+homepage = "https://github.com/ModernRelay/omnigraph"
+documentation = "https://docs.rs/omnigraph-policy"
+
+[dependencies]
+cedar-policy = { workspace = true }
+clap = { workspace = true }
+color-eyre = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+serde_yaml = { workspace = true }
+
+[dev-dependencies]
+tempfile = { workspace = true }

crates/omnigraph-server/Cargo.toml

@@ -21,6 +21,7 @@ aws = ["dep:aws-config", "dep:aws-sdk-secretsmanager"]
 [dependencies]
 omnigraph = { package = "omnigraph-engine", path = "../omnigraph", version = "0.4.2" }
 omnigraph-compiler = { path = "../omnigraph-compiler", version = "0.4.2" }
+omnigraph-policy = { path = "../omnigraph-policy", version = "0.4.2" }
 axum = { workspace = true }
 clap = { workspace = true }
 color-eyre = { workspace = true }
@@ -32,7 +33,6 @@ tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 tower-http = { workspace = true }
 utoipa = { workspace = true }
-cedar-policy = { workspace = true }
 futures = { workspace = true }
 sha2 = { workspace = true }
 subtle = { workspace = true }

crates/omnigraph-server/src/lib.rs

@@ -197,12 +197,29 @@ impl AppState {
             .into_iter()
             .map(|(actor, token)| (hash_bearer_token(&token), Arc::<str>::from(actor)))
             .collect();
+        let policy_engine: Option<Arc<PolicyEngine>> = policy_engine.map(Arc::new);
+        // MR-722 chassis: inject the policy checker into the engine so
+        // `Omnigraph::apply_schema_as` (and PR #3's fan-out of the
+        // remaining writers) gates at engine-layer too. HTTP-layer
+        // `authorize_request` still fires first; the engine-layer gate
+        // is the redundant-but-correct backstop, plus the only path
+        // that protects SDK / embedded callers. PR #3 removes the HTTP
+        // redundancy once we're confident the engine gate covers it.
+        let db = if let Some(engine) = policy_engine.as_ref() {
+            // Unsizing coercion: Arc<PolicyEngine> → Arc<dyn PolicyChecker>.
+            // Needs the explicit `as` cast — Rust 2024 doesn't infer it through
+            // `Arc::clone`.
+            let checker = Arc::clone(engine) as Arc<dyn omnigraph_policy::PolicyChecker>;
+            db.with_policy(checker)
+        } else {
+            db
+        };
         Self {
             uri,
             engine: Arc::new(db),
             workload: Arc::new(workload::WorkloadController::from_env()),
             bearer_tokens: Arc::from(bearer_tokens),
-            policy_engine: policy_engine.map(Arc::new),
+            policy_engine,
         }
     }
 
@@ -443,6 +460,13 @@ impl ApiError {
             ),
             OmniError::Lance(message) => Self::internal(format!("storage: {message}")),
             OmniError::Io(err) => Self::internal(format!("io: {err}")),
+            // Engine-layer policy enforcement (MR-722). All denials and
+            // evaluation failures surface here as 403. The HTTP-layer
+            // `authorize_request` already distinguishes 401 (missing
+            // bearer) from 403 (policy denial), so by the time the
+            // engine gate fires, the bearer is valid — any failure from
+            // the engine is a policy outcome, not an auth one.
+            OmniError::Policy(message) => Self::forbidden(message),
         }
     }
 }
@@ -1077,9 +1101,19 @@ async fn server_schema_apply(
         .map_err(ApiError::from_workload_reject)?;
     let result = {
         let db = &state.engine;
-        db.apply_schema(&request.schema_source)
-            .await
-            .map_err(ApiError::from_omni)?
+        // Engine-layer policy enforcement (MR-722): pass the resolved
+        // actor through so apply_schema_as can call enforce() with the
+        // authoritative identity. With a policy installed in AppState,
+        // engine-side enforcement re-checks the same decision the
+        // HTTP-layer authorize_request just made above. PR #3 collapses
+        // the redundancy.
+        db.apply_schema_as(
+            &request.schema_source,
+            omnigraph::db::SchemaApplyOptions::default(),
+            actor_id,
+        )
+        .await
+        .map_err(ApiError::from_omni)?
     };
     Ok(Json(schema_apply_output(state.uri(), result)))
 }

crates/omnigraph-server/src/policy.rs

@@ -1,844 +1,8 @@
-use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
-use std::fmt;
-use std::fs;
-use std::path::Path;
-use std::str::FromStr;
-
-use cedar_policy::{
-    Authorizer, Context, Decision, Entities, Entity, EntityId, EntityTypeName, EntityUid, Policy,
-    PolicyId, PolicySet, Request, Schema, ValidationMode, Validator,
-};
-use clap::ValueEnum;
-use color_eyre::eyre::{Result, bail, eyre};
-use serde::{Deserialize, Serialize};
-use serde_json::json;
-
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, Serialize, Deserialize, ValueEnum)]
-#[serde(rename_all = "snake_case")]
-pub enum PolicyAction {
-    Read,
-    Export,
-    Change,
-    SchemaApply,
-    BranchCreate,
-    BranchDelete,
-    BranchMerge,
-    Admin,
-}
-
-impl PolicyAction {
-    pub fn as_str(self) -> &'static str {
-        match self {
-            Self::Read => "read",
-            Self::Export => "export",
-            Self::Change => "change",
-            Self::SchemaApply => "schema_apply",
-            Self::BranchCreate => "branch_create",
-            Self::BranchDelete => "branch_delete",
-            Self::BranchMerge => "branch_merge",
-            Self::Admin => "admin",
-        }
-    }
-
-    fn uses_branch_scope(self) -> bool {
-        matches!(self, Self::Read | Self::Export | Self::Change)
-    }
-
-    fn uses_target_branch_scope(self) -> bool {
-        matches!(
-            self,
-            Self::BranchCreate | Self::SchemaApply | Self::BranchDelete | Self::BranchMerge
-        )
-    }
-}
-
-impl fmt::Display for PolicyAction {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str(self.as_str())
-    }
-}
-
-impl FromStr for PolicyAction {
-    type Err = color_eyre::eyre::Error;
-
-    fn from_str(value: &str) -> Result<Self> {
-        match value.trim() {
-            "read" => Ok(Self::Read),
-            "export" => Ok(Self::Export),
-            "change" => Ok(Self::Change),
-            "schema_apply" => Ok(Self::SchemaApply),
-            "branch_create" => Ok(Self::BranchCreate),
-            "branch_delete" => Ok(Self::BranchDelete),
-            "branch_merge" => Ok(Self::BranchMerge),
-            "admin" => Ok(Self::Admin),
-            other => bail!("unknown policy action '{other}'"),
-        }
-    }
-}
-
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum PolicyBranchScope {
-    Any,
-    Protected,
-    Unprotected,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct PolicyActorSelector {
-    pub group: String,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct PolicyAllowRule {
-    pub actors: PolicyActorSelector,
-    pub actions: Vec<PolicyAction>,
-    pub branch_scope: Option<PolicyBranchScope>,
-    pub target_branch_scope: Option<PolicyBranchScope>,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct PolicyRule {
-    pub id: String,
-    pub allow: PolicyAllowRule,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct PolicyConfig {
-    pub version: u32,
-    #[serde(default)]
-    pub groups: BTreeMap<String, Vec<String>>,
-    #[serde(default)]
-    pub protected_branches: Vec<String>,
-    #[serde(default)]
-    pub rules: Vec<PolicyRule>,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct PolicyTestConfig {
-    pub version: u32,
-    #[serde(default)]
-    pub cases: Vec<PolicyTestCase>,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct PolicyTestCase {
-    pub id: String,
-    pub actor: String,
-    pub action: PolicyAction,
-    pub branch: Option<String>,
-    pub target_branch: Option<String>,
-    pub expect: PolicyExpectation,
-}
-
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum PolicyExpectation {
-    Allow,
-    Deny,
-}
-
-#[derive(Debug, Clone)]
-pub struct PolicyRequest {
-    pub actor_id: String,
-    pub action: PolicyAction,
-    pub branch: Option<String>,
-    pub target_branch: Option<String>,
-}
-
-#[derive(Debug, Clone)]
-pub struct PolicyDecision {
-    pub allowed: bool,
-    pub matched_rule_id: Option<String>,
-    pub message: String,
-}
-
-pub struct PolicyCompiler;
-
-#[derive(Clone)]
-pub struct PolicyEngine {
-    repo_id: String,
-    protected_branches: BTreeSet<String>,
-    known_actors: BTreeSet<String>,
-    schema: Schema,
-    entities: Entities,
-    policies: PolicySet,
-    policy_to_rule: HashMap<String, String>,
-}
-
-impl PolicyConfig {
-    pub fn load(path: &Path) -> Result<Self> {
-        let config: Self = serde_yaml::from_str(&fs::read_to_string(path)?)?;
-        config.validate()?;
-        Ok(config)
-    }
-
-    pub fn validate(&self) -> Result<()> {
-        if self.version != 1 {
-            bail!("policy version must be 1");
-        }
-
-        for (group, members) in &self.groups {
-            if group.trim().is_empty() {
-                bail!("policy group names must not be blank");
-            }
-            if members.is_empty() {
-                bail!("policy group '{group}' must not be empty");
-            }
-            for actor in members {
-                if actor.trim().is_empty() {
-                    bail!("policy group '{group}' contains a blank actor id");
-                }
-            }
-        }
-
-        for branch in &self.protected_branches {
-            if branch.trim().is_empty() {
-                bail!("protected branch names must not be blank");
-            }
-        }
-
-        let mut seen_rule_ids = HashSet::new();
-        for rule in &self.rules {
-            if rule.id.trim().is_empty() {
-                bail!("policy rule ids must not be blank");
-            }
-            if !seen_rule_ids.insert(rule.id.clone()) {
-                bail!("duplicate policy rule id '{}'", rule.id);
-            }
-            if rule.allow.actors.group.trim().is_empty() {
-                bail!("policy rule '{}' must reference a non-blank group", rule.id);
-            }
-            if !self.groups.contains_key(rule.allow.actors.group.as_str()) {
-                bail!(
-                    "policy rule '{}' references unknown group '{}'",
-                    rule.id,
-                    rule.allow.actors.group
-                );
-            }
-            if rule.allow.actions.is_empty() {
-                bail!("policy rule '{}' must include at least one action", rule.id);
-            }
-            if rule.allow.branch_scope.is_some() && rule.allow.target_branch_scope.is_some() {
-                bail!(
-                    "policy rule '{}' may specify branch_scope or target_branch_scope, not both",
-                    rule.id
-                );
-            }
-            if let Some(_) = rule.allow.branch_scope {
-                for action in &rule.allow.actions {
-                    if !action.uses_branch_scope() {
-                        bail!(
-                            "policy rule '{}' uses branch_scope with unsupported action '{}'",
-                            rule.id,
-                            action
-                        );
-                    }
-                }
-            }
-            if let Some(_) = rule.allow.target_branch_scope {
-                for action in &rule.allow.actions {
-                    if !action.uses_target_branch_scope() {
-                        bail!(
-                            "policy rule '{}' uses target_branch_scope with unsupported action '{}'",
-                            rule.id,
-                            action
-                        );
-                    }
-                }
-            }
-        }
-
-        Ok(())
-    }
-}
-
-impl PolicyTestConfig {
-    pub fn load(path: &Path) -> Result<Self> {
-        let config: Self = serde_yaml::from_str(&fs::read_to_string(path)?)?;
-        if config.version != 1 {
-            bail!("policy test version must be 1");
-        }
-        let mut seen = HashSet::new();
-        for case in &config.cases {
-            if case.id.trim().is_empty() {
-                bail!("policy test case ids must not be blank");
-            }
-            if !seen.insert(case.id.clone()) {
-                bail!("duplicate policy test case id '{}'", case.id);
-            }
-            if case.actor.trim().is_empty() {
-                bail!("policy test case '{}' must not use a blank actor", case.id);
-            }
-        }
-        Ok(config)
-    }
-}
-
-impl PolicyCompiler {
-    pub fn compile(config: &PolicyConfig, repo_id: &str) -> Result<PolicyEngine> {
-        config.validate()?;
-        let (schema, schema_warnings) = Schema::from_cedarschema_str(policy_schema_source())?;
-        let schema_warnings = schema_warnings
-            .map(|warning| warning.to_string())
-            .collect::<Vec<_>>();
-        if !schema_warnings.is_empty() {
-            bail!("policy schema warnings:\n{}", schema_warnings.join("\n"));
-        }
-        let entities = compile_entities(config, repo_id, &schema)?;
-        let (policies, policy_to_rule) = compile_policies(config, repo_id)?;
-        let validator = Validator::new(schema.clone());
-        let validation = validator.validate(&policies, ValidationMode::Strict);
-        let errors = validation
-            .validation_errors()
-            .map(|err| err.to_string())
-            .collect::<Vec<_>>();
-        if !errors.is_empty() {
-            bail!("policy validation failed:\n{}", errors.join("\n"));
-        }
-
-        let known_actors = config
-            .groups
-            .values()
-            .flat_map(|members| members.iter().cloned())
-            .collect();
-        Ok(PolicyEngine {
-            repo_id: repo_id.to_string(),
-            protected_branches: config.protected_branches.iter().cloned().collect(),
-            known_actors,
-            schema,
-            entities,
-            policies,
-            policy_to_rule,
-        })
-    }
-}
-
-impl PolicyEngine {
-    pub fn load(path: &Path, repo_id: &str) -> Result<Self> {
-        let config = PolicyConfig::load(path)?;
-        PolicyCompiler::compile(&config, repo_id)
-    }
-
-    pub fn authorize(&self, request: &PolicyRequest) -> Result<PolicyDecision> {
-        if !self.known_actors.contains(request.actor_id.as_str()) {
-            return Ok(self.deny(
-                request,
-                None,
-                format!(
-                    "policy denied action '{}' for unknown actor '{}'",
-                    request.action, request.actor_id
-                ),
-            ));
-        }
-
-        let principal = entity_uid("Actor", &request.actor_id)?;
-        let action = entity_uid("Action", request.action.as_str())?;
-        let resource = entity_uid("Repo", &self.repo_id)?;
-        let context_value = json!({
-            "has_branch": request.branch.is_some(),
-            "branch": request.branch.clone().unwrap_or_default(),
-            "has_target_branch": request.target_branch.is_some(),
-            "target_branch": request.target_branch.clone().unwrap_or_default(),
-            "branch_is_protected": request.branch.as_ref().is_some_and(|branch| self.protected_branches.contains(branch)),
-            "target_branch_is_protected": request.target_branch.as_ref().is_some_and(|branch| self.protected_branches.contains(branch)),
-        });
-        let context = Context::from_json_value(context_value, Some((&self.schema, &action)))?;
-        let cedar_request = Request::new(principal, action, resource, context, Some(&self.schema))?;
-        let response =
-            Authorizer::new().is_authorized(&cedar_request, &self.policies, &self.entities);
-        let errors = response
-            .diagnostics()
-            .errors()
-            .map(|err| err.to_string())
-            .collect::<Vec<_>>();
-        if !errors.is_empty() {
-            bail!("policy evaluation failed:\n{}", errors.join("\n"));
-        }
-
-        let matched_rule_id = response
-            .diagnostics()
-            .reason()
-            .filter_map(|policy_id| {
-                let key: &str = policy_id.as_ref();
-                self.policy_to_rule.get(key).cloned()
-            })
-            .min();
-
-        Ok(match response.decision() {
-            Decision::Allow => PolicyDecision {
-                allowed: true,
-                matched_rule_id: matched_rule_id.clone(),
-                message: format!(
-                    "policy allowed action '{}' for actor '{}'",
-                    request.action, request.actor_id
-                ),
-            },
-            Decision::Deny => {
-                let message = format!(
-                    "policy denied action '{}'{}{} for actor '{}'",
-                    request.action,
-                    request
-                        .branch
-                        .as_deref()
-                        .map(|branch| format!(" on branch '{}'", branch))
-                        .unwrap_or_default(),
-                    request
-                        .target_branch
-                        .as_deref()
-                        .map(|branch| format!(" targeting branch '{}'", branch))
-                        .unwrap_or_default(),
-                    request.actor_id
-                );
-                self.deny(request, matched_rule_id, message)
-            }
-        })
-    }
-
-    pub fn validate_request(&self, request: &PolicyRequest) -> Result<()> {
-        let _ = self.authorize(request)?;
-        Ok(())
-    }
-
-    pub fn run_tests(&self, tests: &PolicyTestConfig) -> Result<()> {
-        if tests.version != 1 {
-            bail!("policy test version must be 1");
-        }
-        let mut failures = Vec::new();
-        for case in &tests.cases {
-            let decision = self.authorize(&PolicyRequest {
-                actor_id: case.actor.clone(),
-                action: case.action,
-                branch: case.branch.clone(),
-                target_branch: case.target_branch.clone(),
-            })?;
-            let expected_allowed = matches!(case.expect, PolicyExpectation::Allow);
-            if decision.allowed != expected_allowed {
-                failures.push(format!(
-                    "{}: expected {:?} but got {}",
-                    case.id,
-                    case.expect,
-                    if decision.allowed { "allow" } else { "deny" }
-                ));
-            }
-        }
-        if failures.is_empty() {
-            Ok(())
-        } else {
-            bail!("policy tests failed:\n{}", failures.join("\n"))
-        }
-    }
-
-    pub fn known_actor_count(&self) -> usize {
-        self.known_actors.len()
-    }
-
-    fn deny(
-        &self,
-        _request: &PolicyRequest,
-        matched_rule_id: Option<String>,
-        message: String,
-    ) -> PolicyDecision {
-        PolicyDecision {
-            allowed: false,
-            matched_rule_id,
-            message,
-        }
-    }
-}
-
-fn compile_entities(config: &PolicyConfig, repo_id: &str, schema: &Schema) -> Result<Entities> {
-    let mut group_entities = Vec::new();
-    for group in config.groups.keys() {
-        group_entities.push(Entity::new(
-            entity_uid("Group", group)?,
-            HashMap::new(),
-            HashSet::<EntityUid>::new(),
-        )?);
-    }
-
-    let mut actor_groups: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
-    for (group, members) in &config.groups {
-        for actor in members {
-            actor_groups
-                .entry(actor.clone())
-                .or_default()
-                .insert(group.clone());
-        }
-    }
-
-    let mut actor_entities = Vec::new();
-    for (actor, groups) in actor_groups {
-        let parents = groups
-            .iter()
-            .map(|group| entity_uid("Group", group))
-            .collect::<Result<HashSet<_>>>()?;
-        actor_entities.push(Entity::new(
-            entity_uid("Actor", &actor)?,
-            HashMap::new(),
-            parents,
-        )?);
-    }
-
-    let repo_entity = Entity::new(
-        entity_uid("Repo", repo_id)?,
-        HashMap::new(),
-        HashSet::<EntityUid>::new(),
-    )?;
-
-    let mut entities = Vec::new();
-    entities.extend(group_entities);
-    entities.extend(actor_entities);
-    entities.push(repo_entity);
-    Ok(Entities::from_entities(entities, Some(schema))?)
-}
-
-fn compile_policies(
-    config: &PolicyConfig,
-    repo_id: &str,
-) -> Result<(PolicySet, HashMap<String, String>)> {
-    let mut policies = Vec::new();
-    let mut policy_to_rule = HashMap::new();
-
-    for rule in &config.rules {
-        for action in &rule.allow.actions {
-            let policy_id = PolicyId::new(format!("{}:{}", rule.id, action.as_str()));
-            let source = compile_policy_source(rule, action, repo_id);
-            let policy = Policy::parse(Some(policy_id.clone()), source.as_str())?;
-            policy_to_rule.insert(policy_id.to_string(), rule.id.clone());
-            policies.push(policy);
-        }
-    }
-
-    Ok((PolicySet::from_policies(policies)?, policy_to_rule))
-}
-
-fn compile_policy_source(rule: &PolicyRule, action: &PolicyAction, repo_id: &str) -> String {
-    let mut conditions = Vec::new();
-    if let Some(scope) = rule.allow.branch_scope {
-        conditions.push(branch_scope_condition(scope));
-    }
-    if let Some(scope) = rule.allow.target_branch_scope {
-        conditions.push(target_branch_scope_condition(scope));
-    }
-
-    let when = if conditions.is_empty() {
-        String::new()
-    } else {
-        format!("\nwhen {{ {} }}", conditions.join(" && "))
-    };
-
-    format!(
-        r#"permit (
-    principal in Omnigraph::Group::{group},
-    action == Omnigraph::Action::{action},
-    resource == Omnigraph::Repo::{repo}
-){when};"#,
-        group = cedar_literal(&rule.allow.actors.group),
-        action = cedar_literal(action.as_str()),
-        repo = cedar_literal(repo_id),
-        when = when,
-    )
-}
-
-fn branch_scope_condition(scope: PolicyBranchScope) -> String {
-    match scope {
-        PolicyBranchScope::Any => "true".to_string(),
-        PolicyBranchScope::Protected => {
-            "context.has_branch && context.branch_is_protected".to_string()
-        }
-        PolicyBranchScope::Unprotected => {
-            "context.has_branch && context.branch_is_protected == false".to_string()
-        }
-    }
-}
-
-fn target_branch_scope_condition(scope: PolicyBranchScope) -> String {
-    match scope {
-        PolicyBranchScope::Any => "true".to_string(),
-        PolicyBranchScope::Protected => {
-            "context.has_target_branch && context.target_branch_is_protected".to_string()
-        }
-        PolicyBranchScope::Unprotected => {
-            "context.has_target_branch && context.target_branch_is_protected == false".to_string()
-        }
-    }
-}
-
-fn policy_schema_source() -> &'static str {
-    r#"
-namespace Omnigraph {
-    type RequestContext = {
-        has_branch: Bool,
-        branch: String,
-        has_target_branch: Bool,
-        target_branch: String,
-        branch_is_protected: Bool,
-        target_branch_is_protected: Bool,
-    };
-
-    entity Actor in [Group];
-    entity Group;
-    entity Repo;
-
-    action "read" appliesTo { principal: Actor, resource: Repo, context: RequestContext };
-    action "export" appliesTo { principal: Actor, resource: Repo, context: RequestContext };
-    action "change" appliesTo { principal: Actor, resource: Repo, context: RequestContext };
-    action "schema_apply" appliesTo { principal: Actor, resource: Repo, context: RequestContext };
-    action "branch_create" appliesTo { principal: Actor, resource: Repo, context: RequestContext };
-    action "branch_delete" appliesTo { principal: Actor, resource: Repo, context: RequestContext };
-    action "branch_merge" appliesTo { principal: Actor, resource: Repo, context: RequestContext };
-    action "admin" appliesTo { principal: Actor, resource: Repo, context: RequestContext };
-}
-"#
-}
-
-fn entity_uid(entity_type: &str, id: &str) -> Result<EntityUid> {
-    let typename = EntityTypeName::from_str(&format!("Omnigraph::{entity_type}"))?;
-    let entity_id = EntityId::from_str(id).map_err(|err| eyre!(err.to_string()))?;
-    Ok(EntityUid::from_type_name_and_id(typename, entity_id))
-}
-
-fn cedar_literal(value: &str) -> String {
-    serde_json::to_string(value).expect("string literal should serialize")
-}
-
-impl PolicyRequest {
-    pub fn actor_id(&self) -> &str {
-        &self.actor_id
-    }
-
-    pub fn action(&self) -> PolicyAction {
-        self.action
-    }
-
-    pub fn branch(&self) -> Option<&str> {
-        self.branch.as_deref()
-    }
-
-    pub fn target_branch(&self) -> Option<&str> {
-        self.target_branch.as_deref()
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::{
-        PolicyAction, PolicyCompiler, PolicyConfig, PolicyExpectation, PolicyRequest,
-        PolicyTestCase, PolicyTestConfig,
-    };
-
-    #[test]
-    fn rejects_duplicate_rule_ids() {
-        let policy: PolicyConfig = serde_yaml::from_str(
-            r#"
-version: 1
-groups:
-  team: [act-andrew]
-rules:
-  - id: same
-    allow:
-      actors: { group: team }
-      actions: [read]
-      branch_scope: any
-  - id: same
-    allow:
-      actors: { group: team }
-      actions: [export]
-      branch_scope: any
-"#,
-        )
-        .unwrap();
-
-        let err = policy.validate().unwrap_err();
-        assert!(err.to_string().contains("duplicate policy rule id"));
-    }
-
-    #[test]
-    fn rejects_unknown_group_references() {
-        let policy: PolicyConfig = serde_yaml::from_str(
-            r#"
-version: 1
-groups:
-  team: [act-andrew]
-rules:
-  - id: bad
-    allow:
-      actors: { group: admins }
-      actions: [read]
-      branch_scope: any
-"#,
-        )
-        .unwrap();
-
-        let err = policy.validate().unwrap_err();
-        assert!(err.to_string().contains("references unknown group"));
-    }
-
-    #[test]
-    fn rejects_invalid_scope_action_combinations() {
-        let policy: PolicyConfig = serde_yaml::from_str(
-            r#"
-version: 1
-groups:
-  team: [act-andrew]
-rules:
-  - id: bad
-    allow:
-      actors: { group: team }
-      actions: [branch_merge]
-      branch_scope: protected
-"#,
-        )
-        .unwrap();
-
-        let err = policy.validate().unwrap_err();
-        assert!(err.to_string().contains("unsupported action"));
-    }
-
-    #[test]
-    fn compiles_and_authorizes_branch_and_target_rules() {
-        let policy: PolicyConfig = serde_yaml::from_str(
-            r#"
-version: 1
-groups:
-  team: [act-andrew, act-bruno]
-  admins: [act-andrew]
-protected_branches: [main]
-rules:
-  - id: team-read
-    allow:
-      actors: { group: team }
-      actions: [read, export]
-      branch_scope: any
-  - id: team-write
-    allow:
-      actors: { group: team }
-      actions: [change]
-      branch_scope: unprotected
-  - id: admins-promote
-    allow:
-      actors: { group: admins }
-      actions: [branch_delete, branch_merge]
-      target_branch_scope: protected
-"#,
-        )
-        .unwrap();
-
-        let engine = PolicyCompiler::compile(&policy, "repo").unwrap();
-        let allow = engine
-            .authorize(&PolicyRequest {
-                actor_id: "act-bruno".to_string(),
-                action: PolicyAction::Change,
-                branch: Some("feature".to_string()),
-                target_branch: None,
-            })
-            .unwrap();
-        assert!(allow.allowed);
-        assert_eq!(allow.matched_rule_id.as_deref(), Some("team-write"));
-
-        let deny = engine
-            .authorize(&PolicyRequest {
-                actor_id: "act-bruno".to_string(),
-                action: PolicyAction::BranchDelete,
-                branch: None,
-                target_branch: Some("main".to_string()),
-            })
-            .unwrap();
-        assert!(!deny.allowed);
-
-        let admin = engine
-            .authorize(&PolicyRequest {
-                actor_id: "act-andrew".to_string(),
-                action: PolicyAction::BranchDelete,
-                branch: None,
-                target_branch: Some("main".to_string()),
-            })
-            .unwrap();
-        assert!(admin.allowed);
-        assert_eq!(admin.matched_rule_id.as_deref(), Some("admins-promote"));
-    }
-
-    #[test]
-    fn policy_tests_enforce_expected_outcomes() {
-        let policy: PolicyConfig = serde_yaml::from_str(
-            r#"
-version: 1
-groups:
-  team: [act-andrew]
-protected_branches: [main]
-rules:
-  - id: team-read
-    allow:
-      actors: { group: team }
-      actions: [read]
-      branch_scope: any
-"#,
-        )
-        .unwrap();
-        let engine = PolicyCompiler::compile(&policy, "repo").unwrap();
-        let tests = PolicyTestConfig {
-            version: 1,
-            cases: vec![
-                PolicyTestCase {
-                    id: "allow-read".to_string(),
-                    actor: "act-andrew".to_string(),
-                    action: PolicyAction::Read,
-                    branch: Some("main".to_string()),
-                    target_branch: None,
-                    expect: PolicyExpectation::Allow,
-                },
-                PolicyTestCase {
-                    id: "deny-change".to_string(),
-                    actor: "act-andrew".to_string(),
-                    action: PolicyAction::Change,
-                    branch: Some("main".to_string()),
-                    target_branch: None,
-                    expect: PolicyExpectation::Deny,
-                },
-            ],
-        };
-
-        engine.run_tests(&tests).unwrap();
-    }
-
-    #[test]
-    fn schema_apply_uses_target_branch_scope() {
-        let policy: PolicyConfig = serde_yaml::from_str(
-            r#"
-version: 1
-groups:
-  admins: [act-ragnor]
-protected_branches: [main]
-rules:
-  - id: admins-schema-apply
-    allow:
-      actors: { group: admins }
-      actions: [schema_apply]
-      target_branch_scope: protected
-"#,
-        )
-        .unwrap();
-
-        let engine = PolicyCompiler::compile(&policy, "repo").unwrap();
-        let allow = engine
-            .authorize(&PolicyRequest {
-                actor_id: "act-ragnor".to_string(),
-                action: PolicyAction::SchemaApply,
-                branch: None,
-                target_branch: Some("main".to_string()),
-            })
-            .unwrap();
-        assert!(allow.allowed);
-
-        let deny = engine
-            .authorize(&PolicyRequest {
-                actor_id: "act-ragnor".to_string(),
-                action: PolicyAction::SchemaApply,
-                branch: None,
-                target_branch: Some("feature".to_string()),
-            })
-            .unwrap();
-        assert!(!deny.allowed);
-    }
-}
+// Module shim: PolicyEngine moved to the omnigraph-policy workspace crate
+// (MR-722 chassis core). The re-exports below preserve the existing
+// `omnigraph_server::policy::*` paths so call sites (CLI, tests,
+// downstream consumers) don't have to change in one go. Direct callers
+// should migrate to `omnigraph_policy::*` over time; this shim can
+// be removed once that migration completes.
+
+pub use omnigraph_policy::*;

crates/omnigraph/Cargo.toml

@@ -17,6 +17,7 @@ failpoints = ["dep:fail", "fail/failpoints"]
 
 [dependencies]
 omnigraph-compiler = { path = "../omnigraph-compiler", version = "0.4.2" }
+omnigraph-policy = { path = "../omnigraph-policy", version = "0.4.2" }
 lance = { workspace = true }
 lance-datafusion = { workspace = true }
 datafusion = { workspace = true }

crates/omnigraph/src/db/omnigraph.rs

@@ -129,6 +129,22 @@ pub struct Omnigraph {
     /// every `self.snapshot()` and `self.ensure_commit_graph_initialized()`
     /// call inside the merge body.
     merge_exclusive: Arc<tokio::sync::Mutex<()>>,
+    /// Optional policy checker for engine-layer enforcement (MR-722).
+    /// `None` = no enforcement; mutating methods are unconditionally
+    /// allowed (this is the embedded/dev default). `Some` = every
+    /// mutating method calls `self.enforce(action, scope, actor)` at
+    /// entry; denial returns `OmniError::Policy`.
+    ///
+    /// Per chassis design (see `omnigraph_policy::PolicyChecker`), the
+    /// trait surface is deliberately coarse — action × scope × actor.
+    /// Per-row / per-type / per-column scope lives at the query layer
+    /// (MR-725), which extends the same trait with a different method.
+    /// Don't be tempted to add per-row enforcement here.
+    ///
+    /// Set via `with_policy(checker)` after construction. Today only
+    /// `apply_schema_as` consults this field (PR #2 proof-of-concept);
+    /// PR #3 fans the `enforce()` call out to the remaining writers.
+    policy: Option<Arc<dyn omnigraph_policy::PolicyChecker>>,
 }
 
 /// Whether [`Omnigraph::open`] runs the open-time recovery sweep.
@@ -185,6 +201,7 @@ impl Omnigraph {
             schema_source: Arc::new(ArcSwap::from_pointee(schema_source.to_string())),
             write_queue: Arc::new(crate::db::write_queue::WriteQueueManager::new()),
             merge_exclusive: Arc::new(tokio::sync::Mutex::new(())),
+            policy: None,
         })
     }
 
@@ -272,6 +289,7 @@ impl Omnigraph {
             schema_source: Arc::new(ArcSwap::from_pointee(schema_source)),
             write_queue: Arc::new(crate::db::write_queue::WriteQueueManager::new()),
             merge_exclusive: Arc::new(tokio::sync::Mutex::new(())),
+            policy: None,
         })
     }
 
@@ -304,6 +322,54 @@ impl Omnigraph {
         &self.root_uri
     }
 
+    /// Install a policy checker for engine-layer enforcement (MR-722).
+    /// Builder-style setter — consumes `self`, returns `Self`. Calling
+    /// this on a `Omnigraph` previously without policy enables
+    /// `enforce()` to fire at every mutating engine method that's been
+    /// wired to call it (currently `apply_schema_as`; PR #3 fans out to
+    /// the remaining writers).
+    ///
+    /// Embedded callers that don't care about authorization should
+    /// just not call this. Server / CLI callers that have loaded a
+    /// `PolicyEngine` from `policy.yaml` pass it here.
+    pub fn with_policy(mut self, checker: Arc<dyn omnigraph_policy::PolicyChecker>) -> Self {
+        self.policy = Some(checker);
+        self
+    }
+
+    /// Engine-layer policy enforcement gate (MR-722 chassis core).
+    ///
+    /// * If no policy is installed → no-op (returns `Ok(())`).
+    /// * If policy is installed AND actor is None → denial with a
+    ///   clear "no actor for engine-layer policy check" message.
+    ///   Forces server / CLI / SDK callers to thread an actor through
+    ///   when policy is configured — silent bypass via "I forgot the
+    ///   actor" is exactly the footgun this gate is here to prevent.
+    /// * If policy is installed AND actor is Some → call
+    ///   `PolicyChecker::check(action, scope, actor)`; map denial /
+    ///   internal failure to `OmniError::Policy(...)`.
+    pub(crate) fn enforce(
+        &self,
+        action: omnigraph_policy::PolicyAction,
+        scope: &omnigraph_policy::ResourceScope,
+        actor: Option<&str>,
+    ) -> Result<()> {
+        let Some(checker) = self.policy.as_ref() else {
+            return Ok(());
+        };
+        let Some(actor) = actor else {
+            return Err(OmniError::Policy(
+                "no actor for engine-layer policy check (policy is configured but the call site \
+                 didn't thread an actor through — this is almost certainly a bug, not an \
+                 intended bypass)"
+                    .to_string(),
+            ));
+        };
+        checker
+            .check(action, scope, actor)
+            .map_err(|err| OmniError::Policy(err.to_string()))
+    }
+
     pub(crate) async fn ensure_schema_state_valid(&self) -> Result<()> {
         validate_schema_contract(self.uri(), Arc::clone(&self.storage)).await
     }
@@ -322,7 +388,7 @@ impl Omnigraph {
     }
 
     pub async fn apply_schema(&self, desired_schema_source: &str) -> Result<SchemaApplyResult> {
-        self.apply_schema_with_options(desired_schema_source, SchemaApplyOptions::default())
+        self.apply_schema_as(desired_schema_source, SchemaApplyOptions::default(), None)
             .await
     }
 
@@ -331,7 +397,26 @@ impl Omnigraph {
         desired_schema_source: &str,
         options: SchemaApplyOptions,
     ) -> Result<SchemaApplyResult> {
-        schema_apply::apply_schema(self, desired_schema_source, options).await
+        self.apply_schema_as(desired_schema_source, options, None).await
+    }
+
+    /// Apply a schema migration with an explicit actor for engine-layer
+    /// policy enforcement (MR-722). When a `PolicyChecker` is installed
+    /// via [`Self::with_policy`], this method calls `enforce(SchemaApply,
+    /// Branch("main"), actor)` before any apply work happens. Denial
+    /// returns `OmniError::Policy` and leaves the manifest untouched.
+    ///
+    /// The no-actor variants (`apply_schema`, `apply_schema_with_options`)
+    /// pass `None` here. They work fine without a policy; if a policy IS
+    /// installed and actor is None, enforcement intentionally fails to
+    /// prevent silent-bypass-via-forgetting-the-actor footguns.
+    pub async fn apply_schema_as(
+        &self,
+        desired_schema_source: &str,
+        options: SchemaApplyOptions,
+        actor: Option<&str>,
+    ) -> Result<SchemaApplyResult> {
+        schema_apply::apply_schema(self, desired_schema_source, options, actor).await
     }
 
     pub(crate) async fn ensure_schema_apply_idle(&self, operation: &str) -> Result<()> {

crates/omnigraph/src/db/omnigraph/schema_apply.rs

@@ -52,7 +52,30 @@ pub(super) async fn apply_schema(
     db: &Omnigraph,
     desired_schema_source: &str,
     options: SchemaApplyOptions,
+    actor: Option<&str>,
 ) -> Result<SchemaApplyResult> {
+    // Engine-layer policy gate (MR-722 chassis core).
+    //
+    // Fires BEFORE acquiring the schema-apply lock or doing any other
+    // work. When no PolicyChecker is installed this is a no-op and
+    // the apply path behaves exactly as it did before MR-722. When
+    // a PolicyChecker IS installed and the actor is None, this is a
+    // hard error — see Omnigraph::enforce's docstring for the
+    // forget-the-actor-footgun reasoning.
+    //
+    // Scope is TargetBranch("main") to match the HTTP-layer convention
+    // for SchemaApply: branch=None, target_branch=Some("main"). Cedar
+    // policies in the wild use `target_branch_scope: protected` to
+    // gate schema applies, so the engine-layer call has to set the
+    // target_branch shape that activates that predicate. Wrong scope
+    // here = silent policy mismatch with HTTP. See
+    // `omnigraph_policy::ResourceScope::to_branch_pair` for the mapping.
+    db.enforce(
+        omnigraph_policy::PolicyAction::SchemaApply,
+        &omnigraph_policy::ResourceScope::TargetBranch("main".to_string()),
+        actor,
+    )?;
+
     acquire_schema_apply_lock(db).await?;
     let result = apply_schema_with_lock(db, desired_schema_source, options).await;
     let release_result = release_schema_apply_lock(db).await;

crates/omnigraph/src/error.rs

@@ -85,6 +85,13 @@ pub enum OmniError {
     Manifest(ManifestError),
     #[error("merge conflicts: {0:?}")]
     MergeConflicts(Vec<MergeConflict>),
+    /// Engine-layer policy enforcement (MR-722). Wraps either a policy
+    /// denial ("you can't do that") or a policy-evaluation failure
+    /// ("the policy engine itself blew up"). The HTTP layer maps
+    /// denials to 403 and evaluation failures to 500; CLI and embedded
+    /// callers can match on this variant directly.
+    #[error("policy: {0}")]
+    Policy(String),
 }
 
 impl OmniError {

crates/omnigraph/tests/policy_engine_chassis.rs

@@ -0,0 +1,129 @@
+//! Engine-layer policy enforcement (MR-722 chassis core, PR #2).
+//!
+//! These tests exercise `Omnigraph::with_policy()` + `apply_schema_as()`
+//! via the SDK directly — *no HTTP layer involved*. They're the proof
+//! that engine-layer enforcement works for embedded callers and CLI
+//! direct-engine writes, not just server requests.
+//!
+//! `apply_schema_as` is the only writer wired in PR #2; PR #3 fans the
+//! `enforce()` call out to the other six (`mutate_as`, `load`,
+//! `ingest_as`, `branch_create_from`, `branch_delete`, `branch_merge`).
+
+mod helpers;
+
+use std::fs;
+use std::sync::Arc;
+
+use omnigraph::db::{Omnigraph, SchemaApplyOptions};
+use omnigraph::error::OmniError;
+use omnigraph_policy::{PolicyChecker, PolicyEngine};
+
+use helpers::*;
+
+/// Cedar policy: `act-allowed` may SchemaApply; `act-denied` is in the
+/// known-actors set (so Cedar evaluates the policy, doesn't reject as
+/// unknown) but has no permit rule.
+const POLICY_YAML: &str = r#"
+version: 1
+groups:
+  schema-writers: [act-allowed]
+  readers: [act-denied]
+protected_branches: [main]
+rules:
+  - id: writers-schema-apply
+    allow:
+      actors: { group: schema-writers }
+      actions: [schema_apply]
+      target_branch_scope: any
+"#;
+
+fn additive_schema() -> String {
+    helpers::TEST_SCHEMA.replace("    age: I32?\n}", "    age: I32?\n    nickname: String?\n}")
+}
+
+async fn init_with_policy(dir: &tempfile::TempDir) -> (Omnigraph, Arc<PolicyEngine>) {
+    let db = init_and_load(dir).await;
+    let policy_path = dir.path().join("policy.yaml");
+    fs::write(&policy_path, POLICY_YAML).unwrap();
+    let engine = PolicyEngine::load(&policy_path, dir.path().to_str().unwrap()).unwrap();
+    let engine = Arc::new(engine);
+    let db = db.with_policy(Arc::clone(&engine) as Arc<dyn PolicyChecker>);
+    (db, engine)
+}
+
+#[tokio::test]
+async fn apply_schema_as_denies_when_policy_rejects_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let desired = additive_schema();
+    let result = db
+        .apply_schema_as(&desired, SchemaApplyOptions::default(), Some("act-denied"))
+        .await;
+
+    match result {
+        Err(OmniError::Policy(msg)) => {
+            assert!(
+                msg.contains("denied"),
+                "expected denial message, got: {msg}"
+            );
+        }
+        Err(other) => panic!("expected OmniError::Policy, got: {other:?}"),
+        Ok(_) => panic!("expected denial — act-denied should not be able to SchemaApply"),
+    }
+}
+
+#[tokio::test]
+async fn apply_schema_as_allows_when_policy_permits_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let desired = additive_schema();
+    let result = db
+        .apply_schema_as(&desired, SchemaApplyOptions::default(), Some("act-allowed"))
+        .await
+        .expect("act-allowed should be able to SchemaApply");
+    assert!(result.applied);
+}
+
+#[tokio::test]
+async fn apply_schema_without_actor_when_policy_is_installed_denies() {
+    // MR-722 footgun guard: if a PolicyChecker is installed AND the
+    // call site forgets to pass an actor, enforce() fails hard. Silent
+    // bypass via "I forgot the actor" is exactly what the gate is
+    // here to prevent.
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let desired = additive_schema();
+    // `apply_schema(...)` is the no-actor variant — delegates to
+    // apply_schema_as with actor=None.
+    let result = db.apply_schema(&desired).await;
+
+    match result {
+        Err(OmniError::Policy(msg)) => {
+            assert!(
+                msg.contains("no actor"),
+                "expected 'no actor' message, got: {msg}"
+            );
+        }
+        Err(other) => panic!("expected OmniError::Policy('no actor ...'), got: {other:?}"),
+        Ok(_) => panic!("expected denial — policy is installed but no actor was threaded"),
+    }
+}
+
+#[tokio::test]
+async fn apply_schema_without_policy_still_works() {
+    // Baseline: when no policy is installed (the embedded/dev default),
+    // apply_schema and apply_schema_as both work regardless of whether
+    // an actor is passed. The enforce() gate is a strict no-op in this
+    // shape — proves PR #2 doesn't regress the no-policy path.
+    let dir = tempfile::tempdir().unwrap();
+    let db = init_and_load(&dir).await;
+
+    let desired = additive_schema();
+    // No-actor variant.
+    db.apply_schema(&desired)
+        .await
+        .expect("no policy → no enforcement → apply succeeds");
+}