apunkt/omnigraph
1
0
Fork 0
mirror of https://github.com/ModernRelay/omnigraph.git synced 2026-06-18 02:24:27 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(cli)!: excise omnigraph.yaml from the CLI; policy/queries tooling reads --cluster (#251)

Browse source 
The server already dropped omnigraph.yaml (cluster-only boot). This removes the
CLI's last use of the legacy `OmnigraphConfig`: graphs are addressed only via
`--store`/`--server`/`--cluster`/`--profile`/operator defaults, and actor,
output format, and bearer credentials come from `~/.omnigraph/config.yaml`.
After this change no CLI command reads `omnigraph.yaml` except `config migrate`.

Resolvers (helpers.rs): drop every legacy fallback —
- `resolve_actor` → `--as` > `operator.actor` (no `cli.actor`);
- `resolve_read_format` → `--json`/`--format` > alias > `defaults.output`;
- `resolve_branch`/`resolve_read_target` → `--branch` > alias > "main";
- `resolve_uri`/`resolve_cli_graph` → scope path only; an absent address is a
  loud error;
- `resolve_remote_bearer_token` → operator keyed chain + `OMNIGRAPH_BEARER_TOKEN`.
`GraphClient::resolve`/`resolve_with_policy` drop the `&OmnigraphConfig` param;
direct-store access carries no Cedar policy (policy lives in the cluster/server).

Flags (cli.rs): remove `--config` from every data/query command; it stays only
on `cluster *` (the cluster dir) and `config migrate` (the legacy path).

Re-home control-plane tooling to `--cluster` (RFC-011):
- `policy validate|test|explain` source the Cedar bundle from the cluster's
  applied policies; `--graph` picks a graph's bundle; `policy test` takes
  `--tests <file>`;
- `queries list|validate` source the registry + schemas from the cluster
  serving snapshot; `--graph` scopes to one graph;
- `lint` requires `--schema` (offline) or a direct/cluster graph target;
- `schema plan`/`lint` route their graph-target through the shared direct-scope
  resolver so `--store`/`--profile`/`defaults.store` addressing works.

Tests migrate from `omnigraph.yaml` fixtures to `--store`/operator-config/
`--cluster` (converged-cluster fixtures); the now-impossible command-path
RFC-008 tests are deleted (`config migrate` coverage kept). The
`OmnigraphConfig` type, `load_config`/deprecation machinery, and `config
migrate` are removed in a follow-up.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Andrew Altshuler 2026-06-15 21:48:39 +03:00 • committed by GitHub
parent 8b01c6e547
commit 0bee746a31
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 1464 additions and 2262 deletions
Download patch file Download diff file Expand all files Collapse all files

93
crates/omnigraph-cli/src/cli.rs
View file

@ -18,10 +18,10 @@ any — run against a graph, served (--server / --profile) or embedded (--store
URI): query, mutate, load, branch, snapshot, export, commit, schema show/apply.\n \
served — require a server: graphs.\n \
direct — direct storage access; reject --server (init, optimize, repair, cleanup, \
schema plan, lint, queries validate).\n \
control — manage a cluster via --config: cluster.\n \
local — no graph; local config & tooling: policy, embed, login, logout, config, \
version, queries list.\n\
schema plan, lint).\n \
control — manage or inspect a cluster (cluster via --config; policy & queries via \
--cluster).\n \
local — no graph; local config & tooling: embed, login, logout, config, version.\n\
See the 'Command capabilities' section of the CLI reference for which flags apply where.")]
pub(crate) struct Cli {
/// Actor id for direct-engine writes; overrides `cli.actor`. No effect on
@ -96,8 +96,6 @@ pub(crate) enum Command {
/// the catalog (served — addressed via --server/--profile). With
/// `--query`/`-e`, selects which query in that ad-hoc source to run.
name: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
/// Ad-hoc query file (a `.gq` you're authoring / break-glass).
#[arg(long, conflicts_with = "query_string")]
query: Option<PathBuf>,
@ -126,8 +124,6 @@ pub(crate) enum Command {
/// from the catalog (served — addressed via --server/--profile). With
/// `--query`/`-e`, selects which query in that ad-hoc source to run.
name: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
/// Ad-hoc mutation file (a `.gq` you're authoring / break-glass).
#[arg(long, conflicts_with = "query_string")]
query: Option<PathBuf>,
@ -154,8 +150,6 @@ pub(crate) enum Command {
name: String,
/// Positional args bound to the alias's declared `args` params, in order.
args: Vec<String>,
#[arg(long)]
config: Option<PathBuf>,
#[command(flatten)]
params: ParamsArgs,
#[arg(long, conflicts_with = "json")]
@ -168,8 +162,6 @@ pub(crate) enum Command {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
data: PathBuf,
/// Target branch (defaults to main). Without --from it must exist.
#[arg(long)]
@ -191,8 +183,6 @@ pub(crate) enum Command {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
data: PathBuf,
#[arg(long)]
branch: Option<String>,
@ -213,8 +203,6 @@ pub(crate) enum Command {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
branch: Option<String>,
#[arg(long)]
json: bool,
@ -224,8 +212,6 @@ pub(crate) enum Command {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
branch: Option<String>,
#[arg(long, hide = true)]
jsonl: bool,
@ -270,16 +256,12 @@ pub(crate) enum Command {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
json: bool,
},
/// Classify and explicitly repair manifest/head drift
Repair {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
/// Publish verified maintenance drift. Without this flag, repair only
/// previews what it would do.
#[arg(long)]
@ -295,8 +277,6 @@ pub(crate) enum Command {
Cleanup {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
/// Number of recent versions to keep per table. Either `--keep` or
/// `--older-than` (or both) must be set.
#[arg(long)]
@ -326,8 +306,6 @@ pub(crate) enum Command {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
query: PathBuf,
#[arg(long)]
schema: Option<PathBuf>,
@ -480,8 +458,6 @@ pub(crate) enum GraphsCommand {
#[arg(long)]
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
json: bool,
},
}
@ -494,8 +470,6 @@ pub(crate) enum BranchCommand {
#[arg(long)]
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
from: Option<String>,
name: String,
#[arg(long)]
@ -507,8 +481,6 @@ pub(crate) enum BranchCommand {
#[arg(long)]
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
json: bool,
},
/// Delete a branch
@ -516,8 +488,6 @@ pub(crate) enum BranchCommand {
/// Graph URI
#[arg(long)]
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
name: String,
#[arg(long)]
json: bool,
@ -527,8 +497,6 @@ pub(crate) enum BranchCommand {
/// Graph URI
#[arg(long)]
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
source: String,
#[arg(long)]
into: Option<String>,
@ -544,8 +512,6 @@ pub(crate) enum SchemaCommand {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
schema: PathBuf,
#[arg(long)]
json: bool,
@ -560,8 +526,6 @@ pub(crate) enum SchemaCommand {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
schema: PathBuf,
#[arg(long)]
json: bool,
@ -583,8 +547,6 @@ pub(crate) enum SchemaCommand {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
json: bool,
},
}
@ -597,8 +559,6 @@ pub(crate) enum CommitCommand {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
branch: Option<String>,
#[arg(long)]
json: bool,
@ -608,8 +568,6 @@ pub(crate) enum CommitCommand {
/// Graph URI
#[arg(long)]
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
commit_id: String,
#[arg(long)]
json: bool,
@ -618,20 +576,24 @@ pub(crate) enum CommitCommand {
#[derive(Debug, Subcommand)]
pub(crate) enum PolicyCommand {
/// Validate policy YAML and compiled Cedar policy state
Validate {
#[arg(long)]
config: Option<PathBuf>,
},
/// Run declarative policy tests from policy.tests.yaml
/// Compile and validate the Cedar policy bundle(s) applied in a cluster.
///
/// Sources the bundle(s) from the cluster's applied policies
/// (`--cluster <dir>`); pass the global `--graph <id>` to pick one
/// graph's bundle when several apply.
Validate {},
/// Run declarative policy tests against a cluster's applied bundle.
///
/// The cluster model has no per-bundle tests file, so the cases are
/// supplied explicitly with `--tests <file>` and checked against the
/// bundle selected by `--cluster` (+ optional `--graph`).
Test {
/// Path to a policy.tests.yaml file.
#[arg(long)]
config: Option<PathBuf>,
tests: PathBuf,
},
/// Explain one policy decision locally
/// Explain one policy decision against a cluster's applied bundle.
Explain {
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
actor: String,
#[arg(long)]
@ -645,24 +607,19 @@ pub(crate) enum PolicyCommand {
#[derive(Debug, Subcommand)]
pub(crate) enum QueriesCommand {
/// Type-check the stored-query registry against the live schema.
/// Type-check a cluster's stored-query registry against its schemas.
///
/// Distinct from `omnigraph lint` (which lints one `.gq` file):
/// this validates the whole `queries:` registry — opening the graph
/// to read its schema and confirming every stored query still
/// type-checks. Exits non-zero on any breakage.
/// Distinct from `omnigraph lint` (which lints one `.gq` file): this
/// validates the whole `queries:` registry of a cluster (`--cluster
/// <dir>`, optional `--graph <id>`) by reading each graph's applied
/// schema and confirming every stored query still type-checks. Exits
/// non-zero on any breakage.
Validate {
/// Graph URI
uri: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
json: bool,
},
/// List the registered stored queries (name, MCP exposure, params).
/// List a cluster's registered stored queries (name, params).
List {
#[arg(long)]
config: Option<PathBuf>,
#[arg(long)]
json: bool,
},

109
crates/omnigraph-cli/src/client.rs
View file

@ -40,22 +40,20 @@ use serde_json::Value;
use crate::cli::CliLoadMode;
use crate::helpers::{
ResolvedCliGraph, apply_bearer_token, apply_server_flag, build_http_client, is_remote_uri,
legacy_change_request_body, open_local_db_with_policy, query_params_from_json,
apply_bearer_token, apply_server_flag, build_http_client, is_remote_uri,
legacy_change_request_body, query_params_from_json,
remote_json, remote_url, resolve_cli_actor, resolve_cli_graph, resolve_remote_bearer_token,
resolve_server_flag, select_named_query,
};
use crate::output::{LoadOutput, load_output_from_result, load_output_from_tables};
use omnigraph_server::config::OmnigraphConfig;
pub(crate) enum GraphClient {
/// Local engine at `uri`. Reads (`resolve()`) leave `graph`/`actor`
/// empty and open without policy; writes (`resolve_with_policy()`)
/// fill them, opening through `open_local_db_with_policy` and
/// attributing the resolved actor.
/// Local engine at `uri`. Reads (`resolve()`) leave `actor` empty;
/// writes (`resolve_with_policy()`) attribute the resolved actor.
/// Direct-store access carries no Cedar policy (RFC-011: policy lives
/// in the cluster/server, not in per-operator addressing).
Embedded {
uri: String,
graph: Option<ResolvedCliGraph>,
actor: Option<String>,
},
/// Remote HTTP server. The actor is resolved server-side from the
@ -75,7 +73,6 @@ pub(crate) enum GraphClient {
/// is then correct, or the real request surfaces the failure. Only fires on the
/// no-graph path, so a `--graph`/`default_graph` happy path does no extra I/O.
async fn require_graph_for_multi_graph_server(
config: &OmnigraphConfig,
scope: &crate::scope::ResolvedScope,
) -> Result<()> {
let (Some(server), None) = (scope.server.as_deref(), scope.graph.as_deref()) else {
@ -84,7 +81,7 @@ async fn require_graph_for_multi_graph_server(
let Some(base) = resolve_server_flag(Some(server), None)? else {
return Ok(());
};
let token = resolve_remote_bearer_token(config, Some(&base))?;
let token = resolve_remote_bearer_token(Some(&base))?;
let probe = GraphClient::Remote {
http: build_http_client()?,
base_url: base,
@ -126,7 +123,6 @@ impl GraphClient {
/// path, not the policy-bearing `resolve_cli_graph`). Used by reads
/// and `query` (which opens without policy, like the reads).
pub(crate) async fn resolve(
config: &OmnigraphConfig,
server: Option<&str>,
graph: Option<&str>,
uri: Option<String>,
@ -141,7 +137,7 @@ impl GraphClient {
crate::planes::Capability::Any,
crate::scope::ScopeFlags { profile, store, server, cluster: None, graph, uri },
)?;
require_graph_for_multi_graph_server(config, &scope).await?;
require_graph_for_multi_graph_server(&scope).await?;
let (server, graph, uri) = (
scope.server.as_deref(),
scope.graph.as_deref(),
@ -149,8 +145,8 @@ impl GraphClient {
);
let via_server = server.is_some();
let uri = apply_server_flag(server, graph, uri)?;
let token = resolve_remote_bearer_token(config, uri.as_deref())?;
let uri = crate::helpers::resolve_uri(config, uri)?;
let token = resolve_remote_bearer_token(uri.as_deref())?;
let uri = crate::helpers::resolve_uri(uri)?;
reject_positional_remote(via_server, &uri)?;
if is_remote_uri(&uri) {
Ok(GraphClient::Remote {
@ -159,11 +155,7 @@ impl GraphClient {
token,
})
} else {
Ok(GraphClient::Embedded {
uri,
graph: None,
actor: None,
})
Ok(GraphClient::Embedded { uri, actor: None })
}
}
@ -174,7 +166,6 @@ impl GraphClient {
/// resolution order matches the write arms exactly: server flag →
/// bearer token → graph.
pub(crate) async fn resolve_with_policy(
config: &OmnigraphConfig,
server: Option<&str>,
graph: Option<&str>,
uri: Option<String>,
@ -189,7 +180,7 @@ impl GraphClient {
crate::planes::Capability::Any,
crate::scope::ScopeFlags { profile, store, server, cluster: None, graph, uri },
)?;
require_graph_for_multi_graph_server(config, &scope).await?;
require_graph_for_multi_graph_server(&scope).await?;
let (server, graph, uri) = (
scope.server.as_deref(),
scope.graph.as_deref(),
@ -197,8 +188,8 @@ impl GraphClient {
);
let via_server = server.is_some();
let uri = apply_server_flag(server, graph, uri)?;
let token = resolve_remote_bearer_token(config, uri.as_deref())?;
let resolved = resolve_cli_graph(config, uri)?;
let token = resolve_remote_bearer_token(uri.as_deref())?;
let resolved = resolve_cli_graph(uri)?;
reject_positional_remote(via_server, &resolved.uri)?;
if resolved.is_remote {
// A served write resolves the actor server-side from the bearer
@ -216,10 +207,9 @@ impl GraphClient {
token,
})
} else {
let actor = resolve_cli_actor(cli_as, config)?;
let actor = resolve_cli_actor(cli_as)?;
Ok(GraphClient::Embedded {
uri: resolved.uri.clone(),
graph: Some(resolved),
uri: resolved.uri,
actor,
})
}
@ -233,28 +223,15 @@ impl GraphClient {
}
}
/// The selected graph name, when a policy-bearing embedded client was
/// resolved against a named graph. `None` for remote and for reads.
pub(crate) fn selected(&self) -> Option<&str> {
match self {
GraphClient::Embedded { graph, .. } => graph.as_ref().and_then(ResolvedCliGraph::selected),
GraphClient::Remote { .. } => None,
}
}
pub(crate) fn is_remote(&self) -> bool {
matches!(self, GraphClient::Remote { .. })
}
/// Open the local engine the way the resolved client demands: with
/// policy when a `graph` context is present (write path), bare
/// otherwise (read/`query` path). Captures today's two open paths in
/// one place so each verb stays a single match arm.
async fn open_embedded(uri: &str, graph: &Option<ResolvedCliGraph>) -> Result<Omnigraph> {
match graph {
Some(graph) => open_local_db_with_policy(graph).await,
None => Ok(Omnigraph::open(uri).await?),
}
/// Open the local engine. Direct-store access carries no Cedar policy
/// (RFC-011), so both read and write paths open bare; the actor is still
/// attributed on the write via the `_as` engine APIs.
async fn open_embedded(uri: &str) -> Result<Omnigraph> {
Ok(Omnigraph::open(uri).await?)
}
pub(crate) async fn branch_list(&self) -> Result<BranchListOutput> {
@ -416,8 +393,8 @@ impl GraphClient {
.await?;
Ok(load_output_from_tables(base_url, branch, mode.as_str(), &output))
}
GraphClient::Embedded { uri, graph, actor } => {
let db = Self::open_embedded(uri, graph).await?;
GraphClient::Embedded { uri, actor } => {
let db = Self::open_embedded(uri).await?;
let result = db
.load_file_as(branch, from, data, mode.into(), actor.as_deref())
.await?;
@ -459,8 +436,8 @@ impl GraphClient {
)
.await
}
GraphClient::Embedded { uri, graph, actor } => {
let db = Self::open_embedded(uri, graph).await?;
GraphClient::Embedded { uri, actor } => {
let db = Self::open_embedded(uri).await?;
let result = db
.load_file_as(branch, Some(from), data, mode.into(), actor.as_deref())
.await?;
@ -498,10 +475,10 @@ impl GraphClient {
)
.await
}
GraphClient::Embedded { uri, graph, actor } => {
GraphClient::Embedded { uri, actor } => {
let (selected_name, query_params) = select_named_query(query_source, query_name)?;
let params = query_params_from_json(&query_params, params_json)?;
let db = Self::open_embedded(uri, graph).await?;
let db = Self::open_embedded(uri).await?;
let actor = actor.as_deref();
let result = db
.mutate_as(branch, query_source, &selected_name, &params, actor)
@ -552,10 +529,10 @@ impl GraphClient {
)
.await
}
GraphClient::Embedded { uri, graph, .. } => {
GraphClient::Embedded { uri, .. } => {
let (selected_name, query_params) = select_named_query(query_source, query_name)?;
let params = query_params_from_json(&query_params, params_json)?;
let db = Self::open_embedded(uri, graph).await?;
let db = Self::open_embedded(uri).await?;
let result = db
.query(target.clone(), query_source, &selected_name, &params)
.await?;
@ -631,8 +608,8 @@ impl GraphClient {
)
.await
}
GraphClient::Embedded { uri, graph, actor } => {
let db = Self::open_embedded(uri, graph).await?;
GraphClient::Embedded { uri, actor } => {
let db = Self::open_embedded(uri).await?;
let actor = actor.as_deref();
db.branch_create_from_as(ReadTarget::branch(from), name, actor)
.await?;
@ -662,8 +639,8 @@ impl GraphClient {
)
.await
}
GraphClient::Embedded { uri, graph, actor } => {
let db = Self::open_embedded(uri, graph).await?;
GraphClient::Embedded { uri, actor } => {
let db = Self::open_embedded(uri).await?;
let actor = actor.as_deref();
db.branch_delete_as(name, actor).await?;
Ok(BranchDeleteOutput {
@ -694,8 +671,8 @@ impl GraphClient {
)
.await
}
GraphClient::Embedded { uri, graph, actor } => {
let db = Self::open_embedded(uri, graph).await?;
GraphClient::Embedded { uri, actor } => {
let db = Self::open_embedded(uri).await?;
let actor = actor.as_deref();
let outcome = db.branch_merge_as(source, into, actor).await?;
Ok(BranchMergeOutput {
@ -745,8 +722,8 @@ impl GraphClient {
)
.await
}
GraphClient::Embedded { uri, graph, actor } => {
let db = Self::open_embedded(uri, graph).await?;
GraphClient::Embedded { uri, actor } => {
let db = Self::open_embedded(uri).await?;
let result = db
.apply_schema_as_with_catalog_check(
schema_source,
@ -815,9 +792,9 @@ impl GraphClient {
/// `graphs list` — enumerate the graphs a remote multi-graph server
/// serves (`GET /graphs`). Remote-only by design: there is no local
/// enumeration endpoint, so the Embedded arm fails loudly pointing the
/// operator at `omnigraph.yaml`. Routing it through the enum still buys
/// the shared `resolve()` addressing/token preamble.
/// enumeration endpoint, so the Embedded arm fails loudly. Routing it
/// through the enum still buys the shared `resolve()` addressing/token
/// preamble.
pub(crate) async fn list_graphs(&self) -> Result<GraphListResponse> {
match self {
GraphClient::Remote {
@ -835,9 +812,9 @@ impl GraphClient {
.await
}
GraphClient::Embedded { .. } => bail!(
"`omnigraph graphs list` requires a remote multi-graph server URL \
(http:// or https://). To enumerate local graphs, read `omnigraph.yaml` \
directly."
"`omnigraph graphs list` requires a remote multi-graph server \
(--server <url>). To enumerate the graphs in a cluster, run \
`omnigraph cluster status --config <dir>`."
),
}
}

580
crates/omnigraph-cli/src/helpers.rs
View file

@ -119,231 +119,164 @@ pub(crate) fn bearer_token_from_env(var_name: &str) -> Option<String> {
normalize_bearer_token(std::env::var(var_name).ok())
}
pub(crate) fn parse_env_assignment(line: &str) -> Option<(String, String)> {
let line = line.trim();
if line.is_empty() || line.starts_with('#') {
return None;
}
let line = line.strip_prefix("export ").unwrap_or(line).trim();
let (name, value) = line.split_once('=')?;
let name = name.trim();
if name.is_empty() {
return None;
}
let value = value.trim();
let value = if value.len() >= 2
&& ((value.starts_with('"') && value.ends_with('"'))
|| (value.starts_with('\'') && value.ends_with('\'')))
{
&value[1..value.len() - 1]
} else {
value
};
Some((name.to_string(), value.to_string()))
}
pub(crate) fn bearer_token_from_env_file(path: &Path, var_name: &str) -> Result<Option<String>> {
if !path.exists() {
return Ok(None);
}
for line in fs::read_to_string(path)?.lines() {
let Some((name, value)) = parse_env_assignment(line) else {
continue;
};
if name == var_name {
return Ok(normalize_bearer_token(Some(value)));
}
}
Ok(None)
}
pub(crate) fn load_env_file_into_process(path: &Path) -> Result<()> {
if !path.exists() {
return Ok(());
}
for line in fs::read_to_string(path)?.lines() {
let Some((name, value)) = parse_env_assignment(line) else {
continue;
};
if std::env::var_os(&name).is_none() {
unsafe {
std::env::set_var(name, value);
}
}
}
Ok(())
}
pub(crate) fn load_cli_config(config_path: Option<&PathBuf>) -> Result<OmnigraphConfig> {
let config = load_config(config_path)?;
if let Some(path) = config.resolve_auth_env_file() {
load_env_file_into_process(&path)?;
}
Ok(config)
}
#[derive(Debug, Clone)]
pub(crate) struct ResolvedCliGraph {
pub(crate) uri: String,
pub(crate) selected: Option<String>,
pub(crate) graph_id: String,
pub(crate) policy_file: Option<PathBuf>,
pub(crate) is_remote: bool,
}
impl ResolvedCliGraph {
pub(crate) fn selected(&self) -> Option<&str> {
self.selected.as_deref()
}
}
pub(crate) struct ResolvedPolicyContext {
pub(crate) policy_file: PathBuf,
pub(crate) graph_id: String,
}
pub(crate) fn resolve_policy_context(config: &OmnigraphConfig) -> Result<ResolvedPolicyContext> {
let selected = config.resolve_policy_tooling_graph_selection()?;
let policy_file = config.resolve_policy_file_for(selected).ok_or_else(|| {
color_eyre::eyre::eyre!(
"policy.file or graphs.<name>.policy.file must be set in omnigraph.yaml"
)
})?;
let graph_id = match selected {
Some(name) => graph_resource_id_for_selection(Some(name), ""),
None => graph_resource_id_for_selection(None, "default"),
/// Resolve the cluster for a control-plane tooling command (`policy`,
/// `queries`) from `--cluster`. A configured name (`clusters:` in operator
/// config) is rewritten to its root; a literal dir / `s3://`/`file://` root is
/// passed through. A `--profile`/`OMNIGRAPH_PROFILE` cluster binding also
/// resolves here when `--cluster` is absent. No omnigraph.yaml.
pub(crate) fn require_cluster_scope(
cluster: Option<&str>,
profile: Option<&str>,
command: &str,
) -> Result<String> {
let op = operator::load_operator_config()?;
let resolve_name = |name: &str| {
op.cluster_root(name)
.map(str::to_string)
.unwrap_or_else(|| name.to_string())
};
Ok(ResolvedPolicyContext {
policy_file,
graph_id,
})
if let Some(cluster) = cluster {
return Ok(resolve_name(cluster));
}
// A cluster profile (flag, else OMNIGRAPH_PROFILE) binds the cluster too.
let profile_name = profile
.map(str::to_string)
.or_else(|| std::env::var(scope::PROFILE_ENV).ok().filter(|s| !s.is_empty()));
if let Some(name) = profile_name {
let profile = op.profile(&name).ok_or_else(|| {
color_eyre::eyre::eyre!("unknown profile '{name}' (not defined under `profiles:`)")
})?;
if let crate::operator::ScopeBinding::Cluster(cluster) = profile.binding(&name)? {
return Ok(resolve_name(&cluster));
}
}
bail!(
"`{command}` needs a cluster — pass --cluster <dir|uri> (or a name from `clusters:` \
in ~/.omnigraph/config.yaml), or select a cluster profile"
)
}
pub(crate) fn resolve_policy_engine(context: &ResolvedPolicyContext) -> Result<PolicyEngine> {
PolicyEngine::load_graph(&context.policy_file, &context.graph_id)
/// Read a cluster's serving snapshot for a control-plane tooling command,
/// flattening the readiness `Diagnostic` list into one loud error. The single
/// snapshot entry point for `policy`/`queries` so the not-servable message stays
/// identical across them.
async fn read_serving_snapshot_or_report(
cluster: &str,
) -> Result<omnigraph_cluster::ServingSnapshot> {
omnigraph_cluster::read_serving_snapshot(cluster)
.await
.map_err(|diagnostics| {
color_eyre::eyre::eyre!(
"cluster `{cluster}` is not servable:\n {}",
diagnostics
.iter()
.map(|d| d.message.clone())
.collect::<Vec<_>>()
.join("\n ")
)
})
}
pub(crate) fn resolve_policy_engine_for_graph(graph: &ResolvedCliGraph) -> Result<PolicyEngine> {
let policy_file = graph.policy_file.as_ref().ok_or_else(|| {
color_eyre::eyre::eyre!(
"policy.file or graphs.<name>.policy.file must be set in omnigraph.yaml"
)
})?;
PolicyEngine::load_graph(policy_file, &graph.graph_id)
/// Resolve the Cedar policy bundle(s) for a `--cluster` policy-tooling command
/// (RFC-011). Sources the applied policies from the cluster's serving snapshot;
/// each `ServingPolicy` carries its `source` (digest-verified content) and the
/// scopes it `applies_to` (`cluster` | `graph.<id>`). The optional `graph`
/// selects a graph's bundle when several apply.
pub(crate) async fn read_cluster_policies(
cluster: &str,
) -> Result<Vec<omnigraph_cluster::ServingPolicy>> {
Ok(read_serving_snapshot_or_report(cluster).await?.policies)
}
pub(crate) async fn open_local_db_with_policy(graph: &ResolvedCliGraph) -> Result<Omnigraph> {
let db = Omnigraph::open(&graph.uri).await?;
if graph.policy_file.is_some() {
let engine = Arc::new(resolve_policy_engine_for_graph(graph)?);
Ok(db.with_policy(engine as Arc<dyn omnigraph_policy::PolicyChecker>))
} else {
Ok(db)
/// Pick the single policy bundle that applies to the selection. With `--graph`,
/// the bundle bound to `graph.<id>` (or the cluster-wide one); without it, the
/// sole bundle if there's exactly one. Ambiguity or absence is a loud error.
pub(crate) fn select_cluster_policy<'p>(
cluster: &str,
policies: &'p [omnigraph_cluster::ServingPolicy],
graph: Option<&str>,
) -> Result<&'p omnigraph_cluster::ServingPolicy> {
if let Some(graph_id) = graph {
let graph_ref = format!("graph.{graph_id}");
let matching: Vec<&omnigraph_cluster::ServingPolicy> = policies
.iter()
.filter(|p| {
p.applies_to
.iter()
.any(|s| s == &graph_ref || s == "cluster")
})
.collect();
return match matching.as_slice() {
[only] => Ok(only),
[] => bail!(
"cluster `{cluster}` has no policy bundle bound to graph `{graph_id}` \
(or to the cluster scope)"
),
many => bail!(
"graph `{graph_id}` in cluster `{cluster}` matches {} policy bundles ([{}]); \
the cluster model expects one bundle per graph scope",
many.len(),
many.iter().map(|p| p.name.as_str()).collect::<Vec<_>>().join(", ")
),
};
}
match policies {
[only] => Ok(only),
[] => bail!("cluster `{cluster}` has no applied policy bundles"),
many => bail!(
"cluster `{cluster}` has {} policy bundles ([{}]); pass --graph <id> to select one",
many.len(),
many.iter().map(|p| p.name.as_str()).collect::<Vec<_>>().join(", ")
),
}
}
/// THE actor chain (RFC-007 §D3) — every command that needs an identity
/// THE actor chain (RFC-011) — every command that needs an identity
/// resolves through this one function (one path per concern):
/// `--as` > legacy `cli.actor` in omnigraph.yaml (RFC-008 window) >
/// `operator.actor` in ~/.omnigraph/config.yaml > none.
pub(crate) fn resolve_actor(
cli_as: Option<&str>,
legacy_config_actor: Option<&str>,
) -> Result<Option<String>> {
/// `--as` > `operator.actor` in ~/.omnigraph/config.yaml > none.
pub(crate) fn resolve_actor(cli_as: Option<&str>) -> Result<Option<String>> {
if let Some(actor) = cli_as {
return Ok(Some(actor.to_string()));
}
if let Some(actor) = legacy_config_actor {
return Ok(Some(actor.to_string()));
}
Ok(operator::load_operator_config()?
.actor()
.map(str::to_string))
}
pub(crate) fn resolve_cluster_actor(cli_as: Option<&str>) -> Result<Option<String>> {
if let Some(actor) = cli_as {
return Ok(Some(actor.to_string()));
}
let config = load_config(None).wrap_err(
"resolving the default actor from omnigraph.yaml (pass --as <ACTOR> to skip this lookup)",
)?;
resolve_actor(None, config.cli.actor.as_deref())
resolve_actor(cli_as)
}
pub(crate) fn resolve_cli_actor(
cli_as: Option<&str>,
config: &OmnigraphConfig,
) -> Result<Option<String>> {
resolve_actor(cli_as, config.cli.actor.as_deref())
pub(crate) fn resolve_cli_actor(cli_as: Option<&str>) -> Result<Option<String>> {
resolve_actor(cli_as)
}
pub(crate) fn resolve_policy_tests_path(context: &ResolvedPolicyContext) -> PathBuf {
context.policy_file.with_file_name("policy.tests.yaml")
}
pub(crate) fn normalize_policy_graph_uri(uri: &str) -> Result<String> {
if is_remote_uri(uri) {
Ok(uri.trim_end_matches('/').to_string())
} else {
Ok(normalize_root_uri(uri)?)
}
}
pub(crate) fn resolve_remote_bearer_token(
config: &OmnigraphConfig,
explicit_uri: Option<&str>,
) -> Result<Option<String>> {
// `--target` is gone; the legacy explicit-target name is always None.
let explicit_target: Option<&str> = None;
/// The bearer token for a remote request (RFC-011): the operator keyed chain
/// for the matching server (`OMNIGRAPH_TOKEN_<NAME>` env → 0600 credentials
/// file), then the default `OMNIGRAPH_BEARER_TOKEN` env. No omnigraph.yaml
/// chain.
pub(crate) fn resolve_remote_bearer_token(explicit_uri: Option<&str>) -> Result<Option<String>> {
// The keyed hop (RFC-007 §D4, gh-host model): when the effective remote
// URL belongs to an operator-defined server, that server's keyed chain
// applies first — OMNIGRAPH_TOKEN_<NAME> env, then the 0600 credentials
// file. Ok(None) falls through to the legacy chain unchanged, and the
// keyed token is structurally scoped to its own server (§D5 rule 3):
// a URL matching no operator server never sees it.
if let Some(remote_url) = effective_remote_url(config, explicit_uri, explicit_target) {
// file. The keyed token is structurally scoped to its own server: a URL
// matching no operator server never sees it.
if let Some(remote_url) = explicit_uri.filter(|uri| is_remote_uri(uri)) {
let operator_config = operator::load_operator_config()?;
if let Some(server) = operator_config.find_server_for_url(&remote_url) {
if let Some(server) = operator_config.find_server_for_url(remote_url) {
if let Some(token) = operator::resolve_keyed_token(server)? {
return Ok(Some(token));
}
}
}
let scoped_env =
config.graph_bearer_token_env(explicit_uri, explicit_target, config.cli_graph_name());
let mut env_names = Vec::new();
if let Some(name) = scoped_env {
env_names.push(name.to_string());
}
if env_names
.iter()
.all(|name| name != DEFAULT_BEARER_TOKEN_ENV)
{
env_names.push(DEFAULT_BEARER_TOKEN_ENV.to_string());
}
let env_file = config.resolve_auth_env_file();
for env_name in env_names {
if let Some(token) = bearer_token_from_env(&env_name) {
return Ok(Some(token));
}
if let Some(path) = env_file.as_ref() {
if let Some(token) = bearer_token_from_env_file(path, &env_name)? {
return Ok(Some(token));
}
}
}
Ok(None)
Ok(bearer_token_from_env(DEFAULT_BEARER_TOKEN_ENV))
}
/// `--server <name>` (RFC-007 PR 3): resolve an operator-defined server
@ -391,7 +324,6 @@ pub(crate) fn resolve_server_flag(
/// params. The keyed token applies via the ordinary URL match.
pub(crate) async fn execute_operator_alias(
client: &reqwest::Client,
config: &OmnigraphConfig,
alias_name: &str,
alias: &crate::operator::OperatorAlias,
alias_args: &[String],
@ -399,7 +331,7 @@ pub(crate) async fn execute_operator_alias(
) -> Result<ReadOutput> {
let uri = resolve_server_flag(Some(&alias.server), alias.graph.as_deref())?
.expect("server name is present");
let bearer_token = resolve_remote_bearer_token(config, Some(&uri))?;
let bearer_token = resolve_remote_bearer_token(Some(&uri))?;
let mut params = serde_json::Map::new();
for (key, value) in &alias.params {
@ -454,22 +386,6 @@ pub(crate) fn apply_server_flag(
resolve_server_flag(server, graph)
}
/// The remote base URL a token resolution is FOR — the same scoping
/// `graph_bearer_token_env` uses: an explicit http(s) `--uri` wins, else
/// the config-resolved target's uri (when remote). Local URIs → None.
fn effective_remote_url(
config: &OmnigraphConfig,
explicit_uri: Option<&str>,
explicit_target: Option<&str>,
) -> Option<String> {
if let Some(uri) = explicit_uri {
return is_remote_uri(uri).then(|| uri.to_string());
}
let target = config.resolve_target_name(explicit_uri, explicit_target, config.cli_graph_name())?;
let uri = &config.graphs.get(target)?.uri;
is_remote_uri(uri).then(|| uri.clone())
}
pub(crate) fn build_http_client() -> Result<reqwest::Client> {
Ok(reqwest::Client::new())
}
@ -510,40 +426,31 @@ pub(crate) async fn remote_json<T: DeserializeOwned>(
Ok(serde_json::from_str(&text)?)
}
pub(crate) fn resolve_uri(config: &OmnigraphConfig, cli_uri: Option<String>) -> Result<String> {
// `--target` is gone; the second arg (the legacy explicit-target name) is
// always None. A bare command still falls back to `cli.graph` (the third arg).
config.resolve_target_uri(cli_uri, None, config.cli_graph_name())
/// The graph URI a command addresses (RFC-011): the scope-resolved URI string
/// (positional URI / `--store` / `--profile` / `defaults.store`). No
/// omnigraph.yaml `cli.graph` fallback — an absent address is a loud error.
pub(crate) fn resolve_uri(cli_uri: Option<String>) -> Result<String> {
cli_uri.ok_or_else(|| {
color_eyre::eyre::eyre!(
"no graph addressed — pass a positional URI, --store <uri>, --server <name>, \
--profile <name>, or set a default scope in ~/.omnigraph/config.yaml"
)
})
}
pub(crate) fn resolve_cli_graph(
config: &OmnigraphConfig,
cli_uri: Option<String>,
) -> Result<ResolvedCliGraph> {
let selected = if cli_uri.is_some() {
None
} else {
config.cli_graph_name().map(str::to_string)
};
config.resolve_graph_selection(selected.as_deref())?;
let uri = resolve_uri(config, cli_uri)?;
let normalized_uri = normalize_policy_graph_uri(&uri)?;
let graph_id = graph_resource_id_for_selection(selected.as_deref(), &normalized_uri);
pub(crate) fn resolve_cli_graph(cli_uri: Option<String>) -> Result<ResolvedCliGraph> {
let uri = resolve_uri(cli_uri)?;
Ok(ResolvedCliGraph {
graph_id,
is_remote: is_remote_uri(&uri),
policy_file: config.resolve_policy_file_for(selected.as_deref()),
selected,
uri,
})
}
pub(crate) fn resolve_local_graph(
config: &OmnigraphConfig,
cli_uri: Option<String>,
operation: &str,
) -> Result<ResolvedCliGraph> {
let graph = resolve_cli_graph(config, cli_uri)?;
let graph = resolve_cli_graph(cli_uri)?;
if graph.is_remote {
bail!(
"`{}` is a direct (storage-native) command and needs direct storage \
@ -586,22 +493,19 @@ pub(crate) fn parse_duration_arg(s: &str) -> Result<std::time::Duration> {
Ok(std::time::Duration::from_secs(secs))
}
pub(crate) fn resolve_local_uri(
config: &OmnigraphConfig,
cli_uri: Option<String>,
operation: &str,
) -> Result<String> {
Ok(resolve_local_graph(config, cli_uri, operation)?.uri)
pub(crate) fn resolve_local_uri(cli_uri: Option<String>, operation: &str) -> Result<String> {
Ok(resolve_local_graph(cli_uri, operation)?.uri)
}
/// Resolve a maintenance verb's (optimize/repair/cleanup) address to a direct
/// storage URI through the one RFC-011 scope path. Every primitive funnels
/// here: a positional URI, `--store`, `--cluster <root> --graph <id>`, a
/// `--profile` cluster binding, or operator defaults — all resolved at the
/// `Direct` capability (so a server scope is rejected, a cluster scope is
/// allowed), then mapped to a storage URI by `resolve_storage_uri`.
/// Resolve a direct (storage-native) verb's address to a storage URI through the
/// one RFC-011 scope path — the maintenance verbs (optimize/repair/cleanup) plus
/// `schema plan` and `lint`'s graph-target path. Every primitive funnels here: a
/// positional URI, `--store`, `--cluster <root> --graph <id>`, a `--profile`
/// cluster binding, or operator defaults — all resolved at the `Direct`
/// capability (so a server scope is rejected, a cluster scope is allowed when the
/// verb opts into cluster addressing), then mapped to a storage URI by
/// `resolve_storage_uri`.
pub(crate) async fn resolve_maintenance_uri(
config: &OmnigraphConfig,
profile: Option<&str>,
store: Option<&str>,
cluster: Option<&str>,
@ -622,7 +526,6 @@ pub(crate) async fn resolve_maintenance_uri(
},
)?;
resolve_storage_uri(
config,
scope.uri,
scope.cluster.as_deref(),
scope.cluster_graph.as_deref(),
@ -639,7 +542,6 @@ pub(crate) async fn resolve_maintenance_uri(
/// automatically, otherwise error and list the candidates so the operator can
/// pass `--graph <id>`.
pub(crate) async fn resolve_storage_uri(
config: &OmnigraphConfig,
cli_uri: Option<String>,
cluster: Option<&str>,
cluster_graph: Option<&str>,
@ -651,7 +553,7 @@ pub(crate) async fn resolve_storage_uri(
let graph_id = resolve_sole_cluster_graph(cluster).await?;
resolve_cluster_graph_uri(cluster, &graph_id).await
}
(None, None) => resolve_local_uri(config, cli_uri, operation),
(None, None) => resolve_local_uri(cli_uri, operation),
(None, Some(_)) => {
bail!("internal error: a graph was selected without a cluster scope")
}
@ -687,19 +589,16 @@ async fn resolve_cluster_graph_uri(cluster: &str, graph_id: &str) -> Result<Stri
}
pub(crate) fn resolve_branch(
config: &OmnigraphConfig,
cli_branch: Option<String>,
alias_branch: Option<String>,
default_branch: &str,
) -> String {
cli_branch
.or(alias_branch)
.or_else(|| config.cli.branch.clone())
.unwrap_or_else(|| default_branch.to_string())
}
pub(crate) fn resolve_read_target(
config: &OmnigraphConfig,
cli_branch: Option<String>,
cli_snapshot: Option<String>,
alias_branch: Option<String>,
@ -707,19 +606,15 @@ pub(crate) fn resolve_read_target(
if cli_branch.is_some() && cli_snapshot.is_some() {
bail!("read target may specify branch or snapshot, not both");
}
Ok(read_target_from_cli(
cli_branch
.or(alias_branch)
.or_else(|| config.cli.branch.clone()),
cli_snapshot,
))
Ok(read_target_from_cli(cli_branch.or(alias_branch), cli_snapshot))
}
pub(crate) fn resolve_query_path(
config: &OmnigraphConfig,
explicit_query: Option<&PathBuf>,
alias_query: Option<&str>,
) -> Result<PathBuf> {
// The `.gq` path is resolved plainly (cwd-relative) — no omnigraph.yaml
// `query.roots` search.
explicit_query
.map(PathBuf::from)
.or_else(|| alias_query.map(PathBuf::from))
@ -728,11 +623,9 @@ pub(crate) fn resolve_query_path(
"exactly one of --query, --query-string, or --alias must be provided"
)
})
.and_then(|query_path| config.resolve_query_path(&query_path))
}
pub(crate) fn resolve_query_source(
config: &OmnigraphConfig,
explicit_query: Option<&PathBuf>,
inline_query: Option<&str>,
alias_query: Option<&str>,
@ -744,7 +637,6 @@ pub(crate) fn resolve_query_source(
return Ok(inline.to_string());
}
Ok(fs::read_to_string(resolve_query_path(
config,
explicit_query,
alias_query,
)?)?)
@ -754,11 +646,9 @@ pub(crate) fn parse_alias_value(value: &str) -> Value {
serde_json::from_str(value).unwrap_or_else(|_| Value::String(value.to_string()))
}
/// The format cascade (RFC-007 §D3): `--json` > `--format` > alias format >
/// legacy `cli.output_format` (RFC-008 window) > operator `defaults.output`
/// > table.
/// The format cascade (RFC-011): `--json` > `--format` > alias format >
/// operator `defaults.output` > table.
pub(crate) fn resolve_read_format(
config: &OmnigraphConfig,
cli_format: Option<ReadOutputFormat>,
json: bool,
alias_format: Option<ReadOutputFormat>,
@ -768,7 +658,6 @@ pub(crate) fn resolve_read_format(
}
cli_format
.or(alias_format)
.or(config.cli.output_format)
.or_else(|| {
operator::load_operator_config()
.ok()
@ -825,12 +714,11 @@ pub(crate) fn query_params_from_json(
}
pub(crate) async fn execute_query_lint(
config: &OmnigraphConfig,
cli_uri: Option<String>,
schema_path: Option<&PathBuf>,
query_path: &PathBuf,
) -> Result<QueryLintOutput> {
let resolved_query_path = resolve_query_path(config, Some(query_path), None)?;
let resolved_query_path = resolve_query_path(Some(query_path), None)?;
let query_source = fs::read_to_string(&resolved_query_path)?;
let query_path = resolved_query_path.to_string_lossy().into_owned();
@ -848,12 +736,14 @@ pub(crate) async fn execute_query_lint(
));
}
let has_graph_target = cli_uri.is_some() || config.cli_graph_name().is_some();
if !has_graph_target {
bail!("lint requires --schema <schema.pg> or a resolvable graph target");
if cli_uri.is_none() {
bail!(
"lint requires --schema <schema.pg> (offline) or a graph target \
(--store <uri> / --cluster <dir> --graph <id>)"
);
}
let uri = resolve_local_uri(config, cli_uri, "lint")?;
let uri = resolve_local_uri(cli_uri, "lint")?;
let db = Omnigraph::open(&uri).await?;
Ok(lint_query_file(
&db.catalog(),
@ -863,20 +753,24 @@ pub(crate) async fn execute_query_lint(
))
}
pub(crate) fn resolve_selected_graph(
config: &OmnigraphConfig,
cli_uri: Option<String>,
operation: &str,
) -> Result<(String, Option<String>)> {
let graph = resolve_local_graph(config, cli_uri, operation)?;
Ok((graph.uri, graph.selected))
}
pub(crate) fn load_registry_or_report(
config: &OmnigraphConfig,
selected: Option<&str>,
/// Build a `QueryRegistry` from a cluster serving snapshot's stored queries,
/// optionally scoped to one graph. The `ServingQuery.source` is the
/// digest-verified `.gq` content, so no file I/O or omnigraph.yaml is involved.
fn registry_from_serving_queries(
queries: &[omnigraph_cluster::ServingQuery],
graph: Option<&str>,
) -> Result<QueryRegistry> {
QueryRegistry::load(config, config.query_entries_for(selected)).map_err(|errors| {
let specs: Vec<omnigraph_server::queries::RegistrySpec> = queries
.iter()
.filter(|q| graph.is_none_or(|g| q.graph_id == g))
.map(|q| omnigraph_server::queries::RegistrySpec {
name: q.name.clone(),
source: q.source.clone(),
expose: false,
tool_name: None,
})
.collect();
QueryRegistry::from_specs(specs).map_err(|errors| {
color_eyre::eyre::eyre!(
"stored-query registry failed to load:\n {}",
errors
@ -888,83 +782,58 @@ pub(crate) fn load_registry_or_report(
})
}
pub(crate) fn graph_query_registry_names(config: &OmnigraphConfig) -> Vec<&str> {
config
.graphs
.iter()
.filter_map(|(name, graph)| (!graph.queries.is_empty()).then_some(name.as_str()))
.collect()
}
pub(crate) fn resolve_registry_selection_for_list(
config: &OmnigraphConfig,
) -> Result<Option<String>> {
let selected = config.cli_graph_name().map(str::to_string);
if let Some(name) = selected.as_deref() {
config.resolve_graph_selection(Some(name))?;
return Ok(selected);
}
if !config.query_entries().is_empty() {
return Ok(None);
}
let graph_names = graph_query_registry_names(config);
if graph_names.is_empty() {
return Ok(None);
}
bail!(
"stored-query registries are configured for graph{} {} but no graph was selected. Pass a positional URI or set `cli.graph`.",
if graph_names.len() == 1 { "" } else { "s" },
graph_names.join(", "),
)
}
pub(crate) fn validate_registry_for_catalog(
registry: &QueryRegistry,
catalog: &omnigraph_compiler::catalog::Catalog,
label: &str,
) -> omnigraph::error::Result<()> {
let report = check(registry, catalog);
if report.has_breakages() {
return Err(omnigraph::error::OmniError::manifest(
format_check_breakages(label, &report),
));
}
Ok(())
}
/// `queries validate --cluster <dir>` (RFC-011): type-check every stored query
/// in the cluster catalog against its graph's applied schema. Both the registry
/// and the schemas come from the cluster serving snapshot — no omnigraph.yaml.
/// With `--graph`, scope to a single graph.
pub(crate) async fn execute_queries_validate(
uri: Option<String>,
config_path: Option<&PathBuf>,
cluster: &str,
graph: Option<&str>,
json: bool,
) -> Result<()> {
let config = load_cli_config(config_path)?;
// One selection drives both the schema URI and the registry.
let (uri, selected) = resolve_selected_graph(&config, uri, "queries validate")?;
let registry = load_registry_or_report(&config, selected.as_deref())?;
let db = Omnigraph::open(&uri).await?;
let report = check(&registry, &db.catalog());
let snapshot = read_serving_snapshot_or_report(cluster).await?;
let output = QueriesValidateOutput {
ok: !report.has_breakages(),
breakages: report
.breakages
.iter()
.map(|b| QueriesIssue {
// Type-check per graph: each graph's stored queries against its own schema
// (read from the graph's applied storage root). A `--graph` filter scopes to
// exactly one graph; an unknown id is a loud error.
let mut breakages = Vec::new();
let mut warnings = Vec::new();
let mut total = 0usize;
let mut matched_any = false;
for serving_graph in &snapshot.graphs {
if graph.is_some_and(|g| g != serving_graph.graph_id) {
continue;
}
matched_any = true;
let registry = registry_from_serving_queries(&snapshot.queries, Some(&serving_graph.graph_id))?;
let db = Omnigraph::open(&serving_graph.root.to_string_lossy()).await?;
let report = check(&registry, &db.catalog());
total += registry.len();
for b in &report.breakages {
breakages.push(QueriesIssue {
query: b.query.clone(),
message: b.message.clone(),
})
.collect(),
warnings: report
.warnings
.iter()
.map(|w| QueriesIssue {
});
}
for w in &report.warnings {
warnings.push(QueriesIssue {
query: w.query.clone(),
message: w.message.clone(),
})
.collect(),
});
}
}
if let Some(graph_id) = graph {
if !matched_any {
bail!("graph `{graph_id}` is not applied in cluster `{cluster}`");
}
}
let has_breakages = !breakages.is_empty();
let output = QueriesValidateOutput {
ok: !has_breakages,
breakages,
warnings,
};
if json {
@ -973,8 +842,8 @@ pub(crate) async fn execute_queries_validate(
if output.breakages.is_empty() {
println!(
"OK {} stored quer{} type-check against the schema",
registry.len(),
if registry.len() == 1 { "y" } else { "ies" }
total,
if total == 1 { "y" } else { "ies" }
);
}
for issue in &output.breakages {
@ -985,17 +854,22 @@ pub(crate) async fn execute_queries_validate(
}
}
if report.has_breakages() {
if has_breakages {
io::stdout().flush()?;
std::process::exit(1);
}
Ok(())
}
pub(crate) fn execute_queries_list(config_path: Option<&PathBuf>, json: bool) -> Result<()> {
let config = load_cli_config(config_path)?;
let selected = resolve_registry_selection_for_list(&config)?;
let registry = load_registry_or_report(&config, selected.as_deref())?;
/// `queries list --cluster <dir>` (RFC-011): list the catalog's stored queries.
/// With `--graph`, scope to one graph.
pub(crate) async fn execute_queries_list(
cluster: &str,
graph: Option<&str>,
json: bool,
) -> Result<()> {
let snapshot = read_serving_snapshot_or_report(cluster).await?;
let registry = registry_from_serving_queries(&snapshot.queries, graph)?;
let output = QueriesListOutput {
queries: registry

261
crates/omnigraph-cli/src/main.rs
View file

@ -1,15 +1,11 @@
use std::ffi::OsString;
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, WrapErr, bail};
use color_eyre::eyre::{Result, bail};
use omnigraph::db::{Omnigraph, ReadTarget, SnapshotId};
use omnigraph::loader::LoadMode;
use omnigraph::storage::normalize_root_uri;
use omnigraph_cluster::{
ApplyOptions, ApplyOutput, ApproveOutput, DiagnosticSeverity, ForceUnlockOutput, PlanOutput, StateSyncOutput, StatusOutput,
ValidateOutput, apply_config_dir_with_options, approve_config_dir, force_unlock_config_dir, import_config_dir, plan_config_dir,
@ -26,9 +22,9 @@ use omnigraph_api_types::{
ChangeOutput, CommitOutput, ErrorOutput, IngestOutput, ReadOutput, SchemaApplyOutput,
SnapshotTableOutput,
};
use omnigraph_server::queries::{QueryRegistry, check, format_check_breakages};
use omnigraph_server::queries::{QueryRegistry, check};
use omnigraph_server::{
OmnigraphConfig, PolicyAction, PolicyDecision, PolicyEngine, PolicyRequest,
PolicyAction, PolicyDecision, PolicyEngine, PolicyRequest,
PolicyTestConfig, ReadOutputFormat, graph_resource_id_for_selection, load_config,
};
use reqwest::Method;
@ -170,16 +166,13 @@ async fn main() -> Result<()> {
}
Command::Load {
uri,
config,
data,
branch,
from,
mode,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve_with_policy(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -188,7 +181,7 @@ async fn main() -> Result<()> {
cli.store.as_deref(),
)
.await?;
let branch = resolve_branch(&config, branch, None, "main");
let branch = resolve_branch(branch, None, "main");
if matches!(mode, CliLoadMode::Overwrite) {
confirm_destructive("load --mode overwrite", client.uri(), cli.yes, json)?;
}
@ -204,7 +197,6 @@ async fn main() -> Result<()> {
}
Command::Ingest {
uri,
config,
data,
branch,
from,
@ -216,9 +208,7 @@ async fn main() -> Result<()> {
"warning: `omnigraph ingest` is deprecated and will be removed in a future release; \
use `omnigraph load --from <base> --mode <mode>` (ingest defaults: --from main --mode merge)"
);
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve_with_policy(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -227,8 +217,8 @@ async fn main() -> Result<()> {
cli.store.as_deref(),
)
.await?;
let branch = resolve_branch(&config, branch, None, "main");
let from = resolve_branch(&config, from, None, "main");
let branch = resolve_branch(branch, None, "main");
let from = resolve_branch(from, None, "main");
echo_write_target(cli.quiet, "ingest", client.uri(), client.is_remote());
let payload = client
.ingest(&branch, &from, &data.to_string_lossy(), mode)
@ -242,14 +232,11 @@ async fn main() -> Result<()> {
Command::Branch { command } => match command {
BranchCommand::Create {
uri,
config,
from,
name,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve_with_policy(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -258,7 +245,7 @@ async fn main() -> Result<()> {
cli.store.as_deref(),
)
.await?;
let from = resolve_branch(&config, from, None, "main");
let from = resolve_branch(from, None, "main");
echo_write_target(cli.quiet, "branch create", client.uri(), client.is_remote());
let payload = client.branch_create_from(&from, &name).await?;
if json {
@ -269,12 +256,9 @@ async fn main() -> Result<()> {
}
BranchCommand::List {
uri,
config,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -293,13 +277,10 @@ async fn main() -> Result<()> {
}
BranchCommand::Delete {
uri,
config,
name,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve_with_policy(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -319,14 +300,11 @@ async fn main() -> Result<()> {
}
BranchCommand::Merge {
uri,
config,
source,
into,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve_with_policy(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -335,7 +313,7 @@ async fn main() -> Result<()> {
cli.store.as_deref(),
)
.await?;
let into = resolve_branch(&config, into, None, "main");
let into = resolve_branch(into, None, "main");
echo_write_target(cli.quiet, "branch merge", client.uri(), client.is_remote());
let payload = client.branch_merge(&source, &into).await?;
if json {
@ -353,13 +331,10 @@ async fn main() -> Result<()> {
Command::Commit { command } => match command {
CommitCommand::List {
uri,
config,
branch,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -376,13 +351,10 @@ async fn main() -> Result<()> {
}
CommitCommand::Show {
uri,
config,
commit_id,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -401,13 +373,19 @@ async fn main() -> Result<()> {
Command::Schema { command } => match command {
SchemaCommand::Plan {
uri,
config,
schema,
json,
allow_data_loss,
} => {
let config = load_cli_config(config.as_ref())?;
let uri = resolve_local_uri(&config, uri, "schema plan")?;
let uri = resolve_maintenance_uri(
cli.profile.as_deref(),
cli.store.as_deref(),
cli.cluster.as_deref(),
cli.graph.as_deref(),
uri,
"schema plan",
)
.await?;
let schema_source = fs::read_to_string(&schema)?;
let db = Omnigraph::open(&uri).await?;
let plan = db
@ -430,14 +408,11 @@ async fn main() -> Result<()> {
}
SchemaCommand::Apply {
uri,
config,
schema,
json,
allow_data_loss,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve_with_policy(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -447,25 +422,14 @@ async fn main() -> Result<()> {
)
.await?;
let schema_source = fs::read_to_string(&schema)?;
// The stored-query registry check is an embedded-only concern
// (the remote arm ignores the validator — the server runs its
// own check); build it only for the local path so the remote
// path keeps its no-registry-load behavior.
let registry = if client.is_remote() {
None
} else {
let registry = load_registry_or_report(&config, client.selected())?;
(!registry.is_empty()).then_some(registry)
};
let label = client.selected().unwrap_or(client.uri()).to_string();
// The embedded (direct-store) arm carries no stored-query
// registry — the registry is cluster-owned (RFC-011), so a
// direct apply has nothing to validate against. The served arm
// runs the server's own catalog check. So the validator is a
// no-op here on both arms.
echo_write_target(cli.quiet, "schema apply", client.uri(), client.is_remote());
let output = client
.apply_schema(&schema_source, allow_data_loss, |catalog| {
if let Some(registry) = registry.as_ref() {
validate_registry_for_catalog(registry, catalog, &label)?;
}
Ok(())
})
.apply_schema(&schema_source, allow_data_loss, |_catalog| Ok(()))
.await?;
if json {
print_json(&output)?;
@ -475,12 +439,9 @@ async fn main() -> Result<()> {
}
SchemaCommand::Show {
uri,
config,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -498,41 +459,50 @@ async fn main() -> Result<()> {
},
Command::Lint {
uri,
config,
query,
schema,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let output =
execute_query_lint(&config, uri, schema.as_ref(), &query)
.await?;
// A graph target (when `--schema` is absent) resolves through the
// direct scope path (positional URI / --store / --profile /
// defaults.store). Offline (`--schema`) needs no graph, so leave
// the uri unresolved in that case.
let graph_uri = if schema.is_some() {
uri
} else {
Some(
resolve_maintenance_uri(
cli.profile.as_deref(),
cli.store.as_deref(),
cli.cluster.as_deref(),
cli.graph.as_deref(),
uri,
"lint",
)
.await?,
)
};
let output = execute_query_lint(graph_uri, schema.as_ref(), &query).await?;
finish_query_lint(&output, json)?;
}
Command::Queries { command } => match command {
QueriesCommand::Validate {
uri,
config,
json,
} => {
execute_queries_validate(uri, config.as_ref(), json).await?;
Command::Queries { command } => {
let cluster =
require_cluster_scope(cli.cluster.as_deref(), cli.profile.as_deref(), "queries")?;
match command {
QueriesCommand::Validate { json } => {
execute_queries_validate(&cluster, cli.graph.as_deref(), json).await?;
}
QueriesCommand::List { json } => {
execute_queries_list(&cluster, cli.graph.as_deref(), json).await?;
}
}
QueriesCommand::List {
config,
json,
} => {
execute_queries_list(config.as_ref(), json)?;
}
},
}
Command::Snapshot {
uri,
config,
branch,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -540,7 +510,7 @@ async fn main() -> Result<()> {
cli.store.as_deref(),
)
.await?;
let branch = resolve_branch(&config, branch, None, "main");
let branch = resolve_branch(branch, None, "main");
let payload = client.snapshot(&branch).await?;
if json {
print_json(&payload)?;
@ -550,15 +520,12 @@ async fn main() -> Result<()> {
}
Command::Export {
uri,
config,
branch,
jsonl,
type_names,
table_keys,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
@ -566,7 +533,7 @@ async fn main() -> Result<()> {
cli.store.as_deref(),
)
.await?;
let branch = resolve_branch(&config, branch, None, "main");
let branch = resolve_branch(branch, None, "main");
if jsonl {
eprintln!("warning: --jsonl is deprecated; `omnigraph export` always emits JSONL");
}
@ -579,7 +546,6 @@ async fn main() -> Result<()> {
}
Command::Query {
name,
config,
query,
query_string,
params,
@ -588,9 +554,7 @@ async fn main() -> Result<()> {
format,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
None,
@ -599,12 +563,12 @@ async fn main() -> Result<()> {
)
.await?;
let params_json = load_params_json(&params)?;
let target = resolve_read_target(&config, branch, snapshot, None)?;
let target = resolve_read_target(branch, snapshot, None)?;
let output: ReadOutput = if query.is_some() || query_string.is_some() {
// Ad-hoc lane: run the source; the positional `name` selects
// within it when it holds more than one query.
let query_source =
resolve_query_source(&config, query.as_ref(), query_string.as_deref(), None)?;
resolve_query_source(query.as_ref(), query_string.as_deref(), None)?;
client
.query(target, &query_source, name.as_deref(), params_json.as_ref())
.await?
@ -624,21 +588,18 @@ async fn main() -> Result<()> {
.invoke_named(&name, false, params_json.as_ref(), branch, snapshot)
.await?
};
let format = resolve_read_format(&config, format, json, None);
print_read_output(&output, format, &config)?;
let format = resolve_read_format(format, json, None);
print_read_output(&output, format)?;
}
Command::Mutate {
name,
config,
query,
query_string,
params,
branch,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve_with_policy(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
None,
@ -648,11 +609,11 @@ async fn main() -> Result<()> {
)
.await?;
let params_json = load_params_json(&params)?;
let branch = resolve_branch(&config, branch, None, "main");
let branch = resolve_branch(branch, None, "main");
let output: ChangeOutput = if query.is_some() || query_string.is_some() {
// Ad-hoc lane: run the source; positional `name` selects within it.
let query_source =
resolve_query_source(&config, query.as_ref(), query_string.as_deref(), None)?;
resolve_query_source(query.as_ref(), query_string.as_deref(), None)?;
client
.mutate(&branch, &query_source, name.as_deref(), params_json.as_ref())
.await?
@ -677,12 +638,10 @@ async fn main() -> Result<()> {
Command::Alias {
name,
args,
config,
params,
format,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let operator_config = crate::operator::load_operator_config()?;
let Some(operator_alias) = operator_config.aliases.get(&name) else {
let defined: Vec<&str> =
@ -695,59 +654,64 @@ async fn main() -> Result<()> {
};
let output = execute_operator_alias(
&http_client,
&config,
&name,
operator_alias,
&args,
load_params_json(&params)?,
)
.await?;
let format = resolve_read_format(&config, format, json, operator_alias.format);
print_read_output(&output, format, &config)?;
let format = resolve_read_format(format, json, operator_alias.format);
print_read_output(&output, format)?;
}
Command::Policy { command } => match command {
PolicyCommand::Validate { config } => {
let config = load_cli_config(config.as_ref())?;
let context = resolve_policy_context(&config)?;
let engine = resolve_policy_engine(&context)?;
println!(
"policy valid: {} [{} actors]",
context.policy_file.display(),
engine.known_actor_count()
);
}
PolicyCommand::Test { config } => {
let config = load_cli_config(config.as_ref())?;
let context = resolve_policy_context(&config)?;
let engine = resolve_policy_engine(&context)?;
let tests_path = resolve_policy_tests_path(&context);
let tests = PolicyTestConfig::load(&tests_path)?;
engine.run_tests(&tests)?;
println!("policy tests passed: {} cases", tests.cases.len());
}
PolicyCommand::Explain {
config,
actor,
action,
branch,
target_branch,
} => {
let config = load_cli_config(config.as_ref())?;
let context = resolve_policy_context(&config)?;
let engine = resolve_policy_engine(&context)?;
let request = PolicyRequest {
Command::Policy { command } => {
// Policy tooling sources the Cedar bundle(s) from the cluster's
// applied policies (RFC-011): --cluster <dir>, + the global --graph
// to pick a graph's bundle when several apply.
let cluster =
require_cluster_scope(cli.cluster.as_deref(), cli.profile.as_deref(), "policy")?;
let graph = cli.graph.as_deref();
let graph_id = match graph {
Some(id) => graph_resource_id_for_selection(Some(id), ""),
None => graph_resource_id_for_selection(None, "default"),
};
let policies = read_cluster_policies(&cluster).await?;
match command {
PolicyCommand::Validate {} => {
let bundle = select_cluster_policy(&cluster, &policies, graph)?;
let engine = PolicyEngine::load_graph_from_source(&bundle.source, &graph_id)?;
println!(
"policy valid: bundle '{}' [{} actors]",
bundle.name,
engine.known_actor_count()
);
}
PolicyCommand::Test { tests } => {
let bundle = select_cluster_policy(&cluster, &policies, graph)?;
let engine = PolicyEngine::load_graph_from_source(&bundle.source, &graph_id)?;
let tests = PolicyTestConfig::load(&tests)?;
engine.run_tests(&tests)?;
println!("policy tests passed: {} cases", tests.cases.len());
}
PolicyCommand::Explain {
actor,
action,
branch,
target_branch,
};
let decision = engine.authorize(&actor, &request)?;
print_policy_explain(&decision, &actor, &request);
} => {
let bundle = select_cluster_policy(&cluster, &policies, graph)?;
let engine = PolicyEngine::load_graph_from_source(&bundle.source, &graph_id)?;
let request = PolicyRequest {
action,
branch,
target_branch,
};
let decision = engine.authorize(&actor, &request)?;
print_policy_explain(&decision, &actor, &request);
}
}
},
Command::Optimize { uri, config, json } => {
let config = load_cli_config(config.as_ref())?;
}
Command::Optimize { uri, json } => {
let uri = resolve_maintenance_uri(
&config,
cli.profile.as_deref(),
cli.store.as_deref(),
cli.cluster.as_deref(),
@ -798,14 +762,11 @@ async fn main() -> Result<()> {
}
Command::Repair {
uri,
config,
confirm,
force,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let uri = resolve_maintenance_uri(
&config,
cli.profile.as_deref(),
cli.store.as_deref(),
cli.cluster.as_deref(),
@ -890,15 +851,12 @@ async fn main() -> Result<()> {
}
Command::Cleanup {
uri,
config,
keep,
older_than,
confirm,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let uri = resolve_maintenance_uri(
&config,
cli.profile.as_deref(),
cli.store.as_deref(),
cli.cluster.as_deref(),
@ -1036,12 +994,9 @@ async fn main() -> Result<()> {
Command::Graphs { command } => match command {
GraphsCommand::List {
uri,
config,
json,
} => {
let config = load_cli_config(config.as_ref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,

321
crates/omnigraph-cli/src/main_tests.rs
View file

@ -1,22 +1,16 @@
//! In-source test suite for the CLI binary (moved verbatim from
//! main.rs; `use super::*` resolves through the #[path] declaration).
use std::fs;
use super::{
DEFAULT_BEARER_TOKEN_ENV, apply_bearer_token, bearer_token_from_env_file,
legacy_change_request_body, load_cli_config, load_env_file_into_process,
normalize_bearer_token, parse_env_assignment, resolve_cli_graph, resolve_policy_context,
resolve_remote_bearer_token,
DEFAULT_BEARER_TOKEN_ENV, apply_bearer_token, legacy_change_request_body,
normalize_bearer_token, resolve_remote_bearer_token,
};
use omnigraph_server::load_config;
use reqwest::header::AUTHORIZATION;
use serde_json::json;
use tempfile::tempdir;
#[test]
fn legacy_change_request_body_uses_legacy_field_names() {
// `execute_change_remote` hits `POST /change`, which old
// `mutate`'s remote arm hits `POST /change`, which old
// `omnigraph-server` builds deserialize as `ChangeRequest` with
// **required** `query_source` and optional `query_name` keys.
// Newer servers accept both spellings via serde alias, but a
@ -96,120 +90,20 @@
}
#[test]
fn parse_env_assignment_supports_plain_and_exported_values() {
assert_eq!(
parse_env_assignment("DEMO_TOKEN=demo-token"),
Some(("DEMO_TOKEN".to_string(), "demo-token".to_string()))
);
assert_eq!(
parse_env_assignment("export DEMO_TOKEN=\"quoted-token\""),
Some(("DEMO_TOKEN".to_string(), "quoted-token".to_string()))
);
assert_eq!(parse_env_assignment("# comment"), None);
assert_eq!(parse_env_assignment(" "), None);
}
#[test]
fn bearer_token_from_env_file_reads_named_value() {
let temp = tempdir().unwrap();
let env_file = temp.path().join(".env.omni");
fs::write(
&env_file,
"FIRST=ignore\nexport DEMO_TOKEN=\" demo-token \"\n",
)
.unwrap();
assert_eq!(
bearer_token_from_env_file(&env_file, "DEMO_TOKEN")
.unwrap()
.as_deref(),
Some("demo-token")
);
assert_eq!(
bearer_token_from_env_file(&env_file, "MISSING").unwrap(),
None
);
}
#[test]
fn load_env_file_into_process_sets_missing_values_without_overriding_existing_ones() {
let temp = tempdir().unwrap();
let env_file = temp.path().join(".env.omni");
fs::write(
&env_file,
"AUTOLOAD_ONLY=from-file\nAUTOLOAD_PRESET=from-file\n",
)
.unwrap();
let missing_key = "AUTOLOAD_ONLY";
let preset_key = "AUTOLOAD_PRESET";
let previous_missing = std::env::var_os(missing_key);
let previous_preset = std::env::var_os(preset_key);
unsafe {
std::env::remove_var(missing_key);
std::env::set_var(preset_key, "from-env");
}
load_env_file_into_process(&env_file).unwrap();
assert_eq!(std::env::var(missing_key).unwrap(), "from-file");
assert_eq!(std::env::var(preset_key).unwrap(), "from-env");
unsafe {
if let Some(value) = previous_missing {
std::env::set_var(missing_key, value);
} else {
std::env::remove_var(missing_key);
}
if let Some(value) = previous_preset {
std::env::set_var(preset_key, value);
} else {
std::env::remove_var(preset_key);
}
}
}
#[test]
fn resolve_remote_bearer_token_uses_scoped_env_file_with_global_fallback() {
let temp = tempdir().unwrap();
fs::write(
temp.path().join("omnigraph.yaml"),
r#"
graphs:
demo:
uri: https://example.com
bearer_token_env: DEMO_TOKEN
auth:
env_file: .env.omni
cli:
graph: demo
"#,
)
.unwrap();
fs::write(
temp.path().join(".env.omni"),
"DEMO_TOKEN=scoped-token\nOMNIGRAPH_BEARER_TOKEN=global-token\n",
)
.unwrap();
fn resolve_remote_bearer_token_falls_back_to_default_env() {
// RFC-011: with no operator server matching the URL, the only chain
// left is the default `OMNIGRAPH_BEARER_TOKEN` env (no omnigraph.yaml
// scoped chain). Hermetic: no operator config is read for a literal URL
// that matches no `servers:` entry.
let previous = std::env::var_os(DEFAULT_BEARER_TOKEN_ENV);
let previous_home = std::env::var_os("OMNIGRAPH_HOME");
unsafe {
std::env::remove_var(DEFAULT_BEARER_TOKEN_ENV);
// Hermetic: the keyed hop (RFC-007 PR 2) must not pick up a real
// ~/.omnigraph on the developer's machine — and with no operator
// servers defined, the legacy chain below must behave
// byte-identically to pre-PR-2 (tested-as-untouched).
std::env::set_var("OMNIGRAPH_HOME", temp.path().join("no-operator-config"));
std::env::set_var(DEFAULT_BEARER_TOKEN_ENV, "global-token");
std::env::set_var("OMNIGRAPH_HOME", "/nonexistent/omnigraph-test-home");
}
let config_path = temp.path().join("omnigraph.yaml");
let config = load_config(Some(&config_path)).unwrap();
assert_eq!(
resolve_remote_bearer_token(&config, Some("https://override.example.com"))
resolve_remote_bearer_token(Some("https://override.example.com"))
.unwrap()
.as_deref(),
Some("global-token")
@ -228,196 +122,3 @@ cli:
}
}
}
#[test]
fn load_cli_config_autoloads_env_file_into_process() {
let temp = tempdir().unwrap();
fs::write(
temp.path().join("omnigraph.yaml"),
r#"
auth:
env_file: .env.omni
graphs:
demo:
uri: s3://bucket/prefix
"#,
)
.unwrap();
fs::write(
temp.path().join(".env.omni"),
"AUTOLOAD_FROM_CONFIG=loaded\n",
)
.unwrap();
let key = "AUTOLOAD_FROM_CONFIG";
let previous = std::env::var_os(key);
unsafe {
std::env::remove_var(key);
}
let config_path = temp.path().join("omnigraph.yaml");
let config = load_cli_config(Some(&config_path)).unwrap();
assert_eq!(
config.resolve_target_uri(None, Some("demo"), None).unwrap(),
"s3://bucket/prefix"
);
assert_eq!(std::env::var(key).unwrap(), "loaded");
unsafe {
if let Some(value) = previous {
std::env::set_var(key, value);
} else {
std::env::remove_var(key);
}
}
}
#[test]
fn graph_identity_resolve_policy_context_named_cli_graph_uses_graph_key_not_project_name_or_uri()
{
let temp = tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
project:
name: misleading-project
graphs:
local:
uri: /tmp/local-policy-graph.omni
policy:
file: ./policy.yaml
cli:
graph: local
"#,
)
.unwrap();
let config = load_config(Some(&config_path)).unwrap();
let context = resolve_policy_context(&config).unwrap();
assert_eq!(context.graph_id, "local");
}
#[test]
fn graph_identity_resolve_policy_context_server_graph_uses_graph_key_when_cli_graph_absent() {
let temp = tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
project:
name: misleading-project
graphs:
local:
uri: /tmp/local-policy-graph.omni
policy:
file: ./server-policy.yaml
server:
graph: local
"#,
)
.unwrap();
let config = load_config(Some(&config_path)).unwrap();
let context = resolve_policy_context(&config).unwrap();
assert_eq!(context.graph_id, "local");
assert!(context.policy_file.ends_with("server-policy.yaml"));
}
#[test]
fn graph_identity_resolve_policy_context_anonymous_uses_top_level_default_identity() {
let temp = tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
project:
name: misleading-project
graphs:
local:
uri: /tmp/local-policy-graph.omni
policy:
file: ./top-policy.yaml
"#,
)
.unwrap();
let config = load_config(Some(&config_path)).unwrap();
let context = resolve_policy_context(&config).unwrap();
assert_eq!(context.graph_id, "default");
assert!(context.policy_file.ends_with("top-policy.yaml"));
}
#[test]
fn graph_identity_resolve_cli_graph_named_target_uses_graph_key_not_project_name_or_uri() {
let temp = tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
project:
name: misleading-project
graphs:
prod:
uri: s3://bucket/prod-graph/
policy:
file: ./prod-policy.yaml
cli:
graph: prod
"#,
)
.unwrap();
let config = load_config(Some(&config_path)).unwrap();
// `--target` is removed; the `cli.graph` default drives the same
// graph-key (not project name / URI) selection.
let graph = resolve_cli_graph(&config, None).unwrap();
assert_eq!(graph.selected(), Some("prod"));
assert_eq!(graph.graph_id, "prod");
assert_eq!(graph.uri, "s3://bucket/prod-graph/");
}
#[test]
fn graph_identity_resolve_cli_graph_positional_uri_uses_anonymous_normalized_uri() {
let temp = tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
project:
name: misleading-project
graphs:
local:
uri: /tmp/configured-graph.omni
policy:
file: ./policy.yaml
cli:
graph: local
"#,
)
.unwrap();
let config = load_config(Some(&config_path)).unwrap();
let local_graph_path = temp.path().join("explicit-graph.omni");
let local_graph = resolve_cli_graph(
&config,
Some(format!("file://{}", local_graph_path.display())),
)
.unwrap();
assert_eq!(local_graph.selected(), None);
assert_eq!(
local_graph.graph_id,
local_graph_path.to_string_lossy().as_ref()
);
assert_eq!(local_graph.policy_file, None);
let s3_graph = resolve_cli_graph(
&config,
Some("s3://bucket/anonymous-graph/".to_string()),
)
.unwrap();
assert_eq!(s3_graph.selected(), None);
assert_eq!(s3_graph.graph_id, "s3://bucket/anonymous-graph");
assert_eq!(s3_graph.policy_file, None);
}

24
crates/omnigraph-cli/src/output.rs
View file

@ -734,15 +734,10 @@ pub(crate) fn print_snapshot_human(branch: &str, manifest_version: u64, entries:
pub(crate) fn print_read_output(
output: &ReadOutput,
format: ReadOutputFormat,
config: &OmnigraphConfig,
) -> Result<()> {
println!(
"{}",
render_read(
output,
format,
&resolve_table_render_options(config),
)?
render_read(output, format, &resolve_table_render_options())?
);
Ok(())
}
@ -892,20 +887,11 @@ pub(crate) fn finish_logout(
Ok(())
}
/// Table prefs cascade (RFC-007/008): legacy cli.table_* (window) >
/// operator defaults.table_* > built-in.
pub(crate) fn resolve_table_render_options(config: &OmnigraphConfig) -> ReadRenderOptions {
/// Table prefs cascade (RFC-011): operator defaults.table_* > built-in.
pub(crate) fn resolve_table_render_options() -> ReadRenderOptions {
let operator = crate::operator::load_operator_config().unwrap_or_default();
ReadRenderOptions {
max_column_width: config
.cli
.table_max_column_width
.or(operator.defaults.table_max_column_width)
.unwrap_or(80),
cell_layout: config
.cli
.table_cell_layout
.or(operator.defaults.table_cell_layout)
.unwrap_or_default(),
max_column_width: operator.defaults.table_max_column_width.unwrap_or(80),
cell_layout: operator.defaults.table_cell_layout.unwrap_or_default(),
}
}

37
crates/omnigraph-cli/src/planes.rs
View file

@ -82,9 +82,7 @@ impl Capability {
/// classifier) plus the one Data→Served refinement: `graphs` is remote-only.
///
/// This reflects *current enforced behavior*, so messages stay truthful:
/// `queries list` is `Local` (reads config today) and `queries validate` is
/// `Direct` (opens a graph directly today). Both converge to the RFC end-state
/// (served / control) only when later slices re-route them.
/// `queries`/`policy` read a cluster's applied state (`Control`).
pub(crate) fn command_capability(cmd: &Command) -> Capability {
if let Command::Graphs { .. } = cmd {
return Capability::Served;
@ -120,20 +118,18 @@ pub(crate) fn command_plane(cmd: &Command) -> Plane {
Command::Schema {
command: SchemaCommand::Plan { .. },
} => Plane::Storage,
Command::Queries {
command: QueriesCommand::Validate { .. },
} => Plane::Storage,
Command::Queries {
command: QueriesCommand::List { .. },
} => Plane::Session,
// `queries` and `policy` tooling now source their inputs from a
// cluster's applied state (`--cluster`), so they live on the control
// plane (RFC-011 — omnigraph.yaml excised from the CLI).
Command::Queries { .. } => Plane::Control,
Command::Policy { .. } => Plane::Control,
Command::Init { .. }
| Command::Optimize { .. }
| Command::Repair { .. }
| Command::Cleanup { .. }
| Command::Lint { .. } => Plane::Storage,
Command::Cluster { .. } => Plane::Control,
Command::Policy { .. }
| Command::Embed(_)
Command::Embed(_)
| Command::Login { .. }
| Command::Logout { .. }
| Command::Config { .. }
@ -188,7 +184,17 @@ pub(crate) fn command_label(cmd: &Command) -> &'static str {
pub(crate) fn accepts_cluster_addressing(cmd: &Command) -> bool {
matches!(
cmd,
Command::Optimize { .. } | Command::Repair { .. } | Command::Cleanup { .. }
Command::Optimize { .. }
| Command::Repair { .. }
| Command::Cleanup { .. }
// `lint` can type-check a `.gq` against a cluster graph's schema
// (RFC-011): `--cluster <dir> --graph <id>`.
| Command::Lint { .. }
// The policy/queries tooling addresses a cluster's applied state
// (RFC-011): `--cluster <dir>` selects the cluster, `--graph <id>`
// picks a graph's bundle/registry within it.
| Command::Policy { .. }
| Command::Queries { .. }
)
}
@ -284,7 +290,12 @@ mod tests {
assert_eq!(cap(&["omnigraph", "schema", "plan", "--schema", "s.pg", "graph.omni"]), Capability::Direct);
assert_eq!(cap(&["omnigraph", "cluster", "status", "--config", "."]), Capability::Control);
assert_eq!(cap(&["omnigraph", "version"]), Capability::Local);
assert_eq!(cap(&["omnigraph", "queries", "list"]), Capability::Local);
// `queries`/`policy` tooling reads cluster state now (control plane).
assert_eq!(cap(&["omnigraph", "queries", "list"]), Capability::Control);
assert_eq!(
cap(&["omnigraph", "policy", "validate"]),
Capability::Control
);
}
#[test]