diff --git a/auth/oauth/m2m/m2m.go b/auth/oauth/m2m/m2m.go
index 5ed871ee..4c8a6c95 100644
--- a/auth/oauth/m2m/m2m.go
+++ b/auth/oauth/m2m/m2m.go
@@ -36,6 +36,15 @@ type authClient struct {
 	mx           sync.Mutex
 }
 
+// M2MCredentials exposes the raw client-credentials so the SEA-via-kernel backend
+// can drive the kernel's own M2M flow, keeping cfg.Authenticator the single source
+// of truth for auth mode. It structurally satisfies the M2MCredentialsProvider
+// interface the kernel backend asserts (defined in internal/backend/kernel, so the
+// secret-reading capability is not part of the driver's public API).
+func (c *authClient) M2MCredentials() (clientID, clientSecret string) {
+	return c.clientID, c.clientSecret
+}
+
 // Auth will start the OAuth Authorization Flow to authenticate the cli client
 // using the users credentials in the browser. Compatible with SSO.
 func (c *authClient) Authenticate(r *http.Request) error {
diff --git a/auth/oauth/u2m/authenticator.go b/auth/oauth/u2m/authenticator.go
index 456fba6d..c08f87b7 100644
--- a/auth/oauth/u2m/authenticator.go
+++ b/auth/oauth/u2m/authenticator.go
@@ -78,6 +78,13 @@ type u2mAuthenticator struct {
 	mx          sync.Mutex
 }
 
+// U2MClientID exposes the cloud-inferred OAuth client id so the SEA-via-kernel
+// backend uses the same client id for the kernel's browser/PKCE flow that the
+// Thrift path would, keeping cfg.Authenticator the single source of truth for auth
+// mode. It structurally satisfies the U2MCredentialsProvider interface the kernel
+// backend asserts (defined in internal/backend/kernel, off the public API).
+func (c *u2mAuthenticator) U2MClientID() string { return c.clientID }
+
 // Auth will start the OAuth Authorization Flow to authenticate the cli client
 // using the users credentials in the browser. Compatible with SSO.
 func (c *u2mAuthenticator) Authenticate(r *http.Request) error {
diff --git a/doc.go b/doc.go
index 8f4c69e6..7d5eaa5b 100644
--- a/doc.go
+++ b/doc.go
@@ -197,20 +197,32 @@ path databricks_sql_kernel, which would match nothing):
 	# kernel logs plus its HTTP stack (hyper/reqwest):
 	DBSQL_KERNEL_DEBUG=1 RUST_LOG=debug ./your_app 2>&1
 
-The kernel backend currently supports PAT authentication; reading scalar, nested,
-and complex-typed results (CloudFetch is handled transparently); context
-cancellation during execute (a cancelled ctx fires a real server-side cancel; on
-the read path cancellation is honored at result-batch boundaries, not mid-fetch);
-and the TLS, proxy, and session-conf (query tags, statement timeout, time zone)
-connection options. OAuth (M2M/U2M), initial catalog/schema, and
-WithEnableMetricViewMetadata are not yet supported and return a clear error at
-connect time rather than being silently ignored; likewise WithTimeout (a server
-query timeout the kernel C ABI can't set) and WithRetries used to disable retries
-(the kernel retries internally). Bound query parameters are likewise not yet
-supported and return a clear error at execute time (they arrive per-query, not at
-connect). None of these is silently ignored. (Metadata is issued as ordinary SQL
-— SHOW/DESCRIBE/information_schema — and runs on this backend like any other
-query.)
+The kernel backend currently supports PAT and OAuth (M2M via WithClientCredentials,
+and U2M via the authType=oauthU2M DSN param — the interactive browser/PKCE flow is
+owned by the kernel) authentication; reading scalar, nested, and complex-typed
+results (CloudFetch is handled transparently); context cancellation during execute
+(a cancelled ctx fires a real server-side cancel; on the read path cancellation is
+honored at result-batch boundaries, not mid-fetch); the initial namespace
+(WithInitialNamespace catalog/schema, applied post-connect via USE CATALOG / USE
+SCHEMA since the kernel C ABI has no namespace setter); metric-view metadata
+(WithEnableMetricViewMetadata, sent as the same server session conf the Thrift
+backend uses); and the TLS, proxy, and session-conf (query tags, statement timeout,
+time zone) connection options. WithTimeout (a server query timeout the kernel C ABI
+can't set) and WithRetries used to disable retries (the kernel retries internally)
+return a clear error at connect time rather than being silently ignored; token-
+provider, external/static, and federated authenticators are likewise not supported
+and rejected loudly. Bound query parameters are not yet supported and return a clear
+error at execute time (they arrive per-query, not at connect). None of these is
+silently ignored. (Metadata is issued as ordinary SQL — SHOW/DESCRIBE/
+information_schema — and runs on this backend like any other query.)
+
+OAuth U2M is interactive. On a cache miss (no valid cached refresh token) opening a
+connection launches the system browser and blocks until the user completes login or
+the kernel's built-in callback timeout (~120s) expires — and because the kernel C
+ABI cannot interrupt session open mid-call, a deadline on the connection context is
+not honored during that window (nor can the timeout be shortened; the C ABI exposes
+no override). A cached, still-valid token opens without a browser. Use PAT or OAuth
+M2M for headless / deadline-bound connects.
 
 WithMaxRows and retry tuning (WithRetries with a positive limit) are accepted but
 not applied on the kernel path: the kernel manages result fetching and retries
diff --git a/internal/backend/kernel/auth.go b/internal/backend/kernel/auth.go
new file mode 100644
index 00000000..a78d23f5
--- /dev/null
+++ b/internal/backend/kernel/auth.go
@@ -0,0 +1,65 @@
+package kernel
+
+// This file is intentionally NOT behind the `cgo && databricks_kernel` build tag:
+// the Auth descriptor and the credential-provider interfaces are plain Go with no
+// cgo, so they compile in the default build too. validateKernelConfig (untagged)
+// resolves an Auth from the config; OpenSession (tagged) maps it to the kernel's
+// set_auth_* C setters.
+
+// AuthMode selects which kernel auth form OpenSession applies.
+type AuthMode int
+
+const (
+	AuthPAT AuthMode = iota // personal access token
+	AuthM2M                 // OAuth client-credentials (client id + secret)
+	AuthU2M                 // OAuth user-to-machine (browser/PKCE; kernel-owned flow)
+)
+
+// Auth is the resolved auth descriptor for a kernel connection. Only the fields
+// for Mode are populated. The connector fills it from the driver config (see
+// validateKernelConfig); OpenSession maps it to exactly one
+// kernel_session_config_set_auth_* call.
+// Scopes and RedirectPort map to the optional args of set_auth_u2m and are wired
+// through to it by setAuth, but no Go path populates them today: the driver exposes
+// no user option for U2M scopes or redirect port on either backend (the native
+// Thrift path hardcodes both), so resolveKernelAuth leaves them zero and the kernel
+// applies its defaults. They are kept — rather than dropped and the setter hardcoded
+// to NULL/0 — so kernel.Auth models the full set_auth_u2m surface: adding a future
+// WithOAuthRedirectPort / scopes option (ODBC PR #102 already exposes a redirect
+// port) becomes populating these, not re-plumbing the setter. TestSetAuthByMode's
+// "U2M full" case pins that marshalling so the dormant path stays correct.
+type Auth struct {
+	Mode         AuthMode
+	Token        string   // PAT
+	ClientID     string   // M2M + U2M (U2M: the cloud-inferred Go client id)
+	ClientSecret string   // M2M
+	Scopes       []string // U2M — dormant (see note above); nil → kernel default scopes
+	RedirectPort uint16   // U2M — dormant (see note above); 0 → kernel default port (8020)
+}
+
+// M2MCredentialsProvider is implemented by the OAuth M2M authenticator to expose
+// its raw client-credentials. The kernel backend reads these to drive the kernel's
+// own M2M flow (the kernel owns the token exchange), rather than using the
+// authenticator's Authenticate method. So cfg.Authenticator stays the single source
+// of truth for auth on both backends — the kernel selects M2M by asserting this
+// interface, so the last WithX option applied wins, exactly as on Thrift.
+//
+// It lives in this internal package (not the public auth package) so the
+// secret-reading capability is never exposed on the driver's public API; the
+// unexported concrete m2m authenticator satisfies it structurally.
+type M2MCredentialsProvider interface {
+	// M2MCredentials returns the client id and client secret. (The kernel's C-ABI
+	// M2M setter takes no scopes — it applies its own default scope set, matching
+	// the Go authenticator's default — so scopes are not exposed here.)
+	M2MCredentials() (clientID, clientSecret string)
+}
+
+// U2MCredentialsProvider is implemented by the OAuth U2M authenticator to expose
+// the cloud-inferred client id the kernel should use for its browser/PKCE flow (so
+// the kernel path uses the same client id the Thrift path would). Internal for the
+// same reason as M2MCredentialsProvider; satisfied structurally by the unexported
+// u2m authenticator.
+type U2MCredentialsProvider interface {
+	// U2MClientID returns the OAuth client id for the U2M browser flow.
+	U2MClientID() string
+}
diff --git a/internal/backend/kernel/backend.go b/internal/backend/kernel/backend.go
index 1cb85c10..4bbaa423 100644
--- a/internal/backend/kernel/backend.go
+++ b/internal/backend/kernel/backend.go
@@ -12,6 +12,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/databricks/databricks-sql-go/internal/backend"
@@ -25,7 +26,7 @@ type Config struct {
 	Host        string // workspace hostname, no scheme
 	HTTPPath    string // e.g. /sql/1.0/warehouses/abc123 (carries ?o= org routing)
 	WarehouseID string // bare warehouse id; preferred over HTTPPath when set
-	Token       string // PAT (dapi...)
+	Auth        Auth   // PAT / OAuth M2M / OAuth U2M
 
 	// SessionConf carries server-bound session confs verbatim — the same map the
 	// Thrift backend forwards (STATEMENT_TIMEOUT, QUERY_TAGS, TIMEZONE, …).
@@ -46,6 +47,13 @@ type Config struct {
 	// Location is the session time zone used to render DATE / TIMESTAMP values,
 	// matching the Thrift path which returns them in this location. nil means UTC.
 	Location *time.Location
+
+	// Catalog / Schema select the initial namespace. The kernel C ABI has no
+	// catalog/schema config setter, so OpenSession applies them post-connect by
+	// running USE CATALOG / USE SCHEMA (the OSS ODBC driver's workaround). Empty
+	// leaves the session in the server default namespace.
+	Catalog string
+	Schema  string
 }
 
 // KernelBackend implements backend.Backend over the kernel C ABI. One backend
@@ -119,12 +127,8 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 		}
 	}
 
-	tok := newCStr(k.cfg.Token)
-	defer tok.free()
-	if err := call(func() C.KernelStatusCode {
-		return C.kernel_session_config_set_auth_pat(cfg, tok.c)
-	}); err != nil {
-		return fmt.Errorf("kernel: set_auth_pat: %w", toConnError(err))
+	if err := k.setAuth(cfg); err != nil {
+		return err
 	}
 
 	// TLS: crypto/tls's InsecureSkipVerify accepts any server cert, so relax both
@@ -187,10 +191,123 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 	// The C ABI exposes no formatted session-id accessor; use the handle pointer
 	// as a stable per-conn id for logging / telemetry correlation.
 	k.sessionID = fmt.Sprintf("kernel-%p", sess)
+
+	// Initial namespace: the kernel C ABI has no catalog/schema config setter, so
+	// select it post-connect with USE CATALOG / USE SCHEMA (the OSS ODBC driver's
+	// approach). A failure here means the session is not in the requested namespace
+	// — a correctness precondition, like Thrift's InitialNamespace — so fail the
+	// connect and close the session we just opened (the connector does not call
+	// CloseSession on an OpenSession error).
+	if err := k.applyInitialNamespace(ctx); err != nil {
+		// Close the session we just opened, routing through call() so a failed close
+		// is logged (via lastError's Warn) rather than silently discarded — mirroring
+		// CloseSession. The namespace error is authoritative and returned as-is.
+		if closeErr := call(func() C.KernelStatusCode { return C.kernel_session_close(sess) }); closeErr != nil {
+			klog("close after initial-namespace failure also failed: %v", closeErr)
+		}
+		k.session = nil
+		k.valid = false
+		return err
+	}
+
 	klog("OpenSession OK session=%s", k.sessionID)
 	return nil
 }
 
+// setAuth applies the resolved auth form to the session config via exactly one
+// kernel_session_config_set_auth_* call. PAT and M2M are plain value setters; U2M
+// records the client id / redirect port / scopes and the kernel owns the browser
+// (PKCE) flow, started when the session opens. Empty string args are passed as NULL
+// so the kernel applies its own defaults (e.g. U2M's public client / default port).
+func (k *KernelBackend) setAuth(cfg *C.KernelSessionConfig) error {
+	switch k.cfg.Auth.Mode {
+	case AuthM2M:
+		clientID := newCStr(k.cfg.Auth.ClientID)
+		defer clientID.free()
+		secret := newCStr(k.cfg.Auth.ClientSecret)
+		defer secret.free()
+		if err := call(func() C.KernelStatusCode {
+			return C.kernel_session_config_set_auth_m2m(cfg, clientID.c, secret.c)
+		}); err != nil {
+			return fmt.Errorf("kernel: set_auth_m2m: %w", toConnError(err))
+		}
+	case AuthU2M:
+		// client id / scopes are optional: NULL when empty lets the kernel use its
+		// public client / default scopes. We pass Go's cloud-inferred client id when
+		// set, so the kernel uses the same client id the Thrift path would.
+		clientID := newCStrOrNull(k.cfg.Auth.ClientID)
+		defer clientID.free()
+		scopes := newCStrOrNull(joinScopes(k.cfg.Auth.Scopes))
+		defer scopes.free()
+		if err := call(func() C.KernelStatusCode {
+			return C.kernel_session_config_set_auth_u2m(cfg, clientID.c, C.uint16_t(k.cfg.Auth.RedirectPort), scopes.c)
+		}); err != nil {
+			return fmt.Errorf("kernel: set_auth_u2m: %w", toConnError(err))
+		}
+	default: // AuthPAT
+		tok := newCStr(k.cfg.Auth.Token)
+		defer tok.free()
+		if err := call(func() C.KernelStatusCode {
+			return C.kernel_session_config_set_auth_pat(cfg, tok.c)
+		}); err != nil {
+			return fmt.Errorf("kernel: set_auth_pat: %w", toConnError(err))
+		}
+	}
+	return nil
+}
+
+// joinScopes renders U2M scopes as the comma-separated form the kernel U2M setter
+// expects. Empty (no scopes) yields "" so setAuth passes NULL and the kernel
+// applies its default scope set.
+func joinScopes(scopes []string) string {
+	return strings.Join(scopes, ",")
+}
+
+// trySetAuth allocates a throwaway session config, applies auth to it, and frees
+// it — a test seam so TestSetAuthByMode can exercise the real cgo setter path
+// without putting cgo in a _test.go file (which Go forbids). Not used in
+// production. Returns the setter error (nil on success).
+func trySetAuth(auth Auth) error {
+	var cfg *C.KernelSessionConfig
+	if err := call(func() C.KernelStatusCode { return C.kernel_session_config_new(&cfg) }); err != nil {
+		return fmt.Errorf("config_new: %w", err)
+	}
+	defer C.kernel_session_config_free(cfg)
+	k := &KernelBackend{cfg: Config{Auth: auth}}
+	return k.setAuth(cfg)
+}
+
+// applyInitialNamespace runs USE CATALOG / USE SCHEMA to select the configured
+// initial namespace, since the kernel C ABI exposes no catalog/schema setter.
+// Identifiers are backtick-quoted (quoteIdent) so arbitrary names are safe. A
+// no-op when neither is set.
+func (k *KernelBackend) applyInitialNamespace(ctx context.Context) error {
+	if k.cfg.Catalog != "" {
+		if err := k.runNamespaceStmt(ctx, "USE CATALOG "+quoteIdent(k.cfg.Catalog)); err != nil {
+			return fmt.Errorf("kernel: set initial catalog %q: %w", k.cfg.Catalog, err)
+		}
+	}
+	if k.cfg.Schema != "" {
+		if err := k.runNamespaceStmt(ctx, "USE SCHEMA "+quoteIdent(k.cfg.Schema)); err != nil {
+			return fmt.Errorf("kernel: set initial schema %q: %w", k.cfg.Schema, err)
+		}
+	}
+	return nil
+}
+
+// runNamespaceStmt executes a single side-effecting statement (USE …) and closes
+// the operation. USE produces no rows, so the result stream is not read. execute
+// always returns a non-nil Operation (the Backend contract), so it is closed on
+// both the success and error paths; the execute error is authoritative.
+func (k *KernelBackend) runNamespaceStmt(ctx context.Context, sql string) error {
+	op, err := k.execute(ctx, backend.ExecRequest{Query: sql})
+	_, closeErr := op.Close(ctx)
+	if err != nil {
+		return err
+	}
+	return closeErr
+}
+
 // CloseSession tears down the server-side session. Best-effort: the kernel's
 // close is async (see the C header), so an error is logged, not hard-failed.
 func (k *KernelBackend) CloseSession(ctx context.Context) error {
diff --git a/internal/backend/kernel/cgo.go b/internal/backend/kernel/cgo.go
index 852a467d..1b7f4214 100644
--- a/internal/backend/kernel/cgo.go
+++ b/internal/backend/kernel/cgo.go
@@ -175,6 +175,18 @@ type cStr struct{ c *C.char }
 
 func newCStr(s string) cStr { return cStr{c: C.CString(s)} }
 
+// newCStrOrNull is like newCStr but yields a NULL C pointer for an empty string,
+// for kernel args whose "unset" sentinel is NULL (e.g. the optional U2M client id /
+// scopes, where NULL selects the kernel's own default). C.CString("") would instead
+// pass a non-NULL pointer to an empty string, which the kernel treats as a real
+// (empty) value rather than "use the default".
+func newCStrOrNull(s string) cStr {
+	if s == "" {
+		return cStr{c: nil}
+	}
+	return cStr{c: C.CString(s)}
+}
+
 func (s cStr) free() {
 	if s.c != nil {
 		C.free(unsafe.Pointer(s.c))
diff --git a/internal/backend/kernel/kernel_test.go b/internal/backend/kernel/kernel_test.go
index 973dce99..04be61b6 100644
--- a/internal/backend/kernel/kernel_test.go
+++ b/internal/backend/kernel/kernel_test.go
@@ -11,6 +11,39 @@ import (
 	"github.com/databricks/databricks-sql-go/internal/backend"
 )
 
+// setAuth maps each Auth mode to exactly one kernel_session_config_set_auth_*
+// value-setter. These are pure config setters (no network), so we can assert the
+// call succeeds against a freshly allocated config for every mode — exercising the
+// real cgo path (arg marshaling, NULL-for-empty on the optional U2M args) end to
+// end via the trySetAuth test helper (cgo cannot be used directly in a _test.go
+// file). A failure here means the mode→setter wiring or the C signature drifted.
+func TestSetAuthByMode(t *testing.T) {
+	cases := []struct {
+		name string
+		auth Auth
+	}{
+		{"PAT", Auth{Mode: AuthPAT, Token: "dapi-x"}},
+		{"M2M", Auth{Mode: AuthM2M, ClientID: "cid", ClientSecret: "sec"}},
+		// "U2M full" populates Scopes/RedirectPort, which no production path sets today
+		// (resolveKernelAuth sources only the client id — see kernel.Auth docs). It is
+		// kept deliberately to pin the marshalling of those optional set_auth_u2m args
+		// (joinScopes + uint16 port), so the dormant wiring stays correct for a future
+		// U2M scopes/port option.
+		{"U2M full", Auth{Mode: AuthU2M, ClientID: "u2m-cid", Scopes: []string{"sql", "offline_access"}, RedirectPort: 8030}},
+		// U2M with everything defaulted (the production shape): empty client id / no
+		// scopes / port 0 must pass NULL / 0 so the kernel applies its own defaults
+		// (exercises newCStrOrNull).
+		{"U2M defaults", Auth{Mode: AuthU2M}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			if err := trySetAuth(c.auth); err != nil {
+				t.Errorf("setAuth(%s) = %v, want nil", c.name, err)
+			}
+		})
+	}
+}
+
 // The pure error-classifier tests (TestIsBadConnection, TestIsSessionFatal,
 // TestToConnError, TestToStatementErrorNeverBadConn) live in the untagged
 // errors_classify_test.go so they run under CGO_ENABLED=0. The tests below need a
diff --git a/internal/backend/kernel/namespace.go b/internal/backend/kernel/namespace.go
new file mode 100644
index 00000000..b1f907c3
--- /dev/null
+++ b/internal/backend/kernel/namespace.go
@@ -0,0 +1,17 @@
+package kernel
+
+// This file is intentionally NOT behind the `cgo && databricks_kernel` build tag.
+// quoteIdent is pure Go (no cgo) and is unit-tested in the default CGO_ENABLED=0
+// matrix; OpenSession (tagged) calls it to build the USE CATALOG / USE SCHEMA
+// statements that select the initial namespace.
+
+import "strings"
+
+// quoteIdent renders name as a backtick-quoted Databricks SQL identifier, doubling
+// any embedded backtick. The kernel C ABI exposes no catalog/schema config setter,
+// so the initial namespace is applied post-connect by running USE CATALOG / USE
+// SCHEMA (the same workaround the OSS ODBC driver uses); quoting makes those
+// statements injection-safe for arbitrary identifier text.
+func quoteIdent(name string) string {
+	return "`" + strings.ReplaceAll(name, "`", "``") + "`"
+}
diff --git a/internal/backend/kernel/namespace_test.go b/internal/backend/kernel/namespace_test.go
new file mode 100644
index 00000000..bd005d67
--- /dev/null
+++ b/internal/backend/kernel/namespace_test.go
@@ -0,0 +1,25 @@
+package kernel
+
+import "testing"
+
+func TestQuoteIdent(t *testing.T) {
+	cases := []struct {
+		in   string
+		want string
+	}{
+		{"main", "`main`"},
+		{"my_schema", "`my_schema`"},
+		{"", "``"},
+		{"has space", "`has space`"},
+		// A backtick in the identifier is doubled so it can't terminate the quote.
+		{"a`b", "`a``b`"},
+		// One backtick → doubled to two, wrapped in two more → four total.
+		{"`", "````"},
+		{"weird`;DROP", "`weird``;DROP`"},
+	}
+	for _, c := range cases {
+		if got := quoteIdent(c.in); got != c.want {
+			t.Errorf("quoteIdent(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}
diff --git a/internal/backend/thrift/backend.go b/internal/backend/thrift/backend.go
index 444130eb..0f934ccf 100644
--- a/internal/backend/thrift/backend.go
+++ b/internal/backend/thrift/backend.go
@@ -74,13 +74,10 @@ func (b *Backend) OpenSession(ctx context.Context) error {
 		schemaName = cli_service.TIdentifierPtr(cli_service.TIdentifier(b.cfg.Schema))
 	}
 
-	sessionParams := make(map[string]string)
-	for k, v := range b.cfg.SessionParams {
-		sessionParams[k] = v
-	}
-	if b.cfg.EnableMetricViewMetadata {
-		sessionParams["spark.sql.thriftserver.metadata.metricview.enabled"] = "true"
-	}
+	// EffectiveSessionParams folds any option-derived confs (e.g. metric-view
+	// metadata) into the user's SessionParams, backend-neutrally, so the kernel
+	// backend sends the identical confs without duplicating the derivation.
+	sessionParams := b.cfg.EffectiveSessionParams()
 
 	protocolVersion := int64(b.cfg.ThriftProtocolVersion)
 
diff --git a/internal/config/config.go b/internal/config/config.go
index 12b59395..58c92123 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -50,6 +50,30 @@ type Config struct {
 	ThriftDebugClientProtocol bool
 }
 
+// MetricViewMetadataConfKey is the server session conf that enables metric-view
+// metadata (WithEnableMetricViewMetadata). Despite the "thriftserver" in its name
+// — a historical server-side name — it is an ordinary Spark SQL session conf the
+// server honors regardless of client transport, so both the Thrift and kernel
+// backends send the identical key/value. Defined once here so no backend hardcodes
+// the literal (see EffectiveSessionParams).
+const MetricViewMetadataConfKey = "spark.sql.thriftserver.metadata.metricview.enabled"
+
+// EffectiveSessionParams returns the session confs to send to the server: the
+// user-supplied SessionParams plus any conf derived from a higher-level option
+// (currently metric-view metadata). Both backends call this, so the derivation is
+// backend-neutral — neither special-cases the option nor hardcodes the conf
+// literal. The returned map is always a fresh copy the caller may mutate freely.
+func (c *Config) EffectiveSessionParams() map[string]string {
+	params := make(map[string]string, len(c.SessionParams)+1)
+	for k, v := range c.SessionParams {
+		params[k] = v
+	}
+	if c.EnableMetricViewMetadata {
+		params[MetricViewMetadataConfKey] = "true"
+	}
+	return params
+}
+
 // ToEndpointURL generates the endpoint URL from Config that a Thrift client will connect to
 func (c *Config) ToEndpointURL() (string, error) {
 	var userInfo string
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index 4c70e993..a4b9b54f 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -645,6 +645,48 @@ func TestParseConfig(t *testing.T) {
 	}
 }
 
+func TestEffectiveSessionParams(t *testing.T) {
+	t.Run("no metric view leaves params unchanged", func(t *testing.T) {
+		c := &Config{UserConfig: UserConfig{SessionParams: map[string]string{"QUERY_TAGS": "a:1"}}}
+		got := c.EffectiveSessionParams()
+		if len(got) != 1 || got["QUERY_TAGS"] != "a:1" {
+			t.Errorf("EffectiveSessionParams() = %v, want only QUERY_TAGS", got)
+		}
+		if _, ok := got[MetricViewMetadataConfKey]; ok {
+			t.Errorf("metric-view conf must not be set when EnableMetricViewMetadata is false")
+		}
+	})
+	t.Run("metric view adds the derived conf", func(t *testing.T) {
+		c := &Config{UserConfig: UserConfig{
+			SessionParams:            map[string]string{"QUERY_TAGS": "a:1"},
+			EnableMetricViewMetadata: true,
+		}}
+		got := c.EffectiveSessionParams()
+		if got[MetricViewMetadataConfKey] != "true" {
+			t.Errorf("metric-view conf = %q, want \"true\"", got[MetricViewMetadataConfKey])
+		}
+		if got["QUERY_TAGS"] != "a:1" {
+			t.Errorf("user SessionParams must be preserved, got %v", got)
+		}
+	})
+	t.Run("returns a fresh copy the caller may mutate", func(t *testing.T) {
+		orig := map[string]string{"QUERY_TAGS": "a:1"}
+		c := &Config{UserConfig: UserConfig{SessionParams: orig}}
+		got := c.EffectiveSessionParams()
+		got["QUERY_TAGS"] = "mutated"
+		if orig["QUERY_TAGS"] != "a:1" {
+			t.Errorf("mutating the result mutated the source SessionParams: %v", orig)
+		}
+	})
+	t.Run("nil SessionParams with metric view", func(t *testing.T) {
+		c := &Config{UserConfig: UserConfig{EnableMetricViewMetadata: true}}
+		got := c.EffectiveSessionParams()
+		if got[MetricViewMetadataConfKey] != "true" {
+			t.Errorf("metric-view conf must be set even with nil SessionParams, got %v", got)
+		}
+	})
+}
+
 func TestUserConfig_DeepCopy(t *testing.T) {
 	t.Run("copy empty config", func(t *testing.T) {
 		cfg := UserConfig{}
diff --git a/kernel_backend.go b/kernel_backend.go
index 412ffe78..9c8092ce 100644
--- a/kernel_backend.go
+++ b/kernel_backend.go
@@ -16,11 +16,11 @@ import (
 // connection config, so the user-facing options are unchanged — only the routing
 // differs. The public API adds nothing beyond WithUseKernel.
 func newKernelBackend(_ context.Context, cfg *config.Config) (backend.Backend, error) {
-	// Reject options the kernel path can't honor yet + resolve the PAT. The
+	// Reject options the kernel path can't honor yet + resolve the auth form. The
 	// validation is pure Go and lives in kernel_config.go (untagged) so its tests —
 	// including the exhaustiveness guard against a dropped Config field — run in the
-	// default CGO_ENABLED=0 build.
-	token, err := validateKernelConfig(cfg)
+	// default CGO_ENABLED=0 build. It returns kernel.Auth directly.
+	kauth, err := validateKernelConfig(cfg)
 	if err != nil {
 		return nil, err
 	}
@@ -29,13 +29,18 @@ func newKernelBackend(_ context.Context, cfg *config.Config) (backend.Backend, e
 		Host:        cfg.Host,
 		HTTPPath:    cfg.HTTPPath,
 		WarehouseID: cfg.WarehouseID,
-		Token:       token,
+		Auth:        kauth,
 		Location:    cfg.Location,
-		// Session confs (STATEMENT_TIMEOUT, QUERY_TAGS, TIMEZONE, …) — the same
-		// SessionParams map the Thrift backend forwards, so they flow to the
-		// server identically with no per-backend translation. SPOG org routing
+		// Initial namespace: no kernel config setter, so the kernel backend applies
+		// these post-connect via USE CATALOG / USE SCHEMA.
+		Catalog: cfg.Catalog,
+		Schema:  cfg.Schema,
+		// Session confs (STATEMENT_TIMEOUT, QUERY_TAGS, TIMEZONE, metric-view, …) —
+		// the same effective params the Thrift backend forwards (user SessionParams
+		// plus any option-derived conf like metric-view metadata), so they flow to
+		// the server identically with no per-backend translation. SPOG org routing
 		// rides in HTTPPath's ?o= and is parsed kernel-side.
-		SessionConf: cfg.SessionParams,
+		SessionConf: cfg.EffectiveSessionParams(),
 	}
 	// TLS: the driver honors TLSConfig only for InsecureSkipVerify (see
 	// internal/client), so map exactly that knob to the kernel.
diff --git a/kernel_backend_test.go b/kernel_backend_test.go
index eb3244c4..a6a272da 100644
--- a/kernel_backend_test.go
+++ b/kernel_backend_test.go
@@ -5,6 +5,7 @@ package dbsql
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/databricks/databricks-sql-go/internal/config"
 )
@@ -32,7 +33,7 @@ func TestNewKernelBackend(t *testing.T) {
 
 	t.Run("validation error propagates", func(t *testing.T) {
 		c := base()
-		c.Catalog = "main" // rejected by validateKernelConfig
+		c.QueryTimeout = 30 * time.Second // rejected by validateKernelConfig
 		if _, err := newKernelBackend(context.Background(), c); err == nil {
 			t.Error("newKernelBackend should propagate the validation error")
 		}
diff --git a/kernel_config.go b/kernel_config.go
index b07f52ec..8ed03799 100644
--- a/kernel_config.go
+++ b/kernel_config.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/databricks/databricks-sql-go/auth/noop"
 	"github.com/databricks/databricks-sql-go/auth/pat"
+	"github.com/databricks/databricks-sql-go/internal/backend/kernel"
 	"github.com/databricks/databricks-sql-go/internal/config"
 )
 
@@ -18,68 +19,43 @@ import (
 // validateKernelConfig enforces the kernel backend's "nothing silently ignored"
 // contract: it rejects every option the kernel path can't yet honor with a clear
 // error (rather than dropping it, which would behave differently than Thrift) and
-// resolves the PAT the kernel authenticates with. On success it returns the token
-// to use. Options it does NOT reject are either forwarded by newKernelBackend or
+// resolves the kernel.Auth descriptor the kernel authenticates with (PAT, or OAuth
+// M2M/U2M). Options it does NOT reject are either forwarded by newKernelBackend or
 // intentionally accepted-but-inert (documented in doc.go and asserted by
-// TestKernelConfigFieldsClassified).
-func validateKernelConfig(cfg *config.Config) (token string, err error) {
-	// Initial namespace (WithInitialNamespace): no kernel C-ABI setter yet, so the
-	// session would run in the default namespace and unqualified names would
-	// resolve differently than Thrift.
-	if cfg.Catalog != "" || cfg.Schema != "" {
-		return "", errors.New("databricks: WithInitialNamespace (catalog/schema) is not yet supported by the kernel backend; " +
-			"omit it or use the default (Thrift) backend")
-	}
-	// EnableMetricViewMetadata: maps to a server session conf we want to route
-	// backend-neutrally rather than duplicate here.
-	if cfg.EnableMetricViewMetadata {
-		return "", errors.New("databricks: WithEnableMetricViewMetadata is not yet supported by the kernel backend; " +
-			"omit it or use the default (Thrift) backend")
-	}
+// TestKernelConfigFieldsClassified). It returns kernel.Auth directly (no dbsql-side
+// duplicate) — kernel's auth types are in an untagged file, so this untagged,
+// default-build code can build them without pulling in cgo.
+func validateKernelConfig(cfg *config.Config) (kernel.Auth, error) {
+	// Initial namespace (WithInitialNamespace) is forwarded, not rejected: the
+	// kernel C ABI has no catalog/schema setter, so KernelBackend.OpenSession
+	// selects it post-connect with USE CATALOG / USE SCHEMA (the OSS ODBC driver's
+	// workaround). No per-backend handling needed here.
+	// EnableMetricViewMetadata is forwarded, not rejected: config.EffectiveSessionParams
+	// folds its server conf (spark.sql.thriftserver.metadata.metricview.enabled=true)
+	// into SessionConf backend-neutrally, so the kernel path sends the identical conf
+	// the Thrift path does. No per-backend handling needed here.
 	// Port / Protocol: the kernel C ABI takes only a bare host and connects over
 	// https:443; it has no port or scheme setter. The Thrift path honors a custom
 	// port/scheme via ToEndpointURL, so a non-default value here would be silently
 	// ignored on the kernel path (it would just hit 443) — reject it instead, per
 	// the "nothing silently ignored" contract. Defaults (https/443) are fine.
 	if cfg.Protocol != "" && cfg.Protocol != "https" {
-		return "", errors.New("databricks: a non-https protocol is not supported by the kernel backend " +
+		return kernel.Auth{}, errors.New("databricks: a non-https protocol is not supported by the kernel backend " +
 			"(it connects over https); use the default (Thrift) backend")
 	}
 	if cfg.Port != 0 && cfg.Port != 443 {
-		return "", errors.New("databricks: a non-default port (WithPort) is not supported by the kernel backend " +
+		return kernel.Auth{}, errors.New("databricks: a non-default port (WithPort) is not supported by the kernel backend " +
 			"(it connects on 443); omit it or use the default (Thrift) backend")
 	}
-	// Auth: the kernel backend authenticates with a PAT only. Any other
-	// authenticator sets cfg.Authenticator but leaves cfg.AccessToken empty, so an
-	// empty PAT would reach the kernel and fail with an opaque Unauthenticated
-	// error. Reject it here so the failure names the cause.
-	token = cfg.AccessToken
-	switch a := cfg.Authenticator.(type) {
-	case nil, *noop.NoopAuth:
-		// No explicit authenticator — token comes from cfg.AccessToken (may be
-		// empty; caught below).
-	case *pat.PATAuth:
-		// WithAccessToken sets both cfg.AccessToken and this authenticator, but
-		// WithAuthenticator(&pat.PATAuth{...}) sets only the authenticator and leaves
-		// cfg.AccessToken empty. Take the token from the authenticator when
-		// cfg.AccessToken didn't carry it, so both PAT paths work.
-		if token == "" {
-			token = a.AccessToken
-		}
-	default:
-		return "", errors.New("databricks: only personal access token (WithAccessToken) auth is supported by the kernel backend; " +
-			"OAuth (M2M/U2M), token-provider, external/static, and federated authenticators are not yet supported — " +
-			"use PAT or the default (Thrift) backend")
-	}
-	if token == "" {
-		return "", errors.New("databricks: the kernel backend requires a personal access token; " +
-			"set one with WithAccessToken (or a *pat.PATAuth via WithAuthenticator)")
+	kauth, err := resolveKernelAuth(cfg)
+	if err != nil {
+		return kernel.Auth{}, err
 	}
 	// WithTimeout maps to a per-statement server timeout on Thrift
 	// (TExecuteStatementReq.QueryTimeout); the kernel C ABI exposes no equivalent,
 	// so reject it rather than run the query with no server-side timeout.
 	if cfg.QueryTimeout > 0 {
-		return "", errors.New("databricks: WithTimeout (server query timeout) is not yet supported by the kernel backend; " +
+		return kernel.Auth{}, errors.New("databricks: WithTimeout (server query timeout) is not yet supported by the kernel backend; " +
 			"omit it or use the default (Thrift) backend")
 	}
 	// WithRetries(-1) explicitly disables retries, but the kernel retries
@@ -87,8 +63,60 @@ func validateKernelConfig(cfg *config.Config) (token string, err error) {
 	// would be silently violated. Reject it. Positive/default RetryMax is fine: the
 	// kernel provides retries (just not user-tunable), documented in doc.go.
 	if cfg.RetryMax < 0 {
-		return "", errors.New("databricks: disabling retries via WithRetries is not supported by the kernel backend " +
+		return kernel.Auth{}, errors.New("databricks: disabling retries via WithRetries is not supported by the kernel backend " +
 			"(the kernel retries internally); omit it or use the default (Thrift) backend")
 	}
-	return token, nil
+	return kauth, nil
+}
+
+// resolveKernelAuth picks the kernel auth form from the config. The kernel backend
+// drives the kernel's own OAuth flow from raw credentials (mirroring pyo3/napi and
+// the Node/Python kernel bindings) rather than reusing the Go authenticator's
+// Authenticate method. It reads those credentials off cfg.Authenticator — the
+// single source of truth for auth, so the last WithX option applied wins for both
+// backends (matching Thrift's last-writer-wins on cfg.Authenticator). The M2M/U2M
+// authenticator types are unexported, so it asserts the small
+// kernel.M2MCredentialsProvider / kernel.U2MCredentialsProvider interfaces they
+// satisfy structurally:
+//   - implements M2MCredentialsProvider → M2M (client id + secret)
+//   - implements U2MCredentialsProvider → U2M (browser/PKCE; kernel-owned flow)
+//   - PAT / nil / noop                  → PAT (from AccessToken or a *pat.PATAuth)
+//   - anything else                     → rejected loudly (token-provider / external
+//     / static / federated), so the failure names the cause instead of surfacing as
+//     an opaque Unauthenticated.
+func resolveKernelAuth(cfg *config.Config) (kernel.Auth, error) {
+	switch a := cfg.Authenticator.(type) {
+	case kernel.M2MCredentialsProvider:
+		clientID, clientSecret := a.M2MCredentials()
+		return kernel.Auth{Mode: kernel.AuthM2M, ClientID: clientID, ClientSecret: clientSecret}, nil
+	case kernel.U2MCredentialsProvider:
+		// Go sources only the client id for U2M; kernel.Auth.Scopes / RedirectPort
+		// are left zero so setAuth passes the kernel's defaults. Go exposes no
+		// user-facing option for U2M scopes or redirect port on either backend today
+		// (the native Thrift path hardcodes both), so there is nothing to forward —
+		// but the kernel.Auth fields + setAuth wiring model the full set_auth_u2m
+		// surface for if/when such an option is added (see kernel.Auth docs).
+		return kernel.Auth{Mode: kernel.AuthU2M, ClientID: a.U2MClientID()}, nil
+	case nil, *noop.NoopAuth, *pat.PATAuth:
+		// PAT (or no explicit authenticator). WithAccessToken sets both
+		// cfg.AccessToken and a *pat.PATAuth, but WithAuthenticator(&pat.PATAuth{...})
+		// sets only the authenticator and leaves cfg.AccessToken empty — so take the
+		// token from the authenticator when cfg.AccessToken didn't carry it.
+		token := cfg.AccessToken
+		if token == "" {
+			if p, ok := a.(*pat.PATAuth); ok {
+				token = p.AccessToken
+			}
+		}
+		if token == "" {
+			return kernel.Auth{}, errors.New("databricks: the kernel backend requires a personal access token; " +
+				"set one with WithAccessToken (or a *pat.PATAuth via WithAuthenticator)")
+		}
+		return kernel.Auth{Mode: kernel.AuthPAT, Token: token}, nil
+	default:
+		return kernel.Auth{}, errors.New("databricks: this authenticator is not supported by the kernel backend; " +
+			"PAT (WithAccessToken) and OAuth M2M/U2M (WithClientCredentials / authType) are supported, but " +
+			"token-provider, external/static, and federated authenticators are not — " +
+			"use one of those or the default (Thrift) backend")
+	}
 }
diff --git a/kernel_config_test.go b/kernel_config_test.go
index b82de306..9b718f77 100644
--- a/kernel_config_test.go
+++ b/kernel_config_test.go
@@ -7,15 +7,33 @@ import (
 	"time"
 
 	"github.com/databricks/databricks-sql-go/auth/pat"
+	"github.com/databricks/databricks-sql-go/internal/backend/kernel"
 	"github.com/databricks/databricks-sql-go/internal/config"
 )
 
-// nonPATAuth stands in for any non-PAT authenticator (OAuth / token-provider /
-// external / federated) — the kernel backend must reject it.
+// nonPATAuth stands in for any non-PAT, non-OAuth authenticator (token-provider /
+// external / federated) — the kernel backend must reject it. It implements neither
+// auth.M2MCredentialsProvider nor auth.U2MCredentialsProvider.
 type nonPATAuth struct{}
 
 func (nonPATAuth) Authenticate(*http.Request) error { return nil }
 
+// fakeM2MAuth / fakeU2MAuth implement the credential-provider interfaces the kernel
+// backend asserts on. Used instead of the real m2m/u2m authenticators in unit tests
+// because the real u2m.NewAuthenticator does live OIDC discovery at construction
+// (needs a resolvable host); the kernel only needs the interface, so a fake is both
+// sufficient and hermetic. The real authenticators' method implementations are
+// trivial field returns (verified in auth/oauth/{m2m,u2m}).
+type fakeM2MAuth struct{ id, secret string }
+
+func (fakeM2MAuth) Authenticate(*http.Request) error   { return nil }
+func (f fakeM2MAuth) M2MCredentials() (string, string) { return f.id, f.secret }
+
+type fakeU2MAuth struct{ id string }
+
+func (fakeU2MAuth) Authenticate(*http.Request) error { return nil }
+func (f fakeU2MAuth) U2MClientID() string            { return f.id }
+
 func baseKernelConfig() *config.Config {
 	c := config.WithDefaults()
 	c.Host = "h.databricks.com"
@@ -37,27 +55,38 @@ func TestValidateKernelConfig(t *testing.T) {
 		}
 	})
 
-	t.Run("catalog rejected", func(t *testing.T) {
+	t.Run("catalog accepted (applied post-connect via USE CATALOG)", func(t *testing.T) {
 		c := baseKernelConfig()
 		c.Catalog = "main"
-		if _, err := validateKernelConfig(c); err == nil {
-			t.Error("expected an error when a catalog is set")
+		if _, err := validateKernelConfig(c); err != nil {
+			t.Errorf("initial catalog is now forwarded (USE CATALOG post-connect), want no error, got %v", err)
 		}
 	})
 
-	t.Run("schema rejected", func(t *testing.T) {
+	t.Run("schema accepted (applied post-connect via USE SCHEMA)", func(t *testing.T) {
 		c := baseKernelConfig()
 		c.Schema = "sys"
-		if _, err := validateKernelConfig(c); err == nil {
-			t.Error("expected an error when a schema is set")
+		if _, err := validateKernelConfig(c); err != nil {
+			t.Errorf("initial schema is now forwarded (USE SCHEMA post-connect), want no error, got %v", err)
 		}
 	})
 
-	t.Run("metric view rejected", func(t *testing.T) {
+	t.Run("metric view accepted (folded into session conf)", func(t *testing.T) {
 		c := baseKernelConfig()
 		c.EnableMetricViewMetadata = true
-		if _, err := validateKernelConfig(c); err == nil {
-			t.Error("expected an error when metric-view metadata is enabled")
+		if _, err := validateKernelConfig(c); err != nil {
+			t.Errorf("metric-view metadata is now forwarded backend-neutrally, want no error, got %v", err)
+		}
+	})
+
+	t.Run("PAT resolves to a PAT auth descriptor", func(t *testing.T) {
+		c := baseKernelConfig() // AccessToken = "dapi-x"
+		a, err := validateKernelConfig(c)
+		if err != nil {
+			t.Fatalf("PAT should validate, got %v", err)
+		}
+		if a.Mode != kernel.AuthPAT || a.Token != "dapi-x" {
+			t.Errorf("auth = %+v, want mode=PAT token=dapi-x", a)
 		}
 	})
 
@@ -65,12 +94,61 @@ func TestValidateKernelConfig(t *testing.T) {
 		c := baseKernelConfig()
 		c.AccessToken = ""
 		c.Authenticator = &pat.PATAuth{AccessToken: "dapi-y"}
-		tok, err := validateKernelConfig(c)
+		a, err := validateKernelConfig(c)
 		if err != nil {
 			t.Fatalf("PAT via WithAuthenticator should validate, got %v", err)
 		}
-		if tok != "dapi-y" {
-			t.Errorf("token = %q, want dapi-y (sourced from the authenticator)", tok)
+		if a.Mode != kernel.AuthPAT || a.Token != "dapi-y" {
+			t.Errorf("auth = %+v, want mode=PAT token=dapi-y (sourced from the authenticator)", a)
+		}
+	})
+
+	t.Run("OAuth M2M resolves to an M2M descriptor", func(t *testing.T) {
+		c := baseKernelConfig()
+		c.AccessToken = ""
+		// An M2M authenticator is the single source of truth; resolveKernelAuth reads
+		// the creds off it via the auth.M2MCredentialsProvider interface.
+		c.Authenticator = fakeM2MAuth{id: "cid", secret: "sec"}
+		a, err := validateKernelConfig(c)
+		if err != nil {
+			t.Fatalf("M2M should validate, got %v", err)
+		}
+		if a.Mode != kernel.AuthM2M || a.ClientID != "cid" || a.ClientSecret != "sec" {
+			t.Errorf("auth = %+v, want mode=M2M clientID=cid clientSecret=sec", a)
+		}
+	})
+
+	t.Run("OAuth U2M resolves to a U2M descriptor", func(t *testing.T) {
+		c := baseKernelConfig()
+		c.AccessToken = ""
+		// A U2M authenticator is the single source of truth; resolveKernelAuth reads
+		// its (cloud-inferred) client id via the auth.U2MCredentialsProvider interface.
+		c.Authenticator = fakeU2MAuth{id: "databricks-sql-connector"}
+		a, err := validateKernelConfig(c)
+		if err != nil {
+			t.Fatalf("U2M should validate, got %v", err)
+		}
+		if a.Mode != kernel.AuthU2M || a.ClientID != "databricks-sql-connector" {
+			t.Errorf("auth = %+v, want mode=U2M clientID=databricks-sql-connector", a)
+		}
+	})
+
+	t.Run("last-applied auth wins: M2M then PAT resolves to PAT", func(t *testing.T) {
+		// Regression for the auth-mode divergence: cfg.Authenticator is the single
+		// source of truth, so setting an M2M authenticator and then a PAT (a later
+		// WithAccessToken) must resolve to PAT on the kernel path — matching Thrift's
+		// last-writer-wins on cfg.Authenticator. (Previously a parallel OAuth carrier
+		// field could keep the kernel on M2M while Thrift used PAT.)
+		c := baseKernelConfig()
+		c.Authenticator = fakeM2MAuth{id: "cid", secret: "sec"} // earlier
+		c.Authenticator = &pat.PATAuth{AccessToken: "dapi-z"}   // later wins
+		c.AccessToken = "dapi-z"
+		a, err := validateKernelConfig(c)
+		if err != nil {
+			t.Fatalf("PAT (last applied) should validate, got %v", err)
+		}
+		if a.Mode != kernel.AuthPAT || a.Token != "dapi-z" {
+			t.Errorf("auth = %+v, want mode=PAT token=dapi-z (last-applied wins)", a)
 		}
 	})
 
@@ -83,11 +161,12 @@ func TestValidateKernelConfig(t *testing.T) {
 		}
 	})
 
-	t.Run("non-PAT authenticator rejected", func(t *testing.T) {
+	t.Run("non-PAT/non-OAuth authenticator rejected", func(t *testing.T) {
 		c := baseKernelConfig()
+		c.AccessToken = ""
 		c.Authenticator = nonPATAuth{}
 		if _, err := validateKernelConfig(c); err == nil {
-			t.Error("expected an error for a non-PAT authenticator")
+			t.Error("expected an error for a token-provider/external/federated authenticator")
 		}
 	})
 
@@ -154,20 +233,23 @@ var kernelConfigFieldDisposition = map[string]string{
 	"Host":          "forwarded",
 	"HTTPPath":      "forwarded",
 	"WarehouseID":   "forwarded",
-	"AccessToken":   "forwarded", // as the resolved PAT (kc.Token)
-	"Authenticator": "forwarded", // PAT authenticator resolved to the token
+	"AccessToken":   "forwarded", // as the resolved PAT (kc.Auth.Token)
+	"Authenticator": "forwarded", // resolved to the auth descriptor (PAT/M2M/U2M)
 	"Location":      "forwarded",
 	"SessionParams": "forwarded",
 	"UseKernel":     "forwarded", // the routing flag itself
+	// Folded into SessionConf by config.EffectiveSessionParams (metric-view conf),
+	// sent identically on both backends.
+	"EnableMetricViewMetadata": "forwarded",
+	// Applied post-connect via USE CATALOG / USE SCHEMA (no kernel config setter).
+	"Catalog": "forwarded",
+	"Schema":  "forwarded",
 
 	// Rejected loudly by validateKernelConfig.
-	"Catalog":                  "rejected",
-	"Schema":                   "rejected",
-	"EnableMetricViewMetadata": "rejected",
-	"QueryTimeout":             "rejected", // when > 0 (WithTimeout)
-	"RetryMax":                 "rejected", // when < 0 (disable retries)
-	"Protocol":                 "rejected", // kernel is https-only; non-default rejected
-	"Port":                     "rejected", // kernel connects on 443; non-default rejected
+	"QueryTimeout": "rejected", // when > 0 (WithTimeout)
+	"RetryMax":     "rejected", // when < 0 (disable retries)
+	"Protocol":     "rejected", // kernel is https-only; non-default rejected
+	"Port":         "rejected", // kernel connects on 443; non-default rejected
 
 	// Accepted but intentionally inert on the kernel path (documented in doc.go):
 	// the kernel manages these internally, below the C ABI, with no user knob.
diff --git a/kernel_e2e_test.go b/kernel_e2e_test.go
index 7d847bb4..b2d86fac 100644
--- a/kernel_e2e_test.go
+++ b/kernel_e2e_test.go
@@ -265,3 +265,94 @@ func TestKernelE2ECancellation(t *testing.T) {
 	}
 	t.Logf("cancelled after %v with err=%v", elapsed, err)
 }
+
+// TestKernelE2EInitialNamespace proves WithInitialNamespace selects the initial
+// catalog/schema on the kernel session — applied post-connect via USE CATALOG /
+// USE SCHEMA, since the kernel C ABI has no namespace setter. current_catalog() /
+// current_schema() must return the configured values. Uses system.information_schema,
+// which every workspace has, so the test is workspace-agnostic.
+func TestKernelE2EInitialNamespace(t *testing.T) {
+	const catalog, schema = "system", "information_schema"
+	db := kernelTestDBWith(t, WithInitialNamespace(catalog, schema))
+	defer db.Close()
+
+	var gotCatalog, gotSchema string
+	if err := db.QueryRowContext(context.Background(),
+		"SELECT current_catalog(), current_schema()").Scan(&gotCatalog, &gotSchema); err != nil {
+		t.Fatalf("query current namespace: %v", err)
+	}
+	if gotCatalog != catalog {
+		t.Errorf("current_catalog() = %q, want %q", gotCatalog, catalog)
+	}
+	if gotSchema != schema {
+		t.Errorf("current_schema() = %q, want %q", gotSchema, schema)
+	}
+}
+
+// TestKernelE2EMetricViewMetadata proves WithEnableMetricViewMetadata is accepted
+// by the kernel session — the option is routed as the same server session conf the
+// Thrift path sends (spark.sql.thriftserver.metadata.metricview.enabled), folded in
+// backend-neutrally by config.EffectiveSessionParams.
+//
+// The assertion is "session opens and queries succeed with the conf set", not a SET
+// read-back: this conf is not SET-introspectable on the warehouse — `SET <conf>`
+// errors with CONFIG_NOT_AVAILABLE on BOTH backends (verified against Thrift), so a
+// read-back would be testing the warehouse's SET behavior, not our routing. A
+// successful connect+query with the flag on is the parity bar (the Thrift path sends
+// the identical conf at OpenSession and likewise never reads it back via SET).
+//
+// Scope note: this is intentionally a thin live smoke test — it proves the flag is
+// accepted end to end, not that the conf value is correct. The actual routing (the
+// flag → the metricview.enabled=true session conf) is asserted in the default-build
+// unit test TestEffectiveSessionParams; the SELECT here overlaps TestKernelE2ESelect1
+// by design, since the conf's live effect can't be observed (see above). Kept as a
+// connect-with-flag smoke rather than deleted so the flag has at least one live path.
+func TestKernelE2EMetricViewMetadata(t *testing.T) {
+	db := kernelTestDBWith(t, WithEnableMetricViewMetadata(true))
+	defer db.Close()
+
+	var got int64
+	if err := db.QueryRowContext(context.Background(), "SELECT 1").Scan(&got); err != nil {
+		t.Fatalf("kernel session with metric-view metadata enabled failed to query: %v", err)
+	}
+	if got != 1 {
+		t.Errorf("SELECT 1 = %d, want 1", got)
+	}
+}
+
+// TestKernelE2EM2M authenticates to a real warehouse via OAuth M2M over the kernel
+// (WithClientCredentials → the kernel's own client-credentials flow), then runs
+// SELECT 1. It self-skips unless DATABRICKS_CLIENT_ID / DATABRICKS_CLIENT_SECRET are
+// set (alongside the host/http-path base), mirroring the PAT tests' skip pattern —
+// so CI without a service principal stays green while a credentialed run proves the
+// M2M setter path actually authenticates end to end (not just that set_auth_m2m
+// returns OK, which TestSetAuthByMode already covers).
+func TestKernelE2EM2M(t *testing.T) {
+	host := os.Getenv("DATABRICKS_HOST")
+	httpPath := os.Getenv("DATABRICKS_HTTP_PATH")
+	clientID := os.Getenv("DATABRICKS_CLIENT_ID")
+	clientSecret := os.Getenv("DATABRICKS_CLIENT_SECRET")
+	if host == "" || httpPath == "" || clientID == "" || clientSecret == "" {
+		t.Skip("set DATABRICKS_HOST / DATABRICKS_HTTP_PATH / DATABRICKS_CLIENT_ID / DATABRICKS_CLIENT_SECRET for the M2M e2e")
+	}
+
+	connector, err := NewConnector(
+		WithServerHostname(host),
+		WithHTTPPath(httpPath),
+		WithClientCredentials(clientID, clientSecret),
+		WithUseKernel(true),
+	)
+	if err != nil {
+		t.Fatalf("NewConnector: %v", err)
+	}
+	db := sql.OpenDB(connector)
+	defer db.Close()
+
+	var got int64
+	if err := db.QueryRowContext(context.Background(), "SELECT 1").Scan(&got); err != nil {
+		t.Fatalf("M2M-authenticated query failed: %v", err)
+	}
+	if got != 1 {
+		t.Errorf("SELECT 1 = %d, want 1", got)
+	}
+}