diff --git a/docs/src/python/python.md b/docs/src/python/python.md
index 4d0dfc87a..356b49ae7 100644
--- a/docs/src/python/python.md
+++ b/docs/src/python/python.md
@@ -20,6 +20,17 @@ pip install lancedb
 The following methods describe the synchronous API client. There
 is also an [asynchronous API client](#connections-asynchronous).
 
+Connections to `db://` databases reject local-only options with
+`NotImplementedError`. These include `session` and `manifest_enabled=True` on
+connect, per-table storage and write options on create, and `index_cache_size`,
+`storage_options`, or `location` on open (where the method accepts them).
+Omit these options to use server-managed storage and caching. Connection-level
+`storage_options` supported by an Enterprise deployment remain available.
+
+The synchronous remote `create_table` uses the same positional order as the
+local API: `name, data, schema, mode, exist_ok, on_bad_vectors, fill_value,
+embedding_functions`. Prefer keyword arguments for optional parameters.
+
 ## Connections (Synchronous)
 
 ::: lancedb.connect
diff --git a/docs/web/basics/sdk.mdx b/docs/web/basics/sdk.mdx
index fc386bc7a..b20fc1662 100644
--- a/docs/web/basics/sdk.mdx
+++ b/docs/web/basics/sdk.mdx
@@ -92,6 +92,13 @@ manages background maintenance.
   [background maintenance](/maintenance/background-jobs) rather than calling
   the embedded table's `optimize()` method.
 
+{/* Draft for the remote option validation release.
+- **Remove local storage and cache settings.** `db://` connections reject local
+  sessions, table cache settings, and per-table storage and write options. Omit
+  these options so Enterprise manages storage and caching. Connection-level
+  storage options supported by your Enterprise deployment remain available.
+*/}
+
 For SQL connection details, see [SQL](/basics/sql). For the operating model and
 deployment options, see [LanceDB Enterprise](/enterprise).
 
diff --git a/nodejs/__test__/query.test.ts b/nodejs/__test__/query.test.ts
index 786f04015..068e55c2f 100644
--- a/nodejs/__test__/query.test.ts
+++ b/nodejs/__test__/query.test.ts
@@ -538,11 +538,17 @@ describe("IVF search effort", () => {
     const db = await connect(directory.name);
     const rows = Array.from({ length: 512 }, (_, id) => ({
       id,
-      vector: Array.from(
-        { length: 16 },
-        (_, j) =>
-          (Math.floor(id / 32) === j ? 1 : 0) + Math.sin(id * 17 + j) * 0.001,
-      ),
+      // Keep all eight exact neighbors together even if random IVF training
+      // splits the first cluster. This tests effort, not training recall.
+      vector:
+        id < 8
+          ? [...vector]
+          : Array.from(
+              { length: 16 },
+              (_, j) =>
+                (Math.floor(id / 32) === j ? 1 : 0) +
+                Math.sin(id * 17 + j) * 0.001,
+            ),
     }));
     table = await db.createTable("effort", rows.slice(0, 256));
     await table.add(rows.slice(256));
diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts
index 7e9479f94..d443537b0 100644
--- a/nodejs/__test__/remote.test.ts
+++ b/nodejs/__test__/remote.test.ts
@@ -98,6 +98,55 @@ async function withMockDatabase(
 }
 
 describe("remote connection", () => {
+  it.each([
+    { dataStorageVersion: "2.0" },
+    { enableV2ManifestPaths: false },
+    { storageOptions: { bogusOptionXyz: "1" } },
+  ])("rejects local create options %j before sending data", async (options) => {
+    const requests: string[] = [];
+    await withMockDatabase(
+      (req, res) => {
+        requests.push(req.url ?? "");
+        res.writeHead(200).end();
+      },
+      async (db) => {
+        await expect(
+          db.createTable("test", [{ value: 1 }], {
+            ...options,
+            mode: "overwrite",
+          }),
+        ).rejects.toThrow(/write_options.*remote/);
+        await expect(
+          db.createEmptyTable(
+            "test",
+            new Schema([new Field("value", new Int64(), true)]),
+            options,
+          ),
+        ).rejects.toThrow(/write_options.*remote/);
+      },
+    );
+    expect(requests).toEqual([]);
+  });
+
+  it.each([{ indexCacheSize: 5 }, { storageOptions: { x: "y" } }])(
+    "rejects local open options %j before sending a request",
+    async (options) => {
+      const requests: string[] = [];
+      await withMockDatabase(
+        (req, res) => {
+          requests.push(req.url ?? "");
+          res.writeHead(200).end();
+        },
+        async (db) => {
+          await expect(
+            db.openTable("test", undefined, options),
+          ).rejects.toThrow(/not supported for remote/);
+        },
+      );
+      expect(requests).toEqual([]);
+    },
+  );
+
   it.each([false, true])(
     "normalizes remote blob writes (range target: %s)",
     async (wide) => {
diff --git a/python/python/lancedb/__init__.py b/python/python/lancedb/__init__.py
index a3aa55209..5e879f61d 100644
--- a/python/python/lancedb/__init__.py
+++ b/python/python/lancedb/__init__.py
@@ -281,6 +281,21 @@ def connect(
             "uri is required when not connecting through a namespace client"
         )
     if isinstance(uri, str) and uri.startswith("db://"):
+        # Preserve the deprecated remote timeout options, but validate all other
+        # unknown keywords just as the local connection does.
+        unknown_kwargs = {
+            key: value
+            for key, value in kwargs.items()
+            if key not in ("connection_timeout", "read_timeout")
+        }
+        if unknown_kwargs:
+            raise ValueError(f"Unknown keyword arguments: {unknown_kwargs}")
+        if session is not None:
+            raise NotImplementedError("session is not supported for remote connections")
+        if manifest_enabled:
+            raise NotImplementedError(
+                "manifest_enabled is not supported for remote connections"
+            )
         if api_key is None:
             api_key = os.environ.get("LANCEDB_API_KEY")
         if api_key is None:
diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py
index 4ffd3ac4a..291cc52d6 100644
--- a/python/python/lancedb/remote/db.py
+++ b/python/python/lancedb/remote/db.py
@@ -7,7 +7,6 @@ from __future__ import annotations
 from dataclasses import replace
 from datetime import timedelta
 import json
-import logging
 from concurrent.futures import ThreadPoolExecutor
 import sys
 from typing import (
@@ -502,18 +501,14 @@ class RemoteDBConnection(DBConnection):
 
         if namespace_path is None:
             namespace_path = []
-        if storage_options is not None:
-            logging.info(
-                "storage_options is ignored for remote tables"
-                " (storage is managed; set storage_options on connect() instead)"
+        table = LOOP.run(
+            self._conn.open_table(
+                name,
+                namespace_path=namespace_path,
+                storage_options=storage_options,
+                index_cache_size=index_cache_size,
             )
-        if index_cache_size is not None:
-            logging.info(
-                "index_cache_size is ignored for remote tables"
-                " (there is no local cache to configure)"
-            )
-
-        table = LOOP.run(self._conn.open_table(name, namespace_path=namespace_path))
+        )
         tbl = RemoteTable(
             table,
             self.db_name,
@@ -593,6 +588,9 @@ class RemoteDBConnection(DBConnection):
         embedding_functions: Optional[List[EmbeddingFunctionConfig]] = None,
         *,
         namespace_path: Optional[List[str]] = None,
+        storage_options: Optional[Dict[str, str]] = None,
+        data_storage_version: Optional[str] = None,
+        enable_v2_manifest_paths: Optional[bool] = None,
     ) -> Table:
         """Create a [Table][lancedb.table.Table] in the database.
 
@@ -632,6 +630,12 @@ class RemoteDBConnection(DBConnection):
         embedding_functions: list of EmbeddingFunctionConfig, optional
             The embedding functions to store in the table schema and use to
             generate vectors when creating the table or adding data.
+        storage_options: dict, optional
+            Not supported for remote table creation. Raises NotImplementedError.
+        data_storage_version: str, optional
+            Not supported for remote table creation. Raises NotImplementedError.
+        enable_v2_manifest_paths: bool, optional
+            Not supported for remote table creation. Raises NotImplementedError.
 
         Returns
         -------
@@ -698,6 +702,14 @@ class RemoteDBConnection(DBConnection):
         LanceTable(table4)
 
         """
+        for option, value in (
+            ("data_storage_version", data_storage_version),
+            ("enable_v2_manifest_paths", enable_v2_manifest_paths),
+        ):
+            if value is not None:
+                raise NotImplementedError(
+                    f"{option} is not supported for remote table creation"
+                )
         if exist_ok:
             if mode == "create":
                 mode = "exist_ok"
@@ -719,6 +731,7 @@ class RemoteDBConnection(DBConnection):
                 on_bad_vectors=on_bad_vectors,
                 fill_value=fill_value,
                 embedding_functions=embedding_functions,
+                storage_options=storage_options,
             )
         )
         return RemoteTable(
diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py
index db3394735..32f999d18 100644
--- a/python/python/tests/test_remote_db.py
+++ b/python/python/tests/test_remote_db.py
@@ -1823,12 +1823,12 @@ def test_remote_unsupported_legacy_hnsw_pq():
                 )
 
 
-def test_remote_python_only_messages(caplog):
+def test_remote_python_only_messages():
     with mock_lancedb_connection(remote_unsupported_handler) as db:
-        with caplog.at_level("INFO"):
-            table = db.open_table("test", storage_options={}, index_cache_size=1)
-        assert "storage_options is ignored for remote tables" in caplog.text
-        assert "index_cache_size is ignored for remote tables" in caplog.text
+        for options in [{"storage_options": {}}, {"index_cache_size": 1}]:
+            with pytest.raises(NotImplementedError, match=next(iter(options))):
+                db.open_table("test", **options)
+        table = db.open_table("test")
         for method in ["to_arrow", "to_pandas", "to_polars"]:
             with pytest.raises(NotImplementedError) as error:
                 getattr(table, method)()
diff --git a/python/python/tests/test_remote_options.py b/python/python/tests/test_remote_options.py
new file mode 100644
index 000000000..2022aa29d
--- /dev/null
+++ b/python/python/tests/test_remote_options.py
@@ -0,0 +1,154 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import inspect
+from urllib.parse import parse_qs, urlparse
+
+import lancedb
+import pyarrow as pa
+import pytest
+
+from test_remote_db import mock_lancedb_connection, mock_lancedb_connection_async
+
+
+@pytest.mark.parametrize("uri", ["memory://", "db://dev"])
+def test_unknown_connect_options(uri):
+    with pytest.raises(ValueError, match="Unknown keyword arguments.*bogus"):
+        lancedb.connect(uri, api_key="fake", bogus=1)
+
+
+@pytest.mark.parametrize("option", ["session", "manifest_enabled"])
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.asyncio
+async def test_remote_connect_rejects_local_options(option, asynchronous):
+    value = lancedb.Session() if option == "session" else True
+    with pytest.raises(NotImplementedError, match=option):
+        if asynchronous:
+            await lancedb.connect_async("db://dev", api_key="fake", **{option: value})
+        else:
+            lancedb.connect("db://dev", api_key="fake", **{option: value})
+
+
+def test_remote_create_signature():
+    from lancedb.db import DBConnection
+    from lancedb.remote.db import RemoteDBConnection
+
+    expected = inspect.signature(DBConnection.create_table).parameters
+    actual = inspect.signature(RemoteDBConnection.create_table).parameters
+    assert list(actual) == list(expected)
+    for name in expected:
+        assert actual[name].kind == expected[name].kind
+        if name == "mode":
+            # Remote connections retain their existing server-default mode.
+            assert actual[name].default is None
+        else:
+            assert actual[name].default == expected[name].default
+
+
+def test_remote_create_positional_mode():
+    modes = []
+
+    def handler(request):
+        assert urlparse(request.path).path == "/v1/table/ct_pos/create/"
+        modes.append(parse_qs(urlparse(request.path).query)["mode"])
+        request.send_response(200)
+        request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        data = [{"value": 1}]
+        db.create_table("ct_pos", data, None, "overwrite")
+        db.create_table("ct_pos", data, None, "overwrite")
+        db.create_table("ct_pos", data, None, "create", True)
+        with pytest.raises(ValueError, match="Invalid mode"):
+            db.create_table("ct_pos", data, None, "drop")
+    assert modes == [["overwrite"], ["overwrite"], ["exist_ok"]]
+
+
+CREATE_OPTIONS = [
+    {"new_table_data_storage_version": "2.0"},
+    {"new_table_enable_v2_manifest_paths": "false"},
+    {"bogus_option_xyz": "1"},
+]
+
+
+@pytest.mark.parametrize("storage_options", CREATE_OPTIONS)
+@pytest.mark.parametrize("empty", [False, True])
+@pytest.mark.asyncio
+async def test_remote_create_rejects_write_options(storage_options, empty):
+    requests = []
+
+    def handler(request):
+        requests.append(request.path)
+        request.send_response(200)
+        request.end_headers()
+
+    kwargs = {"schema": pa.schema([("value", pa.int64())])}
+    if not empty:
+        kwargs["data"] = [{"value": 1}]
+    async with mock_lancedb_connection_async(handler) as db:
+        with pytest.raises(NotImplementedError, match="write_options.*remote"):
+            await db.create_table("t", storage_options=storage_options, **kwargs)
+    assert requests == []
+
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        {"data_storage_version": "2.0"},
+        {"enable_v2_manifest_paths": False},
+        {"storage_options": {"x": "1"}},
+    ],
+)
+def test_sync_remote_create_rejects_write_options(options):
+    requests = []
+
+    def handler(request):
+        requests.append(request.path)
+        request.send_response(200)
+        request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        with pytest.raises(NotImplementedError, match="remote"):
+            db.create_table("t", [{"value": 1}], **options)
+    assert requests == []
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize(
+    "options", [{"index_cache_size": 10}, {"storage_options": {"x": "y"}}]
+)
+@pytest.mark.asyncio
+async def test_remote_open_rejects_local_options(options, asynchronous):
+    requests = []
+
+    def handler(request):
+        requests.append(request.path)
+        request.send_response(200)
+        request.end_headers()
+        request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
+
+    if asynchronous:
+        async with mock_lancedb_connection_async(handler) as db:
+            with pytest.raises(NotImplementedError, match="remote"):
+                await db.open_table("t", **options)
+    else:
+        with mock_lancedb_connection(handler) as db:
+            with pytest.raises(NotImplementedError, match="remote"):
+                db.open_table("t", **options)
+    assert requests == []
+
+
+@pytest.mark.asyncio
+async def test_remote_open_rejects_location():
+    requests = []
+
+    def handler(request):
+        requests.append(request.path)
+        request.send_response(200)
+        request.end_headers()
+        request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
+
+    async with mock_lancedb_connection_async(handler) as db:
+        with pytest.raises(NotImplementedError, match="location.*remote"):
+            await db.open_table("t", location="/tmp/zzz")
+    assert requests == []
diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs
index 26afef5e8..bff97bf31 100644
--- a/rust/lancedb/src/connection.rs
+++ b/rust/lancedb/src/connection.rs
@@ -174,6 +174,8 @@ impl OpenTableBuilder {
 
     /// Advanced parameters that can be used to customize table reads
     ///
+    /// Remote connections reject these parameters with [`Error::NotSupported`].
+    ///
     /// If set, these will take precedence over any overlapping `OpenTableOptions` options
     pub fn lance_read_params(mut self, params: ReadParams) -> Self {
         self.request.lance_read_params = Some(params);
@@ -1530,6 +1532,7 @@ impl ConnectBuilder {
 
     /// Enable or disable manifest-backed directory namespace mode for local
     /// native connections.
+    /// Remote connections reject enabling this option with [`Error::NotSupported`].
     ///
     /// When enabled, the connection uses the directory namespace database
     /// directly for all table operations and forces
@@ -1565,6 +1568,7 @@ impl ConnectBuilder {
     }
 
     /// Set a custom session for object stores and caching.
+    /// Remote connections reject this option with [`Error::NotSupported`].
     ///
     /// By default, a new session with default configuration will be created.
     /// This method allows you to provide a custom session with your own
@@ -1596,6 +1600,21 @@ impl ConnectBuilder {
     fn execute_remote(self) -> Result<Connection> {
         use crate::remote::db::RemoteDatabaseOptions;
 
+        for (option, provided) in [
+            ("session", self.request.session.is_some()),
+            ("manifest_enabled", self.request.manifest_enabled),
+            (
+                "namespace_client_properties",
+                !self.request.namespace_client_properties.is_empty(),
+            ),
+        ] {
+            if provided {
+                return Err(Error::NotSupported {
+                    message: format!("{option} is not supported for remote connections"),
+                });
+            }
+        }
+
         let mut merged_options = self.request.options.clone();
         Self::apply_env_defaults(&ENV_VARS_TO_STORAGE_OPTS, &mut merged_options);
         let sql_host_override = merged_options.get(OPT_REMOTE_SQL_HOST_OVERRIDE).cloned();
diff --git a/rust/lancedb/src/connection/create_table.rs b/rust/lancedb/src/connection/create_table.rs
index 39cc82ec0..9d4408ba7 100644
--- a/rust/lancedb/src/connection/create_table.rs
+++ b/rust/lancedb/src/connection/create_table.rs
@@ -45,6 +45,9 @@ impl CreateTableBuilder {
     }
 
     /// Apply the given write options when writing the initial data
+    ///
+    /// Remote connections reject explicit Lance write parameters with
+    /// [`Error::NotSupported`] instead of silently using server defaults.
     pub fn write_options(mut self, write_options: WriteOptions) -> Self {
         self.request.write_options = write_options;
         self
@@ -70,6 +73,8 @@ impl CreateTableBuilder {
 
     /// Set multiple options for the storage layer.
     ///
+    /// Table-level storage options are not supported by remote connections.
+    ///
     /// Options already set on the connection will be inherited by the table,
     /// but can be overridden here.
     ///
diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs
index d33565653..5c299dc5a 100644
--- a/rust/lancedb/src/remote/db.rs
+++ b/rust/lancedb/src/remote/db.rs
@@ -1481,6 +1481,23 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     }
 
     async fn create_table(&self, mut request: CreateTableRequest) -> Result<Arc<dyn BaseTable>> {
+        // The remote create endpoint does not transmit Lance write parameters.
+        // Reject them before consuming data or issuing a request, especially
+        // for overwrite: silently using the server's format can lose intent.
+        for (option, provided) in [
+            (
+                "write_options (including storage_options)",
+                request.write_options.lance_write_params.is_some(),
+            ),
+            ("location", request.location.is_some()),
+            ("namespace_client", request.namespace_client.is_some()),
+        ] {
+            if provided {
+                return Err(Error::NotSupported {
+                    message: format!("{option} is not supported for remote table creation"),
+                });
+            }
+        }
         let data_schema = request.data.schema();
         let body = stream_as_body(request.data.scan_as_stream())?;
 
@@ -1605,6 +1622,23 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     }
 
     async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>> {
+        // Validate before looking in the cache as well as before sending HTTP.
+        for (option, provided) in [
+            ("index_cache_size", request.index_cache_size.is_some()),
+            (
+                "lance_read_params (including storage_options)",
+                request.lance_read_params.is_some(),
+            ),
+            ("location", request.location.is_some()),
+            ("namespace_client", request.namespace_client.is_some()),
+            ("managed_versioning", request.managed_versioning.is_some()),
+        ] {
+            if provided {
+                return Err(Error::NotSupported {
+                    message: format!("{option} is not supported for remote table access"),
+                });
+            }
+        }
         let identifier = build_table_identifier(&request.name, &request.namespace_path)?;
         let cache_key = build_cache_key(&request.name, &request.namespace_path);
 
@@ -2504,14 +2538,14 @@ mod tests {
         let table = conn.open_table("table1").execute().await.unwrap();
         assert_eq!(table.name(), "table1");
 
-        // Storage options should be ignored.
-        let table = conn
+        // Per-table storage options must not be silently ignored.
+        let result = conn
             .open_table("table1")
             .storage_option("key", "value")
             .execute()
-            .await
-            .unwrap();
-        assert_eq!(table.name(), "table1");
+            .await;
+        assert!(matches!(result, Err(Error::NotSupported { message })
+            if message.contains("storage_options")));
     }
 
     #[tokio::test]
@@ -2735,6 +2769,126 @@ mod tests {
         );
     }
 
+    #[rstest]
+    #[case("new_table_data_storage_version", "2.0")]
+    #[case("new_table_enable_v2_manifest_paths", "false")]
+    #[case("bogus_option_xyz", "1")]
+    #[tokio::test]
+    async fn test_create_rejects_local_storage_options(#[case] key: &str, #[case] value: &str) {
+        let conn = Connection::new_with_handler(|_| -> http::Response<&'static str> {
+            panic!("must not send a create request")
+        });
+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+        let data = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+        )
+        .unwrap();
+        for builder in [
+            conn.create_table("table1", data),
+            conn.create_empty_table("table1", schema),
+        ] {
+            let result = builder
+                .mode(CreateTableMode::Overwrite)
+                .storage_option(key, value)
+                .execute()
+                .await;
+            assert!(matches!(result, Err(Error::NotSupported { message })
+                if message.contains("write_options") && message.contains("remote")));
+        }
+    }
+
+    #[tokio::test]
+    async fn test_create_rejects_local_write_params_and_location() {
+        let conn = Connection::new_with_handler(|_| -> http::Response<&'static str> {
+            panic!("must not send a create request")
+        });
+        let schema = Arc::new(Schema::empty());
+        for builder in [
+            conn.create_empty_table("table1", schema.clone())
+                .write_options(crate::table::WriteOptions {
+                    lance_write_params: Some(lance::dataset::WriteParams::default()),
+                }),
+            conn.create_empty_table("table1", schema)
+                .location("/tmp/elsewhere"),
+        ] {
+            assert!(matches!(
+                builder.execute().await,
+                Err(Error::NotSupported { .. })
+            ));
+        }
+    }
+
+    #[rstest]
+    #[tokio::test]
+    async fn test_open_rejects_local_options_even_when_cached(#[values(false, true)] cached: bool) {
+        let mut db = super::RemoteDatabase::new_mock(|request| {
+            assert_eq!(request.url().path(), "/v1/table/table1/create/");
+            http::Response::builder().status(200).body("").unwrap()
+        });
+        db.table_cache = moka::future::Cache::new(10);
+        let table_cache = db.table_cache.clone();
+        let conn = Connection::new(
+            Arc::new(db),
+            Arc::new(crate::embeddings::MemoryRegistry::new()),
+        );
+        if cached {
+            conn.create_empty_table("table1", Arc::new(Schema::empty()))
+                .execute()
+                .await
+                .unwrap();
+        }
+        assert_eq!(
+            table_cache
+                .get(&build_cache_key("table1", &[]))
+                .await
+                .is_some(),
+            cached
+        );
+        for (option, builder) in [
+            (
+                "index_cache_size",
+                conn.open_table("table1").index_cache_size(10),
+            ),
+            (
+                "storage_options",
+                conn.open_table("table1").storage_option("x", "y"),
+            ),
+            ("location", conn.open_table("table1").location("/tmp/zzz")),
+            (
+                "managed_versioning",
+                conn.open_table("table1").managed_versioning(false),
+            ),
+            (
+                "lance_read_params",
+                conn.open_table("table1")
+                    .lance_read_params(Default::default()),
+            ),
+        ] {
+            let result = builder.execute().await;
+            assert!(matches!(result, Err(Error::NotSupported { message })
+                if message.contains(option) && message.contains("remote")));
+        }
+    }
+
+    #[tokio::test]
+    async fn test_connect_rejects_local_options() {
+        for (option, builder) in [
+            (
+                "manifest_enabled",
+                ConnectBuilder::new("db://dev").manifest_enabled(true),
+            ),
+            (
+                "namespace_client_properties",
+                ConnectBuilder::new("db://dev").namespace_client_properties([("root", "/tmp/db")]),
+            ),
+        ] {
+            let result = builder.api_key("fake").region("us-east-1").execute().await;
+            assert!(matches!(result, Err(Error::NotSupported { message })
+                if message.contains(option) && message.contains("remote")));
+        }
+    }
+
     #[rstest]
     #[case::matching(Field::new("a", DataType::Int32, false), true)]
     #[case::different_name(Field::new("x", DataType::Int32, false), false)]