feat(P10b): drift detection (R-018/R-019/R-020, REQ-103..113)

internal/drift/drift.go: Detector (Watch via iter.Seq2, Aggregate,
Remediate with cooldown-on-success, Acknowledge), Config with tiered
cadence (critical 5s + Path units, standard 30s, default 60s).
internal/cli/drift.go: orca drift {show,watch,acknowledge,remediate,
config}. internal/emitter/drift_path.go: systemd Path+service unit
emitter (User=orca, ProtectSystem=strict). scripts/orca-drift-notify.sh
(sha256 event JSON), orca-remediate.sh (cooldown-on-success, transient
retry). Pre-flight gate (R-020, --force + per-ns scoping). orca
system user (REQ-111), NFS detection (D-233), orca job restart for
EnvironmentFile drift (D-235).

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

internal/cli/drift.go

@@ -0,0 +1,391 @@
+// Package cli: drift.go implements the `orca drift` subcommand family
+// (P10b, v0.11; R-018/R-019/R-020, REQ-104). Subcommands:
+//
+//	orca drift                       show current drift state (table)
+//	orca drift watch [--interval=2s] [--paths=...] [--json]
+//	                                 stream drift events (iter.Seq2, D-017)
+//	                                 ctrl-c cancels via signal.NotifyContext (D-023)
+//	orca drift show [--peer <host>]  detailed drift events for a peer
+//	orca drift acknowledge <peer> <path>
+//	                                 record operator acknowledgment
+//	orca drift remediate <peer> <path> [--force]
+//	                                 trigger manual remediation
+//	orca drift config show           show current drift config
+//	orca drift config validate       validate config
+//
+// Plus the `orca job restart <name>` command for EnvironmentFile drift
+// (REQ-113, D-235) — restarts an allocation to pick up env-file drift.
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/drift"
+	"git.cloudinit.dev/coreci/orca/internal/sshpush"
+)
+
+var (
+	driftWatchInterval     time.Duration
+	driftWatchPaths        []string
+	driftShowPeer          string
+	driftConfigPath        string
+	driftRemediateForce    bool
+	driftAckPeer           string
+	driftAckPath           string
+	driftRemediatePeer     string
+	driftRemediatePath     string
+	driftWatchPollOverride time.Duration
+)
+
+// driftTransport is the SSH-push surface the drift CLI needs. It
+// mirrors drift.Transport; tests substitute a mock.
+type driftTransport interface {
+	Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
+	WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error)
+	ReadFile(ctx context.Context, peer string, path string) ([]byte, error)
+}
+
+// driftTransportOverride is the package-level test seam.
+var driftTransportOverride driftTransport
+
+func driftTransportFromCtx() (driftTransport, error) {
+	if driftTransportOverride != nil {
+		return driftTransportOverride, nil
+	}
+	keyPath := certpaths.SSHKeyPath()
+	khPath := certpaths.KnownHostsPath()
+	return sshpush.NewTransport(keyPath, khPath), nil
+}
+
+// driftDetectorOverride is the package-level test seam for the
+// Detector itself. When non-nil it replaces the production detector
+// (which wraps a driftTransport). Tests set it and restore nil.
+var driftDetectorOverride drift.Detector
+
+func driftDetector() (drift.Detector, error) {
+	if driftDetectorOverride != nil {
+		return driftDetectorOverride, nil
+	}
+	t, err := driftTransportFromCtx()
+	if err != nil {
+		return nil, err
+	}
+	return drift.NewDefaultDetector(t), nil
+}
+
+var driftCmd = &cobra.Command{
+	Use:   "drift",
+	Short: "Detect and remediate control-plane drift (P10b)",
+	Long: `Orca's drift detector is a BACKSTOP (R-019): the primary
+consistency mechanism is systemd / Traefik / step-ca / Syncthing
+themselves. The detector polls the lead's aggregated drift state
+(drift-events-aggregated.json) and can trigger orca-remediate.sh for
+auto-remediable paths. Pre-flight drift blocks txn apply (R-020).`,
+}
+
+var driftShowCmd = &cobra.Command{
+	Use:   "show",
+	Short: "Show current drift state (table format)",
+	Long:  `Show the aggregated drift events from the lead. Use --peer to filter to a single peer.`,
+	Args:  cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		d, err := driftDetector()
+		if err != nil {
+			return fmt.Errorf("drift detector: %w", err)
+		}
+		events, err := d.Aggregate(cmd.Context(), driftShowPeer)
+		if err != nil {
+			return fmt.Errorf("aggregate: %w", err)
+		}
+		if driftShowPeer != "" {
+			var filtered []drift.Event
+			for _, e := range events {
+				if e.Host == driftShowPeer {
+					filtered = append(filtered, e)
+				}
+			}
+			events = filtered
+		}
+		if jsonOutput {
+			return printJSON(events)
+		}
+		out := cmd.OutOrStdout()
+		if len(events) == 0 {
+			fmt.Fprintln(out, "No drift events.")
+			return nil
+		}
+		fmt.Fprintf(out, "%-20s  %-20s  %-40s  %-10s  %-10s\n", "EVENT-ID", "HOST", "PATH", "STATUS", "CONFIRMED")
+		for _, e := range events {
+			path := e.Path
+			if len(path) > 40 {
+				path = "..." + path[len(path)-37:]
+			}
+			confirmed := "no"
+			if e.DriftConfirmed {
+				confirmed = "yes"
+			}
+			fmt.Fprintf(out, "%-20s  %-20s  %-40s  %-10s  %-10s\n", e.EventID, e.Host, path, e.Status, confirmed)
+		}
+		return nil
+	},
+}
+
+var driftWatchCmd = &cobra.Command{
+	Use:   "watch",
+	Short: "Stream drift events (ctrl-c to cancel)",
+	Long: `Stream drift events from the lead's aggregated state. Default
+poll is 2s; override with --interval. Use --paths=<glob1>,<glob2> to
+filter. Uses iter.Seq2 (D-017) and signal.NotifyContext for ctrl-c
+(D-023).`,
+	Args: cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		d, err := driftDetector()
+		if err != nil {
+			return fmt.Errorf("drift detector: %w", err)
+		}
+		ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
+		defer cancel()
+		interval := driftWatchInterval
+		if interval <= 0 {
+			interval = 2 * time.Second
+		}
+		var specs []drift.PathSpec
+		for _, p := range driftWatchPaths {
+			if p == "" {
+				continue
+			}
+			specs = append(specs, drift.PathSpec{Pattern: p, Interval: interval})
+		}
+		out := cmd.OutOrStdout()
+		for e, err := range d.Watch(ctx, specs) {
+			if err != nil {
+				if errors.Is(err, context.Canceled) {
+					return nil
+				}
+				fmt.Fprintf(out, "watch error: %v\n", err)
+				continue
+			}
+			if jsonOutput {
+				line, _ := json.Marshal(e)
+				fmt.Fprintln(out, string(line))
+			} else {
+				fmt.Fprintf(out, "%s  [%s]  %s  %s  %s  confirmed=%t\n", e.TS.Format(time.RFC3339), e.EventID, e.Host, e.Path, e.Status, e.DriftConfirmed)
+			}
+		}
+		return nil
+	},
+}
+
+var driftAckCmd = &cobra.Command{
+	Use:   "acknowledge <peer> <path>",
+	Short: "Record operator acknowledgment of drift on a peer",
+	Long: `Record operator acknowledgment for the given path in
+drift-acknowledgments.json on the lead. Acknowledged drift no longer
+blocks txn apply for that namespace (R-020).`,
+	Args: cobra.ExactArgs(2),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		peer := args[0]
+		path := args[1]
+		d, err := driftDetector()
+		if err != nil {
+			return fmt.Errorf("drift detector: %w", err)
+		}
+		if err := d.Acknowledge(cmd.Context(), peer, path); err != nil {
+			return fmt.Errorf("acknowledge: %w", err)
+		}
+		printResult(fmt.Sprintf("✓ Acknowledged drift on %s for %s", peer, path), map[string]any{
+			"peer": peer, "path": path, "status": "acknowledged",
+		})
+		return nil
+	},
+}
+
+var driftRemediateCmd = &cobra.Command{
+	Use:   "remediate <peer> <path>",
+	Short: "Trigger manual remediation of drift on a peer",
+	Long: `Trigger orca-remediate.sh on the lead for the given path.
+--force bypasses the cooldown window (C4).`,
+	Args: cobra.ExactArgs(2),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		peer := args[0]
+		path := args[1]
+		d, err := driftDetector()
+		if err != nil {
+			return fmt.Errorf("drift detector: %w", err)
+		}
+		if err := d.Remediate(cmd.Context(), peer, path, driftRemediateForce); err != nil {
+			if errors.Is(err, drift.ErrCooldown) {
+				printResult(fmt.Sprintf("✗ Remediation in cooldown for %s on %s (use --force to bypass)", path, peer), map[string]any{
+					"peer": peer, "path": path, "status": "cooldown",
+				})
+				return nil
+			}
+			return fmt.Errorf("remediate: %w", err)
+		}
+		printResult(fmt.Sprintf("✓ Remediated drift on %s for %s", peer, path), map[string]any{
+			"peer": peer, "path": path, "status": "remediated", "force": driftRemediateForce,
+		})
+		return nil
+	},
+}
+
+var driftConfigCmd = &cobra.Command{
+	Use:   "config",
+	Short: "Show or validate the drift config",
+}
+
+var driftConfigShowCmd = &cobra.Command{
+	Use:   "show",
+	Short: "Show the current drift config",
+	Args:  cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		cfg, err := drift.LoadConfig(driftConfigPath)
+		if err != nil {
+			return fmt.Errorf("load config: %w", err)
+		}
+		if jsonOutput {
+			return printJSON(cfg)
+		}
+		out := cmd.OutOrStdout()
+		fmt.Fprintf(out, "Polling: enabled=%t default=%s max_peers=%d\n", cfg.Polling.Enabled, cfg.Polling.DefaultInterval, cfg.Polling.MaxConcurrentPeers)
+		fmt.Fprintln(out, "Critical paths:")
+		for _, p := range cfg.Paths.Critical {
+			fmt.Fprintf(out, "  [%s] %s (interval=%s, path_unit=%t)\n", p.Tier, p.Pattern, p.Interval, p.SystemdPathUnit)
+		}
+		fmt.Fprintln(out, "Standard paths:")
+		for _, p := range cfg.Paths.Standard {
+			fmt.Fprintf(out, "  [%s] %s (interval=%s)\n", p.Tier, p.Pattern, p.Interval)
+		}
+		fmt.Fprintln(out, "Excluded paths:")
+		for _, p := range cfg.Paths.Excluded {
+			fmt.Fprintf(out, "  %s\n", p)
+		}
+		fmt.Fprintf(out, "Remediation: auto=%t notify=%t\n", cfg.Remediate.Auto, cfg.Remediate.NotifyOnRemediation)
+		if len(cfg.Remediate.AutoPaths) > 0 {
+			fmt.Fprintln(out, "  auto_paths:")
+			for _, p := range cfg.Remediate.AutoPaths {
+				fmt.Fprintf(out, "    %s\n", p)
+			}
+		}
+		if len(cfg.Remediate.RequireApproval) > 0 {
+			fmt.Fprintln(out, "  require_approval:")
+			for _, p := range cfg.Remediate.RequireApproval {
+				fmt.Fprintf(out, "    %s\n", p)
+			}
+		}
+		return nil
+	},
+}
+
+var driftConfigValidateCmd = &cobra.Command{
+	Use:   "validate",
+	Short: "Validate the drift config",
+	Args:  cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		cfg, err := drift.LoadConfig(driftConfigPath)
+		if err != nil {
+			return fmt.Errorf("load config: %w", err)
+		}
+		if err := drift.ValidateConfig(cfg); err != nil {
+			printResult(fmt.Sprintf("✗ Config invalid: %v", err), map[string]any{"valid": false, "error": err.Error()})
+			return err
+		}
+		printResult("✓ Config valid", map[string]any{"valid": true})
+		return nil
+	},
+}
+
+// jobRestartCmd implements `orca job restart <name>` (REQ-113, D-235):
+// restart an allocation on its peer to pick up EnvironmentFile drift.
+var jobRestartCmd = &cobra.Command{
+	Use:   "restart <name>",
+	Short: "Restart an allocation to pick up EnvironmentFile drift (REQ-113)",
+	Long: `SSH to the peer running allocation <name> and run
+systemctl restart orca-alloc-<id>.service. This is the normal
+allocation lifecycle (NOT file-level remediation) and is triggered
+when /etc/orca/allocs/<id>/env drifts.`,
+	Args: cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		name := args[0]
+		peer := jobRestartPeer
+		if peer == "" {
+			return fmt.Errorf("--peer is required for job restart")
+		}
+		transport, err := driftTransportFromCtx()
+		if err != nil {
+			return fmt.Errorf("ssh transport: %w", err)
+		}
+		unit := fmt.Sprintf("orca-alloc-%s.service", name)
+		restartCmd := fmt.Sprintf("systemctl restart %s", shellQuoteDrift(unit))
+		out, err := transport.Exec(cmd.Context(), peer, restartCmd)
+		if err != nil {
+			return fmt.Errorf("restart %s on %s: %w (output: %s)", unit, peer, err, string(out))
+		}
+		printResult(fmt.Sprintf("✓ Restarted %s on %s", unit, peer), map[string]any{
+			"unit": unit, "peer": peer, "status": "restarted", "output": string(out),
+		})
+		return nil
+	},
+}
+
+var jobRestartPeer string
+
+func shellQuoteDrift(s string) string {
+	return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'"
+}
+
+func init() {
+	driftWatchCmd.Flags().DurationVar(&driftWatchInterval, "interval", 2*time.Second, "poll interval (default 2s)")
+	driftWatchCmd.Flags().StringSliceVar(&driftWatchPaths, "paths", nil, "comma-separated glob patterns to watch (default: all)")
+	driftShowCmd.Flags().StringVar(&driftShowPeer, "peer", "", "filter to a single peer host")
+	driftRemediateCmd.Flags().BoolVar(&driftRemediateForce, "force", false, "bypass the cooldown window (C4)")
+	driftConfigCmd.PersistentFlags().StringVar(&driftConfigPath, "config", "", "path to drift config JSON (default: built-in)")
+	jobRestartCmd.Flags().StringVar(&jobRestartPeer, "peer", "", "peer address (host:port) running the allocation")
+
+	driftCmd.AddCommand(driftShowCmd)
+	driftCmd.AddCommand(driftWatchCmd)
+	driftCmd.AddCommand(driftAckCmd)
+	driftCmd.AddCommand(driftRemediateCmd)
+	driftCmd.AddCommand(driftConfigCmd)
+	driftConfigCmd.AddCommand(driftConfigShowCmd)
+	driftConfigCmd.AddCommand(driftConfigValidateCmd)
+	rootCmd.AddCommand(driftCmd)
+
+	jobCmd.AddCommand(jobRestartCmd)
+}
+
+// writeClusterDefaultDriftConfig writes the canonical default drift
+// config to the cluster state dir so the lead has a reference copy.
+// Best-effort; caller logs failures.
+func writeClusterDefaultDriftConfig() error {
+	dir := filepath.Join(os.Getenv("ORCA_HOME"), "cluster", "state")
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return fmt.Errorf("mkdir cluster state: %w", err)
+	}
+	path := filepath.Join(dir, "drift.json")
+	cfg := drift.DefaultConfig()
+	raw, err := json.MarshalIndent(cfg, "", "  ")
+	if err != nil {
+		return fmt.Errorf("marshal drift config: %w", err)
+	}
+	tmp := path + ".tmp"
+	if err := os.WriteFile(tmp, raw, 0o644); err != nil {
+		return fmt.Errorf("write drift config: %w", err)
+	}
+	if err := os.Rename(tmp, path); err != nil {
+		return fmt.Errorf("rename drift config: %w", err)
+	}
+	return nil
+}

internal/cli/drift_test.go

@@ -0,0 +1,631 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"iter"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"git.cloudinit.dev/coreci/orca/internal/drift"
+)
+
+type mockDriftTransport struct {
+	execOut []byte
+	execErr error
+	readOut []byte
+	readErr error
+	writes  []mockDriftWrite
+	execFn  func(ctx context.Context, peer string, cmd string) ([]byte, error)
+	execs   []string
+}
+
+type mockDriftWrite struct {
+	peer    string
+	path    string
+	content []byte
+	mode    os.FileMode
+}
+
+func (m *mockDriftTransport) Exec(ctx context.Context, peer string, cmd string) ([]byte, error) {
+	if m.execFn != nil {
+		return m.execFn(ctx, peer, cmd)
+	}
+	m.execs = append(m.execs, cmd)
+	return m.execOut, m.execErr
+}
+
+func (m *mockDriftTransport) WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error) {
+	m.writes = append(m.writes, mockDriftWrite{peer, path, content, mode})
+	return true, nil
+}
+
+func (m *mockDriftTransport) ReadFile(ctx context.Context, peer string, path string) ([]byte, error) {
+	return m.readOut, m.readErr
+}
+
+// fakeDetector is a record-replay drift.Detector for CLI tests.
+type fakeDetector struct {
+	aggEvents      []drift.Event
+	aggErr         error
+	remediateErr   error
+	ackWrites      int
+	remediateCalls []remediateCall
+}
+
+type remediateCall struct {
+	peer  string
+	path  string
+	force bool
+}
+
+func (f *fakeDetector) Watch(ctx context.Context, paths []drift.PathSpec) iter.Seq2[drift.Event, error] {
+	return func(yield func(drift.Event, error) bool) {
+		for _, e := range f.aggEvents {
+			if !yield(e, nil) {
+				return
+			}
+		}
+		<-ctx.Done()
+	}
+}
+
+func (f *fakeDetector) Aggregate(ctx context.Context, leadPeer string) ([]drift.Event, error) {
+	return f.aggEvents, f.aggErr
+}
+
+func (f *fakeDetector) Remediate(ctx context.Context, leadPeer, path string, force bool) error {
+	f.remediateCalls = append(f.remediateCalls, remediateCall{leadPeer, path, force})
+	return f.remediateErr
+}
+
+func (f *fakeDetector) Acknowledge(ctx context.Context, leadPeer, path string) error {
+	f.ackWrites++
+	return nil
+}
+
+func setupDriftCLITest(t *testing.T) string {
+	t.Helper()
+	home := t.TempDir()
+	t.Setenv("ORCA_HOME", home)
+	t.Setenv("ORCA_LEAD_STATE_DIR", filepath.Join(home, "state"))
+	return home
+}
+
+func TestDriftCmdRegistered(t *testing.T) {
+	for _, c := range rootCmd.Commands() {
+		if c.Name() == "drift" {
+			return
+		}
+	}
+	t.Fatal("drift command not registered on root")
+}
+
+func TestDriftSubcommandsRegistered(t *testing.T) {
+	for _, c := range rootCmd.Commands() {
+		if c.Name() != "drift" {
+			continue
+		}
+		want := map[string]bool{
+			"show":        false,
+			"watch":       false,
+			"acknowledge": false,
+			"remediate":   false,
+			"config":      false,
+		}
+		for _, sub := range c.Commands() {
+			if _, ok := want[sub.Name()]; ok {
+				want[sub.Name()] = true
+			}
+		}
+		for name, found := range want {
+			if !found {
+				t.Errorf("drift subcommand %q not registered", name)
+			}
+		}
+		return
+	}
+	t.Fatal("drift command not registered")
+}
+
+func TestDriftConfigSubcommands(t *testing.T) {
+	for _, c := range rootCmd.Commands() {
+		if c.Name() != "drift" {
+			continue
+		}
+		for _, sub := range c.Commands() {
+			if sub.Name() != "config" {
+				continue
+			}
+			want := map[string]bool{"show": false, "validate": false}
+			for _, s := range sub.Commands() {
+				if _, ok := want[s.Name()]; ok {
+					want[s.Name()] = true
+				}
+			}
+			for name, found := range want {
+				if !found {
+					t.Errorf("drift config subcommand %q not registered", name)
+				}
+			}
+			return
+		}
+	}
+	t.Fatal("drift config not registered")
+}
+
+func TestJobRestartRegistered(t *testing.T) {
+	for _, c := range rootCmd.Commands() {
+		if c.Name() != "job" {
+			continue
+		}
+		for _, sub := range c.Commands() {
+			if sub.Name() == "restart" {
+				return
+			}
+		}
+	}
+	t.Fatal("job restart not registered")
+}
+
+func TestDriftShowEmpty(t *testing.T) {
+	setupDriftCLITest(t)
+	fd := &fakeDetector{aggEvents: nil}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "show"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift show: %v", err)
+	}
+	if !bytesContains(buf.String(), "No drift events") {
+		t.Errorf("empty show output: %s", buf.String())
+	}
+}
+
+func TestDriftShowTable(t *testing.T) {
+	setupDriftCLITest(t)
+	fd := &fakeDetector{aggEvents: []drift.Event{
+		{EventID: "E1", Host: "peer1", Path: "/etc/traefik/dynamic/orca.yml", Status: drift.StatusModified, DriftConfirmed: true},
+	}}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "show"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift show: %v", err)
+	}
+	out := buf.String()
+	for _, want := range []string{"E1", "peer1", "modified"} {
+		if !bytesContains(out, want) {
+			t.Errorf("output missing %q: %s", want, out)
+		}
+	}
+}
+
+func TestDriftShowJSON(t *testing.T) {
+	setupDriftCLITest(t)
+	fd := &fakeDetector{aggEvents: []drift.Event{
+		{EventID: "E1", Host: "peer1", Path: "/etc/x", Status: drift.StatusCreated, DriftConfirmed: true},
+	}}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	_ = rootCmd.PersistentFlags().Set("json", "true")
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "show"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift show --json: %v", err)
+	}
+	if !bytesContains(buf.String(), `"event_id"`) {
+		t.Errorf("json output missing event_id: %s", buf.String())
+	}
+}
+
+func TestDriftShowPeerFilter(t *testing.T) {
+	setupDriftCLITest(t)
+	fd := &fakeDetector{aggEvents: []drift.Event{
+		{EventID: "E1", Host: "peer1", Path: "/a", Status: drift.StatusModified, DriftConfirmed: true},
+		{EventID: "E2", Host: "peer2", Path: "/b", Status: drift.StatusModified, DriftConfirmed: true},
+	}}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "show", "--peer", "peer1"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift show --peer: %v", err)
+	}
+	out := buf.String()
+	if !bytesContains(out, "E1") {
+		t.Errorf("filtered output should have E1: %s", out)
+	}
+	if bytesContains(out, "E2") {
+		t.Errorf("filtered output should NOT have E2: %s", out)
+	}
+}
+
+func TestDriftWatchStreamsAndCancels(t *testing.T) {
+	setupDriftCLITest(t)
+	// Use a fakeDetector whose Watch yields one event then blocks on
+	// ctx so the stream terminates when ctrl-c (signal.NotifyContext)
+	// cancels. We simulate the cancel by constructing a fake that yields
+	// then returns when the consumer stops pulling OR ctx is cancelled.
+	fd := &drainingFakeDetector{events: []drift.Event{
+		{EventID: "E1", Host: "p", Path: "/etc/x", Status: drift.StatusModified, DriftConfirmed: true},
+	}}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "watch", "--interval", "10ms"})
+	// Inject a context that auto-cancels after the events drain so
+	// the watch loop exits without polluting rootCmd's context (which
+	// is shared across tests). We use PersistentPreRunE's context by
+	// overriding it here and restoring after.
+	origCtx := rootCmd.Context()
+	ctx, cancel := context.WithCancel(origCtx)
+	defer cancel()
+	rootCmd.SetContext(ctx)
+	fd.cancelAfterYield = cancel
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift watch: %v", err)
+	}
+	// Restore rootCmd context for subsequent tests.
+	rootCmd.SetContext(origCtx)
+	if !bytesContains(buf.String(), "E1") {
+		t.Errorf("watch output missing E1: %s", buf.String())
+	}
+}
+
+// drainingFakeDetector yields the events then cancels the provided
+// cancel func (so the watch loop's signal.NotifyContext ctx is
+// cancelled and the stream terminates cleanly).
+type drainingFakeDetector struct {
+	events           []drift.Event
+	cancelAfterYield context.CancelFunc
+	remediateCalls   []remediateCall
+	ackWrites        int
+}
+
+func (d *drainingFakeDetector) Watch(ctx context.Context, paths []drift.PathSpec) iter.Seq2[drift.Event, error] {
+	return func(yield func(drift.Event, error) bool) {
+		for _, e := range d.events {
+			if !yield(e, nil) {
+				return
+			}
+		}
+		if d.cancelAfterYield != nil {
+			d.cancelAfterYield()
+		}
+		<-ctx.Done()
+	}
+}
+func (d *drainingFakeDetector) Aggregate(ctx context.Context, leadPeer string) ([]drift.Event, error) {
+	return d.events, nil
+}
+func (d *drainingFakeDetector) Remediate(ctx context.Context, leadPeer, path string, force bool) error {
+	d.remediateCalls = append(d.remediateCalls, remediateCall{leadPeer, path, force})
+	return nil
+}
+func (d *drainingFakeDetector) Acknowledge(ctx context.Context, leadPeer, path string) error {
+	d.ackWrites++
+	return nil
+}
+
+func TestDriftAcknowledge(t *testing.T) {
+	setupDriftCLITest(t)
+	fd := &fakeDetector{}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "acknowledge", "peer1", "/etc/traefik/dynamic/orca.yml"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift acknowledge: %v", err)
+	}
+	if fd.ackWrites != 1 {
+		t.Errorf("ackWrites = %d, want 1", fd.ackWrites)
+	}
+	if !bytesContains(buf.String(), "Acknowledged") {
+		t.Errorf("output missing Acknowledged: %s", buf.String())
+	}
+}
+
+func TestDriftRemediate(t *testing.T) {
+	setupDriftCLITest(t)
+	fd := &fakeDetector{}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "remediate", "peer1", "/etc/x"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift remediate: %v", err)
+	}
+	if len(fd.remediateCalls) != 1 {
+		t.Fatalf("remediateCalls = %d, want 1", len(fd.remediateCalls))
+	}
+	if fd.remediateCalls[0].peer != "peer1" || fd.remediateCalls[0].path != "/etc/x" {
+		t.Errorf("remediate call: %+v", fd.remediateCalls[0])
+	}
+	if fd.remediateCalls[0].force {
+		t.Errorf("force should be false without --force")
+	}
+	if !bytesContains(buf.String(), "Remediated") {
+		t.Errorf("output missing Remediated: %s", buf.String())
+	}
+}
+
+func TestDriftRemediateForce(t *testing.T) {
+	setupDriftCLITest(t)
+	fd := &fakeDetector{}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "remediate", "peer1", "/etc/x", "--force"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift remediate --force: %v", err)
+	}
+	if len(fd.remediateCalls) != 1 {
+		t.Fatalf("remediateCalls = %d, want 1", len(fd.remediateCalls))
+	}
+	if !fd.remediateCalls[0].force {
+		t.Errorf("force should be true with --force")
+	}
+}
+
+func TestDriftRemediateCooldown(t *testing.T) {
+	setupDriftCLITest(t)
+	fd := &fakeDetector{remediateErr: drift.ErrCooldown}
+	driftDetectorOverride = fd
+	defer func() { driftDetectorOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "remediate", "peer1", "/etc/x"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift remediate cooldown: %v", err)
+	}
+	if !bytesContains(buf.String(), "cooldown") {
+		t.Errorf("output should mention cooldown: %s", buf.String())
+	}
+}
+
+func TestDriftConfigShow(t *testing.T) {
+	setupDriftCLITest(t)
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "config", "show"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift config show: %v", err)
+	}
+	out := buf.String()
+	for _, want := range []string{"Polling:", "Critical paths:", "Standard paths:", "Excluded paths:", "Remediation:"} {
+		if !bytesContains(out, want) {
+			t.Errorf("config show missing %q: %s", want, out)
+		}
+	}
+}
+
+func TestDriftConfigValidate(t *testing.T) {
+	setupDriftCLITest(t)
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "config", "validate"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("drift config validate: %v", err)
+	}
+	if !bytesContains(buf.String(), "valid") {
+		t.Errorf("output missing 'valid': %s", buf.String())
+	}
+}
+
+func TestDriftConfigValidateFails(t *testing.T) {
+	setupDriftCLITest(t)
+	cfgPath := filepath.Join(t.TempDir(), "drift.json")
+	bad := `{"polling":{"enabled":true,"default_interval":60000000000,"max_concurrent_peers":4},"paths":{"critical":[{"tier":"critical","pattern":"","interval":5000000000}]},"remediate":{"auto":true}}`
+	if err := os.WriteFile(cfgPath, []byte(bad), 0o644); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"drift", "config", "validate", "--config", cfgPath})
+	err := rootCmd.Execute()
+	if err == nil {
+		t.Fatal("expected validate error")
+	}
+}
+
+func TestJobRestartRequiresPeer(t *testing.T) {
+	setupDriftCLITest(t)
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"job", "restart", "alloc1"})
+	err := rootCmd.Execute()
+	if err == nil {
+		t.Fatal("expected error for missing --peer")
+	}
+}
+
+func TestJobRestartExecs(t *testing.T) {
+	setupDriftCLITest(t)
+	mt := &mockDriftTransport{execOut: []byte("restarted")}
+	driftTransportOverride = mt
+	defer func() { driftTransportOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"job", "restart", "alloc1", "--peer", "peer1:22"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("job restart: %v", err)
+	}
+	if len(mt.execs) != 1 {
+		t.Fatalf("execs = %d, want 1", len(mt.execs))
+	}
+	if !bytesContains(mt.execs[0], "orca-alloc-alloc1.service") {
+		t.Errorf("restart cmd missing: %s", mt.execs[0])
+	}
+}
+
+func TestJobRestartTransientError(t *testing.T) {
+	setupDriftCLITest(t)
+	mt := &mockDriftTransport{execErr: fmt.Errorf("connection refused")}
+	driftTransportOverride = mt
+	defer func() { driftTransportOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"job", "restart", "alloc1", "--peer", "peer1:22"})
+	err := rootCmd.Execute()
+	if err == nil {
+		t.Fatal("expected error for exec failure")
+	}
+	if !errors.Is(err, err) {
+	}
+}
+
+func TestPeerSetupCmdRegistered(t *testing.T) {
+	for _, c := range rootCmd.Commands() {
+		if c.Name() == "peer-setup" {
+			return
+		}
+	}
+	t.Fatal("peer-setup command not registered")
+}
+
+func TestPeerSetupCreatesUserAndDir(t *testing.T) {
+	setupDriftCLITest(t)
+	mt := &mockDriftTransport{execOut: []byte("ext4")}
+	peerSetupTransportOverride = mt
+	defer func() { peerSetupTransportOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"peer-setup", "peer1:22"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("peer-setup: %v", err)
+	}
+	if len(mt.execs) < 2 {
+		t.Fatalf("execs = %d, want >= 2", len(mt.execs))
+	}
+	useraddSeen := false
+	mkdirSeen := false
+	statSeen := false
+	for _, c := range mt.execs {
+		if bytesContains(c, "useradd -r orca") {
+			useraddSeen = true
+		}
+		if bytesContains(c, "mkdir -p /etc/orca/state/drift-events") {
+			mkdirSeen = true
+		}
+		if bytesContains(c, "stat -f") {
+			statSeen = true
+		}
+	}
+	if !useraddSeen {
+		t.Errorf("useradd not run")
+	}
+	if !mkdirSeen {
+		t.Errorf("mkdir drift-events not run")
+	}
+	if !statSeen {
+		t.Errorf("stat (NFS detect) not run")
+	}
+	if !bytesContains(buf.String(), "nfs=false") {
+		t.Errorf("output should report nfs=false: %s", buf.String())
+	}
+}
+
+func TestPeerSetupNoOrcaUser(t *testing.T) {
+	setupDriftCLITest(t)
+	mt := &mockDriftTransport{execOut: []byte("ext4")}
+	peerSetupTransportOverride = mt
+	defer func() { peerSetupTransportOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"peer-setup", "peer1:22", "--no-orca-user"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("peer-setup --no-orca-user: %v", err)
+	}
+	useraddSeen := false
+	for _, c := range mt.execs {
+		if bytesContains(c, "useradd -r orca") {
+			useraddSeen = true
+		}
+	}
+	if useraddSeen {
+		t.Errorf("useradd should NOT run with --no-orca-user")
+	}
+	if !bytesContains(buf.String(), "user=false") {
+		t.Errorf("output should report user=false: %s", buf.String())
+	}
+}
+
+func TestPeerSetupDetectsNFS(t *testing.T) {
+	setupDriftCLITest(t)
+	mt := &mockDriftTransport{execOut: []byte("nfs4")}
+	peerSetupTransportOverride = mt
+	defer func() { peerSetupTransportOverride = nil }()
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"peer-setup", "peer1:22"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("peer-setup: %v", err)
+	}
+	if !bytesContains(buf.String(), "nfs=true") {
+		t.Errorf("output should report nfs=true: %s", buf.String())
+	}
+}

internal/cli/namespace_test.go

@@ -49,6 +49,13 @@ func resetCommandFlags() {
 	txnApplyNamespace = ""
 	txnApplyTimeout = 5 * time.Minute
 	txnApplyLead, txnRollbackLead = "", ""
+	driftWatchInterval = 2 * time.Second
+	driftWatchPaths = nil
+	driftShowPeer = ""
+	driftConfigPath = ""
+	driftRemediateForce = false
+	jobRestartPeer = ""
+	peerSetupNoOrcaUser = false
 	resetNSFlags()
 	// Reset per-command output writers so tests that polluted them
 	// (e.g. daemon tests calling cmd.SetOut(&buf)) don't leak into

internal/cli/peer_setup.go

@@ -0,0 +1,136 @@
+// Package cli: peer_setup.go implements the orca system-user setup and
+// NFS detection on peers (P10b-T8/T9, v0.11, REQ-111, REQ-112/D-233).
+//
+// `orca node join` now also creates the `orca` system user on the peer
+// (so the systemd Path-unit services, which run as User=orca, have a
+// uid to run as). It also detects whether /etc/orca is on an NFS mount
+// and, when it is, skips emitting Path units for paths under /etc/orca
+// (falling back to polling for those paths — systemd Path units on NFS
+// are unreliable because inotify does not fire reliably over NFS).
+//
+// The setup is idempotent: re-running on an already-configured peer is
+// a no-op. A --no-orca-user flag skips user creation (for environments
+// with existing service accounts).
+package cli
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/spf13/cobra"
+)
+
+var peerSetupNoOrcaUser bool
+
+// peerSetupTransport is the SSH surface the peer-setup code needs. It
+// mirrors driftTransport; tests substitute a mock.
+type peerSetupTransport interface {
+	Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
+}
+
+// peerSetupTransportOverride is the test seam.
+var peerSetupTransportOverride peerSetupTransport
+
+func peerSetupTransportFromCtx() (peerSetupTransport, error) {
+	if peerSetupTransportOverride != nil {
+		return peerSetupTransportOverride, nil
+	}
+	t, err := driftTransportFromCtx()
+	if err != nil {
+		return nil, err
+	}
+	return t, nil
+}
+
+// PeerSetupResult records what the peer setup did.
+type PeerSetupResult struct {
+	UserCreated bool   `json:"user_created"`
+	EventsDir   string `json:"events_dir"`
+	NFSOnOrca   bool   `json:"nfs_on_orca"`
+	NFSMsg      string `json:"nfs_msg,omitempty"`
+}
+
+// setupOrcaUser runs the idempotent `useradd -r orca` and creates the
+// drift-events directory owned by orca:orca on the peer. Returns the
+// result; a transient SSH failure returns the error (no partial state).
+func setupOrcaUser(ctx context.Context, transport peerSetupTransport, peer string) (*PeerSetupResult, error) {
+	if peer == "" {
+		return nil, fmt.Errorf("peer setup: peer is empty")
+	}
+	res := &PeerSetupResult{EventsDir: "/etc/orca/state/drift-events"}
+
+	if !peerSetupNoOrcaUser {
+		useraddCmd := "useradd -r orca -s /usr/sbin/nologin 2>/dev/null || true"
+		if _, err := transport.Exec(ctx, peer, useraddCmd); err != nil {
+			return nil, fmt.Errorf("peer setup: useradd: %w", err)
+		}
+		res.UserCreated = true
+	}
+
+	mkdirCmd := fmt.Sprintf("mkdir -p %s && %s", res.EventsDir, chownDriftEvents(res.EventsDir))
+	if _, err := transport.Exec(ctx, peer, mkdirCmd); err != nil {
+		return nil, fmt.Errorf("peer setup: mkdir drift-events: %w", err)
+	}
+
+	nfs, msg := detectNFS(ctx, transport, peer, "/etc/orca")
+	res.NFSOnOrca = nfs
+	res.NFSMsg = msg
+	return res, nil
+}
+
+// chownDriftEvents returns the chown command for the drift-events dir.
+// When --no-orca-user is set the orca user may not exist; chown only
+// when the user was created.
+func chownDriftEvents(dir string) string {
+	if peerSetupNoOrcaUser {
+		return "true"
+	}
+	return fmt.Sprintf("chown orca:orca %s 2>/dev/null || true", dir)
+}
+
+// detectNFS checks whether the given path is on an NFS mount by running
+// `stat -f -c %T <path>` on the peer. When the fs type contains "nfs"
+// it returns (true, msg). Best-effort: a stat failure returns
+// (false, "stat unavailable").
+func detectNFS(ctx context.Context, transport peerSetupTransport, peer, path string) (bool, string) {
+	out, err := transport.Exec(ctx, peer, fmt.Sprintf("stat -f -c %%T %s 2>/dev/null || echo unknown", shellQuoteDrift(path)))
+	if err != nil {
+		return false, "stat unavailable: " + err.Error()
+	}
+	fsType := strings.TrimSpace(string(out))
+	if strings.Contains(fsType, "nfs") {
+		return true, fmt.Sprintf("%s is on NFS (%s); skipping Path units for /etc/orca paths", path, fsType)
+	}
+	return false, fsType
+}
+
+var peerSetupCmd = &cobra.Command{
+	Use:   "peer-setup <peer>",
+	Short: "Create the orca system user + drift-events dir on a peer (REQ-111)",
+	Long: `SSH to <peer> and idempotently create the orca system user
+(useradd -r orca -s /usr/sbin/nologin) and /etc/orca/state/drift-events/
+owned by orca:orca. Also detects NFS on /etc/orca (REQ-112/D-233) and
+logs a warning when /etc/orca is on NFS (Path units are skipped for
+those paths in that case). Use --no-orca-user to skip user creation
+(for environments with an existing service account).`,
+	Args: cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		peer := args[0]
+		transport, err := peerSetupTransportFromCtx()
+		if err != nil {
+			return fmt.Errorf("ssh transport: %w", err)
+		}
+		res, err := setupOrcaUser(cmd.Context(), transport, peer)
+		if err != nil {
+			return err
+		}
+		printResult(fmt.Sprintf("✓ Peer %s set up (user=%t, nfs=%t)", peer, res.UserCreated, res.NFSOnOrca), res)
+		return nil
+	},
+}
+
+func init() {
+	peerSetupCmd.Flags().BoolVar(&peerSetupNoOrcaUser, "no-orca-user", false, "skip orca system user creation (env has existing service account)")
+	rootCmd.AddCommand(peerSetupCmd)
+}