tests: policy chassis e2e gap-fills (MR-722 follow-up) (#106)

* tests: policy chassis e2e gap-fills (MR-722 follow-up)

Audit after PRs #101-105 surfaced real e2e gaps in the policy chassis
that could let regressions ride through silently. Coverage was strong
at the SDK level (18 chassis tests) and reasonable at HTTP (12+ policy
tests), but the CLI×writer matrix was asymmetric (only `change` tested
end-to-end), the `cli.actor` config-only precedence path was untested,
the `OMNIGRAPH_UNAUTHENTICATED` env-var read path was unexercised,
`serve()`'s startup-refusal propagation was structural-review only,
and engine↔HTTP decision parity was a structural property without a
test pinning it. This commit closes those gaps.

Added (15 new tests, all test-only):

* `policy_engine_chassis.rs` (+2): `load_file_as` allow + deny pair —
  PR #104 added the actor-aware mirror of `load_file` but it was only
  exercised via CLI integration; this is direct-SDK coverage.

* `omnigraph-server/src/lib.rs` mod tests (+2):
  - `unauthenticated_env_var_classification` — consolidated single
    test (process-global env var; running parallel would race) that
    pins truthy values, falsy values, unset, and CLI-flag-overrides-
    env behavior of the `OMNIGRAPH_UNAUTHENTICATED` read path inside
    `load_server_settings`.
  - `serve_refuses_to_start_in_state_1_without_unauthenticated` —
    `#[serial]` integration test. Clears all bearer-token env vars,
    builds a `ServerConfig` with no policy file and no flag, calls
    `serve(config).await`, asserts Err before any side-effecting
    work (Lance dataset open, TcpListener::bind). Guards the
    classifier→serve propagation path so a future refactor that
    drops the call turns red.

* `omnigraph-server/tests/server.rs` (+4): `policy_decision_parity_*`
  — four cases (Change×allowed+denied, BranchMerge×allowed+denied).
  Each case runs the same Cedar decision via both SDK
  (`Omnigraph::with_policy().mutate_as` / `branch_merge_as`) and HTTP
  (`POST /change` / `POST /branches/merge`) and asserts both either
  Allow or Deny. The structural property (both paths call
  `PolicyChecker::check`) is now test-asserted.

* `omnigraph-cli/tests/system_local.rs` (+8): the CLI×writer matrix
  fan-out:
  - `local_cli_load_enforces_engine_layer_policy`
  - `local_cli_ingest_enforces_engine_layer_policy`
  - `local_cli_schema_apply_enforces_engine_layer_policy`
  - `local_cli_branch_create_enforces_engine_layer_policy`
  - `local_cli_branch_delete_enforces_engine_layer_policy`
  - `local_cli_branch_merge_enforces_engine_layer_policy`
  Each: one denied case (`--as act-bruno` against protected main) +
  one allowed case (`--as act-ragnor` via existing/extended admins-*
  rules).
  Plus:
  - `local_cli_actor_from_config_used_when_no_flag` — proves the
    config-only precedence path works.
  - `local_cli_actor_flag_overrides_config_actor` — proves the
    `--as` flag wins over `cli.actor` in the config.
  Adds `local_policy_config_with_actor` helper. Extends
  `POLICY_E2E_YAML` with `admins-branch-ops` (BranchCreate +
  BranchDelete) and `admins-schema-apply` rules so the CLI×writer
  matrix has positive-case rule coverage.

Verification: all new tests pass; full `cargo test --workspace
--locked` is green; `scripts/check-agents-md.sh` passes.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* tests: serialize env-touching server lib tests to fix CI flake

CI flake on PR #106's Test Workspace job: two of the new tests
(`serve_refuses_to_start_in_state_1_without_unauthenticated` and
`unauthenticated_env_var_classification`) raced against
`server_bearer_tokens_from_env_reads_legacy_token_and_token_file`,
which sets `OMNIGRAPH_SERVER_BEARER_TOKEN` via `EnvGuard`.

While `serve_refuses` was mid-execution with its EnvGuard cleared,
the bearer-token test's EnvGuard had `OMNIGRAPH_SERVER_BEARER_TOKEN`
set; `resolve_token_source()` saw it and classified the runtime
state as `DefaultDeny` rather than refusing — so the test panicked
with "Dataset at path X not found" instead of the expected refusal
message. The unauthenticated test had the symmetric failure: its
`OMNIGRAPH_UNAUTHENTICATED="anything"` got overwritten by a peer
`EnvGuard` drop.

Fix: mark every test that uses `EnvGuard` with `#[serial]` so they
serialize against each other (default key). Already on
`serve_refuses_to_start_in_state_1_without_unauthenticated`; added
to `unauthenticated_env_var_classification` and
`server_bearer_tokens_from_env_reads_legacy_token_and_token_file`.
The `parse_bearer_tokens_json_*` tests don't touch env vars and
stay parallel.

Locally green (36 tests pass on my workstation); the parallelism
issue is CI-runner-specific (more aggressive thread interleaving)
but the fix is universal.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

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

crates/omnigraph-server/tests/server.rs

@@ -8,8 +8,10 @@ use axum::body::{Body, to_bytes};
 use axum::http::header::AUTHORIZATION;
 use axum::http::{Method, Request, StatusCode};
 use lance_index::traits::DatasetIndexExt;
-use omnigraph::db::{Omnigraph, ReadTarget};
+use omnigraph::db::{Omnigraph, ReadTarget, SchemaApplyOptions};
+use omnigraph::error::OmniError;
 use omnigraph::loader::{LoadMode, load_jsonl};
+use omnigraph_policy::{PolicyChecker, PolicyEngine};
 use omnigraph_server::api::{
     BranchCreateRequest, BranchMergeRequest, ChangeRequest, ErrorOutput, ExportRequest,
     IngestRequest, ReadRequest, SchemaApplyRequest, SchemaOutput,
@@ -3787,3 +3789,234 @@ async fn default_deny_mode_rejects_schema_apply_with_forbidden() {
         error.error
     );
 }
+
+// ─── SDK ↔ HTTP decision parity (MR-722 PR A) ─────────────────────────────
+//
+// Engine and HTTP both consult Cedar via `PolicyChecker::check()`; by
+// construction they cannot disagree on a decision. These tests pin that
+// property explicitly so a future refactor that introduces a separate
+// auth path (or copy-pastes Cedar evaluation logic) turns red.
+//
+// Four cases cover the per-action scope shapes:
+// * Change on a protected branch via `mutate_as` / POST /change
+// * Change with an actor that has no permit
+// * BranchMerge to a protected target via `branch_merge_as` / POST /branches/merge
+// * BranchMerge with an actor that has no permit
+
+const PARITY_POLICY_YAML: &str = r#"
+version: 1
+groups:
+  team: [act-bruno]
+  admins: [act-ragnor]
+protected_branches: [main]
+rules:
+  - id: admins-change-anywhere
+    allow:
+      actors: { group: admins }
+      actions: [change]
+      branch_scope: any
+  - id: admins-merge-to-protected
+    allow:
+      actors: { group: admins }
+      actions: [branch_merge]
+      target_branch_scope: protected
+"#;
+
+#[derive(Clone, Copy, Debug)]
+enum ParityDecision {
+    Allow,
+    Deny,
+}
+
+async fn build_parity_repo() -> (tempfile::TempDir, PathBuf, PathBuf) {
+    // Build a repo with `main` loaded and a `feature` branch ready for
+    // merge. Returns the repo path and a written policy.yaml path.
+    let temp = init_loaded_repo().await;
+    let repo = repo_path(temp.path());
+    {
+        let db = Omnigraph::open(repo.to_str().unwrap()).await.unwrap();
+        db.branch_create_from(ReadTarget::branch("main"), "feature")
+            .await
+            .unwrap();
+        db.load_as(
+            "feature",
+            r#"{"type":"Person","data":{"name":"ParityEve","age":29}}"#,
+            LoadMode::Append,
+            None,
+        )
+        .await
+        .unwrap();
+    }
+    let policy_path = temp.path().join("policy.yaml");
+    fs::write(&policy_path, PARITY_POLICY_YAML).unwrap();
+    (temp, repo, policy_path)
+}
+
+async fn sdk_change_decision(repo: &Path, policy_path: &Path, actor: &str) -> ParityDecision {
+    let policy = PolicyEngine::load(policy_path, repo.to_string_lossy().as_ref()).unwrap();
+    let db = Omnigraph::open(repo.to_str().unwrap())
+        .await
+        .unwrap()
+        .with_policy(Arc::new(policy) as Arc<dyn PolicyChecker>);
+    let mut params: omnigraph_compiler::ParamMap = Default::default();
+    // Parameter keys are bare names (no `$` prefix); the runtime resolves
+    // `$name` references in the query body to `params["name"]`.
+    params.insert(
+        "name".to_string(),
+        omnigraph_compiler::Literal::String("ParityCharlie".to_string()),
+    );
+    params.insert("age".to_string(), omnigraph_compiler::Literal::Integer(30));
+    let result = db
+        .mutate_as("main", MUTATION_QUERIES, "insert_person", &params, Some(actor))
+        .await;
+    match result {
+        Ok(_) => ParityDecision::Allow,
+        Err(OmniError::Policy(_)) => ParityDecision::Deny,
+        Err(other) => panic!("unexpected SDK error for change: {other:?}"),
+    }
+}
+
+async fn http_change_decision(
+    repo: &Path,
+    policy_path: &PathBuf,
+    actor: &str,
+    token: &str,
+) -> ParityDecision {
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        repo.to_string_lossy().to_string(),
+        vec![(actor.to_string(), token.to_string())],
+        Some(policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+    let req = ChangeRequest {
+        query_source: MUTATION_QUERIES.to_string(),
+        query_name: Some("insert_person".to_string()),
+        params: Some(json!({ "name": "ParityCharlie", "age": 30 })),
+        branch: Some("main".to_string()),
+    };
+    let (status, _body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/change")
+            .method(Method::POST)
+            .header(AUTHORIZATION, format!("Bearer {token}"))
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&req).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    match status {
+        StatusCode::OK => ParityDecision::Allow,
+        StatusCode::FORBIDDEN => ParityDecision::Deny,
+        other => panic!("unexpected HTTP status for change: {other}"),
+    }
+}
+
+async fn sdk_merge_decision(repo: &Path, policy_path: &Path, actor: &str) -> ParityDecision {
+    let policy = PolicyEngine::load(policy_path, repo.to_string_lossy().as_ref()).unwrap();
+    let db = Omnigraph::open(repo.to_str().unwrap())
+        .await
+        .unwrap()
+        .with_policy(Arc::new(policy) as Arc<dyn PolicyChecker>);
+    let result = db.branch_merge_as("feature", "main", Some(actor)).await;
+    match result {
+        Ok(_) => ParityDecision::Allow,
+        Err(OmniError::Policy(_)) => ParityDecision::Deny,
+        Err(other) => panic!("unexpected SDK error for branch_merge: {other:?}"),
+    }
+}
+
+async fn http_merge_decision(
+    repo: &Path,
+    policy_path: &PathBuf,
+    actor: &str,
+    token: &str,
+) -> ParityDecision {
+    let state = AppState::open_with_bearer_tokens_and_policy(
+        repo.to_string_lossy().to_string(),
+        vec![(actor.to_string(), token.to_string())],
+        Some(policy_path),
+    )
+    .await
+    .unwrap();
+    let app = build_app(state);
+    let req = BranchMergeRequest {
+        source: "feature".to_string(),
+        target: Some("main".to_string()),
+    };
+    let (status, _body) = json_response(
+        &app,
+        Request::builder()
+            .uri("/branches/merge")
+            .method(Method::POST)
+            .header(AUTHORIZATION, format!("Bearer {token}"))
+            .header("content-type", "application/json")
+            .body(Body::from(serde_json::to_vec(&req).unwrap()))
+            .unwrap(),
+    )
+    .await;
+    match status {
+        StatusCode::OK => ParityDecision::Allow,
+        StatusCode::FORBIDDEN => ParityDecision::Deny,
+        other => panic!("unexpected HTTP status for branch_merge: {other}"),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_decision_parity_change_admin_on_main_allowed() {
+    // (act-ragnor, change, main) — admins-change-anywhere rule applies.
+    // Both SDK and HTTP must allow. Each path uses its own fresh repo
+    // because allow→side-effects.
+    let (_t1, repo1, policy1) = build_parity_repo().await;
+    let sdk = sdk_change_decision(&repo1, &policy1, "act-ragnor").await;
+    let (_t2, repo2, policy2) = build_parity_repo().await;
+    let http = http_change_decision(&repo2, &policy2, "act-ragnor", "ragnor-token").await;
+    assert!(
+        matches!(sdk, ParityDecision::Allow) && matches!(http, ParityDecision::Allow),
+        "SDK={sdk:?} HTTP={http:?} — should both Allow",
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_decision_parity_change_team_on_main_denied() {
+    // (act-bruno, change, main) — no rule grants bruno change on
+    // protected. Both SDK and HTTP must deny. Same repo is reusable
+    // because deny→no side-effects.
+    let (_temp, repo, policy) = build_parity_repo().await;
+    let sdk = sdk_change_decision(&repo, &policy, "act-bruno").await;
+    let http = http_change_decision(&repo, &policy, "act-bruno", "bruno-token").await;
+    assert!(
+        matches!(sdk, ParityDecision::Deny) && matches!(http, ParityDecision::Deny),
+        "SDK={sdk:?} HTTP={http:?} — should both Deny",
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_decision_parity_branch_merge_admin_allowed() {
+    // (act-ragnor, branch_merge, feature→main) — admins-merge-to-protected
+    // rule applies. Both Allow. Each path uses its own fresh repo —
+    // a successful merge consumes the feature branch's commit on main.
+    let (_t1, repo1, policy1) = build_parity_repo().await;
+    let sdk = sdk_merge_decision(&repo1, &policy1, "act-ragnor").await;
+    let (_t2, repo2, policy2) = build_parity_repo().await;
+    let http = http_merge_decision(&repo2, &policy2, "act-ragnor", "ragnor-token").await;
+    assert!(
+        matches!(sdk, ParityDecision::Allow) && matches!(http, ParityDecision::Allow),
+        "SDK={sdk:?} HTTP={http:?} — should both Allow",
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn policy_decision_parity_branch_merge_team_denied() {
+    // (act-bruno, branch_merge, feature→main) — no rule grants bruno
+    // branch_merge. Both Deny.
+    let (_temp, repo, policy) = build_parity_repo().await;
+    let sdk = sdk_merge_decision(&repo, &policy, "act-bruno").await;
+    let http = http_merge_decision(&repo, &policy, "act-bruno", "bruno-token").await;
+    assert!(
+        matches!(sdk, ParityDecision::Deny) && matches!(http, ParityDecision::Deny),
+        "SDK={sdk:?} HTTP={http:?} — should both Deny",
+    );
+}