feat: migration framework (#127)

README.md

@@ -1,6 +1,6 @@
 # Spinifex: An Open Source AWS-Compatible Stack for Bare-Metal, Edge, and On-Prem Deployments
 
-(Developer Preview) - Spinifex developed by [Mulga Defense Corporation](https://mulgadc.com/) is an open source infrastructure platform that brings the core services of AWS—like EC2, VPC, EBS, and S3—to environments where running in the cloud isn't an option. Whether you're deploying to edge sites, private data-centers, or need to operate in low-connectivity or highly contested environments, Spinifex gives you AWS-style workflows on your own hardware.
+Spinifex developed by [Mulga Defense Corporation](https://mulgadc.com) is an open source infrastructure platform that brings the core services of AWS—like EC2, VPC, EBS, and S3—to environments where running in the cloud isn't an option. Whether you're deploying to edge sites, private data-centers, or need to operate in low-connectivity or highly contested environments, Spinifex gives you AWS-style workflows on your own hardware.
 
 ## What is Spinifex?
 
@@ -61,11 +61,11 @@ Installation requires an Ubuntu / Debian system. See the detailed documentation
 The installation is straightforward to set up and running on a single node for testing purposes.
 
 ```bash
-curl https://install.mulgadc.com | bash
+curl -fsSL https://install.mulgadc.com | bash
 
 sudo /usr/local/share/spinifex/setup-ovn.sh --management
 
-sudo spx admin init
+sudo spx admin init --node node1 --nodes 1
 
 sudo systemctl start spinifex.target
 
@@ -76,7 +76,7 @@ aws ec2 describe-instance-types
 
 ### Development Setup
 
-For a complete development environment see the [Source Install](https://docs.mulgadc.com/docs/source-install/) documentation
+For a complete development environment see the [Source Install](https://docs.mulgadc.com/docs/install-source) documentation
 
 ### Component Repositories
 

cmd/spinifex/cmd/admin.go

@@ -217,6 +217,9 @@ func init() {
 
 	adminCmd.AddCommand(certCmd)
 	certCmd.AddCommand(certRenewCmd)
+
+	adminCmd.AddCommand(upgradeCmd)
+	upgradeCmd.Flags().Bool("yes", false, "Apply migrations without prompting")
 	certRenewCmd.Flags().StringSlice("extra-ip", nil, "Additional IP addresses to include in SANs")
 	certRenewCmd.Flags().StringSlice("extra-dns", nil, "Additional DNS names to include in SANs")
 	accountCreateCmd.Flags().String("name", "", "Account name (required)")

cmd/spinifex/cmd/admin_upgrade.go

@@ -0,0 +1,92 @@
+package cmd
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/mulgadc/spinifex/spinifex/migrate"
+	"github.com/spf13/cobra"
+)
+
+var upgradeCmd = &cobra.Command{
+	Use:   "upgrade",
+	Short: "Apply pending config migrations",
+	Long: `Apply pending configuration file migrations for upgrades between Spinifex versions.
+
+When run interactively (without --yes), shows pending migrations and prompts
+for confirmation. When called from setup.sh with --yes, applies immediately.
+
+Operators can skip migrations during install by setting INSTALL_SPINIFEX_SKIP_MIGRATE=1,
+then run 'spx admin upgrade' manually to review and apply.`,
+	Run: runAdminUpgrade,
+}
+
+func runAdminUpgrade(cmd *cobra.Command, _ []string) {
+	configDir, _ := cmd.Root().Flags().GetString("config-dir")
+	dataDir, _ := cmd.Root().Flags().GetString("spinifex-dir")
+	yes, _ := cmd.Flags().GetBool("yes")
+
+	// Check that the installation exists.
+	spinifexToml := configDir + "/spinifex.toml"
+	if _, err := os.Stat(spinifexToml); os.IsNotExist(err) {
+		fmt.Fprintf(os.Stderr, "No Spinifex installation found at %s\nRun 'spx admin init' first.\n", configDir)
+		os.Exit(1)
+	}
+
+	// Show current versions.
+	versions, err := migrate.DefaultRegistry.ConfigVersions(configDir)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error reading config versions: %v\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Println("Reading current config versions...")
+	for name, v := range versions {
+		if v == 0 {
+			fmt.Printf("  %-20s %d (no version marker)\n", name+":", v)
+		} else {
+			fmt.Printf("  %-20s %d\n", name+":", v)
+		}
+	}
+	fmt.Println()
+
+	// Check for pending migrations.
+	pending, err := migrate.DefaultRegistry.PendingConfig(configDir)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error checking pending migrations: %v\n", err)
+		os.Exit(1)
+	}
+
+	if len(pending) == 0 {
+		fmt.Println("No pending config migrations.")
+		return
+	}
+
+	fmt.Println("Pending config migrations:")
+	for _, p := range pending {
+		fmt.Printf("  [config] %s %d → %d: %s\n", p.Target, p.FromVersion, p.ToVersion, p.Description)
+	}
+	fmt.Println()
+
+	// Prompt for confirmation unless --yes.
+	if !yes {
+		fmt.Print("Apply? [y/N] ")
+		reader := bufio.NewReader(os.Stdin)
+		answer, _ := reader.ReadString('\n')
+		answer = strings.TrimSpace(strings.ToLower(answer))
+		if answer != "y" && answer != "yes" {
+			fmt.Println("Aborted.")
+			return
+		}
+	}
+
+	// Apply migrations.
+	if err := migrate.DefaultRegistry.RunAllConfig(configDir, dataDir); err != nil {
+		fmt.Fprintf(os.Stderr, "Migration failed: %v\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Println("\nDone. Restart services to apply: sudo systemctl restart spinifex.target")
+}

cmd/spinifex/cmd/templates/nats.conf

@@ -1,3 +1,4 @@
+# spinifex-config-version: 1
 # NATS Server Configuration
 # Generated by Spinifex admin init
 

docs/COMMANDS.md

@@ -72,6 +72,12 @@ Operational commands for inspecting cluster state. These fan out NATS requests t
 |---------|-------|---------------|-------------|------------|--------|
 | `spx admin cert renew` | `--extra-ip` (additional IPs for SANs), `--extra-dns` (additional DNS names for SANs) | Existing CA from `spx admin init` | Reads existing CA → regenerates server certificate with all current network interface IPs and machine hostname in SANs → writes new cert. Use after adding a new network interface or changing IP addresses. | 1. Renew with auto-detected IPs<br>2. Renew with extra IPs<br>3. Renew with extra DNS names | **DONE** |
 
+### Upgrade Management
+
+| Command | Flags | Prerequisites | Basic Logic | Test Cases | Status |
+|---------|-------|---------------|-------------|------------|--------|
+| `spx admin upgrade` | `--yes` (apply without prompting), `--config-dir` (persistent, default: `~/spinifex/config`), `--spinifex-dir` (persistent, default: `~/spinifex`) | Existing installation (`spinifex.toml` must exist in config dir) | Reads current config versions from registry → computes pending config migrations (from→to per target) → prints version summary and pending list → prompts for confirmation unless `--yes` → runs `migrate.DefaultRegistry.RunAllConfig()` to apply migrations to config files. Intended for upgrades between Spinifex versions. Operators can skip migrations during install by setting `INSTALL_SPINIFEX_SKIP_MIGRATE=1`, then run `spx admin upgrade` manually to review and apply. After completion, services must be restarted with `sudo systemctl restart spinifex.target`. Invoked non-interactively by `scripts/setup.sh` with `--yes`. | 1. No installation present (error, suggests `spx admin init`)<br>2. No pending migrations (prints "No pending config migrations")<br>3. Pending migrations with interactive confirmation (y/yes applies, anything else aborts)<br>4. Pending migrations with `--yes` (applies without prompt)<br>5. Migration failure surfaces error and exits non-zero | **DONE** |
+
 ### Account Management
 
 | Command | Flags | Prerequisites | Basic Logic | Test Cases | Status |

scripts/setup.sh

@@ -370,6 +370,93 @@ EOF
     fi
 }
 
+# --- Fix file ownership for upgrades from v1 ---
+fix_file_ownership() {
+    info "Fixing file ownership for privilege separation..."
+
+    # Per-service data dirs — recursive chown so existing files are accessible
+    for entry in \
+        nats:spinifex-nats \
+        predastore:spinifex-storage \
+        spinifex:spinifex-daemon \
+        viperblock:spinifex-viperblock \
+        vpcd:spinifex-vpcd \
+        awsgw:spinifex-gw; do
+        IFS=: read -r dir svc_user <<< "$entry"
+        if [ -d "/var/lib/spinifex/$dir" ]; then
+            $SUDO chown -R "$svc_user:$SPINIFEX_GROUP" "/var/lib/spinifex/$dir" \
+                || fatal "Failed to set ownership on /var/lib/spinifex/$dir"
+        fi
+    done
+
+    # Per-service config dirs — recursive chown
+    if [ -d /etc/spinifex/nats ]; then
+        $SUDO chown -R "spinifex-nats:$SPINIFEX_GROUP" /etc/spinifex/nats \
+            || fatal "Failed to set ownership on /etc/spinifex/nats"
+    fi
+    if [ -d /etc/spinifex/predastore ]; then
+        $SUDO chown -R "spinifex-storage:$SPINIFEX_GROUP" /etc/spinifex/predastore \
+            || fatal "Failed to set ownership on /etc/spinifex/predastore"
+    fi
+
+    # Shared config files — root:spinifex with per-file modes
+    for f in spinifex.toml master.key server.key; do
+        if [ -f "/etc/spinifex/$f" ]; then
+            $SUDO chown "root:$SPINIFEX_GROUP" "/etc/spinifex/$f" \
+                || fatal "Failed to set ownership on /etc/spinifex/$f"
+            $SUDO chmod 0640 "/etc/spinifex/$f" \
+                || fatal "Failed to set permissions on /etc/spinifex/$f"
+        fi
+    done
+    for f in server.pem ca.pem; do
+        if [ -f "/etc/spinifex/$f" ]; then
+            $SUDO chown "root:$SPINIFEX_GROUP" "/etc/spinifex/$f" \
+                || fatal "Failed to set ownership on /etc/spinifex/$f"
+            $SUDO chmod 0644 "/etc/spinifex/$f" \
+                || fatal "Failed to set permissions on /etc/spinifex/$f"
+        fi
+    done
+
+    # CA private key — root-only
+    if [ -f /etc/spinifex/ca.key ]; then
+        $SUDO chown root:root /etc/spinifex/ca.key \
+            || fatal "Failed to set ownership on /etc/spinifex/ca.key"
+        $SUDO chmod 0600 /etc/spinifex/ca.key \
+            || fatal "Failed to set permissions on /etc/spinifex/ca.key"
+    fi
+
+    # Shared data dirs — root:spinifex 0770 (daemon + admin CLI write, services read)
+    for d in images amis volumes state; do
+        if [ -d "/var/lib/spinifex/$d" ]; then
+            $SUDO chown -R "root:$SPINIFEX_GROUP" "/var/lib/spinifex/$d" \
+                || fatal "Failed to set ownership on /var/lib/spinifex/$d"
+            $SUDO chmod 0770 "/var/lib/spinifex/$d" \
+                || fatal "Failed to set permissions on /var/lib/spinifex/$d"
+        fi
+    done
+
+    info "File ownership updated"
+}
+
+# --- Run config migrations ---
+run_migrations() {
+    # Only run if spinifex is already initialized (config exists)
+    if [ ! -f /etc/spinifex/spinifex.toml ]; then
+        info "Fresh install detected, skipping migrations"
+        return
+    fi
+
+    if [ "${INSTALL_SPINIFEX_SKIP_MIGRATE:-0}" = "1" ]; then
+        info "INSTALL_SPINIFEX_SKIP_MIGRATE=1, skipping migrations"
+        info "Run 'spx admin upgrade' manually to apply pending migrations"
+        return
+    fi
+
+    info "Running config migrations..."
+    $SUDO /usr/local/bin/spx admin upgrade --yes \
+        || fatal "Config migration failed. See errors above."
+}
+
 # --- Install systemd units ---
 install_systemd() {
     info "Installing systemd units..."
@@ -471,7 +558,9 @@ main() {
     download_spinifex
     install_files
     create_directories
+    fix_file_ownership
     install_systemd
+    run_migrations
     install_logrotate
     rm -rf "$SPINIFEX_TMPDIR"
     restart_if_needed

spinifex/daemon/jetstream.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go"
@@ -83,8 +84,8 @@ func (m *JetStreamManager) InitKVBucket() error {
 	}
 
 	m.kv = kv
-	if err := utils.WriteVersion(kv, InstanceStateBucketVersion); err != nil {
-		return fmt.Errorf("write version to %s: %w", InstanceStateBucket, err)
+	if err := migrate.DefaultRegistry.RunKV(InstanceStateBucket, kv, InstanceStateBucketVersion); err != nil {
+		return fmt.Errorf("migrate %s: %w", InstanceStateBucket, err)
 	}
 	return nil
 }
@@ -113,8 +114,8 @@ func (m *JetStreamManager) InitClusterStateBucket() error {
 	}
 
 	m.clusterKV = kv
-	if err := utils.WriteVersion(kv, ClusterStateBucketVersion); err != nil {
-		return fmt.Errorf("write version to %s: %w", ClusterStateBucket, err)
+	if err := migrate.DefaultRegistry.RunKV(ClusterStateBucket, kv, ClusterStateBucketVersion); err != nil {
+		return fmt.Errorf("migrate %s: %w", ClusterStateBucket, err)
 	}
 	return nil
 }
@@ -144,8 +145,8 @@ func (m *JetStreamManager) InitTerminatedInstanceBucket() error {
 	}
 
 	m.terminatedKV = kv
-	if err := utils.WriteVersion(kv, TerminatedInstanceBucketVersion); err != nil {
-		return fmt.Errorf("write version to %s: %w", TerminatedInstanceBucket, err)
+	if err := migrate.DefaultRegistry.RunKV(TerminatedInstanceBucket, kv, TerminatedInstanceBucketVersion); err != nil {
+		return fmt.Errorf("migrate %s: %w", TerminatedInstanceBucket, err)
 	}
 	return nil
 }
@@ -199,8 +200,9 @@ func (m *JetStreamManager) recoverBucket(cfg *nats.KeyValueConfig, field *nats.K
 		return nil, err
 	}
 
-	if err := utils.WriteVersion(kv, version); err != nil {
-		slog.Error("Failed to write version to recreated bucket", "bucket", cfg.Bucket, "err", err)
+	if err := migrate.DefaultRegistry.RunKV(cfg.Bucket, kv, version); err != nil {
+		slog.Error("Failed to run migrations on recreated bucket", "bucket", cfg.Bucket, "err", err)
+		return nil, fmt.Errorf("migrate recreated bucket %s: %w", cfg.Bucket, err)
 	}
 
 	*field = kv

spinifex/handlers/ec2/account/service_impl.go

@@ -9,6 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/config"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -44,8 +45,8 @@ func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Co
 	if err != nil {
 		return nil, fmt.Errorf("failed to create account settings KV bucket: %w", err)
 	}
-	if err := utils.WriteVersion(settingsKV, KVBucketAccountSettingsVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketAccountSettings, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketAccountSettings, settingsKV, KVBucketAccountSettingsVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketAccountSettings, err)
 	}
 
 	slog.Info("Account settings service initialized with JetStream KV", "bucket", KVBucketAccountSettings)

spinifex/handlers/ec2/eigw/service_impl.go

@@ -14,6 +14,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/config"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -53,8 +54,8 @@ func NewEgressOnlyIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEgressOnlyIGW, err)
 	}
-	if err := utils.WriteVersion(eigwKV, KVBucketEgressOnlyIGWVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketEgressOnlyIGW, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketEgressOnlyIGW, eigwKV, KVBucketEgressOnlyIGWVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketEgressOnlyIGW, err)
 	}
 
 	// Get or create VPC KV bucket for cross-resource ownership validation

spinifex/handlers/ec2/eip/service_impl.go

@@ -13,6 +13,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -48,8 +49,8 @@ func NewEIPServiceImpl(natsConn *nats.Conn, externalIPAM *handlers_ec2_vpc.Exter
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEIPs, err)
 	}
-	if err := utils.WriteVersion(eipKV, KVBucketEIPsVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketEIPs, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketEIPs, eipKV, KVBucketEIPsVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketEIPs, err)
 	}
 
 	slog.Info("EIP service initialized with JetStream KV", "bucket", KVBucketEIPs)

spinifex/handlers/ec2/igw/service_impl.go

@@ -14,6 +14,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/config"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
@@ -55,8 +56,8 @@ func NewIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*IGWSer
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketIGW, err)
 	}
-	if err := utils.WriteVersion(igwKV, KVBucketIGWVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketIGW, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketIGW, igwKV, KVBucketIGWVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketIGW, err)
 	}
 
 	// Get or create VPC KV bucket for cross-resource ownership validation

spinifex/handlers/ec2/natgw/service_impl.go

@@ -14,6 +14,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
 	handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -41,8 +42,8 @@ func NewNatGatewayServiceImplWithNATS(natsConn *nats.Conn) (*NatGatewayServiceIm
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketNatGateways, err)
 	}
-	if err := utils.WriteVersion(natgwKV, KVBucketNatGatewaysVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketNatGateways, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketNatGateways, natgwKV, KVBucketNatGatewaysVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketNatGateways, err)
 	}
 
 	// Deleted NAT Gateways bucket with 1-hour TTL — keys auto-expire.
@@ -59,8 +60,8 @@ func NewNatGatewayServiceImplWithNATS(natsConn *nats.Conn) (*NatGatewayServiceIm
 			return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketDeletedNatGateways, err)
 		}
 	}
-	if err := utils.WriteVersion(deletedKV, KVBucketDeletedNatGatewaysVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketDeletedNatGateways, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketDeletedNatGateways, deletedKV, KVBucketDeletedNatGatewaysVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketDeletedNatGateways, err)
 	}
 
 	eipKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_eip.KVBucketEIPs, 10)

spinifex/handlers/ec2/placementgroup/service_impl.go

@@ -13,6 +13,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/config"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -57,8 +58,8 @@ func NewPlacementGroupServiceImplWithNATS(cfg *config.Config, natsConn *nats.Con
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketPlacementGroups, err)
 	}
-	if err := utils.WriteVersion(kv, KVBucketPlacementGroupsVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketPlacementGroups, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketPlacementGroups, kv, KVBucketPlacementGroupsVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketPlacementGroups, err)
 	}
 
 	slog.Info("Placement group service initialized with JetStream KV", "bucket", KVBucketPlacementGroups)

spinifex/handlers/ec2/routetable/service_impl.go

@@ -16,6 +16,7 @@ import (
 	handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
 	handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -45,8 +46,8 @@ func NewRouteTableServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketRouteTables, err)
 	}
-	if err := utils.WriteVersion(rtbKV, KVBucketRouteTablesVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketRouteTables, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketRouteTables, rtbKV, KVBucketRouteTablesVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketRouteTables, err)
 	}
 
 	vpcKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketVPCs, 10)

spinifex/handlers/ec2/snapshot/service_impl.go

@@ -18,6 +18,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/config"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/utils"
@@ -75,8 +76,8 @@ func NewSnapshotServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*S
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVolumeSnapshots, err)
 	}
-	if err := utils.WriteVersion(kv, KVBucketVolumeSnapshotsVersion); err != nil {
-		return nil, nil, fmt.Errorf("write version to %s: %w", KVBucketVolumeSnapshots, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketVolumeSnapshots, kv, KVBucketVolumeSnapshotsVersion); err != nil {
+		return nil, nil, fmt.Errorf("migrate %s: %w", KVBucketVolumeSnapshots, err)
 	}
 
 	slog.Info("Snapshot service initialized with JetStream KV", "bucket", KVBucketVolumeSnapshots)

spinifex/handlers/ec2/vpc/external_ipam.go

@@ -8,6 +8,7 @@ import (
 	"math/big"
 	"net"
 
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -71,8 +72,8 @@ func NewExternalIPAM(js nats.JetStreamContext, pools []ExternalPoolConfig) (*Ext
 	if err != nil {
 		return nil, fmt.Errorf("create external IPAM KV bucket: %w", err)
 	}
-	if err := utils.WriteVersion(kv, KVBucketExternalIPAMVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketExternalIPAM, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketExternalIPAM, kv, KVBucketExternalIPAMVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketExternalIPAM, err)
 	}
 	ipam := &ExternalIPAM{kv: kv, pools: pools}
 	if err := ipam.initPools(); err != nil {

spinifex/handlers/ec2/vpc/ipam.go

@@ -8,6 +8,7 @@ import (
 	"math/big"
 	"net"
 
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -35,8 +36,8 @@ func NewIPAM(js nats.JetStreamContext) (*IPAM, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to create IPAM KV bucket: %w", err)
 	}
-	if err := utils.WriteVersion(kv, KVBucketIPAMVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketIPAM, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketIPAM, kv, KVBucketIPAMVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketIPAM, err)
 	}
 	return &IPAM{kv: kv}, nil
 }

spinifex/handlers/ec2/vpc/service_impl.go

@@ -15,6 +15,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/config"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -96,40 +97,40 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVPCs, err)
 	}
-	if err := utils.WriteVersion(vpcKV, KVBucketVPCsVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketVPCs, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketVPCs, vpcKV, KVBucketVPCsVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketVPCs, err)
 	}
 
 	subnetKV, err := utils.GetOrCreateKVBucket(js, KVBucketSubnets, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSubnets, err)
 	}
-	if err := utils.WriteVersion(subnetKV, KVBucketSubnetsVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketSubnets, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketSubnets, subnetKV, KVBucketSubnetsVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketSubnets, err)
 	}
 
 	vniKV, err := utils.GetOrCreateKVBucket(js, KVBucketVNICounter, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVNICounter, err)
 	}
-	if err := utils.WriteVersion(vniKV, KVBucketVNICounterVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketVNICounter, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketVNICounter, vniKV, KVBucketVNICounterVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketVNICounter, err)
 	}
 
 	eniKV, err := utils.GetOrCreateKVBucket(js, KVBucketENIs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketENIs, err)
 	}
-	if err := utils.WriteVersion(eniKV, KVBucketENIsVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketENIs, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketENIs, eniKV, KVBucketENIsVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketENIs, err)
 	}
 
 	sgKV, err := utils.GetOrCreateKVBucket(js, KVBucketSecurityGroups, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSecurityGroups, err)
 	}
-	if err := utils.WriteVersion(sgKV, KVBucketSecurityGroupsVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketSecurityGroups, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketSecurityGroups, sgKV, KVBucketSecurityGroupsVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketSecurityGroups, err)
 	}
 
 	rtbKV, err := utils.GetOrCreateKVBucket(js, "spinifex-vpc-route-tables", 10)

spinifex/handlers/elbv2/store.go

@@ -7,6 +7,7 @@ import (
 	"log/slog"
 	"strings"
 
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -38,8 +39,8 @@ func NewStore(nc *nats.Conn) (*Store, error) {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketELBv2, err)
 	}
 
-	if err := utils.WriteVersion(kv, KVBucketELBv2Version); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketELBv2, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketELBv2, kv, KVBucketELBv2Version); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketELBv2, err)
 	}
 
 	slog.Info("ELBv2 store initialized", "bucket", KVBucketELBv2)

spinifex/handlers/iam/service_impl.go

@@ -16,6 +16,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/mulgadc/spinifex/spinifex/admin"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -69,40 +70,40 @@ func NewIAMServiceImpl(natsConn *nats.Conn, masterKey []byte, clusterSize int) (
 	if err != nil {
 		return nil, fmt.Errorf("init users bucket: %w", err)
 	}
-	if err := utils.WriteVersion(usersBucket, KVBucketUsersVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketUsers, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketUsers, usersBucket, KVBucketUsersVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketUsers, err)
 	}
 
 	accessKeysBucket, err := getOrCreateBucket(js, KVBucketAccessKeys, 5, replicas)
 	if err != nil {
 		return nil, fmt.Errorf("init access keys bucket: %w", err)
 	}
-	if err := utils.WriteVersion(accessKeysBucket, KVBucketAccessKeysVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketAccessKeys, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketAccessKeys, accessKeysBucket, KVBucketAccessKeysVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketAccessKeys, err)
 	}
 
 	policiesBucket, err := getOrCreateBucket(js, KVBucketPolicies, 10, replicas)
 	if err != nil {
 		return nil, fmt.Errorf("init policies bucket: %w", err)
 	}
-	if err := utils.WriteVersion(policiesBucket, KVBucketPoliciesVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketPolicies, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketPolicies, policiesBucket, KVBucketPoliciesVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketPolicies, err)
 	}
 
 	accountsBucket, err := getOrCreateBucket(js, KVBucketAccounts, 5, replicas)
 	if err != nil {
 		return nil, fmt.Errorf("init accounts bucket: %w", err)
 	}
-	if err := utils.WriteVersion(accountsBucket, KVBucketAccountsVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketAccounts, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketAccounts, accountsBucket, KVBucketAccountsVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketAccounts, err)
 	}
 
 	accountCounterBucket, err := getOrCreateBucket(js, KVBucketAccountCounter, 5, replicas)
 	if err != nil {
 		return nil, fmt.Errorf("init account counter bucket: %w", err)
 	}
-	if err := utils.WriteVersion(accountCounterBucket, KVBucketAccountCounterVersion); err != nil {
-		return nil, fmt.Errorf("write version to %s: %w", KVBucketAccountCounter, err)
+	if err := migrate.DefaultRegistry.RunKV(KVBucketAccountCounter, accountCounterBucket, KVBucketAccountCounterVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketAccountCounter, err)
 	}
 
 	decrypter, err := NewDecrypter(masterKey)

spinifex/migrate/001_nats_token.go

@@ -0,0 +1,97 @@
+package migrate
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"regexp"
+)
+
+func init() {
+	DefaultRegistry.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+
+	DefaultRegistry.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0,
+		ToVersion:   1,
+		Description: "Enable NATS auth token and relocate log_file to /var/log/spinifex",
+		Run: func(ctx ConfigContext) error {
+			path := filepath.Join(ctx.ConfigDir, "nats", "nats.conf")
+			content, err := os.ReadFile(path)
+			if err != nil {
+				return err
+			}
+
+			updated := content
+			var changed bool
+
+			// 1. Uncomment the token line inside the authorization block.
+			// Matches "#  token:" with any leading/interstitial whitespace.
+			commentedRe := regexp.MustCompile(`(?m)^(\s*)#(\s*token:\s*)`)
+			uncommentedRe := regexp.MustCompile(`(?m)^\s*token:`)
+
+			if commentedRe.Match(updated) {
+				updated = commentedRe.ReplaceAll(updated, []byte("${1}${2}"))
+				changed = true
+			} else if !uncommentedRe.Match(updated) {
+				return fmt.Errorf("nats.conf migration failed: token line not found (neither commented nor uncommented)")
+			}
+
+			// 2. Convert cluster route URLs from nats:// to nats-route://token@.
+			// Existing deployments have: "nats://host:port"
+			// Need:                      "nats-route://TOKEN@host:port"
+			// Extract the token value from the (now uncommented) token line.
+			tokenValueRe := regexp.MustCompile(`(?m)^\s*token:\s*"([^"]+)"`)
+			tokenMatch := tokenValueRe.FindSubmatch(updated)
+
+			if tokenMatch != nil {
+				token := string(tokenMatch[1])
+				// Match route lines: "nats://host:port" (without token already embedded).
+				routeRe := regexp.MustCompile(`"nats://([^@"]+)"`)
+				if routeRe.Match(updated) {
+					replacement := fmt.Sprintf(`"nats-route://%s@${1}"`, token)
+					updated = routeRe.ReplaceAll(updated, []byte(replacement))
+					changed = true
+				}
+			}
+
+			// 3. Relocate log_file from data dir to /var/log/spinifex.
+			// v1.0.2 rendered: log_file: "/var/lib/spinifex/logs/nats.log"
+			// New location:    log_file: "/var/log/spinifex/nats.log"
+			// The new systemd unit sandbox only permits writes under /var/log/spinifex,
+			// so leaving the old path causes nats to crash-loop on "permission denied".
+			oldLogRe := regexp.MustCompile(`(?m)^log_file:\s*"/var/lib/spinifex/logs/nats\.log"`)
+			newLogLine := `log_file: "/var/log/spinifex/nats.log"`
+			if oldLogRe.Match(updated) {
+				updated = oldLogRe.ReplaceAll(updated, []byte(newLogLine))
+				changed = true
+			}
+
+			if !changed {
+				ctx.Logger.Info("nats.conf: already migrated, skipping")
+				return nil
+			}
+
+			info, err := os.Stat(path)
+			if err != nil {
+				return err
+			}
+			if err := os.WriteFile(path, updated, info.Mode()); err != nil {
+				return err
+			}
+
+			// Post-migration validation.
+			result, err := os.ReadFile(path)
+			if err != nil {
+				return fmt.Errorf("nats.conf migration: post-migration validation read failed: %w", err)
+			}
+			if !uncommentedRe.Match(result) {
+				return fmt.Errorf("nats.conf migration failed: token line not found after uncommenting")
+			}
+			// Validate log_file no longer points at the old data-dir location.
+			if oldLogRe.Match(result) {
+				return fmt.Errorf("nats.conf migration failed: log_file still points to /var/lib/spinifex/logs after rewrite")
+			}
+			return nil
+		},
+	})
+}

spinifex/migrate/migrate.go

@@ -0,0 +1,330 @@
+package migrate
+
+import (
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"sort"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+)
+
+// KVMigration represents a versioned transformation of KV bucket data.
+type KVMigration struct {
+	FromVersion int
+	ToVersion   int
+	Description string
+	Run         func(ctx KVContext) error
+}
+
+// KVContext provides KV migration functions access to the bucket being migrated.
+type KVContext struct {
+	KV     nats.KeyValue
+	Logger *slog.Logger
+}
+
+// ConfigMigration represents a versioned transformation of on-disk config files.
+type ConfigMigration struct {
+	FromVersion int
+	ToVersion   int
+	Description string
+	Run         func(ctx ConfigContext) error
+}
+
+// ConfigContext provides config migration functions access to the filesystem.
+type ConfigContext struct {
+	ConfigDir string // path to /etc/spinifex or equivalent
+	DataDir   string // path to /var/lib/spinifex or equivalent
+	Logger    *slog.Logger
+}
+
+// PendingMigration describes a migration that has not yet been applied.
+type PendingMigration struct {
+	Target      string
+	FromVersion int
+	ToVersion   int
+	Description string
+}
+
+// configTarget bundles the relative path and version reader for a config file.
+type configTarget struct {
+	path   string // relative to configDir (e.g. "nats/nats.conf")
+	reader ConfigVersionReader
+}
+
+// Registry holds all registered migrations, keyed by target name.
+type Registry struct {
+	kvMigrations     map[string][]KVMigration
+	configMigrations map[string][]ConfigMigration
+	configTargets    map[string]configTarget
+}
+
+// DefaultRegistry is the global migration registry. Migrations self-register
+// via init() functions so they are available before any service starts.
+var DefaultRegistry = NewRegistry()
+
+// NewRegistry creates an empty migration registry.
+func NewRegistry() *Registry {
+	return &Registry{
+		kvMigrations:     make(map[string][]KVMigration),
+		configMigrations: make(map[string][]ConfigMigration),
+		configTargets:    make(map[string]configTarget),
+	}
+}
+
+// RegisterKV adds a KV bucket migration. Migrations are kept sorted by FromVersion.
+func (r *Registry) RegisterKV(bucket string, m KVMigration) {
+	r.kvMigrations[bucket] = append(r.kvMigrations[bucket], m)
+	sort.Slice(r.kvMigrations[bucket], func(i, j int) bool {
+		return r.kvMigrations[bucket][i].FromVersion < r.kvMigrations[bucket][j].FromVersion
+	})
+}
+
+// RegisterConfigTarget registers a config file target with its relative path
+// and version reader. Must be called before RegisterConfig for the same target.
+func (r *Registry) RegisterConfigTarget(name, relPath string, reader ConfigVersionReader) {
+	r.configTargets[name] = configTarget{path: relPath, reader: reader}
+}
+
+// RegisterConfig adds a config file migration. Migrations are kept sorted by FromVersion.
+func (r *Registry) RegisterConfig(target string, m ConfigMigration) {
+	r.configMigrations[target] = append(r.configMigrations[target], m)
+	sort.Slice(r.configMigrations[target], func(i, j int) bool {
+		return r.configMigrations[target][i].FromVersion < r.configMigrations[target][j].FromVersion
+	})
+}
+
+// RunKV executes all pending migrations for a KV bucket up to targetVersion.
+// It reads the current version via utils.ReadVersion, runs each migration in
+// order, and stamps the version via utils.WriteVersion after each step.
+//
+// When no migrations are registered for a bucket, RunKV stamps targetVersion
+// directly — equivalent to a bare utils.WriteVersion. This is the common case
+// for buckets that have not yet needed a schema change.
+//
+// If targetVersion is ahead of the registered migration chain, RunKV returns
+// an error — migration authors must register migrations before bumping version
+// constants.
+func (r *Registry) RunKV(bucket string, kv nats.KeyValue, targetVersion int) error {
+	current, err := utils.ReadVersion(kv)
+	if err != nil {
+		return fmt.Errorf("read version for %s: %w", bucket, err)
+	}
+
+	if current >= targetVersion {
+		return nil
+	}
+
+	all := r.kvMigrations[bucket]
+
+	// Fresh bucket with no registered migrations — stamp directly. This is
+	// the common first-init path for buckets that have not yet needed a
+	// schema change. Equivalent to the pre-framework utils.WriteVersion call.
+	// Note: we do NOT stamp directly for a non-fresh bucket with no migrations
+	// (e.g. someone bumped a BucketVersion constant from 1 to 2 without adding
+	// a migration) — that case falls through and errors with "no migrations
+	// registered" to force the author to add the migration.
+	if current == 0 && len(all) == 0 {
+		return utils.WriteVersion(kv, targetVersion)
+	}
+
+	// Migrations are registered (or an existing bucket has a missing chain) —
+	// require a complete chain from current to target.
+	var pending []KVMigration
+	for _, m := range all {
+		if m.FromVersion >= current && m.ToVersion <= targetVersion {
+			pending = append(pending, m)
+		}
+	}
+
+	if len(pending) == 0 {
+		return fmt.Errorf("no migrations registered for %s from version %d to %d", bucket, current, targetVersion)
+	}
+
+	// Validate contiguous chain.
+	expected := current
+	for _, m := range pending {
+		if m.FromVersion != expected {
+			return fmt.Errorf("migration chain gap for %s: expected from %d, got from %d", bucket, expected, m.FromVersion)
+		}
+		expected = m.ToVersion
+	}
+	if expected != targetVersion {
+		return fmt.Errorf("migration chain for %s ends at version %d, target is %d", bucket, expected, targetVersion)
+	}
+
+	logger := slog.Default()
+	for _, m := range pending {
+		logger.Info("Running KV migration", "bucket", bucket, "from", m.FromVersion, "to", m.ToVersion, "description", m.Description)
+		ctx := KVContext{KV: kv, Logger: logger}
+		if err := m.Run(ctx); err != nil {
+			return fmt.Errorf("KV migration %s %d→%d failed: %w", bucket, m.FromVersion, m.ToVersion, err)
+		}
+		if err := utils.WriteVersion(kv, m.ToVersion); err != nil {
+			return fmt.Errorf("stamp version %d on %s: %w", m.ToVersion, bucket, err)
+		}
+	}
+
+	return nil
+}
+
+// RunConfig executes all pending config migrations for a target.
+// Before each migration step, a timestamped backup is created.
+// Returns an error on failure (setup.sh should abort).
+func (r *Registry) RunConfig(target string, configDir, dataDir string) error {
+	t, ok := r.configTargets[target]
+	if !ok {
+		return fmt.Errorf("unknown config target: %s", target)
+	}
+
+	migrations := r.configMigrations[target]
+	if len(migrations) == 0 {
+		return nil
+	}
+
+	fullPath := filepath.Join(configDir, t.path)
+
+	// If the config file doesn't exist (fresh install), skip migrations.
+	if _, err := os.Stat(fullPath); err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return fmt.Errorf("stat config file %s: %w", fullPath, err)
+	}
+
+	current, err := t.reader.ReadVersion(fullPath)
+	if err != nil {
+		return fmt.Errorf("read version for %s: %w", target, err)
+	}
+
+	var pending []ConfigMigration
+	for _, m := range migrations {
+		if m.FromVersion >= current {
+			pending = append(pending, m)
+		}
+	}
+
+	if len(pending) == 0 {
+		return nil
+	}
+
+	// Validate contiguous chain (same check as RunKV).
+	expected := current
+	for _, m := range pending {
+		if m.FromVersion != expected {
+			return fmt.Errorf("config migration chain gap for %s: expected from %d, got from %d", target, expected, m.FromVersion)
+		}
+		expected = m.ToVersion
+	}
+
+	logger := slog.Default()
+	for _, m := range pending {
+		// Create backup before migration.
+		backupPath, err := BackupConfig(fullPath, m.FromVersion, m.ToVersion)
+		if err != nil {
+			return fmt.Errorf("backup %s before migration %d→%d: %w", target, m.FromVersion, m.ToVersion, err)
+		}
+		logger.Info("Created config backup", "target", target, "backup", backupPath)
+
+		logger.Info("Running config migration", "target", target, "from", m.FromVersion, "to", m.ToVersion, "description", m.Description)
+		ctx := ConfigContext{ConfigDir: configDir, DataDir: dataDir, Logger: logger}
+		if err := m.Run(ctx); err != nil {
+			return fmt.Errorf("config migration %s %d→%d failed: %w", target, m.FromVersion, m.ToVersion, err)
+		}
+
+		// Stamp new version after successful migration.
+		if err := t.reader.WriteVersion(fullPath, m.ToVersion); err != nil {
+			return fmt.Errorf("stamp version %d on %s: %w", m.ToVersion, target, err)
+		}
+	}
+
+	return nil
+}
+
+// RunAllConfig executes pending config migrations for all registered targets.
+// Used by `spx admin upgrade`.
+func (r *Registry) RunAllConfig(configDir, dataDir string) error {
+	for target := range r.configTargets {
+		if err := r.RunConfig(target, configDir, dataDir); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// PendingConfig returns all pending config migrations across all targets.
+func (r *Registry) PendingConfig(configDir string) ([]PendingMigration, error) {
+	var result []PendingMigration
+	for name, t := range r.configTargets {
+		fullPath := filepath.Join(configDir, t.path)
+
+		if _, err := os.Stat(fullPath); err != nil {
+			if os.IsNotExist(err) {
+				continue
+			}
+			return nil, fmt.Errorf("stat config file %s: %w", fullPath, err)
+		}
+
+		current, err := t.reader.ReadVersion(fullPath)
+		if err != nil {
+			return nil, fmt.Errorf("read version for %s: %w", name, err)
+		}
+
+		for _, m := range r.configMigrations[name] {
+			if m.FromVersion >= current {
+				result = append(result, PendingMigration{
+					Target:      name,
+					FromVersion: m.FromVersion,
+					ToVersion:   m.ToVersion,
+					Description: m.Description,
+				})
+			}
+		}
+	}
+	return result, nil
+}
+
+// ConfigVersions returns current versions for all registered config targets.
+// Targets whose file does not exist are omitted.
+func (r *Registry) ConfigVersions(configDir string) (map[string]int, error) {
+	versions := make(map[string]int)
+	for name, t := range r.configTargets {
+		fullPath := filepath.Join(configDir, t.path)
+		if _, err := os.Stat(fullPath); err != nil {
+			if os.IsNotExist(err) {
+				continue
+			}
+			return nil, fmt.Errorf("stat config file %s: %w", fullPath, err)
+		}
+		v, err := t.reader.ReadVersion(fullPath)
+		if err != nil {
+			return nil, fmt.Errorf("read version for %s: %w", name, err)
+		}
+		versions[name] = v
+	}
+	return versions, nil
+}
+
+// BackupConfig creates a timestamped backup of a config file before migration.
+// The backup is stored alongside the original file.
+func BackupConfig(path string, fromVersion, toVersion int) (string, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return "", fmt.Errorf("read file for backup: %w", err)
+	}
+
+	info, err := os.Stat(path)
+	if err != nil {
+		return "", fmt.Errorf("stat file for backup: %w", err)
+	}
+
+	backupPath := fmt.Sprintf("%s.pre-migrate-%dto%d.%d", path, fromVersion, toVersion, time.Now().Unix())
+	if err := os.WriteFile(backupPath, data, info.Mode()); err != nil {
+		return "", fmt.Errorf("write backup: %w", err)
+	}
+
+	return backupPath, nil
+}

spinifex/migrate/migrate_test.go

@@ -0,0 +1,738 @@
+package migrate
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// startTestNATS starts an embedded NATS server with JetStream for testing.
+func startTestNATS(t *testing.T) (*server.Server, *nats.Conn) {
+	t.Helper()
+	ns, nc, _ := testutil.StartTestJetStream(t)
+	return ns, nc
+}
+
+func createTestBucket(t *testing.T, nc *nats.Conn, name string) nats.KeyValue {
+	t.Helper()
+	js, err := nc.JetStream()
+	require.NoError(t, err)
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: name, History: 1})
+	require.NoError(t, err)
+	return kv
+}
+
+// --- Registry validation tests ---
+
+func TestRegistry_ValidatesChainNoGaps(t *testing.T) {
+	r := NewRegistry()
+	r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "first", Run: func(KVContext) error { return nil }})
+	r.RegisterKV("test-bucket", KVMigration{FromVersion: 3, ToVersion: 4, Description: "gap", Run: func(KVContext) error { return nil }})
+
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+	// Stamp version 1 to simulate existing bucket.
+	_, err := kv.PutString("_version", "1")
+	require.NoError(t, err)
+
+	err = r.RunKV("test-bucket", kv, 4)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "gap")
+}
+
+func TestRegistry_RejectsDuplicateVersions(t *testing.T) {
+	r := NewRegistry()
+	r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "first", Run: func(KVContext) error { return nil }})
+	r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "duplicate", Run: func(KVContext) error { return nil }})
+
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+	_, err := kv.PutString("_version", "1")
+	require.NoError(t, err)
+
+	// Two migrations with the same FromVersion — chain validation will catch this
+	// because after running first 1→2, the second 1→2 won't match expected=2.
+	err = r.RunKV("test-bucket", kv, 2)
+	// Should succeed (first 1→2 runs, second is filtered out since FromVersion < current after first runs).
+	// Actually with our filtering, both have FromVersion=1 >= current=1 and ToVersion=2 <= target=2,
+	// so both are in pending. Chain validation: expected=1, first.From=1 ✓, expected=2, second.From=1 ✗.
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "gap")
+}
+
+// --- RunKV tests ---
+
+func TestRunKV_NoPendingMigrations_NoOp(t *testing.T) {
+	r := NewRegistry()
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+
+	// Stamp version 1 directly.
+	_, err := kv.PutString("_version", "1")
+	require.NoError(t, err)
+
+	// RunKV with target=1, current=1 → no-op.
+	err = r.RunKV("test-bucket", kv, 1)
+	assert.NoError(t, err)
+}
+
+func TestRunKV_FreshBucket_StampsVersion(t *testing.T) {
+	r := NewRegistry()
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+
+	// No version set → version 0 (fresh bucket).
+	err := r.RunKV("test-bucket", kv, 1)
+	assert.NoError(t, err)
+
+	// Verify version was stamped.
+	entry, err := kv.Get("_version")
+	require.NoError(t, err)
+	assert.Equal(t, "1", string(entry.Value()))
+}
+
+func TestRunKV_ExecutesMigrationsInOrder(t *testing.T) {
+	var order []int
+	r := NewRegistry()
+	r.RegisterKV("test-bucket", KVMigration{
+		FromVersion: 1, ToVersion: 2, Description: "step 1",
+		Run: func(KVContext) error { order = append(order, 1); return nil },
+	})
+	r.RegisterKV("test-bucket", KVMigration{
+		FromVersion: 2, ToVersion: 3, Description: "step 2",
+		Run: func(KVContext) error { order = append(order, 2); return nil },
+	})
+
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+	_, err := kv.PutString("_version", "1")
+	require.NoError(t, err)
+
+	err = r.RunKV("test-bucket", kv, 3)
+	assert.NoError(t, err)
+	assert.Equal(t, []int{1, 2}, order)
+
+	// Verify final version.
+	entry, err := kv.Get("_version")
+	require.NoError(t, err)
+	assert.Equal(t, "3", string(entry.Value()))
+}
+
+func TestRunKV_StampsAfterEachStep(t *testing.T) {
+	r := NewRegistry()
+	r.RegisterKV("test-bucket", KVMigration{
+		FromVersion: 1, ToVersion: 2, Description: "step 1",
+		Run: func(ctx KVContext) error {
+			// After this runs, version should be stamped to 2 by RunKV.
+			return nil
+		},
+	})
+	r.RegisterKV("test-bucket", KVMigration{
+		FromVersion: 2, ToVersion: 3, Description: "step 2 fails",
+		Run: func(KVContext) error { return errors.New("boom") },
+	})
+
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+	_, err := kv.PutString("_version", "1")
+	require.NoError(t, err)
+
+	err = r.RunKV("test-bucket", kv, 3)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "boom")
+
+	// Version should be 2 (stamped after step 1, before step 2 failed).
+	entry, err := kv.Get("_version")
+	require.NoError(t, err)
+	assert.Equal(t, "2", string(entry.Value()))
+}
+
+func TestRunKV_StopsOnFailure_VersionNotBumped(t *testing.T) {
+	r := NewRegistry()
+	r.RegisterKV("test-bucket", KVMigration{
+		FromVersion: 1, ToVersion: 2, Description: "fails",
+		Run: func(KVContext) error { return errors.New("migration error") },
+	})
+
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+	_, err := kv.PutString("_version", "1")
+	require.NoError(t, err)
+
+	err = r.RunKV("test-bucket", kv, 2)
+	assert.Error(t, err)
+
+	// Version remains at 1.
+	entry, err := kv.Get("_version")
+	require.NoError(t, err)
+	assert.Equal(t, "1", string(entry.Value()))
+}
+
+func TestRunKV_RejectsMissingMigration(t *testing.T) {
+	r := NewRegistry()
+	// No migrations registered, but bucket is at version 1 and target is 2.
+
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+	_, err := kv.PutString("_version", "1")
+	require.NoError(t, err)
+
+	err = r.RunKV("test-bucket", kv, 2)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "no migrations registered")
+}
+
+func TestRunKV_Idempotent(t *testing.T) {
+	runCount := 0
+	r := NewRegistry()
+	r.RegisterKV("test-bucket", KVMigration{
+		FromVersion: 1, ToVersion: 2, Description: "add field",
+		Run: func(ctx KVContext) error {
+			runCount++
+			// Idempotent: write a key, re-running writes the same value.
+			_, err := ctx.KV.PutString("data.key1", `{"field":"value"}`)
+			return err
+		},
+	})
+
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+	_, err := kv.PutString("_version", "1")
+	require.NoError(t, err)
+
+	// First run.
+	err = r.RunKV("test-bucket", kv, 2)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, runCount)
+
+	// Second run — already at version 2, should be no-op.
+	err = r.RunKV("test-bucket", kv, 2)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, runCount) // Not incremented.
+}
+
+// --- Config version reader tests ---
+
+func TestTOMLVersionReader_ReadStringVersion(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "spinifex.toml")
+	require.NoError(t, os.WriteFile(path, []byte(`version = "1.0"
+[server]
+port = 8080
+`), 0o644))
+
+	r := &TOMLVersionReader{}
+	v, err := r.ReadVersion(path)
+	require.NoError(t, err)
+	assert.Equal(t, 1, v)
+}
+
+func TestTOMLVersionReader_ReadIntVersion(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "config.toml")
+	require.NoError(t, os.WriteFile(path, []byte(`version = 2
+[server]
+port = 8080
+`), 0o644))
+
+	r := &TOMLVersionReader{}
+	v, err := r.ReadVersion(path)
+	require.NoError(t, err)
+	assert.Equal(t, 2, v)
+}
+
+func TestTOMLVersionReader_WriteVersion(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "config.toml")
+	require.NoError(t, os.WriteFile(path, []byte(`version = "1.0"
+[server]
+port = 8080
+`), 0o644))
+
+	r := &TOMLVersionReader{}
+	require.NoError(t, r.WriteVersion(path, 2))
+
+	data, err := os.ReadFile(path)
+	require.NoError(t, err)
+	assert.Contains(t, string(data), "version = 2")
+	assert.NotContains(t, string(data), `"1.0"`)
+	assert.Contains(t, string(data), "port = 8080") // rest preserved
+}
+
+func TestNATSConfVersionReader_ReadNoMarker(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "nats.conf")
+	require.NoError(t, os.WriteFile(path, []byte(`# NATS Server Configuration
+listen: 0.0.0.0:4222
+`), 0o644))
+
+	r := &NATSConfVersionReader{}
+	v, err := r.ReadVersion(path)
+	require.NoError(t, err)
+	assert.Equal(t, 0, v)
+}
+
+func TestNATSConfVersionReader_ReadWithMarker(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "nats.conf")
+	require.NoError(t, os.WriteFile(path, []byte(`# spinifex-config-version: 1
+# NATS Server Configuration
+listen: 0.0.0.0:4222
+`), 0o644))
+
+	r := &NATSConfVersionReader{}
+	v, err := r.ReadVersion(path)
+	require.NoError(t, err)
+	assert.Equal(t, 1, v)
+}
+
+func TestNATSConfVersionReader_WriteVersion_Prepend(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "nats.conf")
+	require.NoError(t, os.WriteFile(path, []byte(`# NATS Server Configuration
+listen: 0.0.0.0:4222
+`), 0o644))
+
+	r := &NATSConfVersionReader{}
+	require.NoError(t, r.WriteVersion(path, 1))
+
+	data, err := os.ReadFile(path)
+	require.NoError(t, err)
+	lines := strings.Split(string(data), "\n")
+	assert.Equal(t, "# spinifex-config-version: 1", lines[0])
+	assert.Equal(t, "# NATS Server Configuration", lines[1])
+}
+
+func TestNATSConfVersionReader_WriteVersion_Replace(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "nats.conf")
+	require.NoError(t, os.WriteFile(path, []byte(`# spinifex-config-version: 1
+# NATS Server Configuration
+listen: 0.0.0.0:4222
+`), 0o644))
+
+	r := &NATSConfVersionReader{}
+	require.NoError(t, r.WriteVersion(path, 2))
+
+	data, err := os.ReadFile(path)
+	require.NoError(t, err)
+	lines := strings.Split(string(data), "\n")
+	assert.Equal(t, "# spinifex-config-version: 2", lines[0])
+	assert.Equal(t, "# NATS Server Configuration", lines[1])
+}
+
+// --- Config migration tests ---
+
+func TestRunConfig_CreatesBackup(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+authorization {
+#  token: "testtoken123"
+}
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0, ToVersion: 1, Description: "test migration",
+		Run: func(ctx ConfigContext) error {
+			return nil // no-op for backup test
+		},
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	require.NoError(t, err)
+
+	// Check backup exists.
+	entries, err := os.ReadDir(natsDir)
+	require.NoError(t, err)
+	var found bool
+	for _, e := range entries {
+		if strings.HasPrefix(e.Name(), "nats.conf.pre-migrate-0to1.") {
+			found = true
+			break
+		}
+	}
+	assert.True(t, found, "backup file should exist")
+}
+
+func TestRunConfig_SkipsFreshInstall(t *testing.T) {
+	dir := t.TempDir()
+	// No nats.conf file exists — fresh install.
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0, ToVersion: 1, Description: "should not run",
+		Run: func(ctx ConfigContext) error {
+			t.Fatal("migration should not run on fresh install")
+			return nil
+		},
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	assert.NoError(t, err)
+}
+
+func TestRunConfig_AlreadyMigrated_NoOp(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# spinifex-config-version: 1
+# NATS Server Configuration
+authorization {
+  token: "testtoken123"
+}
+`), 0o640))
+
+	ran := false
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0, ToVersion: 1, Description: "already done",
+		Run: func(ctx ConfigContext) error { ran = true; return nil },
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	assert.NoError(t, err)
+	assert.False(t, ran)
+}
+
+// --- BackupConfig tests ---
+
+func TestBackupConfig_CreatesTimestampedBackup(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "test.conf")
+	content := []byte("original content")
+	require.NoError(t, os.WriteFile(path, content, 0o640))
+
+	backupPath, err := BackupConfig(path, 0, 1)
+	require.NoError(t, err)
+	assert.Contains(t, backupPath, "pre-migrate-0to1")
+
+	backupData, err := os.ReadFile(backupPath)
+	require.NoError(t, err)
+	assert.Equal(t, content, backupData)
+}
+
+// --- NATS token migration tests ---
+
+func TestNATSTokenMigration_UncommentsToken(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+authorization {
+#  token: "mytoken123"
+}
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0,
+		ToVersion:   1,
+		Description: "Enable NATS authorization token",
+		Run:         DefaultRegistry.configMigrations["nats.conf"][0].Run,
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	require.NoError(t, err)
+
+	data, err := os.ReadFile(confPath)
+	require.NoError(t, err)
+	assert.Contains(t, string(data), `  token: "mytoken123"`)
+	assert.NotContains(t, string(data), `#  token:`)
+}
+
+func TestNATSTokenMigration_ConvertsRoutes(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+cluster {
+  name: spinifex
+  listen: 10.11.12.1:4248
+
+  routes = [
+    "nats://10.11.12.2:4248",
+    "nats://10.11.12.3:4248",
+  ]
+}
+
+authorization {
+#  token: "mytoken123"
+}
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0,
+		ToVersion:   1,
+		Description: "Enable NATS authorization token",
+		Run:         DefaultRegistry.configMigrations["nats.conf"][0].Run,
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	require.NoError(t, err)
+
+	data, err := os.ReadFile(confPath)
+	require.NoError(t, err)
+	content := string(data)
+	assert.Contains(t, content, `  token: "mytoken123"`)
+	assert.Contains(t, content, `"nats-route://mytoken123@10.11.12.2:4248"`)
+	assert.Contains(t, content, `"nats-route://mytoken123@10.11.12.3:4248"`)
+	assert.NotContains(t, content, `"nats://10.11.12`)
+}
+
+func TestNATSTokenMigration_AlreadyUncommented(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+authorization {
+  token: "mytoken123"
+}
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0,
+		ToVersion:   1,
+		Description: "Enable NATS authorization token",
+		Run:         DefaultRegistry.configMigrations["nats.conf"][0].Run,
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	require.NoError(t, err)
+}
+
+func TestNATSTokenMigration_RewritesLogFilePath(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	// Simulates a v1.0.2-rendered nats.conf — the log_file path lives under the
+	// data dir, which the new systemd sandbox blocks writes to.
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+authorization {
+#  token: "mytoken123"
+}
+log_file: "/var/lib/spinifex/logs/nats.log"
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0,
+		ToVersion:   1,
+		Description: "Enable NATS auth token and relocate log_file",
+		Run:         DefaultRegistry.configMigrations["nats.conf"][0].Run,
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	require.NoError(t, err)
+
+	data, err := os.ReadFile(confPath)
+	require.NoError(t, err)
+	content := string(data)
+	assert.Contains(t, content, `log_file: "/var/log/spinifex/nats.log"`)
+	assert.NotContains(t, content, `"/var/lib/spinifex/logs/nats.log"`)
+	// Token rewrite still applied alongside the log_file rewrite.
+	assert.Contains(t, content, `  token: "mytoken123"`)
+}
+
+func TestNATSTokenMigration_ToleratesWhitespaceVariations(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+
+	tests := []struct {
+		name    string
+		input   string
+		wantTok string
+	}{
+		{"tab indented", "authorization {\n\t# token: \"tok1\"\n}", "\t token: \"tok1\""},
+		{"extra spaces", "authorization {\n   #   token: \"tok2\"\n}", "      token: \"tok2\""},
+	}
+
+	for i, tt := range tests {
+		confPath := filepath.Join(natsDir, "nats.conf")
+		require.NoError(t, os.WriteFile(confPath, []byte(tt.input), 0o640))
+
+		r := NewRegistry()
+		r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+		r.RegisterConfig("nats.conf", ConfigMigration{
+			FromVersion: 0,
+			ToVersion:   1,
+			Description: "test " + strconv.Itoa(i),
+			Run:         DefaultRegistry.configMigrations["nats.conf"][0].Run,
+		})
+
+		err := r.RunConfig("nats.conf", dir, dir)
+		require.NoError(t, err, tt.name)
+
+		data, err := os.ReadFile(confPath)
+		require.NoError(t, err)
+		assert.Contains(t, string(data), tt.wantTok, tt.name)
+	}
+}
+
+// --- PendingConfig tests ---
+
+func TestPendingConfig_ReturnsPending(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+authorization {
+#  token: "testtoken"
+}
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0,
+		ToVersion:   1,
+		Description: "Enable NATS authorization token",
+		Run:         func(ConfigContext) error { return nil },
+	})
+
+	pending, err := r.PendingConfig(dir)
+	require.NoError(t, err)
+	require.Len(t, pending, 1)
+	assert.Equal(t, "nats.conf", pending[0].Target)
+	assert.Equal(t, 0, pending[0].FromVersion)
+	assert.Equal(t, 1, pending[0].ToVersion)
+}
+
+func TestPendingConfig_NoPending(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# spinifex-config-version: 1
+# NATS Server Configuration
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0,
+		ToVersion:   1,
+		Description: "already done",
+		Run:         func(ConfigContext) error { return nil },
+	})
+
+	pending, err := r.PendingConfig(dir)
+	require.NoError(t, err)
+	assert.Empty(t, pending)
+}
+
+// --- Config migration error and chain tests ---
+
+func TestRunConfig_StopsOnFailure_VersionNotBumped(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+listen: 0.0.0.0:4222
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0,
+		ToVersion:   1,
+		Description: "fails",
+		Run:         func(ConfigContext) error { return errors.New("migration error") },
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "migration error")
+
+	// Version should remain at 0 (no version marker).
+	reader := &NATSConfVersionReader{}
+	v, err := reader.ReadVersion(confPath)
+	require.NoError(t, err)
+	assert.Equal(t, 0, v)
+}
+
+func TestRunConfig_ExecutesMultiStepChainInOrder(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+listen: 0.0.0.0:4222
+`), 0o640))
+
+	var order []int
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0, ToVersion: 1, Description: "step 1",
+		Run: func(ConfigContext) error { order = append(order, 1); return nil },
+	})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 1, ToVersion: 2, Description: "step 2",
+		Run: func(ConfigContext) error { order = append(order, 2); return nil },
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	require.NoError(t, err)
+	assert.Equal(t, []int{1, 2}, order)
+
+	// Verify final version is 2.
+	reader := &NATSConfVersionReader{}
+	v, err := reader.ReadVersion(confPath)
+	require.NoError(t, err)
+	assert.Equal(t, 2, v)
+}
+
+func TestRunConfig_RejectsChainGap(t *testing.T) {
+	dir := t.TempDir()
+	natsDir := filepath.Join(dir, "nats")
+	require.NoError(t, os.MkdirAll(natsDir, 0o755))
+	confPath := filepath.Join(natsDir, "nats.conf")
+	require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration
+listen: 0.0.0.0:4222
+`), 0o640))
+
+	r := NewRegistry()
+	r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 0, ToVersion: 1, Description: "step 1",
+		Run: func(ConfigContext) error { return nil },
+	})
+	r.RegisterConfig("nats.conf", ConfigMigration{
+		FromVersion: 3, ToVersion: 4, Description: "gap",
+		Run: func(ConfigContext) error { return nil },
+	})
+
+	err := r.RunConfig("nats.conf", dir, dir)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "gap")
+}

spinifex/migrate/version_readers.go

@@ -0,0 +1,128 @@
+package migrate
+
+import (
+	"fmt"
+	"os"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+// ConfigVersionReader reads and writes the schema version for a config file.
+type ConfigVersionReader interface {
+	ReadVersion(path string) (int, error)
+	WriteVersion(path string, version int) error
+}
+
+// TOMLVersionReader reads/writes the version field in TOML config files.
+// Handles both legacy string values ("1.0" → 1) and integer values (2 → 2).
+// New writes always use integers (version = 2).
+type TOMLVersionReader struct{}
+
+var tomlVersionRe = regexp.MustCompile(`(?m)^version\s*=\s*(?:"([^"]+)"|(\d+))\s*$`)
+
+func (r *TOMLVersionReader) ReadVersion(path string) (int, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return 0, err
+	}
+
+	match := tomlVersionRe.FindSubmatch(data)
+	if match == nil {
+		return 0, nil // no version field
+	}
+
+	var raw string
+	if len(match[1]) > 0 {
+		raw = string(match[1]) // quoted string like "1.0"
+	} else {
+		raw = string(match[2]) // bare integer like 2
+	}
+
+	// Handle "1.0" style — take the integer part before the dot.
+	if idx := strings.Index(raw, "."); idx >= 0 {
+		raw = raw[:idx]
+	}
+
+	v, err := strconv.Atoi(raw)
+	if err != nil {
+		return 0, fmt.Errorf("parse TOML version %q: %w", raw, err)
+	}
+	return v, nil
+}
+
+func (r *TOMLVersionReader) WriteVersion(path string, version int) error {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return err
+	}
+
+	replacement := fmt.Sprintf("version = %d", version)
+
+	if tomlVersionRe.Match(data) {
+		data = tomlVersionRe.ReplaceAll(data, []byte(replacement))
+	} else {
+		// Prepend version line if not present.
+		data = append([]byte(replacement+"\n"), data...)
+	}
+
+	info, err := os.Stat(path)
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(path, data, info.Mode())
+}
+
+// NATSConfVersionReader reads/writes the schema version for nats.conf files.
+// Version is tracked via a comment on the first line: # spinifex-config-version: 1
+// If absent, the file is treated as version 0 (pre-versioning).
+type NATSConfVersionReader struct{}
+
+var natsVersionRe = regexp.MustCompile(`^#\s*spinifex-config-version:\s*(\d+)\s*$`)
+
+func (r *NATSConfVersionReader) ReadVersion(path string) (int, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return 0, err
+	}
+
+	// Check first line only.
+	firstLine := strings.SplitN(string(data), "\n", 2)[0]
+	match := natsVersionRe.FindStringSubmatch(firstLine)
+	if match == nil {
+		return 0, nil // no version marker
+	}
+
+	v, err := strconv.Atoi(match[1])
+	if err != nil {
+		return 0, fmt.Errorf("parse NATS conf version %q: %w", match[1], err)
+	}
+	return v, nil
+}
+
+func (r *NATSConfVersionReader) WriteVersion(path string, version int) error {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return err
+	}
+
+	versionLine := fmt.Sprintf("# spinifex-config-version: %d", version)
+	content := string(data)
+
+	// Check if first line already has a version marker.
+	lines := strings.SplitN(content, "\n", 2)
+	if natsVersionRe.MatchString(lines[0]) {
+		// Replace existing version line.
+		lines[0] = versionLine
+		content = strings.Join(lines, "\n")
+	} else {
+		// Prepend version line.
+		content = versionLine + "\n" + content
+	}
+
+	info, err := os.Stat(path)
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(path, []byte(content), info.Mode())
+}

spinifex/services/nats/nats_test.go

@@ -4,6 +4,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"strings"
 	"testing"
 	"text/template"
 	"time"
@@ -86,3 +87,17 @@ func TestRenderedConfig_EnforcesAuth(t *testing.T) {
 		assert.True(t, nc.IsConnected())
 	})
 }
+
+// TestRenderedConfig_HasMigrationVersionMarker ensures the nats.conf template
+// stamps the current migration version on the first line. Without this marker,
+// the migration framework treats fresh installs as version 0 and tries to run
+// the 0→1 migration on next upgrade. The marker must be the literal first line
+// — NATSConfVersionReader only checks line 0.
+func TestRenderedConfig_HasMigrationVersionMarker(t *testing.T) {
+	raw, err := os.ReadFile(templatePath(t))
+	require.NoError(t, err)
+
+	firstLine := strings.SplitN(string(raw), "\n", 2)[0]
+	assert.Equal(t, "# spinifex-config-version: 1", firstLine,
+		"nats.conf template must start with the current migration version marker")
+}