policy: chassis fan-out — _as variants on the remaining 6 writers (MR-722) (#103)

PR #102 wired apply_schema_as. This PR completes the chassis-side
coverage so every public mutating engine entry point hits the same
Omnigraph::enforce(action, scope, actor) gate regardless of transport:

- mutate_as → enforce(Change, Branch(branch), actor)
- load_as → enforce(Change, Branch(branch), actor)
- ingest_as → enforce(Change, Branch(branch), actor); also threads
  actor through the implicit branch_create_from_as so fresh-branch
  ingest correctly hits BranchCreate too
- branch_create_as → enforce(BranchCreate, TargetBranch(name), actor)
- branch_create_from_as → enforce(BranchCreate,
  BranchTransition { source, target }, actor)
- branch_delete_as → enforce(BranchDelete, TargetBranch(name), actor)
- branch_merge_as → enforce(BranchMerge,
  BranchTransition { source, target }, actor)

Three new _as variants for branch ops (create, create_from, delete)
that had no actor surface before; existing actor-less variants delegate
with actor=None so the no-policy path is a strict no-op.

HTTP handlers updated to thread the resolved actor into the new _as
variants for branch_create and branch_delete (was previously dropped).

14 new SDK chassis tests (one allow + one deny pair per wired writer);
the existing 4 apply_schema_as tests stay. All 18 pass.

docs/user/policy.md updated to describe engine-wide enforcement and the
coarse-vs-fine layer split (engine = action gate, query layer per-row =
MR-725 future). AGENTS.md capability matrix updated to match.

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

crates/omnigraph-server/src/lib.rs

@@ -1298,9 +1298,13 @@ async fn server_branch_create(
         .map_err(ApiError::from_workload_reject)?;
     {
         let db = &state.engine;
-        db.branch_create_from(ReadTarget::branch(&from), &request.name)
-            .await
-            .map_err(ApiError::from_omni)?;
+        db.branch_create_from_as(
+            ReadTarget::branch(&from),
+            &request.name,
+            actor.as_ref().map(|Extension(a)| a.as_str()),
+        )
+        .await
+        .map_err(ApiError::from_omni)?;
     }
     Ok(Json(BranchCreateOutput {
         uri: state.uri().to_string(),
@@ -1359,7 +1363,7 @@ async fn server_branch_delete(
         .map_err(ApiError::from_workload_reject)?;
     {
         let db = &state.engine;
-        db.branch_delete(&branch)
+        db.branch_delete_as(&branch, actor_id)
             .await
             .map_err(ApiError::from_omni)?;
     }

crates/omnigraph/src/db/omnigraph.rs

@@ -991,6 +991,22 @@ impl Omnigraph {
     }
 
     pub async fn branch_create(&self, name: &str) -> Result<()> {
+        self.branch_create_as(name, None).await
+    }
+
+    /// Create a branch from the coordinator's currently-open snapshot,
+    /// with an explicit actor for engine-layer policy enforcement
+    /// (MR-722 fan-out). Scope is `TargetBranch(name)` — symmetric with
+    /// `branch_delete_as`: the branch being acted upon is the target.
+    /// Cedar rules using `target_branch_scope: protected` therefore see
+    /// the new-branch name and can deny e.g. creating any branch named
+    /// `main` from a non-privileged actor.
+    pub async fn branch_create_as(&self, name: &str, actor: Option<&str>) -> Result<()> {
+        self.enforce(
+            omnigraph_policy::PolicyAction::BranchCreate,
+            &omnigraph_policy::ResourceScope::TargetBranch(name.to_string()),
+            actor,
+        )?;
         self.ensure_schema_state_valid().await?;
         self.ensure_schema_apply_idle("branch_create").await?;
         ensure_public_branch_ref(name, "branch_create")?;
@@ -1002,8 +1018,41 @@ impl Omnigraph {
         from: impl Into<ReadTarget>,
         name: &str,
     ) -> Result<()> {
+        self.branch_create_from_as(from, name, None).await
+    }
+
+    /// Create a branch from a specific source branch with an explicit
+    /// actor for engine-layer policy enforcement (MR-722 fan-out).
+    ///
+    /// Scope is `BranchTransition { source, target }` — matches the
+    /// HTTP-layer convention at `server_branch_create`
+    /// (branch=Some(from), target_branch=Some(name)), so engine and
+    /// HTTP fire the same Cedar decision. Pinned-snapshot sources
+    /// (which aren't a branch ref) materialize as the sentinel
+    /// `<snapshot>` for the policy check; Cedar rules using
+    /// `branch_scope: any` still match, rules pinning a specific
+    /// source branch correctly do not.
+    pub async fn branch_create_from_as(
+        &self,
+        from: impl Into<ReadTarget>,
+        name: &str,
+        actor: Option<&str>,
+    ) -> Result<()> {
+        let target = from.into();
+        let source_branch = match &target {
+            ReadTarget::Branch(b) => b.clone(),
+            _ => "<snapshot>".to_string(),
+        };
+        self.enforce(
+            omnigraph_policy::PolicyAction::BranchCreate,
+            &omnigraph_policy::ResourceScope::BranchTransition {
+                source: source_branch,
+                target: name.to_string(),
+            },
+            actor,
+        )?;
         self.ensure_schema_apply_idle("branch_create_from").await?;
-        self.branch_create_from_impl(from, name, false).await
+        self.branch_create_from_impl(target, name, false).await
     }
 
     async fn branch_create_from_impl(
@@ -1049,6 +1098,22 @@ impl Omnigraph {
     }
 
     pub async fn branch_delete(&self, name: &str) -> Result<()> {
+        self.branch_delete_as(name, None).await
+    }
+
+    /// Delete a branch with an explicit actor for engine-layer policy
+    /// enforcement (MR-722 fan-out). Scope is `TargetBranch(name)` —
+    /// matches the HTTP-layer convention at `server_branch_delete`
+    /// (branch=None, target_branch=Some(name)). Cedar rules using
+    /// `target_branch_scope: protected` therefore correctly gate
+    /// deletion of protected branches (e.g. deny BranchDelete against
+    /// `main`).
+    pub async fn branch_delete_as(&self, name: &str, actor: Option<&str>) -> Result<()> {
+        self.enforce(
+            omnigraph_policy::PolicyAction::BranchDelete,
+            &omnigraph_policy::ResourceScope::TargetBranch(name.to_string()),
+            actor,
+        )?;
         self.ensure_schema_state_valid().await?;
         self.ensure_schema_apply_idle("branch_delete").await?;
         ensure_public_branch_ref(name, "branch_delete")?;

crates/omnigraph/src/exec/merge.rs

@@ -1062,6 +1062,21 @@ impl Omnigraph {
         target: &str,
         actor_id: Option<&str>,
     ) -> Result<MergeOutcome> {
+        // Engine-layer policy gate (MR-722 fan-out / PR #3). Scope is
+        // `BranchTransition { source, target }` — matches the HTTP-layer
+        // convention at `server_branch_merge` (branch=Some(source),
+        // target_branch=Some(target)). Cedar rules using
+        // `target_branch_scope: protected` therefore correctly gate
+        // merges INTO protected branches without forbidding the
+        // (symmetric) source-side reference.
+        self.enforce(
+            omnigraph_policy::PolicyAction::BranchMerge,
+            &omnigraph_policy::ResourceScope::BranchTransition {
+                source: source.to_string(),
+                target: target.to_string(),
+            },
+            actor_id,
+        )?;
         self.ensure_schema_apply_idle("branch_merge").await?;
         self.branch_merge_impl(source, target, actor_id).await
     }

crates/omnigraph/src/exec/mutation.rs

@@ -692,6 +692,16 @@ impl Omnigraph {
         params: &ParamMap,
         actor_id: Option<&str>,
     ) -> Result<MutationResult> {
+        // Engine-layer policy gate (MR-722 fan-out / PR #3). Scope is
+        // `Branch(branch)` to match the HTTP-layer convention at
+        // `server_change` (branch=Some(branch), target_branch=None). When no
+        // PolicyChecker is installed this is a no-op; with policy installed
+        // and actor=None this fails hard (forget-the-actor footgun guard).
+        self.enforce(
+            omnigraph_policy::PolicyAction::Change,
+            &omnigraph_policy::ResourceScope::Branch(branch.to_string()),
+            actor_id,
+        )?;
         self.mutate_with_current_actor(branch, query_source, query_name, params, actor_id)
             .await
     }

crates/omnigraph/src/loader/mod.rs

@@ -90,6 +90,18 @@ impl Omnigraph {
         mode: LoadMode,
         actor_id: Option<&str>,
     ) -> Result<IngestResult> {
+        // Engine-layer policy gate (MR-722 fan-out / PR #3). Scope is
+        // `Branch(branch)` for the data-write portion. If ingest creates
+        // a new branch as a side-effect (target branch doesn't exist),
+        // the inner `branch_create_from_as` call below additionally
+        // checks `BranchCreate` — both authorities are genuinely needed
+        // for "ingest into a fresh branch", so the layered check is
+        // correct, not redundant.
+        self.enforce(
+            omnigraph_policy::PolicyAction::Change,
+            &omnigraph_policy::ResourceScope::Branch(branch.to_string()),
+            actor_id,
+        )?;
         self.ingest_with_current_actor(branch, from, data, mode, actor_id)
             .await
     }
@@ -135,8 +147,18 @@ impl Omnigraph {
             .iter()
             .any(|name| name == &target_branch);
         if branch_created {
-            self.branch_create_from(crate::db::ReadTarget::branch(&base_branch), &target_branch)
-                .await?;
+            // Thread the actor through to the implicit BranchCreate so
+            // policy decisions match what an explicit `branch_create_from_as`
+            // call would see. Calling the no-actor variant here would
+            // bypass BranchCreate enforcement when policy is installed —
+            // the footgun guard catches that case too, but threading is
+            // the correct fix.
+            self.branch_create_from_as(
+                crate::db::ReadTarget::branch(&base_branch),
+                &target_branch,
+                actor_id,
+            )
+            .await?;
         }
 
         let result = self.load_as(&target_branch, data, mode, actor_id).await?;
@@ -160,6 +182,17 @@ impl Omnigraph {
         mode: LoadMode,
         actor_id: Option<&str>,
     ) -> Result<LoadResult> {
+        // Engine-layer policy gate (MR-722 fan-out / PR #3). Scope is
+        // `Branch(branch)` to match the HTTP-layer Change convention.
+        // `ingest_as` also calls `load_as` after enforcing its own
+        // Change gate — that double-check is fine because both gates
+        // resolve to identical Cedar decisions for the same actor +
+        // branch (the second check is a structurally-correct no-op).
+        self.enforce(
+            omnigraph_policy::PolicyAction::Change,
+            &omnigraph_policy::ResourceScope::Branch(branch.to_string()),
+            actor_id,
+        )?;
         self.ensure_schema_state_valid().await?;
         // Reject internal `__run__*` / system-prefixed branches at the
         // public write boundary. Direct-publish paths assert this

crates/omnigraph/tests/policy_engine_chassis.rs

@@ -1,39 +1,59 @@
-//! Engine-layer policy enforcement (MR-722 chassis core, PR #2).
+//! Engine-layer policy enforcement (MR-722 chassis core, PR #2 + PR #3).
 //!
-//! These tests exercise `Omnigraph::with_policy()` + `apply_schema_as()`
+//! These tests exercise `Omnigraph::with_policy()` + every `_as` writer
 //! 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`).
+//! PR #2 wired `apply_schema_as`. PR #3 fans the same `enforce()` call
+//! out to the remaining six writers — `mutate_as`, `load_as`,
+//! `ingest_as`, `branch_create_as` / `branch_create_from_as`,
+//! `branch_delete_as`, `branch_merge_as`. Each writer pair below
+//! covers allow + deny via the engine-side gate; the allow case proves
+//! the enforce call is correctly scoped (i.e. doesn't reject a legit
+//! actor), the deny case proves it actually denies an unauthorized
+//! actor — and both together pin the action × scope shape to match the
+//! HTTP-layer authorize_request convention so engine and HTTP fire the
+//! same Cedar decision.
 
 mod helpers;
 
 use std::fs;
+use std::path::Path;
 use std::sync::Arc;
 
-use omnigraph::db::{Omnigraph, SchemaApplyOptions};
+use omnigraph::db::{Omnigraph, ReadTarget, SchemaApplyOptions};
+use omnigraph::loader::LoadMode;
 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.
+/// Cedar policy: `act-allowed` may do every write; `act-denied` is in
+/// the known-actors set (so Cedar evaluates the policy and doesn't
+/// reject as unknown) but has no permit rule and is therefore implicitly
+/// denied for every action.
+///
+/// The rule split mirrors the per-action scope convention: Change uses
+/// `branch_scope`; SchemaApply, BranchCreate, BranchDelete, BranchMerge
+/// use `target_branch_scope` (see `PolicyAction::uses_branch_scope` and
+/// `uses_target_branch_scope` in `omnigraph-policy`).
 const POLICY_YAML: &str = r#"
 version: 1
 groups:
-  schema-writers: [act-allowed]
+  writers: [act-allowed]
   readers: [act-denied]
 protected_branches: [main]
 rules:
-  - id: writers-schema-apply
+  - id: writers-data
     allow:
-      actors: { group: schema-writers }
-      actions: [schema_apply]
+      actors: { group: writers }
+      actions: [change]
+      branch_scope: any
+  - id: writers-branches-schema
+    allow:
+      actors: { group: writers }
+      actions: [schema_apply, branch_create, branch_delete, branch_merge]
       target_branch_scope: any
 "#;
 
@@ -41,16 +61,54 @@ 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");
+fn install_policy(db: Omnigraph, dir_path: &Path) -> (Omnigraph, Arc<PolicyEngine>) {
+    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 = 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)
 }
 
+async fn init_with_policy(dir: &tempfile::TempDir) -> (Omnigraph, Arc<PolicyEngine>) {
+    let db = init_and_load(dir).await;
+    install_policy(db, dir.path())
+}
+
+/// Variant for tests that need a pre-created feature branch (branch_delete /
+/// branch_merge setup). Create the branch BEFORE wrapping with policy so the
+/// setup itself doesn't need to satisfy BranchCreate.
+async fn init_with_policy_and_feature_branch(
+    dir: &tempfile::TempDir,
+    branch: &str,
+) -> (Omnigraph, Arc<PolicyEngine>) {
+    let db = init_and_load(dir).await;
+    db.branch_create_from(ReadTarget::branch("main"), branch)
+        .await
+        .expect("setup: create feature branch before installing policy");
+    install_policy(db, dir.path())
+}
+
+// `MUTATION_QUERIES` from helpers/mod.rs already defines `insert_person($name, $age)`
+// — reuse it rather than redefining one here, so this test exercises the
+// same surface the engine integration tests do.
+
+/// One JSONL record for `load_as` / `ingest_as` exercises.
+const ONE_PERSON_JSONL: &str = r#"{"type": "Person", "data": {"name": "Eve"}}"#;
+
+fn assert_denied(result: Result<impl std::fmt::Debug, OmniError>, what: &str) {
+    match result {
+        Err(OmniError::Policy(msg)) => {
+            assert!(
+                msg.contains("denied"),
+                "{what}: expected denial message, got: {msg}"
+            );
+        }
+        Err(other) => panic!("{what}: expected OmniError::Policy, got: {other:?}"),
+        Ok(value) => panic!("{what}: expected denial, got Ok({value:?})"),
+    }
+}
+
 #[tokio::test]
 async fn apply_schema_as_denies_when_policy_rejects_actor() {
     let dir = tempfile::tempdir().unwrap();
@@ -127,3 +185,191 @@ async fn apply_schema_without_policy_still_works() {
         .await
         .expect("no policy → no enforcement → apply succeeds");
 }
+
+// ─── PR #3 writer fan-out ─────────────────────────────────────────────────
+//
+// One allow + one deny test per newly-wired writer. The allow case
+// proves the enforce scope is correctly shaped (i.e. doesn't reject a
+// legit actor whose policy permit matches the engine-side scope). The
+// deny case proves the gate actually fires for an unauthorized actor.
+// Footgun-guard (no-actor + policy-installed) is already proved by
+// `apply_schema_without_actor_when_policy_is_installed_denies` and
+// applies identically to every `_as` variant — duplicating it per
+// writer would be redundant.
+
+#[tokio::test]
+async fn mutate_as_denies_when_policy_rejects_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let params = mixed_params(&[("$name", "Eve")], &[("$age", 22)]);
+    let result = db
+        .mutate_as(
+            "main",
+            MUTATION_QUERIES,
+            "insert_person",
+            &params,
+            Some("act-denied"),
+        )
+        .await;
+    assert_denied(result, "mutate_as");
+}
+
+#[tokio::test]
+async fn mutate_as_allows_when_policy_permits_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let params = mixed_params(&[("$name", "Eve")], &[("$age", 22)]);
+    db.mutate_as(
+        "main",
+        MUTATION_QUERIES,
+        "insert_person",
+        &params,
+        Some("act-allowed"),
+    )
+    .await
+    .expect("act-allowed should be able to Change on main");
+}
+
+#[tokio::test]
+async fn load_as_denies_when_policy_rejects_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let result = db
+        .load_as("main", ONE_PERSON_JSONL, LoadMode::Merge, Some("act-denied"))
+        .await;
+    assert_denied(result, "load_as");
+}
+
+#[tokio::test]
+async fn load_as_allows_when_policy_permits_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    db.load_as(
+        "main",
+        ONE_PERSON_JSONL,
+        LoadMode::Merge,
+        Some("act-allowed"),
+    )
+    .await
+    .expect("act-allowed should be able to load on main");
+}
+
+#[tokio::test]
+async fn ingest_as_denies_when_policy_rejects_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let result = db
+        .ingest_as(
+            "main",
+            Some("main"),
+            ONE_PERSON_JSONL,
+            LoadMode::Merge,
+            Some("act-denied"),
+        )
+        .await;
+    assert_denied(result, "ingest_as");
+}
+
+#[tokio::test]
+async fn ingest_as_allows_when_policy_permits_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    db.ingest_as(
+        "main",
+        Some("main"),
+        ONE_PERSON_JSONL,
+        LoadMode::Merge,
+        Some("act-allowed"),
+    )
+    .await
+    .expect("act-allowed should be able to ingest on main");
+}
+
+#[tokio::test]
+async fn branch_create_as_denies_when_policy_rejects_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let result = db.branch_create_as("feature", Some("act-denied")).await;
+    assert_denied(result, "branch_create_as");
+}
+
+#[tokio::test]
+async fn branch_create_as_allows_when_policy_permits_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    db.branch_create_as("feature", Some("act-allowed"))
+        .await
+        .expect("act-allowed should be able to BranchCreate");
+}
+
+#[tokio::test]
+async fn branch_create_from_as_denies_when_policy_rejects_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    let result = db
+        .branch_create_from_as(ReadTarget::branch("main"), "feature", Some("act-denied"))
+        .await;
+    assert_denied(result, "branch_create_from_as");
+}
+
+#[tokio::test]
+async fn branch_create_from_as_allows_when_policy_permits_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy(&dir).await;
+
+    db.branch_create_from_as(ReadTarget::branch("main"), "feature", Some("act-allowed"))
+        .await
+        .expect("act-allowed should be able to BranchCreate from main");
+}
+
+#[tokio::test]
+async fn branch_delete_as_denies_when_policy_rejects_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy_and_feature_branch(&dir, "feature").await;
+
+    let result = db.branch_delete_as("feature", Some("act-denied")).await;
+    assert_denied(result, "branch_delete_as");
+}
+
+#[tokio::test]
+async fn branch_delete_as_allows_when_policy_permits_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy_and_feature_branch(&dir, "feature").await;
+
+    db.branch_delete_as("feature", Some("act-allowed"))
+        .await
+        .expect("act-allowed should be able to BranchDelete");
+}
+
+#[tokio::test]
+async fn branch_merge_as_denies_when_policy_rejects_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy_and_feature_branch(&dir, "feature").await;
+
+    let result = db
+        .branch_merge_as("feature", "main", Some("act-denied"))
+        .await;
+    assert_denied(result, "branch_merge_as");
+}
+
+#[tokio::test]
+async fn branch_merge_as_allows_when_policy_permits_actor() {
+    let dir = tempfile::tempdir().unwrap();
+    let (db, _engine) = init_with_policy_and_feature_branch(&dir, "feature").await;
+
+    // No diverging writes on feature → merge is a no-op fast-forward,
+    // but it still goes through enforce(BranchMerge, ...). That's the
+    // path under test; the actual merge outcome is incidental.
+    db.branch_merge_as("feature", "main", Some("act-allowed"))
+        .await
+        .expect("act-allowed should be able to BranchMerge");
+}