diff --git a/driver/kubernetes/driver_test.go b/driver/kubernetes/driver_test.go
index 3e3f2ec7eaaa..bd72a7824b48 100644
--- a/driver/kubernetes/driver_test.go
+++ b/driver/kubernetes/driver_test.go
@@ -1,6 +1,8 @@
 package kubernetes
 
 import (
+	"context"
+	stderrors "errors"
 	"testing"
 	"time"
 
@@ -39,3 +41,39 @@ func TestCalculateBackoff(t *testing.T) {
 		}
 	}
 }
+
+func TestTryWithBackoffRetriesTransient(t *testing.T) {
+	var calls int
+	err := tryWithBackoff(context.Background(), "test-pod", func() error {
+		calls++
+		if calls == 1 {
+			return stderrors.New("unable to upgrade connection: remote error: tls: internal error")
+		}
+		return nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, 2, calls, "second attempt should succeed after a transient failure")
+}
+
+func TestTryWithBackoffPermanentError(t *testing.T) {
+	permErr := stderrors.New("pods is forbidden")
+	var calls int
+	err := tryWithBackoff(context.Background(), "test-pod", func() error {
+		calls++
+		return permErr
+	})
+	require.ErrorIs(t, err, permErr)
+	require.Equal(t, 1, calls, "a permanent error must not be retried")
+}
+
+func TestTryWithBackoffContextCancelled(t *testing.T) {
+	ctx, cancel := context.WithCancelCause(context.Background())
+	var calls int
+	err := tryWithBackoff(ctx, "test-pod", func() error {
+		calls++
+		cancel(context.Canceled)
+		return stderrors.New("tls: internal error")
+	})
+	require.ErrorIs(t, err, context.Canceled)
+	require.Equal(t, 1, calls, "retry loop must stop once the context is cancelled")
+}
diff --git a/driver/kubernetes/execconn/execconn.go b/driver/kubernetes/execconn/execconn.go
index 17a787ed2a93..e5b878578749 100644
--- a/driver/kubernetes/execconn/execconn.go
+++ b/driver/kubernetes/execconn/execconn.go
@@ -9,12 +9,15 @@ import (
 	"time"
 
 	"github.com/docker/buildx/driver/kubernetes/kubeclient"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/remotecommand"
 )
 
+var errStreamEndedBeforeReady = errors.New("exec stream ended before the connection was established")
+
 func ExecConn(ctx context.Context, restClient rest.Interface, restConfig *rest.Config, namespace, pod, container string, cmd []string) (net.Conn, error) {
 	req := restClient.
 		Post().
@@ -34,12 +37,19 @@ func ExecConn(ctx context.Context, restClient rest.Interface, restConfig *rest.C
 	if err != nil {
 		return nil, err
 	}
-	return newExecConn(ctx, exec), nil
+	return newExecConn(ctx, exec)
 }
 
 // newExecConn wires a remotecommand.Executor's stdin/stdout streams up as a net.Conn.
 // It is split from ExecConn to ease testing.
-func newExecConn(ctx context.Context, exec remotecommand.Executor) net.Conn {
+//
+// StreamWithContext sets up the exec stream synchronously and then blocks for its
+// whole lifetime, so it runs in a goroutine. newExecConn waits for the stream to be
+// established before returning, otherwise a failed setup (e.g. "tls: internal error"
+// on a not-yet-ready node) would only show up on the first Read/Write, after Dial has
+// already reported success and past its retry. The executor reads stdin only once the
+// stream is established, so the first stdin read is used as the readiness signal.
+func newExecConn(ctx context.Context, exec remotecommand.Executor) (net.Conn, error) {
 	stdinR, stdinW := io.Pipe()
 	stdoutR, stdoutW := io.Pipe()
 	kc := &kubeConn{
@@ -48,9 +58,14 @@ func newExecConn(ctx context.Context, exec remotecommand.Executor) net.Conn {
 		localAddr:  dummyAddr{network: "dummy", s: "dummy-0"},
 		remoteAddr: dummyAddr{network: "dummy", s: "dummy-1"},
 	}
+
+	ready := make(chan struct{})
+	stdin := &readyReader{r: stdinR, ready: ready}
+
+	streamErr := make(chan error, 1)
 	go func() {
 		serr := exec.StreamWithContext(ctx, remotecommand.StreamOptions{
-			Stdin:  stdinR,
+			Stdin:  stdin,
 			Stdout: stdoutW,
 			Stderr: os.Stderr,
 			Tty:    false,
@@ -61,8 +76,35 @@ func newExecConn(ctx context.Context, exec remotecommand.Executor) net.Conn {
 		// Ensure the pipes are closed to unblock Read/Write on kubeConn and avoid infinite hangs.
 		stdoutW.CloseWithError(serr)
 		stdinR.CloseWithError(serr)
+		streamErr <- serr
 	}()
-	return kc
+
+	select {
+	case <-ready:
+		return kc, nil
+	case serr := <-streamErr:
+		_ = kc.Close()
+		if serr == nil {
+			serr = errStreamEndedBeforeReady
+		}
+		return nil, serr
+	case <-ctx.Done():
+		_ = kc.Close()
+		return nil, context.Cause(ctx)
+	}
+}
+
+// readyReader closes ready on the first Read. The exec stream reads stdin only once its
+// streams have been created, so the first read marks the connection as established.
+type readyReader struct {
+	r     io.Reader
+	once  sync.Once
+	ready chan struct{}
+}
+
+func (r *readyReader) Read(p []byte) (int, error) {
+	r.once.Do(func() { close(r.ready) })
+	return r.r.Read(p)
 }
 
 type kubeConn struct {
diff --git a/driver/kubernetes/execconn/execconn_test.go b/driver/kubernetes/execconn/execconn_test.go
index 0cb446735802..73de2789f9f7 100644
--- a/driver/kubernetes/execconn/execconn_test.go
+++ b/driver/kubernetes/execconn/execconn_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net"
 	"testing"
 	"time"
 
@@ -11,40 +12,169 @@ import (
 	"k8s.io/client-go/tools/remotecommand"
 )
 
-// fakeExecutor is a fake remotecommand.Executor whose StreamWithContext blocks until
-// unblock is closed, then returns err. It stands in for the real SPDY exec stream
-// to a builder pod, so the pipe-closing behavior in newExecConn can be tested
-// without a real Kubernetes API server.
+// fakeExecutor stands in for the real SPDY exec stream. The connection is only
+// considered established once StreamWithContext reads stdin, matching the real
+// executor which copies stdin to the pod once all its streams are created. It
+// reads stdin just enough to signal readiness and then stops consuming it, so a
+// write after the stream ends observes the closed pipe rather than being read.
 type fakeExecutor struct {
+	// startErr, if set, is returned before the stream is established, without ever
+	// reading stdin (a failed SPDY/TLS upgrade).
+	startErr error
+	// stall, if set, blocks without establishing the stream until the context is
+	// cancelled (an upgrade that never completes).
+	stall bool
+	// echo, if set, copies stdin to stdout once the stream is established.
+	echo bool
+	// unblock ends an established stream when closed; endErr is then returned.
 	unblock chan struct{}
-	err     error
+	endErr  error
+	// finished is closed when StreamWithContext returns.
+	finished chan struct{}
+	// readerDone is closed when the stdin reader has stopped (non-echo streams).
+	readerDone chan struct{}
 }
 
 func (f *fakeExecutor) Stream(_ remotecommand.StreamOptions) error {
 	panic("unimplemented")
 }
 
-func (f *fakeExecutor) StreamWithContext(ctx context.Context, _ remotecommand.StreamOptions) error {
+func (f *fakeExecutor) StreamWithContext(ctx context.Context, opts remotecommand.StreamOptions) error {
+	if f.finished != nil {
+		defer close(f.finished)
+	}
+
+	if f.startErr != nil {
+		return f.startErr
+	}
+	if f.stall {
+		<-ctx.Done()
+		return context.Cause(ctx)
+	}
+
+	go func() {
+		if f.echo {
+			_, _ = io.Copy(opts.Stdout, opts.Stdin)
+			return
+		}
+		if f.readerDone != nil {
+			defer close(f.readerDone)
+		}
+		// A single read is enough to signal readiness; stop consuming afterwards.
+		_, _ = opts.Stdin.Read(make([]byte, 1))
+	}()
+
 	select {
 	case <-f.unblock:
-		return f.err
+		return f.endErr
 	case <-ctx.Done():
 		return context.Cause(ctx)
 	}
 }
 
-func TestNewExecConnPropagatesStreamEnd(t *testing.T) {
+func TestNewExecConnReadyWithData(t *testing.T) {
+	fe := &fakeExecutor{
+		echo:     true,
+		unblock:  make(chan struct{}),
+		finished: make(chan struct{}),
+	}
+	defer close(fe.unblock)
+
+	conn, err := newExecConn(context.Background(), fe)
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+
+	_, err = conn.Write([]byte("ping"))
+	require.NoError(t, err)
+
+	buf := make([]byte, 4)
+	_, err = io.ReadFull(conn, buf)
+	require.NoError(t, err)
+	require.Equal(t, "ping", string(buf))
+}
+
+func TestNewExecConnStartupFailure(t *testing.T) {
+	startErr := errors.New("unable to upgrade connection: remote error: tls: internal error")
+	fe := &fakeExecutor{
+		startErr: startErr,
+		finished: make(chan struct{}),
+	}
+
+	type result struct {
+		conn net.Conn
+		err  error
+	}
+	done := make(chan result, 1)
+	go func() {
+		conn, err := newExecConn(context.Background(), fe)
+		done <- result{conn, err}
+	}()
+
+	select {
+	case r := <-done:
+		require.ErrorIs(t, r.err, startErr)
+		require.Nil(t, r.conn)
+	case <-time.After(5 * time.Second):
+		t.Fatal("newExecConn blocked on a startup failure instead of returning the error")
+	}
+
+	select {
+	case <-fe.finished:
+	case <-time.After(time.Second):
+		t.Fatal("executor goroutine leaked after a startup failure")
+	}
+}
+
+func TestNewExecConnContextCancellation(t *testing.T) {
+	fe := &fakeExecutor{
+		stall:    true,
+		finished: make(chan struct{}),
+	}
+	ctx, cancel := context.WithCancelCause(context.Background())
+
+	type result struct {
+		conn net.Conn
+		err  error
+	}
+	done := make(chan result, 1)
+	go func() {
+		conn, err := newExecConn(ctx, fe)
+		done <- result{conn, err}
+	}()
+
+	cancel(context.Canceled)
+
+	select {
+	case r := <-done:
+		require.ErrorIs(t, r.err, context.Canceled)
+		require.Nil(t, r.conn)
+	case <-time.After(5 * time.Second):
+		t.Fatal("newExecConn did not unblock on context cancellation")
+	}
+
+	select {
+	case <-fe.finished:
+	case <-time.After(time.Second):
+		t.Fatal("executor goroutine leaked after context cancellation")
+	}
+}
+
+func TestNewExecConnStreamEndPropagates(t *testing.T) {
 	t.Run("stream ends with an error", func(t *testing.T) {
 		streamErr := errors.New("exec stream terminated")
 		fe := &fakeExecutor{
-			unblock: make(chan struct{}),
-			err:     streamErr,
+			unblock:    make(chan struct{}),
+			endErr:     streamErr,
+			readerDone: make(chan struct{}),
 		}
 
-		conn := newExecConn(context.Background(), fe)
+		conn, err := newExecConn(context.Background(), fe)
+		require.NoError(t, err)
+
 		close(fe.unblock)
+		waitClosed(t, fe.readerDone)
 
-		_, err := conn.Read(make([]byte, 16))
+		_, err = conn.Read(make([]byte, 16))
 		require.ErrorIs(t, err, streamErr)
 
 		_, err = conn.Write([]byte("test"))
@@ -52,21 +182,28 @@ func TestNewExecConnPropagatesStreamEnd(t *testing.T) {
 	})
 
 	t.Run("stream ends with no error", func(t *testing.T) {
-		fe := &fakeExecutor{unblock: make(chan struct{})}
+		fe := &fakeExecutor{
+			unblock:    make(chan struct{}),
+			readerDone: make(chan struct{}),
+		}
 
-		conn := newExecConn(context.Background(), fe)
-		close(fe.unblock) // StreamWithContext returns nil
+		conn, err := newExecConn(context.Background(), fe)
+		require.NoError(t, err)
 
-		_, err := conn.Read(make([]byte, 16))
+		close(fe.unblock)
+		waitClosed(t, fe.readerDone)
+
+		_, err = conn.Read(make([]byte, 16))
 		require.ErrorIs(t, err, io.EOF)
 
 		_, err = conn.Write([]byte("test"))
 		require.ErrorIs(t, err, io.ErrClosedPipe)
 	})
 
-	t.Run("stream still active: reads stay blocked, not closed early", func(t *testing.T) {
+	t.Run("stream still active: reads stay blocked", func(t *testing.T) {
 		fe := &fakeExecutor{unblock: make(chan struct{})}
-		conn := newExecConn(context.Background(), fe)
+		conn, err := newExecConn(context.Background(), fe)
+		require.NoError(t, err)
 		defer close(fe.unblock)
 
 		done := make(chan struct{})
@@ -79,40 +216,15 @@ func TestNewExecConnPropagatesStreamEnd(t *testing.T) {
 		case <-done:
 			t.Fatal("Read returned before the exec stream ended; it should still be blocked")
 		case <-time.After(200 * time.Millisecond):
-			// expected: still blocked, exactly like a real in-progress build
 		}
 	})
+}
 
-	t.Run("stream still active: writes stay blocked, not closed early", func(t *testing.T) {
-		fe := &fakeExecutor{unblock: make(chan struct{})}
-		conn := newExecConn(context.Background(), fe)
-		defer close(fe.unblock)
-
-		done := make(chan struct{})
-		go func() {
-			_, _ = conn.Write([]byte("test")) //nolint:errcheck
-			close(done)
-		}()
-		select {
-		case <-done:
-			t.Fatal("Write returned before the exec stream ended; it should still be blocked")
-		case <-time.After(200 * time.Millisecond):
-			// expected: still blocked, exactly like a real in-progress build
-		}
-	})
-
-	t.Run("stream cancelled by context", func(t *testing.T) {
-		fe := &fakeExecutor{unblock: make(chan struct{})}
-		ctx, cancel := context.WithCancelCause(context.Background())
-		conn := newExecConn(ctx, fe)
-		defer close(fe.unblock)
-
-		cancel(context.Canceled)
-
-		_, err := conn.Read(make([]byte, 16))
-		require.ErrorIs(t, err, context.Canceled)
-
-		_, err = conn.Write([]byte("test"))
-		require.ErrorIs(t, err, context.Canceled)
-	})
+func waitClosed(t *testing.T, ch <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatal("fake executor kept reading stdin after the stream ended")
+	}
 }