fix(cluster): validate policy bundle semantics

crates/omnigraph-cli/tests/cli_cluster.rs

@@ -25,6 +25,98 @@ fn cluster_validate_config_success() {
     assert!(stdout.contains("cluster config valid"), "{stdout}");
 }
 
+#[test]
+fn cluster_validate_rejects_semantically_invalid_policy() {
+    let temp = tempdir().unwrap();
+    write_cluster_config_fixture(temp.path());
+    fs::write(
+        temp.path().join("base.policy.yaml"),
+        r#"
+version: 1
+groups:
+  team: [act-andrew]
+rules:
+  - id: invalid-invoke-scope
+    allow:
+      actors: { group: team }
+      actions: [invoke_query]
+      branch_scope: any
+"#,
+    )
+    .unwrap();
+
+    let output = output_failure(
+        cli()
+            .arg("cluster")
+            .arg("validate")
+            .arg("--config")
+            .arg(temp.path()),
+    );
+    let stdout = stdout_string(&output);
+    assert!(
+        stdout.contains("ERROR policy_invalid policies.base.file"),
+        "{stdout}"
+    );
+    assert!(
+        stdout.contains("branch_scope") && stdout.contains("invoke_query"),
+        "{stdout}"
+    );
+}
+
+#[test]
+fn cluster_validate_rejects_policy_binding_kind_mismatch() {
+    for (applies_to, action, scope, expected_kind) in [
+        ("knowledge", "graph_list", "", "server-scoped"),
+        (
+            "cluster",
+            "read",
+            "      branch_scope: any\n",
+            "per-graph",
+        ),
+    ] {
+        let temp = tempdir().unwrap();
+        write_cluster_config_fixture(temp.path());
+        let config_path = temp.path().join("cluster.yaml");
+        let config = fs::read_to_string(&config_path)
+            .unwrap()
+            .replace("applies_to: [knowledge]", &format!("applies_to: [{applies_to}]"));
+        fs::write(config_path, config).unwrap();
+        fs::write(
+            temp.path().join("base.policy.yaml"),
+            format!(
+                r#"
+version: 1
+groups:
+  team: [act-andrew]
+rules:
+  - id: wrong-kind
+    allow:
+      actors: {{ group: team }}
+      actions: [{action}]
+{scope}"#
+            ),
+        )
+        .unwrap();
+
+        let output = output_failure(
+            cli()
+                .arg("cluster")
+                .arg("validate")
+                .arg("--config")
+                .arg(temp.path()),
+        );
+        let stdout = stdout_string(&output);
+        assert!(
+            stdout.contains("ERROR policy_invalid policies.base.file"),
+            "{stdout}"
+        );
+        assert!(
+            stdout.contains(expected_kind) && stdout.contains(action),
+            "{stdout}"
+        );
+    }
+}
+
 #[test]
 fn cluster_validate_json_is_stable() {
     let temp = tempdir().unwrap();
@@ -989,7 +1081,11 @@ fn applied_two_graph_cluster() -> tempfile::TempDir {
         "node Person {\n  name: String @key\n  age: I32?\n}\n",
     )
     .unwrap();
-    fs::write(root.join("base.policy.yaml"), "rules: []\n").unwrap();
+    fs::write(
+        root.join("base.policy.yaml"),
+        "version: 1\nrules: []\n",
+    )
+    .unwrap();
     fs::write(
         root.join("cluster.yaml"),
         r#"

crates/omnigraph-cli/tests/cli_data.rs

@@ -1029,49 +1029,6 @@ fn policy_validate_accepts_cluster_bundle() {
     assert!(stdout.contains("[2 actors]"));
 }
 
-#[test]
-fn policy_validate_fails_for_invalid_cluster_bundle() {
-    // The cluster does not validate a policy bundle's internal rules, so an
-    // applied-but-malformed bundle reaches `policy validate`, which compiles it
-    // and surfaces the error (here: a duplicate rule id).
-    let cluster = converged_loaded_cluster(
-        "knowledge",
-        Some(
-            r#"
-version: 1
-groups:
-  team: [act-andrew]
-rules:
-  - id: duplicate
-    allow:
-      actors: { group: team }
-      actions: [read]
-      branch_scope: any
-  - id: duplicate
-    allow:
-      actors: { group: team }
-      actions: [export]
-      branch_scope: any
-"#,
-        ),
-    );
-
-    let output = output_failure(
-        cli()
-            .arg("policy")
-            .arg("validate")
-            .arg("--cluster")
-            .arg(cluster.path())
-            .arg("--graph")
-            .arg("knowledge"),
-    );
-    let stderr = String::from_utf8(output.stderr).unwrap();
-    assert!(
-        stderr.contains("duplicate policy rule id"),
-        "expected a duplicate-rule error; got: {stderr}"
-    );
-}
-
 #[test]
 fn policy_test_runs_declarative_cases_against_cluster_bundle() {
     let cluster = converged_loaded_cluster("knowledge", Some(POLICY_YAML));

crates/omnigraph-cli/tests/support/mod.rs

@@ -712,8 +712,16 @@ query find_service($name: String) {
 "#,
     )
     .unwrap();
-    fs::write(root.join("cluster_wide.policy.yaml"), "rules: []\n").unwrap();
-    fs::write(root.join("shared.policy.yaml"), "rules: []\n").unwrap();
+    fs::write(
+        root.join("cluster_wide.policy.yaml"),
+        "version: 1\nrules: []\n",
+    )
+    .unwrap();
+    fs::write(
+        root.join("shared.policy.yaml"),
+        "version: 1\nrules: []\n",
+    )
+    .unwrap();
     fs::write(
         root.join("cluster.yaml"),
         r#"

crates/omnigraph-cluster/src/config.rs

@@ -949,12 +949,16 @@ pub(crate) fn load_desired(config_dir: &Path) -> LoadOutcome {
 
         let policy_address = policy_address(policy_name);
         let mut normalized_bindings: Vec<String> = Vec::new();
+        let mut binds_cluster = false;
+        let mut graph_binding = None;
         for (idx, target) in policy.applies_to.iter().enumerate() {
             match normalize_policy_target(target) {
                 PolicyTarget::Cluster => {
+                    binds_cluster = true;
                     normalized_bindings.push("cluster".to_string());
                 }
                 PolicyTarget::Graph(graph_id) => {
+                    graph_binding.get_or_insert_with(|| graph_id.clone());
                     normalized_bindings.push(graph_address(&graph_id));
                     if raw.graphs.contains_key(&graph_id) {
                         dependencies.insert(Dependency {
@@ -981,20 +985,45 @@ pub(crate) fn load_desired(config_dir: &Path) -> LoadOutcome {
 
         normalized_bindings.sort();
         normalized_bindings.dedup();
-        policy_bindings.insert(policy_address.clone(), normalized_bindings);
 
         let policy_path = resolve_config_path(&config_dir, &policy.file);
-        match fs::read(&policy_path) {
-            Ok(bytes) => {
+        match fs::read_to_string(&policy_path) {
+            Ok(source) => {
                 resources.insert(
                     policy_address.clone(),
                     ResourceSummary {
-                        address: policy_address,
+                        address: policy_address.clone(),
                         kind: "policy".to_string(),
-                        digest: sha256_hex(&bytes),
+                        digest: sha256_hex(source.as_bytes()),
                         path: Some(display_path(&policy_path)),
                     },
                 );
+                let validation = omnigraph_policy::PolicyConfig::from_source(&source)
+                    .and_then(|_| {
+                        if binds_cluster {
+                            omnigraph_policy::PolicyEngine::load_server_from_source(&source)
+                                .map(|_| ())
+                        } else {
+                            Ok(())
+                        }
+                    })
+                    .and_then(|_| {
+                        if let Some(graph_id) = graph_binding.as_deref() {
+                            omnigraph_policy::PolicyEngine::load_graph_from_source(
+                                &source, graph_id,
+                            )
+                            .map(|_| ())
+                        } else {
+                            Ok(())
+                        }
+                    });
+                if let Err(err) = validation {
+                    diagnostics.push(Diagnostic::error(
+                        "policy_invalid",
+                        format!("policies.{policy_name}.file"),
+                        format!("policy file '{}' is invalid: {err}", policy_path.display()),
+                    ));
+                }
             }
             Err(err) => diagnostics.push(Diagnostic::error(
                 "policy_file_missing",
@@ -1005,6 +1034,7 @@ pub(crate) fn load_desired(config_dir: &Path) -> LoadOutcome {
                 ),
             )),
         }
+        policy_bindings.insert(policy_address, normalized_bindings);
     }
 
     let mut resource_digests = BTreeMap::new();