feat(maintenance)!: add served optimization and interrupted-operation recovery

changelog.d/served-optimize-cli.added.md

@@ -0,0 +1,9 @@
+- `omnigraph optimize` supports registered server targets with an explicitly
+  selected graph. `--no-wait` returns after acceptance, `--operation-id` observes
+  an existing operation without resubmission, and `--timeout` bounds observation
+  without cancelling server work. The CLI retains the original operation ID
+  through uncertain delivery; direct-storage optimize keeps its existing output.
+  Administrators can reconcile an interrupted original ID with
+  `--writers-stopped` on an explicit direct-root target after writer and
+  accepted-I/O exclusion. This never runs another optimization or unlocks a
+  held owner.

changelog.d/served-optimize.breaking.md

@@ -0,0 +1,19 @@
+- `optimize` now checks the installed graph policy's explicit `optimize` action
+on target branch `main`, including direct engine and CLI access. Existing
+policies must grant this action before those actors can optimize. Graphs
+without an installed policy retain direct storage behavior.
+
+- Cluster servers can opt into bounded, authenticated optimization with durable
+original-ID receipts. It runs under the existing native owner, serializes with
+configuration deployment and preserves unresolved work without replay.
+Source inventory ceilings do not replace process CPU, memory or I/O limits.
+The complete result envelope is reserved before effects. Original-ID recovery
+records interrupted publication under stopped-writer exclusion without replay.
+Root-bound signed-token/OIDC servers refuse new optimization until resource
+admission is supported; direct HTTP cannot bypass that gate.
+
+- Vector index reconciliation now propagates statistics, storage and builder
+failures instead of reporting them as pending training data. An existing index
+is deferred only when it has no non-null vectors; Lance chooses the supported
+training layout for positive input.
+Served source preflight also refuses unsupported index formats before rewriting.

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

@@ -2013,6 +2013,10 @@ fn access_slot(action: omnigraph_policy::PolicyAction) -> Option<AccessSlot> {
         P::InvokeQuery => Some(AccessSlot::GraphWide),
         // Server- or operator-scoped: no graph's access report names them.
         P::Admin | P::GraphList | P::ConfigManage => None,
+        // This report has a closed wire vocabulary consumed by older clients.
+        // Maintenance status checks Optimize separately on main; do not add a
+        // new AccessAction without deliberately evolving the report contract.
+        P::Optimize => None,
     }
 }
 
@@ -2036,7 +2040,8 @@ impl From<AccessAction> for omnigraph_policy::PolicyAction {
 
 /// What the caller's own credential may do on one graph, by the graph's
 /// installed policy. It names actions, never the rules that allow them, and
-/// says nothing about any other actor.
+/// says nothing about any other actor. This closed vocabulary omits optimize;
+/// maintenance status checks that permission separately on main.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
 pub struct AccessOutput {
     /// The actor the credential resolves to. Absent only on a server that runs
@@ -3593,6 +3598,14 @@ mod tests {
             transitions,
             [AccessAction::BranchCreate, AccessAction::BranchMerge]
         );
+        assert_eq!(access_slot(omnigraph_policy::PolicyAction::Optimize), None);
+        assert!(
+            AccessAction::ALL
+                .into_iter()
+                .all(|action| omnigraph_policy::PolicyAction::from(action)
+                    != omnigraph_policy::PolicyAction::Optimize),
+            "the existing closed access vocabulary must stay readable by older clients"
+        );
     }
 
     #[test]

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

@@ -210,6 +210,39 @@ pub const HTTP_OPERATIONS: &[HttpOperation] = &[
         Surface::absent(DEPLOYMENT),
         Surface::bound(&["cluster plan"]),
     ),
+    operation(
+        "maintenance status",
+        HttpMethod::Get,
+        "/cluster/maintenance",
+        Capability::Read,
+        None,
+        Surface::absent("Maintenance is not an MCP operation."),
+        Surface::restricted(
+            &["optimize"],
+            "Read native admission and resource ceilings before submission.",
+        ),
+    ),
+    operation(
+        "optimize",
+        HttpMethod::Post,
+        "/cluster/maintenance",
+        Capability::Write,
+        None,
+        Surface::absent("Maintenance is not an MCP operation."),
+        Surface::bound(&["optimize"]),
+    ),
+    operation(
+        "maintenance lookup",
+        HttpMethod::Get,
+        "/cluster/maintenance/{operation_id}",
+        Capability::Read,
+        None,
+        Surface::absent("Maintenance is not an MCP operation."),
+        Surface::restricted(
+            &["optimize"],
+            "--operation-id observes the original invocation without execution.",
+        ),
+    ),
     operation(
         "list graphs",
         HttpMethod::Get,
@@ -621,7 +654,6 @@ pub const CLI_ONLY_OPERATIONS: &[CliOnlyOperation] = &[
     cli_only("schema upgrade-system-columns", STORAGE),
     cli_only("init", STORAGE),
     cli_only("upgrade", STORAGE),
-    cli_only("optimize", STORAGE),
     cli_only("rebuild-full-text-indexes", STORAGE),
     cli_only("repair", STORAGE),
     cli_only("cleanup", STORAGE),

crates/omnigraph-catalog/src/lib.rs

@@ -22,7 +22,7 @@ mod row;
 mod schema_publication;
 pub use schema_publication::{
     SchemaPublicationCandidate, SchemaPublicationEvidence, read_schema_publication_at,
-    read_schema_publication_candidate_at,
+    read_schema_publication_candidate_at, read_schema_publication_candidate_at_with_session,
 };
 
 use std::collections::{HashMap, HashSet};

crates/omnigraph-catalog/src/schema_publication.rs

@@ -43,6 +43,24 @@ pub async fn read_schema_publication_candidate_at(
     version: u64,
     original_id: Option<&str>,
     settlement_id: Option<&str>,
+) -> Result<Option<SchemaPublicationCandidate>> {
+    read_schema_publication_candidate_at_with_session(
+        root_uri,
+        version,
+        original_id,
+        settlement_id,
+        None,
+    )
+    .await
+}
+
+/// The same exact-version proof, retaining an already captured provider.
+pub async fn read_schema_publication_candidate_at_with_session(
+    root_uri: &str,
+    version: u64,
+    original_id: Option<&str>,
+    settlement_id: Option<&str>,
+    session: Option<&std::sync::Arc<lance::session::Session>>,
 ) -> Result<Option<SchemaPublicationCandidate>> {
     if version == 0
         || original_id
@@ -58,7 +76,7 @@ pub async fn read_schema_publication_candidate_at(
     let dataset = match open_dataset(
         &manifest_uri(root_uri),
         VersionResolution::At(version),
-        None,
+        session,
         crate::instrumentation::manifest_wrapper(),
     )
     .await

crates/omnigraph-cli/src/app.rs

@@ -201,6 +201,7 @@ fn finish(
     if error
         .downcast_ref::<cluster_remote::WaitTimeout>()
         .is_some()
+        || error.is::<commands::maintenance::remote::WaitTimeout>()
     {
         stdout.flush()?;
         return Ok(Completion::exit(5));

crates/omnigraph-cli/src/cli.rs

@@ -47,6 +47,7 @@ pub(crate) struct Cli {
     /// On a served write, and on every request `mcp` relays, it is an assertion:
     /// the request carries `omnigraph-if-actor`, and the server refuses it
     /// before anything runs when the bearer token resolves to another actor.
+    /// Served optimize derives its actor only from the credential and refuses --as.
     /// With a policy configured but no actor set, a direct operation is
     /// denied — see docs/user/operations/policy.md.
     #[arg(long = "as", global = true, value_name = "ACTOR", value_parser = crate::invocation::single_selector("--as"))]
@@ -145,7 +146,7 @@ pub(crate) enum Command {
     /// Convert a standalone graph's storage offline from format v8, v9 or v13 to
     /// the current format this binary serves (v16); other formats are rebuilt by export and load
     Upgrade(UpgradeArgs),
-    /// Compact small Lance fragments in every backing dataset of the graph
+    /// Compact small Lance fragments in every backing dataset of the graph on main
     Optimize(OptimizeArgs),
     /// Rebuild all full-text indexes from one branch's current rows
     ///

crates/omnigraph-cli/src/cluster_remote.rs

@@ -417,7 +417,7 @@ fn is_wait_timeout(error: &color_eyre::Report, deadline: Instant) -> bool {
     error.is::<tokio::time::error::Elapsed>() || Instant::now() >= deadline
 }
 
-fn retry_observation(error: &color_eyre::Report) -> bool {
+pub(crate) fn retry_observation(error: &color_eyre::Report) -> bool {
     error.is::<tokio::time::error::Elapsed>()
         || error.downcast_ref::<reqwest::Error>().is_some_and(|error| {
             error.is_timeout()

crates/omnigraph-cli/src/commands/maintenance/args.rs

@@ -1,4 +1,4 @@
-//! Arguments for direct graph maintenance.
+//! Arguments for graph maintenance.
 
 use clap::Args;
 
@@ -16,6 +16,19 @@ pub(crate) struct UpgradeArgs {
 
 #[derive(Debug, Args)]
 pub(crate) struct OptimizeArgs {
+    /// Return after the server accepts the operation; work continues remotely
+    #[arg(long)]
+    pub(crate) no_wait: bool,
+    /// Observe an existing operation; never submit another optimization
+    #[arg(long, value_name = "ID")]
+    pub(crate) operation_id: Option<String>,
+    /// Reconcile this original operation on an explicit direct root after
+    /// excluding prior writers and accepted graph/control I/O
+    #[arg(long, requires = "operation_id", conflicts_with_all = ["no_wait", "timeout"])]
+    pub(crate) writers_stopped: bool,
+    /// Maximum seconds to observe server work (default 300); never cancels work
+    #[arg(long, value_parser = clap::value_parser!(u64).range(1..=3600))]
+    pub(crate) timeout: Option<u64>,
     #[arg(long)]
     pub(crate) json: bool,
 }

crates/omnigraph-cli/src/commands/maintenance/mod.rs

@@ -1,4 +1,4 @@
-//! Direct graph maintenance through the existing admission and engine owners.
+//! Graph maintenance through the existing admission and engine owners.
 
 use std::io::Write;
 
@@ -10,6 +10,7 @@ use crate::helpers::{confirm_destructive, parse_duration_arg, resolve_actor};
 use crate::output::{write_cleanup, write_full_text_rebuild, write_optimize, write_repair};
 
 pub(crate) mod args;
+pub(crate) mod remote;
 pub(crate) use args::{
     CleanupArgs, OptimizeArgs, RebuildFullTextIndexesArgs, RepairArgs, UpgradeArgs,
 };
@@ -39,7 +40,16 @@ pub(crate) async fn run_optimize(
     stdout: &mut dyn Write,
     stderr: &mut dyn Write,
 ) -> Result<Completion> {
-    let OptimizeArgs { json } = args;
+    if args.writers_stopped {
+        return recover_optimize(args, scope, stdout, stderr).await;
+    }
+    if matches!(
+        scope.target.binding,
+        crate::targets::TargetBinding::Server { .. }
+    ) {
+        return remote::run(args, scope, stdout, stderr).await;
+    }
+    let OptimizeArgs { json, .. } = args;
     let uri = crate::scope::direct_graph_uri(scope.target, scope.graph).await?;
     let binding = crate::scope::capture_storage_binding(scope.target)?.for_uri(&uri)?;
     crate::admission::ensure_graph_with_binding(&binding, stderr).await?;
@@ -48,11 +58,116 @@ pub(crate) async fn run_optimize(
         omnigraph::storage::storage_for_binding(&binding)?,
     )
     .await?;
-    let stats = db.optimize().await?;
+    let actor = resolve_actor(scope.operator, scope.actor, stderr)?;
+    let stats = db.optimize_as(actor.as_deref()).await?;
     write_optimize(&uri, &stats, json, stdout)?;
     Ok(Completion::success())
 }
 
+async fn recover_optimize(
+    args: OptimizeArgs,
+    scope: MaintenanceScope<'_>,
+    stdout: &mut dyn Write,
+    stderr: &mut dyn Write,
+) -> Result<Completion> {
+    use color_eyre::eyre::{WrapErr, eyre};
+    use omnigraph_cluster::{
+        ClusterAdmissionPurpose, IdentityAuthorization, MaintenanceCompletion,
+        MaintenancePublication,
+    };
+    if !scope.target.is_explicit()
+        || !matches!(
+            scope.target.binding,
+            crate::targets::TargetBinding::DirectRoot { .. }
+        )
+    {
+        bail!("optimization recovery requires an explicit direct-root --target");
+    }
+    let id = args
+        .operation_id
+        .as_deref()
+        .ok_or_else(|| eyre!("recovery requires --operation-id"))?;
+    let graph = crate::scope::graph_selection(scope.target, scope.graph)?
+        .ok_or_else(|| eyre!("optimization recovery requires a graph"))?;
+    let actor = resolve_actor(scope.operator, scope.actor, stderr)?
+        .ok_or_else(|| eyre!("optimization recovery requires an explicit or configured actor"))?;
+    let diagnostic = |error: omnigraph_cluster::Diagnostic| {
+        eyre!("[{}] {}: {}", error.code, error.path, error.message)
+    };
+    let identity = IdentityAuthorization::authenticated(actor).map_err(diagnostic)?;
+    let binding = crate::scope::capture_storage_binding(scope.target)?;
+    let admission = omnigraph_cluster::acquire_cluster_admission_with_binding(
+        &binding,
+        ClusterAdmissionPurpose::MaintenanceRecovery {
+            operation_id: id.to_owned(),
+        },
+    )
+    .await
+    .map_err(diagnostic)?
+    .ok_or_else(|| eyre!("optimization recovery requires an initialized cluster"))?;
+    let custody = format!(
+        "Operation-ID: {id}; cluster-root: {}; lock-ID: {}",
+        omnigraph::storage::redacted_storage_uri(admission.canonical_root()),
+        admission.lock_id(),
+    );
+    writeln!(stderr, "{custody}")
+        .and_then(|()| stderr.flush())
+        .wrap_err(custody.clone())?;
+    let result = omnigraph_cluster::reconcile_maintenance(&admission, graph, id, &identity)
+        .await
+        .map_err(diagnostic)
+        .wrap_err(custody.clone())?;
+    let delivery = (|| -> Result<()> {
+        if args.json {
+            crate::output::write_json(
+                &serde_json::json!({
+                    "maintenance": {"status": "complete", "result": result},
+                    "in_progress": false,
+                }),
+                stdout,
+            )?;
+        } else {
+            writeln!(stdout, "Operation-ID: {id}")?;
+            match &result.completion {
+                MaintenanceCompletion::Success { .. } => writeln!(
+                    stdout,
+                    "Original optimization completed; exact result retained."
+                )?,
+                MaintenanceCompletion::Refused { code, message } => {
+                    writeln!(stdout, "Original optimization refused: {code}: {message}")?
+                }
+                MaintenanceCompletion::Interrupted {
+                    publication: MaintenancePublication::Unpublished,
+                } => writeln!(
+                    stdout,
+                    "Original optimization interrupted before graph publication; receipt reconciled."
+                )?,
+                MaintenanceCompletion::Interrupted {
+                    publication: MaintenancePublication::Published { report },
+                } => writeln!(
+                    stdout,
+                    "Original optimization published commit {}; later derived work is not proven complete. Receipt reconciled.",
+                    report.after_commit
+                )?,
+            }
+        }
+        stdout.flush()?;
+        Ok(())
+    })();
+    delivery.wrap_err(custody.clone())?;
+    // Release only after native reconciliation and exact-result delivery. The
+    // admission closes and settles its I/O scope; uncertainty retains the lock.
+    tokio::time::timeout(
+        std::time::Duration::from_secs(45),
+        admission.release_after_settlement(),
+    )
+    .await
+    .wrap_err(custody.clone())?
+    .map_err(diagnostic)
+    .wrap_err(custody)?;
+    Ok(Completion::success())
+}
+
 pub(crate) async fn run_full_text_rebuild(
     args: RebuildFullTextIndexesArgs,
     scope: MaintenanceScope<'_>,