feat(streaming): complete graph-first firehose controls

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

@@ -5,8 +5,9 @@
 
 use omnigraph::db::{
     GraphCommit, GraphStreamDeclaration, GraphStreamDeclarationStatus,
-    GraphStreamDriverErrorStatus, GraphStreamDriverStatus, GraphStreamOperationalStatus,
-    GraphStreamPendingStatus, GraphStreamRebuildBlocker, GraphStreamRebuildStatus,
+    GraphStreamDriverErrorStatus, GraphStreamDriverStatus, GraphStreamEnsureIndicesResult,
+    GraphStreamOperationalStatus, GraphStreamOptimizeResult, GraphStreamPendingStatus,
+    GraphStreamRebuildBlocker, GraphStreamRebuildStatus, GraphStreamResumeResult,
     GraphStreamTokenCounts, MergeOutcome, ReadTarget, SchemaApplyResult, Snapshot,
 };
 use omnigraph::error::{MergeConflict, MergeConflictKind};
@@ -825,6 +826,61 @@ pub struct StreamStatusOutput {
     pub rebuild: StreamRebuildStatusOutput,
 }
 
+/// Aggregate result of reopening every sealed streaming declaration in a
+/// graph. Declaration, table, lane, dataset, and recovery identities are
+/// deliberately absent from this graph-level control-plane shape.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+pub struct StreamResumeOutput {
+    pub profile_revision: u64,
+    pub enrolled_declarations: u64,
+    pub resumed_declarations: u64,
+    pub already_open_declarations: u64,
+}
+
+/// Aggregate result of graph-wide checked index refresh. Any enrolled
+/// declaration changed by the operation is required to be sealed.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+pub struct StreamEnsureIndicesOutput {
+    pub changed: bool,
+    pub pending_index_count: u64,
+}
+
+/// Aggregate result of graph-wide checked stream optimization. Any enrolled
+/// declaration changed by the operation is required to be sealed. Physical
+/// fragment and dataset details stay inside the engine.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+pub struct StreamOptimizeOutput {
+    pub changed: bool,
+    pub pending_index_count: u64,
+    pub requires_repair: bool,
+}
+
+pub fn stream_resume_output(value: GraphStreamResumeResult) -> StreamResumeOutput {
+    StreamResumeOutput {
+        profile_revision: value.profile_revision,
+        enrolled_declarations: value.enrolled_declarations,
+        resumed_declarations: value.resumed_declarations,
+        already_open_declarations: value.already_open_declarations,
+    }
+}
+
+pub fn stream_ensure_indices_output(
+    value: GraphStreamEnsureIndicesResult,
+) -> StreamEnsureIndicesOutput {
+    StreamEnsureIndicesOutput {
+        changed: value.changed,
+        pending_index_count: value.pending_index_count,
+    }
+}
+
+pub fn stream_optimize_output(value: GraphStreamOptimizeResult) -> StreamOptimizeOutput {
+    StreamOptimizeOutput {
+        changed: value.changed,
+        pending_index_count: value.pending_index_count,
+        requires_repair: value.requires_repair,
+    }
+}
+
 fn stream_profile_mode_output(
     value: &str,
 ) -> std::result::Result<StreamProfileModeOutput, &'static str> {

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: stream ingest/status (graph scope) and graphs (registry scope).\n  \
+served — require a server: stream ingest/status/resume/maintenance (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 \
@@ -394,6 +394,34 @@ pub(crate) enum StreamCommand {
         #[arg(long)]
         json: bool,
     },
+    /// Reopen every sealed declaration in the served graph.
+    Resume {
+        /// Emit the graph-level result as JSON.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Run checked graph-wide maintenance; any enrolled declaration changed
+    /// by the operation must be sealed.
+    Maintenance {
+        #[command(subcommand)]
+        command: StreamMaintenanceCommand,
+    },
+}
+
+#[derive(Debug, Subcommand)]
+pub(crate) enum StreamMaintenanceCommand {
+    /// Reconcile declared indexes across the graph.
+    EnsureIndices {
+        /// Emit the graph-level result as JSON.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Compact the graph through one coordinated publish.
+    Optimize {
+        /// Emit the graph-level result as JSON.
+        #[arg(long)]
+        json: bool,
+    },
 }
 
 #[derive(Debug, Subcommand)]
@@ -566,8 +594,6 @@ pub(crate) enum StreamDeadLetterCommand {
 pub(crate) enum StreamBlockCommand {
     /// Revalidate and print one bounded page of correction evidence.
     Show {
-        /// Exact manifest table key, for example node:Person.
-        table_key: String,
         /// Cluster config directory containing cluster.yaml.
         #[arg(long, default_value = ".")]
         config: PathBuf,
@@ -586,8 +612,6 @@ pub(crate) enum StreamBlockCommand {
     },
     /// Apply one ordered REPLACE/WITHDRAW plan to the exact blocked cut.
     Correct {
-        /// Exact manifest table key, for example node:Person.
-        table_key: String,
         /// Cluster config directory containing cluster.yaml.
         #[arg(long, default_value = ".")]
         config: PathBuf,

crates/omnigraph-cli/src/client.rs

@@ -34,8 +34,8 @@ use omnigraph_api_types::{
     ErrorOutput, ExportRequest, GraphListResponse, IngestOutput, IngestRequest,
     InvokeStoredQueryRequest, ReadOutput,
     ReadRequest, SchemaApplyOutput, SchemaApplyRequest, SchemaOutput, SnapshotOutput,
-    StreamStatusOutput, commit_output, ingest_output, read_output, schema_apply_output,
-    snapshot_payload,
+    StreamEnsureIndicesOutput, StreamOptimizeOutput, StreamResumeOutput, StreamStatusOutput,
+    commit_output, ingest_output, read_output, schema_apply_output, snapshot_payload,
 };
 use omnigraph_compiler::catalog::Catalog;
 use reqwest::Method;
@@ -341,6 +341,18 @@ impl GraphClient {
             .await
     }
 
+    /// Resolve one graph-wide served stream control. The public surface has no
+    /// declaration/table selector and never accepts a client-supplied actor.
+    pub(crate) async fn resolve_stream_control(
+        command: &str,
+        server: Option<&str>,
+        graph: Option<&str>,
+        profile: Option<&str>,
+        store: Option<&str>,
+    ) -> Result<Self> {
+        Self::resolve_selected_served_graph(command, server, graph, None, profile, store).await
+    }
+
     /// Shared graph-selection owner for the served-only stream family.
     /// `command` is threaded only into the observable missing-graph error;
     /// keeping the ingest spelling here preserves its existing contract.
@@ -1071,6 +1083,79 @@ impl GraphClient {
         }
     }
 
+    /// Reopen every currently sealed declaration in the selected graph.
+    pub(crate) async fn stream_resume(&self) -> Result<StreamResumeOutput> {
+        match self {
+            GraphClient::Remote {
+                http,
+                base_url,
+                token,
+            } => {
+                remote_json(
+                    http,
+                    Method::POST,
+                    remote_url(base_url, &["stream", "resume"], &[])?,
+                    None,
+                    token.as_deref(),
+                )
+                .await
+            }
+            GraphClient::Embedded { .. } => bail!(
+                "internal error: `stream resume` reached an embedded client — stream controls \
+                 always resolve a server"
+            ),
+        }
+    }
+
+    /// Reconcile declared indexes across the graph. Any enrolled declaration
+    /// changed by the operation must be sealed; physical datasets stay private
+    /// behind the graph coordinator.
+    pub(crate) async fn stream_ensure_indices(&self) -> Result<StreamEnsureIndicesOutput> {
+        match self {
+            GraphClient::Remote {
+                http,
+                base_url,
+                token,
+            } => {
+                remote_json(
+                    http,
+                    Method::POST,
+                    remote_url(base_url, &["stream", "maintenance", "ensure-indices"], &[])?,
+                    None,
+                    token.as_deref(),
+                )
+                .await
+            }
+            GraphClient::Embedded { .. } => bail!(
+                "internal error: `stream maintenance ensure-indices` reached an embedded client"
+            ),
+        }
+    }
+
+    /// Compact the graph through the existing coordinated publish. Any
+    /// enrolled declaration changed by the operation must be sealed.
+    pub(crate) async fn stream_optimize(&self) -> Result<StreamOptimizeOutput> {
+        match self {
+            GraphClient::Remote {
+                http,
+                base_url,
+                token,
+            } => {
+                remote_json(
+                    http,
+                    Method::POST,
+                    remote_url(base_url, &["stream", "maintenance", "optimize"], &[])?,
+                    None,
+                    token.as_deref(),
+                )
+                .await
+            }
+            GraphClient::Embedded { .. } => bail!(
+                "internal error: `stream maintenance optimize` reached an embedded client"
+            ),
+        }
+    }
+
     /// `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
@@ -1516,6 +1601,69 @@ mod tests {
         server.await.unwrap();
     }
 
+    #[tokio::test]
+    async fn graph_stream_controls_post_bodyless_aggregate_requests() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base_url = format!("http://{}/graphs/knowledge", listener.local_addr().unwrap());
+        let server = tokio::spawn(async move {
+            for (path, response) in [
+                (
+                    "/graphs/knowledge/stream/resume",
+                    r#"{"profile_revision":4,"enrolled_declarations":3,"resumed_declarations":2,"already_open_declarations":1}"#,
+                ),
+                (
+                    "/graphs/knowledge/stream/maintenance/ensure-indices",
+                    r#"{"changed":true,"pending_index_count":0}"#,
+                ),
+                (
+                    "/graphs/knowledge/stream/maintenance/optimize",
+                    r#"{"changed":false,"pending_index_count":1,"requires_repair":false}"#,
+                ),
+            ] {
+                let (mut stream, _) = listener.accept().await.unwrap();
+                let request = read_http_request(&mut stream).await;
+                assert!(
+                    request
+                        .head
+                        .starts_with(&format!("POST {path} HTTP/1.1")),
+                    "{}",
+                    request.head
+                );
+                assert!(
+                    request
+                        .head
+                        .to_ascii_lowercase()
+                        .contains("authorization: bearer manage-token")
+                );
+                assert!(request.body.is_empty());
+                write_response(
+                    &mut stream,
+                    "200 OK",
+                    &[("Content-Type", "application/json")],
+                    response.as_bytes(),
+                )
+                .await;
+            }
+        });
+
+        let client = GraphClient::Remote {
+            http: reqwest::Client::new(),
+            base_url,
+            token: Some("manage-token".to_string()),
+        };
+        let resumed = client.stream_resume().await.unwrap();
+        assert_eq!(resumed.resumed_declarations, 2);
+        assert_eq!(resumed.already_open_declarations, 1);
+        let indexed = client.stream_ensure_indices().await.unwrap();
+        assert!(indexed.changed);
+        assert_eq!(indexed.pending_index_count, 0);
+        let optimized = client.stream_optimize().await.unwrap();
+        assert!(!optimized.changed);
+        assert_eq!(optimized.pending_index_count, 1);
+        assert!(!optimized.requires_repair);
+        server.await.unwrap();
+    }
+
     #[tokio::test]
     async fn stream_commands_require_and_resolve_a_selected_graph_without_network_io() {
         let error = match GraphClient::resolve_stream_ingest(
@@ -1564,5 +1712,22 @@ mod tests {
         .await
         .unwrap();
         assert_eq!(client.uri(), "http://127.0.0.1:9/graphs/knowledge");
+
+        let error = match GraphClient::resolve_stream_control(
+            "stream resume",
+            Some("http://127.0.0.1:9"),
+            None,
+            None,
+            None,
+        )
+        .await
+        {
+            Ok(_) => panic!("stream resume accepted a server without a selected graph"),
+            Err(error) => error.to_string(),
+        };
+        assert!(
+            error.contains("`stream resume` requires one selected graph"),
+            "{error}"
+        );
     }
 }