diff --git a/connector.go b/connector.go
index b23ea897..24149a89 100644
--- a/connector.go
+++ b/connector.go
@@ -620,6 +620,13 @@ func WithKernelClientCertificate(cert, key []byte) ConnOption {
 		// caller mutating the buffers between NewConnector and Connect can't change
 		// the identity out from under us.
 		k := kernelExperimental(c)
+		// Record that mTLS was explicitly requested, independent of whether the
+		// caller passed non-empty bytes. Without this marker an empty cert+key
+		// pair (e.g. from a failed PEM load) is indistinguishable from the option
+		// never being called, and validateKernelConfig would accept it while
+		// applyKernelTLS silently skipped the setter — connecting with no client
+		// identity. The marker lets validation reject the incomplete request loudly.
+		k.TLSClientCertConfigured = true
 		if len(cert) > 0 {
 			k.TLSClientCertPEM = append([]byte(nil), cert...)
 		} else {
diff --git a/doc.go b/doc.go
index 7e630edf..de9e1191 100644
--- a/doc.go
+++ b/doc.go
@@ -195,19 +195,31 @@ level the lines are suppressed with no cost.
 
 	dbsql.SetLogLevel("debug") // or DATABRICKS_LOG_LEVEL=debug
 
-The same level is mapped into the kernel's internal (Rust) log subscriber, with two
-caveats specific to the Rust lines: they go to stderr directly (not affected by
-logger.SetLogOutput), and their verbosity is fixed when the first kernel session in
-the process is opened — set the level before that first connect to control them.
+The same level is mapped into the kernel's internal (Rust) log subscriber, which is
+routed into the driver's logger via a reverse-call callback (kernel_set_log_callback):
+kernel-internal tracing events are forwarded into the SAME unified sink as the Go
+binding lines — including a custom logger.SetLogOutput — rather than going to stderr.
+One caveat specific to the Rust lines: BOTH their verbosity AND their output
+destination are captured when the first kernel session in the process is opened (the
+forwarding sink snapshots the logger then). A later dbsql.SetLogLevel or
+logger.SetLogOutput re-targets the Go binding lines but NOT the already-forwarded
+kernel lines, so set the level and any custom output before that first connect to
+govern them.
 
 For finer control of the Rust verbosity independent of the driver level, set
-DBSQL_KERNEL_DEBUG to any non-empty value: it forces the kernel subscriber on and
-defers to RUST_LOG. Filter on the target databricks::sql::kernel (note the colons):
+DBSQL_KERNEL_DEBUG to any non-empty value: the callback then defers its level to
+RUST_LOG. Filter on the target databricks::sql::kernel (note the colons):
 
-	# kernel logs only, at the kernel's own verbosity:
-	DBSQL_KERNEL_DEBUG=1 RUST_LOG=databricks::sql::kernel=debug ./your_app 2>&1
+	# kernel logs at the kernel's own verbosity, forwarded into the driver logger:
+	DBSQL_KERNEL_DEBUG=1 RUST_LOG=databricks::sql::kernel=debug ./your_app
 	# kernel logs plus its HTTP stack:
-	DBSQL_KERNEL_DEBUG=1 RUST_LOG=debug ./your_app 2>&1
+	DBSQL_KERNEL_DEBUG=1 RUST_LOG=debug ./your_app
+
+The kernel exposes one process-global tracing subscriber reachable two mutually
+exclusive ways over the C ABI (kernel_init_logging → stderr, kernel_set_log_callback
+→ the driver logger); the driver installs the callback. A host that has already
+installed its own global tracing subscriber will see the registration report a
+benign, logged failure and kernel events simply won't forward.
 
 Supported on the kernel backend: PAT and OAuth (M2M via WithClientCredentials, U2M
 via the authType=oauthU2M DSN param); reading scalar, nested, and complex-typed
diff --git a/internal/backend/kernel/abi.go b/internal/backend/kernel/abi.go
new file mode 100644
index 00000000..1f8a10a7
--- /dev/null
+++ b/internal/backend/kernel/abi.go
@@ -0,0 +1,22 @@
+package kernel
+
+import "fmt"
+
+// compareABI reports whether the C-ABI version compiled into the linked kernel
+// library (got) matches the version the driver's header declares (want): it
+// returns a descriptive, operator-facing error on mismatch and nil when they
+// agree.
+//
+// It is split out from checkABIVersion (in the cgo-tagged cgo.go) as pure,
+// cgo-free logic so the mismatch verdict and its message are unit-testable under
+// the default CGO_ENABLED=0 build. The negative path can't otherwise be reached
+// from a test: a real build always links a .a and header produced together, and
+// abiVersions() reads fixed cgo constants with no injection seam. checkABIVersion
+// runs abiVersions() through this on every connect.
+func compareABI(got, want uint32) error {
+	if got != want {
+		return fmt.Errorf("databricks: kernel ABI version mismatch: linked library reports %d, "+
+			"driver header expects %d; rebuild the kernel static lib and header together (make kernel-lib)", got, want)
+	}
+	return nil
+}
diff --git a/internal/backend/kernel/abi_test.go b/internal/backend/kernel/abi_test.go
new file mode 100644
index 00000000..0059a4c3
--- /dev/null
+++ b/internal/backend/kernel/abi_test.go
@@ -0,0 +1,37 @@
+package kernel
+
+import (
+	"strings"
+	"testing"
+)
+
+// TestCompareABI covers both verdicts of the ABI-version handshake without the
+// cgo symbols, so it runs under the default CGO_ENABLED=0 build. The matching
+// (happy) path is additionally proven against the real linked library by
+// TestABIVersionMatches in the cgo build; the mismatch path — the runtime hazard
+// the check exists for (a driver header linked against a differently-built
+// prebuilt .a) — is otherwise unreachable from a test, since a real build always
+// links a matching .a + header.
+func TestCompareABI(t *testing.T) {
+	// Matching versions must produce no error: checkABIVersion opens the session.
+	for _, v := range []uint32{0, 1, 42} {
+		if err := compareABI(v, v); err != nil {
+			t.Errorf("compareABI(%d, %d) = %v, want nil for matching versions", v, v, err)
+		}
+	}
+
+	// Mismatched versions must produce an error so checkABIVersion refuses to open
+	// (rather than silently misread status codes / error-struct fields). The
+	// message must name both versions and the remediation so an operator hitting a
+	// stale prebuilt .a knows what to rebuild.
+	err := compareABI(2, 1)
+	if err == nil {
+		t.Fatal("compareABI(2, 1) = nil, want an error for mismatched versions")
+	}
+	msg := err.Error()
+	for _, want := range []string{"ABI version mismatch", "reports 2", "expects 1", "make kernel-lib"} {
+		if !strings.Contains(msg, want) {
+			t.Errorf("mismatch error %q does not contain %q", msg, want)
+		}
+	}
+}
diff --git a/internal/backend/kernel/cgo.go b/internal/backend/kernel/cgo.go
index 48d13c53..34dbd52e 100644
--- a/internal/backend/kernel/cgo.go
+++ b/internal/backend/kernel/cgo.go
@@ -101,56 +101,45 @@ func klogCtx(ctx context.Context, format string, args ...any) {
 	).Debug().Msgf("[kernel] "+format, args...)
 }
 
-// initLoggingOnce guards kernel_init_logging, which is process-wide and
-// first-call-wins in the kernel. We install the kernel subscriber lazily on the
-// first session open rather than in init(), so a process that never opens a
-// kernel session installs nothing.
-var initLoggingOnce sync.Once
-
-// initKernelLogging turns on the kernel's own Rust (tracing) logs and points their
-// verbosity at the driver's log level, so DATABRICKS_LOG_LEVEL drives both the Go
-// binding lines and the kernel's Rust lines from one knob. The mapped level is
-// passed to kernel_init_logging (Go zerolog level → the kernel's OFF/ERROR/WARN/
-// INFO/DEBUG/TRACE string); DBSQL_KERNEL_DEBUG forces the subscriber on with a
-// NULL level so the kernel honors RUST_LOG instead (the advanced override for
-// tuning kernel-only verbosity). file_path=NULL sends kernel logs to stderr — the
-// kernel ABI has no sink hook, so the Rust lines always go to stderr and are NOT
-// routed through logger.SetLogOutput (unlike the Go binding lines).
+// initKernelLogging routes the kernel's own Rust (tracing) logs into the driver's
+// logger and points their verbosity at the driver's log level, so
+// DATABRICKS_LOG_LEVEL drives both the Go binding lines and the kernel's Rust
+// lines from one knob.
+//
+// The kernel exposes exactly one process-global tracing subscriber, reachable
+// over the C ABI two mutually-exclusive ways: kernel_init_logging (RUST_LOG →
+// stderr) and kernel_set_log_callback (reverse-call → a host sink). We install
+// the CALLBACK (see installKernelLogCallback) so kernel-internal events flow into
+// logger.Logger — the SAME unified stream as the Go binding lines (klog/klogCtx)
+// and honouring logger.SetLogOutput — instead of only reaching stderr.
 //
-// Best-effort: an Internal return (e.g. the host already installed a global
-// subscriber) is a documented, benign outcome — logged at Warn, never fatal to
-// connect. The subscriber installs at whatever level is mapped in; a driver left at
-// the default Warn level (benchmarks included, and never having set
-// DBSQL_KERNEL_DEBUG) installs it at WARN, so the kernel emits nothing below Warn
-// and there is no hot-path cost.
+// resolveKernelLogArg maps the driver's zerolog level to the kernel's
+// OFF/ERROR/WARN/INFO/DEBUG/TRACE string; DBSQL_KERNEL_DEBUG instead defers to
+// RUST_LOG (useNULL → an empty level string → NULL to the kernel, the advanced
+// override for tuning kernel-only verbosity). The callback filters at that level
+// with the same precedence kernel_init_logging used, so the swap is level-transparent.
 //
-// Scope caveat: the kernel subscriber is PROCESS-WIDE, first-call-wins, and never
+// Best-effort: a non-Success install (e.g. the host already installed a global
+// subscriber) is a documented, benign outcome — logged, never fatal to connect
+// (handled inside installKernelLogCallback). A driver left at the default Warn
+// level (benchmarks included, DBSQL_KERNEL_DEBUG unset) installs at WARN, so the
+// kernel emits nothing below Warn and there is no hot-path cost.
+//
+// Scope caveat: the subscriber is PROCESS-WIDE, first-call-wins, and never
 // uninstalled — in a long-lived multi-tenant process the first kernel session's
-// level/destination applies to ALL subsequent kernel sessions, with no way to
-// re-scope or turn it off afterward. That is a kernel-ABI property, not a Go one.
-// A direct consequence: the driver level is sampled HERE, once, at the first kernel
-// session — a later dbsql.SetLogLevel re-levels the Go binding lines (klog/klogCtx
-// re-read GetLevel per call) but NOT the already-installed Rust subscriber. Set the
-// level before opening the first kernel connection to govern the Rust logs.
+// level applies to ALL subsequent kernel sessions. The driver level is sampled
+// HERE, once, at the first kernel session — a later dbsql.SetLogLevel re-levels the
+// Go binding lines (klog/klogCtx re-read GetLevel per call) but NOT the
+// already-installed Rust subscriber. Set the level before opening the first kernel
+// connection to govern the Rust logs.
 func initKernelLogging() {
-	initLoggingOnce.Do(func() {
-		// resolveKernelLogArg decides the level (or NULL for the DBSQL_KERNEL_DEBUG
-		// override, which lets the kernel honor RUST_LOG). The pure decision lives in
-		// logging_level.go so it's unit-tested without cgo.
-		var level cStr
-		if lvl, useNULL := resolveKernelLogArg(); !useNULL {
-			level = newCStr(lvl)
-			defer level.free()
-		} // else level stays {c: nil} → NULL → kernel honors RUST_LOG
-		if err := call(func() C.KernelStatusCode {
-			return C.kernel_init_logging(level.c, nil)
-		}); err != nil {
-			// The kernel subscriber didn't install (commonly: the host already
-			// installed a global tracing subscriber). Non-fatal — surface it through
-			// the shared logger so it's visible without a separate stderr scrape.
-			logger.Logger.Warn().Msgf("databricks: kernel_init_logging: %v (kernel logs unavailable; proceeding)", err)
-		}
-	})
+	// resolveKernelLogArg decides the level string (or "" for the
+	// DBSQL_KERNEL_DEBUG override, so the kernel honors RUST_LOG). The pure
+	// decision lives in logging_level.go so it's unit-tested without cgo.
+	level, _ := resolveKernelLogArg()
+	// installKernelLogCallback self-guards with logCallbackOnce (process-wide,
+	// first-call-wins), so no extra sync.Once here.
+	installKernelLogCallback(level)
 }
 
 // abiCheckOnce ensures the ABI-version handshake runs at most once per process
@@ -169,15 +158,14 @@ var (
 // download-a-prebuilt-lib distribution path). A mismatch means the
 // KernelStatusCode enum / KernelError struct layout the driver reads may not
 // match what the library writes, so we refuse to open rather than silently
-// misread status codes / error fields. Runs once; the result is cached.
+// misread status codes / error fields. Runs once; the result is cached. The
+// pure got-vs-want verdict lives in compareABI (untagged) so its mismatch path
+// is testable without the cgo symbols.
 func checkABIVersion() error {
 	abiCheckOnce.Do(func() {
 		got, want := abiVersions()
 		klog("checkABIVersion got=%d want=%d", got, want)
-		if got != want {
-			abiCheckErr = fmt.Errorf("databricks: kernel ABI version mismatch: linked library reports %d, "+
-				"driver header expects %d; rebuild the kernel static lib and header together (make kernel-lib)", got, want)
-		}
+		abiCheckErr = compareABI(got, want)
 	})
 	return abiCheckErr
 }
diff --git a/internal/backend/kernel/kernel_test.go b/internal/backend/kernel/kernel_test.go
index 44b752f8..8d2e8546 100644
--- a/internal/backend/kernel/kernel_test.go
+++ b/internal/backend/kernel/kernel_test.go
@@ -9,6 +9,7 @@ import (
 	"encoding/json"
 	"errors"
 	"os"
+	"runtime/cgo"
 	"testing"
 
 	"github.com/databricks/databricks-sql-go/driverctx"
@@ -100,6 +101,110 @@ func TestABIVersionMatches(t *testing.T) {
 	}
 }
 
+// TestLogCallbackRoundTrip exercises the full reverse-call path: wrap an
+// observable *logSink in a cgo.Handle, drive the C→Go trampoline exactly as the
+// kernel's drain thread would (via the test seam), and assert the event arrived
+// with the handle correctly unwrapped and the fields intact. This proves the
+// cgo.Handle round-trip + recover firewall + sink dispatch that a real kernel log
+// event flows through (kernel-side delivery itself is covered by the kernel's own
+// integration tests).
+func TestLogCallbackRoundTrip(t *testing.T) {
+	type got struct {
+		level           int
+		target, message string
+	}
+	ch := make(chan got, 1)
+	sink := &logSink{observe: func(level int, target, message string) {
+		ch <- got{level, target, message}
+	}}
+	h := cgo.NewHandle(sink)
+	defer h.Delete()
+
+	invokeLogTrampolineForTest(h, kernelLevelWarn, "databricks::sql::kernel", "retrying request")
+
+	select {
+	case g := <-ch:
+		if g.level != kernelLevelWarn || g.target != "databricks::sql::kernel" || g.message != "retrying request" {
+			t.Errorf("delivered event = %+v, want level=WARN target=databricks::sql::kernel msg='retrying request'", g)
+		}
+	default:
+		t.Fatal("the trampoline did not deliver the event to the sink")
+	}
+}
+
+// captureDriverLog points the global driver logger at a buffer (TraceLevel) for
+// the duration of a test, returning the buffer and restoring the logger after.
+// Used to assert a trampoline guard branch delivered NOTHING (the branches bail
+// before reaching a sink, so there's no observe hook to check — the driver
+// logger output is the observable).
+func captureDriverLog(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var buf bytes.Buffer
+	prev := logger.Logger.GetLevel()
+	logger.SetLogOutput(&buf)
+	_ = logger.SetLogLevel("trace")
+	t.Cleanup(func() {
+		logger.SetLogOutput(os.Stderr)
+		logger.Logger.Logger = logger.Logger.Level(prev)
+	})
+	return &buf
+}
+
+// A zero ctx must be a no-op (no crash, no delivery). cgo.Handle(0) casts to a
+// NULL void* ctx, so the trampoline returns at its `ctx == nil` guard before ever
+// calling Value() (calling Value() on the zero handle would itself panic — the
+// recover firewall would catch it, but the nil guard means we never reach it).
+func TestLogCallbackTrampolineNilCtxSafe(t *testing.T) {
+	buf := captureDriverLog(t)
+	invokeLogTrampolineForTest(cgo.Handle(0), kernelLevelError, "databricks::sql::kernel", "should-not-appear")
+	if buf.Len() != 0 {
+		t.Errorf("nil-ctx trampoline delivered a line, want none: %q", buf.String())
+	}
+}
+
+// TestLogCallbackTrampolineWrongTypeSafe covers the "handle holds a non-*logSink
+// value" branch: a live, non-nil handle whose Value() type-asserts to false. The
+// trampoline must return without panicking AND without delivering (asserted via
+// the captured driver logger staying empty).
+func TestLogCallbackTrampolineWrongTypeSafe(t *testing.T) {
+	buf := captureDriverLog(t)
+	h := cgo.NewHandle("not a logSink")
+	defer h.Delete()
+	invokeLogTrampolineForTest(h, kernelLevelError, "databricks::sql::kernel", "should-not-appear")
+	if buf.Len() != 0 {
+		t.Errorf("wrong-type trampoline delivered a line, want none: %q", buf.String())
+	}
+}
+
+// TestLogCallbackPanicFirewall is the load-bearing safety test: a sink whose
+// routing panics must NOT crash the process — the trampoline's defer recover()
+// converts it into a dropped line (a panic unwinding across the cgo boundary
+// would abort). We drive a panicking observe hook, assert the call returns, and
+// assert a subsequent well-formed sink still delivers (the firewall didn't wedge
+// anything process-global).
+func TestLogCallbackPanicFirewall(t *testing.T) {
+	panicSink := &logSink{observe: func(int, string, string) { panic("boom in the sink") }}
+	ph := cgo.NewHandle(panicSink)
+	defer ph.Delete()
+	// Must not crash the test binary.
+	invokeLogTrampolineForTest(ph, kernelLevelError, "databricks::sql::kernel", "will panic")
+
+	// A subsequent well-formed delivery still works.
+	ch := make(chan string, 1)
+	okSink := &logSink{observe: func(_ int, _, message string) { ch <- message }}
+	oh := cgo.NewHandle(okSink)
+	defer oh.Delete()
+	invokeLogTrampolineForTest(oh, kernelLevelWarn, "databricks::sql::kernel", "recovered")
+	select {
+	case got := <-ch:
+		if got != "recovered" {
+			t.Errorf("post-panic delivery = %q, want 'recovered'", got)
+		}
+	default:
+		t.Fatal("delivery after a panicking callback did not work — firewall wedged")
+	}
+}
+
 // TestKernelLogLevel and TestResolveKernelLogArg — the pure level-resolution tests —
 // live in the untagged logging_level_test.go so they run under CGO_ENABLED=0. The
 // tests below exercise klog/klogCtx and so need the cgo build.
diff --git a/internal/backend/kernel/log_callback.go b/internal/backend/kernel/log_callback.go
new file mode 100644
index 00000000..84f23062
--- /dev/null
+++ b/internal/backend/kernel/log_callback.go
@@ -0,0 +1,134 @@
+//go:build cgo && databricks_kernel
+
+package kernel
+
+/*
+#include <stdlib.h>
+#include "databricks_kernel.h"
+
+// Adapter matching the kernel_log_callback typedef (const char*) that forwards
+// to the //export'd Go trampoline. cgo generates the trampoline's prototype in
+// _cgo_export.h with non-const char* params, which don't match the typedef, so
+// this shim bridges the two and gives kernel_set_log_callback one clean address
+// to take. Forward-declared with the SAME (char*) signature cgo generates to
+// avoid a conflicting-declaration error; the kernel only ever reads the strings.
+void kernelLogTrampoline(void* ctx, int level, char* target, char* message);
+// Hand kernel_set_log_callback the trampoline's address as a kernel_log_callback.
+// The //export'd trampoline's generated signature uses non-const char* while the
+// typedef uses const char*, which are function-pointer compatible (a const-only
+// difference); do the cast once here, on the C side, so the Go call site stays
+// clean and there is no separately-linked adapter symbol.
+static kernel_log_callback kernel_log_cb(void) {
+    return (kernel_log_callback)kernelLogTrampoline;
+}
+// Cast a cgo.Handle (an integer token) to the opaque void* ctx on the C side,
+// so Go never does unsafe.Pointer(uintptr) directly (which `go vet` flags).
+static void* kernel_handle_to_ctx(uintptr_t h) { return (void*)h; }
+// Test-only: invoke the trampoline exactly as the kernel drain thread would.
+// (A C func-pointer value can't be called directly from Go, so a helper does it.)
+static void kernel_invoke_trampoline_for_test(void* ctx, int level, char* target, char* message) {
+    kernelLogTrampoline(ctx, level, target, message);
+}
+*/
+import "C"
+
+import (
+	"runtime/cgo"
+	"sync"
+	"unsafe"
+
+	"github.com/databricks/databricks-sql-go/logger"
+)
+
+// This file holds the reverse-call machinery for the kernel log callback:
+// a cgo-exported Go function the kernel invokes from its log-drain thread, made
+// safe with a recover() panic firewall (a panic across the cgo boundary would
+// abort the process) and a runtime/cgo.Handle so a Go pointer (the *logSink) can
+// round-trip through the C void* ctx under cgo's pointer rules.
+
+// logCallbackOnce guards the one-time callback registration; logCallbackHandle
+// keeps the cgo.Handle alive for the process (the kernel holds its numeric value
+// as the opaque ctx and there is no detach in the v0 C ABI, so we never Delete
+// it — a deliberate process-lifetime pin, not a leak).
+var (
+	logCallbackOnce   sync.Once
+	logCallbackHandle cgo.Handle
+)
+
+// invokeLogTrampolineForTest drives the C→Go trampoline exactly as the kernel's
+// drain thread would — allocating C strings, casting a cgo.Handle to void*, and
+// calling through kernel_log_cb() — so a test can assert the full reverse-call
+// round-trip (handle unwrap + recover firewall + sink dispatch) without a live
+// kernel event. Not used in production.
+func invokeLogTrampolineForTest(h cgo.Handle, level int, target, message string) {
+	ct := C.CString(target)
+	defer C.free(unsafe.Pointer(ct))
+	cm := C.CString(message)
+	defer C.free(unsafe.Pointer(cm))
+	ctx := C.kernel_handle_to_ctx(C.uintptr_t(h))
+	C.kernel_invoke_trampoline_for_test(ctx, C.int(level), ct, cm)
+}
+
+//export kernelLogTrampoline
+func kernelLogTrampoline(ctx unsafe.Pointer, level C.int, target, message *C.char) {
+	// Panic firewall: a Go panic unwinding across the cgo boundary aborts the
+	// process. Convert any panic in the sink routing into a dropped log line.
+	defer func() { _ = recover() }()
+	if ctx == nil {
+		return
+	}
+	sink, ok := cgo.Handle(uintptr(ctx)).Value().(*logSink)
+	if !ok || sink == nil {
+		return
+	}
+	// Copy the borrowed C strings out immediately — they are valid only for the
+	// duration of this call.
+	sink.forward(int(level), C.GoString(target), C.GoString(message))
+}
+
+// installKernelLogCallback registers the trampoline as the kernel's log sink,
+// once per process, wrapping a *logSink in a cgo.Handle passed as the opaque ctx.
+// The kernel filters forwarded events at `level` (an OFF/ERROR/WARN/INFO/DEBUG/
+// TRACE string, or "" → NULL to defer to RUST_LOG), which the caller maps from
+// the driver's own log level so kernel Rust lines follow the one
+// DATABRICKS_LOG_LEVEL knob — the same string kernel_init_logging would have
+// received (see initKernelLogging).
+//
+// Best-effort: a non-Success status (e.g. Internal because a global tracing
+// subscriber was already installed) is logged and ignored — kernel logs simply
+// won't flow through the callback, which is a documented, benign outcome. Returns
+// nothing; the caller does not gate connect on it.
+func installKernelLogCallback(level string) {
+	logCallbackOnce.Do(func() {
+		// Snapshot the driver logger at install (TraceLevel, immutable) so the
+		// kernel drain thread never re-gates already-approved events and never
+		// reads the mutable global logger.Logger (see logSink.log).
+		logCallbackHandle = cgo.NewHandle(newLogSink())
+		// Pass the handle (an integer token) as the opaque ctx via a C-side cast,
+		// so Go doesn't do unsafe.Pointer(uintptr) directly.
+		ctx := C.kernel_handle_to_ctx(C.uintptr_t(logCallbackHandle))
+		// An empty level maps to a NULL char* (kernel defers to RUST_LOG); a
+		// non-empty level is passed as a C string the kernel parses with the same
+		// precedence as kernel_init_logging (explicit wins > RUST_LOG > warn).
+		clevel := newCStrOrNull(level)
+		defer clevel.free()
+		if err := call(func() C.KernelStatusCode {
+			return C.kernel_set_log_callback(C.kernel_log_cb(), ctx, clevel.c)
+		}); err != nil {
+			// On a non-Success return the kernel installed no subscriber and did
+			// NOT retain ctx (it stores the sink only on the success path), so the
+			// process-lifetime pin at logCallbackHandle would be a true leak here —
+			// free it. (On success we deliberately never Delete: the kernel holds
+			// the handle value as its opaque ctx for the process lifetime.)
+			logCallbackHandle.Delete()
+			logCallbackHandle = 0
+			// Surface at Warn (visible at the default level), NOT klog: klog is
+			// Debug-gated and no-ops at the default, and this install runs once so
+			// a later level change can't re-surface it. The message names the
+			// benign common cause (a global subscriber already installed) so a
+			// logging no-op isn't mistaken for a connect failure.
+			logger.Logger.Warn().Msgf(
+				"databricks: kernel_set_log_callback: %v (kernel logs not forwarded; proceeding)", err)
+		}
+	})
+}
diff --git a/internal/backend/kernel/logforward.go b/internal/backend/kernel/logforward.go
new file mode 100644
index 00000000..a7a62d55
--- /dev/null
+++ b/internal/backend/kernel/logforward.go
@@ -0,0 +1,92 @@
+package kernel
+
+// This file is intentionally NOT behind the cgo build tag: the level mapping
+// and sink-routing logic are pure Go, so their tests run under CGO_ENABLED=0.
+// The cgo trampoline that calls into these lives in log_callback.go (tagged).
+
+import (
+	"github.com/databricks/databricks-sql-go/logger"
+	"github.com/rs/zerolog"
+)
+
+// Kernel log levels as delivered over the C ABI (1=ERROR..5=TRACE), mirroring
+// the kernel's numeric severity. 0 is unused on the wire (the kernel never
+// forwards an OFF event).
+const (
+	kernelLevelError = 1
+	kernelLevelWarn  = 2
+	kernelLevelInfo  = 3
+	kernelLevelDebug = 4
+	kernelLevelTrace = 5
+)
+
+// logSink receives forwarded kernel log events and routes them into the driver's
+// logger, so kernel-internal diagnostics (HTTP stack, CloudFetch, retry) join the
+// unified Go+kernel debug stream instead of only reaching stderr via RUST_LOG.
+// A dedicated type (rather than calling the logger directly from the trampoline)
+// keeps the routing pure-Go and unit-testable, and gives the cgo.Handle something
+// concrete to wrap.
+type logSink struct {
+	// log is the destination for forwarded events. It is a SNAPSHOT of the
+	// driver logger taken once at install (newLogSink), pinned to TraceLevel, and
+	// never reassigned. Two reasons:
+	//   - No double gating. The KERNEL already filters events against the level
+	//     the driver passed to kernel_set_log_callback (DATABRICKS_LOG_LEVEL, or
+	//     RUST_LOG under DBSQL_KERNEL_DEBUG); that decision is authoritative. If
+	//     we routed through the live logger.Logger — gated at the driver level —
+	//     a DEBUG kernel event under the DBSQL_KERNEL_DEBUG override would be
+	//     re-dropped by a driver still at Warn, defeating the override. Pinning
+	//     the sink at TraceLevel lets every already-approved event through.
+	//   - No data race. The kernel drain thread (a non-Go OS thread) calls
+	//     forward asynchronously; reading the mutable global logger.Logger here
+	//     would race a concurrent dbsql.SetLogLevel/SetLogOutput. A value copy
+	//     captured at install is immutable, so the drain thread reads only its
+	//     own snapshot. (The writer/output is captured at install; a later
+	//     SetLogOutput does not re-target already-forwarded kernel lines — the
+	//     same "level fixed at first connect" caveat the kernel subscriber has.)
+	log zerolog.Logger
+
+	// observe, when non-nil, is invoked with each forwarded event before the
+	// logger routing. A test seam so a round-trip test can assert the trampoline
+	// unwrapped the handle and delivered the event; nil in production.
+	observe func(level int, target, message string)
+}
+
+// newLogSink snapshots the current driver logger at TraceLevel for the sink to
+// forward through. Called once at install so the drain thread never touches the
+// mutable global logger. Kept separate from the zero value so tests can build a
+// sink with an explicit writer.
+func newLogSink() *logSink {
+	return &logSink{log: logger.Logger.Level(zerolog.TraceLevel)}
+}
+
+// forward routes one kernel log event into the sink's logger at the mapped level.
+// The sink logger is a TraceLevel snapshot of the driver logger (see logSink.log),
+// so events the kernel already approved are emitted without being re-gated by the
+// driver level. Kept small and allocation-light: it is on the (debug-only) log
+// path, not a hot query path. target and message are already Go strings (copied
+// out of the C buffers by the trampoline before this is called).
+func (s *logSink) forward(level int, target, message string) {
+	if s == nil {
+		return
+	}
+	if s.observe != nil {
+		s.observe(level, target, message)
+	}
+	switch level {
+	case kernelLevelError:
+		s.log.Error().Str("target", target).Msg(message)
+	case kernelLevelWarn:
+		s.log.Warn().Str("target", target).Msg(message)
+	case kernelLevelInfo:
+		s.log.Info().Str("target", target).Msg(message)
+	case kernelLevelDebug:
+		s.log.Debug().Str("target", target).Msg(message)
+	case kernelLevelTrace:
+		s.log.Trace().Str("target", target).Msg(message)
+	default:
+		// Unknown level: don't drop it — surface at Debug so it's still visible
+		// without being mistaken for a warning/error.
+		s.log.Debug().Str("target", target).Int("kernelLevel", level).Msg(message)
+	}
+}
diff --git a/internal/backend/kernel/logforward_test.go b/internal/backend/kernel/logforward_test.go
new file mode 100644
index 00000000..fccec27c
--- /dev/null
+++ b/internal/backend/kernel/logforward_test.go
@@ -0,0 +1,97 @@
+package kernel
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/databricks/databricks-sql-go/logger"
+	"github.com/rs/zerolog"
+)
+
+// The level mapping and nil-sink no-op are pure Go, so they run under
+// CGO_ENABLED=0. Actual kernel→callback delivery is exercised by the tagged
+// TestLogCallbackRoundTrip (which needs the linked kernel).
+
+// A nil sink must be a safe no-op (the trampoline guards against a missing/typed
+// handle by calling forward on a possibly-nil sink).
+func TestLogSinkForwardNilIsNoOp(t *testing.T) {
+	var s *logSink
+	// Must not panic.
+	s.forward(kernelLevelError, "databricks::sql::kernel", "boom")
+}
+
+// forward must map each kernel level to the right zerolog level and carry the
+// target + message through. A buffer-backed TraceLevel sink (mirroring the
+// production snapshot) lets us assert the emitted JSON rather than only "no
+// panic" — a bug swapping e.g. ERROR→Debug would otherwise ship green.
+func TestLogSinkForwardMapsLevels(t *testing.T) {
+	cases := []struct {
+		kernelLevel int
+		wantLevel   string // zerolog "level" field
+		unknown     bool   // unmapped → Debug WITH a kernelLevel diagnostic field
+	}{
+		{kernelLevelError, "error", false},
+		{kernelLevelWarn, "warn", false},
+		{kernelLevelInfo, "info", false},
+		{kernelLevelDebug, "debug", false},
+		{kernelLevelTrace, "trace", false},
+		{0, "debug", true},
+		{99, "debug", true},
+		{-1, "debug", true},
+	}
+	for _, c := range cases {
+		var buf bytes.Buffer
+		s := &logSink{log: zerolog.New(&buf).Level(zerolog.TraceLevel)}
+		s.forward(c.kernelLevel, "databricks::sql::kernel", "hello")
+		var rec map[string]any
+		if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil {
+			t.Fatalf("kernelLevel=%d: emitted non-JSON %q: %v", c.kernelLevel, buf.String(), err)
+		}
+		if rec["level"] != c.wantLevel {
+			t.Errorf("kernelLevel=%d: level=%v, want %s", c.kernelLevel, rec["level"], c.wantLevel)
+		}
+		// The unknown-level branch adds a kernelLevel field — the ONLY thing
+		// distinguishing an unmapped level from a real Debug line. Assert it so
+		// dropping that field ships red, not green.
+		if c.unknown {
+			if got, ok := rec["kernelLevel"]; !ok || got != float64(c.kernelLevel) {
+				t.Errorf("kernelLevel=%d: expected kernelLevel field=%d, got %v (present=%t)",
+					c.kernelLevel, c.kernelLevel, got, ok)
+			}
+		} else if _, ok := rec["kernelLevel"]; ok {
+			t.Errorf("kernelLevel=%d: mapped level should NOT carry a kernelLevel field", c.kernelLevel)
+		}
+		if rec["target"] != "databricks::sql::kernel" || rec["message"] != "hello" {
+			t.Errorf("kernelLevel=%d: target/message not carried through: %v", c.kernelLevel, rec)
+		}
+	}
+}
+
+// A TraceLevel snapshot must emit events the driver's own level would suppress —
+// this is the anti-double-gating property (forwarded kernel events the kernel
+// already approved must not be re-dropped by a driver still at Warn). Drives the
+// PRODUCTION newLogSink() against a Warn-level global logger, so a regression of
+// newLogSink to `&logSink{log: logger.Logger}` (routing through the live,
+// driver-gated logger) fails here rather than passing against a hand-rolled replica.
+func TestLogSinkForwardNotReGatedByDriverLevel(t *testing.T) {
+	// Point the global driver logger at a buffer and pin it to Warn — the state
+	// newLogSink() reads at install. Restore afterward so other tests are
+	// unaffected.
+	var buf bytes.Buffer
+	prevLevel := logger.Logger.GetLevel()
+	logger.SetLogOutput(&buf)
+	_ = logger.SetLogLevel("warn")
+	t.Cleanup(func() {
+		logger.SetLogOutput(os.Stderr)
+		logger.Logger.Logger = logger.Logger.Level(prevLevel)
+	})
+
+	s := newLogSink() // snapshots logger.Logger at TraceLevel — the real path
+	s.forward(kernelLevelDebug, "databricks::sql::kernel", "debug-line")
+	if !strings.Contains(buf.String(), "debug-line") {
+		t.Errorf("a DEBUG kernel event was dropped despite newLogSink's TraceLevel snapshot: %q", buf.String())
+	}
+}
diff --git a/internal/config/config.go b/internal/config/config.go
index e7b56897..b22e13a1 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -83,6 +83,16 @@ type KernelExperimentalConfig struct {
 	// kernel_session_config_set_tls_client_certificate. The key is never logged.
 	TLSClientCertPEM []byte
 	TLSClientKeyPEM  []byte
+	// TLSClientCertConfigured records that WithKernelClientCertificate was
+	// invoked, independent of whether the caller passed non-empty bytes. It
+	// exists because an empty cert+key pair is otherwise indistinguishable from
+	// the option never being called (KernelExperimental can be non-nil from any
+	// other WithKernel* option), which would let a caller who explicitly asked
+	// for mTLS — e.g. after a failed/empty PEM load — silently connect with no
+	// client identity (applyKernelTLS gates the setter on a non-empty cert).
+	// validateKernelConfig uses this marker to reject an incomplete mTLS request
+	// loudly instead of failing open. Never forwarded to the kernel C ABI.
+	TLSClientCertConfigured bool
 	// CloudFetchEnabled toggles CloudFetch on the kernel path. Tri-state: nil
 	// keeps the kernel default (on); a non-nil *false forces the inline-Arrow
 	// path. A *bool (not a plain bool) so "unset" is distinguishable from
@@ -98,7 +108,10 @@ func (k *KernelExperimentalConfig) DeepCopy() *KernelExperimentalConfig {
 	if k == nil {
 		return nil
 	}
-	cp := &KernelExperimentalConfig{TLSSkipHostnameVerify: k.TLSSkipHostnameVerify}
+	cp := &KernelExperimentalConfig{
+		TLSSkipHostnameVerify:   k.TLSSkipHostnameVerify,
+		TLSClientCertConfigured: k.TLSClientCertConfigured,
+	}
 	if k.TLSTrustedCertsPEM != nil {
 		cp.TLSTrustedCertsPEM = append([]byte(nil), k.TLSTrustedCertsPEM...)
 	}
diff --git a/kernel_config.go b/kernel_config.go
index a147403b..9f6d4ea0 100644
--- a/kernel_config.go
+++ b/kernel_config.go
@@ -87,17 +87,24 @@ func validateKernelConfig(cfg *config.Config) (kernel.Auth, error) {
 			"(the kernel retries internally); omit it or use the default (Thrift) backend", dbsqlerr.ErrNotSupportedByKernel)
 	}
 	// mTLS client identity (WithKernelClientCertificate) must be a complete pair:
-	// mTLS needs both a client cert and its private key. Reject an unpaired
-	// credential loudly at connect — otherwise applyKernelTLS (which gates the mTLS
-	// forward on the cert being present) would silently drop a lone key and connect
-	// with no client identity, contradicting the "both halves required together /
-	// never silently dropped" contract. The cert-without-key case would fail at the
-	// kernel setter (it rejects a null/empty key), but validating both directions
-	// here surfaces the error earlier and with a clearer message.
+	// mTLS needs both a non-empty client cert and its private key. Reject any
+	// incomplete request loudly at connect — otherwise applyKernelTLS (which gates
+	// the mTLS forward on the cert being non-empty) would silently skip the setter
+	// and connect with no client identity, a fail-open that contradicts the "both
+	// halves required together / never silently dropped" contract.
+	//
+	// "Incomplete" covers three cases: cert-without-key, key-without-cert, and —
+	// critically — the option being invoked with both halves empty/nil (e.g. from
+	// a failed PEM load), which the marker distinguishes from the option never
+	// being called. We also treat a bare cert/key present as a request even
+	// without the marker, so a config assembled without the option can't fail open
+	// either.
 	if ke := cfg.KernelExperimental; ke != nil {
-		if (len(ke.TLSClientCertPEM) == 0) != (len(ke.TLSClientKeyPEM) == 0) {
+		certLen, keyLen := len(ke.TLSClientCertPEM), len(ke.TLSClientKeyPEM)
+		mtlsRequested := ke.TLSClientCertConfigured || certLen > 0 || keyLen > 0
+		if mtlsRequested && (certLen == 0 || keyLen == 0) {
 			return kernel.Auth{}, fmt.Errorf("databricks: WithKernelClientCertificate requires " +
-				"both a client certificate and a private key; one was empty")
+				"both a non-empty client certificate and a non-empty private key")
 		}
 	}
 	return kauth, nil
diff --git a/kernel_config_test.go b/kernel_config_test.go
index dfe07b40..1f5edfe2 100644
--- a/kernel_config_test.go
+++ b/kernel_config_test.go
@@ -154,6 +154,36 @@ func TestValidateKernelConfig(t *testing.T) {
 			t.Errorf("a complete mTLS pair should validate, got %v", err)
 		}
 	})
+	// Fail-open guard: WithKernelClientCertificate invoked with empty bytes (e.g.
+	// a failed PEM load) sets the configured marker with no cert/key. Validation
+	// must reject it rather than let applyKernelTLS silently skip the setter and
+	// connect with no client identity.
+	t.Run("mTLS configured but empty rejected", func(t *testing.T) {
+		c := baseKernelConfig()
+		c.KernelExperimental = &config.KernelExperimentalConfig{TLSClientCertConfigured: true}
+		if _, err := validateKernelConfig(c); err == nil {
+			t.Fatal("expected an error when WithKernelClientCertificate was used with empty cert/key")
+		}
+	})
+	// The same request expressed through the public option (nil, nil): it must be
+	// rejected all the way through, not just when the config is hand-built.
+	t.Run("mTLS via option with empty args rejected", func(t *testing.T) {
+		c := baseKernelConfig()
+		WithKernelClientCertificate(nil, nil)(c)
+		if _, err := validateKernelConfig(c); err == nil {
+			t.Fatal("expected an error when WithKernelClientCertificate(nil, nil) is used")
+		}
+	})
+	// A non-mTLS experimental config (only some other WithKernel* option set) must
+	// still validate — the marker, not merely a non-nil KernelExperimental, gates
+	// the mTLS completeness check.
+	t.Run("non-mTLS experimental config validates", func(t *testing.T) {
+		c := baseKernelConfig()
+		c.KernelExperimental = &config.KernelExperimentalConfig{TLSSkipHostnameVerify: true}
+		if _, err := validateKernelConfig(c); err != nil {
+			t.Errorf("a non-mTLS experimental config should validate, got %v", err)
+		}
+	})
 
 	t.Run("PAT via WithAuthenticator resolves the token", func(t *testing.T) {
 		c := baseKernelConfig()
diff --git a/kernel_experimental_test.go b/kernel_experimental_test.go
index f8c7ac2b..f131ead1 100644
--- a/kernel_experimental_test.go
+++ b/kernel_experimental_test.go
@@ -17,19 +17,22 @@ import (
 // kernel_config tests.
 
 // kernelExperimentalFieldDisposition records how each experimental (kernel-only)
-// option is handled. Every experimental field is forwarded to the kernel C ABI
-// (there is no "inert" experimental knob — they exist precisely because the kernel
-// supports them) and rejected on the Thrift path (the connector fails loud when
-// KernelExperimental is non-nil). A new field on config.KernelExperimentalConfig
-// without an entry here fails TestKernelExperimentalFieldsClassified, forcing a
-// deliberate decision and a setter in KernelBackend.OpenSession so it can't be
+// option is handled. Most experimental fields are "forwarded" to the kernel C ABI
+// (they exist precisely because the kernel supports them) and rejected on the
+// Thrift path (the connector fails loud when KernelExperimental is non-nil). The
+// lone exception is "validation-only" metadata (TLSClientCertConfigured), which is
+// consumed by validateKernelConfig and deliberately never forwarded. A new field
+// on config.KernelExperimentalConfig without an entry here fails
+// TestKernelExperimentalFieldsClassified, forcing a deliberate decision (and, for
+// a forwarded field, a setter in KernelBackend.OpenSession) so it can't be
 // silently dropped.
 var kernelExperimentalFieldDisposition = map[string]string{
-	"TLSTrustedCertsPEM":    "forwarded", // set_tls_trusted_certs
-	"TLSSkipHostnameVerify": "forwarded", // set_tls_skip_hostname_verification
-	"TLSClientCertPEM":      "forwarded", // set_tls_client_certificate (cert half)
-	"TLSClientKeyPEM":       "forwarded", // set_tls_client_certificate (key half)
-	"CloudFetchEnabled":     "forwarded", // set_cloudfetch_enabled
+	"TLSTrustedCertsPEM":      "forwarded",       // set_tls_trusted_certs
+	"TLSSkipHostnameVerify":   "forwarded",       // set_tls_skip_hostname_verification
+	"TLSClientCertPEM":        "forwarded",       // set_tls_client_certificate (cert half)
+	"TLSClientKeyPEM":         "forwarded",       // set_tls_client_certificate (key half)
+	"TLSClientCertConfigured": "validation-only", // consumed by validateKernelConfig; never forwarded
+	"CloudFetchEnabled":       "forwarded",       // set_cloudfetch_enabled
 }
 
 func TestKernelExperimentalFieldsClassified(t *testing.T) {
@@ -70,7 +73,13 @@ func TestWithKernelTLSOptionsSetExperimental(t *testing.T) {
 			return k.TLSSkipHostnameVerify
 		}},
 		{"client certificate", WithKernelClientCertificate([]byte("cert"), []byte("key")), func(k *config.KernelExperimentalConfig) bool {
-			return string(k.TLSClientCertPEM) == "cert" && string(k.TLSClientKeyPEM) == "key"
+			return string(k.TLSClientCertPEM) == "cert" && string(k.TLSClientKeyPEM) == "key" && k.TLSClientCertConfigured
+		}},
+		{"client certificate empty still marks configured", WithKernelClientCertificate(nil, nil), func(k *config.KernelExperimentalConfig) bool {
+			// Even with empty bytes the option must mark itself configured, so
+			// validateKernelConfig can reject the incomplete mTLS request rather
+			// than fail open (connect with no client identity).
+			return k.TLSClientCertConfigured && len(k.TLSClientCertPEM) == 0 && len(k.TLSClientKeyPEM) == 0
 		}},
 		{"cloudfetch off", WithKernelCloudFetch(false), func(k *config.KernelExperimentalConfig) bool {
 			return k.CloudFetchEnabled != nil && !*k.CloudFetchEnabled
@@ -158,11 +167,12 @@ func TestWithKernelTrustedCertsCopiesPEM(t *testing.T) {
 func TestKernelExperimentalDeepCopy(t *testing.T) {
 	cf := false
 	orig := &config.KernelExperimentalConfig{
-		TLSTrustedCertsPEM:    []byte("ca-bundle"),
-		TLSSkipHostnameVerify: true,
-		TLSClientCertPEM:      []byte("cert-pem"),
-		TLSClientKeyPEM:       []byte("key-pem"),
-		CloudFetchEnabled:     &cf,
+		TLSTrustedCertsPEM:      []byte("ca-bundle"),
+		TLSSkipHostnameVerify:   true,
+		TLSClientCertPEM:        []byte("cert-pem"),
+		TLSClientKeyPEM:         []byte("key-pem"),
+		TLSClientCertConfigured: true,
+		CloudFetchEnabled:       &cf,
 	}
 	cp := orig.DeepCopy()
 	if cp == nil || string(cp.TLSTrustedCertsPEM) != "ca-bundle" || !cp.TLSSkipHostnameVerify {
@@ -171,6 +181,9 @@ func TestKernelExperimentalDeepCopy(t *testing.T) {
 	if string(cp.TLSClientCertPEM) != "cert-pem" || string(cp.TLSClientKeyPEM) != "key-pem" {
 		t.Fatalf("DeepCopy lost the mTLS cert/key: %+v", cp)
 	}
+	if !cp.TLSClientCertConfigured {
+		t.Fatalf("DeepCopy lost the TLSClientCertConfigured marker: %+v", cp)
+	}
 	if cp.CloudFetchEnabled == nil || *cp.CloudFetchEnabled != false {
 		t.Fatalf("DeepCopy lost CloudFetchEnabled: %+v", cp)
 	}