feat(cli)!: replace legacy routing with explicit targets

README.md

@@ -149,8 +149,8 @@ idempotent; re-running is always safe.
 
 ```bash
 omnigraph cluster validate   # parse + typecheck everything
-omnigraph cluster plan       # preview what apply would do
-omnigraph cluster apply      # converge
+omnigraph cluster plan --target ./company-brain    # preview
+omnigraph cluster apply --target ./company-brain   # converge
 
 # Boot the server from the cluster dir; storage resolves through cluster.yaml
 omnigraph-server --cluster company-brain --bind 0.0.0.0:8080
@@ -164,7 +164,8 @@ containers, AWS/Railway/Azure, auth, and the object-store environment contracts.
 
 ## Query and mutate
 
-Set a default server and graph once in `~/.omnigraph/config.yaml`, and the
+Define a server target with `default_graph` and set `defaults.target` once in
+`~/.omnigraph/config.yaml` ([configuration](docs/user/cli/index.md)), and the
 everyday commands stay short. Stored queries and mutations run **by name**:
 
 ```bash
@@ -176,10 +177,10 @@ omnigraph branch create --from main agent/ingest-42
 omnigraph branch merge  agent/ingest-42 --into main
 ```
 
-An **alias** is shorter still: bind a server, graph, and stored query to one
+An **alias** is shorter still: bind a target, graph, and stored query to one
 name, then `omnigraph alias triage` runs it. For an ad-hoc target, any command
-still takes `--server <name|url> --graph <id>` (or `--store <uri>` for a local
-graph). See the [CLI reference](docs/user/cli/reference.md).
+takes `--target NAME|ADDRESS`, plus `--graph ID` for a graph within a server
+or direct root. Use `--target ./graph.omni` for a standalone graph. See the [CLI reference](docs/user/cli/reference.md).
 
 ## Security & governance
 

changelog.d/cli-target-clean-slate.breaking.md

@@ -0,0 +1,3 @@
+- The CLI now selects one complete backend with `--target NAME|ADDRESS` or `defaults.target`. Legacy routing flags, profiles, folder selection and `use` are removed; `--config` supplies source files only, and resolved targets and known write scopes are always printed to stderr. Cloud bindings preserve native authentication while pinning destination, signing and conditional-write settings; unsupported ambient transport overrides refuse. Run `omnigraph target import` for legacy operator configuration and lifecycle custody, then review the complete target and endpoint-bound credentials before use. See the [CLI reference][target-cli].
+
+[target-cli]: ../docs/user/cli/reference.md

crates/omnigraph-catalog/src/graph.rs

@@ -188,7 +188,10 @@ pub(crate) async fn init_manifest_graph(
     let manifest_path = manifest_uri(root);
     let params = WriteParams {
         mode: WriteMode::Create,
-        store_params: Some(crate::storage::lance_store_params_for_uri(&manifest_path)?),
+        store_params: Some(crate::storage::lance_store_params_for_session(
+            &manifest_path,
+            control_session,
+        )?),
         enable_stable_row_ids: true,
         data_storage_version: Some(LanceFileVersion::V2_2),
         auto_cleanup: None,
@@ -347,15 +350,15 @@ pub(crate) async fn snapshot_state_at(
     branch: Option<&str>,
     version: u64,
     history: &ExtentCache,
+    control_session: &Arc<lance::session::Session>,
 ) -> Result<(Dataset, VersionState)> {
-    let control_session = crate::lance_access::control_session();
     let root = root_uri.trim_end_matches('/');
-    let dataset = open_manifest_dataset_with_session(root, branch, &control_session).await?;
+    let dataset = open_manifest_dataset_with_session(root, branch, control_session).await?;
     let dataset = dataset
         .checkout_version(version)
         .await
         .map_err(OmniError::storage)?;
-    let state = match crate::legacy::at_version(root, &control_session, history, &dataset).await? {
+    let state = match crate::legacy::at_version(root, control_session, history, &dataset).await? {
         None => VersionState::Rows(read_manifest_state(&dataset).await?),
         Some(at) => VersionState::Legacy(at),
     };
@@ -488,7 +491,10 @@ async fn create_empty_dataset(
     let reader = RecordBatchIterator::new(vec![Ok(batch)], schema);
     let params = WriteParams {
         mode: WriteMode::Create,
-        store_params: Some(crate::storage::lance_store_params_for_uri(uri)?),
+        store_params: Some(crate::storage::lance_store_params_for_session(
+            uri,
+            control_session,
+        )?),
         enable_stable_row_ids: true,
         data_storage_version: Some(LanceFileVersion::V2_2),
         allow_external_blob_outside_bases: true,

crates/omnigraph-catalog/src/history.rs

@@ -893,7 +893,7 @@ async fn store(
 ) -> Result<(Arc<ObjectStore>, Path)> {
     let uri = history_uri(root_uri);
     let wrapper = crate::instrumentation::history_wrapper();
-    let mut params = crate::storage::lance_store_params_for_uri(&uri)?;
+    let mut params = crate::storage::lance_store_params_for_session(&uri, session)?;
     params.object_store_wrapper = wrapper.clone();
     let (store, path) = ObjectStore::from_uri_and_params(session.store_registry(), &uri, &params)
         .await

crates/omnigraph-catalog/src/legacy.rs

@@ -15,7 +15,7 @@ mod record;
 pub use census::{
     CensusCounts, CensusError, CensusHead, CensusInput, CensusRef, FindingCode, LegacyCensus,
     LegacyFinding, LegacyPlan, MAX_CENSUS_CELLS, MAX_CENSUS_SNAPSHOT_BYTES, census, census_within,
-    conversion_batch, materialize, plan,
+    census_within_with_session, conversion_batch, materialize, plan,
 };
 
 use std::collections::{BTreeMap, BTreeSet, HashMap};

crates/omnigraph-catalog/src/legacy/census.rs

@@ -222,9 +222,27 @@ pub async fn census_within(
     max_cells: u64,
     max_snapshot_bytes: u64,
 ) -> CensusResult<LegacyCensus> {
-    let session = crate::lance_access::control_session();
+    census_within_with_session(
+        root_uri,
+        input,
+        source,
+        max_cells,
+        max_snapshot_bytes,
+        &crate::lance_access::control_session(),
+    )
+    .await
+}
+
+pub async fn census_within_with_session(
+    root_uri: &str,
+    input: &CensusInput,
+    source: &dyn LegacyManifestSource,
+    max_cells: u64,
+    max_snapshot_bytes: u64,
+    session: &Arc<lance::session::Session>,
+) -> CensusResult<LegacyCensus> {
     let main =
-        crate::layout::open_manifest_dataset_native_with_session(root_uri, None, &session).await?;
+        crate::layout::open_manifest_dataset_native_with_session(root_uri, None, session).await?;
     let mut counts = CensusCounts {
         live_refs: input.live.len() as u64,
         retired_refs: input.retired.len() as u64,

crates/omnigraph-catalog/src/lib.rs

@@ -203,8 +203,16 @@ pub struct CommitOutcome {
 /// `migrations::guard_stamp`, which the open paths use instead). Surfaces
 /// the storage version to operators (`omnigraph snapshot`).
 pub async fn internal_schema_stamp_at(root_uri: &str, branch: Option<&str>) -> Result<Option<u32>> {
-    let control_session = crate::lance_access::control_session();
-    let dataset = open_manifest_dataset_with_session(root_uri, branch, &control_session).await?;
+    internal_schema_stamp_at_with_session(root_uri, branch, &crate::lance_access::control_session())
+        .await
+}
+
+pub async fn internal_schema_stamp_at_with_session(
+    root_uri: &str,
+    branch: Option<&str>,
+    session: &Arc<lance::session::Session>,
+) -> Result<Option<u32>> {
+    let dataset = open_manifest_dataset_with_session(root_uri, branch, session).await?;
     Ok(migrations::read_stamp(&dataset))
 }
 
@@ -228,8 +236,20 @@ pub async fn manifest_has_external_base_paths(
     root_uri: &str,
     branch: Option<&str>,
 ) -> Result<bool> {
-    let control_session = crate::lance_access::control_session();
-    let dataset = open_manifest_dataset_with_session(root_uri, branch, &control_session).await?;
+    manifest_has_external_base_paths_with_session(
+        root_uri,
+        branch,
+        &crate::lance_access::control_session(),
+    )
+    .await
+}
+
+pub async fn manifest_has_external_base_paths_with_session(
+    root_uri: &str,
+    branch: Option<&str>,
+    session: &Arc<lance::session::Session>,
+) -> Result<bool> {
+    let dataset = open_manifest_dataset_with_session(root_uri, branch, session).await?;
     Ok(!dataset.manifest().base_paths.is_empty())
 }
 
@@ -784,7 +804,12 @@ impl Snapshot {
                     )
                     .await
             }
-            None => open_dataset_entry(entry, &self.root_uri, None).await,
+            None => {
+                let session = self.manifest_dataset.as_ref().map(|dataset| {
+                    crate::lance_access::statement_session(dataset.session().store_registry(), 0, 0)
+                });
+                open_dataset_entry(entry, &self.root_uri, session.as_ref()).await
+            }
         }
     }
 
@@ -1173,8 +1198,14 @@ impl CollectorBranch {
         history: &crate::commit_graph::HistoryCache,
     ) -> Result<Option<Snapshot>> {
         let branch = self.branch.as_deref();
-        let (dataset, state) =
-            snapshot_state_at(&self.root_uri, branch, version, history.objects()).await?;
+        let (dataset, state) = snapshot_state_at(
+            &self.root_uri,
+            branch,
+            version,
+            history.objects(),
+            &self.dataset.session(),
+        )
+        .await?;
         ManifestCoordinator::version_snapshot(&self.root_uri, branch, version, dataset, state)
     }
 }
@@ -1506,7 +1537,17 @@ impl ManifestCoordinator {
                      contract cannot be read",
                 ));
             };
-            let control_session = crate::lance_access::control_session();
+            let control_session = snapshot
+                .manifest_dataset
+                .as_ref()
+                .map(|dataset| dataset.session())
+                .or_else(|| {
+                    snapshot
+                        .read_caches
+                        .as_ref()
+                        .map(|cache| cache.session.clone())
+                })
+                .unwrap_or_else(crate::lance_access::control_session);
             return history::read_schema(&snapshot.root_uri, &control_session, digest, head).await;
         }
         let dataset =
@@ -1803,9 +1844,43 @@ impl ManifestCoordinator {
         branch: Option<&str>,
         version: u64,
         history: &HistoryCache,
+    ) -> Result<Snapshot> {
+        Self::snapshot_at_in_with_session(
+            root_uri,
+            branch,
+            version,
+            history,
+            &crate::lance_access::control_session(),
+        )
+        .await
+    }
+
+    pub async fn snapshot_at_with_session(
+        root_uri: &str,
+        branch: Option<&str>,
+        version: u64,
+        session: &Arc<lance::session::Session>,
+    ) -> Result<Snapshot> {
+        Self::snapshot_at_in_with_session(
+            root_uri,
+            branch,
+            version,
+            &HistoryCache::default(),
+            session,
+        )
+        .await
+    }
+
+    pub async fn snapshot_at_in_with_session(
+        root_uri: &str,
+        branch: Option<&str>,
+        version: u64,
+        history: &HistoryCache,
+        session: &Arc<lance::session::Session>,
     ) -> Result<Snapshot> {
         let root = root_uri.trim_end_matches('/');
-        let (dataset, state) = snapshot_state_at(root, branch, version, history.objects()).await?;
+        let (dataset, state) =
+            snapshot_state_at(root, branch, version, history.objects(), session).await?;
         let native = dataset.manifest().branch.clone();
         Self::version_snapshot(root, branch, version, dataset, state)?
             .ok_or_else(|| crate::legacy::pre_genesis(native.as_deref(), version))

crates/omnigraph-catalog/src/retention.rs

@@ -319,12 +319,21 @@ impl ManifestCoordinator {
         root_uri: &str,
         commit: &GraphCommit,
     ) -> Result<PinnedGraphManifest> {
-        let main = super::open_manifest_dataset_native_with_session(
+        Self::pinned_graph_commit_with_session(
             root_uri,
-            None,
+            commit,
             &crate::lance_access::control_session(),
         )
-        .await?;
+        .await
+    }
+
+    pub async fn pinned_graph_commit_with_session(
+        root_uri: &str,
+        commit: &GraphCommit,
+        session: &Arc<lance::session::Session>,
+    ) -> Result<PinnedGraphManifest> {
+        let main =
+            super::open_manifest_dataset_native_with_session(root_uri, None, session).await?;
         let legacy = ExtentCache::default();
         let mut candidates = Vec::new();
         match commit.graph_branch.as_deref() {
@@ -387,12 +396,21 @@ impl ManifestCoordinator {
         root_uri: &str,
         history: &HistoryCache,
     ) -> Result<Vec<(String, CommitGraph)>> {
-        let main = super::open_manifest_dataset_native_with_session(
+        Self::retired_commit_graphs_with_session(
             root_uri,
-            None,
+            history,
             &crate::lance_access::control_session(),
         )
-        .await?;
+        .await
+    }
+
+    pub async fn retired_commit_graphs_with_session(
+        root_uri: &str,
+        history: &HistoryCache,
+        session: &Arc<lance::session::Session>,
+    ) -> Result<Vec<(String, CommitGraph)>> {
+        let main =
+            super::open_manifest_dataset_native_with_session(root_uri, None, session).await?;
         let mut names = retired_manifest_branches(&main)
             .await?
             .into_keys()

crates/omnigraph-cli/src/admission.rs

@@ -5,6 +5,7 @@ use std::io::Write;
 use std::{collections::BTreeMap, future::Future, sync::Arc};
 
 use color_eyre::eyre::{Result, eyre};
+use omnigraph::storage::{CapturedStorageBinding, storage_for_binding};
 use omnigraph_cluster::{ClusterAdmission, ClusterAdmissionPurpose};
 use tokio::sync::Mutex;
 
@@ -20,14 +21,19 @@ pub(crate) async fn scope<F: Future>(future: F) -> F::Output {
 
 /// Read commands never acquire writer admission or run the writable capability
 /// probe. Cluster authority is observed around the open, not held as a lease.
-pub(crate) async fn open_read_only(
-    uri: &str,
+pub(crate) async fn open_read_only_with_binding(
+    binding: &CapturedStorageBinding,
     expected_state_cas: Option<&str>,
 ) -> Result<omnigraph::db::Omnigraph> {
-    let authority = omnigraph_cluster::GraphReadAuthority::capture(uri, expected_state_cas)
-        .await
-        .map_err(report)?;
-    let db = omnigraph::db::Omnigraph::open_read_only(uri).await?;
+    let authority =
+        omnigraph_cluster::GraphReadAuthority::capture_with_binding(binding, expected_state_cas)
+            .await
+            .map_err(report)?;
+    let db = omnigraph::db::Omnigraph::open_read_only_with_storage(
+        binding.uri(),
+        storage_for_binding(binding)?,
+    )
+    .await?;
     if let Some(authority) = authority {
         authority.validate_opened(&db).await.map_err(report)?;
     }
@@ -36,33 +42,29 @@ pub(crate) async fn open_read_only(
 
 /// Acquire before the writable opener (including its capability probe), and
 /// reuse only the same canonical root's admission during this command.
-pub(crate) async fn ensure_graph(uri: &str, stderr: &mut dyn Write) -> Result<()> {
-    let Some(root) = omnigraph_cluster::cluster_root_for_graph_uri(uri)
+pub(crate) async fn ensure_graph_with_binding(
+    binding: &CapturedStorageBinding,
+    stderr: &mut dyn Write,
+) -> Result<()> {
+    let uri = binding.uri();
+    let Some(root) = omnigraph_cluster::cluster_root_for_graph_binding(binding)
         .await
         .map_err(report)?
     else {
         return Ok(());
     };
-    let root = omnigraph::storage::normalize_root_uri(&root)?;
-    let key = if omnigraph::storage::storage_kind_for_uri(&root)?
-        == omnigraph::storage::StorageKind::Local
-    {
-        std::fs::canonicalize(&root)?
-            .to_str()
-            .ok_or_else(|| eyre!("cluster root is not valid UTF-8"))?
-            .to_owned()
-    } else {
-        root
-    };
+    let key = binding.for_uri(&root)?.identity();
     let owners = COMMAND.try_with(Arc::clone).unwrap_or_default();
     let mut owners = owners.lock().await;
     if let Some(owner) = owners.get(&key) {
         return owner.validate_graph_uri(uri).await.map_err(report);
     }
-    if let Some(owner) =
-        omnigraph_cluster::acquire_graph_admission(uri, ClusterAdmissionPurpose::GraphOperation)
-            .await
-            .map_err(report)?
+    if let Some(owner) = omnigraph_cluster::acquire_graph_admission_with_binding(
+        binding,
+        ClusterAdmissionPurpose::GraphOperation,
+    )
+    .await
+    .map_err(report)?
     {
         // Report before dispatch as well as before any command-local exit or
         // lost output. Success of an arbitrary CLI command is not a qualified