Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 89 additions & 4 deletions crates/omnigraph-cli/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -21,22 +21,25 @@
//! `apply_schema` catalog-validator closure that is not object-safe.
//! Same one-body-two-impls collapse, less ceremony.

use std::io::Write;

use color_eyre::Result;
use color_eyre::eyre::bail;
use omnigraph::db::{Omnigraph, ReadTarget};
use omnigraph_api_types::{
BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, BranchListOutput,
BranchMergeOutput, BranchMergeRequest, ChangeOutput, CommitListOutput, CommitOutput,
IngestOutput, IngestRequest, ReadOutput, ReadRequest, SchemaApplyOutput, SchemaApplyRequest,
SchemaOutput, SnapshotOutput, commit_output, ingest_output, read_output, schema_apply_output,
snapshot_payload,
ErrorOutput, ExportRequest, GraphListResponse, IngestOutput, IngestRequest, ReadOutput,
ReadRequest, SchemaApplyOutput, SchemaApplyRequest, SchemaOutput, SnapshotOutput, commit_output,
ingest_output, read_output, schema_apply_output, snapshot_payload,
};
use omnigraph_compiler::catalog::Catalog;
use reqwest::Method;
use serde_json::Value;

use crate::cli::CliLoadMode;
use crate::helpers::{
ResolvedCliGraph, apply_server_flag, build_http_client, is_remote_uri,
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, remote_branch_url,
remote_json, remote_url, resolve_cli_actor, resolve_cli_graph, resolve_remote_bearer_token,
select_named_query,
Expand Down Expand Up @@ -615,4 +618,86 @@ impl GraphClient {
}
}
}

/// `export` — stream the branch as JSONL into `writer`. The streaming
/// shape (a `W: Write`, not a returned DTO) is why this lands in 3c
/// rather than 3b. Opens WITHOUT policy (like reads), so it is reached
/// via `resolve()`; the Embedded arm opens bare. The Remote arm streams
/// the chunked response body straight through (no buffering the whole
/// export in memory).
pub(crate) async fn export<W: Write>(
&self,
branch: &str,
type_names: &[String],
table_keys: &[String],
writer: &mut W,
) -> Result<()> {
match self {
GraphClient::Remote {
http,
base_url,
token,
} => {
let request = apply_bearer_token(
http.request(Method::POST, remote_url(base_url, "/export")),
token.as_deref(),
)
.json(&ExportRequest {
branch: Some(branch.to_string()),
type_names: type_names.to_vec(),
table_keys: table_keys.to_vec(),
});
let mut response = request.send().await?;
let status = response.status();
if !status.is_success() {
let text = response.text().await?;
if let Ok(error) = serde_json::from_str::<ErrorOutput>(&text) {
bail!(error.error);
}
bail!("server returned {}: {}", status, text);
}
while let Some(chunk) = response.chunk().await? {
writer.write_all(&chunk)?;
}
writer.flush()?;
Ok(())
}
GraphClient::Embedded { uri, .. } => {
let db = Omnigraph::open(uri).await?;
db.export_jsonl_to_writer(branch, type_names, table_keys, writer)
.await?;
writer.flush()?;
Ok(())
}
}
}

/// `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.
pub(crate) async fn list_graphs(&self) -> Result<GraphListResponse> {
match self {
GraphClient::Remote {
http,
base_url,
token,
} => {
remote_json(
http,
Method::GET,
remote_url(base_url, "/graphs"),
None,
token.as_deref(),
)
.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."
),
}
}
}
65 changes: 0 additions & 65 deletions crates/omnigraph-cli/src/helpers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -678,22 +678,6 @@ pub(crate) fn normalize_legacy_alias_uri(
}


pub(crate) fn inferred_config_path(uri: &str) -> Result<PathBuf> {
if uri.contains("://") {
return Ok(omnigraph_server::config::default_config_path());
}

let path = Path::new(uri);
let base = if path.is_absolute() {
path.parent()
.map(Path::to_path_buf)
.unwrap_or(std::env::current_dir()?)
} else {
std::env::current_dir()?.join(path.parent().unwrap_or_else(|| Path::new(".")))
};
Ok(base.join(omnigraph_server::config::DEFAULT_CONFIG_FILE))
}

pub(crate) fn read_target_from_cli(branch: Option<String>, snapshot: Option<String>) -> ReadTarget {
if let Some(snapshot) = snapshot {
ReadTarget::snapshot(SnapshotId::new(snapshot))
Expand Down Expand Up @@ -998,55 +982,6 @@ pub(crate) fn legacy_change_request_body(
body
}

pub(crate) async fn execute_export_to_writer<W: Write>(
uri: &str,
branch: &str,
type_names: &[String],
table_keys: &[String],
writer: &mut W,
) -> Result<()> {
let db = Omnigraph::open(uri).await?;
db.export_jsonl_to_writer(branch, type_names, table_keys, writer)
.await?;
writer.flush()?;
Ok(())
}

pub(crate) async fn execute_export_remote_to_writer<W: Write>(
client: &reqwest::Client,
uri: &str,
branch: &str,
type_names: &[String],
table_keys: &[String],
bearer_token: Option<&str>,
writer: &mut W,
) -> Result<()> {
let request = apply_bearer_token(
client.request(Method::POST, remote_url(uri, "/export")),
bearer_token,
)
.json(&ExportRequest {
branch: Some(branch.to_string()),
type_names: type_names.to_vec(),
table_keys: table_keys.to_vec(),
});
let mut response = request.send().await?;
let status = response.status();
if !status.is_success() {
let text = response.text().await?;
if let Ok(error) = serde_json::from_str::<ErrorOutput>(&text) {
bail!(error.error);
}
bail!("server returned {}: {}", status, text);
}

while let Some(chunk) = response.chunk().await? {
writer.write_all(&chunk)?;
}
writer.flush()?;
Ok(())
}

pub(crate) fn rewrite_deprecated_argv(args: Vec<OsString>) -> Vec<OsString> {
if args.len() >= 3 {
let sub = args[1].to_str();
Expand Down
60 changes: 19 additions & 41 deletions crates/omnigraph-cli/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@ use omnigraph_compiler::{
json_params_to_param_map, lint_query_file,
};
use omnigraph_api_types::{
ChangeOutput, CommitOutput, ErrorOutput, ExportRequest, GraphListResponse, IngestOutput,
ReadOutput, SchemaApplyOutput, SnapshotTableOutput,
ChangeOutput, CommitOutput, ErrorOutput, IngestOutput, ReadOutput, SchemaApplyOutput,
SnapshotTableOutput,
};
use omnigraph_server::queries::{QueryRegistry, check, format_check_breakages};
use omnigraph_server::{
Expand Down Expand Up @@ -525,33 +525,23 @@ async fn main() -> Result<()> {
table_keys,
} => {
let config = load_cli_config(config.as_ref())?;
let uri =
apply_server_flag(cli.server.as_deref(), cli.graph.as_deref(), uri, target.as_deref())?;
let bearer_token =
resolve_remote_bearer_token(&config, uri.as_deref(), target.as_deref())?;
let uri = resolve_uri(&config, uri, target.as_deref())?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
target.as_deref(),
)?;
let branch = resolve_branch(&config, branch, None, "main");
if jsonl {
eprintln!("warning: --jsonl is deprecated; `omnigraph export` always emits JSONL");
}

let stdout = io::stdout();
let mut stdout = stdout.lock();
if is_remote_uri(&uri) {
execute_export_remote_to_writer(
&http_client,
&uri,
&branch,
&type_names,
&table_keys,
bearer_token.as_deref(),
&mut stdout,
)
client
.export(&branch, &type_names, &table_keys, &mut stdout)
.await?;
} else {
execute_export_to_writer(&uri, &branch, &type_names, &table_keys, &mut stdout)
.await?;
}
}
Command::Query {
uri,
Expand Down Expand Up @@ -1047,26 +1037,14 @@ async fn main() -> Result<()> {
json,
} => {
let config = load_cli_config(config.as_ref())?;
let uri =
apply_server_flag(cli.server.as_deref(), cli.graph.as_deref(), uri, target.as_deref())?;
let bearer_token =
resolve_remote_bearer_token(&config, uri.as_deref(), target.as_deref())?;
let uri = resolve_uri(&config, uri, target.as_deref())?;
if !is_remote_uri(&uri) {
bail!(
"`omnigraph graphs list` requires a remote multi-graph server URL \
(http:// or https://). To enumerate local graphs, read `omnigraph.yaml` \
directly."
);
}
let payload = remote_json::<GraphListResponse>(
&http_client,
Method::GET,
remote_url(&uri, "/graphs"),
None,
bearer_token.as_deref(),
)
.await?;
let client = client::GraphClient::resolve(
&config,
cli.server.as_deref(),
cli.graph.as_deref(),
uri,
target.as_deref(),
)?;
let payload = client.list_graphs().await?;
if json {
print_json(&payload)?;
} else {
Expand Down
4 changes: 0 additions & 4 deletions crates/omnigraph-cli/src/output.rs
Original file line number Diff line number Diff line change
Expand Up @@ -812,10 +812,6 @@ pub(crate) fn print_policy_explain(decision: &PolicyDecision, actor_id: &str, re
println!("message: {}", decision.message);
}

pub(crate) fn yaml_string(value: &str) -> String {
format!("'{}'", value.replace('\'', "''"))
}

#[derive(serde::Serialize)]
pub(crate) struct QueriesIssue {
pub(crate) query: String,
Expand Down
29 changes: 29 additions & 0 deletions crates/omnigraph-cli/tests/parity_matrix.rs
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,35 @@ fn parity_load() {
assert_parity("load", &l, &r);
}

#[test]
fn parity_export() {
let p = parity();
let (l, r) = p.run(&["export"]);
// export emits a JSONL STREAM, not a single `--json` document, so the
// scrubbed-single-doc `assert_parity` doesn't apply — compare line-wise.
// The twin graphs are byte-copies of one loaded fixture, so rows carry
// identical ids/versions and need no scrubbing; sort the lines so any
// cross-arm row-ordering difference doesn't masquerade as a divergence.
assert_eq!(
l.status.code(),
r.status.code(),
"export: exit codes diverge\nlocal {l:?}\nremote {r:?}"
);
assert!(l.status.success(), "export local arm failed: {l:?}");
let mut local_lines: Vec<&str> = std::str::from_utf8(&l.stdout).unwrap().lines().collect();
let mut remote_lines: Vec<&str> = std::str::from_utf8(&r.stdout).unwrap().lines().collect();
assert!(
!local_lines.is_empty(),
"export produced no rows — the parity check would be vacuous"
);
local_lines.sort_unstable();
remote_lines.sort_unstable();
assert_eq!(
local_lines, remote_lines,
Comment on lines +199 to +206

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Missing non-empty guard on the remote output

assert!(!local_lines.is_empty()) guards against the fixture producing no rows on the local arm, but the remote arm has no equivalent check. If the remote arm silently returns a 200 with an empty body (e.g., a chunked-transfer bug where the first chunk() is None), remote_lines will be empty while local_lines is non-empty, and the test will still fail — but with the generic "JSONL streams diverge" message rather than a diagnostic one that points directly at the remote arm being empty. Adding a parallel assert for remote_lines mirrors the stated intent and makes the failure mode self-documenting.

Fix in Claude Code

"export: JSONL streams diverge (left=local, right=remote)"
);
}

// ---- error parity: exit codes must match for shared failure cases ----

#[test]
Expand Down
Loading