fix(P09): migration + operational safety — job stop, retention, logs cap (REQ-158)

- job stop: real systemctl stop via SSH (was DB-only soft stop)
  resolves node from alloc_history or --peer flag
- doctor db-retention: row count check for jobs/tasks/audit_log
  warns at 100k rows, suggests backup + cleanup
- logs --lines: cap at 50000 (default 1000); --since upper bound 7d
  prevents OOM from unbounded journalctl
- cache DB mode 0600 (was 0644; matches store.Open)
- upgrade cutover: backup file + atomic rename (was sed -i)
  rollback restores from backup on failure

Tests: job stop SSH, DB retention warning, logs lines cap, cache mode,
cutover backup-restore + atomic rename.

---ci---
project: orca
phase: 9
milestone: v0.13
status: complete
requirements:
  covered: [158]
---/ci---

internal/cache/cache.go

@@ -73,6 +73,11 @@ func Open(path string) (*Cache, error) {
 		_ = db.Close()
 		return nil, fmt.Errorf("ping cache sqlite: %w", err)
 	}
+	// REQ-158 / P09 T4: enforce 0600 on the cache DB file (SQLite
+	// creates it at umask, typically 0644). Match store.Open which
+	// chmods after open+ping (the file exists at this point). Non-fatal
+	// if chmod fails (e.g. the DB is at a path we don't own).
+	_ = os.Chmod(path, 0o600)
 	const schema = `CREATE TABLE IF NOT EXISTS cache_entries (
 		class       TEXT NOT NULL,
 		key         TEXT NOT NULL,

internal/cache/cache_test.go

@@ -2,6 +2,7 @@ package cache
 
 import (
 	"errors"
+	"os"
 	"path/filepath"
 	"testing"
 	"time"
@@ -225,3 +226,49 @@ func BenchmarkCacheHit(b *testing.B) {
 		}
 	}
 }
+
+
+// TestCache_FileMode0600 verifies that the cache DB file is created
+// with mode 0600 (not the default umask 0644) (REQ-158, P09 T4).
+func TestCache_FileMode0600(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "orca_cache.db")
+	c, err := Open(path)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	defer c.Close()
+
+	info, err := os.Stat(path)
+	if err != nil {
+		t.Fatalf("stat cache db: %v", err)
+	}
+	got := info.Mode().Perm()
+	if got != 0o600 {
+		t.Errorf("cache db mode = %04o, want 0600", got)
+	}
+}
+
+// TestCache_FileMode0600DefaultPath verifies that the cache DB at the
+// default path (ORCA_HOME) also gets 0600 (REQ-158, P09 T4).
+func TestCache_FileMode0600DefaultPath(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("ORCA_HOME", dir)
+	c, err := Open("")
+	if err != nil {
+		t.Fatalf("open default path: %v", err)
+	}
+	defer c.Close()
+
+	// The default path is paths.CacheDB() which is under ORCA_HOME.
+	// Find the db file.
+	dbPath := filepath.Join(dir, "orca_cache.db")
+	info, err := os.Stat(dbPath)
+	if err != nil {
+		t.Fatalf("stat cache db at %s: %v", dbPath, err)
+	}
+	got := info.Mode().Perm()
+	if got != 0o600 {
+		t.Errorf("cache db mode = %04o, want 0600", got)
+	}
+}

internal/cli/doctor.go

@@ -384,7 +384,81 @@ func checkOIDCHealth(ctx context.Context) []oidcCheckResult {
 	return results
 }
 
+// doctorDBRetentionCmd implements `orca doctor db-retention` (REQ-158,
+// P09 T2). Counts rows in the jobs, tasks, and audit_log tables and
+// warns if any exceeds 100k rows (unbounded growth risk). Suggests
+// `orca backup` + manual cleanup.
+var doctorDBRetentionCmd = &cobra.Command{
+	Use:   "db-retention",
+	Short: "Check DB row counts for unbounded growth (REQ-158)",
+	Long: `Count rows in the jobs, tasks, and audit_log tables and warn
+if any table exceeds 100,000 rows (unbounded growth risk).
+
+Large tables degrade query performance and inflate backup size. Run
+'orca backup' to capture a snapshot, then prune old rows manually
+(e.g. DELETE FROM tasks WHERE created_at < <cutoff>).
+
+Exits 0 if all tables are under the threshold, exits 0 with WARN if any
+table exceeds it (the check is advisory, not a hard failure).`,
+	Args: cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		ctx, cancel := context.WithTimeout(cmd.Context(), 10*time.Second)
+		defer cancel()
+
+		db, closer, err := openDB()
+		if err != nil {
+			return fmt.Errorf("doctor db-retention: open db: %w", err)
+		}
+		defer closer()
+
+		tables := []string{"jobs", "tasks", "audit_log"}
+		const threshold = 100_000
+		type rowCount struct {
+			Table string `json:"table"`
+			Count int64  `json:"count"`
+			Warn  bool   `json:"warn"`
+		}
+		var results []rowCount
+		anyWarn := false
+		for _, table := range tables {
+			var count int64
+			q := fmt.Sprintf("SELECT COUNT(*) FROM %s", table)
+			if err := db.QueryRowContext(ctx, q).Scan(&count); err != nil {
+				return fmt.Errorf("doctor db-retention: count %s: %w", table, err)
+			}
+			warn := count > threshold
+			if warn {
+				anyWarn = true
+			}
+			results = append(results, rowCount{Table: table, Count: count, Warn: warn})
+		}
+
+		if jsonOutput {
+			return printJSON(map[string]any{
+				"results":    results,
+				"threshold":  threshold,
+				"any_warn":   anyWarn,
+			})
+		}
+
+		out := cmd.OutOrStdout()
+		for _, r := range results {
+			status := "ok"
+			if r.Warn {
+				status = "WARN"
+			}
+			fmt.Fprintf(out, "%-12s %-5s  %d rows (threshold: %d)\n", r.Table, status, r.Count, threshold)
+		}
+		if anyWarn {
+			fmt.Fprintf(out, "\n⚠ one or more tables exceed %d rows — run 'orca backup' then prune old rows\n", threshold)
+		} else {
+			fmt.Fprintln(out, "\n✓ all tables under retention threshold")
+		}
+		return nil
+	},
+}
+
 func init() {
-	doctorCmd.AddCommand(doctorCertCmd, doctorNetworkCmd, doctorDBCmd, doctorOSCmd, doctorProxmoxCmd, noOrcaOnServerCmd, doctorNftCmd, doctorAuditCmd, doctorModesCmd, doctorOIDCCmd)
+	doctorCmd.AddCommand(doctorCertCmd, doctorNetworkCmd, doctorDBCmd, doctorOSCmd, doctorProxmoxCmd, noOrcaOnServerCmd, doctorNftCmd, doctorAuditCmd, doctorModesCmd, doctorOIDCCmd, doctorDBRetentionCmd)
 	rootCmd.AddCommand(doctorCmd)
 }

internal/cli/doctor_test.go

@@ -2,9 +2,14 @@ package cli
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"strings"
 	"testing"
+	"time"
+
+	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/store"
 )
 
 func TestDoctorText(t *testing.T) {
@@ -194,3 +199,92 @@ func TestDoctorProxmoxJSON(t *testing.T) {
 		t.Errorf("doctor proxmox --json missing Name: %v", result)
 	}
 }
+
+
+// TestDoctorDBRetention verifies that `orca doctor db-retention` counts
+// rows in jobs, tasks, and audit_log and warns when a table exceeds
+// 100k rows (REQ-158, P09 T7).
+func TestDoctorDBRetention(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	if err := runInit(discardWriter{}); err != nil {
+		t.Fatalf("init: %v", err)
+	}
+
+	// Insert 100001 rows into the audit_log table to trigger the warning.
+	// Use a multi-row VALUES insert in batches for speed.
+	db, err := store.Open(certpaths.DBPath())
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer db.Close()
+	ctx := context.Background()
+	// Build a batch insert: 500 rows per INSERT in a transaction.
+	// SQLite handles this much faster than 100k individual inserts.
+	const totalRows = 100001
+	const batchSize = 500
+	inserted := 0
+	for inserted < totalRows {
+		remaining := totalRows - inserted
+		batch := batchSize
+		if remaining < batch {
+			batch = remaining
+		}
+		var placeholders strings.Builder
+		var args []any
+		for j := 0; j < batch; j++ {
+			if j > 0 {
+				placeholders.WriteString(",")
+			}
+			placeholders.WriteString("(?, 'test', 'test.action', 'test-resource', 'success')")
+			args = append(args, time.Now().UTC())
+		}
+		q := "INSERT INTO audit_log (timestamp, actor, action, resource, result) VALUES " + placeholders.String()
+		if _, err := db.ExecContext(ctx, q, args...); err != nil {
+			t.Fatalf("batch insert at offset %d: %v", inserted, err)
+		}
+		inserted += batch
+	}
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"doctor", "db-retention"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("doctor db-retention: %v", err)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "audit_log") {
+		t.Errorf("output missing audit_log table: %s", out)
+	}
+	if !strings.Contains(out, "WARN") {
+		t.Errorf("output should contain WARN for audit_log exceeding threshold: %s", out)
+	}
+	if !strings.Contains(out, "backup") {
+		t.Errorf("output should suggest 'orca backup': %s", out)
+	}
+}
+
+// TestDoctorDBRetentionNoWarn verifies that with a small DB no warning
+// is emitted (REQ-158, P09 T7).
+func TestDoctorDBRetentionNoWarn(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	if err := runInit(discardWriter{}); err != nil {
+		t.Fatalf("init: %v", err)
+	}
+
+	resetRootFlags(t)
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"doctor", "db-retention"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("doctor db-retention: %v", err)
+	}
+	out := buf.String()
+	if strings.Contains(out, "WARN") {
+		t.Errorf("output should NOT contain WARN for small DB: %s", out)
+	}
+}

internal/cli/job.go

@@ -2,6 +2,7 @@ package cli
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -14,9 +15,11 @@ import (
 	"github.com/google/uuid"
 	"github.com/spf13/cobra"
 
+	"git.cloudinit.dev/coreci/orca/internal/certpaths"
 	"git.cloudinit.dev/coreci/orca/internal/engine"
 	"git.cloudinit.dev/coreci/orca/internal/jobspec"
 	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/sshpush"
 	"git.cloudinit.dev/coreci/orca/internal/store"
 )
 
@@ -290,11 +293,79 @@ func renderJobTable(jobs []*model.Job) string {
 	return out
 }
 
+// jobStopTransport is the SSH command-execution seam used by
+// `orca job stop`. *sshpush.Transport satisfies it via Exec; tests
+// inject a mock (same pattern as driftTransport / drainExecer).
+type jobStopTransport interface {
+	Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
+}
+
+// jobStopTransportOverride is the package-level test seam for the
+// SSH transport used by `orca job stop`. When non-nil it replaces the
+// production transport; tests set it and restore nil in cleanup.
+var jobStopTransportOverride jobStopTransport
+
+// jobStopTimeout is the SSH command timeout for `orca job stop`.
+var jobStopTimeout time.Duration
+
+// jobStopPeer is the optional --peer override for `orca job stop`.
+// When empty, the node is looked up from the alloc_history table
+// (latest entry for the job id). When set, the SSH stop targets that
+// peer directly.
+var jobStopPeer string
+
+func jobStopTransportFromCtx() (jobStopTransport, error) {
+	if jobStopTransportOverride != nil {
+		return jobStopTransportOverride, nil
+	}
+	keyPath := certpaths.SSHKeyPath()
+	khPath := certpaths.KnownHostsPath()
+	return sshpush.NewTransport(keyPath, khPath), nil
+}
+
+// nodeForJob looks up the node that ran (or is running) a job by
+// searching the alloc_history table for the latest entry for the
+// given job id. Returns nil if no history entry exists (the job may
+// have been run locally or pre-dates alloc_history).
+func nodeForJob(ctx context.Context, db *sql.DB, jobID string) (*model.Node, error) {
+	hist := store.NewAllocHistoryRepo(db)
+	if err := hist.EnsureSchema(ctx); err != nil {
+		return nil, fmt.Errorf("alloc history schema: %w", err)
+	}
+	entries, err := hist.List(ctx, store.HistoryFilter{JobID: jobID})
+	if err != nil {
+		return nil, fmt.Errorf("alloc history list: %w", err)
+	}
+	if len(entries) == 0 {
+		return nil, nil
+	}
+	// Pick the latest entry (List returns ASC; take the last).
+	latest := entries[len(entries)-1]
+	if latest.NodeID == "" {
+		return nil, nil
+	}
+	nodeRepo := store.NewNodeRepo(db)
+	n, err := nodeRepo.Get(ctx, latest.NodeID)
+	if err != nil {
+		if errors.Is(err, store.ErrNotFound) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("lookup node %s: %w", latest.NodeID, err)
+	}
+	return n, nil
+}
+
 var jobStopCmd = &cobra.Command{
 	Use:   "stop [job-id]",
 	Short: "Stop a running job",
-	Long:  "Mark a job as stopped. Note: this is a soft stop (cancel context for the daemon).",
-	Args:  cobra.MaximumNArgs(1),
+	Long: `Stop a running job by sending 'systemctl stop orca-alloc-<name>-*'
+to the node running the allocation via SSH, then mark the job as
+stopped in the DB (REQ-158, P09 T1).
+
+If --peer is not given, the node is looked up from the allocation
+history. If no node is found, the DB status is updated anyway (soft
+stop fallback for local-run jobs).`,
+	Args: cobra.MaximumNArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
 		id := stopID
 		if id == "" && len(args) > 0 {
@@ -303,8 +374,6 @@ var jobStopCmd = &cobra.Command{
 		if id == "" {
 			return fmt.Errorf("job id required (--id or argument)")
 		}
-		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
-		defer cancel()
 
 		db, closer, err := openDB()
 		if err != nil {
@@ -312,6 +381,9 @@ var jobStopCmd = &cobra.Command{
 		}
 		defer closer()
 
+		ctx, cancel := context.WithTimeout(cmd.Context(), 30*time.Second)
+		defer cancel()
+
 		repo := store.NewJobRepo(db)
 		job, err := repo.Get(ctx, id)
 		if err != nil {
@@ -320,6 +392,52 @@ var jobStopCmd = &cobra.Command{
 			}
 			return err
 		}
+
+		// Determine the peer to SSH to. --peer takes precedence;
+		// otherwise look up the node from alloc_history.
+		peer := jobStopPeer
+		var node *model.Node
+		if peer == "" {
+			node, err = nodeForJob(ctx, db, id)
+			if err != nil {
+				return fmt.Errorf("lookup node for job %s: %w", id, err)
+			}
+			if node != nil {
+				peer = peerAddrForNode(node)
+			}
+		}
+
+		// Validate the job name before interpolation into the shell
+		// command (same injection guard as logs --job / stopAlloc).
+		jobName := job.Name
+		if !validSafeName(jobName) {
+			return fmt.Errorf("job stop: invalid job name %q (allowed: A-Z a-z 0-9 _ -)", jobName)
+		}
+
+		sshRan := false
+		if peer != "" {
+			transport, terr := jobStopTransportFromCtx()
+			if terr != nil {
+				return fmt.Errorf("job stop: ssh transport: %w", terr)
+			}
+			stopCtx, stopCancel := sshCmdCtx(ctx, jobStopTimeout)
+			defer stopCancel()
+			// Match the drift.go job restart unit pattern: orca-alloc-<name>.
+			// Use a glob (orca-alloc-<name>-*) to stop all task units in
+			// a multi-task allocation group.
+			unitPattern := fmt.Sprintf("orca-alloc-%s-*", jobName)
+			stopCmd := fmt.Sprintf("systemctl stop %s", shellQuote(unitPattern))
+			out, sErr := transport.Exec(stopCtx, peer, stopCmd)
+			if sErr != nil {
+				// Non-fatal: the unit may not be running (already
+				// stopped) or SSH may fail. We still update the DB
+				// status so the operator's intent is recorded.
+				fmt.Fprintf(cmd.ErrOrStderr(), "⚠ job stop: SSH systemctl stop failed on %s: %v (output: %s)\n", peer, sErr, strings.TrimSpace(string(out)))
+			} else {
+				sshRan = true
+			}
+		}
+
 		if err := repo.UpdateStatus(ctx, id, model.JobStatusStopped, 130); err != nil {
 			return err
 		}
@@ -327,9 +445,20 @@ var jobStopCmd = &cobra.Command{
 		// `orca job list` reflects the just-stopped job.
 		cacheInvalidate(cacheJobClass)
 		if jsonOutput {
-			return printJSON(map[string]any{"id": id, "status": "stopped", "previous_status": job.Status})
+			result := map[string]any{"id": id, "status": "stopped", "previous_status": job.Status}
+			if peer != "" {
+				result["peer"] = peer
+				result["ssh_stop"] = sshRan
+			}
+			return printJSON(result)
+		}
+		if peer != "" && sshRan {
+			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (systemctl stop on %s)\n", id, peer)
+		} else if peer != "" {
+			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (DB only; SSH stop failed — see stderr)\n", id)
+		} else {
+			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (DB only; no node found)\n", id)
 		}
-		fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s\n", id)
 		return nil
 	},
 }
@@ -383,6 +512,8 @@ var jobLogsCmd = &cobra.Command{
 
 func init() {
 	jobStopCmd.Flags().StringVar(&stopID, "id", "", "job id")
+	jobStopCmd.Flags().StringVar(&jobStopPeer, "peer", "", "peer address (host:port) running the allocation (auto-detected from alloc history if empty)")
+	jobStopCmd.Flags().DurationVar(&jobStopTimeout, "timeout", sshCmdDefaultTimeout, "SSH command timeout")
 	jobLogsCmd.Flags().StringVar(&stopID, "id", "", "job id")
 	jobRunCmd.Flags().StringVar(&runTarget, "target", "", "pin job to a specific node id (overrides bin-packing)")
 	jobRunCmd.Flags().StringVar(&runIDKey, "idempotency-key", "", "X-Orca-Idempotency-Key for cross-node dispatch dedupe")

internal/cli/job_test.go

@@ -2,11 +2,14 @@ package cli
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"git.cloudinit.dev/coreci/orca/internal/certpaths"
 	"git.cloudinit.dev/coreci/orca/internal/model"
@@ -310,3 +313,193 @@ func seedJob(t *testing.T, name string, status model.JobStatus) string {
 	}
 	return j.ID
 }
+
+
+// mockJobStopExec is a record-and-replay SSH execer for `orca job stop`
+// tests (same pattern as mockDrainExec / mockLogsExec).
+type mockJobStopExec struct {
+	mu        sync.Mutex
+	responses []jobStopMockResp
+	calls     []jobStopMockCall
+}
+
+type jobStopMockResp struct {
+	match string
+	out   string
+	exit  int
+}
+
+type jobStopMockCall struct {
+	peer string
+	cmd  string
+}
+
+func (m *mockJobStopExec) Exec(_ context.Context, peer, cmd string) ([]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.calls = append(m.calls, jobStopMockCall{peer: peer, cmd: cmd})
+	for _, r := range m.responses {
+		if r.match == "" || strings.Contains(cmd, r.match) {
+			return []byte(r.out), nil
+		}
+	}
+	return []byte(""), nil
+}
+
+func (m *mockJobStopExec) callsFor(match string) []jobStopMockCall {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var out []jobStopMockCall
+	for _, c := range m.calls {
+		if strings.Contains(c.cmd, match) {
+			out = append(out, c)
+		}
+	}
+	return out
+}
+
+// TestJobStopSSH verifies that `orca job stop` sends a real
+// 'systemctl stop' via SSH to the target node when the job has a
+// recorded allocation history (REQ-158, P09 T6).
+func TestJobStopSSH(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	if err := runInit(discardWriter{}); err != nil {
+		t.Fatalf("init: %v", err)
+	}
+
+	// Seed a node and a job, then record an alloc_history entry
+	// linking the job to the node.
+	nodeID := seedNode(t, "worker-1", "worker-1:8443")
+	jobID := seedJob(t, "webapp", model.JobStatusRunning)
+
+	db, err := store.Open(certpaths.DBPath())
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer db.Close()
+	hist := store.NewAllocHistoryRepo(db)
+	ctx := context.Background()
+	if err := hist.EnsureSchema(ctx); err != nil {
+		t.Fatalf("ensure schema: %v", err)
+	}
+	if err := hist.Record(ctx, store.AllocHistoryEntry{
+		AllocID:   "default/webapp-0",
+		JobID:     jobID,
+		NodeID:    nodeID,
+		Namespace: "default",
+		ToState:   "created",
+		Timestamp: time.Now().UTC(),
+	}); err != nil {
+		t.Fatalf("record alloc history: %v", err)
+	}
+
+	// Wire the mock SSH transport (must be after resetRootFlags so
+	// resetCommandFlags doesn't nil it out).
+	resetRootFlags(t)
+	mock := &mockJobStopExec{}
+	prev := jobStopTransportOverride
+	jobStopTransportOverride = mock
+	t.Cleanup(func() { jobStopTransportOverride = prev })
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"job", "stop", jobID})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("job stop: %v", err)
+	}
+
+	// Verify systemctl stop was called via SSH.
+	stopCalls := mock.callsFor("systemctl stop")
+	if len(stopCalls) == 0 {
+		t.Fatalf("expected systemctl stop SSH call, got %d calls: %v", len(mock.calls), mock.calls)
+	}
+	if !strings.Contains(stopCalls[0].cmd, "orca-alloc-webapp-*") {
+		t.Errorf("expected 'orca-alloc-webapp-*' in cmd, got: %s", stopCalls[0].cmd)
+	}
+	if !strings.Contains(stopCalls[0].peer, "worker-1") {
+		t.Errorf("expected peer to contain 'worker-1', got: %s", stopCalls[0].peer)
+	}
+
+	// Verify the DB status was updated.
+	repo := store.NewJobRepo(db)
+	job, err := repo.Get(ctx, jobID)
+	if err != nil {
+		t.Fatalf("get job: %v", err)
+	}
+	if job.Status != model.JobStatusStopped {
+		t.Errorf("job status = %v, want stopped", job.Status)
+	}
+}
+
+// TestJobStopSSHPeerOverride verifies that --peer bypasses the
+// alloc_history lookup and uses the given peer directly (REQ-158).
+func TestJobStopSSHPeerOverride(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	if err := runInit(discardWriter{}); err != nil {
+		t.Fatalf("init: %v", err)
+	}
+
+	jobID := seedJob(t, "webapp2", model.JobStatusRunning)
+
+	resetRootFlags(t)
+	mock := &mockJobStopExec{}
+	prev := jobStopTransportOverride
+	jobStopTransportOverride = mock
+	t.Cleanup(func() { jobStopTransportOverride = prev })
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"job", "stop", jobID, "--peer", "10.0.0.5:22"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("job stop: %v", err)
+	}
+
+	stopCalls := mock.callsFor("systemctl stop")
+	if len(stopCalls) == 0 {
+		t.Fatalf("expected systemctl stop SSH call, got %d calls", len(mock.calls))
+	}
+	if stopCalls[0].peer != "10.0.0.5:22" {
+		t.Errorf("peer = %s, want 10.0.0.5:22", stopCalls[0].peer)
+	}
+}
+
+// TestJobStopNoNodeFallback verifies that when no node is found in
+// alloc_history, the job is still stopped in the DB (soft stop
+// fallback) without attempting SSH (REQ-158).
+func TestJobStopNoNodeFallback(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+	if err := runInit(discardWriter{}); err != nil {
+		t.Fatalf("init: %v", err)
+	}
+
+	jobID := seedJob(t, "localjob", model.JobStatusRunning)
+
+	resetRootFlags(t)
+	mock := &mockJobStopExec{}
+	prev := jobStopTransportOverride
+	jobStopTransportOverride = mock
+	t.Cleanup(func() { jobStopTransportOverride = prev })
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"job", "stop", jobID})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("job stop: %v", err)
+	}
+
+	// No SSH calls should have been made (no node found).
+	if len(mock.calls) > 0 {
+		t.Errorf("expected 0 SSH calls, got %d: %v", len(mock.calls), mock.calls)
+	}
+
+	// Verify the output mentions "DB only".
+	if !strings.Contains(buf.String(), "DB only") {
+		t.Errorf("output should mention 'DB only', got: %s", buf.String())
+	}
+}

internal/cli/logs.go

@@ -101,8 +101,20 @@ var (
 	logsJob      string
 	logsSince    string
 	logsJSON     bool
+	logsLines    int
 )
 
+// logsMaxLines is the hard cap on --lines to prevent OOM from
+// unbounded journalctl output (REQ-158, P09 T3).
+const logsMaxLines = 50000
+
+// logsDefaultLines is the default --lines value.
+const logsDefaultLines = 1000
+
+// logsMaxSince is the maximum lookback for --since (7 days) to
+// prevent OOM from unbounded journalctl queries (REQ-158, P09 T3).
+const logsMaxSince = 7 * 24 * time.Hour
+
 var logsCmd = &cobra.Command{
 	Use:   "logs",
 	Short: "Aggregate journald logs across nodes (REQ-117)",
@@ -139,6 +151,25 @@ Ctrl-C cancels the fan-out via signal.NotifyContext.`,
 		if err != nil {
 			return err
 		}
+		// REQ-158 / P09 T3: clamp --since to 7 days max to prevent
+		// OOM from unbounded journalctl queries. If the requested
+		// lookback exceeds the cap, clamp it and warn.
+		now := time.Now().UTC()
+		maxSince := now.Add(-logsMaxSince)
+		if since.Before(maxSince) {
+			fmt.Fprintf(cmd.ErrOrStderr(), "⚠ --since %s exceeds 7d cap; clamping to 7d\n", logsSince)
+			since = maxSince
+		}
+
+		// REQ-158 / P09 T3: clamp --lines to [1, logsMaxLines].
+		lines := logsLines
+		if lines <= 0 {
+			lines = logsDefaultLines
+		}
+		if lines > logsMaxLines {
+			fmt.Fprintf(cmd.ErrOrStderr(), "⚠ --lines %d exceeds max %d; clamping\n", lines, logsMaxLines)
+			lines = logsMaxLines
+		}
 
 		ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
 		defer cancel()
@@ -158,7 +189,7 @@ Ctrl-C cancels the fan-out via signal.NotifyContext.`,
 
 		out := cmd.OutOrStdout()
 		multi := len(nodes) > 1
-		for line := range streamLogs(ctx, ex, nodes, since, logsJob) {
+		for line := range streamLogs(ctx, ex, nodes, since, logsJob, lines) {
 			if logsJSON {
 				raw, _ := json.Marshal(line)
 				fmt.Fprintln(out, string(raw))
@@ -227,7 +258,7 @@ func resolveLogNodes(ctx context.Context) ([]*model.Node, error) {
 // JSON entry immediately. The stream ends when every node has
 // completed (or the context is cancelled). The caller drives the
 // iteration via range-over-func (D-017 iter.Seq pattern).
-func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since time.Time, job string) iter.Seq[LogLine] {
+func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since time.Time, job string, lines int) iter.Seq[LogLine] {
 	return func(yield func(LogLine) bool) {
 		merged := make(chan LogLine)
 		var wg sync.WaitGroup
@@ -235,7 +266,7 @@ func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since t
 			wg.Add(1)
 			go func(n *model.Node) {
 				defer wg.Done()
-				streamNodeLines(ctx, ex, n, since, job, merged)
+				streamNodeLines(ctx, ex, n, since, job, lines, merged)
 			}(n)
 		}
 		done := make(chan struct{})
@@ -267,7 +298,7 @@ func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since t
 // context is cancelled); the caller is responsible for waiting on the
 // goroutine. Send is non-blocking via select on ctx.Done so a slow
 // consumer does not stall the fanout forever.
-func streamNodeLines(ctx context.Context, ex logsExecer, n *model.Node, since time.Time, job string, out chan<- LogLine) {
+func streamNodeLines(ctx context.Context, ex logsExecer, n *model.Node, since time.Time, job string, lines int, out chan<- LogLine) {
 	peer := peerAddrForNode(n)
 	if peer == "" {
 		slog.Default().Warn("logs: cannot resolve SSH address for node", "node", n.Name)
@@ -278,9 +309,14 @@ func streamNodeLines(ctx context.Context, ex logsExecer, n *model.Node, since ti
 		unitPattern = "orca-alloc-" + job + "-*"
 	}
 	sinceStr := since.Format("2006-01-02 15:04:05")
+	// REQ-158 / P09 T3: pass --lines=N to journalctl to cap output
+	// and prevent OOM from unbounded log queries.
+	if lines <= 0 {
+		lines = logsDefaultLines
+	}
 	// F1: shellQuote (single-quote wrap) instead of %q — %q does not
 	// escape backticks, enabling command substitution in double quotes.
-	cmd := fmt.Sprintf("journalctl -u %s --since %s --output json --no-pager", shellQuote(unitPattern), shellQuote(sinceStr))
+	cmd := fmt.Sprintf("journalctl -u %s --since %s --lines %d --output json --no-pager", shellQuote(unitPattern), shellQuote(sinceStr), lines)
 	raw, err := ex.Exec(ctx, peer, cmd)
 	if err != nil {
 		slog.Default().Warn("logs: exec failed", "node", n.Name, "peer", peer, "error", err)
@@ -324,7 +360,8 @@ func init() {
 	logsCmd.Flags().BoolVar(&logsAllNodes, "all-nodes", false, "fan out to all registered nodes")
 	logsCmd.Flags().StringVar(&logsNode, "node", "", "restrict to a single node (name or id)")
 	logsCmd.Flags().StringVar(&logsJob, "job", "", "filter by job name (matches orca-alloc-<name>-* units)")
-	logsCmd.Flags().StringVar(&logsSince, "since", "5m", "duration lookback (e.g. 5m, 1h, 30m); default 5m")
+	logsCmd.Flags().StringVar(&logsSince, "since", "5m", "duration lookback (e.g. 5m, 1h, 30m); default 5m; max 7d")
+	logsCmd.Flags().IntVar(&logsLines, "lines", logsDefaultLines, fmt.Sprintf("max number of journal lines per node (default %d, max %d)", logsDefaultLines, logsMaxLines))
 	logsCmd.Flags().BoolVar(&logsJSON, "json", false, "output raw JSON (one LogLine per line)")
 	rootCmd.AddCommand(logsCmd)
 }

internal/cli/logs_test.go

@@ -63,6 +63,18 @@ func (m *mockLogsExec) countCalls(match string) int {
 	return n
 }
 
+func (m *mockLogsExec) callsFor(match string) []logsMockCall {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var out []logsMockCall
+	for _, c := range m.calls {
+		if strings.Contains(c.cmd, match) {
+			out = append(out, c)
+		}
+	}
+	return out
+}
+
 // logsTestEnv wires a mockLogsExec into logsExecOverride and returns
 // the mock + a cleanup func. Tests MUST defer the cleanup.
 func logsTestEnv(t *testing.T) *mockLogsExec {
@@ -325,7 +337,7 @@ func TestLogsCancelStopsStream(t *testing.T) {
 		{ID: "n1", Name: "cancelnode", Address: "cancelnode:8443"},
 	}
 	consumed := 0
-	for range streamLogs(ctx, ex, nodes, time.Now().UTC().Add(-1*time.Minute), "") {
+	for range streamLogs(ctx, ex, nodes, time.Now().UTC().Add(-1*time.Minute), "", 1000) {
 		consumed++
 	}
 	if consumed > 1 {
@@ -357,3 +369,146 @@ func TestLogsParseJournalLine_InvalidJSON(t *testing.T) {
 		t.Error("expected error for invalid json, got nil")
 	}
 }
+
+
+// TestLogsLinesFlag verifies that --lines is passed through to the
+// journalctl command as --lines=N (REQ-158, P09 T8).
+func TestLogsLinesFlag(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+
+	logsNodeForTest(t, "linesnode", "linesnode:8443")
+	mx := logsTestEnv(t)
+
+	ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+	mx.responses = []logsMockResp{
+		{match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "line test", "6") + "\n"},
+	}
+
+	_, err := runLogsCmd(t, []string{"logs", "--node", "linesnode", "--since", "1m", "--lines", "500"})
+	if err != nil {
+		t.Fatalf("logs: %v", err)
+	}
+	calls := mx.callsFor("journalctl")
+	if len(calls) == 0 {
+		t.Fatal("expected journalctl call")
+	}
+	if !strings.Contains(calls[0].cmd, "--lines 500") {
+		t.Errorf("expected '--lines 500' in cmd, got: %s", calls[0].cmd)
+	}
+}
+
+// TestLogsLinesDefault verifies that the default --lines value (1000)
+// is passed to journalctl when --lines is not specified (REQ-158).
+func TestLogsLinesDefault(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+
+	logsNodeForTest(t, "defnode", "defnode:8443")
+	mx := logsTestEnv(t)
+
+	ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+	mx.responses = []logsMockResp{
+		{match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "default lines", "6") + "\n"},
+	}
+
+	_, err := runLogsCmd(t, []string{"logs", "--node", "defnode", "--since", "1m"})
+	if err != nil {
+		t.Fatalf("logs: %v", err)
+	}
+	calls := mx.callsFor("journalctl")
+	if len(calls) == 0 {
+		t.Fatal("expected journalctl call")
+	}
+	if !strings.Contains(calls[0].cmd, "--lines 1000") {
+		t.Errorf("expected default '--lines 1000' in cmd, got: %s", calls[0].cmd)
+	}
+}
+
+// TestLogsLinesClamp verifies that --lines exceeding the max (50000) is
+// clamped (REQ-158, P09 T8).
+func TestLogsLinesClamp(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+
+	logsNodeForTest(t, "clampnode", "clampnode:8443")
+	mx := logsTestEnv(t)
+
+	ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+	mx.responses = []logsMockResp{
+		{match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "clamp test", "6") + "\n"},
+	}
+
+	_, err := runLogsCmd(t, []string{"logs", "--node", "clampnode", "--since", "1m", "--lines", "999999"})
+	if err != nil {
+		t.Fatalf("logs: %v", err)
+	}
+	calls := mx.callsFor("journalctl")
+	if len(calls) == 0 {
+		t.Fatal("expected journalctl call")
+	}
+	if !strings.Contains(calls[0].cmd, "--lines 50000") {
+		t.Errorf("expected clamped '--lines 50000' in cmd, got: %s", calls[0].cmd)
+	}
+}
+
+// TestLogsSinceClamp verifies that --since exceeding 7 days is
+// clamped and a warning is printed (REQ-158, P09 T8).
+func TestLogsSinceClamp(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+
+	logsNodeForTest(t, "sincenode", "sincenode:8443")
+	mx := logsTestEnv(t)
+
+	ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+	mx.responses = []logsMockResp{
+		{match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "since test", "6") + "\n"},
+	}
+
+	var buf bytes.Buffer
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	resetRootFlags(t)
+	rootCmd.SetOut(&buf)
+	rootCmd.SetErr(&buf)
+	rootCmd.SetArgs([]string{"logs", "--node", "sincenode", "--since", "720h"})
+	err := rootCmd.Execute()
+	if err != nil {
+		t.Fatalf("logs: %v", err)
+	}
+	out := buf.String()
+	if !strings.Contains(out, "clamping to 7d") {
+		t.Errorf("expected warning about clamping --since to 7d, got: %s", out)
+	}
+	calls := mx.callsFor("journalctl")
+	if len(calls) == 0 {
+		t.Fatal("expected journalctl call")
+	}
+}
+
+// TestLogsLinesFlagJSON verifies --lines is passed through in JSON mode.
+func TestLogsLinesFlagJSON(t *testing.T) {
+	_, cleanup := initTestEnv(t)
+	defer cleanup()
+
+	logsNodeForTest(t, "jsonlines", "jsonlines:8443")
+	mx := logsTestEnv(t)
+
+	ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+	mx.responses = []logsMockResp{
+		{match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "json lines test", "6") + "\n"},
+	}
+
+	_, err := runLogsCmd(t, []string{"logs", "--node", "jsonlines", "--since", "1m", "--lines", "200", "--json"})
+	if err != nil {
+		t.Fatalf("logs: %v", err)
+	}
+	calls := mx.callsFor("journalctl")
+	if len(calls) == 0 {
+		t.Fatal("expected journalctl call")
+	}
+	if !strings.Contains(calls[0].cmd, "--lines 200") {
+		t.Errorf("expected '--lines 200' in cmd, got: %s", calls[0].cmd)
+	}
+}

internal/cli/namespace_test.go

@@ -57,6 +57,11 @@ func resetCommandFlags() {
 	driftConfigPath = ""
 	driftRemediateForce = false
 	jobRestartPeer = ""
+	jobStopPeer = ""
+	jobStopTimeout = 0
+	jobStopTransportOverride = nil
+	logsLines = logsDefaultLines
+	cutoverFSOverride = nil
 	jobLintExplain = false
 	jobLintFormat = "text"
 	jobVerifyLead = ""

internal/cli/upgrade.go

@@ -69,6 +69,41 @@ var upgradeTransportOverride upgradeTransport
 // peers to create the orca user on. Returns a list of peer addresses.
 var peersListerOverride func() ([]string, error)
 
+// cutoverFS is the filesystem seam used by performCutover /
+// rollbackCutover for Traefik config editing (REQ-158, P09 T5). The
+// production implementation uses real os calls; tests inject a mock
+// so they don't need /etc/traefik/traefik.yml to exist.
+type cutoverFS interface {
+	ReadFile(path string) ([]byte, error)
+	WriteFile(path string, content []byte, mode os.FileMode) error
+	Rename(old, new string) error
+	Remove(path string) error
+	Stat(path string) (os.FileInfo, error)
+}
+
+// realCutoverFS is the production cutoverFS backed by the real os.
+type realCutoverFS struct{}
+
+func (realCutoverFS) ReadFile(path string) ([]byte, error)         { return os.ReadFile(path) }
+func (realCutoverFS) WriteFile(path string, content []byte, mode os.FileMode) error {
+	return os.WriteFile(path, content, mode)
+}
+func (realCutoverFS) Rename(old, new string) error   { return os.Rename(old, new) }
+func (realCutoverFS) Remove(path string) error        { return os.Remove(path) }
+func (realCutoverFS) Stat(path string) (os.FileInfo, error) { return os.Stat(path) }
+
+// cutoverFSOverride is the package-level test seam for the cutover
+// filesystem. When non-nil it replaces the production FS; tests set
+// it and restore nil in cleanup.
+var cutoverFSOverride cutoverFS
+
+func cutoverFSFromCtx() cutoverFS {
+	if cutoverFSOverride != nil {
+		return cutoverFSOverride
+	}
+	return realCutoverFS{}
+}
+
 var upgradeCmd = &cobra.Command{
 	Use:   "upgrade",
 	Short: "Upgrade orca to a new version (REQ-115, R-017 cutover)",
@@ -310,11 +345,43 @@ func detectOldTraefikBinding() bool {
 // return 200. On failure, rolls back (restores :443, removes nft rules)
 // and returns (false, nil). On success returns (true, nil). With
 // force=true, verification is skipped.
+//
+// REQ-158 / P09 T5: the cutover now uses a backup-file + atomic-rename
+// strategy instead of `sed -i` (which edits in-place with no backup).
+// The Traefik config is copied to traefik.yml.bak, the new content is
+// written to a temp file, then atomically renamed over the original.
+// If any step fails, the backup is restored. This prevents a partial
+// edit from leaving Traefik in a broken state.
 func performCutover(ctx context.Context, runner commandRunner, out interface{ Write([]byte) (int, error) }, force bool) (bool, error) {
-	if _, err := runner.Run(ctx, "sed", "-i", "s/:443/127.0.0.1:8443/g", "/etc/traefik/traefik.yml"); err != nil {
-		return false, fmt.Errorf("cutover: edit traefik.yml: %w", err)
+	cfs := cutoverFSFromCtx()
+	traefikYml := "/etc/traefik/traefik.yml"
+	backupPath := traefikYml + ".bak"
+
+	// Step 1: read the current config and create a backup.
+	original, err := cfs.ReadFile(traefikYml)
+	if err != nil {
+		return false, fmt.Errorf("cutover: read traefik.yml: %w", err)
 	}
+	if err := cfs.WriteFile(backupPath, original, 0o644); err != nil {
+		return false, fmt.Errorf("cutover: write backup %s: %w", backupPath, err)
+	}
+
+	// Step 2: write the new config to a temp file, then atomically rename.
+	newContent := strings.ReplaceAll(string(original), ":443", "127.0.0.1:8443")
+	tmpPath := traefikYml + ".tmp"
+	if err := cfs.WriteFile(tmpPath, []byte(newContent), 0o644); err != nil {
+		return false, fmt.Errorf("cutover: write temp %s: %w", tmpPath, err)
+	}
+	if err := cfs.Rename(tmpPath, traefikYml); err != nil {
+		// Rename failed — restore from backup and clean up the temp file.
+		_ = cfs.Remove(tmpPath)
+		_ = cfs.Rename(backupPath, traefikYml)
+		return false, fmt.Errorf("cutover: atomic rename %s → %s: %w", tmpPath, traefikYml, err)
+	}
+
 	if _, err := runner.Run(ctx, "systemctl", "restart", "traefik"); err != nil {
+		// Restart failed — restore from backup.
+		_ = cfs.Rename(backupPath, traefikYml)
 		return false, fmt.Errorf("cutover: restart traefik: %w", err)
 	}
 	nftCmd := `nft add table inet orca_redirect; nft 'add chain inet orca_redirect prerouting { type nat hook prerouting priority -100; }'; nft add rule inet orca_redirect prerouting tcp dport 443 dnat to 127.0.0.1:8443`
@@ -324,6 +391,8 @@ func performCutover(ctx context.Context, runner commandRunner, out interface{ Wr
 
 	if force {
 		fmt.Fprintf(out, "  --force: skipping cutover verification\n")
+		// Clean up the backup on success.
+		_ = cfs.Remove(backupPath)
 		return true, nil
 	}
 
@@ -336,6 +405,8 @@ func performCutover(ctx context.Context, runner commandRunner, out interface{ Wr
 		return false, nil
 	}
 	fmt.Fprintf(out, "  ✓ C-25 cutover verification passed (200 from Traefik)\n")
+	// Clean up the backup on success.
+	_ = cfs.Remove(backupPath)
 	return true, nil
 }
 
@@ -390,9 +461,33 @@ func verifyCutover(out interface{ Write([]byte) (int, error) }) error {
 }
 
 // rollbackCutover restores Traefik to :443 and removes nftables rules.
+// REQ-158 / P09 T5: restore from the backup file (traefik.yml.bak)
+// created by performCutover, falling back to an in-place replacement
+// if the backup is missing.
 func rollbackCutover(ctx context.Context, runner commandRunner) error {
-	if _, err := runner.Run(ctx, "sed", "-i", "s/127.0.0.1:8443/:443/g", "/etc/traefik/traefik.yml"); err != nil {
-		return fmt.Errorf("rollback: edit traefik.yml: %w", err)
+	cfs := cutoverFSFromCtx()
+	traefikYml := "/etc/traefik/traefik.yml"
+	backupPath := traefikYml + ".bak"
+	// Try restoring from the backup first.
+	if _, err := cfs.Stat(backupPath); err == nil {
+		if err := cfs.Rename(backupPath, traefikYml); err != nil {
+			return fmt.Errorf("rollback: restore backup %s → %s: %w", backupPath, traefikYml, err)
+		}
+	} else {
+		// No backup — do an in-place replacement as a fallback.
+		current, rErr := cfs.ReadFile(traefikYml)
+		if rErr != nil {
+			return fmt.Errorf("rollback: read traefik.yml: %w", rErr)
+		}
+		restored := strings.ReplaceAll(string(current), "127.0.0.1:8443", ":443")
+		tmpPath := traefikYml + ".tmp"
+		if err := cfs.WriteFile(tmpPath, []byte(restored), 0o644); err != nil {
+			return fmt.Errorf("rollback: write temp %s: %w", tmpPath, err)
+		}
+		if err := cfs.Rename(tmpPath, traefikYml); err != nil {
+			_ = cfs.Remove(tmpPath)
+			return fmt.Errorf("rollback: atomic rename: %w", err)
+		}
 	}
 	if _, err := runner.Run(ctx, "systemctl", "restart", "traefik"); err != nil {
 		return fmt.Errorf("rollback: restart traefik: %w", err)

internal/cli/upgrade_test.go

@@ -8,6 +8,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"git.cloudinit.dev/coreci/orca/internal/migration"
 	"git.cloudinit.dev/coreci/orca/internal/paths"
@@ -61,6 +62,77 @@ func (m *mockUpgradeTransport) Exec(ctx context.Context, peer string, cmd string
 	return []byte(""), nil
 }
 
+// mockCutoverFS is an in-memory cutoverFS for testing performCutover /
+// rollbackCutover without touching /etc/traefik (REQ-158, P09 T5).
+type mockCutoverFS struct {
+	files map[string][]byte
+	errs  map[string]error // keyed by operation: "read:<path>", "write:<path>", "rename:<old>", "stat:<path>"
+}
+
+func newMockCutoverFS() *mockCutoverFS {
+	return &mockCutoverFS{
+		files: make(map[string][]byte),
+		errs:  make(map[string]error),
+	}
+}
+
+func (m *mockCutoverFS) ReadFile(path string) ([]byte, error) {
+	if err, ok := m.errs["read:"+path]; ok {
+		return nil, err
+	}
+	if data, ok := m.files[path]; ok {
+		return data, nil
+	}
+	return nil, fmt.Errorf("mock: %s not found", path)
+}
+
+func (m *mockCutoverFS) WriteFile(path string, content []byte, mode os.FileMode) error {
+	if err, ok := m.errs["write:"+path]; ok {
+		return err
+	}
+	cp := make([]byte, len(content))
+	copy(cp, content)
+	m.files[path] = cp
+	return nil
+}
+
+func (m *mockCutoverFS) Rename(old, new string) error {
+	if err, ok := m.errs["rename:"+old]; ok {
+		return err
+	}
+	data, ok := m.files[old]
+	if !ok {
+		return fmt.Errorf("mock: rename source %s not found", old)
+	}
+	m.files[new] = data
+	delete(m.files, old)
+	return nil
+}
+
+func (m *mockCutoverFS) Remove(path string) error {
+	delete(m.files, path)
+	return nil
+}
+
+func (m *mockCutoverFS) Stat(path string) (os.FileInfo, error) {
+	if err, ok := m.errs["stat:"+path]; ok {
+		return nil, err
+	}
+	if _, ok := m.files[path]; ok {
+		return mockFileInfo{name: path}, nil
+	}
+	return nil, fmt.Errorf("mock: %s not found", path)
+}
+
+type mockFileInfo struct{ name string }
+
+func (m mockFileInfo) Name() string       { return m.name }
+func (m mockFileInfo) Size() int64        { return 0 }
+func (m mockFileInfo) Mode() os.FileMode   { return 0o644 }
+func (m mockFileInfo) ModTime() time.Time  { return time.Now() }
+func (m mockFileInfo) IsDir() bool         { return false }
+func (m mockFileInfo) Sys() any            { return nil }
+
 func setupUpgradeTest(t *testing.T) {
 	t.Helper()
 	t.Setenv("ORCA_HOME", t.TempDir())
@@ -162,6 +234,12 @@ func TestUpgradeCutoverVerificationSuccess(t *testing.T) {
 	upgradeRunnerOverride = runner
 	httpClientOverride = func(url string) (int, error) { return 200, nil }
 
+	// Provide a mock Traefik config so performCutover can read it.
+	cfs := newMockCutoverFS()
+	cfs.files["/etc/traefik/traefik.yml"] = []byte("entrypoint: :443\n")
+	cutoverFSOverride = cfs
+	t.Cleanup(func() { cutoverFSOverride = nil })
+
 	rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0", "--force"})
 	if err := rootCmd.Execute(); err != nil {
 		t.Fatalf("upgrade with cutover: %v", err)
@@ -180,6 +258,12 @@ func TestUpgradeCutoverRollback(t *testing.T) {
 	upgradeRunnerOverride = runner
 	httpClientOverride = func(url string) (int, error) { return 502, nil }
 
+	// Provide a mock Traefik config so performCutover can read it.
+	cfs := newMockCutoverFS()
+	cfs.files["/etc/traefik/traefik.yml"] = []byte("entrypoint: :443\n")
+	cutoverFSOverride = cfs
+	t.Cleanup(func() { cutoverFSOverride = nil })
+
 	rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0"})
 	err := rootCmd.Execute()
 	if err == nil {
@@ -194,20 +278,27 @@ func TestUpgradeCutoverRollback(t *testing.T) {
 		t.Errorf("output should mention rollback: %s", out)
 	}
 
+	// Verify rollback: the traefik.yml content should be restored to
+	// :443 (the backup was renamed back over the modified file).
+	restored, ok := cfs.files["/etc/traefik/traefik.yml"]
+	if !ok {
+		t.Fatal("rollback: traefik.yml missing after rollback")
+	}
+	if !strings.Contains(string(restored), ":443") {
+		t.Errorf("rollback: traefik.yml not restored to :443, got: %s", string(restored))
+	}
+	if strings.Contains(string(restored), "127.0.0.1:8443") {
+		t.Errorf("rollback: traefik.yml still has 127.0.0.1:8443 after rollback: %s", string(restored))
+	}
+
 	foundRollback := false
 	for _, call := range runner.calls {
-		if call.name == "sed" && len(call.args) >= 2 {
-			joined := strings.Join(call.args, " ")
-			if strings.Contains(joined, "127.0.0.1:8443") && strings.Contains(joined, ":443") {
-				foundRollback = true
-			}
-		}
 		if call.name == "nft" && len(call.args) >= 2 && call.args[0] == "delete" {
 			foundRollback = true
 		}
 	}
 	if !foundRollback {
-		t.Errorf("rollback commands not detected (calls: %v)", runner.calls)
+		t.Errorf("rollback nft delete command not detected (calls: %v)", runner.calls)
 	}
 }
 
@@ -227,6 +318,12 @@ func TestUpgradeCutoverForceSkipsVerification(t *testing.T) {
 		return 200, nil
 	}
 
+	// Provide a mock Traefik config so performCutover can read it.
+	cfs := newMockCutoverFS()
+	cfs.files["/etc/traefik/traefik.yml"] = []byte("entrypoint: :443\n")
+	cutoverFSOverride = cfs
+	t.Cleanup(func() { cutoverFSOverride = nil })
+
 	rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0", "--force"})
 	if err := rootCmd.Execute(); err != nil {
 		t.Fatalf("upgrade with --force: %v", err)
@@ -339,3 +436,211 @@ func TestUpgradeFullMigration(t *testing.T) {
 		t.Errorf("install.sh was not invoked (calls: %v)", runner.calls)
 	}
 }
+
+
+// TestCutoverBackupRestoreOnFailure verifies that when the cutover
+// verification fails, the Traefik config is restored from the backup
+// file (REQ-158, P09 T10). This is a unit-level test that calls
+// performCutover directly with a mock FS.
+func TestCutoverBackupRestoreOnFailure(t *testing.T) {
+	// Set up a mock FS with a Traefik config containing :443.
+	cfs := newMockCutoverFS()
+	original := []byte("entrypoint:\n  - :443\n")
+	cfs.files["/etc/traefik/traefik.yml"] = original
+	cutoverFSOverride = cfs
+	t.Cleanup(func() { cutoverFSOverride = nil })
+
+	// Mock runner that succeeds for systemctl restart.
+	runner := &mockUpgradeRunner{
+		outputs: make(map[string][]byte),
+	}
+	// Mock HTTP check returns 502 (failure).
+	prevHTTP := httpClientOverride
+	httpClientOverride = func(url string) (int, error) { return 502, nil }
+	t.Cleanup(func() { httpClientOverride = prevHTTP })
+
+	var buf bytes.Buffer
+	ok, err := performCutover(context.Background(), runner, &buf, false)
+	if err != nil {
+		t.Fatalf("performCutover: %v", err)
+	}
+	if ok {
+		t.Fatal("expected cutover to fail (ok=false)")
+	}
+
+	// Verify the Traefik config was restored from backup.
+	restored, exists := cfs.files["/etc/traefik/traefik.yml"]
+	if !exists {
+		t.Fatal("traefik.yml missing after rollback")
+	}
+	if string(restored) != string(original) {
+		t.Errorf("traefik.yml not restored to original, got: %s", string(restored))
+	}
+	// Verify 127.0.0.1:8443 is NOT in the restored file.
+	if strings.Contains(string(restored), "127.0.0.1:8443") {
+		t.Errorf("traefik.yml still has 127.0.0.1:8443 after rollback: %s", string(restored))
+	}
+	// The backup file should have been consumed by rollbackCutover's rename.
+	if _, bakExists := cfs.files["/etc/traefik/traefik.yml.bak"]; bakExists {
+		t.Error("backup file still exists after rollback (should have been renamed)")
+	}
+}
+
+// TestCutoverAtomicRenameSuccess verifies that the cutover writes the
+// new config via atomic rename (temp file → original) and cleans up
+// the backup on success (REQ-158, P09 T10).
+func TestCutoverAtomicRenameSuccess(t *testing.T) {
+	cfs := newMockCutoverFS()
+	original := []byte("entrypoint:\n  - :443\n")
+	cfs.files["/etc/traefik/traefik.yml"] = original
+	cutoverFSOverride = cfs
+	t.Cleanup(func() { cutoverFSOverride = nil })
+
+	runner := &mockUpgradeRunner{
+		outputs: make(map[string][]byte),
+	}
+	prevHTTP := httpClientOverride
+	httpClientOverride = func(url string) (int, error) { return 200, nil }
+	t.Cleanup(func() { httpClientOverride = prevHTTP })
+
+	var buf bytes.Buffer
+	ok, err := performCutover(context.Background(), runner, &buf, false)
+	if err != nil {
+		t.Fatalf("performCutover: %v", err)
+	}
+	if !ok {
+		t.Fatal("expected cutover to succeed (ok=true)")
+	}
+
+	// Verify the config was updated to 127.0.0.1:8443.
+	updated, exists := cfs.files["/etc/traefik/traefik.yml"]
+	if !exists {
+		t.Fatal("traefik.yml missing after cutover")
+	}
+	if !strings.Contains(string(updated), "127.0.0.1:8443") {
+		t.Errorf("traefik.yml should have 127.0.0.1:8443, got: %s", string(updated))
+	}
+	if strings.Contains(string(updated), ":443\n") && !strings.Contains(string(updated), "127.0.0.1:8443") {
+		t.Errorf("traefik.yml should not have bare :443 anymore, got: %s", string(updated))
+	}
+	// The temp file should not exist.
+	if _, tmpExists := cfs.files["/etc/traefik/traefik.yml.tmp"]; tmpExists {
+		t.Error("temp file still exists after atomic rename")
+	}
+	// The backup should have been cleaned up on success.
+	if _, bakExists := cfs.files["/etc/traefik/traefik.yml.bak"]; bakExists {
+		t.Error("backup file still exists after successful cutover (should be cleaned up)")
+	}
+}
+
+// TestCutoverBackupCreated verifies that a backup file is created
+// before the cutover edits the config (REQ-158, P09 T10). Uses a
+// custom mock FS that records the sequence of operations so we can
+// assert the backup was written before the temp file.
+func TestCutoverBackupCreated(t *testing.T) {
+	// Use a recording mock FS that fails on the rename step so the
+	// backup write is observable before the rollback consumes it.
+	cfs := newMockCutoverFS()
+	original := []byte("entrypoint:\n  - :443\n")
+	cfs.files["/etc/traefik/traefik.yml"] = original
+	// Track write order via a custom FS that records operations.
+	var writeOrder []string
+	recordingCFS := &recordingCutoverFS{
+		inner:     cfs,
+		writeOrder: &writeOrder,
+	}
+	// Make the rename of the temp file fail so the cutover aborts.
+	cfs.errs["rename:/etc/traefik/traefik.yml.tmp"] = fmt.Errorf("rename failed")
+	cutoverFSOverride = recordingCFS
+	t.Cleanup(func() { cutoverFSOverride = nil })
+
+	runner := &mockUpgradeRunner{
+		outputs: make(map[string][]byte),
+	}
+
+	var buf bytes.Buffer
+	_, err := performCutover(context.Background(), runner, &buf, false)
+	if err == nil {
+		t.Fatal("expected error from failed rename")
+	}
+
+	// Verify the backup was written BEFORE the temp file.
+	// writeOrder records WriteFile calls in order.
+	bakIdx := -1
+	tmpIdx := -1
+	for i, p := range writeOrder {
+		if p == "/etc/traefik/traefik.yml.bak" {
+			bakIdx = i
+		}
+		if p == "/etc/traefik/traefik.yml.tmp" {
+			tmpIdx = i
+		}
+	}
+	if bakIdx == -1 {
+		t.Fatal("backup file was not written before cutover")
+	}
+	if tmpIdx == -1 {
+		t.Fatal("temp file was not written")
+	}
+	if bakIdx > tmpIdx {
+		t.Errorf("backup written after temp file (bakIdx=%d, tmpIdx=%d) — backup should come first", bakIdx, tmpIdx)
+	}
+	// The original should have been restored from backup on failure.
+	restored, exists := cfs.files["/etc/traefik/traefik.yml"]
+	if !exists {
+		t.Fatal("traefik.yml missing after failed rename + restore")
+	}
+	if string(restored) != string(original) {
+		t.Errorf("traefik.yml not restored to original after failed rename, got: %s", string(restored))
+	}
+}
+
+// recordingCutoverFS wraps a cutoverFS and records WriteFile call
+// paths so tests can assert the order of operations (REQ-158, P09 T10).
+type recordingCutoverFS struct {
+	inner      cutoverFS
+	writeOrder *[]string
+}
+
+func (r *recordingCutoverFS) ReadFile(path string) ([]byte, error) {
+	return r.inner.ReadFile(path)
+}
+func (r *recordingCutoverFS) WriteFile(path string, content []byte, mode os.FileMode) error {
+	*r.writeOrder = append(*r.writeOrder, path)
+	return r.inner.WriteFile(path, content, mode)
+}
+func (r *recordingCutoverFS) Rename(old, new string) error {
+	return r.inner.Rename(old, new)
+}
+func (r *recordingCutoverFS) Remove(path string) error {
+	return r.inner.Remove(path)
+}
+func (r *recordingCutoverFS) Stat(path string) (os.FileInfo, error) {
+	return r.inner.Stat(path)
+}
+
+// TestCutoverNoSedDirectly verifies that the cutover does NOT use
+// `sed -i` (the old unsafe approach). The mock runner records all
+// calls; none should be `sed` (REQ-158, P09 T5).
+func TestCutoverNoSedDirectly(t *testing.T) {
+	cfs := newMockCutoverFS()
+	cfs.files["/etc/traefik/traefik.yml"] = []byte("entrypoint:\n  - :443\n")
+	cutoverFSOverride = cfs
+	t.Cleanup(func() { cutoverFSOverride = nil })
+
+	runner := &mockUpgradeRunner{
+		outputs: make(map[string][]byte),
+	}
+	prevHTTP := httpClientOverride
+	httpClientOverride = func(url string) (int, error) { return 200, nil }
+	t.Cleanup(func() { httpClientOverride = prevHTTP })
+
+	var buf bytes.Buffer
+	_, _ = performCutover(context.Background(), runner, &buf, false)
+
+	for _, call := range runner.calls {
+		if call.name == "sed" {
+			t.Errorf("cutover should not use 'sed' (uses atomic rename now), found call: %s %v", call.name, call.args)
+		}
+	}
+}