feat(cli): run optimize through served maintenance on a server target

changelog.d/ensure-indices-refreshes-coverage.changed.md

@@ -1,3 +1,3 @@
-- `ensure_indices` and `ensure_indices_on` also rebuild lagging BTREE/vector
-  indexes, report lagging full-text coverage as pending, and under a policy
-  need an actor with `Change`: use `ensure_indices_on_as`.
+- The engine's index build (`ensure_indices_as`, `ensure_indices_on_as`) also
+  rebuilds lagging BTREE and vector indexes and reports lagging full-text
+  coverage as pending.

changelog.d/full-text-rebuild-change-grant.deprecated.md

@@ -0,0 +1,5 @@
+- The full-text rebuild accepts `change` on its branch for one more release;
+  after that it needs `maintain`. Grant `maintain` to the accounts that rebuild
+  full-text indexes. See [rebuild full-text indexes][full-text-rebuild-change-grant].
+
+[full-text-rebuild-change-grant]: ../docs/user/operations/maintenance.md#rebuild-full-text-indexes

changelog.d/maintenance-entry-points-take-an-actor.breaking.md

@@ -0,0 +1,10 @@
+- An embedded host with a policy checker refuses maintenance until its policy
+  grants `maintain` or `reclaim`. Engine maintenance methods take an actor:
+  `optimize_as`, `cleanup_as`, `ensure_indices_as`, `ensure_indices_on_as` and
+  `rebuild_full_text_indices_on_as`. `repair` is the read-only
+  `repair_report_as`, without `RepairOptions`, `healed`, `forced` or
+  `graph_manifest_version`. See [policy actions][maintenance-actions-upgrade]
+  and [repair][maintenance-entry-points-repair].
+
+[maintenance-actions-upgrade]: ../docs/user/operations/policy.md#actions
+[maintenance-entry-points-repair]: ../docs/user/operations/maintenance.md#repair

changelog.d/maintenance-policy-actions.added.md

@@ -0,0 +1,5 @@
+- Cedar policy has two maintenance actions: `maintain` for optimize, index
+  builds, the full-text rebuild and repair, and `reclaim` for cleanup. `change`
+  implies neither. See [policy actions][maintenance-policy-actions].
+
+[maintenance-policy-actions]: ../docs/user/operations/policy.md#actions

changelog.d/optimize-artifact-size-limit.added.md

@@ -0,0 +1,3 @@
+- Embedded hosts can bound the adjacency artifact optimize writes with
+  `OptimizeLimits::artifact_bytes`; a larger one is skipped, before or during
+  its edge scan, and nothing is written.

changelog.d/optimize-before-publish-hook.added.md

@@ -0,0 +1,3 @@
+- Embedded hosts can pass `OptimizeRequest::with_before_publish`: it runs once
+  before an optimize run publishes, with the commit's nonce and input head, and
+  its failure publishes nothing.

changelog.d/optimize-cancellation.added.md

@@ -0,0 +1,2 @@
+- Embedded hosts can cancel an optimize run with `OptimizeRequest::with_cancellation`:
+  it stops before the next table or the publication and publishes nothing.

changelog.d/optimize-gate-hold-bound.added.md

@@ -0,0 +1,3 @@
+- `optimize --max-gate-hold` bounds how long a run holds main's writer gates and names the table the next run resumes at; `--tables` runs only the listed tables. See [bounding a run][optimize-bound].
+
+[optimize-bound]: ../docs/user/operations/maintenance.md#bounding-a-run

changelog.d/optimize-report-outcomes.changed.md

@@ -0,0 +1,3 @@
+- `optimize` reports one outcome per table, both graph commits and the gate hold; a failed configuration clear or adjacency artifact no longer fails a published run. See [bounding a run][optimize-report].
+
+[optimize-report]: ../docs/user/operations/maintenance.md#bounding-a-run

changelog.d/optimize-resource-ceilings.added.md

@@ -0,0 +1,3 @@
+- Embedded hosts can bound optimize per table (`TableCeilings`) and per run
+  (`run_source_bytes`); a table over a ceiling is deferred before its first
+  write, with nothing written for it.

changelog.d/optimize-table-failure-isolation.changed.md

@@ -0,0 +1,2 @@
+- A table whose own optimize work fails is reported `failed` with its error, and
+  the other tables still publish; before, one table's failure failed the run.

changelog.d/optimize-takes-a-request.breaking.md

@@ -0,0 +1,8 @@
+- Embedded hosts: `optimize_as` takes an `OptimizeRequest` (`OptimizeRequest::default()`
+  is a full run) and returns an `OptimizeReport` whose `tables` holds the per-table
+  stats; `committed` and `skipped` are methods. `optimize --json` reports the
+  `__manifest` row as `configuration_clear` and drops the always-null
+  `published_dataset_version` and `lance_head_version`. See
+  [bounding a run][optimize-request-upgrade].
+
+[optimize-request-upgrade]: ../docs/user/operations/maintenance.md#bounding-a-run

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

@@ -0,0 +1,5 @@
+- `omnigraph optimize` against a server target runs served maintenance: it
+  prints the operation id, waits for the record, and never resubmits after a
+  lost response. See [served maintenance][served-optimize-cli].
+
+[served-optimize-cli]: ../docs/user/operations/served-maintenance.md#from-the-cli

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

@@ -0,0 +1,5 @@
+- A cluster server can run optimize on a graph's `main` when
+  `OMNIGRAPH_MAINTENANCE_ROUTES=true`, keeping a durable record a caller looks
+  up by its own id. See [served maintenance][served-maintenance].
+
+[served-maintenance]: ../docs/user/operations/served-maintenance.md

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

@@ -2058,6 +2058,9 @@ 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,
+        // Maintenance runs only through the direct CLI and embedded callers;
+        // no served operation performs it, so the report does not name it.
+        P::Maintain | P::Reclaim => None,
     }
 }
 

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

@@ -145,6 +145,10 @@ const PROBE: &str = "Process probe, not a graph command.";
 const OAUTH: &str = "Protocol-defined OAuth discovery, not a graph operation.";
 const DEPLOYMENT: &str = "Deployment control is not an MCP agent tool.";
 const BYTE_STREAM: &str = "MCP tools do not carry this byte-stream contract.";
+const MAINTENANCE: &str =
+    "Served maintenance runs under a durable operation record, not as an MCP agent tool.";
+const MAINTENANCE_CANCEL_CLI: &str =
+    "The CLI does not cancel a served maintenance operation; a scheduler or an HTTP client does.";
 
 /// Every operation in the generated HTTP specification. Protocol endpoints
 /// (`/mcp`, `/openapi.json`) and the catalogue's reserved-id dispatch retain
@@ -410,6 +414,36 @@ pub const HTTP_OPERATIONS: &[HttpOperation] = &[
         Surface::bound(&["commits_publish"]),
         Surface::bound(&["commit publish"]),
     ),
+    operation(
+        "start optimize",
+        HttpMethod::Post,
+        "/graphs/{graph_id}/maintenance/optimize",
+        Capability::Write,
+        None,
+        Surface::absent(MAINTENANCE),
+        Surface::bound(&["optimize"]),
+    ),
+    operation(
+        "get maintenance operation",
+        HttpMethod::Get,
+        "/graphs/{graph_id}/maintenance/{maintenance_id}",
+        Capability::Read,
+        None,
+        Surface::absent(MAINTENANCE),
+        Surface::restricted(
+            &["optimize"],
+            "--operation observes an existing operation, and the wait polls it.",
+        ),
+    ),
+    operation(
+        "cancel maintenance operation",
+        HttpMethod::Post,
+        "/graphs/{graph_id}/maintenance/{maintenance_id}/cancel",
+        Capability::Write,
+        None,
+        Surface::absent(MAINTENANCE),
+        Surface::absent(MAINTENANCE_CANCEL_CLI),
+    ),
     operation(
         "diff",
         HttpMethod::Get,
@@ -621,7 +655,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("index build", STORAGE),
     cli_only("index rebuild", STORAGE),
     cli_only("repair", STORAGE),

crates/omnigraph-bench/src/dataset_worker.rs

@@ -803,7 +803,10 @@ pub(crate) async fn build_dataset(
                 Engine::V2,
             )?;
             if plan.needs_indices {
-                session.ensure_indices().await.map_err(|e| e.to_string())?;
+                session
+                    .ensure_indices_as(None)
+                    .await
+                    .map_err(|e| e.to_string())?;
             }
             (session, case, Some(exact))
         }

crates/omnigraph-bench/src/legacy/branch_merge_oracle.rs

@@ -28,32 +28,33 @@ pub async fn initialize_local_fixture(
     let optimized_user_tables = match plan.compaction_recency {
         CompactionRecency::Optimized => {
             let outcomes = db
-                .optimize()
+                .optimize_as(None, omnigraph::db::OptimizeRequest::default())
                 .await
                 .map_err(|error| fixture_error(format!("optimize fixture main branch: {error}")))?;
             let intended_keys = (0..plan.node_tables())
                 .map(node_table_key)
                 .chain((0..plan.edge_tables()).map(edge_table_key));
             for key in intended_keys {
                 let outcome = outcomes
+                    .tables
                     .iter()
                     .find(|outcome| outcome.type_key == key)
                     .ok_or_else(|| {
                         fixture_error(format!(
                             "optimized fixture returned no outcome for intended user table {key}"
                         ))
                     })?;
-                if outcome.skipped.is_some()
-                    || !outcome.committed
+                if outcome.skipped().is_some()
+                    || !outcome.committed()
                     || outcome.fragments_removed == 0
                     || outcome.fragments_added == 0
                 {
                     return Err(fixture_error(format!(
                         "optimized fixture did not productively compact {key}: committed={}, fragments_removed={}, fragments_added={}, skipped={:?}",
-                        outcome.committed,
+                        outcome.committed(),
                         outcome.fragments_removed,
                         outcome.fragments_added,
-                        outcome.skipped
+                        outcome.skipped()
                     )));
                 }
             }

crates/omnigraph-cli/src/client.rs

@@ -2061,6 +2061,54 @@ impl GraphClient {
     }
 
     /// Create `name` at `start`: a branch's head, or a published graph commit.
+    /// Submit a served optimize; the record, or the server's refusal.
+    pub(crate) async fn start_optimize(&self, body: Value) -> Result<Value> {
+        let GraphClient::Remote {
+            http,
+            base_url,
+            token,
+            response_limit,
+            ..
+        } = self
+        else {
+            bail!("served maintenance needs a server target");
+        };
+        remote_json_bounded(
+            http,
+            Method::POST,
+            remote_url(base_url, &["maintenance", "optimize"], &[])?,
+            Some(body),
+            token.as_deref(),
+            None,
+            *response_limit,
+        )
+        .await
+    }
+
+    /// Look a served maintenance operation up by its id.
+    pub(crate) async fn maintenance_operation(&self, maintenance_id: &str) -> Result<Value> {
+        let GraphClient::Remote {
+            http,
+            base_url,
+            token,
+            response_limit,
+            ..
+        } = self
+        else {
+            bail!("served maintenance needs a server target");
+        };
+        remote_json_bounded(
+            http,
+            Method::GET,
+            remote_url(base_url, &["maintenance", maintenance_id], &[])?,
+            None,
+            token.as_deref(),
+            None,
+            *response_limit,
+        )
+        .await
+    }
+
     pub(crate) async fn branch_create(
         &self,
         start: ReadTarget,

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

@@ -60,6 +60,27 @@ pub(crate) struct UpgradeArgs {
 
 #[derive(Debug, Args)]
 pub(crate) struct OptimizeArgs {
+    /// Only these tables, in this order: `node:<Type>` or `edge:<Type>`,
+    /// comma-separated or repeated. Default: every table on main.
+    #[arg(long, value_delimiter = ',')]
+    pub(crate) tables: Vec<String>,
+    /// Stop starting tables once the run has held main's writer gates this
+    /// long, and report the rest deferred. The first table always runs.
+    /// Accepts Go-style durations: `30s`, `5m`.
+    #[arg(long)]
+    pub(crate) max_gate_hold: Option<String>,
+    /// Server targets: return once the server has accepted the run, without
+    /// waiting for its terminal record.
+    #[arg(long)]
+    pub(crate) no_wait: bool,
+    /// Server targets: observe the operation with this id instead of
+    /// submitting a new one.
+    #[arg(long, value_name = "ID")]
+    pub(crate) operation: Option<String>,
+    /// Server targets: stop waiting after this long (`10m`); the operation
+    /// keeps running and `--operation` observes it.
+    #[arg(long)]
+    pub(crate) timeout: Option<String>,
     #[arg(long)]
     pub(crate) json: bool,
 }

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

@@ -3,7 +3,7 @@
 use std::io::Write;
 
 use color_eyre::eyre::{Result, bail};
-use omnigraph::db::{CleanupPolicyOptions, Omnigraph, RepairAction, RepairOptions};
+use omnigraph::db::{CleanupPolicyOptions, Omnigraph, RepairAction};
 
 use crate::app::Completion;
 use crate::helpers::{confirm_destructive, parse_duration_arg, resolve_actor};
@@ -12,6 +12,7 @@ use crate::output::{
 };
 
 pub(crate) mod args;
+mod served;
 pub(crate) use args::{
     CleanupArgs, IndexBuildArgs, IndexCommand, IndexRebuildArgs, IndexRebuildKind, OptimizeArgs,
     RebuildFullTextIndexesArgs, RepairArgs, UpgradeArgs,
@@ -42,17 +43,68 @@ pub(crate) async fn run_optimize(
     stdout: &mut dyn Write,
     stderr: &mut dyn Write,
 ) -> Result<Completion> {
-    let OptimizeArgs { json } = args;
+    let OptimizeArgs {
+        tables,
+        max_gate_hold,
+        no_wait,
+        operation,
+        timeout,
+        json,
+    } = args;
+    let max_gate_hold = max_gate_hold
+        .as_deref()
+        .map(parse_duration_arg)
+        .transpose()?;
+    if matches!(
+        scope.target.binding,
+        crate::targets::TargetBinding::Server { .. } | crate::targets::TargetBinding::Managed(_)
+    ) {
+        let client = crate::client::GraphClient::from_target(
+            scope.operator,
+            scope.target,
+            scope.graph,
+            None,
+            stderr,
+        )
+        .await?;
+        let timeout = timeout.as_deref().map(parse_duration_arg).transpose()?;
+        return served::run(
+            &client,
+            served::ServedOptimize {
+                tables,
+                max_gate_hold,
+                no_wait,
+                operation,
+                timeout,
+                json,
+            },
+            stdout,
+            stderr,
+        )
+        .await;
+    }
+    if no_wait || operation.is_some() || timeout.is_some() {
+        bail!("--no-wait, --operation and --timeout apply to a server target only");
+    }
+    let mut request = omnigraph::db::OptimizeRequest::default();
+    if !tables.is_empty() {
+        request = request.with_tables(tables);
+    }
+    if let Some(hold) = max_gate_hold {
+        request =
+            request.with_limits(omnigraph::db::OptimizeLimits::default().with_max_gate_hold(hold));
+    }
     let uri = crate::scope::direct_graph_uri(scope.target, scope.graph).await?;
     let binding = crate::scope::capture_storage_binding(scope.target)?.for_uri(&uri)?;
+    let actor = resolve_actor(scope.operator, scope.actor, stderr)?;
     crate::admission::ensure_graph_with_binding(&binding, stderr).await?;
     let db = Omnigraph::open_with_storage(
         binding.uri(),
         omnigraph::storage::storage_for_binding(&binding)?,
     )
     .await?;
-    let stats = db.optimize().await?;
-    write_optimize(&uri, &stats, json, stdout)?;
+    let report = db.optimize_as(actor.as_deref(), request).await?;
+    write_optimize(&uri, &report, json, stdout)?;
     Ok(Completion::success())
 }
 
@@ -141,13 +193,14 @@ pub(crate) async fn run_repair(
     let RepairArgs { 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)?;
+    let actor = resolve_actor(scope.operator, scope.actor, stderr)?;
     crate::admission::ensure_graph_with_binding(&binding, stderr).await?;
     let db = Omnigraph::open_with_storage(
         binding.uri(),
         omnigraph::storage::storage_for_binding(&binding)?,
     )
     .await?;
-    let stats = db.repair(RepairOptions::default()).await?;
+    let stats = db.repair_report_as(actor.as_deref()).await?;
     let refused_count = stats
         .datasets
         .iter()
@@ -207,13 +260,14 @@ pub(crate) async fn run_cleanup(
         keep_versions: keep,
         older_than: older_than_dur,
     };
+    let actor = resolve_actor(scope.operator, scope.actor, stderr)?;
     crate::admission::ensure_graph_with_binding(&binding, stderr).await?;
     let db = Omnigraph::open_with_storage(
         binding.uri(),
         omnigraph::storage::storage_for_binding(&binding)?,
     )
     .await?;
-    let stats = db.cleanup(options.clone()).await?;
+    let stats = db.cleanup_as(options.clone(), actor.as_deref()).await?;
     write_cleanup(&uri, &options, &policy_desc, &stats, json, stdout)?;
     Ok(Completion::success())
 }