diff --git a/pkg/svc/provider/hetzner/export_test.go b/pkg/svc/provider/hetzner/export_test.go
index a163246c52..48febcd92a 100644
--- a/pkg/svc/provider/hetzner/export_test.go
+++ b/pkg/svc/provider/hetzner/export_test.go
@@ -18,7 +18,7 @@ func (p *Provider) WaitForServersStatusForTest(
 	clusterName string,
 	status hcloud.ServerStatus,
 ) error {
-	return p.waitForServersStatus(ctx, clusterName, status)
+	return p.waitForServersStatus(ctx, clusterName, status, DefaultActionTimeout)
 }
 
 // SourceIPsEqualForTest exports sourceIPsEqual for testing.
@@ -131,3 +131,8 @@ func NewSnapshotManagerWithUploaderForTest(
 
 // BuildSSHFirewallRulesForTest exports buildSSHFirewallRules for testing.
 var BuildSSHFirewallRulesForTest = buildSSHFirewallRules
+
+// SetGracefulShutdownTimeoutForTest shortens the graceful shutdown window StopNodes waits for.
+func (p *Provider) SetGracefulShutdownTimeoutForTest(timeout time.Duration) {
+	p.gracefulShutdownTimeout = timeout
+}
diff --git a/pkg/svc/provider/hetzner/provider.go b/pkg/svc/provider/hetzner/provider.go
index cded0a0b6e..1e71eade99 100644
--- a/pkg/svc/provider/hetzner/provider.go
+++ b/pkg/svc/provider/hetzner/provider.go
@@ -16,6 +16,9 @@ import (
 const (
 	// DefaultActionTimeout is the timeout for waiting on Hetzner actions.
 	DefaultActionTimeout = 5 * time.Minute
+	// DefaultGracefulShutdownTimeout is how long StopNodes waits for a graceful
+	// (ACPI) shutdown before it powers off the servers that are still on.
+	DefaultGracefulShutdownTimeout = 3 * time.Minute
 	// DefaultOperationTimeout is the timeout for individual operations.
 	DefaultOperationTimeout = 2 * time.Minute
 	// DefaultPollingInterval is the interval between status checks.
@@ -48,9 +51,14 @@ const (
 // ErrHetznerActionFailed indicates that a Hetzner action failed.
 var ErrHetznerActionFailed = errors.New("hetzner action failed")
 
+// ErrServerStatusTimeout indicates that servers did not reach a status in time.
+var ErrServerStatusTimeout = errors.New("timeout waiting for servers to reach status")
+
 // Provider implements provider.Provider for Hetzner Cloud servers.
 type Provider struct {
 	client *hcloud.Client
+	// gracefulShutdownTimeout overrides DefaultGracefulShutdownTimeout when positive.
+	gracefulShutdownTimeout time.Duration
 }
 
 // NewProvider creates a new Hetzner Cloud provider with the given client.
@@ -87,7 +95,12 @@ func (p *Provider) StartNodes(ctx context.Context, clusterName string) error {
 }
 
 // StopNodes stops all servers for the given cluster.
-// Uses graceful shutdown (ACPI signal) and waits for servers to be off.
+//
+// It sends a graceful shutdown (ACPI signal) and waits up to the graceful
+// shutdown timeout for every server to be off. Servers still on after that
+// window are powered off, the same as pulling the plug, so a node whose
+// operating system never finishes its shutdown cannot leave the cluster
+// half-stopped.
 func (p *Provider) StopNodes(ctx context.Context, clusterName string) error {
 	err := p.forEachServer(ctx, clusterName, func(server *hcloud.Server) (*hcloud.Action, error) {
 		// Skip if already off
@@ -106,8 +119,47 @@ func (p *Provider) StopNodes(ctx context.Context, clusterName string) error {
 		return err
 	}
 
-	// Wait for all servers to be off
-	return p.waitForServersStatus(ctx, clusterName, hcloud.ServerStatusOff)
+	gracefulErr := p.waitForServersStatus(
+		ctx,
+		clusterName,
+		hcloud.ServerStatusOff,
+		p.gracefulShutdownWindow(),
+	)
+	// Force servers off only when the graceful window itself ran out. An API
+	// failure, or a caller that cancelled the stop, is returned unchanged.
+	if gracefulErr == nil || !errors.Is(gracefulErr, ErrServerStatusTimeout) || ctx.Err() != nil {
+		return gracefulErr
+	}
+
+	err = p.forEachServer(ctx, clusterName, func(server *hcloud.Server) (*hcloud.Action, error) {
+		if server.Status == hcloud.ServerStatusOff {
+			return nil, provider.ErrSkipAction
+		}
+
+		action, _, poweroffErr := p.client.Server.Poweroff(ctx, server)
+		if poweroffErr != nil {
+			return nil, fmt.Errorf("failed to power off server %s: %w", server.Name, poweroffErr)
+		}
+
+		return action, nil
+	})
+	if err != nil {
+		return fmt.Errorf("graceful shutdown did not finish (%w), and forcing power off failed: %w",
+			gracefulErr, err)
+	}
+
+	return p.waitForServersStatus(ctx, clusterName, hcloud.ServerStatusOff, DefaultActionTimeout)
+}
+
+// gracefulShutdownWindow returns how long StopNodes waits for an ACPI shutdown.
+//
+//nolint:funcorder // Grouped with StopNodes for logical code organization
+func (p *Provider) gracefulShutdownWindow() time.Duration {
+	if p.gracefulShutdownTimeout > 0 {
+		return p.gracefulShutdownTimeout
+	}
+
+	return DefaultGracefulShutdownTimeout
 }
 
 // waitForServersStatus polls until all servers in the cluster reach the desired status.
@@ -117,6 +169,7 @@ func (p *Provider) waitForServersStatus(
 	ctx context.Context,
 	clusterName string,
 	desiredStatus hcloud.ServerStatus,
+	timeout time.Duration,
 ) error {
 	var (
 		condErr error
@@ -126,7 +179,7 @@ func (p *Provider) waitForServersStatus(
 	err := wait.PollUntilContextTimeout(
 		ctx,
 		DefaultPollingInterval,
-		DefaultActionTimeout,
+		timeout,
 		false, // match the legacy ticker: wait one interval before the first check.
 		func(ctx context.Context) (bool, error) {
 			ready, checkErr := p.allServersAtStatus(ctx, clusterName, desiredStatus)
@@ -151,7 +204,8 @@ func (p *Provider) waitForServersStatus(
 	}
 
 	return fmt.Errorf(
-		"timeout waiting for servers to reach status %s: %w",
+		"%w %s: %w",
+		ErrServerStatusTimeout,
 		desiredStatus,
 		err,
 	)
diff --git a/pkg/svc/provider/hetzner/stop_nodes_test.go b/pkg/svc/provider/hetzner/stop_nodes_test.go
new file mode 100644
index 0000000000..0bf2c5da0d
--- /dev/null
+++ b/pkg/svc/provider/hetzner/stop_nodes_test.go
@@ -0,0 +1,247 @@
+package hetzner_test
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/devantler-tech/ksail/v7/pkg/svc/provider/hetzner"
+	"github.com/hetznercloud/hcloud-go/v2/hcloud/schema"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const stopTestCluster = "stop-test"
+
+// stopTestAPI is a fake Hetzner API holding the power state of a cluster's
+// servers. Servers whose name is in ignoreShutdown stay running after a graceful
+// shutdown, like a node whose operating system never finishes shutting down.
+type stopTestAPI struct {
+	mu             sync.Mutex
+	status         map[string]string
+	ids            map[int64]string
+	ignoreShutdown map[string]bool
+	shutdowns      []string
+	poweroffs      []string
+	// stallStatusRead, when set, holds the first server list that follows the
+	// graceful shutdowns open until the client gives up on it, like a slow API
+	// whose status read is still in flight when the graceful window ends.
+	stallStatusRead bool
+	stalled         bool
+}
+
+// stallsThisRead reports whether a server list request must be held open.
+func (api *stopTestAPI) stallsThisRead(nameFilter string) bool {
+	api.mu.Lock()
+	defer api.mu.Unlock()
+
+	if !api.stallStatusRead || api.stalled || nameFilter != "" || len(api.shutdowns) == 0 {
+		return false
+	}
+
+	api.stalled = true
+
+	return true
+}
+
+// recorded returns the servers shut down, the servers powered off, and whether
+// the API held a status read open, all read under the lock its handlers write
+// them under. A stalled handler can still be running after the client gave up
+// on it, so a test must never read these fields directly.
+func (api *stopTestAPI) recorded() ([]string, []string, bool) {
+	api.mu.Lock()
+	defer api.mu.Unlock()
+
+	return slices.Clone(api.shutdowns), slices.Clone(api.poweroffs), api.stalled
+}
+
+func newStopTestAPI(status map[string]string, ignoreShutdown ...string) *stopTestAPI {
+	api := &stopTestAPI{
+		status:         status,
+		ids:            map[int64]string{},
+		ignoreShutdown: map[string]bool{},
+	}
+
+	var nextID int64 = 1
+	for name := range status {
+		api.ids[nextID] = name
+		nextID++
+	}
+
+	for _, name := range ignoreShutdown {
+		api.ignoreShutdown[name] = true
+	}
+
+	return api
+}
+
+func (api *stopTestAPI) servers(nameFilter string) []schema.Server {
+	api.mu.Lock()
+	defer api.mu.Unlock()
+
+	servers := []schema.Server{}
+
+	for serverID, name := range api.ids {
+		if nameFilter != "" && nameFilter != name {
+			continue
+		}
+
+		servers = append(servers, schema.Server{
+			ID:     serverID,
+			Name:   name,
+			Status: api.status[name],
+			Labels: map[string]string{
+				hetzner.LabelOwned:       "true",
+				hetzner.LabelClusterName: stopTestCluster,
+			},
+		})
+	}
+
+	return servers
+}
+
+func (api *stopTestAPI) act(t *testing.T, writer http.ResponseWriter, request *http.Request) {
+	t.Helper()
+
+	serverID, err := strconv.ParseInt(request.PathValue("id"), 10, 64)
+	if err != nil {
+		t.Errorf("parse id: %v", err)
+		http.Error(writer, err.Error(), http.StatusBadRequest)
+
+		return
+	}
+
+	command := request.PathValue("command")
+
+	api.mu.Lock()
+
+	name := api.ids[serverID]
+
+	switch command {
+	case "shutdown":
+		api.shutdowns = append(api.shutdowns, name)
+		if !api.ignoreShutdown[name] {
+			api.status[name] = "off"
+		}
+	case "poweroff":
+		api.poweroffs = append(api.poweroffs, name)
+		api.status[name] = "off"
+	default:
+		t.Errorf("unexpected server action %q", command)
+	}
+
+	api.mu.Unlock()
+
+	writeJSONResponse(t, writer, map[string]any{"action": successTestAction(serverID)})
+}
+
+func (api *stopTestAPI) start(t *testing.T) *httptest.Server {
+	t.Helper()
+
+	mux := http.NewServeMux()
+
+	mux.HandleFunc("GET /servers", func(writer http.ResponseWriter, request *http.Request) {
+		if api.stallsThisRead(request.URL.Query().Get("name")) {
+			<-request.Context().Done()
+
+			return
+		}
+
+		writeJSONResponse(
+			t,
+			writer,
+			schema.ServerListResponse{Servers: api.servers(request.URL.Query().Get("name"))},
+		)
+	})
+
+	mux.HandleFunc(
+		"POST /servers/{id}/actions/{command}",
+		func(writer http.ResponseWriter, request *http.Request) {
+			api.act(t, writer, request)
+		},
+	)
+
+	mux.HandleFunc("GET /actions", func(writer http.ResponseWriter, request *http.Request) {
+		actionID, err := strconv.ParseInt(request.URL.Query().Get("id"), 10, 64)
+		if err != nil {
+			t.Errorf("parse id: %v", err)
+			http.Error(writer, err.Error(), http.StatusBadRequest)
+
+			return
+		}
+
+		writeJSONResponse(t, writer, map[string]any{"actions": []any{successTestAction(actionID)}})
+	})
+
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+
+	return srv
+}
+
+// TestStopNodes_GracefulShutdownNeedsNoPowerOff proves that servers which shut
+// down gracefully are never forced off.
+func TestStopNodes_GracefulShutdownNeedsNoPowerOff(t *testing.T) {
+	t.Parallel()
+
+	api := newStopTestAPI(map[string]string{"cp-1": "running", "worker-1": "running"})
+	prov := hetzner.NewProvider(newTestHcloudClient(t, api.start(t).URL))
+	prov.SetGracefulShutdownTimeoutForTest(hetzner.DefaultPollingInterval + 3*time.Second)
+
+	require.NoError(t, prov.StopNodes(t.Context(), stopTestCluster))
+
+	shutdowns, poweroffs, _ := api.recorded()
+
+	assert.ElementsMatch(t, []string{"cp-1", "worker-1"}, shutdowns)
+	assert.Empty(t, poweroffs, "a graceful shutdown that finished must not be forced")
+}
+
+// TestStopNodes_PowersOffServersThatIgnoreShutdown proves that a server still
+// running after the graceful window is powered off, so the stop succeeds
+// instead of timing out, and that only that server is forced.
+func TestStopNodes_PowersOffServersThatIgnoreShutdown(t *testing.T) {
+	t.Parallel()
+
+	api := newStopTestAPI(
+		map[string]string{"cp-1": "running", "worker-1": "running", "worker-2": "off"},
+		"cp-1",
+	)
+	prov := hetzner.NewProvider(newTestHcloudClient(t, api.start(t).URL))
+	prov.SetGracefulShutdownTimeoutForTest(hetzner.DefaultPollingInterval + time.Second)
+
+	require.NoError(t, prov.StopNodes(t.Context(), stopTestCluster))
+
+	shutdowns, poweroffs, _ := api.recorded()
+
+	assert.ElementsMatch(t, []string{"cp-1", "worker-1"}, shutdowns,
+		"a server that is already off is not shut down again")
+	assert.Equal(t, []string{"cp-1"}, poweroffs,
+		"only the server still running after the graceful window is forced off")
+}
+
+// TestStopNodes_PowersOffWhenTheLastStatusReadFailsAtTheWindowDeadline proves
+// that a status read still in flight when the graceful window ends does not
+// stop the power off. The provider cannot tell a read its own deadline
+// interrupted from one that failed with a matching error at that moment, and
+// in both cases the window has run out. The power off step lists the servers
+// again, so it acts on their current state and fails if the API is down.
+func TestStopNodes_PowersOffWhenTheLastStatusReadFailsAtTheWindowDeadline(t *testing.T) {
+	t.Parallel()
+
+	api := newStopTestAPI(map[string]string{"cp-1": "running", "worker-1": "running"}, "cp-1")
+	api.stallStatusRead = true
+	prov := hetzner.NewProvider(newTestHcloudClient(t, api.start(t).URL))
+	prov.SetGracefulShutdownTimeoutForTest(hetzner.DefaultPollingInterval + time.Second)
+
+	require.NoError(t, prov.StopNodes(t.Context(), stopTestCluster))
+
+	_, poweroffs, stalled := api.recorded()
+
+	assert.True(t, stalled, "the status read at the window deadline must have been interrupted")
+	assert.Equal(t, []string{"cp-1"}, poweroffs,
+		"only the server still running after the graceful window is forced off")
+}