Activate served graph firehose ingest

.github/workflows/ci.yml

@@ -395,6 +395,8 @@ jobs:
 
       - name: Test omnigraph-server with aws feature
         if: needs.classify_changes.outputs.run_full_ci == 'true'
+        env:
+          RUST_MIN_STACK: 16777216
         run: cargo test --locked -p omnigraph-server --features aws
 
   rustfs_integration:

crates/omnigraph-api-types/src/lib.rs

@@ -543,6 +543,96 @@ pub struct ExportRequest {
     pub table_keys: Vec<String>,
 }
 
+/// Effect-free precondition challenge for graph-native streaming ingest.
+///
+/// The same value is returned as a strong `ETag` response header. Clients
+/// retry the request with that tag in `If-Match`; this convenience copy keeps
+/// the graph authority token distinct from the per-row sequencing
+/// [`StreamIngestLineOutput::stream_token`].
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct StreamIngestChallenge {
+    pub graph_token: String,
+}
+
+/// Logical declaration kind selected by one graph-native stream row.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum StreamIngestKindOutput {
+    Node,
+    Edge,
+}
+
+/// Whether one stream result applies to a single row or blocks the graph-wide
+/// remainder of the request.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum StreamIngestScopeOutput {
+    Row,
+    Graph,
+}
+
+/// Stable, graph-logical status vocabulary for one streaming-ingest line.
+///
+/// Several private physical-authority transitions intentionally collapse to
+/// `stream_authority_changed`; the transport never exposes lane, binding,
+/// shard, epoch, generation, dataset, or recovery-sidecar identity.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum StreamIngestStatusOutput {
+    Durable,
+    AckUnknown,
+    AlreadyDurable,
+    Withdrawn,
+    DeadLettered,
+    Invalid,
+    StreamInputTooLarge,
+    StreamAuthorityChanged,
+    StreamSequenceConflict,
+    StreamIdempotencyConflict,
+    StreamFoldRequired,
+    StreamBackpressure,
+    RecoveryRequired,
+    StreamRetryRequired,
+}
+
+/// One ordered, newline-delimited result from graph-native streaming ingest.
+///
+/// Every field is caller-logical or directly actionable retry evidence.
+/// Physical table and MemWAL identities are deliberately absent.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct StreamIngestLineOutput {
+    pub ordinal: u64,
+    pub status: StreamIngestStatusOutput,
+    pub scope: StreamIngestScopeOutput,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub kind: Option<StreamIngestKindOutput>,
+    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
+    pub type_name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub id: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub write_id: Option<String>,
+    /// Confirmed per-row sequencing token. This is not the graph-level ETag.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub stream_token: Option<String>,
+    /// Candidate token whose durability acknowledgement is unknown.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub unconfirmed_candidate_token: Option<String>,
+    /// Current per-row token returned with a sequencing or terminal outcome.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub current_token: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub message: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub limit: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub actual: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub blocking_ordinal: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub blocking_status: Option<StreamIngestStatusOutput>,
+}
+
 #[derive(Debug, Clone, Deserialize, IntoParams)]
 pub struct SnapshotQuery {
     pub branch: Option<String>,

crates/omnigraph-cli/Cargo.toml

@@ -24,8 +24,9 @@ color-eyre = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }
-tokio = { workspace = true }
-reqwest = { workspace = true, features = ["blocking"] }
+tokio = { workspace = true, features = ["fs", "io-std", "io-util"] }
+reqwest = { workspace = true, features = ["blocking", "stream"] }
+futures = { workspace = true }
 
 [dev-dependencies]
 assert_cmd = "2"

crates/omnigraph-cli/src/cli.rs

@@ -16,7 +16,7 @@ pub(crate) const DEFAULT_BEARER_TOKEN_ENV: &str = "OMNIGRAPH_BEARER_TOKEN";
 COMMANDS BY CAPABILITY:\n  \
 any — run against a graph, served (--server / --profile) or embedded (--store / a \
 URI): query, mutate, load, branch, snapshot, export, commit, schema show/apply.\n  \
-served — require a server (registry scope, --server/--profile only): graphs.\n  \
+served — require a server: stream ingest (graph scope) and graphs (registry scope).\n  \
 direct — direct storage access; reject --server (init, optimize, repair, cleanup, \
 schema plan, lint).\n  \
 control — manage or inspect a cluster (cluster via --config; policy & queries via \
@@ -247,6 +247,11 @@ pub(crate) enum Command {
         #[command(subcommand)]
         command: GraphsCommand,
     },
+    /// Stream graph rows durably through a served firehose.
+    Stream {
+        #[command(subcommand)]
+        command: StreamCommand,
+    },
 
     // ── Storage / local graph ops ── direct storage or local files; reject --server.
     /// Initialize a new graph from a schema
@@ -371,6 +376,20 @@ pub(crate) enum Command {
     Version,
 }
 
+#[derive(Debug, Subcommand)]
+pub(crate) enum StreamCommand {
+    /// Stream newline-delimited graph rows to the served graph.
+    Ingest {
+        /// NDJSON input path, or `-` for stdin.
+        #[arg(long, value_name = "PATH|-", default_value = "-")]
+        data: PathBuf,
+        /// Opaque graph-ingest eligibility token. Omit it to obtain one with a
+        /// bodyless preflight before the input is opened.
+        #[arg(long, value_name = "TOKEN")]
+        graph_token: Option<String>,
+    },
+}
+
 #[derive(Debug, Subcommand)]
 pub(crate) enum ProfileCommand {
     /// List the profiles defined in ~/.omnigraph/config.yaml.