refactor(vm,daemon): post-extraction cleanup groups 1+2 (#194)

spinifex/daemon/daemon.go

@@ -49,7 +49,6 @@ import (
 	handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2"
 	"github.com/mulgadc/spinifex/spinifex/instancetypes"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
-	"github.com/mulgadc/spinifex/spinifex/qmp"
 	"github.com/mulgadc/spinifex/spinifex/tags"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/utils"
@@ -149,14 +148,19 @@ type Daemon struct {
 	// JetStream manager for KV state storage (nil if JetStream disabled)
 	jsManager *JetStreamManager
 
+	// stateStore is the vm.StateStore-shaped view over jsManager. Both the
+	// vm.Manager and daemon-side handlers route VM-instance state I/O
+	// through it. Initialized after initJetStream succeeds.
+	stateStore vm.StateStore
+
 	// Delay after QMP device_del before blockdev-del (default 1s, 0 in tests)
 	detachDelay time.Duration
 
 	// NATS connect retry options (nil uses defaults: 5min max, 500ms initial delay)
 	natsRetryOpts []utils.RetryOption
 
-	// NetworkPlumber handles tap device lifecycle for VPC networking
-	networkPlumber NetworkPlumber
+	// networkPlumber handles tap device lifecycle for VPC networking.
+	networkPlumber vm.NetworkPlumber
 
 	// Management NIC infrastructure: bridge IP + IP allocator for system instances.
 	// Populated at startup when br-mgmt is detected; nil/empty otherwise.
@@ -897,7 +901,7 @@ func (d *Daemon) Start() error {
 
 	d.waitForClusterReady()
 	d.upgradeJetStreamReplicas()
-	d.restoreInstances()
+	d.vmMgr.Restore()
 
 	// Rebuild mgmt IP allocator from restored VMs so we don't re-allocate IPs
 	// that are already in use by running system instances.
@@ -982,6 +986,8 @@ func (d *Daemon) initJetStream() error {
 		retryDelay = min(retryDelay*2, 10*time.Second)
 	}
 
+	d.stateStore = newStateStoreAdapter(d.jsManager)
+
 	// Replica upgrade is deferred to after all services have created their
 	// KV buckets and the cluster is ready (see upgradeJetStreamReplicas).
 
@@ -1094,19 +1100,6 @@ func (d *Daemon) checkPredastoreReady() bool {
 	return true
 }
 
-// restoreInstances delegates to vm.Manager.Restore. Daemon-side shim
-// preserved so existing tests keep their bare-method call shape; phase 2f
-// migrates them into vm/ and removes the wrapper.
-func (d *Daemon) restoreInstances() {
-	d.vmMgr.Restore()
-}
-
-// areVolumeSocketsValid is a daemon-side shim around vm.AreVolumeSocketsValid
-// kept so the existing TestAreVolumeSocketsValid_* cases call the same name.
-func (d *Daemon) areVolumeSocketsValid(instance *vm.VM) bool {
-	return vm.AreVolumeSocketsValid(instance)
-}
-
 // awaitShutdown blocks until the daemon's shutdown wait group completes.
 func (d *Daemon) awaitShutdown() {
 	done := make(chan struct{})
@@ -1201,7 +1194,7 @@ func (d *Daemon) ClusterManager() error {
 		}
 	})
 
-	// Load TLS certificate (C-5: serve over HTTPS instead of plaintext HTTP)
+	// Load TLS certificate.
 	// Resolve relative cert paths against config directory (cert lives alongside spinifex.toml).
 	// For binary installs, systemd sets absolute paths via env vars; for dev, the config
 	// stores relative paths like "config/server.pem" which need resolution.
@@ -1270,77 +1263,6 @@ func (d *Daemon) WriteState() error {
 	return nil
 }
 
-// LoadState loads the instance state from JetStream KV store (required).
-// Replaces the manager's running set with the loaded map.
-func (d *Daemon) LoadState() error {
-	if d.jsManager == nil {
-		return fmt.Errorf("JetStream manager not initialized - cannot load state")
-	}
-
-	loaded, err := d.jsManager.LoadState(d.node)
-	if err != nil {
-		slog.Error("JetStream load failed", "error", err)
-		return fmt.Errorf("failed to load state from JetStream: %w", err)
-	}
-
-	d.vmMgr.Replace(loaded)
-	return nil
-}
-
-func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId string) (*qmp.QMPResponse, error) {
-	// Confirm QMP client is initialized
-	if q == nil || q.Encoder == nil || q.Decoder == nil {
-		return nil, fmt.Errorf("QMP client is not initialized")
-	}
-
-	// Lock the QMP client
-	q.Mu.Lock()
-	defer q.Mu.Unlock()
-
-	// Set a read deadline so we don't block forever on a hung QEMU process
-	if err := q.Conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil {
-		return nil, fmt.Errorf("set read deadline: %w", err)
-	}
-	defer func() { _ = q.Conn.SetReadDeadline(time.Time{}) }() // clear deadline after command
-
-	if err := q.Encoder.Encode(cmd); err != nil {
-		return nil, fmt.Errorf("encode error: %w", err)
-	}
-
-	for {
-		var msg map[string]any
-		if err := q.Decoder.Decode(&msg); err != nil {
-			return nil, fmt.Errorf("decode error: %w", err)
-		}
-
-		if _, ok := msg["event"]; ok {
-			// QMP events are informational only — state transitions are driven
-			// by the command handlers that initiate the action, avoiding races
-			// between event-driven and command-driven transitions.
-			slog.Info("QMP event", "event", msg["event"], "instanceId", instanceId)
-			// Extend deadline after receiving an event (QEMU is alive, just chatty)
-			if err := q.Conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil {
-				return nil, fmt.Errorf("set read deadline: %w", err)
-			}
-			continue
-		}
-		if errObj, ok := msg["error"].(map[string]any); ok {
-			return nil, fmt.Errorf("QMP error: %s: %s", errObj["class"], errObj["desc"])
-		}
-		if _, ok := msg["return"]; ok {
-			respBytes, err := json.Marshal(msg)
-			if err != nil {
-				return nil, fmt.Errorf("marshal QMP response: %w", err)
-			}
-			var resp qmp.QMPResponse
-			if err := json.Unmarshal(respBytes, &resp); err != nil {
-				return nil, fmt.Errorf("unmarshal error: %w", err)
-			}
-			return &resp, nil
-		}
-	}
-}
-
 // setupReload registers a SIGHUP handler that reloads GPU config without restarting.
 func (d *Daemon) setupReload() {
 	d.shutdownWg.Go(func() {

spinifex/daemon/daemon_handlers_instance.go

@@ -72,8 +72,9 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
 		respondWithError(msg, awserrors.ErrorInvalidAMIIDNotFound)
 		return
 	}
-	// Verify the caller can use this AMI: must own it or it must be a system/pre-phase4 AMI.
-	// System AMIs have non-account-ID owner aliases (e.g. "self", "spinifex", empty).
+	// Verify the caller can use this AMI: must own it or the AMI must
+	// have a non-account-ID owner alias (e.g. "self", "spinifex", empty)
+	// indicating a system/legacy AMI.
 	amiOwner := amiMeta.ImageOwnerAlias
 	if amiOwner != "" && amiOwner != accountID {
 		if utils.IsAccountID(amiOwner) {
@@ -690,7 +691,8 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) {
 	d.vmMgr.View(func(vms map[string]*vm.VM) {
 		for _, instance := range vms {
 			// Skip instances not owned by the caller's account.
-			// Pre-Phase4 instances (empty AccountID) are only visible to root.
+			// Instances with an empty AccountID (legacy/migration data)
+			// are only visible to root.
 			if !isInstanceVisible(accountID, instance.AccountID) {
 				continue
 			}
@@ -837,14 +839,14 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) {
 		return
 	}
 
-	if d.jsManager == nil {
-		slog.Error("handleEC2StartStoppedInstance: JetStream not available")
+	if d.stateStore == nil {
+		slog.Error("handleEC2StartStoppedInstance: state store not available")
 		respondWithError(msg, awserrors.ErrorServerInternal)
 		return
 	}
 
 	// Load instance from shared KV
-	instance, err := d.jsManager.LoadStoppedInstance(req.InstanceID)
+	instance, err := d.stateStore.LoadStoppedInstance(req.InstanceID)
 	if err != nil {
 		slog.Error("handleEC2StartStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err)
 		respondWithError(msg, awserrors.ErrorServerInternal)
@@ -927,10 +929,10 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) {
 
 	// Remove from shared KV now that it's running locally.
 	// Retry once on failure — a stale KV entry risks duplicate starts.
-	if err := d.jsManager.DeleteStoppedInstance(req.InstanceID); err != nil {
+	if err := d.stateStore.DeleteStoppedInstance(req.InstanceID); err != nil {
 		slog.Warn("handleEC2StartStoppedInstance: first KV delete failed, retrying",
 			"instanceId", req.InstanceID, "err", err)
-		if retryErr := d.jsManager.DeleteStoppedInstance(req.InstanceID); retryErr != nil {
+		if retryErr := d.stateStore.DeleteStoppedInstance(req.InstanceID); retryErr != nil {
 			slog.Error("handleEC2StartStoppedInstance: KV delete failed after retry, instance is running locally but stale entry remains in shared KV",
 				"instanceId", req.InstanceID, "err", retryErr)
 		}
@@ -964,14 +966,14 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
 		return
 	}
 
-	if d.jsManager == nil {
-		slog.Error("handleEC2TerminateStoppedInstance: JetStream not available")
+	if d.stateStore == nil {
+		slog.Error("handleEC2TerminateStoppedInstance: state store not available")
 		respondWithError(msg, awserrors.ErrorServerInternal)
 		return
 	}
 
 	// Load instance from shared KV
-	instance, err := d.jsManager.LoadStoppedInstance(req.InstanceID)
+	instance, err := d.stateStore.LoadStoppedInstance(req.InstanceID)
 	if err != nil {
 		slog.Error("handleEC2TerminateStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err)
 		respondWithError(msg, awserrors.ErrorServerInternal)
@@ -1066,7 +1068,7 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
 	// Write to terminated KV bucket FIRST so the instance is visible in DescribeInstances.
 	// If this fails, the instance remains in the stopped bucket (safe to retry).
 	instance.Status = vm.StateTerminated
-	if err := d.jsManager.WriteTerminatedInstance(req.InstanceID, instance); err != nil {
+	if err := d.stateStore.WriteTerminatedInstance(req.InstanceID, instance); err != nil {
 		slog.Error("handleEC2TerminateStoppedInstance: failed to write to terminated KV, aborting", "instanceId", req.InstanceID, "err", err)
 		respondWithError(msg, awserrors.ErrorServerInternal)
 		return
@@ -1074,10 +1076,10 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
 
 	// Now safe to remove from shared stopped KV — instance already exists in terminated bucket.
 	// Retry once on failure to avoid duplicate entries in DescribeInstances.
-	if err := d.jsManager.DeleteStoppedInstance(req.InstanceID); err != nil {
+	if err := d.stateStore.DeleteStoppedInstance(req.InstanceID); err != nil {
 		slog.Warn("handleEC2TerminateStoppedInstance: first stopped KV delete failed, retrying",
 			"instanceId", req.InstanceID, "err", err)
-		if retryErr := d.jsManager.DeleteStoppedInstance(req.InstanceID); retryErr != nil {
+		if retryErr := d.stateStore.DeleteStoppedInstance(req.InstanceID); retryErr != nil {
 			slog.Error("handleEC2TerminateStoppedInstance: stopped KV delete failed after retry, instance may appear in both buckets",
 				"instanceId", req.InstanceID, "err", retryErr)
 		}
@@ -1092,22 +1094,25 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
 
 // handleEC2DescribeStoppedInstances returns stopped instances from shared KV.
 func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) {
-	d.describeInstancesFromKV(msg, d.jsManager.ListStoppedInstances, 80, "stopped", "handleEC2DescribeStoppedInstances")
+	if d.stateStore == nil {
+		respondWithError(msg, awserrors.ErrorServerInternal)
+		return
+	}
+	d.describeInstancesFromKV(msg, d.stateStore.ListStoppedInstances, 80, "stopped", "handleEC2DescribeStoppedInstances")
 }
 
 // handleEC2DescribeTerminatedInstances returns terminated instances from the terminated KV bucket.
 func (d *Daemon) handleEC2DescribeTerminatedInstances(msg *nats.Msg) {
-	d.describeInstancesFromKV(msg, d.jsManager.ListTerminatedInstances, 48, "terminated", "handleEC2DescribeTerminatedInstances")
+	if d.stateStore == nil {
+		respondWithError(msg, awserrors.ErrorServerInternal)
+		return
+	}
+	d.describeInstancesFromKV(msg, d.stateStore.ListTerminatedInstances, 48, "terminated", "handleEC2DescribeTerminatedInstances")
 }
 
 // describeInstancesFromKV is a shared helper for DescribeStopped/TerminatedInstances handlers.
 // It lists instances from a KV bucket, filters by account/instance ID, and responds with reservations.
 func (d *Daemon) describeInstancesFromKV(msg *nats.Msg, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, handlerName string) {
-	if d.jsManager == nil {
-		respondWithError(msg, awserrors.ErrorServerInternal)
-		return
-	}
-
 	accountID := utils.AccountIDFromMsg(msg)
 
 	describeInput := &ec2.DescribeInstancesInput{}
@@ -1227,13 +1232,13 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) {
 		return
 	}
 
-	if d.jsManager == nil {
-		slog.Error("handleEC2ModifyInstanceAttribute: JetStream not available")
+	if d.stateStore == nil {
+		slog.Error("handleEC2ModifyInstanceAttribute: state store not available")
 		respondWithError(msg, awserrors.ErrorServerInternal)
 		return
 	}
 
-	instance, err := d.jsManager.LoadStoppedInstance(instanceID)
+	instance, err := d.stateStore.LoadStoppedInstance(instanceID)
 	if err != nil {
 		slog.Error("handleEC2ModifyInstanceAttribute: failed to load stopped instance", "instanceId", instanceID, "err", err)
 		respondWithError(msg, awserrors.ErrorServerInternal)
@@ -1289,7 +1294,7 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) {
 		}
 	}
 
-	if err := d.jsManager.WriteStoppedInstance(instanceID, instance); err != nil {
+	if err := d.stateStore.WriteStoppedInstance(instanceID, instance); err != nil {
 		slog.Error("handleEC2ModifyInstanceAttribute: failed to write modified instance to KV",
 			"instanceId", instanceID, "err", err)
 		respondWithError(msg, awserrors.ErrorServerInternal)
@@ -1335,12 +1340,12 @@ func (d *Daemon) handleEC2DescribeInstanceAttribute(msg *nats.Msg) {
 	}
 
 	if instance == nil {
-		if d.jsManager == nil {
-			slog.Error("handleEC2DescribeInstanceAttribute: JetStream not available")
+		if d.stateStore == nil {
+			slog.Error("handleEC2DescribeInstanceAttribute: state store not available")
 			respondWithError(msg, awserrors.ErrorServerInternal)
 			return
 		}
-		stopped, err := d.jsManager.LoadStoppedInstance(instanceID)
+		stopped, err := d.stateStore.LoadStoppedInstance(instanceID)
 		if err != nil {
 			slog.Error("handleEC2DescribeInstanceAttribute: failed to load stopped instance",
 				"instanceId", instanceID, "err", err)

spinifex/daemon/daemon_handlers_test.go

@@ -86,6 +86,7 @@ func createFullTestDaemonWithJetStream(t *testing.T, natsURL string) *Daemon {
 	require.NoError(t, err)
 	err = daemon.jsManager.InitTerminatedInstanceBucket()
 	require.NoError(t, err)
+	daemon.stateStore = newStateStoreAdapter(daemon.jsManager)
 
 	return daemon
 }
@@ -660,7 +661,7 @@ func TestHandleEC2Events_RebootRunningInstance(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, reply)
 
-	// With nil QMPClient encoder/decoder, SendQMPCommand returns error,
+	// With nil QMPClient encoder/decoder, the QMP send returns error,
 	// so we expect an error response (ServerInternal).
 	var errResp map[string]any
 	err = json.Unmarshal(reply.Data, &errResp)

spinifex/daemon/daemon_handlers_volume.go

@@ -31,7 +31,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceComm
 	volumeID := command.AttachVolumeData.VolumeID
 
 	// Surface IncorrectInstanceState before any volume-side lookups so the
-	// AWS-SDK error matches the pre-2d ordering.
+	// caller sees the instance-state error first when both apply.
 	status := d.vmMgr.Status(instance)
 	if status != vm.StateRunning {
 		slog.Error("AttachVolume: instance not running",

spinifex/daemon/daemon_system_instance.go

@@ -243,7 +243,7 @@ func (d *Daemon) LaunchSystemInstance(input *handlers_elbv2.SystemInstanceInput)
 		slog.Warn("LaunchSystemInstance: failed to write state", "instanceId", instance.ID, "err", err)
 	}
 
-	// Subscribe to per-instance NATS topic for terminate commands
+	// Subscribe to per-instance NATS topic for terminate commands.
 	d.mu.Lock()
 	sub, subErr := d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events)
 	if subErr != nil {

spinifex/daemon/daemon_test.go

@@ -107,13 +107,12 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon {
 
 	// Wire the minimum vm.Deps that handler tests rely on. Lifecycle (Run/Start/
 	// Stop/Terminate) tests still set up their own deps; this gives the
-	// post-2d AttachVolume / DetachVolume manager methods enough plumbing to
-	// drive ebs.mount/unmount over NATS using the test's connection.
-	volState := newVolumeStateUpdaterAdapter(daemon.volumeService)
+	// AttachVolume / DetachVolume manager methods enough plumbing to drive
+	// ebs.mount/unmount over NATS using the test's connection.
 	daemon.vmMgr.SetDeps(vm.Deps{
 		NodeID:             daemon.node,
-		VolumeMounter:      newVolumeMounterAdapter(daemon.natsConn, daemon.node, volState),
-		VolumeStateUpdater: volState,
+		VolumeMounter:      newVolumeMounterAdapter(daemon.natsConn, daemon.node, daemon.volumeService),
+		VolumeStateUpdater: daemon.volumeService,
 		DetachDelay:        daemon.detachDelay,
 	})
 
@@ -960,9 +959,10 @@ func TestDaemon_BootAllocation(t *testing.T) {
 	err = daemon.jsManager.WriteState("node-1", vms)
 	require.NoError(t, err)
 
-	// Manually trigger the LoadState and allocation logic normally found in Start()
-	err = daemon.LoadState()
+	// Manually trigger the load and allocation logic normally found in Start()
+	loaded, err := daemon.jsManager.LoadState(daemon.node)
 	require.NoError(t, err)
+	daemon.vmMgr.Replace(loaded)
 
 	// Simulate the allocation loop in Start()
 	for _, instance := range daemon.vmMgr.Snapshot() {
@@ -2269,8 +2269,8 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 	})
 
 	t.Run("QMPDeviceDelFails_NoForce", func(t *testing.T) {
-		// With nil QMPClient encoder/decoder, SendQMPCommand returns error.
-		// Without force=true, this should return ServerInternal.
+		// With nil QMPClient encoder/decoder, the QMP device_del returns
+		// error. Without force=true, this should return ServerInternal.
 		command := types.EC2InstanceCommand{
 			ID: instanceID,
 			Attributes: types.EC2CommandAttributes{
@@ -2998,7 +2998,7 @@ func TestInstanceTypeMemoryMiB_NilSafety(t *testing.T) {
 	}))
 }
 
-// --- Daemon.WriteState / Daemon.LoadState nil jsManager ---
+// --- Daemon.WriteState nil jsManager ---
 
 func TestDaemon_WriteState_NilJSManager(t *testing.T) {
 	d := &Daemon{jsManager: nil}
@@ -3007,13 +3007,6 @@ func TestDaemon_WriteState_NilJSManager(t *testing.T) {
 	assert.Contains(t, err.Error(), "JetStream manager not initialized")
 }
 
-func TestDaemon_LoadState_NilJSManager(t *testing.T) {
-	d := &Daemon{jsManager: nil}
-	err := d.LoadState()
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "JetStream manager not initialized")
-}
-
 // --- GetAvailableInstanceTypeInfos edge cases ---
 
 func TestGetAvailableInstanceTypeInfos_Overcommitted(t *testing.T) {
@@ -3098,8 +3091,8 @@ func TestNewDaemon_WalDirPreservedIfSet(t *testing.T) {
 
 // TestVolumeMounterAdapter_UnmountOne_Success verifies that the adapter's
 // UnmountOne sends an ebs.unmount NATS request and handles a successful
-// response. UnmountOne is the post-2d successor to Daemon.rollbackEBSMount;
-// it is the AttachVolume rollback path and DetachVolume Phase 3.
+// response. UnmountOne is shared by the AttachVolume rollback path and the
+// DetachVolume ebs.unmount step.
 func TestVolumeMounterAdapter_UnmountOne_Success(t *testing.T) {
 	natsURL := sharedNATSURL
 

spinifex/daemon/network.go

@@ -26,24 +26,11 @@ var sudoCommand = func(name string, args ...string) *exec.Cmd {
 	return exec.Command("sudo", append([]string{name}, args...)...)
 }
 
-// NetworkPlumber handles tap device and OVS bridge operations for VM networking.
-// The live implementation runs system commands (ip, ovs-vsctl); tests use a mock.
-type NetworkPlumber interface {
-	// SetupTapDevice creates a tap device and adds it to the OVS br-int bridge
-	// with the correct iface-id for OVN port binding.
-	SetupTapDevice(eniId, mac string) error
-
-	// CleanupTapDevice removes the tap device from br-int and deletes it.
-	CleanupTapDevice(eniId string) error
-}
-
-// OVSNetworkPlumber implements NetworkPlumber using system commands.
+// OVSNetworkPlumber implements vm.NetworkPlumber using system commands
+// (ip, ovs-vsctl); tests use a mock satisfying the same interface.
 type OVSNetworkPlumber struct{}
 
-var (
-	_ NetworkPlumber    = (*OVSNetworkPlumber)(nil)
-	_ vm.NetworkPlumber = (*OVSNetworkPlumber)(nil)
-)
+var _ vm.NetworkPlumber = (*OVSNetworkPlumber)(nil)
 
 func (p *OVSNetworkPlumber) SetupTapDevice(eniId, mac string) error {
 	tapName := vm.TapDeviceName(eniId)

spinifex/daemon/network_test.go

@@ -44,7 +44,7 @@ type MockNetworkPlumber struct {
 	CleanupErr   error
 }
 
-var _ NetworkPlumber = (*MockNetworkPlumber)(nil)
+var _ vm.NetworkPlumber = (*MockNetworkPlumber)(nil)
 
 type mockSetupCall struct {
 	ENIId string

spinifex/daemon/ownership.go

@@ -10,14 +10,15 @@ import (
 
 // checkInstanceOwnership verifies the caller owns the instance.
 // Returns true if access is allowed, false if denied (error already sent).
-// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID).
+// Instances with an empty AccountID (legacy/migration data) are only
+// visible to root (GlobalAccountID).
 func checkInstanceOwnership(msg *nats.Msg, instanceID, ownerAccountID string) bool {
 	callerAccountID := utils.AccountIDFromMsg(msg)
 
-	// Pre-Phase4 instance: only root can access
+	// Untenanted instance: only root can access.
 	if ownerAccountID == "" {
 		if callerAccountID != utils.GlobalAccountID {
-			slog.Warn("Pre-Phase4 instance access denied (not root)",
+			slog.Warn("Untenanted instance access denied (not root)",
 				"instanceId", instanceID, "callerAccount", callerAccountID)
 			respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
 			return false
@@ -36,7 +37,8 @@ func checkInstanceOwnership(msg *nats.Msg, instanceID, ownerAccountID string) bo
 }
 
 // isInstanceVisible checks if the caller can see this instance (for Describe operations).
-// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID).
+// Instances with an empty AccountID (legacy/migration data) are only
+// visible to root (GlobalAccountID).
 func isInstanceVisible(callerAccountID, ownerAccountID string) bool {
 	if ownerAccountID == "" {
 		return callerAccountID == utils.GlobalAccountID
@@ -45,10 +47,10 @@ func isInstanceVisible(callerAccountID, ownerAccountID string) bool {
 }
 
 // volumeVisibleTo reports whether callerAccountID may operate on a volume with
-// the given tenantID. Pre-Phase4 volumes (empty tenantID) are root-only —
-// without this, the legacy short-circuit `tenantID != "" && tenantID != caller`
-// matched every caller and let any tenant attach a legacy/migration volume by
-// ID alone.
+// the given tenantID. Volumes with an empty tenantID (legacy/migration
+// data) are root-only — without this, the short-circuit
+// `tenantID != "" && tenantID != caller` matches every caller and lets any
+// tenant attach a legacy/migration volume by ID alone.
 func volumeVisibleTo(tenantID, callerAccountID string) bool {
 	if tenantID == "" {
 		return callerAccountID == utils.GlobalAccountID

spinifex/daemon/ownership_test.go

@@ -26,13 +26,13 @@ func TestIsInstanceVisible(t *testing.T) {
 			expected:        false,
 		},
 		{
-			name:            "pre-Phase4 instance visible to root",
+			name:            "untenanted instance visible to root",
 			callerAccountID: utils.GlobalAccountID,
 			ownerAccountID:  "",
 			expected:        true,
 		},
 		{
-			name:            "pre-Phase4 instance hidden from non-root",
+			name:            "untenanted instance hidden from non-root",
 			callerAccountID: "123456789012",
 			ownerAccountID:  "",
 			expected:        false,
@@ -82,19 +82,19 @@ func TestVolumeVisibleTo(t *testing.T) {
 			expected:        false,
 		},
 		{
-			name:            "pre-Phase4 volume visible to root",
+			name:            "untenanted volume visible to root",
 			tenantID:        "",
 			callerAccountID: utils.GlobalAccountID,
 			expected:        true,
 		},
 		{
-			name:            "pre-Phase4 volume hidden from non-root tenant",
+			name:            "untenanted volume hidden from non-root tenant",
 			tenantID:        "",
 			callerAccountID: "123456789012",
 			expected:        false,
 		},
 		{
-			name:            "pre-Phase4 volume denied to empty caller",
+			name:            "untenanted volume denied to empty caller",
 			tenantID:        "",
 			callerAccountID: "",
 			expected:        false,

spinifex/daemon/state_test.go

@@ -40,6 +40,7 @@ func createDaemonWithJetStream(t *testing.T) *Daemon {
 	require.NoError(t, daemon.jsManager.InitKVBucket())
 	require.NoError(t, daemon.jsManager.InitClusterStateBucket())
 	require.NoError(t, daemon.jsManager.InitTerminatedInstanceBucket())
+	daemon.stateStore = newStateStoreAdapter(daemon.jsManager)
 
 	// Wire just enough vm.Deps for manager-driven state operations to work
 	// (migrate, MarkFailed, Restore classification). Full wiring (network
@@ -48,7 +49,7 @@ func createDaemonWithJetStream(t *testing.T) *Daemon {
 	// a real VM lifecycle.
 	daemon.vmMgr.SetDeps(vm.Deps{
 		NodeID:                     daemon.node,
-		StateStore:                 newStateStoreAdapter(daemon.jsManager),
+		StateStore:                 daemon.stateStore,
 		TransitionState:            daemon.TransitionState,
 		InstanceTypes:              newInstanceTypeResolverAdapter(daemon.resourceMgr),
 		Resources:                  newResourceControllerAdapter(daemon.resourceMgr),
@@ -569,17 +570,17 @@ func TestTransitionState_MultipleInstancesIndependent(t *testing.T) {
 // --- Recovery / restoreInstances tests ---
 //
 // These tests simulate a daemon restart by writing instance state to JetStream,
-// then calling restoreInstances on a fresh daemon. Since no QEMU process is
+// then calling vmMgr.Restore() on a fresh daemon. Since no QEMU process is
 // running, isInstanceProcessRunning returns false for all instances, which
 // exercises the recovery state resolution logic.
 
-// simulateCleanRestore writes a clean shutdown marker then calls restoreInstances.
-// Without the marker, restoreInstances sleeps 3s waiting for stale QEMU PIDs —
-// unnecessary in tests since no QEMU process ever runs.
+// simulateCleanRestore writes a clean shutdown marker then drives the
+// manager's restore path. Without the marker, Restore sleeps 3s waiting for
+// stale QEMU PIDs — unnecessary in tests since no QEMU process ever runs.
 func simulateCleanRestore(t *testing.T, daemon *Daemon) {
 	t.Helper()
 	require.NoError(t, daemon.jsManager.WriteShutdownMarker(daemon.node))
-	daemon.restoreInstances()
+	daemon.vmMgr.Restore()
 }
 
 // TestRestoreInstances_StoppingFinalizedToStopped verifies that an instance
@@ -905,7 +906,9 @@ func TestStatePersistence_RoundTrip(t *testing.T) {
 
 	// Simulate restart: clear and reload.
 	daemon.vmMgr.Replace(map[string]*vm.VM{})
-	require.NoError(t, daemon.LoadState())
+	snapshot, err := daemon.jsManager.LoadState(daemon.node)
+	require.NoError(t, err)
+	daemon.vmMgr.Replace(snapshot)
 
 	loaded, _ := daemon.vmMgr.Get("i-roundtrip")
 	require.NotNil(t, loaded)
@@ -1037,9 +1040,6 @@ func TestPendingWatchdog_MarksStuckInstanceFailed(t *testing.T) {
 }
 
 func TestAreVolumeSocketsValid_ValidSocket(t *testing.T) {
-	daemon := createDaemonWithJetStream(t)
-
-	// Create a real Unix socket listener
 	sockPath := filepath.Join(t.TempDir(), "test.sock")
 	listener, err := net.Listen("unix", sockPath)
 	require.NoError(t, err)
@@ -1053,12 +1053,10 @@ func TestAreVolumeSocketsValid_ValidSocket(t *testing.T) {
 			},
 		},
 	}
-	assert.True(t, daemon.areVolumeSocketsValid(instance))
+	assert.True(t, vm.AreVolumeSocketsValid(instance))
 }
 
 func TestAreVolumeSocketsValid_MissingSocket(t *testing.T) {
-	daemon := createDaemonWithJetStream(t)
-
 	instance := &vm.VM{
 		ID: "i-missing-sock",
 		EBSRequests: types.EBSRequests{
@@ -1067,18 +1065,16 @@ func TestAreVolumeSocketsValid_MissingSocket(t *testing.T) {
 			},
 		},
 	}
-	assert.False(t, daemon.areVolumeSocketsValid(instance))
+	assert.False(t, vm.AreVolumeSocketsValid(instance))
 }
 
 func TestAreVolumeSocketsValid_StaleSocketFile(t *testing.T) {
-	daemon := createDaemonWithJetStream(t)
-
-	// Create a socket file with no listener — simulates viperblock crash
-	// leaving a stale socket on disk
+	// Socket file with no listener — simulates viperblock crash leaving a
+	// stale socket on disk. os.Stat would return true here, but dial fails.
 	sockPath := filepath.Join(t.TempDir(), "stale.sock")
 	listener, err := net.Listen("unix", sockPath)
 	require.NoError(t, err)
-	listener.Close() // close listener but leave file on disk
+	listener.Close()
 
 	instance := &vm.VM{
 		ID: "i-stale-sock",
@@ -1088,23 +1084,19 @@ func TestAreVolumeSocketsValid_StaleSocketFile(t *testing.T) {
 			},
 		},
 	}
-	// os.Stat would return true here, but dial should fail
-	assert.False(t, daemon.areVolumeSocketsValid(instance))
+	assert.False(t, vm.AreVolumeSocketsValid(instance))
 }
 
 func TestAreVolumeSocketsValid_EmptyRequests(t *testing.T) {
-	daemon := createDaemonWithJetStream(t)
-
 	instance := &vm.VM{
 		ID:          "i-no-vols",
 		EBSRequests: types.EBSRequests{},
 	}
-	assert.True(t, daemon.areVolumeSocketsValid(instance))
+	assert.True(t, vm.AreVolumeSocketsValid(instance))
 }
 
 func TestAreVolumeSocketsValid_TCPTransport(t *testing.T) {
-	daemon := createDaemonWithJetStream(t)
-
+	// TCP transport can't be validated locally — should return true.
 	instance := &vm.VM{
 		ID: "i-tcp",
 		EBSRequests: types.EBSRequests{
@@ -1113,8 +1105,7 @@ func TestAreVolumeSocketsValid_TCPTransport(t *testing.T) {
 			},
 		},
 	}
-	// TCP transport can't be validated locally — should return true
-	assert.True(t, daemon.areVolumeSocketsValid(instance))
+	assert.True(t, vm.AreVolumeSocketsValid(instance))
 }
 
 func TestMarkInstanceFailed_AlreadyShuttingDown(t *testing.T) {
@@ -1168,10 +1159,12 @@ func TestRestoreInstances_ProvisioningInstanceLaunchTimeReset(t *testing.T) {
 	before := time.Now()
 	simulateCleanRestore(t, daemon)
 
-	// LoadState inside restoreInstances creates fresh objects, so look up
-	// from the map. The instance may have been removed by markInstanceFailed
-	// (launch fails in test env), so load directly from JetStream state.
-	require.NoError(t, daemon.LoadState())
+	// Restore creates fresh objects, so look up from the map. The instance
+	// may have been removed by markInstanceFailed (launch fails in test
+	// env), so reload directly from JetStream state.
+	loaded, err := daemon.jsManager.LoadState(daemon.node)
+	require.NoError(t, err)
+	daemon.vmMgr.Replace(loaded)
 	instance, _ := daemon.vmMgr.Get("i-provisioning")
 	if instance == nil {
 		// Instance was cleaned up by markInstanceFailed — that's fine.

spinifex/daemon/vm_adapters.go

@@ -2,17 +2,14 @@ package daemon
 
 import (
 	"encoding/json"
-	"errors"
 	"fmt"
 	"log/slog"
-	"os/exec"
 	"strings"
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
-	"github.com/mulgadc/spinifex/spinifex/qmp"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go"
@@ -188,8 +185,8 @@ func (a *volumeMounterAdapter) MountOne(req *types.EBSRequest) error {
 }
 
 // UnmountOne sends ebs.unmount for a single request. Best-effort: errors
-// are logged. Mirrors the pre-2d Daemon.rollbackEBSMount semantics so
-// AttachVolume rollback and DetachVolume Phase 3 share one code path.
+// are logged. Shared by AttachVolume rollback and the DetachVolume
+// ebs.unmount step.
 func (a *volumeMounterAdapter) UnmountOne(req types.EBSRequest) {
 	payload, err := json.Marshal(req)
 	if err != nil {
@@ -221,86 +218,6 @@ func (a *volumeMounterAdapter) UnmountOne(req types.EBSRequest) {
 	slog.Info("UnmountOne: volume unmounted successfully", "volume", req.Name)
 }
 
-// qmpClientFactoryAdapter satisfies vm.QMPClientFactory. It performs only
-// the connect + qmp_capabilities handshake; the manager owns starting the
-// heartbeat goroutine on the returned client.
-type qmpClientFactoryAdapter struct{}
-
-var _ vm.QMPClientFactory = (*qmpClientFactoryAdapter)(nil)
-
-func newQMPClientFactoryAdapter() *qmpClientFactoryAdapter { return &qmpClientFactoryAdapter{} }
-
-func (a *qmpClientFactoryAdapter) Create(v *vm.VM) (*qmp.QMPClient, error) {
-	client, err := qmp.NewQMPClient(v.Config.QMPSocket)
-	if err != nil {
-		return nil, fmt.Errorf("connect QMP socket %s: %w", v.Config.QMPSocket, err)
-	}
-
-	if _, err := sendQMPHandshake(client, v.ID); err != nil {
-		_ = client.Conn.Close()
-		return nil, err
-	}
-
-	return client, nil
-}
-
-// sendQMPHandshake issues qmp_capabilities under the QMPClient's mutex. Mirrors
-// the slim version of Daemon.SendQMPCommand for the handshake; the daemon's
-// full helper still owns long-running command dispatch.
-func sendQMPHandshake(client *qmp.QMPClient, instanceID string) (*qmp.QMPResponse, error) {
-	if client == nil || client.Encoder == nil || client.Decoder == nil {
-		return nil, errors.New("QMP client is not initialized")
-	}
-
-	client.Mu.Lock()
-	defer client.Mu.Unlock()
-
-	if err := client.Conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil {
-		return nil, fmt.Errorf("set read deadline: %w", err)
-	}
-	defer func() { _ = client.Conn.SetReadDeadline(time.Time{}) }()
-
-	if err := client.Encoder.Encode(qmp.QMPCommand{Execute: "qmp_capabilities"}); err != nil {
-		return nil, fmt.Errorf("encode qmp_capabilities: %w", err)
-	}
-
-	for {
-		var msg map[string]any
-		if err := client.Decoder.Decode(&msg); err != nil {
-			return nil, fmt.Errorf("decode qmp_capabilities response: %w", err)
-		}
-		if _, ok := msg["event"]; ok {
-			continue
-		}
-		raw, err := json.Marshal(msg)
-		if err != nil {
-			return nil, fmt.Errorf("marshal qmp response: %w", err)
-		}
-		var resp qmp.QMPResponse
-		if err := json.Unmarshal(raw, &resp); err != nil {
-			return nil, fmt.Errorf("unmarshal qmp response: %w", err)
-		}
-		if resp.Error != nil {
-			return nil, fmt.Errorf("qmp_capabilities failed: %s", resp.Error.Desc)
-		}
-		slog.Debug("QMP handshake complete", "instance", instanceID)
-		return &resp, nil
-	}
-}
-
-// processLauncherAdapter satisfies vm.ProcessLauncher. The real
-// implementation is a 1:1 wrapper over Config.Execute; the seam exists so
-// tests can substitute scripted commands.
-type processLauncherAdapter struct{}
-
-var _ vm.ProcessLauncher = (*processLauncherAdapter)(nil)
-
-func newProcessLauncherAdapter() *processLauncherAdapter { return &processLauncherAdapter{} }
-
-func (a *processLauncherAdapter) Launch(cfg *vm.Config) (*exec.Cmd, error) {
-	return cfg.Execute()
-}
-
 // instanceTypeResolverAdapter satisfies vm.InstanceTypeResolver. It looks up
 // the EC2 InstanceTypeInfo on the daemon's ResourceManager and projects the
 // QEMU-relevant numbers (vCPU, memory MiB, architecture) into the
@@ -368,37 +285,13 @@ func (a *resourceControllerAdapter) CanAllocate(instanceType string, count int)
 	return a.rm.canAllocate(it, count)
 }
 
-// volumeStateUpdaterAdapter satisfies vm.VolumeStateUpdater by delegating to
-// the daemon's volume service.
-type volumeStateUpdaterAdapter struct {
-	svc volumeStateUpdater
-}
-
-// volumeStateUpdater is the narrow slice of handlers_ec2_volume.VolumeServiceImpl
-// that the manager touches. Defining it locally avoids dragging the full
-// volume-service surface into the adapter.
-type volumeStateUpdater interface {
-	UpdateVolumeState(volumeID, state, instanceID, attachmentDevice string) error
-}
-
-var _ vm.VolumeStateUpdater = (*volumeStateUpdaterAdapter)(nil)
-
-func newVolumeStateUpdaterAdapter(svc volumeStateUpdater) *volumeStateUpdaterAdapter {
-	return &volumeStateUpdaterAdapter{svc: svc}
-}
-
-func (a *volumeStateUpdaterAdapter) UpdateVolumeState(volumeID, state, instanceID, attachmentDevice string) error {
-	return a.svc.UpdateVolumeState(volumeID, state, instanceID, attachmentDevice)
-}
-
 // onInstanceRecoveringHook returns the daemon's OnInstanceRecovering
 // callback. Restore fires it once per instance about to be relaunched so
 // concurrent terminate commands can land on this node before launch
 // completes — without it, ec2.cmd.<id> would only be subscribed by
 // onInstanceUpHook after launch success, leaving a window where the
 // instance is reachable by DescribeInstances (in StatePending) but not
-// by EC2 commands. Mirrors the pre-2e early-subscribe block in
-// daemon.restoreInstances.
+// by EC2 commands.
 func (d *Daemon) onInstanceRecoveringHook() func(*vm.VM) {
 	return func(instance *vm.VM) {
 		d.mu.Lock()
@@ -520,17 +413,14 @@ func (d *Daemon) onInstanceDownHook() func(string) {
 // All collaborators must already be initialized; callers are expected to
 // invoke this from Daemon.Start after services and JetStream are ready.
 func (d *Daemon) buildVMManagerDeps() vm.Deps {
-	volState := newVolumeStateUpdaterAdapter(d.volumeService)
 	return vm.Deps{
 		NodeID:             d.node,
-		StateStore:         newStateStoreAdapter(d.jsManager),
-		VolumeMounter:      newVolumeMounterAdapter(d.natsConn, d.node, volState),
-		QMPClientFactory:   newQMPClientFactoryAdapter(),
-		ProcessLauncher:    newProcessLauncherAdapter(),
+		StateStore:         d.stateStore,
+		VolumeMounter:      newVolumeMounterAdapter(d.natsConn, d.node, d.volumeService),
 		NetworkPlumber:     d.networkPlumber,
 		InstanceTypes:      newInstanceTypeResolverAdapter(d.resourceMgr),
 		Resources:          newResourceControllerAdapter(d.resourceMgr),
-		VolumeStateUpdater: volState,
+		VolumeStateUpdater: d.volumeService,
 		InstanceCleaner:    newInstanceCleanerAdapter(d),
 		Hooks: vm.ManagerHooks{
 			OnInstanceUp:         d.onInstanceUpHook(),
@@ -563,8 +453,7 @@ func newInstanceCleanerAdapter(d *Daemon) *instanceCleanerAdapter {
 
 // DeleteVolumes deletes EFI / cloud-init internal volumes via ebs.delete
 // and user volumes flagged DeleteOnTermination via the volume service.
-// Errors are logged per volume; partial failure is tolerated to match
-// pre-2c stopInstance behaviour.
+// Errors are logged per volume; partial failure is tolerated.
 func (a *instanceCleanerAdapter) DeleteVolumes(instance *vm.VM) {
 	instance.EBSRequests.Mu.Lock()
 	defer instance.EBSRequests.Mu.Unlock()

spinifex/handlers/ec2/volume/service_impl.go

@@ -22,6 +22,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/mulgadc/viperblock/viperblock"
 	s3backend "github.com/mulgadc/viperblock/viperblock/backends/s3"
 	"github.com/nats-io/nats.go"
@@ -32,6 +33,10 @@ const defaultGP3IOPS = 3000
 // Ensure VolumeServiceImpl implements VolumeService
 var _ VolumeService = (*VolumeServiceImpl)(nil)
 
+// Ensure VolumeServiceImpl satisfies vm.VolumeStateUpdater so the manager
+// can call UpdateVolumeState directly without a daemon-side adapter.
+var _ vm.VolumeStateUpdater = (*VolumeServiceImpl)(nil)
+
 // VolumeServiceImpl handles EBS volume operations with S3 storage
 type VolumeServiceImpl struct {
 	config     *config.Config

spinifex/vm/crash_recovery.go

@@ -110,8 +110,6 @@ func (m *Manager) HandleCrash(instance *VM, waitErr error) {
 		if v.Health.FirstCrashTime.IsZero() {
 			v.Health.FirstCrashTime = now
 		}
-		v.Running = false
-		v.PID = 0
 	})
 
 	if m.deps.Resources != nil && instance.InstanceType != "" {

spinifex/vm/crash_recovery_test.go

@@ -128,8 +128,6 @@ func TestHandleCrash_CoreFlow(t *testing.T) {
 	instance := &VM{
 		ID:           "i-core",
 		Status:       StateRunning,
-		Running:      true,
-		PID:          12345,
 		InstanceType: "t3.micro",
 		Config:       Config{QMPSocket: qmpPath},
 	}
@@ -147,9 +145,6 @@ func TestHandleCrash_CoreFlow(t *testing.T) {
 	assert.Equal(t, "unknown", instance.Health.LastCrashReason)
 	assert.False(t, instance.Health.FirstCrashTime.IsZero())
 
-	assert.False(t, instance.Running, "Running must be cleared")
-	assert.Zero(t, instance.PID, "PID must be cleared")
-
 	assert.Equal(t, 1, rc.deallocateCount("t3.micro"), "crashed instance resources must be deallocated")
 
 	_, err := os.Stat(qmpPath)
@@ -162,8 +157,6 @@ func TestHandleCrash_FirstCrashSetsTime(t *testing.T) {
 	instance := &VM{
 		ID:           "i-firstcrash",
 		Status:       StateRunning,
-		Running:      true,
-		PID:          111,
 		InstanceType: "t3.micro",
 	}
 	m.Insert(instance)
@@ -175,8 +168,6 @@ func TestHandleCrash_FirstCrashSetsTime(t *testing.T) {
 
 	m.UpdateState(instance.ID, func(v *VM) {
 		v.Status = StateRunning
-		v.Running = true
-		v.PID = 222
 	})
 
 	time.Sleep(time.Millisecond)
@@ -188,17 +179,15 @@ func TestHandleCrash_FirstCrashSetsTime(t *testing.T) {
 }
 
 // TestHandleCrash_UnknownInstanceType verifies a crash on a VM with an
-// instance type not in the resolver still completes the state transition,
-// health bookkeeping, and pid/running clear — even though MaybeRestart will
-// later refuse to restart it.
+// instance type not in the resolver still completes the state transition
+// and health bookkeeping — even though MaybeRestart will later refuse to
+// restart it.
 func TestHandleCrash_UnknownInstanceType(t *testing.T) {
 	m, _, _, _ := crashTestManager(t)
 
 	instance := &VM{
 		ID:           "i-unknown",
 		Status:       StateRunning,
-		Running:      true,
-		PID:          333,
 		InstanceType: "z99.nonexistent",
 	}
 	m.Insert(instance)
@@ -207,8 +196,6 @@ func TestHandleCrash_UnknownInstanceType(t *testing.T) {
 
 	assert.Equal(t, StateError, m.Status(instance))
 	assert.Equal(t, 1, instance.Health.CrashCount)
-	assert.False(t, instance.Running)
-	assert.Zero(t, instance.PID)
 }
 
 func TestMaybeRestart_ExceedsMaxInWindow(t *testing.T) {

spinifex/vm/errors.go

@@ -30,6 +30,5 @@ var ErrVolumeDeviceMismatch = errors.New("volume device name mismatch")
 // ErrMountAmbiguous is returned by VolumeMounter.MountOne when the
 // backend acknowledged the mount but returned an empty URI. The mount may
 // or may not have started serving NBD; AttachVolume must invoke UnmountOne
-// defensively so a half-started backend mount is not orphaned. Mirrors the
-// pre-2d daemon-side rollback that triggered on this exact response shape.
+// defensively so a half-started backend mount is not orphaned.
 var ErrMountAmbiguous = errors.New("ebs.mount succeeded with empty URI")

spinifex/vm/lifecycle.go

@@ -72,8 +72,7 @@ func (m *Manager) launchStillValid(instance *VM) bool {
 }
 
 // launch is the orchestrator: pid check, mount volumes, exec QEMU, attach
-// QMP, fire OnInstanceUp, transition to Running. Mirrors the pre-2b
-// Daemon.LaunchInstance flow.
+// QMP, fire OnInstanceUp, transition to Running.
 func (m *Manager) launch(instance *VM) error {
 	if !m.launchStillValid(instance) {
 		return nil
@@ -109,7 +108,7 @@ func (m *Manager) launch(instance *VM) error {
 		return err
 	}
 
-	qmpClient, err := m.deps.QMPClientFactory.Create(instance)
+	qmpClient, err := newQMPClientWithHandshake(instance)
 	if err != nil {
 		slog.Error("Failed to create QMP client", "err", err)
 		return err
@@ -149,8 +148,7 @@ func (m *Manager) launch(instance *VM) error {
 
 	if m.deps.Hooks.OnInstanceUp != nil {
 		// Launch path: per-instance subscribe failures are logged and the
-		// launch still succeeds, mirroring the pre-2e early-subscribe block
-		// in handleEC2RunInstances. The instance is reachable via cluster
+		// launch still succeeds. The instance is reachable via cluster
 		// fan-out (DescribeInstances) and the next OnInstanceUp on a
 		// state-touching event will reinstall the subs idempotently.
 		if err := m.deps.Hooks.OnInstanceUp(instance); err != nil {
@@ -163,7 +161,7 @@ func (m *Manager) launch(instance *VM) error {
 }
 
 // startQEMU launches the QEMU process for instance and waits for startup to
-// confirm. Mirrors the pre-2b Daemon.StartInstance flow.
+// confirm.
 func (m *Manager) startQEMU(instance *VM) error {
 	pidFile, err := utils.GeneratePidFile(instance.ID)
 	if err != nil {
@@ -281,7 +279,7 @@ func (m *Manager) startQEMU(instance *VM) error {
 	startupConfirmed := make(chan bool, 1)
 
 	go func() {
-		cmd, err := m.deps.ProcessLauncher.Launch(&instance.Config)
+		cmd, err := instance.Config.Execute()
 		if err != nil {
 			slog.Error("Failed to execute VM", "err", err)
 			processChan <- 0
@@ -384,7 +382,6 @@ func (m *Manager) startQEMU(instance *VM) error {
 }
 
 // appendDevHostfwdNIC adds a user-mode NIC with SSH hostfwd for dev access.
-// Mirrors the pre-2b inline DEV_NETWORKING block in StartInstance.
 func (m *Manager) appendDevHostfwdNIC(instance *VM) {
 	sshDebugAddr, err := viperblock.FindFreePort()
 	if err != nil {
@@ -436,10 +433,10 @@ func (m *Manager) appendDevHostfwdNIC(instance *VM) {
 
 // AttachQMP connects a QMP client to a QEMU process that already exists
 // (e.g. a daemon restart finding a still-running QEMU) and starts the
-// heartbeat goroutine. The launch path uses QMPClientFactory + qmpHeartbeat
-// directly; AttachQMP is the matching seam for reconnect callers.
+// heartbeat goroutine. AttachQMP is the matching seam for reconnect
+// callers; the launch path calls the same helper inline.
 func (m *Manager) AttachQMP(instance *VM) error {
-	client, err := m.deps.QMPClientFactory.Create(instance)
+	client, err := newQMPClientWithHandshake(instance)
 	if err != nil {
 		return err
 	}
@@ -448,6 +445,22 @@ func (m *Manager) AttachQMP(instance *VM) error {
 	return nil
 }
 
+// newQMPClientWithHandshake dials the instance's QMP socket and runs the
+// qmp_capabilities handshake. The caller owns starting the heartbeat
+// goroutine on the returned client.
+func newQMPClientWithHandshake(v *VM) (*qmp.QMPClient, error) {
+	client, err := qmp.NewQMPClient(v.Config.QMPSocket)
+	if err != nil {
+		return nil, fmt.Errorf("connect QMP socket %s: %w", v.Config.QMPSocket, err)
+	}
+	if _, err := sendQMPCommand(client, qmp.QMPCommand{Execute: "qmp_capabilities"}, v.ID); err != nil {
+		_ = client.Conn.Close()
+		return nil, err
+	}
+	slog.Debug("QMP handshake complete", "instance", v.ID)
+	return client, nil
+}
+
 // qmpHeartbeat sends a query-status QMP command every 30s to confirm the
 // QEMU process is still healthy. Exits when the instance reaches a terminal
 // or transitional state. Closes the QMP connection on exit.
@@ -479,9 +492,7 @@ func (m *Manager) qmpHeartbeat(instance *VM) {
 }
 
 // sendQMPCommand encodes cmd onto client and decodes the matching response,
-// skipping informational events. Mirrors the pre-2b Daemon.SendQMPCommand
-// helper but without the daemon receiver — used by manager-internal callers
-// (heartbeat).
+// skipping informational events.
 func sendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceID string) (*qmp.QMPResponse, error) {
 	if q == nil || q.Encoder == nil || q.Decoder == nil {
 		return nil, fmt.Errorf("QMP client is not initialized")

spinifex/vm/manager.go

@@ -34,8 +34,6 @@ type Deps struct {
 
 	StateStore         StateStore
 	VolumeMounter      VolumeMounter
-	QMPClientFactory   QMPClientFactory
-	ProcessLauncher    ProcessLauncher
 	NetworkPlumber     NetworkPlumber
 	InstanceTypes      InstanceTypeResolver
 	Resources          ResourceController

spinifex/vm/manager_deps_test.go

@@ -213,12 +213,8 @@ func TestDeps_StateStoreSatisfiesInterface(t *testing.T) {
 	if err := m.deps.StateStore.WriteStoppedInstance("i-1", &VM{ID: "i-1"}); err != nil {
 		t.Fatalf("WriteStoppedInstance: %v", err)
 	}
-	got, err := m.deps.StateStore.LoadStoppedInstance("i-1")
-	if err != nil {
-		t.Fatalf("LoadStoppedInstance: %v", err)
-	}
-	if got == nil || got.ID != "i-1" {
-		t.Fatalf("LoadStoppedInstance: got %v, want VM{ID:i-1}", got)
+	if got := store.stopped["i-1"]; got == nil || got.ID != "i-1" {
+		t.Fatalf("WriteStoppedInstance: got %v, want VM{ID:i-1}", got)
 	}
 
 	if err := m.deps.StateStore.SaveRunningState("n", map[string]*VM{"i-1": {ID: "i-1"}}); err != nil {

spinifex/vm/process_launcher.go

@@ -1,10 +0,0 @@
-package vm
-
-import "os/exec"
-
-// ProcessLauncher resolves a Config into an *exec.Cmd ready to Start(). The
-// real implementation is a 1-line wrapper over Config.Execute(); the seam
-// exists so tests can substitute scripted commands without spawning QEMU.
-type ProcessLauncher interface {
-	Launch(cfg *Config) (*exec.Cmd, error)
-}

spinifex/vm/qmp_client.go

@@ -1,10 +0,0 @@
-package vm
-
-import "github.com/mulgadc/spinifex/spinifex/qmp"
-
-// QMPClientFactory connects a QMP client to a VM's QMP socket and runs the
-// qmp_capabilities handshake. The returned client is ready for command
-// dispatch. Heartbeats are owned by the manager, not the factory.
-type QMPClientFactory interface {
-	Create(v *VM) (*qmp.QMPClient, error)
-}

spinifex/vm/restore.go

@@ -245,11 +245,11 @@ func (m *Manager) finalizeTransitionalRestore(instance *VM) bool {
 	return true
 }
 
-// relaunchAll kicks the recovery launch loop. Per pre-2e behaviour, we
-// announce each instance via OnInstanceRecovering before launching so the
-// daemon can early-subscribe ec2.cmd.<id>; that lets a concurrent terminate
-// reach this node while the relaunch is still pending. The OnInstanceUp
-// hook fired after a successful launch reinstalls both per-instance subs
+// relaunchAll kicks the recovery launch loop. Each instance is announced
+// via OnInstanceRecovering before launching so the daemon can
+// early-subscribe ec2.cmd.<id>; that lets a concurrent terminate reach
+// this node while the relaunch is still pending. The OnInstanceUp hook
+// fired after a successful launch reinstalls both per-instance subs
 // idempotently.
 func (m *Manager) relaunchAll(toLaunch []*VM) {
 	if m.deps.Hooks.OnInstanceRecovering != nil {

spinifex/vm/shutdown.go

@@ -129,7 +129,7 @@ func (m *Manager) Terminate(id string) error {
 // launch errors mid-way: callers (NATS RunInstances handler, recovery
 // worker, pending watchdog, system-instance launcher) get back control
 // immediately and do not block on volume unmount, ENI delete, or KV
-// writes. Mirrors the pre-2c `go d.finalizeTermination(instance)` pattern.
+// writes.
 //
 // Tolerates instances already in a cleanup state (no-op) and instances
 // that may or may not be present in the running-VM map.
@@ -350,9 +350,8 @@ func (m *Manager) transitionWithPrecheck(instance *VM, target InstanceState) err
 }
 
 // writeRunningState persists the current running-VM map via the StateStore.
-// The View callback holds the manager lock across the marshal+put — same
-// constraint as the daemon's pre-2c WriteState. Splitting marshal from put
-// is a 2f cleanup item.
+// The View callback holds the manager lock across the marshal+put so VM
+// fields can't change mid-encode; splitting marshal from put is deferred.
 func (m *Manager) writeRunningState() error {
 	if m.deps.StateStore == nil {
 		return nil

spinifex/vm/shutdown_test.go

@@ -132,9 +132,7 @@ func TestMarkFailed_AlreadyTerminated_NoOp(t *testing.T) {
 
 // TestStop_DoesNotCallDeleteVolumes locks down the architectural
 // invariant that Stop must never delete volumes — a regression here
-// would silently destroy user data on every stop. Phase 2c removed
-// TestStopInstance_NoDelete_OnStop from daemon_test.go without a
-// vm-package replacement; this is that replacement.
+// would silently destroy user data on every stop.
 func TestStop_DoesNotCallDeleteVolumes(t *testing.T) {
 	m, _, _, cleaner, _ := shutdownTestManager(t)
 

spinifex/vm/state_store.go

@@ -1,11 +1,12 @@
 package vm
 
-// StateStore persists VM state across the three locations spinifex models:
-// per-node "running" state, the cluster-shared "stopped" bucket, and the
-// cluster-shared "terminated" bucket.
+// StateStore persists VM state to the per-node "running" bucket and to the
+// cluster-shared "stopped" / "terminated" buckets.
 //
 // The interface lives in the vm package; the live implementation is a thin
-// adapter in the daemon package wrapping the JetStream-backed bucket.
+// adapter in the daemon package wrapping the JetStream-backed bucket. Both
+// vm.Manager and daemon-side handlers route VM-instance state I/O through
+// this interface so the daemon can be exercised against an in-memory fake.
 type StateStore interface {
 	// SaveRunningState writes the snapshot of VMs currently running on the
 	// given node. The supplied map is owned by the caller and must not be

spinifex/vm/vm.go

@@ -35,8 +35,6 @@ type ExtraENI struct {
 
 type VM struct {
 	ID           string        `json:"id"`
-	PID          int           `json:"pid"`
-	Running      bool          `json:"running"`
 	Status       InstanceState `json:"status"`
 	InstanceType string        `json:"instance_type"`
 	Config       Config        `json:"config"`
@@ -70,7 +68,8 @@ type VM struct {
 	Health InstanceHealthState `json:"health"`
 
 	// AccountID is the AWS account that owns this instance.
-	// Empty for pre-Phase4 resources (treated as visible to all accounts).
+	// Empty AccountID identifies legacy/migration resources that are only
+	// visible to the root account.
 	AccountID string `json:"account_id,omitempty"`
 
 	// VPC networking: ENI attached to this instance (set by RunInstances when VPC mode is active)
@@ -130,8 +129,6 @@ type VM struct {
 // that last ran this instance. Must be called after deserializing a VM from
 // shared KV before launching it on a new node.
 func (v *VM) ResetNodeLocalState() {
-	v.PID = 0
-	v.Running = false
 	v.MetadataServerAddress = ""
 	v.QMPClient = &qmp.QMPClient{}
 	v.EBSRequests.Mu = sync.Mutex{}

spinifex/vm/vm_test.go

@@ -233,16 +233,12 @@ func argExists(args []string, flag string) bool {
 func TestResetNodeLocalState(t *testing.T) {
 	v := &VM{
 		ID:                    "i-abc123",
-		PID:                   12345,
-		Running:               true,
 		MetadataServerAddress: "127.0.0.1:9999",
 		Status:                StateRunning,
 	}
 
 	v.ResetNodeLocalState()
 
-	assert.Equal(t, 0, v.PID)
-	assert.False(t, v.Running)
 	assert.Empty(t, v.MetadataServerAddress)
 	assert.NotNil(t, v.QMPClient)
 	// ID and Status should be unchanged

spinifex/vm/volumes.go

@@ -16,7 +16,7 @@ import (
 
 // blockdevDelMaxAttempts caps the bounded retry on "node is in use" while
 // QEMU drains pending I/O after device_del. 20 × DetachDelay (default 1s)
-// gives a 20s budget that matches the pre-2d daemon-side helper.
+// gives a 20s budget.
 const blockdevDelMaxAttempts = 20
 
 // AttachVolumeResult carries the device names produced by AttachVolume:
@@ -270,10 +270,9 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string,
 	nodeName := fmt.Sprintf("nbd-%s", volumeID)
 	iothreadID := fmt.Sprintf("ioth-%s", volumeID)
 
-	// Phase 1: device_del. Idempotent on DeviceNotFound so the second
-	// AWS-CLI retry can drive blockdev-del to completion when a prior
-	// detach left the guest device gone but the block node intact
-	// (mulga-siv-41).
+	// device_del is idempotent on DeviceNotFound so a second AWS-CLI
+	// retry can drive blockdev-del to completion when a prior detach left
+	// the guest device gone but the block node intact.
 	_, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
 		Execute:   "device_del",
 		Arguments: map[string]any{"id": deviceID},
@@ -295,14 +294,14 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string,
 		time.Sleep(m.deps.DetachDelay)
 	}
 
-	// Phase 2: blockdev-del with bounded retry on "node is in use".
+	// blockdev-del with bounded retry on "node is in use".
 	if blockdevErr := m.tryBlockdevDel(instance, nodeName); blockdevErr != nil {
 		slog.Error("DetachVolume: QMP blockdev-del failed, leaving volume state intact",
 			"volumeId", volumeID, "err", blockdevErr)
 		return "", fmt.Errorf("QMP blockdev-del: %w", blockdevErr)
 	}
 
-	// Phase 2b: object-del (best-effort).
+	// object-del (best-effort).
 	if _, iothreadErr := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
 		Execute:   "object-del",
 		Arguments: map[string]any{"id": iothreadID},
@@ -311,7 +310,7 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string,
 			"volumeId", volumeID, "err", iothreadErr)
 	}
 
-	// Phase 3: ebs.unmount (best-effort).
+	// ebs.unmount (best-effort).
 	if m.deps.VolumeMounter != nil {
 		m.deps.VolumeMounter.UnmountOne(ebsReq)
 	}
@@ -392,7 +391,7 @@ func (m *Manager) tryBlockdevDel(instance *VM, nodeName string) error {
 
 // isQMPDeviceNotFound returns true when err is a QMP DeviceNotFound class
 // error from device_del. Used to make device_del idempotent across detach
-// retries (mulga-siv-41).
+// retries.
 func isQMPDeviceNotFound(err error) bool {
 	if err == nil {
 		return false

spinifex/vm/volumes_test.go

@@ -411,12 +411,10 @@ func attachVolumeRunningInstance(t *testing.T, qmpClient *qmp.QMPClient, mounter
 	return m, v
 }
 
-// TestAttachVolume_MountAmbiguous_TriggersUnmountOne covers the regression
-// from issue #1: pre-2d the daemon called rollbackEBSMount when the backend
-// acknowledged the mount with an empty URI. 2d collapsed the empty-URI case
-// into an opaque MountOne error and dropped the rollback. The fix introduces
-// ErrMountAmbiguous so AttachVolume can detect this specific case and call
-// UnmountOne defensively.
+// TestAttachVolume_MountAmbiguous_TriggersUnmountOne asserts that an
+// ebs.mount response with an empty URI is treated as ambiguous: AttachVolume
+// returns ErrMountAmbiguous and invokes UnmountOne so a half-started
+// backend mount cannot be orphaned.
 func TestAttachVolume_MountAmbiguous_TriggersUnmountOne(t *testing.T) {
 	mounter := &fakeVolumeMounter{mountOneErr: ErrMountAmbiguous}
 	m, _ := attachVolumeRunningInstance(t, nil, mounter)
@@ -430,8 +428,7 @@ func TestAttachVolume_MountAmbiguous_TriggersUnmountOne(t *testing.T) {
 
 // TestAttachVolume_GenericMountError_DoesNotUnmount confirms that other
 // MountOne failures (NATS timeout, backend explicit Error, marshal errors)
-// do NOT trigger UnmountOne — matches pre-2d behaviour where the daemon
-// only rolled back on the empty-URI case, not on mountResp.Error != "".
+// do NOT trigger UnmountOne — only the empty-URI ambiguous case rolls back.
 func TestAttachVolume_GenericMountError_DoesNotUnmount(t *testing.T) {
 	mounter := &fakeVolumeMounter{mountOneErr: errors.New("ebs.mount NATS request: timeout")}
 	m, _ := attachVolumeRunningInstance(t, nil, mounter)
@@ -440,7 +437,7 @@ func TestAttachVolume_GenericMountError_DoesNotUnmount(t *testing.T) {
 	require.Error(t, err)
 	assert.NotErrorIs(t, err, ErrMountAmbiguous)
 	assert.Empty(t, mounter.unmountedOne,
-		"non-ambiguous MountOne errors must not trigger UnmountOne (matches pre-2d)")
+		"non-ambiguous MountOne errors must not trigger UnmountOne")
 }
 
 // TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne covers the rollback