feat(daemon): local-first instance state persistence (1a)

Make /var/lib/spinifex/state/instance-state.json the source of truth for a
node's VMs; JetStream KV becomes a best-effort cluster cache. The daemon now
survives KV unavailability without losing instance state.

- New local_state.go: atomic write (tmp+fsync+rename), schema_version gate,
  missing-file = fresh install, corrupt = refuse start.
- Daemon.WriteState writes the local file under Instances.Mu, then calls
  jsManager.WriteStateBestEffort with a 1s timeout. KV failures log and
  continue; clients never see "KV down" during partition.
- Daemon.LoadState reads from the local file only — no KV fallback (a fresh
  node has empty KV anyway, so silent fallback would mask bugs).
- restoreInstances now returns error and Start() aborts on corrupt local
  state rather than silently starting empty.
- jsManager.WriteState no longer self-locks (caller holds Mu); added
  WriteStateBestEffort and shared marshalInstanceState helper.

Reconnect-driven KV resync deferred to siv-16 (1c owns the callback wiring).

Bead: mulga-siv-15
Plan: docs/development/feature/daemon-local-state-persistence.md

spinifex/daemon/daemon.go

@@ -807,7 +807,9 @@ func (d *Daemon) Start() error {
 
 	d.waitForClusterReady()
 	d.upgradeJetStreamReplicas()
-	d.restoreInstances()
+	if err := d.restoreInstances(); err != nil {
+		return fmt.Errorf("restore instances: %w", err)
+	}
 
 	// Rebuild mgmt IP allocator from restored VMs so we don't re-allocate IPs
 	// that are already in use by running system instances.
@@ -1034,8 +1036,10 @@ func (d *Daemon) migrateTerminatedToKV(instance *vm.VM) bool {
 const maxConcurrentRecovery = 2
 
 // restoreInstances loads persisted VM state and re-launches instances that are
-// neither terminated nor flagged as user-stopped.
-func (d *Daemon) restoreInstances() {
+// neither terminated nor flagged as user-stopped. Returns an error only when
+// local state is unreadable (corrupt JSON or unknown schema_version) — those
+// are fatal because silently dropping instance state would orphan running VMs.
+func (d *Daemon) restoreInstances() error {
 	// Check for clean shutdown marker
 	cleanShutdown := false
 	if d.jsManager != nil {
@@ -1053,10 +1057,8 @@ func (d *Daemon) restoreInstances() {
 		time.Sleep(3 * time.Second)
 	}
 
-	err := d.LoadState()
-	if err != nil {
-		slog.Warn("Failed to load state, continuing with empty state", "error", err)
-		return
+	if err := d.LoadState(); err != nil {
+		return fmt.Errorf("load state: %w", err)
 	}
 
 	slog.Info("Loaded state", "instance count", len(d.Instances.VMS))
@@ -1258,6 +1260,7 @@ func (d *Daemon) restoreInstances() {
 	if err := d.WriteState(); err != nil {
 		slog.Error("Failed to persist state after restore", "error", err)
 	}
+	return nil
 }
 
 // isInstanceProcessRunning checks if the QEMU process for an instance is still alive.
@@ -1481,33 +1484,64 @@ func (d *Daemon) ClusterManager() error {
 	return nil
 }
 
-// WriteState writes the instance state to JetStream KV store (required).
-// It acquires d.Instances.Mu internally.
+// kvSyncTimeout bounds the best-effort cluster sync so a degraded NATS does
+// not stall every state transition. 1s is well above healthy KV.Put latency
+// and well below a user-visible delay.
+const kvSyncTimeout = time.Second
+
+// localStatePath returns the on-disk path to this daemon's instance state file.
+func (d *Daemon) localStatePath() string {
+	if d.config == nil {
+		return LocalStatePath("", "")
+	}
+	return LocalStatePath(d.config.DataDir, d.config.BaseDir)
+}
+
+// WriteState persists the instance state. Local file is the source of truth;
+// JetStream KV is best-effort cluster cache. The local write is fatal on
+// failure; KV failures are logged and swallowed so partition-time clients
+// never see "KV down" errors.
+//
+// Acquires d.Instances.Mu internally. Lock is held across the local write and
+// the bounded KV sync so the snapshot stays consistent.
 func (d *Daemon) WriteState() error {
-	if d.jsManager == nil {
-		return fmt.Errorf("JetStream manager not initialized - cannot write state")
+	d.Instances.Mu.Lock()
+	defer d.Instances.Mu.Unlock()
+
+	path := d.localStatePath()
+	if err := WriteLocalState(path, &d.Instances); err != nil {
+		slog.Error("Local state write failed", "path", path, "error", err)
+		return fmt.Errorf("write local state: %w", err)
 	}
-	if err := d.jsManager.WriteState(d.node, &d.Instances); err != nil {
-		slog.Error("JetStream write failed", "error", err)
-		return fmt.Errorf("failed to write state to JetStream: %w", err)
+
+	if d.jsManager != nil {
+		d.jsManager.WriteStateBestEffort(d.node, &d.Instances, kvSyncTimeout)
 	}
+
 	return nil
 }
 
-// LoadState loads the instance state from JetStream KV store (required)
+// LoadState loads the instance state from the local file. Missing file is the
+// fresh-install signal (start with empty map). Corrupt or unknown-schema files
+// are fatal — caller refuses start rather than silently losing data. There is
+// no KV fallback: a fresh node has empty KV anyway, so silent KV-fallback on
+// file loss would just mask bugs.
 func (d *Daemon) LoadState() error {
-	if d.jsManager == nil {
-		return fmt.Errorf("JetStream manager not initialized - cannot load state")
-	}
-
-	instances, err := d.jsManager.LoadState(d.node)
+	path := d.localStatePath()
+	state, err := ReadLocalState(path)
 	if err != nil {
-		slog.Error("JetStream load failed", "error", err)
-		return fmt.Errorf("failed to load state from JetStream: %w", err)
+		slog.Error("Local state load failed", "path", path, "error", err)
+		return fmt.Errorf("read local state: %w", err)
 	}
 
-	// Copy only the VMS map, not the mutex
-	d.Instances.VMS = instances.VMS
+	if state == nil {
+		d.Instances.VMS = make(map[string]*vm.VM)
+		slog.Info("No local state file, starting with empty instance map", "path", path)
+		return nil
+	}
+
+	d.Instances.VMS = state.VMS
+	slog.Info("Loaded local state", "path", path, "instances", len(state.VMS))
 	return nil
 }
 

spinifex/daemon/daemon_test.go

@@ -915,9 +915,10 @@ func TestDaemon_BootAllocation(t *testing.T) {
 	err = daemon.jsManager.InitKVBucket()
 	require.NoError(t, err)
 
-	// Pre-populate JetStream with test state
+	// Pre-populate the local state file with test state. Post-1a, LoadState
+	// reads from the local file (KV is best-effort cache only).
 	testInstances := &vm.Instances{VMS: vms}
-	err = daemon.jsManager.WriteState("node-1", testInstances)
+	err = WriteLocalState(daemon.localStatePath(), testInstances)
 	require.NoError(t, err)
 
 	// Manually trigger the LoadState and allocation logic normally found in Start()
@@ -3123,19 +3124,62 @@ func TestInstanceTypeMemoryMiB_NilSafety(t *testing.T) {
 }
 
 // --- Daemon.WriteState / Daemon.LoadState nil jsManager ---
+//
+// Post-1a: local file is the source of truth, KV is best-effort. A nil
+// jsManager is a valid configuration (e.g. standalone daemon, fresh install
+// before NATS is up) and must not block local persistence.
 
 func TestDaemon_WriteState_NilJSManager(t *testing.T) {
-	d := &Daemon{jsManager: nil}
-	err := d.WriteState()
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "JetStream manager not initialized")
+	tmpDir := t.TempDir()
+	d := &Daemon{
+		jsManager: nil,
+		config:    &config.Config{BaseDir: tmpDir},
+		Instances: vm.Instances{VMS: map[string]*vm.VM{"i-1": {ID: "i-1"}}},
+	}
+	require.NoError(t, d.WriteState())
+
+	state, err := ReadLocalState(LocalStatePath("", tmpDir))
+	require.NoError(t, err)
+	require.NotNil(t, state)
+	assert.Contains(t, state.VMS, "i-1")
 }
 
 func TestDaemon_LoadState_NilJSManager(t *testing.T) {
-	d := &Daemon{jsManager: nil}
+	tmpDir := t.TempDir()
+	require.NoError(t, WriteLocalState(LocalStatePath("", tmpDir), &vm.Instances{
+		VMS: map[string]*vm.VM{"i-seed": {ID: "i-seed"}},
+	}))
+
+	d := &Daemon{
+		jsManager: nil,
+		config:    &config.Config{BaseDir: tmpDir},
+	}
+	require.NoError(t, d.LoadState())
+	assert.Contains(t, d.Instances.VMS, "i-seed")
+}
+
+func TestDaemon_LoadState_MissingFileIsFreshInstall(t *testing.T) {
+	d := &Daemon{
+		jsManager: nil,
+		config:    &config.Config{BaseDir: t.TempDir()},
+	}
+	require.NoError(t, d.LoadState())
+	assert.Empty(t, d.Instances.VMS)
+}
+
+func TestDaemon_LoadState_CorruptFileFatal(t *testing.T) {
+	tmpDir := t.TempDir()
+	path := LocalStatePath("", tmpDir)
+	require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o750))
+	require.NoError(t, os.WriteFile(path, []byte("{not json"), 0o600))
+
+	d := &Daemon{
+		jsManager: nil,
+		config:    &config.Config{BaseDir: tmpDir},
+	}
 	err := d.LoadState()
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "JetStream manager not initialized")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "read local state")
 }
 
 // --- GetAvailableInstanceTypeInfos edge cases ---

spinifex/daemon/jetstream.go

@@ -382,23 +382,14 @@ func (m *JetStreamManager) WriteServiceManifest(nodeID string, services []string
 }
 
 // WriteState writes the instance state to the KV store for the given node.
-// It acquires instances.Mu internally.
+// Caller must hold instances.Mu — caller is the local-state writer and the lock
+// is needed across the local file write + KV sync to keep the snapshot consistent.
 func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) error {
-	instances.Mu.Lock()
-	defer instances.Mu.Unlock()
-
 	if m.kv == nil {
 		return errors.New("KV bucket not initialized")
 	}
 
-	// Create a struct without the mutex to avoid copying the lock
-	state := struct {
-		VMS map[string]*vm.VM `json:"vms"`
-	}{
-		VMS: instances.VMS,
-	}
-
-	jsonData, err := json.Marshal(state)
+	jsonData, err := marshalInstanceState(instances)
 	if err != nil {
 		return err
 	}
@@ -425,6 +416,59 @@ func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) er
 	return nil
 }
 
+// WriteStateBestEffort attempts to push instance state to KV with a deadline.
+// On timeout or error, it logs a warning and returns — never blocks the caller
+// past `timeout` and never returns an error. Used when the local state file is
+// the source of truth and KV is a best-effort cache.
+//
+// Caller must hold instances.Mu so the marshalled snapshot is consistent.
+//
+// Note: kv.Put has no context API. On timeout, the in-flight Put goroutine
+// continues and completes (or fails) on its own. This leaks at most one
+// goroutine per write; under sustained partition the leak is bounded by
+// WriteState call cadence (per-state-transition, not a tight loop).
+func (m *JetStreamManager) WriteStateBestEffort(nodeID string, instances *vm.Instances, timeout time.Duration) {
+	if m.kv == nil {
+		slog.Debug("KV bucket not initialized, skipping cluster sync", "node", nodeID)
+		return
+	}
+
+	jsonData, err := marshalInstanceState(instances)
+	if err != nil {
+		slog.Warn("KV sync skipped: marshal failed", "node", nodeID, "err", err)
+		return
+	}
+
+	key := InstanceStatePrefix + nodeID
+	done := make(chan error, 1)
+	go func() {
+		_, putErr := m.kv.Put(key, jsonData)
+		done <- putErr
+	}()
+
+	select {
+	case putErr := <-done:
+		if putErr != nil {
+			slog.Warn("KV sync failed (best-effort)", "key", key, "err", putErr)
+			return
+		}
+		slog.Debug("Wrote state to KV (best-effort)", "key", key, "instances", len(instances.VMS))
+	case <-time.After(timeout):
+		slog.Warn("KV sync timed out (best-effort)", "key", key, "timeout", timeout)
+	}
+}
+
+// marshalInstanceState produces the JSON wire form of instances without copying
+// the embedded sync.Mutex. Caller must hold instances.Mu.
+func marshalInstanceState(instances *vm.Instances) ([]byte, error) {
+	state := struct {
+		VMS map[string]*vm.VM `json:"vms"`
+	}{
+		VMS: instances.VMS,
+	}
+	return json.Marshal(state)
+}
+
 // LoadState loads the instance state from the KV store for the given node
 func (m *JetStreamManager) LoadState(nodeID string) (*vm.Instances, error) {
 	if m.kv == nil {

spinifex/daemon/local_state.go

@@ -0,0 +1,138 @@
+package daemon
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+
+	"github.com/mulgadc/spinifex/spinifex/vm"
+)
+
+const (
+	// LocalStateSchemaVersion is the on-disk schema version for instance-state.json.
+	// Bump on any breaking change to the LocalState shape; daemon refuses to start
+	// on an unknown version rather than silently losing data.
+	LocalStateSchemaVersion = 1
+
+	// DefaultLocalStateDir is the default directory under DataDir for local state files.
+	DefaultLocalStateDir = "state"
+
+	// LocalStateFileName is the on-disk filename for per-node instance state.
+	LocalStateFileName = "instance-state.json"
+
+	// DefaultDataDir is the fallback DataDir when config.DataDir is empty.
+	DefaultDataDir = "/var/lib/spinifex"
+)
+
+// LocalState is the on-disk representation of a node's instance state.
+// SchemaVersion gates compatibility; unknown versions are rejected.
+type LocalState struct {
+	SchemaVersion int               `json:"schema_version"`
+	VMS           map[string]*vm.VM `json:"vms"`
+}
+
+// LocalStatePath returns the absolute path to the per-node instance state file.
+// Resolution order: dataDir, then baseDir, then DefaultDataDir. The first
+// non-empty value wins. Production sets dataDir (/var/lib/spinifex); tests
+// typically set baseDir to a t.TempDir().
+func LocalStatePath(dataDir, baseDir string) string {
+	root := dataDir
+	if root == "" {
+		root = baseDir
+	}
+	if root == "" {
+		root = DefaultDataDir
+	}
+	return filepath.Join(root, DefaultLocalStateDir, LocalStateFileName)
+}
+
+// WriteLocalState atomically writes the instance state to path.
+// Caller must hold instances.Mu.
+//
+// Atomicity: marshal → write to <path>.tmp → fsync → rename. The rename is
+// atomic on POSIX so a concurrent reader sees either the old file or the new
+// one, never a half-written file.
+func WriteLocalState(path string, instances *vm.Instances) error {
+	state := LocalState{
+		SchemaVersion: LocalStateSchemaVersion,
+		VMS:           instances.VMS,
+	}
+
+	data, err := json.Marshal(state)
+	if err != nil {
+		return fmt.Errorf("marshal local state: %w", err)
+	}
+
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o750); err != nil {
+		return fmt.Errorf("mkdir %s: %w", dir, err)
+	}
+
+	tmp := path + ".tmp"
+	f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
+	if err != nil {
+		return fmt.Errorf("open tmp %s: %w", tmp, err)
+	}
+
+	if _, err := f.Write(data); err != nil {
+		f.Close()
+		os.Remove(tmp)
+		return fmt.Errorf("write tmp %s: %w", tmp, err)
+	}
+
+	if err := f.Sync(); err != nil {
+		f.Close()
+		os.Remove(tmp)
+		return fmt.Errorf("fsync tmp %s: %w", tmp, err)
+	}
+
+	if err := f.Close(); err != nil {
+		os.Remove(tmp)
+		return fmt.Errorf("close tmp %s: %w", tmp, err)
+	}
+
+	if err := os.Rename(tmp, path); err != nil {
+		os.Remove(tmp)
+		return fmt.Errorf("rename %s -> %s: %w", tmp, path, err)
+	}
+
+	return nil
+}
+
+// ReadLocalState reads the per-node instance state from path.
+//
+// Returns (nil, nil) if the file does not exist — fresh-install signal,
+// caller should start with an empty instance map.
+//
+// Returns an error on:
+//   - JSON parse failure (corruption — caller refuses start)
+//   - Unknown SchemaVersion (caller refuses start; safer than silent migration)
+func ReadLocalState(path string) (*LocalState, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			slog.Debug("Local state file missing, treating as fresh install", "path", path)
+			return nil, nil
+		}
+		return nil, fmt.Errorf("read local state %s: %w", path, err)
+	}
+
+	var state LocalState
+	if err := json.Unmarshal(data, &state); err != nil {
+		return nil, fmt.Errorf("parse local state %s: %w", path, err)
+	}
+
+	if state.SchemaVersion != LocalStateSchemaVersion {
+		return nil, fmt.Errorf("local state %s: unknown schema_version %d (expected %d)",
+			path, state.SchemaVersion, LocalStateSchemaVersion)
+	}
+
+	if state.VMS == nil {
+		state.VMS = make(map[string]*vm.VM)
+	}
+
+	return &state, nil
+}

spinifex/daemon/local_state_test.go

@@ -0,0 +1,122 @@
+package daemon
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLocalStatePath_PrefersDataDir(t *testing.T) {
+	got := LocalStatePath("/data", "/base")
+	assert.Equal(t, "/data/state/instance-state.json", got)
+}
+
+func TestLocalStatePath_FallsBackToBaseDir(t *testing.T) {
+	got := LocalStatePath("", "/base")
+	assert.Equal(t, "/base/state/instance-state.json", got)
+}
+
+func TestLocalStatePath_DefaultWhenBothEmpty(t *testing.T) {
+	got := LocalStatePath("", "")
+	assert.Equal(t, "/var/lib/spinifex/state/instance-state.json", got)
+}
+
+func TestWriteLocalState_RoundTrip(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state", "instance-state.json")
+
+	instances := &vm.Instances{
+		VMS: map[string]*vm.VM{
+			"i-aaa": {ID: "i-aaa", InstanceType: "t3.micro"},
+			"i-bbb": {ID: "i-bbb", InstanceType: "m5.large"},
+		},
+	}
+
+	require.NoError(t, WriteLocalState(path, instances))
+
+	loaded, err := ReadLocalState(path)
+	require.NoError(t, err)
+	require.NotNil(t, loaded)
+	assert.Equal(t, LocalStateSchemaVersion, loaded.SchemaVersion)
+	assert.Len(t, loaded.VMS, 2)
+	assert.Equal(t, "t3.micro", loaded.VMS["i-aaa"].InstanceType)
+	assert.Equal(t, "m5.large", loaded.VMS["i-bbb"].InstanceType)
+}
+
+func TestWriteLocalState_MkdirParent(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "deep", "nested", "state", "instance-state.json")
+	require.NoError(t, WriteLocalState(path, &vm.Instances{VMS: map[string]*vm.VM{}}))
+	_, err := os.Stat(path)
+	require.NoError(t, err)
+}
+
+func TestWriteLocalState_NoTmpLeftBehind(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "instance-state.json")
+	require.NoError(t, WriteLocalState(path, &vm.Instances{VMS: map[string]*vm.VM{}}))
+
+	entries, err := os.ReadDir(dir)
+	require.NoError(t, err)
+	for _, e := range entries {
+		assert.NotContains(t, e.Name(), ".tmp", "stale tmp file left behind")
+	}
+}
+
+func TestWriteLocalState_AtomicReplace(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "instance-state.json")
+
+	require.NoError(t, WriteLocalState(path, &vm.Instances{VMS: map[string]*vm.VM{
+		"i-1": {ID: "i-1"},
+	}}))
+
+	require.NoError(t, WriteLocalState(path, &vm.Instances{VMS: map[string]*vm.VM{
+		"i-2": {ID: "i-2"},
+	}}))
+
+	loaded, err := ReadLocalState(path)
+	require.NoError(t, err)
+	require.NotNil(t, loaded)
+	assert.Len(t, loaded.VMS, 1)
+	_, ok := loaded.VMS["i-2"]
+	assert.True(t, ok, "second write should replace, not merge")
+}
+
+func TestReadLocalState_Missing(t *testing.T) {
+	state, err := ReadLocalState(filepath.Join(t.TempDir(), "absent.json"))
+	require.NoError(t, err)
+	assert.Nil(t, state, "missing file is the fresh-install signal")
+}
+
+func TestReadLocalState_CorruptJSON(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "instance-state.json")
+	require.NoError(t, os.WriteFile(path, []byte("{not json"), 0o600))
+
+	_, err := ReadLocalState(path)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "parse local state")
+}
+
+func TestReadLocalState_UnknownSchema(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "instance-state.json")
+	data, _ := json.Marshal(map[string]any{"schema_version": 99, "vms": map[string]any{}})
+	require.NoError(t, os.WriteFile(path, data, 0o600))
+
+	_, err := ReadLocalState(path)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "unknown schema_version 99")
+}
+
+func TestReadLocalState_NilVMSMapInitialized(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "instance-state.json")
+	data, _ := json.Marshal(map[string]any{"schema_version": LocalStateSchemaVersion})
+	require.NoError(t, os.WriteFile(path, data, 0o600))
+
+	state, err := ReadLocalState(path)
+	require.NoError(t, err)
+	require.NotNil(t, state)
+	assert.NotNil(t, state.VMS)
+}

spinifex/daemon/state_test.go

@@ -573,7 +573,7 @@ func TestTransitionState_MultipleInstancesIndependent(t *testing.T) {
 func simulateCleanRestore(t *testing.T, daemon *Daemon) {
 	t.Helper()
 	require.NoError(t, daemon.jsManager.WriteShutdownMarker(daemon.node))
-	daemon.restoreInstances()
+	require.NoError(t, daemon.restoreInstances())
 }
 
 // TestRestoreInstances_StoppingFinalizedToStopped verifies that an instance