policy: CLI policy injection — local writes go through engine enforce (MR-722) (#104)

Closes the CLI side of the policy chassis fan-out. Before this commit,
CLI direct-engine writes bypassed Cedar entirely because the CLI never
called `Omnigraph::with_policy(...)` for non-`policy validate|test|explain`
subcommands. After this commit, every CLI direct-engine writer
(change, load, ingest, branch create/delete/merge, schema apply) opens
the engine via a new `open_local_db_with_policy(uri, &config)` helper
that installs the configured `PolicyEngine` when `policy.file` is set,
and threads the resolved actor through to the `_as` writer methods.

Actor identity resolution:

- New top-level `--as <ACTOR>` global flag on the CLI overrides config.
- New `cli.actor` field in `omnigraph.yaml` provides a default actor.
- Precedence: `--as` > `cli.actor` > None.
- When policy is configured and neither is set, the engine-layer
  footgun guard fires and the write is denied — silent bypass via
  "I forgot the actor" is exactly what the guard prevents.
- Remote HTTP writes ignore both — bearer-token-resolved server-side.

Helpers added in main.rs:

- `open_local_db_with_policy(uri, &config) -> Result<Omnigraph>` —
  opens the DB and installs the PolicyEngine when configured. Without
  policy this is identical to a bare `Omnigraph::open`.
- `resolve_cli_actor(cli_as, &config) -> Option<&str>` — implements
  the flag > config > None precedence.

Engine: added `load_file_as` to the loader as the actor-aware mirror of
`load_file`, so CLI file-path loads flow through the same enforce gate
as in-memory `load_as` calls.

Test rewrite: `local_cli_policy_tooling_is_end_to_end_while_local_writes_stay_unenforced`
was the explicit assertion of the pre-chassis hole. Renamed and split:

- `local_cli_policy_tooling_is_end_to_end` — sanity for the read-only
  policy CLI surfaces (validate/test/explain), unchanged behavior.
- `local_cli_change_enforces_engine_layer_policy` — the new assertion:
  policy installed + no actor → footgun-guard denial; `--as act-bruno`
  on protected main → Cedar denial; `--as act-ragnor` (admins-write
  rule) on main → permit, write committed.

POLICY_E2E_YAML gains an `admins-write` rule so the permit case has
a non-trivial actor to exercise.

docs/user/policy.md updated with `cli.actor` + `--as <ACTOR>` usage.

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

crates/omnigraph-cli/src/main.rs

@@ -2,6 +2,7 @@ use std::fs;
 use std::io::{self, Write};
 use std::path::Path;
 use std::path::PathBuf;
+use std::sync::Arc;
 
 use clap::{Arg, ArgAction, Args, CommandFactory, FromArgMatches, Parser, Subcommand, ValueEnum};
 use color_eyre::eyre::{Result, bail};
@@ -44,6 +45,17 @@ const DEFAULT_BEARER_TOKEN_ENV: &str = "OMNIGRAPH_BEARER_TOKEN";
 #[command(about = "Omnigraph graph database CLI")]
 #[command(version = env!("CARGO_PKG_VERSION"), disable_version_flag = true)]
 struct Cli {
+    /// Actor identity for direct-engine writes (MR-722). Overrides
+    /// `cli.actor` from `omnigraph.yaml`. When the configured policy
+    /// is in effect, Cedar evaluates this actor against the requested
+    /// action and scope; with policy configured but neither this flag
+    /// nor `cli.actor` set, the engine-layer footgun guard fires and
+    /// the write is denied (no silent bypass). Has no effect on remote
+    /// HTTP writes — those resolve their actor server-side from the
+    /// bearer token.
+    #[arg(long = "as", global = true, value_name = "ACTOR")]
+    as_actor: Option<String>,
+
     #[command(subcommand)]
     command: Command,
 }
@@ -685,6 +697,37 @@ fn resolve_policy_engine(config: &OmnigraphConfig) -> Result<PolicyEngine> {
     PolicyEngine::load(&policy_file, &policy_repo_id(config))
 }
 
+/// Open a local-URI repo and, when `policy.file` is configured in
+/// `omnigraph.yaml`, install the resolved `PolicyEngine` on the engine
+/// handle so every direct-engine write goes through
+/// `Omnigraph::enforce(...)` (MR-722). Without a configured policy this
+/// is identical to a bare `Omnigraph::open`.
+///
+/// Returns owned `Omnigraph`; chained on top of `Omnigraph::open(...)`'s
+/// existing future to keep call sites narrow.
+async fn open_local_db_with_policy(uri: &str, config: &OmnigraphConfig) -> Result<Omnigraph> {
+    let db = Omnigraph::open(uri).await?;
+    if config.resolve_policy_file().is_some() {
+        let engine = Arc::new(resolve_policy_engine(config)?);
+        Ok(db.with_policy(engine as Arc<dyn omnigraph_policy::PolicyChecker>))
+    } else {
+        Ok(db)
+    }
+}
+
+/// Resolve the CLI's effective actor identity for engine-layer policy
+/// (MR-722). Precedence: `--as <ACTOR>` (top-level flag) overrides
+/// `cli.actor` from `omnigraph.yaml`; both unset returns `None`. When
+/// policy is configured and this returns `None`, the engine-layer
+/// footgun guard intentionally denies — silent bypass via "I forgot the
+/// actor" is what the guard prevents.
+fn resolve_cli_actor<'a>(
+    cli_as: Option<&'a str>,
+    config: &'a OmnigraphConfig,
+) -> Option<&'a str> {
+    cli_as.or(config.cli.actor.as_deref())
+}
+
 fn resolve_policy_tests_path(config: &OmnigraphConfig) -> Result<PathBuf> {
     config.resolve_policy_tests_file().ok_or_else(|| {
         color_eyre::eyre::eyre!(
@@ -1553,19 +1596,22 @@ async fn execute_change(
     query_name: Option<&str>,
     branch: &str,
     params_json: Option<&Value>,
+    config: &OmnigraphConfig,
+    cli_as_actor: Option<&str>,
 ) -> Result<ChangeOutput> {
     let (selected_name, query_params) = select_named_query(query_source, query_name)?;
     let params = query_params_from_json(&query_params, params_json)?;
-    let mut db = Omnigraph::open(uri).await?;
+    let db = open_local_db_with_policy(uri, config).await?;
+    let actor = resolve_cli_actor(cli_as_actor, config);
     let result = db
-        .mutate(branch, query_source, &selected_name, &params)
+        .mutate_as(branch, query_source, &selected_name, &params, actor)
         .await?;
     Ok(ChangeOutput {
         branch: branch.to_string(),
         query_name: selected_name,
         affected_nodes: result.affected_nodes,
         affected_edges: result.affected_edges,
-        actor_id: None,
+        actor_id: actor.map(String::from),
     })
 }
 
@@ -1689,9 +1735,10 @@ async fn main() -> Result<()> {
             let config = load_cli_config(config.as_ref())?;
             let uri = resolve_local_uri(&config, uri, target.as_deref(), "load")?;
             let branch = resolve_branch(&config, branch, None, "main");
-            let mut db = Omnigraph::open(&uri).await?;
+            let db = open_local_db_with_policy(&uri, &config).await?;
+            let actor = resolve_cli_actor(cli.as_actor.as_deref(), &config);
             let result = db
-                .load_file(&branch, &data.to_string_lossy(), mode.into())
+                .load_file_as(&branch, &data.to_string_lossy(), mode.into(), actor)
                 .await?;
             let payload = LoadOutput {
                 uri: &uri,
@@ -1748,9 +1795,16 @@ async fn main() -> Result<()> {
                 )
                 .await?
             } else {
-                let mut db = Omnigraph::open(&uri).await?;
+                let db = open_local_db_with_policy(&uri, &config).await?;
+                let actor = resolve_cli_actor(cli.as_actor.as_deref(), &config);
                 let result = db
-                    .ingest_file(&branch, Some(&from), &data.to_string_lossy(), mode.into())
+                    .ingest_file_as(
+                        &branch,
+                        Some(&from),
+                        &data.to_string_lossy(),
+                        mode.into(),
+                        actor,
+                    )
                     .await?;
                 ingest_output(&uri, &result, None)
             };
@@ -1787,14 +1841,15 @@ async fn main() -> Result<()> {
                     )
                     .await?
                 } else {
-                    let mut db = Omnigraph::open(&uri).await?;
-                    db.branch_create_from(ReadTarget::branch(&from), &name)
+                    let db = open_local_db_with_policy(&uri, &config).await?;
+                    let actor = resolve_cli_actor(cli.as_actor.as_deref(), &config);
+                    db.branch_create_from_as(ReadTarget::branch(&from), &name, actor)
                         .await?;
                     BranchCreateOutput {
                         uri: uri.clone(),
                         from: from.clone(),
                         name: name.clone(),
-                        actor_id: None,
+                        actor_id: actor.map(String::from),
                     }
                 };
                 if json {
@@ -1857,12 +1912,13 @@ async fn main() -> Result<()> {
                     )
                     .await?
                 } else {
-                    let mut db = Omnigraph::open(&uri).await?;
-                    db.branch_delete(&name).await?;
+                    let db = open_local_db_with_policy(&uri, &config).await?;
+                    let actor = resolve_cli_actor(cli.as_actor.as_deref(), &config);
+                    db.branch_delete_as(&name, actor).await?;
                     BranchDeleteOutput {
                         uri: uri.clone(),
                         name: name.clone(),
-                        actor_id: None,
+                        actor_id: actor.map(String::from),
                     }
                 };
                 if json {
@@ -1897,13 +1953,14 @@ async fn main() -> Result<()> {
                     )
                     .await?
                 } else {
-                    let mut db = Omnigraph::open(&uri).await?;
-                    let outcome = db.branch_merge(&source, &into).await?;
+                    let db = open_local_db_with_policy(&uri, &config).await?;
+                    let actor = resolve_cli_actor(cli.as_actor.as_deref(), &config);
+                    let outcome = db.branch_merge_as(&source, &into, actor).await?;
                     BranchMergeOutput {
                         source: source.clone(),
                         target: into.clone(),
                         outcome: outcome.into(),
-                        actor_id: None,
+                        actor_id: actor.map(String::from),
                     }
                 };
                 if json {
@@ -2051,11 +2108,13 @@ async fn main() -> Result<()> {
                     )
                     .await?
                 } else {
-                    let mut db = Omnigraph::open(&uri).await?;
+                    let db = open_local_db_with_policy(&uri, &config).await?;
+                    let actor = resolve_cli_actor(cli.as_actor.as_deref(), &config);
                     let result = db
-                        .apply_schema_with_options(
+                        .apply_schema_as(
                             &schema_source,
                             omnigraph::db::SchemaApplyOptions { allow_data_loss },
+                            actor,
                         )
                         .await?;
                     schema_apply_output(&uri, result)
@@ -2340,6 +2399,8 @@ async fn main() -> Result<()> {
                     query_name.as_deref(),
                     &branch,
                     params_json.as_ref(),
+                    &config,
+                    cli.as_actor.as_deref(),
                 )
                 .await?
             };
@@ -2397,7 +2458,7 @@ async fn main() -> Result<()> {
         } => {
             let config = load_cli_config(config.as_ref())?;
             let uri = resolve_uri(&config, uri, target.as_deref())?;
-            let mut db = Omnigraph::open(&uri).await?;
+            let db = Omnigraph::open(&uri).await?;
             let stats = db.optimize().await?;
             if json {
                 let value = serde_json::json!({