diff --git a/KERNEL_REV b/KERNEL_REV
index 95cfce81..87114f3d 100644
--- a/KERNEL_REV
+++ b/KERNEL_REV
@@ -1 +1 @@
-eff8950428f4e6cc9975c663ec919f334962f7d0
+2dd4739f1c20e3a560bcaf52abed8aab9bd957cf
diff --git a/doc.go b/doc.go
index 18673400..3f0ac892 100644
--- a/doc.go
+++ b/doc.go
@@ -195,19 +195,21 @@ 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. Its
+records are forwarded into the driver's logger, so they use the same output as the
+Go and Thrift paths, including a local file configured with logger.SetLogOutput.
+Changing SetLogOutput later safely retargets all three paths. The Rust verbosity is
+fixed when the first kernel session in the process is opened, so set the level
+before that first connect.
 
 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 filtering 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
+	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
 
 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/cgo.go b/internal/backend/kernel/cgo.go
index 3fc26289..20d5b403 100644
--- a/internal/backend/kernel/cgo.go
+++ b/internal/backend/kernel/cgo.go
@@ -37,7 +37,6 @@ import (
 	"context"
 	"fmt"
 	"runtime"
-	"sync"
 	"unsafe"
 
 	"github.com/databricks/databricks-sql-go/driverctx"
@@ -101,56 +100,25 @@ 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 records into the
+// driver's shared logger and maps the driver level into the kernel subscriber.
+// Consequently DATABRICKS_LOG_LEVEL controls both paths and SetLogOutput controls
+// their common destination. DBSQL_KERNEL_DEBUG keeps its advanced behavior: the
+// callback is installed with a NULL level so the kernel honors RUST_LOG instead.
 //
-// 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.
+// Best-effort: failure to install the process-global callback is logged at Warn
+// and never fails a connection. At the default Warn level the kernel filters out
+// lower-level events before crossing cgo.
 //
 // Scope caveat: the kernel 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 later sessions. The driver level is sampled here, once;
+// set it before the first kernel connection. The output is different: the shared
+// logger uses a stable writer proxy, so a later SetLogOutput safely retargets both
+// Go and forwarded Rust records.
 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)
-		}
-	})
+	level, useNULL := resolveKernelLogArg()
+	installKernelLogCallback(level, useNULL)
 }
 
 // call runs a fallible kernel entry point and, on a non-Success status, reads
diff --git a/internal/backend/kernel/log_callback.go b/internal/backend/kernel/log_callback.go
new file mode 100644
index 00000000..bdb7279f
--- /dev/null
+++ b/internal/backend/kernel/log_callback.go
@@ -0,0 +1,86 @@
+//go:build cgo && databricks_kernel
+
+package kernel
+
+/*
+#include <stdlib.h>
+#include "databricks_kernel.h"
+
+// The Go export below is generated with mutable char* parameters, but
+// KernelLogCallback lends read-only const char*. kernelLogAdapter has the exact
+// KernelLogCallback signature and forwards to the Go export, so the function
+// pointer handed to the kernel needs no incompatible function-pointer cast.
+void kernelLogTrampoline(char* level, char* target, char* message, void* user_data);
+static void kernelLogAdapter(const char* level, const char* target,
+                             const char* message, void* user_data) {
+    kernelLogTrampoline((char*)level, (char*)target, (char*)message, user_data);
+}
+static KernelLogCallback kernel_log_cb(void) { return kernelLogAdapter; }
+*/
+import "C"
+
+import (
+	"sync"
+	"time"
+	"unsafe"
+
+	"github.com/databricks/databricks-sql-go/logger"
+)
+
+// This file is the thin cgo layer of the kernel log bridge: the exported callback
+// trampoline and the one-time kernel_init_logging_callback install. The pure-Go
+// pipeline it drives (queue, drain, flush, drop accounting) lives untagged in
+// logforward_async.go so its tests run in the default CGO_ENABLED=0 build.
+
+// logCallbackOnce guards the process-wide, first-call-wins install.
+var logCallbackOnce sync.Once
+
+//export kernelLogTrampoline
+func kernelLogTrampoline(level, target, message *C.char, _ unsafe.Pointer) {
+	// A panic must never cross the C ABI. user_data is deliberately unused: the
+	// kernel is given NULL, and the destination is reached through logQueue, so no
+	// Go pointer is ever fabricated into a C void* (which the GC could fault on).
+	defer func() { _ = recover() }()
+	// time.Now() here is the emission time — the callback fires synchronously on the
+	// kernel thread as the event is logged. C.GoString copies each borrowed string
+	// into owned Go memory before the record can outlive the callback; the rest is
+	// pure Go (see enqueueKernelLog).
+	enqueueKernelLog(time.Now(), C.GoString(level), C.GoString(target), C.GoString(message))
+}
+
+func installKernelLogCallback(level string, useNULL bool) {
+	logCallbackOnce.Do(func() {
+		// OFF intentionally installs no subscriber and starts no drain.
+		if !useNULL && level == "OFF" {
+			return
+		}
+
+		ch := make(chan kernelLogRecord, kernelLogChannelCapacity)
+		// Publish before installing so a callback that fires during
+		// kernel_init_logging_callback already has a channel to enqueue onto;
+		// records buffer until the drain starts just below.
+		logQueue.Store(&ch)
+
+		var clevel cStr
+		if !useNULL {
+			clevel = newCStr(level)
+			defer clevel.free()
+		}
+		// NULL user_data: the drain goroutine owns the sink, so nothing Go-managed
+		// crosses into C as a pointer.
+		if err := call(func() C.KernelStatusCode {
+			return C.kernel_init_logging_callback(clevel.c, C.kernel_log_cb(), nil)
+		}); err != nil {
+			// Install failed, so the callback layer was not installed. Unpublish the
+			// channel; no drain was started and nothing references it, so it is simply
+			// collected — no close (and thus no send-on-closed race to reason about).
+			logQueue.Store(nil)
+			logger.Logger.Warn().Msgf(
+				"databricks: kernel_init_logging_callback: %v (kernel logs not forwarded; proceeding)", err)
+			return
+		}
+		// Installed: start the single drain goroutine. Any records enqueued during
+		// the call above are buffered and delivered once it runs.
+		go drainKernelLogs(ch, newLogSink())
+	})
+}
diff --git a/internal/backend/kernel/log_callback_test.go b/internal/backend/kernel/log_callback_test.go
new file mode 100644
index 00000000..e3247a92
--- /dev/null
+++ b/internal/backend/kernel/log_callback_test.go
@@ -0,0 +1,87 @@
+//go:build cgo && databricks_kernel
+
+package kernel
+
+import (
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/databricks/databricks-sql-go/logger"
+)
+
+const (
+	logFileHelperEnv = "DBSQL_KERNEL_LOG_FILE_HELPER"
+	logFilePathEnv   = "DBSQL_KERNEL_LOG_FILE_PATH"
+	goLogFileProbe   = "go local-file logging probe"
+	rustLogFileProbe = "retry max_wait_ms is below min_wait_ms"
+)
+
+// TestKernelCallbackWritesConfiguredFileEndToEnd proves the user-visible parity
+// contract in a fresh process: the same file passed to logger.SetLogOutput gets a
+// native Go record and a real Rust tracing record delivered through the C ABI.
+// A subprocess is required because the kernel tracing subscriber is process-wide
+// and first-call-wins.
+func TestKernelCallbackWritesConfiguredFileEndToEnd(t *testing.T) {
+	if os.Getenv(logFileHelperEnv) == "1" {
+		path := os.Getenv(logFilePathEnv)
+		file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) //nolint:gosec // Parent supplies its temp path.
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := logger.SetLogLevel("warn"); err != nil {
+			t.Fatal(err)
+		}
+		logger.SetLogOutput(file)
+
+		initKernelLogging()
+		logger.Logger.Warn().Msg(goLogFileProbe)
+		// The C ABI corrects this inverted range and emits a Rust klog::warn!,
+		// giving the test a deterministic kernel-owned record without a server.
+		err = trySetRetry(Config{Retry: &RetryConfig{
+			MinWait:    5 * time.Second,
+			MaxWait:    time.Second,
+			MaxRetries: 1,
+		}})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// The kernel record crosses an async drain goroutine, so flush it into the
+		// file before retargeting the output or closing it — otherwise the drain
+		// could write to stderr (post-retarget) or after Close.
+		if !flushKernelLogs(5 * time.Second) {
+			t.Fatal("kernel log flush timed out")
+		}
+
+		logger.SetLogOutput(os.Stderr)
+		if err := file.Sync(); err != nil {
+			t.Fatal(err)
+		}
+		if err := file.Close(); err != nil {
+			t.Fatal(err)
+		}
+		return
+	}
+
+	logPath := filepath.Join(t.TempDir(), "driver-and-kernel.log")
+	cmd := exec.Command(os.Args[0], "-test.run=^TestKernelCallbackWritesConfiguredFileEndToEnd$") //nolint:gosec // Re-executes this test binary only.
+	cmd.Env = append(os.Environ(), logFileHelperEnv+"=1", logFilePathEnv+"="+logPath)
+	if output, err := cmd.CombinedOutput(); err != nil {
+		t.Fatalf("logging helper failed: %v\n%s", err, output)
+	}
+	contents, err := os.ReadFile(logPath) //nolint:gosec // Test-owned temporary path.
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := string(contents)
+	if !strings.Contains(got, goLogFileProbe) {
+		t.Errorf("local log file is missing Go record: %q", got)
+	}
+	if !strings.Contains(got, rustLogFileProbe) {
+		t.Errorf("local log file is missing Rust kernel record: %q", got)
+	}
+}
diff --git a/internal/backend/kernel/logforward.go b/internal/backend/kernel/logforward.go
new file mode 100644
index 00000000..673de63a
--- /dev/null
+++ b/internal/backend/kernel/logforward.go
@@ -0,0 +1,70 @@
+package kernel
+
+import (
+	"strings"
+	"time"
+
+	"github.com/databricks/databricks-sql-go/logger"
+	"github.com/rs/zerolog"
+)
+
+// logSink is the Go destination for kernel tracing records. It forwards through
+// logger.ForwardingSink (not a logger derived from logger.Logger), so it (a) still
+// follows SetLogOutput, (b) is ungated at TraceLevel because the kernel already
+// applied its configured level, (c) carries no auto-timestamp hook — forward stamps
+// each record with the emission time captured on the kernel thread, not the drain
+// time — and (d) cannot be round-tripped into SetLogOutput (ForwardingSink is not
+// an io.Writer), which would deadlock.
+type logSink struct {
+	sink    *logger.ForwardingSink
+	observe func(level, target, message string)
+}
+
+func newLogSink() *logSink {
+	return &logSink{sink: logger.NewForwardingSink()}
+}
+
+// forward writes one kernel record. emittedAt is the time the kernel emitted the
+// event (captured in the cgo callback), stamped as the record's timestamp so a
+// backed-up drain does not skew kernel log times toward drain time.
+func (s *logSink) forward(emittedAt time.Time, level, target, message string) {
+	if s == nil {
+		return
+	}
+	if s.observe != nil {
+		s.observe(level, target, message)
+	}
+	s.event(level).Time(zerolog.TimestampFieldName, emittedAt).Str("target", target).Msg(message)
+}
+
+// warnDropped emits a one-shot advisory that forwarded records were dropped. It
+// goes through the sink's own immutable logger — not logger.Logger, whose embedded
+// value SetLogLevel reassigns — so the long-lived drain goroutine never races
+// SetLogLevel. Like forwarded records it is ungated, which is what we want: log loss
+// should surface regardless of the driver level.
+func (s *logSink) warnDropped(dropped uint64) {
+	s.sink.Event(zerolog.WarnLevel).
+		Uint64("dropped", dropped).
+		Time(zerolog.TimestampFieldName, time.Now()).
+		Msg("[kernel] kernel log records dropped; the log sink is not keeping up " +
+			"(raise capacity or lower kernel verbosity)")
+}
+
+// event picks the zerolog event for a kernel level string. An unknown level maps
+// to Debug and preserves the raw kernel level as a field.
+func (s *logSink) event(level string) *zerolog.Event {
+	switch strings.ToLower(level) {
+	case "error":
+		return s.sink.Event(zerolog.ErrorLevel)
+	case "warn":
+		return s.sink.Event(zerolog.WarnLevel)
+	case "info":
+		return s.sink.Event(zerolog.InfoLevel)
+	case "debug":
+		return s.sink.Event(zerolog.DebugLevel)
+	case "trace":
+		return s.sink.Event(zerolog.TraceLevel)
+	default:
+		return s.sink.Event(zerolog.DebugLevel).Str("kernelLevel", level)
+	}
+}
diff --git a/internal/backend/kernel/logforward_async.go b/internal/backend/kernel/logforward_async.go
new file mode 100644
index 00000000..e1b080ba
--- /dev/null
+++ b/internal/backend/kernel/logforward_async.go
@@ -0,0 +1,136 @@
+package kernel
+
+// This file is intentionally NOT behind the `cgo && databricks_kernel` build tag
+// (matching logging_level.go and logforward.go). It holds the pure-Go async
+// forwarding pipeline — the bounded hand-off queue, its drain, the flush barrier,
+// drop accounting, and panic containment. Only the cgo trampoline and the
+// kernel_init_logging_callback call actually need cgo; keeping the rest untagged
+// lets its tests (FIFO flush, drop policy, panic containment) run in the default
+// CGO_ENABLED=0 build rather than only in the kernel-linked lane.
+
+import (
+	"sync/atomic"
+	"time"
+)
+
+// kernelLogRecord is an owned copy of one kernel tracing record. The C strings are
+// valid only for the duration of the callback, so the trampoline copies them before
+// the record leaves the kernel thread. A record with a non-nil done channel is a
+// flush barrier: the drain closes done (in FIFO order, after every earlier record is
+// written) and forwards nothing — see flushKernelLogs.
+type kernelLogRecord struct {
+	emittedAt time.Time
+	level     string
+	target    string
+	message   string
+	done      chan struct{}
+}
+
+// kernelLogChannelCapacity bounds the hand-off buffer between kernel threads and the
+// drain goroutine. Bursts beyond this are dropped rather than blocking a kernel
+// thread — logs are advisory and must never back-pressure a kernel path.
+const kernelLogChannelCapacity = 4096
+
+var (
+	// logQueue publishes the bounded hand-off channel to the trampoline. It is an
+	// atomic pointer so the read on a kernel thread synchronizes with the write on
+	// the installing goroutine; nil until (and unless) logging installs.
+	logQueue atomic.Pointer[chan kernelLogRecord]
+	// logDropped counts records discarded because the buffer was full — a growing
+	// value means the sink cannot keep up. Exposed via kernelLogDropped.
+	logDropped atomic.Uint64
+)
+
+// kernelLogDropped reports how many forwarded kernel records were dropped because
+// the bounded hand-off buffer was full. Safe to call at any time.
+func kernelLogDropped() uint64 { return logDropped.Load() }
+
+// enqueueKernelLog hands one already-owned record to the drain goroutine without
+// blocking. Split out of the cgo trampoline so the enqueue/drop policy is testable
+// without cgo (import "C" is not allowed in _test.go files).
+func enqueueKernelLog(emittedAt time.Time, level, target, message string) {
+	qp := logQueue.Load()
+	if qp == nil {
+		return
+	}
+	rec := kernelLogRecord{emittedAt: emittedAt, level: level, target: target, message: message}
+	// Non-blocking hand-off: never stall a kernel thread on a slow, contended, or
+	// re-entrant user writer. A full buffer drops the record and counts it.
+	select {
+	case *qp <- rec:
+	default:
+		logDropped.Add(1)
+	}
+}
+
+// drainKernelLogs is the single goroutine that moves records off kernel threads and
+// into the shared logger. Running the arbitrary user writer here — not in the
+// trampoline — keeps user I/O (and any driver re-entry it triggers) off the kernel
+// thread, honoring the C ABI's "return promptly / no re-entry" contract.
+func drainKernelLogs(ch <-chan kernelLogRecord, sink *logSink) {
+	baselineDrops := logDropped.Load()
+	warnedDrop := false
+	for rec := range ch {
+		// A flush barrier carries no record: closing done signals that every earlier
+		// record has been written (channel + drain are FIFO).
+		if rec.done != nil {
+			close(rec.done)
+			continue
+		}
+		// Every write below goes to the user's writer, which may panic — an
+		// unrecovered goroutine panic is fatal to the process, so contain each one.
+		contain(func() {
+			sink.forward(rec.emittedAt, rec.level, rec.target, rec.message)
+		})
+		// Surface log loss the first time the sink falls behind. Routed through the
+		// sink's own (immutable) logger, not logger.Logger, so this long-lived
+		// goroutine never races SetLogLevel's reassignment of Logger.Logger. One-shot
+		// so a burst can't turn into log spam; the total stays in kernelLogDropped().
+		if !warnedDrop && logDropped.Load() > baselineDrops {
+			warnedDrop = true
+			dropped := logDropped.Load() - baselineDrops
+			contain(func() { sink.warnDropped(dropped) })
+		}
+	}
+}
+
+// contain runs fn, swallowing any panic. A misbehaving user writer (reached via the
+// sink) must never take down the drain goroutine.
+func contain(fn func()) {
+	defer func() { _ = recover() }()
+	fn()
+}
+
+// flushKernelLogs blocks until every kernel record already queued has been written,
+// or until timeout elapses; it returns whether the flush completed. It is a no-op
+// returning true when kernel logging was never installed.
+//
+// It drains the asynchronous hand-off so records are not lost or misrouted when the
+// log writer is closed, output is retargeted, or the process exits. There is
+// currently no public entry point that calls it — only the end-to-end test does;
+// exposing a supported flush API (or wiring it into a shutdown path) is a separate
+// change. Best-effort: records already dropped for a full buffer are gone, and
+// records enqueued after this call are not waited on.
+func flushKernelLogs(timeout time.Duration) bool {
+	qp := logQueue.Load()
+	if qp == nil {
+		return true
+	}
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	done := make(chan struct{})
+	// Enqueue the barrier behind everything already queued. A full buffer means the
+	// drain is behind; wait (bounded) for room rather than dropping the barrier.
+	select {
+	case *qp <- kernelLogRecord{done: done}:
+	case <-timer.C:
+		return false
+	}
+	select {
+	case <-done:
+		return true
+	case <-timer.C:
+		return false
+	}
+}
diff --git a/internal/backend/kernel/logforward_async_test.go b/internal/backend/kernel/logforward_async_test.go
new file mode 100644
index 00000000..e18b3c42
--- /dev/null
+++ b/internal/backend/kernel/logforward_async_test.go
@@ -0,0 +1,165 @@
+package kernel
+
+// These exercise the pure-Go async forwarding pipeline (logforward_async.go) and
+// are intentionally untagged, so the FIFO-flush, drop-policy, and panic-containment
+// guarantees run in the default CGO_ENABLED=0 build — not only the kernel-linked
+// lane. The cgo trampoline that feeds this pipeline is covered by the end-to-end
+// test in log_callback_test.go.
+
+import (
+	"io"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/databricks/databricks-sql-go/logger"
+)
+
+// The forward path enqueues an owned record onto the bounded channel. (The cgo
+// trampoline copies the borrowed C strings via C.GoString, then calls this; that
+// tiny boundary can't be driven from a _test.go file because import "C" is
+// disallowed there, so the testable logic lives in enqueueKernelLog.)
+func TestEnqueueKernelLog(t *testing.T) {
+	ch := make(chan kernelLogRecord, 1)
+	prev := logQueue.Swap(&ch)
+	t.Cleanup(func() { logQueue.Store(prev) })
+
+	emittedAt := time.Now()
+	enqueueKernelLog(emittedAt, "debug", "databricks::sql::kernel", "callback probe")
+	select {
+	case got := <-ch:
+		if got.level != "debug" || got.target != "databricks::sql::kernel" || got.message != "callback probe" {
+			t.Fatalf("enqueued record = %#v", got)
+		}
+		if !got.emittedAt.Equal(emittedAt) {
+			t.Fatalf("emittedAt = %v, want %v", got.emittedAt, emittedAt)
+		}
+	default:
+		t.Fatal("record was not enqueued")
+	}
+}
+
+// A full buffer drops the record and counts it rather than blocking the kernel
+// thread. A nil queue (logging not installed) is a safe no-op.
+func TestEnqueueKernelLogDropsWhenFullAndNoopWhenUnset(t *testing.T) {
+	// Unset queue: must not block or panic.
+	prev := logQueue.Swap(nil)
+	t.Cleanup(func() { logQueue.Store(prev) })
+	enqueueKernelLog(time.Now(), "warn", "t", "before install")
+
+	ch := make(chan kernelLogRecord, 1)
+	logQueue.Store(&ch)
+	before := kernelLogDropped()
+	enqueueKernelLog(time.Now(), "warn", "t", "keeps the one slot") // fills the buffer
+	enqueueKernelLog(time.Now(), "warn", "t", "must be dropped")    // buffer full → dropped
+	if got := kernelLogDropped() - before; got != 1 {
+		t.Fatalf("dropped delta = %d, want 1", got)
+	}
+	if len(ch) != 1 {
+		t.Fatalf("channel len = %d, want 1 (non-blocking drop)", len(ch))
+	}
+}
+
+// A panicking writer must not kill the drain goroutine (an unrecovered goroutine
+// panic is fatal to the process); the drain contains it and keeps processing.
+func TestDrainRecoversFromWriterPanic(t *testing.T) {
+	t.Cleanup(func() { logger.SetLogOutput(os.Stderr) })
+	logger.SetLogOutput(io.Discard) // discard the sink's forwarded records
+	done := make(chan string, 1)
+	sink := newLogSink()
+	sink.observe = func(_, _, message string) {
+		if message == "boom" {
+			panic("writer failure")
+		}
+		done <- message
+	}
+
+	ch := make(chan kernelLogRecord, 2)
+	go drainKernelLogs(ch, sink)
+	defer close(ch)
+
+	ch <- kernelLogRecord{level: "error", target: "t", message: "boom"} // triggers the panic
+	ch <- kernelLogRecord{level: "info", target: "t", message: "after"} // must still be delivered
+	select {
+	case got := <-done:
+		if got != "after" {
+			t.Fatalf("delivered %q, want %q", got, "after")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("drain goroutine did not survive a writer panic")
+	}
+}
+
+// flushKernelLogs returns only after every record queued before it is written.
+func TestFlushKernelLogsWaitsForQueued(t *testing.T) {
+	t.Cleanup(func() { logger.SetLogOutput(os.Stderr) })
+	logger.SetLogOutput(io.Discard) // discard the sink's forwarded records
+	seen := make(chan string, 8)
+	sink := newLogSink()
+	sink.observe = func(_, _, message string) { seen <- message }
+
+	ch := make(chan kernelLogRecord, 8)
+	prev := logQueue.Swap(&ch)
+	t.Cleanup(func() { logQueue.Store(prev) })
+	go drainKernelLogs(ch, sink)
+	defer close(ch)
+
+	enqueueKernelLog(time.Now(), "info", "t", "one")
+	enqueueKernelLog(time.Now(), "info", "t", "two")
+	if !flushKernelLogs(2 * time.Second) {
+		t.Fatal("flush timed out")
+	}
+	// The barrier is FIFO-ordered behind both records, so both are delivered.
+	if got := len(seen); got != 2 {
+		t.Fatalf("after flush, delivered %d records, want 2", got)
+	}
+}
+
+// flushKernelLogs is a no-op returning true when logging was never installed.
+func TestFlushKernelLogsNoopWhenUnset(t *testing.T) {
+	prev := logQueue.Swap(nil)
+	t.Cleanup(func() { logQueue.Store(prev) })
+	if !flushKernelLogs(time.Second) {
+		t.Fatal("flush with no queue should be a no-op returning true")
+	}
+}
+
+type panicWriter struct{}
+
+func (panicWriter) Write([]byte) (int, error) { panic("writer always panics") }
+
+// A panicking writer combined with dropped records must not crash the drain: the
+// forwarded record AND the one-shot drop warning both write to that writer, and
+// both must be contained. observe advances the drop counter past the drain's
+// baseline (so the warning fires) and panics (so forward is exercised too); the
+// shared output is the panicking writer (so the warning write panics).
+func TestDrainContainsPanicFromForwardAndDropWarning(t *testing.T) {
+	prevLevel := logger.Logger.GetLevel()
+	t.Cleanup(func() {
+		logger.SetLogOutput(os.Stderr)
+		logger.Logger.Logger = logger.Logger.Level(prevLevel)
+	})
+	if err := logger.SetLogLevel("warn"); err != nil {
+		t.Fatal(err)
+	}
+	logger.SetLogOutput(panicWriter{})
+
+	sink := newLogSink()
+	sink.observe = func(_, _, _ string) {
+		logDropped.Add(1) // advance past the drain's baseline, deterministically
+		panic("forward failure")
+	}
+
+	ch := make(chan kernelLogRecord, 4)
+	prev := logQueue.Swap(&ch)
+	t.Cleanup(func() { logQueue.Store(prev) })
+	go drainKernelLogs(ch, sink)
+	defer close(ch)
+
+	enqueueKernelLog(time.Now(), "info", "databricks::sql::kernel", "boom")
+
+	// The drain reaches the flush barrier only if it survived both panics.
+	if !flushKernelLogs(2 * time.Second) {
+		t.Fatal("drain did not survive a panicking writer combined with dropped records")
+	}
+}
diff --git a/internal/backend/kernel/logforward_test.go b/internal/backend/kernel/logforward_test.go
new file mode 100644
index 00000000..b0c80683
--- /dev/null
+++ b/internal/backend/kernel/logforward_test.go
@@ -0,0 +1,102 @@
+package kernel
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/databricks/databricks-sql-go/logger"
+	"github.com/rs/zerolog"
+)
+
+func TestLogSinkForwardMapsLevels(t *testing.T) {
+	cases := []struct {
+		level string
+		want  string
+	}{
+		{"error", "error"},
+		{"warn", "warn"},
+		{"info", "info"},
+		{"debug", "debug"},
+		{"trace", "trace"},
+		{"future", "debug"},
+	}
+	t.Cleanup(func() { logger.SetLogOutput(os.Stderr) })
+	emittedAt := time.Now()
+	for _, tc := range cases {
+		var buf bytes.Buffer
+		// The sink forwards through the shared output; point it at a buffer to
+		// capture the single record and confirm it is one hook-free JSON line whose
+		// only timestamp is the emission time forward stamps.
+		logger.SetLogOutput(&buf)
+		sink := newLogSink()
+		sink.forward(emittedAt, tc.level, "databricks::sql::kernel", "hello")
+		var record map[string]any
+		if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &record); err != nil {
+			t.Fatalf("level %q: %v", tc.level, err)
+		}
+		if record["level"] != tc.want || record["target"] != "databricks::sql::kernel" || record["message"] != "hello" {
+			t.Errorf("level %q: record = %#v", tc.level, record)
+		}
+		if _, ok := record[zerolog.TimestampFieldName]; !ok {
+			t.Errorf("level %q: record missing %q field: %#v", tc.level, zerolog.TimestampFieldName, record)
+		}
+	}
+}
+
+func TestLogSinkFollowsLocalFileRetarget(t *testing.T) {
+	prevLevel := logger.Logger.GetLevel()
+	t.Cleanup(func() {
+		logger.SetLogOutput(os.Stderr)
+		logger.Logger.Logger = logger.Logger.Level(prevLevel)
+	})
+
+	first, err := os.CreateTemp(t.TempDir(), "kernel-first-*.log")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := first.Close(); err != nil {
+			t.Errorf("close first log: %v", err)
+		}
+	})
+	second, err := os.CreateTemp(t.TempDir(), "kernel-second-*.log")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := second.Close(); err != nil {
+			t.Errorf("close second log: %v", err)
+		}
+	})
+
+	logger.SetLogOutput(first)
+	sink := newLogSink()
+	sink.forward(time.Now(), "debug", "databricks::sql::kernel", "kernel first destination")
+	logger.SetLogOutput(second)
+	sink.forward(time.Now(), "warn", "databricks::sql::kernel", "kernel second destination")
+
+	if err := first.Sync(); err != nil {
+		t.Fatal(err)
+	}
+	if err := second.Sync(); err != nil {
+		t.Fatal(err)
+	}
+	firstBytes, err := os.ReadFile(first.Name())
+	if err != nil {
+		t.Fatal(err)
+	}
+	secondBytes, err := os.ReadFile(second.Name())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := string(firstBytes); !strings.Contains(got, "kernel first destination") || strings.Contains(got, "kernel second destination") {
+		t.Fatalf("first log contents = %q", got)
+	}
+	if got := string(secondBytes); !strings.Contains(got, "kernel second destination") || strings.Contains(got, "kernel first destination") {
+		t.Fatalf("second log contents = %q", got)
+	}
+}
diff --git a/logger/logger.go b/logger/logger.go
index 683501a1..e858dcf3 100644
--- a/logger/logger.go
+++ b/logger/logger.go
@@ -4,6 +4,7 @@ import (
 	"io"
 	"os"
 	"runtime"
+	"sync/atomic"
 	"time"
 
 	"github.com/mattn/go-isatty"
@@ -14,6 +15,70 @@ type DBSQLLogger struct {
 	zerolog.Logger
 }
 
+// sharedOutput is the process-wide destination behind every driver logger,
+// including logger values derived before a later SetLogOutput call. The current
+// destination is held behind an atomic pointer and wrapped in a per-destination
+// synchronized, level-aware writer. This gives three properties the driver's
+// logging (Go, Thrift, and forwarded kernel records) relies on:
+//
+//   - Retargeting (set) is a single atomic store: it never waits on an in-flight
+//     user Write, is never blocked by a slow/stuck writer, and never holds a lock
+//     across arbitrary user code — so SetLogOutput can't self-deadlock and a stuck
+//     writer can always be replaced.
+//   - Concurrent records to one destination stay intact (zerolog.SyncWriter
+//     serializes writes per destination).
+//   - A zerolog.LevelWriter destination keeps severity-aware routing: the proxy
+//     itself implements LevelWriter, so zerolog calls WriteLevel on it and it
+//     forwards WriteLevel to the destination.
+type sharedOutput struct {
+	dst atomic.Pointer[zerolog.LevelWriter]
+}
+
+func newSharedOutput(w io.Writer) *sharedOutput {
+	o := &sharedOutput{}
+	o.set(w)
+	return o
+}
+
+// set publishes w as the current destination. A nil writer is normalized to
+// io.Discard, matching the historical zerolog.New(nil) behavior (before the proxy
+// existed, SetLogOutput(nil) → Logger.Output(nil) → io.Discard); without this the
+// next log would panic on a nil-interface Write.
+func (o *sharedOutput) set(w io.Writer) {
+	if w == nil {
+		w = io.Discard
+	}
+	// SyncWriter serializes concurrent writes to this destination and preserves a
+	// LevelWriter's WriteLevel (a plain writer is adapted). The result always
+	// implements LevelWriter; keep a fallback adapter in case that ever changes.
+	sw := zerolog.SyncWriter(w)
+	lw, ok := sw.(zerolog.LevelWriter)
+	if !ok {
+		lw = plainLevelWriter{sw}
+	}
+	o.dst.Store(&lw)
+}
+
+func (o *sharedOutput) current() zerolog.LevelWriter {
+	return *o.dst.Load()
+}
+
+func (o *sharedOutput) Write(p []byte) (int, error) {
+	return o.current().Write(p)
+}
+
+func (o *sharedOutput) WriteLevel(l zerolog.Level, p []byte) (int, error) {
+	return o.current().WriteLevel(l, p)
+}
+
+// plainLevelWriter adapts an io.Writer that is not a zerolog.LevelWriter, routing
+// WriteLevel to Write (dropping the level, as zerolog's own adapter does).
+type plainLevelWriter struct{ io.Writer }
+
+func (p plainLevelWriter) WriteLevel(_ zerolog.Level, b []byte) (int, error) {
+	return p.Write(b)
+}
+
 // Track is a simple utility function to use with logger to log a message with a timestamp.
 // Recommended to use in conjunction with Duration.
 //
@@ -36,15 +101,15 @@ func (l *DBSQLLogger) Duration(msg string, start time.Time) {
 	l.Debug().Msgf("%v elapsed time: %v", msg, time.Since(start))
 }
 
-var Logger = &DBSQLLogger{
-	zerolog.New(os.Stderr).With().Timestamp().Logger(),
-}
+var output = newSharedOutput(os.Stderr)
+
+var Logger = &DBSQLLogger{zerolog.New(output).With().Timestamp().Logger()}
 
 // Enable pretty printing for interactive terminals and json for production.
 func init() {
 	// for tty terminal enable pretty logs
 	if isatty.IsTerminal(os.Stdout.Fd()) && runtime.GOOS != "windows" {
-		Logger = &DBSQLLogger{Logger.Output(zerolog.ConsoleWriter{Out: os.Stderr})}
+		output.set(zerolog.ConsoleWriter{Out: os.Stderr})
 	}
 	// by default only log warns or above
 	loglvl := zerolog.WarnLevel
@@ -71,8 +136,49 @@ func SetLogLevel(l string) error {
 }
 
 // Sets logging output. Default is os.Stderr. If in terminal, pretty logs are enabled.
+// A nil writer is treated as io.Discard. Existing logger values (and the kernel
+// log bridge) follow later calls to this function.
+//
+// Writes are serialized per destination. If you hot-swap the output under
+// concurrent logging while reusing the same underlying writer across swaps, pass a
+// writer that is safe for concurrent use (os.File is; a bare bytes.Buffer is not):
+// records in flight across a swap are serialized by that writer, not by the driver.
+//
+// The writer must not log through this driver from within its own Write/WriteLevel,
+// directly or indirectly: per-destination serialization holds a non-reentrant lock
+// across the write, so a re-entrant writer deadlocks (and a truly self-referential
+// one would recurse without bound regardless). As with any logging library, do not
+// log from your log destination.
 func SetLogOutput(w io.Writer) {
-	Logger.Logger = Logger.Output(w)
+	output.set(w)
+}
+
+// ForwardingSink emits already-rendered log records from an external source (such
+// as the Rust kernel) to the driver's shared destination. It follows SetLogOutput
+// and bypasses Logger's level gate and timestamp hook, so the source can supply its
+// own level and its own (emission-time) timestamp.
+//
+// It is deliberately NOT an io.Writer — and neither is the *zerolog.Event it hands
+// out. A forwarding sink that is an io.Writer can be passed to SetLogOutput, which
+// wraps it in a SyncWriter and stores it as the destination; the sink's own writes
+// then route back through that same SyncWriter (output → SyncWriter → sink →
+// output), and because SyncWriter holds its mutex across the write, the next record
+// deadlocks. Handing out a Logger fails the same way (zerolog.Logger implements
+// io.Writer). A method-only sink cannot form that cycle.
+type ForwardingSink struct {
+	log zerolog.Logger
+}
+
+// NewForwardingSink returns a sink over the shared output, ungated at TraceLevel
+// because the external source has already applied its own level filter.
+func NewForwardingSink() *ForwardingSink {
+	return &ForwardingSink{log: zerolog.New(output).Level(zerolog.TraceLevel)}
+}
+
+// Event begins a record at level on the shared destination; the caller adds fields
+// and terminates with Msg. Like Logger, it follows SetLogOutput.
+func (s *ForwardingSink) Event(level zerolog.Level) *zerolog.Event {
+	return s.log.WithLevel(level)
 }
 
 // Sets log to trace. -1
diff --git a/logger/logger_test.go b/logger/logger_test.go
new file mode 100644
index 00000000..b5448f03
--- /dev/null
+++ b/logger/logger_test.go
@@ -0,0 +1,226 @@
+package logger
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/rs/zerolog"
+)
+
+// Existing logger values must follow SetLogOutput. The kernel callback keeps an
+// immutable Trace-level logger value for thread safety, so retargeting only works
+// when every value writes through the same stable output proxy.
+func TestSetLogOutputRetargetsExistingLogger(t *testing.T) {
+	prevLevel := Logger.GetLevel()
+	t.Cleanup(func() {
+		SetLogOutput(os.Stderr)
+		Logger.Logger = Logger.Level(prevLevel)
+	})
+
+	first, err := os.CreateTemp(t.TempDir(), "driver-first-*.log")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := first.Close(); err != nil {
+			t.Errorf("close first log: %v", err)
+		}
+	})
+	second, err := os.CreateTemp(t.TempDir(), "driver-second-*.log")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := second.Close(); err != nil {
+			t.Errorf("close second log: %v", err)
+		}
+	})
+
+	snapshot := Logger.Level(zerolog.TraceLevel)
+	SetLogOutput(first)
+	snapshot.Info().Msg("first destination")
+	SetLogOutput(second)
+	snapshot.Info().Msg("second destination")
+
+	if err := first.Sync(); err != nil {
+		t.Fatal(err)
+	}
+	if err := second.Sync(); err != nil {
+		t.Fatal(err)
+	}
+	firstBytes, err := os.ReadFile(first.Name())
+	if err != nil {
+		t.Fatal(err)
+	}
+	secondBytes, err := os.ReadFile(second.Name())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := string(firstBytes); !strings.Contains(got, "first destination") || strings.Contains(got, "second destination") {
+		t.Fatalf("first log contents = %q", got)
+	}
+	if got := string(secondBytes); !strings.Contains(got, "second destination") || strings.Contains(got, "first destination") {
+		t.Fatalf("second log contents = %q", got)
+	}
+}
+
+// SetLogOutput(nil) must normalize to io.Discard, not store a nil writer that
+// panics on the next enabled log. (Before the shared proxy, zerolog.New(nil)
+// handled this; the proxy must preserve it.)
+func TestSetLogOutputNilDiscards(t *testing.T) {
+	prevLevel := Logger.GetLevel()
+	t.Cleanup(func() {
+		SetLogOutput(os.Stderr)
+		Logger.Logger = Logger.Level(prevLevel)
+	})
+
+	SetLogOutput(nil)
+	Logger.Logger = Logger.Level(zerolog.TraceLevel)
+	// An enabled log actually reaches the writer; a nil writer would panic here.
+	Logger.Info().Msg("after nil output is discarded, not panicked")
+}
+
+// recordingLevelWriter records whether zerolog reaches it via WriteLevel (level
+// preserved) or Write (level lost).
+type recordingLevelWriter struct {
+	mu     sync.Mutex
+	levels []zerolog.Level
+	plain  int
+}
+
+func (w *recordingLevelWriter) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.plain++
+	return len(p), nil
+}
+
+func (w *recordingLevelWriter) WriteLevel(l zerolog.Level, p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.levels = append(w.levels, l)
+	return len(p), nil
+}
+
+// A LevelWriter destination supplied via SetLogOutput must keep receiving
+// WriteLevel (severity-aware routing), not be flattened to Write by the proxy.
+func TestSetLogOutputPreservesLevelWriter(t *testing.T) {
+	prevLevel := Logger.GetLevel()
+	t.Cleanup(func() {
+		SetLogOutput(os.Stderr)
+		Logger.Logger = Logger.Level(prevLevel)
+	})
+
+	lw := &recordingLevelWriter{}
+	SetLogOutput(lw)
+	Logger.Logger = Logger.Level(zerolog.TraceLevel)
+	Logger.Warn().Msg("severity aware")
+
+	lw.mu.Lock()
+	defer lw.mu.Unlock()
+	if lw.plain != 0 {
+		t.Fatalf("destination received %d plain Write calls; expected WriteLevel only", lw.plain)
+	}
+	foundWarn := false
+	for _, l := range lw.levels {
+		if l == zerolog.WarnLevel {
+			foundWarn = true
+		}
+	}
+	if !foundWarn {
+		t.Fatalf("WriteLevel not called with WarnLevel; severity routing lost (levels=%v)", lw.levels)
+	}
+}
+
+// A ForwardingSink must not be an io.Writer. If it were, SetLogOutput(sink) would
+// store it as the destination and its own writes would route back through the
+// SyncWriter that wraps it (output -> SyncWriter -> sink -> output), deadlocking
+// the next record on the re-entered mutex — the same trap that made returning a
+// zerolog.Logger (which implements io.Writer) unsafe.
+func TestForwardingSinkIsNotAnIOWriter(t *testing.T) {
+	if _, ok := any(NewForwardingSink()).(io.Writer); ok {
+		t.Fatal("ForwardingSink must not implement io.Writer (would enable a self-referential SetLogOutput deadlock)")
+	}
+}
+
+// countingWriter is a self-synchronized destination: it counts complete records
+// and flags any that isn't a single well-formed JSON line (which is what byte
+// interleaving from unsynchronized concurrent writes would produce).
+type countingWriter struct {
+	mu    sync.Mutex
+	lines int
+	bad   int
+}
+
+func (w *countingWriter) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if json.Valid(bytes.TrimSpace(p)) {
+		w.lines++
+	} else {
+		w.bad++
+	}
+	return len(p), nil
+}
+
+// Hot-swapping SetLogOutput under concurrent logging must not corrupt output when
+// the destination is safe for concurrent use — even when the same writer is
+// reapplied (each set builds a fresh SyncWriter, so the writer's own lock, not the
+// driver's, is what serializes writes straddling a swap). Run under -race.
+func TestConcurrentRetargetToSyncWriter(t *testing.T) {
+	prevLevel := Logger.GetLevel()
+	t.Cleanup(func() {
+		SetLogOutput(os.Stderr)
+		Logger.Logger = Logger.Level(prevLevel)
+	})
+
+	w := &countingWriter{}
+	SetLogOutput(w)
+	snapshot := Logger.Level(zerolog.TraceLevel)
+
+	const writers, perWriter = 8, 200
+
+	// Retargeter: reapply the same writer until stopped, each call replacing the
+	// SyncWriter wrapper while writes are in flight.
+	stop := make(chan struct{})
+	retargeterDone := make(chan struct{})
+	go func() {
+		defer close(retargeterDone)
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+				SetLogOutput(w)
+			}
+		}
+	}()
+
+	var wg sync.WaitGroup
+	wg.Add(writers)
+	for i := 0; i < writers; i++ {
+		go func() {
+			defer wg.Done()
+			for j := 0; j < perWriter; j++ {
+				snapshot.Info().Int("j", j).Msg("concurrent")
+			}
+		}()
+	}
+	wg.Wait()
+	close(stop)
+	<-retargeterDone
+
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.bad != 0 {
+		t.Fatalf("observed %d corrupted (non-JSON) writes", w.bad)
+	}
+	if w.lines != writers*perWriter {
+		t.Fatalf("wrote %d complete records, want %d", w.lines, writers*perWriter)
+	}
+}