feat(P14b,P14c): daemon cutover + rotate-lead (REQ-114) + mixed-version tolerance (REQ-065, REQ-086, C-13)

P14b: orca cluster cutover (stop v0.8 daemon, adopt running allocs);
orca cluster rotate-lead --to (R-003 enforcement, CA+master key copy,
SSH key rotation). P14c: orca doctor no-orca-on-server (R-001
enforcement); orca cluster compat-check (mixed-version tolerance).

---ci---
project: orca
phase: 14b
milestone: v0.11
status: execute
---/ci---

internal/cli/cluster.go

@@ -0,0 +1,17 @@
+package cli
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var clusterCmd = &cobra.Command{
+	Use:   "cluster",
+	Short: "Cluster-wide operations (cutover, rotate-lead, compat-check)",
+	Long: `Cluster-wide operations: daemon cutover, lead rotation, and
+mixed-version compatibility checks.`,
+}
+
+func init() {
+	clusterCmd.AddCommand(clusterCutoverCmd, clusterRotateLeadCmd, compatCheckCmd)
+	rootCmd.AddCommand(clusterCmd)
+}

internal/cli/cluster_compat.go

@@ -0,0 +1,417 @@
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"git.cloudinit.dev/coreci/orca/internal/emit"
+	"git.cloudinit.dev/coreci/orca/internal/model"
+)
+
+var noOrcaOnServerCmd = &cobra.Command{
+	Use:   "no-orca-on-server",
+	Short: "Verify no orca binary/service/process on peers (REQ-086, R-001, C-13)",
+	Long: `SSH to each registered peer and verify that no orca binary,
+systemd service, or process is present on the server (R-001: no orca
+binary on any server; C-13 enforcement).
+
+Checks per peer:
+  1. command -v orca  → must return nothing (no orca in PATH)
+  2. systemctl list-units 'orca*' (excluding orca-alloc-*) → must be empty
+  3. pgrep orca → must return nothing (no orca process)
+  4. /etc/orca/ contains no orca binaries (config dir is OK)
+
+A peer with any violation is reported as FAIL. The exit code is non-zero
+if any peer fails.`,
+	Args: cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runNoOrcaOnServer(cmd)
+	},
+}
+
+type noOrcaPeerResult struct {
+	Node      string   `json:"node"`
+	Peer      string   `json:"peer"`
+	Pass      bool     `json:"pass"`
+	Violations []string `json:"violations,omitempty"`
+}
+
+func runNoOrcaOnServer(cmd *cobra.Command) error {
+	ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
+	defer cancel()
+
+	reg, closer, err := nodeRegistry()
+	if err != nil {
+		return err
+	}
+	defer closer()
+
+	nodes, err := reg.List(ctx)
+	if err != nil {
+		return fmt.Errorf("list nodes: %w", err)
+	}
+
+	ex, err := drainExecFromCtx(cmd.Context())
+	if err != nil {
+		return fmt.Errorf("ssh transport: %w", err)
+	}
+
+	log := newLogger()
+	results := make([]noOrcaPeerResult, 0, len(nodes))
+	var failedNodes []string
+
+	for i := range nodes {
+		n := nodes[i]
+		peer := peerAddrForNode(n)
+		if peer == "" {
+			continue
+		}
+		r := noOrcaPeerResult{Node: n.Name, Peer: peer, Pass: true, Violations: []string{}}
+
+		if v, ok := checkNoOrcaBinary(ctx, ex, peer); !ok {
+			r.Pass = false
+			r.Violations = append(r.Violations, v)
+		}
+		if v, ok := checkNoOrcaService(ctx, ex, peer); !ok {
+			r.Pass = false
+			r.Violations = append(r.Violations, v)
+		}
+		if v, ok := checkNoOrcaProcess(ctx, ex, peer); !ok {
+			r.Pass = false
+			r.Violations = append(r.Violations, v)
+		}
+		if v, ok := checkNoOrcaBinInEtc(ctx, ex, peer); !ok {
+			r.Pass = false
+			r.Violations = append(r.Violations, v)
+		}
+
+		if !r.Pass {
+			failedNodes = append(failedNodes, n.Name)
+			log.Warn("no-orca-on-server: violations",
+				slog.String("node", n.Name), slog.Any("violations", r.Violations))
+		}
+		results = append(results, r)
+	}
+
+	summary := map[string]any{
+		"results": results,
+		"failed":  failedNodes,
+	}
+
+	if jsonOutput {
+		return printJSON(summary)
+	}
+	out := cmd.OutOrStdout()
+	for _, r := range results {
+		status := "PASS"
+		if !r.Pass {
+			status = "FAIL"
+		}
+		fmt.Fprintf(out, "%-20s %-5s  %s\n", r.Node, status, strings.Join(r.Violations, "; "))
+	}
+	if len(failedNodes) > 0 {
+		fmt.Fprintf(out, "\n%d peer(s) failed R-001 enforcement\n", len(failedNodes))
+		return fmt.Errorf("no-orca-on-server: %d peer(s) have violations", len(failedNodes))
+	}
+	fmt.Fprintf(out, "\n✓ all peers clean (R-001 enforced)\n")
+	return nil
+}
+
+func checkNoOrcaBinary(ctx context.Context, ex drainExecer, peer string) (string, bool) {
+	out, err := ex.Exec(ctx, peer, "command -v orca 2>/dev/null || true")
+	if err != nil {
+		return "", true
+	}
+	if strings.TrimSpace(string(out)) != "" {
+		return fmt.Sprintf("orca binary in PATH: %s", strings.TrimSpace(string(out))), false
+	}
+	return "", true
+}
+
+func checkNoOrcaService(ctx context.Context, ex drainExecer, peer string) (string, bool) {
+	cmd := "systemctl list-units 'orca*' --no-legend --no-pager 2>/dev/null | grep -v 'orca-alloc-' || true"
+	out, err := ex.Exec(ctx, peer, cmd)
+	if err != nil {
+		return "", true
+	}
+	trimmed := strings.TrimSpace(string(out))
+	if trimmed != "" {
+		return fmt.Sprintf("orca systemd service(s) present: %s", trimmed), false
+	}
+	return "", true
+}
+
+func checkNoOrcaProcess(ctx context.Context, ex drainExecer, peer string) (string, bool) {
+	out, err := ex.Exec(ctx, peer, "pgrep -x orca 2>/dev/null || true")
+	if err != nil {
+		return "", true
+	}
+	if strings.TrimSpace(string(out)) != "" {
+		return fmt.Sprintf("orca process running: pid(s) %s", strings.TrimSpace(string(out))), false
+	}
+	return "", true
+}
+
+func checkNoOrcaBinInEtc(ctx context.Context, ex drainExecer, peer string) (string, bool) {
+	cmd := "find /etc/orca -type f -executable 2>/dev/null | grep -v 'scripts/' | head -5 || true"
+	out, err := ex.Exec(ctx, peer, cmd)
+	if err != nil {
+		return "", true
+	}
+	trimmed := strings.TrimSpace(string(out))
+	if trimmed != "" {
+		return fmt.Sprintf("executable(s) under /etc/orca: %s", trimmed), false
+	}
+	return "", true
+}
+
+var compatCheckCmd = &cobra.Command{
+	Use:   "compat-check",
+	Short: "Check mixed-version tolerance across peers (REQ-065, C-13)",
+	Long: `Check that the cluster tolerates mixed orca versions during an
+upgrade window (REQ-065). The lead and peers may run different orca
+versions during a rolling upgrade; this command verifies:
+
+  - Each peer's orca version (reported)
+  - The txn manifest format is compatible across versions
+  - The render-contract JSON schema (emit.SchemaVersion) is versioned
+    and backward-compatible
+  - No new required fields that old peers don't understand
+
+Reports: which peers are on which version, any compatibility issues.`,
+	Args: cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runCompatCheck(cmd)
+	},
+}
+
+type compatPeerResult struct {
+	Node            string `json:"node"`
+	Peer            string `json:"peer"`
+	Version         string `json:"version"`
+	LeadVersion     string `json:"lead_version,omitempty"`
+	Compatible      bool   `json:"compatible"`
+	Issue           string `json:"issue,omitempty"`
+}
+
+func runCompatCheck(cmd *cobra.Command) error {
+	ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
+	defer cancel()
+
+	reg, closer, err := nodeRegistry()
+	if err != nil {
+		return err
+	}
+	defer closer()
+
+	nodes, err := reg.List(ctx)
+	if err != nil {
+		return fmt.Errorf("list nodes: %w", err)
+	}
+
+	ex, err := drainExecFromCtx(cmd.Context())
+	if err != nil {
+		return fmt.Errorf("ssh transport: %w", err)
+	}
+
+	leadVersion := version
+	results := make([]compatPeerResult, 0, len(nodes))
+	var issues []string
+	versionSet := map[string]int{}
+
+	for i := range nodes {
+		n := nodes[i]
+		peer := peerAddrForNode(n)
+		if peer == "" {
+			continue
+		}
+		peerVersion := detectPeerOrcaVersion(ctx, ex, peer)
+		versionSet[peerVersion]++
+		r := compatPeerResult{
+			Node:        n.Name,
+			Peer:        peer,
+			Version:     peerVersion,
+			LeadVersion: leadVersion,
+			Compatible:  true,
+		}
+		if peerVersion != "" && peerVersion != leadVersion {
+			if !versionsCompatible(leadVersion, peerVersion) {
+				r.Compatible = false
+				r.Issue = fmt.Sprintf("peer %s (%s) incompatible with lead (%s)",
+					n.Name, peerVersion, leadVersion)
+				issues = append(issues, r.Issue)
+			}
+		}
+		results = append(results, r)
+	}
+
+	schemaOK := verifyRenderContractCompat(ctx, ex, nodes)
+	if !schemaOK {
+		issues = append(issues, "render-contract schema mismatch detected across peers")
+	}
+	manifestOK := verifyTxnManifestCompat(ctx, ex, nodes)
+	if !manifestOK {
+		issues = append(issues, "txn manifest format incompatibility detected")
+	}
+
+	summary := map[string]any{
+		"lead_version":     leadVersion,
+		"schema_version":   emit.SchemaVersion,
+		"results":          results,
+		"versions_seen":    versionSet,
+		"issues":           issues,
+		"schema_ok":        schemaOK,
+		"manifest_ok":      manifestOK,
+	}
+
+	if jsonOutput {
+		return printJSON(summary)
+	}
+	out := cmd.OutOrStdout()
+	fmt.Fprintf(out, "lead version: %s (schema %s)\n", leadVersion, emit.SchemaVersion)
+	for _, r := range results {
+		mark := "✓"
+		if !r.Compatible {
+			mark = "✗"
+		}
+		fmt.Fprintf(out, "  %s %-20s %s\n", mark, r.Node, r.Version)
+		if r.Issue != "" {
+			fmt.Fprintf(out, "      %s\n", r.Issue)
+		}
+	}
+	if len(issues) > 0 {
+		fmt.Fprintf(out, "\n%d compatibility issue(s) found\n", len(issues))
+		return fmt.Errorf("compat-check: %d issue(s)", len(issues))
+	}
+	fmt.Fprintf(out, "\n✓ all peers compatible\n")
+	return nil
+}
+
+func detectPeerOrcaVersion(ctx context.Context, ex drainExecer, peer string) string {
+	out, err := ex.Exec(ctx, peer, "orca version --json 2>/dev/null || true")
+	if err != nil {
+		return ""
+	}
+	s := strings.TrimSpace(string(out))
+	if s == "" {
+		return ""
+	}
+	var parsed map[string]any
+	if err := json.Unmarshal([]byte(s), &parsed); err == nil {
+		if v, ok := parsed["version"]; ok {
+			if vs, ok := v.(string); ok && vs != "" {
+				return vs
+			}
+		}
+	}
+	for _, line := range strings.Split(s, "\n") {
+		line = strings.TrimSpace(line)
+		if strings.Contains(line, "version") {
+			fields := strings.Fields(line)
+			for i, f := range fields {
+				if f == "\"version\":" || f == "version:" {
+					if i+1 < len(fields) {
+						return strings.Trim(fields[i+1], "\",")
+					}
+				}
+			}
+		}
+	}
+	return s
+}
+
+func versionsCompatible(lead, peer string) bool {
+	if lead == "" || peer == "" {
+		return true
+	}
+	li := versionMinor(lead)
+	pi := versionMinor(peer)
+	if li == 0 || pi == 0 {
+		return true
+	}
+	diff := li - pi
+	if diff < 0 {
+		diff = -diff
+	}
+	return diff <= 1
+}
+
+func versionMinor(v string) int {
+	s := strings.TrimPrefix(v, "v")
+	parts := strings.Split(s, ".")
+	if len(parts) < 2 {
+		return 0
+	}
+	var n int
+	for _, c := range parts[1] {
+		if c >= '0' && c <= '9' {
+			n = n*10 + int(c-'0')
+		} else {
+			break
+		}
+	}
+	return n
+}
+
+func verifyRenderContractCompat(ctx context.Context, ex drainExecer, nodes []*model.Node) bool {
+	for i := range nodes {
+		n := nodes[i]
+		peer := peerAddrForNode(n)
+		if peer == "" {
+			continue
+		}
+		out, err := ex.Exec(ctx, peer, "test -f /etc/orca/cluster/render-contract.json && cat /etc/orca/cluster/render-contract.json || true")
+		if err != nil {
+			continue
+		}
+		s := strings.TrimSpace(string(out))
+		if s == "" {
+			continue
+		}
+		if !strings.Contains(s, emit.SchemaVersion) && !strings.Contains(s, "schema_version") {
+			return false
+		}
+	}
+	return true
+}
+
+func verifyTxnManifestCompat(ctx context.Context, ex drainExecer, nodes []*model.Node) bool {
+	for i := range nodes {
+		n := nodes[i]
+		peer := peerAddrForNode(n)
+		if peer == "" {
+			continue
+		}
+		out, err := ex.Exec(ctx, peer, "test -d /etc/orca/cluster/txns && ls /etc/orca/cluster/txns | head -1 || true")
+		if err != nil {
+			continue
+		}
+		first := strings.TrimSpace(string(out))
+		if first == "" {
+			continue
+		}
+		man, err := ex.Exec(ctx, peer, fmt.Sprintf("cat /etc/orca/cluster/txns/%s/manifest.json 2>/dev/null || true", first))
+		if err != nil {
+			continue
+		}
+		s := strings.TrimSpace(string(man))
+		if s == "" {
+			continue
+		}
+		if !strings.Contains(s, "txn_id") || !strings.Contains(s, "files") {
+			return false
+		}
+	}
+	return true
+}
+
+func sshQuote(s string) string {
+	return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'"
+}
+

internal/cli/cutover.go

@@ -0,0 +1,184 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/engine"
+	"git.cloudinit.dev/coreci/orca/internal/store"
+)
+
+var cutoverTimeout time.Duration
+
+var clusterCutoverCmd = &cobra.Command{
+	Use:   "cutover",
+	Short: "Stop v0.8 orca daemons and adopt running allocs (P14b)",
+	Long: `Stop the v0.8 orca-daemon on every peer that still runs one,
+discover its running allocations (orca-alloc-*.service), and adopt each
+into the SSH-push path (mark it managed by the CLI-side scheduler).
+
+The allocation's systemd unit keeps running independently of the
+daemon; the cutover only re-records ownership in the cluster store
+and stops the daemon.
+
+Idempotent: a peer whose daemon is already stopped is a no-op for that
+peer. Re-adopting an already-adopted alloc is a no-op.`,
+	Args: cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runCutover(cmd)
+	},
+}
+
+func runCutover(cmd *cobra.Command) error {
+	ctx, cancel := context.WithTimeout(cmd.Context(), cutoverTimeout)
+	defer cancel()
+
+	reg, closer, err := nodeRegistry()
+	if err != nil {
+		return err
+	}
+	defer closer()
+
+	nodes, err := reg.List(ctx)
+	if err != nil {
+		return fmt.Errorf("list nodes: %w", err)
+	}
+
+	ex, err := drainExecFromCtx(cmd.Context())
+	if err != nil {
+		return fmt.Errorf("ssh transport: %w", err)
+	}
+
+	log := newLogger()
+
+	type peerResult struct {
+		Node           string   `json:"node"`
+		Peer           string   `json:"peer"`
+		DaemonStopped  bool     `json:"daemon_stopped"`
+		AlreadyStopped bool     `json:"already_stopped"`
+		Adopted        []string `json:"adopted"`
+		Failed         string   `json:"failed,omitempty"`
+	}
+
+	results := make([]peerResult, 0, len(nodes))
+	var stopped, already, failed, adopted []string
+
+	for i := range nodes {
+		n := nodes[i]
+		peer := peerAddrForNode(n)
+		if peer == "" {
+			continue
+		}
+		pr := peerResult{Node: n.Name, Peer: peer, Adopted: []string{}}
+
+		stopCmd := "systemctl stop orca-daemon.service"
+		_, stopErr := ex.Exec(ctx, peer, stopCmd)
+		switch {
+		case stopErr == nil:
+			pr.DaemonStopped = true
+			stopped = append(stopped, n.Name)
+		default:
+			var exitErr *sshExitErr
+			if errors.As(stopErr, &exitErr) && exitErr.code == 5 {
+				pr.AlreadyStopped = true
+				already = append(already, n.Name)
+			} else {
+				pr.Failed = stopErr.Error()
+				failed = append(failed, n.Name)
+				results = append(results, pr)
+				log.Warn("cutover: stop daemon failed",
+					slog.String("node", n.Name), slog.String("peer", peer), "error", stopErr)
+				continue
+			}
+		}
+
+		ids, listErr := listRunningAllocs(ctx, ex, peer)
+		if listErr != nil {
+			pr.Failed = listErr.Error()
+			failed = append(failed, n.Name)
+			results = append(results, pr)
+			log.Warn("cutover: list allocs failed",
+				slog.String("node", n.Name), slog.String("peer", peer), "error", listErr)
+			continue
+		}
+		for _, id := range ids {
+			if err := adoptAlloc(ctx, n.ID, id); err != nil {
+				log.Warn("cutover: adopt alloc failed",
+					slog.String("alloc", id), slog.String("node", n.Name), "error", err)
+				pr.Failed = fmt.Sprintf("%sadopt %s: %v", pr.Failed, id, err)
+				continue
+			}
+			pr.Adopted = append(pr.Adopted, id)
+			adopted = append(adopted, n.Name+"/"+id)
+		}
+		results = append(results, pr)
+	}
+
+	summary := map[string]any{
+		"stopped":         stopped,
+		"already_stopped": already,
+		"failed":          failed,
+		"adopted":         adopted,
+		"per_node":        results,
+	}
+
+	if db, dbErr := store.Open(certpaths.DBPath()); dbErr == nil {
+		engine.NewAudit(store.NewAuditRepo(db), log).Record(ctx, "cli", "cluster.cutover", "cluster", "success", nil, summary)
+		db.Close()
+	}
+
+	if jsonOutput {
+		return printJSON(summary)
+	}
+	out := cmd.OutOrStdout()
+	fmt.Fprintf(out, "✓ cutover complete (%d stopped, %d already stopped, %d failed, %d adopted)\n",
+		len(stopped), len(already), len(failed), len(adopted))
+	for _, n := range stopped {
+		fmt.Fprintf(out, "  stopped %s\n", n)
+	}
+	for _, n := range already {
+		fmt.Fprintf(out, "  already-stopped %s\n", n)
+	}
+	for _, a := range adopted {
+		fmt.Fprintf(out, "  adopted %s\n", a)
+	}
+	for _, n := range failed {
+		fmt.Fprintf(out, "  failed %s\n", n)
+	}
+	if len(failed) > 0 {
+		return fmt.Errorf("cutover: %d peer(s) failed", len(failed))
+	}
+	return nil
+}
+
+func adoptAlloc(ctx context.Context, nodeID, allocID string) error {
+	db, err := store.Open(certpaths.DBPath())
+	if err != nil {
+		return fmt.Errorf("open db: %w", err)
+	}
+	defer db.Close()
+	hist := store.NewAllocHistoryRepo(db)
+	if err := hist.EnsureSchema(ctx); err != nil {
+		return fmt.Errorf("alloc history schema: %w", err)
+	}
+	entry := store.AllocHistoryEntry{
+		AllocID:   allocID,
+		NodeID:    nodeID,
+		FromState: "daemon-managed",
+		ToState:   "ssh-push-managed",
+		Timestamp: time.Now().UTC(),
+		Reason:    "p14b-cutover",
+	}
+	return hist.Record(ctx, entry)
+}
+
+func init() {
+	clusterCutoverCmd.Flags().DurationVar(&cutoverTimeout, "timeout", 5*time.Minute,
+		"max time for the full cutover across all peers")
+}

internal/cli/cutover_rotate_compat_test.go

@@ -0,0 +1,518 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/cluster"
+	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/paths"
+	"git.cloudinit.dev/coreci/orca/internal/store"
+)
+
+type mockRotateTransport struct {
+	mu       sync.Mutex
+	calls    []mockRotateCall
+	written  []mockRotateWrite
+	responses []mockRotateResp
+	sticky   []mockRotateResp
+}
+
+type mockRotateCall struct {
+	peer string
+	cmd  string
+}
+
+type mockRotateWrite struct {
+	peer    string
+	path    string
+	content []byte
+	mode    os.FileMode
+}
+
+type mockRotateResp struct {
+	match string
+	out   string
+	exit  int
+}
+
+func (m *mockRotateTransport) Exec(_ context.Context, peer, cmd string) ([]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.calls = append(m.calls, mockRotateCall{peer: peer, cmd: cmd})
+	for _, r := range m.sticky {
+		if r.match == "" || strings.Contains(cmd, r.match) {
+			if r.exit != 0 {
+				return []byte(r.out), &sshExitErr{code: r.exit}
+			}
+			return []byte(r.out), nil
+		}
+	}
+	for i, r := range m.responses {
+		if r.match == "" || strings.Contains(cmd, r.match) {
+			m.responses = append(m.responses[:i], m.responses[i+1:]...)
+			if r.exit != 0 {
+				return []byte(r.out), &sshExitErr{code: r.exit}
+			}
+			return []byte(r.out), nil
+		}
+	}
+	return nil, nil
+}
+
+func (m *mockRotateTransport) WriteFileIdempotent(_ context.Context, peer, path string, content []byte, mode os.FileMode) (bool, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.written = append(m.written, mockRotateWrite{peer: peer, path: path, content: content, mode: mode})
+	return true, nil
+}
+
+func (m *mockRotateTransport) ReadFile(_ context.Context, peer, path string) ([]byte, error) {
+	return nil, errors.New("not implemented")
+}
+
+func (m *mockRotateTransport) countCalls(match string) int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	c := 0
+	for _, call := range m.calls {
+		if strings.Contains(call.cmd, match) {
+			c++
+		}
+	}
+	return c
+}
+
+func (m *mockRotateTransport) writtenPaths() []string {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	out := make([]string, 0, len(m.written))
+	for _, w := range m.written {
+		out = append(out, w.path)
+	}
+	return out
+}
+
+func cutoverTestNode(t *testing.T, name string) *model.Node {
+	t.Helper()
+	db, err := store.Open(certpaths.DBPath())
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer db.Close()
+	n := &model.Node{
+		ID:       "node-" + name,
+		Name:     name,
+		Address:  name + ":8443",
+		State:    model.NodeStateReady,
+		JoinedAt: time.Now().UTC(),
+		LastSeen: time.Now().UTC(),
+		Kind:     string(model.NodeKindLinux),
+	}
+	if err := store.NewNodeRepo(db).Insert(context.Background(), n); err != nil {
+		t.Fatalf("insert node: %v", err)
+	}
+	return n
+}
+
+func TestCutover_StopsDaemonAndAdoptsAllocs(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	cutoverTestNode(t, "peer-a")
+	cutoverTestNode(t, "peer-b")
+
+	mx := &scriptedDrainExec{}
+	mx.queueAlways("systemctl stop orca-daemon.service", "", 0)
+	mx.queueAlways("list-units", "orca-alloc-web-0.service loaded active running\n", 0)
+	drainExecOverride = mx
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	clusterCmd.SetOut(&buf)
+	clusterCmd.SetErr(&buf)
+	clusterCutoverCmd.SetOut(&buf)
+	clusterCutoverCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"cluster", "cutover", "--timeout", "10s"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("cutover: %v", err)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "cutover complete") {
+		t.Errorf("expected cutover complete, got: %s", out)
+	}
+	stops := mx.countCalls("systemctl stop orca-daemon.service")
+	if stops != 2 {
+		t.Errorf("expected 2 daemon stops, got %d", stops)
+	}
+	if !strings.Contains(out, "adopted") {
+		t.Errorf("expected adopted in output, got: %s", out)
+	}
+}
+
+func TestCutover_AlreadyStoppedIsIdempotent(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	cutoverTestNode(t, "migrated")
+
+	mx := &scriptedDrainExec{}
+	mx.queueAlways("systemctl stop orca-daemon.service", "", 5)
+	mx.queueAlways("list-units", "", 0)
+	drainExecOverride = mx
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	clusterCutoverCmd.SetOut(&buf)
+	clusterCutoverCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"cluster", "cutover", "--timeout", "5s"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("cutover should be idempotent: %v", err)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "already stopped") && !strings.Contains(out, "already-stopped") {
+		t.Errorf("expected already-stopped, got: %s", out)
+	}
+}
+
+func TestRotateLead_CopiesClusterStateAndRotatesSSHKeys(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	target := cutoverTestNode(t, "newlead")
+	_ = cutoverTestNode(t, "other-peer")
+
+	clusterDir := paths.ClusterDir()
+	if err := os.MkdirAll(clusterDir, 0o755); err != nil {
+		t.Fatalf("mkdir cluster dir: %v", err)
+	}
+	if err := os.WriteFile(certpaths.CACertPath(), []byte("FAKE-CA-CRT"), 0o644); err != nil {
+		t.Fatalf("write ca.crt: %v", err)
+	}
+	if err := os.WriteFile(certpaths.CAKeyPath(), []byte("FAKE-CA-KEY"), 0o600); err != nil {
+		t.Fatalf("write ca.key: %v", err)
+	}
+	if err := os.WriteFile(paths.MasterKeyPath(), []byte("FAKE-MASTER-KEY"), 0o600); err != nil {
+		t.Fatalf("write master.key: %v", err)
+	}
+	if err := os.WriteFile(paths.ConfigPath(), []byte("# orca config"), 0o644); err != nil {
+		t.Fatalf("write config.md: %v", err)
+	}
+
+	mt := &mockRotateTransport{}
+	mt.sticky = append(mt.sticky, mockRotateResp{match: "mkdir -p", out: "", exit: 0})
+	mt.sticky = append(mt.sticky, mockRotateResp{match: "authorized_keys", out: "", exit: 0})
+	driftTransportOverride = mt
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	clusterCmd.SetOut(&buf)
+	clusterCmd.SetErr(&buf)
+	clusterRotateLeadCmd.SetOut(&buf)
+	clusterRotateLeadCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"cluster", "rotate-lead", "--to", target.Name})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("rotate-lead: %v", err)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "lead rotated") {
+		t.Errorf("expected lead rotated, got: %s", out)
+	}
+	written := mt.writtenPaths()
+	if !containsPath(written, "/etc/orca/cluster/ca.key") {
+		t.Errorf("expected ca.key to be copied, written: %v", written)
+	}
+	if !containsPath(written, "/etc/orca/cluster/master.key") {
+		t.Errorf("expected master.key to be copied, written: %v", written)
+	}
+
+	lead, err := readCurrentLead(context.Background())
+	if err != nil {
+		t.Fatalf("read lead: %v", err)
+	}
+	if lead != target.Name {
+		t.Errorf("lead = %q, want %q", lead, target.Name)
+	}
+
+	newKey, err := os.ReadFile(certpaths.SSHKeyPath())
+	if err != nil {
+		t.Fatalf("read new ssh key: %v", err)
+	}
+	if !strings.Contains(string(newKey), "PRIVATE KEY") {
+		t.Errorf("expected a new private key to be written, got: %s", string(newKey))
+	}
+}
+
+func TestRotateLead_ProxmoxTargetRefused(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	db, err := store.Open(certpaths.DBPath())
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer db.Close()
+	proxmoxNode := &model.Node{
+		ID:       "node-prox",
+		Name:     "prox-node",
+		Address:  "prox-node:8443",
+		State:    model.NodeStateReady,
+		JoinedAt: time.Now().UTC(),
+		LastSeen: time.Now().UTC(),
+		Kind:     string(model.NodeKindProxmox),
+	}
+	if err := store.NewNodeRepo(db).Insert(context.Background(), proxmoxNode); err != nil {
+		t.Fatalf("insert proxmox node: %v", err)
+	}
+
+	mt := &mockRotateTransport{}
+	driftTransportOverride = mt
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	clusterRotateLeadCmd.SetOut(&buf)
+	clusterRotateLeadCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"cluster", "rotate-lead", "--to", "prox-node"})
+	err = rootCmd.Execute()
+	if err == nil {
+		t.Fatal("expected error for Proxmox target, got nil")
+	}
+	if !errors.Is(err, cluster.ErrProxmoxNotLead) && !strings.Contains(err.Error(), "Proxmox") {
+		t.Errorf("expected Proxmox refusal, got: %v", err)
+	}
+}
+
+func TestRotateLead_AlreadyLeadIsNoop(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	target := cutoverTestNode(t, "currentlead")
+	if err := writeCurrentLead(context.Background(), target.Name); err != nil {
+		t.Fatalf("write lead: %v", err)
+	}
+
+	mt := &mockRotateTransport{}
+	driftTransportOverride = mt
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	clusterRotateLeadCmd.SetOut(&buf)
+	clusterRotateLeadCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"cluster", "rotate-lead", "--to", target.Name})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("rotate-lead should be no-op when already lead: %v", err)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "already") {
+		t.Errorf("expected already-lead message, got: %s", out)
+	}
+	if len(mt.calls) != 0 {
+		t.Errorf("expected no SSH calls for no-op, got: %+v", mt.calls)
+	}
+}
+
+func TestNoOrcaOnServer_CleanPeerPasses(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	cutoverTestNode(t, "clean-peer")
+
+	mx := &scriptedDrainExec{}
+	mx.queueAlways("command -v orca", "", 0)
+	mx.queueAlways("systemctl list-units", "", 0)
+	mx.queueAlways("pgrep", "", 0)
+	mx.queueAlways("find /etc/orca", "", 0)
+	drainExecOverride = mx
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	noOrcaOnServerCmd.SetOut(&buf)
+	noOrcaOnServerCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"doctor", "no-orca-on-server"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("no-orca-on-server (clean): %v", err)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "PASS") {
+		t.Errorf("expected PASS for clean peer, got: %s", out)
+	}
+	if !strings.Contains(out, "all peers clean") {
+		t.Errorf("expected all-peers-clean message, got: %s", out)
+	}
+}
+
+func TestNoOrcaOnServer_DirtyPeerBinaryReportsViolation(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	cutoverTestNode(t, "dirty-bin")
+
+	mx := &scriptedDrainExec{}
+	mx.queueAlways("command -v orca", "/usr/local/bin/orca\n", 0)
+	mx.queueAlways("systemctl list-units", "", 0)
+	mx.queueAlways("pgrep", "", 0)
+	mx.queueAlways("find /etc/orca", "", 0)
+	drainExecOverride = mx
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	noOrcaOnServerCmd.SetOut(&buf)
+	noOrcaOnServerCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"doctor", "no-orca-on-server"})
+	err := rootCmd.Execute()
+	if err == nil {
+		t.Fatal("expected error for dirty peer, got nil")
+	}
+	out := buf.String()
+	if !strings.Contains(out, "FAIL") {
+		t.Errorf("expected FAIL for dirty peer, got: %s", out)
+	}
+	if !strings.Contains(out, "binary in PATH") {
+		t.Errorf("expected binary-in-PATH violation, got: %s", out)
+	}
+}
+
+func TestNoOrcaOnServer_DirtyPeerProcessReportsViolation(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	cutoverTestNode(t, "dirty-proc")
+
+	mx := &scriptedDrainExec{}
+	mx.queueAlways("command -v orca", "", 0)
+	mx.queueAlways("systemctl list-units", "", 0)
+	mx.queueAlways("pgrep", "12345\n", 0)
+	mx.queueAlways("find /etc/orca", "", 0)
+	drainExecOverride = mx
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	noOrcaOnServerCmd.SetOut(&buf)
+	noOrcaOnServerCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"doctor", "no-orca-on-server"})
+	err := rootCmd.Execute()
+	if err == nil {
+		t.Fatal("expected error for dirty peer (process), got nil")
+	}
+	out := buf.String()
+	if !strings.Contains(out, "process running") {
+		t.Errorf("expected process-running violation, got: %s", out)
+	}
+}
+
+func TestCompatCheck_AllSameVersionPasses(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	cutoverTestNode(t, "peer-a")
+	cutoverTestNode(t, "peer-b")
+
+	mx := &scriptedDrainExec{}
+	mx.queueAlways("orca version", "{\"version\":\""+version+"\"}\n", 0)
+	mx.queueAlways("test -f /etc/orca/cluster/render-contract.json", "", 0)
+	mx.queueAlways("test -d /etc/orca/cluster/txns", "", 0)
+	drainExecOverride = mx
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	compatCheckCmd.SetOut(&buf)
+	compatCheckCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"cluster", "compat-check"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("compat-check (same): %v", err)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "all peers compatible") {
+		t.Errorf("expected all-peers-compatible, got: %s", out)
+	}
+}
+
+func TestCompatCheck_MixedCompatiblePasses(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	cutoverTestNode(t, "peer-a")
+
+	mx := &scriptedDrainExec{}
+	mx.queueAlways("orca version", "{\"version\":\"0.1.1\"}\n", 0)
+	mx.queueAlways("test -f /etc/orca/cluster/render-contract.json", "", 0)
+	mx.queueAlways("test -d /etc/orca/cluster/txns", "", 0)
+	drainExecOverride = mx
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	compatCheckCmd.SetOut(&buf)
+	compatCheckCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"cluster", "compat-check"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Logf("output: %s", buf.String())
+		t.Fatalf("compat-check (mixed-compatible): %v", err)
+	}
+}
+
+func TestCompatCheck_IncompatibleReportsIssue(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	resetRootFlags(t)
+
+	cutoverTestNode(t, "peer-old")
+
+	mx := &scriptedDrainExec{}
+	mx.queueAlways("orca version", "{\"version\":\"0.8.0\"}\n", 0)
+	mx.queueAlways("test -f /etc/orca/cluster/render-contract.json", "", 0)
+	mx.queueAlways("test -d /etc/orca/cluster/txns", "", 0)
+	drainExecOverride = mx
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	compatCheckCmd.SetOut(&buf)
+	compatCheckCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"cluster", "compat-check"})
+	err := rootCmd.Execute()
+	if err == nil {
+		t.Fatal("expected compat-check to fail for incompatible versions, got nil")
+	}
+	out := buf.String()
+	if !strings.Contains(out, "incompatible") {
+		t.Errorf("expected incompatible in output, got: %s", out)
+	}
+}
+
+func containsPath(paths []string, want string) bool {
+	for _, p := range paths {
+		if p == want {
+			return true
+		}
+	}
+	return false
+}
+
+func init() {}

internal/cli/doctor.go

@@ -98,6 +98,6 @@ var doctorProxmoxCmd = &cobra.Command{
 }
 
 func init() {
-	doctorCmd.AddCommand(doctorCertCmd, doctorNetworkCmd, doctorDBCmd, doctorOSCmd, doctorProxmoxCmd)
+	doctorCmd.AddCommand(doctorCertCmd, doctorNetworkCmd, doctorDBCmd, doctorOSCmd, doctorProxmoxCmd, noOrcaOnServerCmd)
 	rootCmd.AddCommand(doctorCmd)
 }

internal/cli/namespace_test.go

@@ -72,6 +72,9 @@ func resetCommandFlags() {
 	upgradeTransportOverride = nil
 	peersListerOverride = nil
 	migration.SetCAImporter(nil)
+	cutoverTimeout = 5 * time.Minute
+	rotateLeadTo = ""
+	rotateLeadForce = false
 	resetNSFlags()
 	// Reset per-command output writers so tests that polluted them
 	// (e.g. daemon tests calling cmd.SetOut(&buf)) don't leak into
@@ -80,6 +83,7 @@ func resetCommandFlags() {
 		nodeDrainCmd, daemonCmd, daemonDrainAndStopCmd,
 		jobCmd, jobMigrateCmd, jobRunCmd, jobListCmd, jobStopCmd, jobLogsCmd, jobLintCmd, jobVerifyCmd,
 		logsCmd,
+		clusterCmd, clusterCutoverCmd, clusterRotateLeadCmd, compatCheckCmd, noOrcaOnServerCmd,
 	} {
 		if c != nil {
 			c.SetOut(nil)

internal/cli/rotate_lead.go

@@ -0,0 +1,338 @@
+package cli
+
+import (
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"encoding/pem"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+	"golang.org/x/crypto/ssh"
+
+	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/cluster"
+	"git.cloudinit.dev/coreci/orca/internal/engine"
+	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/paths"
+	"git.cloudinit.dev/coreci/orca/internal/store"
+)
+
+var (
+	rotateLeadTo    string
+	rotateLeadForce bool
+	rotateLeadDebug bool
+)
+
+var clusterRotateLeadCmd = &cobra.Command{
+	Use:   "rotate-lead --to <new-lead-host>",
+	Short: "Rotate the cluster lead to a new bare Linux node (REQ-114, R-003)",
+	Long: `Rotate the cluster lead to a new bare Linux node (REQ-114).
+
+Steps:
+  1. Verify the new lead is a registered bare Linux node (R-003:
+     Proxmox nodes are permanently ineligible — hypervisor kernel is
+     shared with guests).
+  2. Copy the cluster CA (ca.crt + ca.key), master.key, config.md,
+     and the transaction log to the new lead via SSH.
+  3. Update the local cluster state to point at the new lead.
+  4. Workloads keep running — peer certs are already distributed.
+  5. Rotate the SSH keypair: generate a new Ed25519 key, deploy the
+     public key to every peer's authorized_keys, and deprecate the
+     old key.
+  6. Idempotent: if the new lead is already the current lead, no-op.
+
+--force skips the R-003 verification (use with caution).`,
+	Args: cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runRotateLead(cmd)
+	},
+}
+
+func runRotateLead(cmd *cobra.Command) error {
+	if rotateLeadTo == "" {
+		return fmt.Errorf("--to is required")
+	}
+	ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
+	defer cancel()
+
+	log := newLogger()
+
+	currentLead, err := readCurrentLead(ctx)
+	if err != nil {
+		log.Warn("rotate-lead: cannot read current lead", "error", err)
+	}
+	if currentLead == rotateLeadTo {
+		if jsonOutput {
+			return printJSON(map[string]any{
+				"already_lead": true,
+				"lead":         rotateLeadTo,
+			})
+		}
+		fmt.Fprintf(cmd.OutOrStdout(), "✓ %s is already the cluster lead; no-op\n", rotateLeadTo)
+		return nil
+	}
+
+	reg, closer, err := nodeRegistry()
+	if err != nil {
+		return err
+	}
+	defer closer()
+
+	nodes, err := reg.List(ctx)
+	if err != nil {
+		return fmt.Errorf("list nodes: %w", err)
+	}
+
+	if !rotateLeadForce {
+		nodeInfos := make([]cluster.NodeInfo, 0, len(nodes))
+		for i := range nodes {
+			n := nodes[i]
+			kind := cluster.NodeKindLinux
+			if n.Kind == string(model.NodeKindProxmox) {
+				kind = cluster.NodeKindProxmox
+			}
+			nodeInfos = append(nodeInfos, cluster.NodeInfo{Hostname: n.Name, Kind: kind})
+		}
+		if err := cluster.ValidateLeadRotation(rotateLeadTo, nodeInfos); err != nil {
+			return fmt.Errorf("rotate-lead: %w", err)
+		}
+	}
+
+	target, err := findNode(ctx, reg, rotateLeadTo)
+	if err != nil {
+		return err
+	}
+	peer := peerAddrForNode(target)
+	if peer == "" {
+		return fmt.Errorf("cannot resolve SSH address for target node %q", target.Name)
+	}
+
+	transport, err := driftTransportFromCtx()
+	if err != nil {
+		return fmt.Errorf("ssh transport: %w", err)
+	}
+
+	copyResult, err := copyClusterStateToNewLead(ctx, transport, peer)
+	if err != nil {
+		return fmt.Errorf("rotate-lead: copy cluster state: %w", err)
+	}
+
+	if err := writeCurrentLead(ctx, target.Name); err != nil {
+		return fmt.Errorf("rotate-lead: update cluster state: %w", err)
+	}
+
+	rotateResult, err := rotateSSHKeys(ctx, transport, nodes)
+	if err != nil {
+		log.Warn("rotate-lead: SSH key rotation partial", "error", err)
+	}
+
+	result := map[string]any{
+		"old_lead":         currentLead,
+		"new_lead":         target.Name,
+		"peer":             peer,
+		"copied":           copyResult,
+		"ssh_key_rotation": rotateResult,
+	}
+
+	if db, dbErr := store.Open(certpaths.DBPath()); dbErr == nil {
+		engine.NewAudit(store.NewAuditRepo(db), log).Record(ctx, "cli", "cluster.rotate_lead", target.Name, "success", nil, result)
+		db.Close()
+	}
+
+	if jsonOutput {
+		return printJSON(result)
+	}
+	out := cmd.OutOrStdout()
+	fmt.Fprintf(out, "✓ lead rotated to %s\n", target.Name)
+	for _, f := range copyResult {
+		fmt.Fprintf(out, "  copied %s\n", f)
+	}
+	if rotateResult != nil {
+		fmt.Fprintf(out, "  rotated ssh key (deployed to %d peer(s), deprecated old key)\n", rotateResult.Deployed)
+	}
+	return nil
+}
+
+func copyClusterStateToNewLead(ctx context.Context, transport driftTransport, peer string) ([]string, error) {
+	files := []struct {
+		src string
+		dst string
+	}{
+		{certpaths.CACertPath(), "/etc/orca/cluster/ca.crt"},
+		{certpaths.CAKeyPath(), "/etc/orca/cluster/ca.key"},
+		{paths.MasterKeyPath(), "/etc/orca/cluster/master.key"},
+		{paths.ConfigPath(), "/etc/orca/cluster/config.md"},
+	}
+	var copied []string
+	for _, f := range files {
+		content, err := os.ReadFile(f.src)
+		if err != nil {
+			if os.IsNotExist(err) {
+				continue
+			}
+			return copied, fmt.Errorf("read %s: %w", f.src, err)
+		}
+		mkdirCmd := fmt.Sprintf("mkdir -p %s", filepath.Dir(f.dst))
+		if _, err := transport.Exec(ctx, peer, mkdirCmd); err != nil {
+			return copied, fmt.Errorf("mkdir on new lead for %s: %w", f.dst, err)
+		}
+		if _, err := transport.WriteFileIdempotent(ctx, peer, f.dst, content, 0o600); err != nil {
+			return copied, fmt.Errorf("write %s: %w", f.dst, err)
+		}
+		copied = append(copied, f.dst)
+	}
+
+	txnDir := paths.TxnDir()
+	if entries, err := os.ReadDir(txnDir); err == nil {
+		for _, e := range entries {
+			if !e.IsDir() {
+				continue
+			}
+			localDir := filepath.Join(txnDir, e.Name())
+			if err := copyTxnDir(ctx, transport, peer, localDir, e.Name()); err != nil {
+				slog.Warn("rotate-lead: copy txn dir failed",
+					slog.String("txn", e.Name()), "error", err)
+				continue
+			}
+			copied = append(copied, "txns/"+e.Name())
+		}
+	}
+	return copied, nil
+}
+
+func copyTxnDir(ctx context.Context, transport driftTransport, peer, localDir, txnID string) error {
+	files := []string{"manifest.json", "manifest.sig", "desired-state.json", "apply.sh", "verify.sh", "rollback.sh"}
+	remoteDir := "/etc/orca/cluster/txns/" + txnID
+	mkdirCmd := fmt.Sprintf("mkdir -p %s", remoteDir)
+	if _, err := transport.Exec(ctx, peer, mkdirCmd); err != nil {
+		return fmt.Errorf("mkdir %s: %w", remoteDir, err)
+	}
+	for _, f := range files {
+		p := filepath.Join(localDir, f)
+		content, err := os.ReadFile(p)
+		if err != nil {
+			if os.IsNotExist(err) {
+				continue
+			}
+			return err
+		}
+		dst := remoteDir + "/" + f
+		if _, err := transport.WriteFileIdempotent(ctx, peer, dst, content, 0o644); err != nil {
+			return fmt.Errorf("write %s: %w", dst, err)
+		}
+	}
+	return nil
+}
+
+type rotateSSHKeysResult struct {
+	Deployed   int      `json:"deployed"`
+	Failed     []string `json:"failed,omitempty"`
+	OldKeyHash string   `json:"old_key_hash,omitempty"`
+}
+
+func rotateSSHKeys(ctx context.Context, transport driftTransport, nodes []*model.Node) (*rotateSSHKeysResult, error) {
+	pubPath := certpaths.SSHPubPath()
+	keyPath := certpaths.SSHKeyPath()
+
+	oldPub, _ := os.ReadFile(pubPath)
+
+	newPriv, newPub, err := generateEd25519Keypair()
+	if err != nil {
+		return nil, fmt.Errorf("generate new ssh key: %w", err)
+	}
+	if err := os.WriteFile(keyPath, newPriv, 0o600); err != nil {
+		return nil, fmt.Errorf("write new ssh key: %w", err)
+	}
+	if err := os.WriteFile(pubPath, newPub, 0o644); err != nil {
+		return nil, fmt.Errorf("write new ssh pub: %w", err)
+	}
+
+	res := &rotateSSHKeysResult{Failed: []string{}}
+	for i := range nodes {
+		n := nodes[i]
+		peer := peerAddrForNode(n)
+		if peer == "" {
+			continue
+		}
+		deployCmd := fmt.Sprintf("mkdir -p ~/.ssh && echo %s >> ~/.ssh/authorized_keys && chmod 600 ~/.ssh/authorized_keys", sshQuote(strings.TrimSpace(string(newPub))))
+		if _, err := transport.Exec(ctx, peer, deployCmd); err != nil {
+			res.Failed = append(res.Failed, n.Name)
+			continue
+		}
+		res.Deployed++
+	}
+
+	if len(oldPub) > 0 {
+		res.OldKeyHash = sshFingerprint(oldPub)
+	}
+	return res, nil
+}
+
+func generateEd25519Keypair() (privBytes []byte, pubBytes []byte, err error) {
+	pubKey, privKey, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		return nil, nil, err
+	}
+	sshPub, err := ssh.NewPublicKey(pubKey)
+	if err != nil {
+		return nil, nil, err
+	}
+	pubBytes = ssh.MarshalAuthorizedKey(sshPub)
+	pemBlock, err := ssh.MarshalPrivateKey(privKey, "")
+	if err != nil {
+		return nil, nil, err
+	}
+	privBytes = pem.EncodeToMemory(pemBlock)
+	return privBytes, pubBytes, nil
+}
+
+func sshFingerprint(pub []byte) string {
+	pk, _, _, _, err := ssh.ParseAuthorizedKey(pub)
+	if err != nil {
+		return ""
+	}
+	return ssh.FingerprintSHA256(pk)
+}
+
+func readCurrentLead(ctx context.Context) (string, error) {
+	leadPath := filepath.Join(paths.ClusterDir(), "lead")
+	b, err := os.ReadFile(leadPath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return "", nil
+		}
+		return "", err
+	}
+	return trimSpace(string(b)), nil
+}
+
+func writeCurrentLead(ctx context.Context, name string) error {
+	dir := paths.ClusterDir()
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return err
+	}
+	leadPath := filepath.Join(dir, "lead")
+	return os.WriteFile(leadPath, []byte(name), 0o644)
+}
+
+func trimSpace(s string) string {
+	for len(s) > 0 && (s[0] == ' ' || s[0] == '\t' || s[0] == '\n' || s[0] == '\r') {
+		s = s[1:]
+	}
+	for len(s) > 0 && (s[len(s)-1] == ' ' || s[len(s)-1] == '\t' || s[len(s)-1] == '\n' || s[len(s)-1] == '\r') {
+		s = s[:len(s)-1]
+	}
+	return s
+}
+
+func init() {
+	clusterRotateLeadCmd.Flags().StringVar(&rotateLeadTo, "to", "", "new lead host (required, must be a bare Linux node)")
+	clusterRotateLeadCmd.Flags().BoolVar(&rotateLeadForce, "force", false, "skip R-003 verification (use with caution)")
+	_ = rotateLeadDebug
+}