feat(cluster): storage root — ledger, catalog, and graphs on the StorageAdapter (RFC-006 PR 2/3)

crates/omnigraph-cli/src/main.rs

@@ -3720,7 +3720,7 @@ async fn main() -> Result<()> {
                 finish_cluster_approve(&output, json)?;
             }
             ClusterCommand::Status { config, json } => {
-                let output = status_config_dir(config);
+                let output = status_config_dir(config).await;
                 finish_cluster_status(&output, json)?;
             }
             ClusterCommand::Refresh { config, json } => {
@@ -3736,7 +3736,7 @@ async fn main() -> Result<()> {
                 config,
                 json,
             } => {
-                let output = force_unlock_config_dir(config, lock_id);
+                let output = force_unlock_config_dir(config, lock_id).await;
                 finish_cluster_force_unlock(&output, json)?;
             }
         },

crates/omnigraph-cluster/Cargo.toml

@@ -23,6 +23,10 @@ serde_yaml = { workspace = true }
 sha2 = { workspace = true }
 thiserror = { workspace = true }
 time = { workspace = true }
+# Runtime handle only — best-effort async lock release in
+# StateLockGuard::drop on object-store backends (cluster commands always
+# run inside the caller's tokio runtime).
+tokio = { workspace = true }
 ulid = { workspace = true }
 
 [dev-dependencies]

crates/omnigraph-cluster/src/config.rs

@@ -239,8 +239,33 @@ pub(crate) fn validate_cluster_header(
         }
     }
 
+    if let Some(storage) = raw.storage.as_deref() {
+        let trimmed = storage.trim();
+        if trimmed.is_empty() {
+            diagnostics.push(Diagnostic::error(
+                "invalid_storage_root",
+                "storage",
+                "storage must be a non-empty URI (e.g. s3://bucket/prefix) when provided",
+            ));
+        } else if let Some(rest) = trimmed.strip_prefix("s3://") {
+            if rest.trim_start_matches('/').is_empty() {
+                diagnostics.push(Diagnostic::error(
+                    "invalid_storage_root",
+                    "storage",
+                    "storage s3:// URI must name a bucket",
+                ));
+            }
+        }
+    }
+
     ClusterSettings {
         state_lock: raw.state.lock.unwrap_or(true),
+        storage_root: raw
+            .storage
+            .as_deref()
+            .map(str::trim)
+            .filter(|storage| !storage.is_empty())
+            .map(|storage| storage.trim_end_matches('/').to_string()),
     }
 }
 
@@ -271,19 +296,19 @@ pub(crate) fn initial_import_state(desired: &DesiredCluster) -> ClusterState {
 }
 
 
-pub(crate) async fn observe_declared_graphs(desired: &DesiredCluster, state: &mut ClusterState) -> usize {
+pub(crate) async fn observe_declared_graphs(
+    desired: &DesiredCluster,
+    backend: &ClusterStore,
+    state: &mut ClusterState,
+) -> usize {
     let mut graph_error_count = 0;
     for graph in &desired.graphs {
         let graph_address = graph_address(&graph.id);
         let schema_address = schema_address(&graph.id);
-        let graph_path = desired
-            .config_dir
-            .join(CLUSTER_GRAPHS_DIR)
-            .join(format!("{}.omni", graph.id));
-        let graph_uri = display_path(&graph_path);
+        let graph_uri = backend.graph_root(&graph.id);
         let observed_at = now_rfc3339();
 
-        if !graph_path.exists() {
+        if !backend.graph_root_exists(&graph_uri).await {
             state.applied_revision.resources.remove(&graph_address);
             state.applied_revision.resources.remove(&schema_address);
             state.observations.insert(
@@ -737,6 +762,7 @@ pub(crate) fn load_desired(config_dir: &Path) -> LoadOutcome {
         desired: Some(DesiredCluster {
             config_dir: config_dir.clone(),
             config_digest,
+            storage_root: settings.storage_root.clone(),
             state_lock: settings.state_lock,
             graphs,
             resource_digests,

crates/omnigraph-cluster/src/diff.rs

@@ -142,7 +142,7 @@ pub(crate) fn compute_approvals(
 /// Near-misses — an artifact for the same resource whose bound digests no
 /// longer match — warn as `approval_stale` and never authorize anything.
 pub(crate) fn approved_resources(
-    artifacts: &[(PathBuf, ApprovalArtifact)],
+    artifacts: &[(String, ApprovalArtifact)],
     changes: &[PlanChange],
     config_digest: &str,
     diagnostics: &mut Vec<Diagnostic>,