refactor: accounts

hive/daemon/daemon.go

@@ -454,40 +454,55 @@ func (d *Daemon) Start() error {
 	d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store)
 	d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config)
 	d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config, d.natsConn)
-	snapSvc, snapshotKV, err := d.initSnapshotService()
-	if err != nil {
-		return fmt.Errorf("failed to initialize snapshot service with NATS KV: %w", err)
+
+	type snapResult struct {
+		svc *handlers_ec2_snapshot.SnapshotServiceImpl
+		kv  nats.KeyValue
 	}
-	d.snapshotService = snapSvc
-	d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn, snapshotKV)
+	snap, err := initServiceWithRetry("snapshot service", func() (snapResult, error) {
+		svc, kv, err := handlers_ec2_snapshot.NewSnapshotServiceImplWithNATS(d.config, d.natsConn)
+		return snapResult{svc, kv}, err
+	})
+	if err != nil {
+		return fmt.Errorf("failed to initialize snapshot service: %w", err)
+	}
+	d.snapshotService = snap.svc
+
+	d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn, snap.kv)
 	d.tagsService = handlers_ec2_tags.NewTagsServiceImpl(d.config)
 
-	if eigwSvc, err := handlers_ec2_eigw.NewEgressOnlyIGWServiceImplWithNATS(d.config, d.natsConn); err != nil {
-		slog.Warn("Failed to initialize EIGW service, falling back to in-memory", "error", err)
-		d.eigwService = handlers_ec2_eigw.NewEgressOnlyIGWServiceImpl(d.config)
-	} else {
-		d.eigwService = eigwSvc
+	d.eigwService, err = initServiceWithRetry("EIGW service", func() (*handlers_ec2_eigw.EgressOnlyIGWServiceImpl, error) {
+		return handlers_ec2_eigw.NewEgressOnlyIGWServiceImplWithNATS(d.config, d.natsConn)
+	})
+	if err != nil {
+		return fmt.Errorf("failed to initialize EIGW service: %w", err)
 	}
-	if igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(d.config, d.natsConn); err != nil {
-		slog.Warn("Failed to initialize IGW service, falling back to in-memory", "error", err)
-		d.igwService = handlers_ec2_igw.NewIGWServiceImpl(d.config)
-	} else {
-		d.igwService = igwSvc
+
+	d.igwService, err = initServiceWithRetry("IGW service", func() (*handlers_ec2_igw.IGWServiceImpl, error) {
+		return handlers_ec2_igw.NewIGWServiceImplWithNATS(d.config, d.natsConn)
+	})
+	if err != nil {
+		return fmt.Errorf("failed to initialize IGW service: %w", err)
 	}
-	if vpcSvc, err := handlers_ec2_vpc.NewVPCServiceImplWithNATS(d.config, d.natsConn); err != nil {
-		slog.Warn("Failed to initialize VPC service, falling back to in-memory", "error", err)
-		d.vpcService = handlers_ec2_vpc.NewVPCServiceImpl(d.config)
-	} else {
-		d.vpcService = vpcSvc
+
+	d.vpcService, err = initServiceWithRetry("VPC service", func() (*handlers_ec2_vpc.VPCServiceImpl, error) {
+		return handlers_ec2_vpc.NewVPCServiceImplWithNATS(d.config, d.natsConn)
+	})
+	if err != nil {
+		return fmt.Errorf("failed to initialize VPC service: %w", err)
 	}
-	if err := d.initAccountService(); err != nil {
+
+	d.accountService, err = initServiceWithRetry("account settings service", func() (*handlers_ec2_account.AccountSettingsServiceImpl, error) {
+		return handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn)
+	})
+	if err != nil {
 		return fmt.Errorf("failed to initialize account settings service: %w", err)
 	}
 
 	// Ensure default VPC exists (matches AWS: every account has a default VPC)
 	if d.vpcService != nil {
-		if err := d.vpcService.EnsureDefaultVPC(handlers_ec2_vpc.GlobalAccountID); err != nil {
-			slog.Warn("Failed to ensure default VPC", "error", err)
+		if err := d.vpcService.EnsureDefaultVPC(utils.GlobalAccountID); err != nil {
+			slog.Error("Failed to ensure default VPC", "error", err)
 		}
 	}
 
@@ -603,10 +618,11 @@ func (d *Daemon) initJetStream() error {
 	return nil
 }
 
-// initSnapshotService initializes the snapshot service with retry/backoff.
-// During cluster restarts, JetStream KV may be temporarily unavailable while
-// NATS routes re-establish and the cluster forms quorum.
-func (d *Daemon) initSnapshotService() (*handlers_ec2_snapshot.SnapshotServiceImpl, nats.KeyValue, error) {
+// initServiceWithRetry initializes a service using the provided init function,
+// retrying with exponential backoff (500ms→10s) for up to 5 minutes. During
+// cluster restarts, JetStream KV may be temporarily unavailable while NATS
+// routes re-establish and the cluster forms quorum.
+func initServiceWithRetry[T any](name string, initFn func() (T, error)) (T, error) {
 	const maxWait = 5 * time.Minute
 	retryDelay := 500 * time.Millisecond
 	start := time.Now()
@@ -614,49 +630,21 @@ func (d *Daemon) initSnapshotService() (*handlers_ec2_snapshot.SnapshotServiceIm
 
 	for {
 		attempt++
-		snapSvc, snapshotKV, err := handlers_ec2_snapshot.NewSnapshotServiceImplWithNATS(d.config, d.natsConn)
+		result, err := initFn()
 		if err == nil {
 			if attempt > 1 {
-				slog.Info("Snapshot service initialized successfully", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second))
+				slog.Info(name+" initialized successfully", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second))
 			}
-			return snapSvc, snapshotKV, nil
+			return result, nil
 		}
 
 		elapsed := time.Since(start)
 		if elapsed >= maxWait {
-			return nil, nil, fmt.Errorf("snapshot service unavailable after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err)
+			var zero T
+			return zero, fmt.Errorf("%s unavailable after %s (%d attempts): %w", name, elapsed.Round(time.Second), attempt, err)
 		}
 
-		slog.Warn("Failed to init snapshot service", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second))
-		time.Sleep(retryDelay)
-		retryDelay = min(retryDelay*2, 10*time.Second)
-	}
-}
-
-// initAccountService initializes the account settings service with retry/backoff.
-func (d *Daemon) initAccountService() error {
-	const maxWait = 5 * time.Minute
-	retryDelay := 500 * time.Millisecond
-	start := time.Now()
-	attempt := 0
-
-	for {
-		attempt++
-		svc, err := handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn)
-		if err == nil {
-			if attempt > 1 {
-				slog.Info("Account settings service initialized successfully", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second))
-			}
-			d.accountService = svc
-			return nil
-		}
-
-		elapsed := time.Since(start)
-		if elapsed >= maxWait {
-			return fmt.Errorf("account settings service unavailable after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err)
-		}
-
-		slog.Warn("Failed to init account settings service", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second))
+		slog.Warn("Failed to init "+name, "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second))
 		time.Sleep(retryDelay)
 		retryDelay = min(retryDelay*2, 10*time.Second)
 	}
@@ -1384,7 +1372,7 @@ func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) er
 					NetworkInterfaceId: &instance.ENIId,
 				}, instance.AccountID)
 				if eniErr != nil {
-					slog.Warn("Failed to delete ENI on termination", "eni", instance.ENIId, "err", eniErr)
+					slog.Error("Failed to delete ENI on termination", "eni", instance.ENIId, "err", eniErr)
 				} else {
 					slog.Info("Deleted ENI on termination", "eni", instance.ENIId, "instanceId", instance.ID)
 				}

hive/daemon/daemon_handlers.go

@@ -73,11 +73,8 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
 		return
 	}
 
-	// Verify the caller owns this instance (skip for pre-Phase4 instances with no AccountID)
-	callerAccountID := utils.AccountIDFromMsg(msg)
-	if callerAccountID != "" && instance.AccountID != "" && instance.AccountID != callerAccountID {
-		slog.Warn("Account does not own this instance", "instanceId", command.ID, "callerAccount", callerAccountID, "ownerAccount", instance.AccountID)
-		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+	// Verify the caller owns this instance
+	if !checkInstanceOwnership(msg, command.ID, instance.AccountID) {
 		return
 	}
 

hive/daemon/daemon_handlers_console.go

@@ -32,7 +32,6 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) {
 	}
 
 	instanceID := *input.InstanceId
-	accountID := utils.AccountIDFromMsg(msg)
 
 	// Find the instance on this node
 	d.Instances.Mu.Lock()
@@ -45,8 +44,7 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) {
 	}
 
 	// Verify the caller owns this instance
-	if accountID != "" && instance.AccountID != "" && instance.AccountID != accountID {
-		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+	if !checkInstanceOwnership(msg, instanceID, instance.AccountID) {
 		return
 	}
 

hive/daemon/daemon_handlers_image.go

@@ -66,9 +66,7 @@ func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) {
 	}
 
 	// Verify the caller owns this instance
-	if accountID != "" && instance.AccountID != "" && instance.AccountID != accountID {
-		slog.Warn("CreateImage: account does not own instance", "instanceId", instanceID, "accountID", accountID, "ownerAccountID", instance.AccountID)
-		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+	if !checkInstanceOwnership(msg, instanceID, instance.AccountID) {
 		return
 	}
 

hive/daemon/daemon_handlers_instance.go

@@ -74,19 +74,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
 	// System AMIs have non-account-ID owner aliases (e.g. "self", "hive", empty).
 	amiOwner := amiMeta.ImageOwnerAlias
 	if amiOwner != "" && amiOwner != accountID {
-		// Check if this is a system/pre-phase4 AMI (non-account-ID owner)
-		isSystem := len(amiOwner) != 12
-		if !isSystem {
-			allDigits := true
-			for _, c := range amiOwner {
-				if c < '0' || c > '9' {
-					allDigits = false
-					break
-				}
-			}
-			isSystem = !allDigits
-		}
-		if !isSystem {
+		if utils.IsAccountID(amiOwner) {
 			slog.Warn("handleEC2RunInstances AMI not owned by caller", "imageId", *runInstancesInput.ImageId, "amiOwner", amiOwner, "accountID", accountID)
 			respondWithError(msg, awserrors.ErrorInvalidAMIIDNotFound)
 			return
@@ -519,8 +507,8 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) {
 	// Iterate through all instances on this node
 	for _, instance := range d.Instances.VMS {
 		// Skip instances not owned by the caller's account.
-		// Pre-Phase4 instances (empty AccountID) are visible to all accounts.
-		if accountID != "" && instance.AccountID != "" && instance.AccountID != accountID {
+		// Pre-Phase4 instances (empty AccountID) are only visible to root.
+		if !isInstanceVisible(accountID, instance.AccountID) {
 			continue
 		}
 
@@ -654,9 +642,6 @@ type startStoppedInstanceRequest struct {
 // handleEC2StartStoppedInstance picks up a stopped instance from shared KV,
 // re-launches it on this daemon node, and removes it from shared KV.
 func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) {
-	// Extract account ID from NATS header for ownership verification
-	callerAccountID := utils.AccountIDFromMsg(msg)
-
 	var req startStoppedInstanceRequest
 	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("handleEC2StartStoppedInstance: failed to unmarshal request", "err", err)
@@ -696,9 +681,7 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) {
 	}
 
 	// Verify the caller owns this instance
-	if callerAccountID != "" && instance.AccountID != "" && instance.AccountID != callerAccountID {
-		slog.Warn("handleEC2StartStoppedInstance: account does not own instance", "instanceId", req.InstanceID, "callerAccount", callerAccountID, "ownerAccount", instance.AccountID)
-		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+	if !checkInstanceOwnership(msg, req.InstanceID, instance.AccountID) {
 		return
 	}
 
@@ -769,9 +752,6 @@ type terminateStoppedInstanceRequest struct {
 // handleEC2TerminateStoppedInstance picks up a stopped instance from shared KV,
 // deletes its volumes, and removes it from shared KV.
 func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
-	// Extract account ID from NATS header for ownership verification
-	callerAccountID := utils.AccountIDFromMsg(msg)
-
 	var req terminateStoppedInstanceRequest
 	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("handleEC2TerminateStoppedInstance: failed to unmarshal request", "err", err)
@@ -811,9 +791,7 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
 	}
 
 	// Verify the caller owns this instance
-	if callerAccountID != "" && instance.AccountID != "" && instance.AccountID != callerAccountID {
-		slog.Warn("handleEC2TerminateStoppedInstance: account does not own instance", "instanceId", req.InstanceID, "callerAccount", callerAccountID, "ownerAccount", instance.AccountID)
-		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+	if !checkInstanceOwnership(msg, req.InstanceID, instance.AccountID) {
 		return
 	}
 
@@ -910,8 +888,8 @@ func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) {
 
 	for _, instance := range instances {
 		// Skip instances not owned by the caller's account.
-		// Pre-Phase4 instances (empty AccountID) are visible to all accounts.
-		if accountID != "" && instance.AccountID != "" && instance.AccountID != accountID {
+		// Pre-Phase4 instances (empty AccountID) are only visible to root.
+		if !isInstanceVisible(accountID, instance.AccountID) {
 			continue
 		}
 
@@ -977,9 +955,6 @@ func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) {
 // handleEC2ModifyInstanceAttribute modifies attributes of a stopped instance in shared KV.
 // All supported attributes (InstanceType, UserData) require the instance to be stopped.
 func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) {
-	// Extract account ID from NATS header for ownership verification
-	callerAccountID := utils.AccountIDFromMsg(msg)
-
 	var input ec2.ModifyInstanceAttributeInput
 	if err := json.Unmarshal(msg.Data, &input); err != nil {
 		slog.Error("handleEC2ModifyInstanceAttribute: failed to unmarshal request", "err", err)
@@ -1020,9 +995,7 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) {
 	}
 
 	// Verify the caller owns this instance
-	if callerAccountID != "" && instance.AccountID != "" && instance.AccountID != callerAccountID {
-		slog.Warn("handleEC2ModifyInstanceAttribute: account does not own instance", "instanceId", instanceID, "callerAccount", callerAccountID, "ownerAccount", instance.AccountID)
-		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+	if !checkInstanceOwnership(msg, instanceID, instance.AccountID) {
 		return
 	}
 

hive/daemon/daemon_handlers_test.go

@@ -58,7 +58,6 @@ func createFullTestDaemonWithStore(t *testing.T, natsURL string) (*Daemon, *obje
 	daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImplWithStore(cfg, memStore, daemon.natsConn)
 	daemon.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImplWithStore(cfg, memStore, daemon.natsConn)
 	daemon.tagsService = handlers_ec2_tags.NewTagsServiceImplWithStore(cfg, memStore)
-	daemon.eigwService = handlers_ec2_eigw.NewEgressOnlyIGWServiceImpl(cfg)
 	initAccountServiceForTest(t, daemon)
 
 	return daemon, memStore
@@ -543,6 +542,7 @@ func TestHandleEC2Events_StopInstance(t *testing.T) {
 		Status:       vm.StateRunning,
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
+		AccountID:    testAccountID,
 	}
 
 	sub, err := daemon.natsConn.Subscribe(
@@ -558,7 +558,7 @@ func TestHandleEC2Events_StopInstance(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(cmd)
 
-	reply, err := daemon.natsConn.Request(
+	reply, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -588,6 +588,7 @@ func TestHandleEC2Events_TerminateInstance(t *testing.T) {
 		Status:       vm.StateRunning,
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
+		AccountID:    testAccountID,
 	}
 
 	sub, err := daemon.natsConn.Subscribe(
@@ -603,7 +604,7 @@ func TestHandleEC2Events_TerminateInstance(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(cmd)
 
-	reply, err := daemon.natsConn.Request(
+	reply, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -914,8 +915,9 @@ func TestHandleEC2CreateImage_InvalidState(t *testing.T) {
 	// Add an instance in "pending" state (not running or stopped)
 	daemon.Instances.Mu.Lock()
 	daemon.Instances.VMS["i-pending123"] = &vm.VM{
-		ID:     "i-pending123",
-		Status: vm.StatePending,
+		ID:        "i-pending123",
+		Status:    vm.StatePending,
+		AccountID: testAccountID,
 		Instance: &ec2.Instance{
 			InstanceId: aws.String("i-pending123"),
 			ImageId:    aws.String("ami-source"),
@@ -938,7 +940,7 @@ func TestHandleEC2CreateImage_InvalidState(t *testing.T) {
 		Name:       aws.String("my-image"),
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.CreateImage", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var errResp map[string]any
@@ -955,8 +957,9 @@ func TestHandleEC2CreateImage_NoRootVolume(t *testing.T) {
 	// Add instance with no block device mappings
 	daemon.Instances.Mu.Lock()
 	daemon.Instances.VMS["i-novol123"] = &vm.VM{
-		ID:     "i-novol123",
-		Status: vm.StateRunning,
+		ID:        "i-novol123",
+		Status:    vm.StateRunning,
+		AccountID: testAccountID,
 		Instance: &ec2.Instance{
 			InstanceId:          aws.String("i-novol123"),
 			ImageId:             aws.String("ami-source"),
@@ -974,7 +977,7 @@ func TestHandleEC2CreateImage_NoRootVolume(t *testing.T) {
 		Name:       aws.String("my-image"),
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.CreateImage", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var errResp map[string]any
@@ -1067,6 +1070,7 @@ func TestHandleEC2StartStoppedInstance_NotStoppedState(t *testing.T) {
 		ID:           "i-running-shared",
 		Status:       vm.StateRunning,
 		InstanceType: getTestInstanceType(),
+		AccountID:    testAccountID,
 	}
 	err := daemon.jsManager.WriteStoppedInstance(runningVM.ID, runningVM)
 	require.NoError(t, err)
@@ -1076,7 +1080,7 @@ func TestHandleEC2StartStoppedInstance_NotStoppedState(t *testing.T) {
 	defer sub.Unsubscribe()
 
 	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-running-shared"})
-	reply, err := daemon.natsConn.Request("ec2.start", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.start", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var errResp map[string]any
@@ -1101,6 +1105,7 @@ func TestHandleEC2DescribeStoppedInstances_ReturnsStoppedInstances(t *testing.T)
 		Status:       vm.StateStopped,
 		InstanceType: getTestInstanceType(),
 		LastNode:     "node-1",
+		AccountID:    testAccountID,
 		Reservation: &ec2.Reservation{
 			ReservationId: aws.String("r-test-001"),
 			OwnerId:       aws.String("123456789012"),
@@ -1119,7 +1124,7 @@ func TestHandleEC2DescribeStoppedInstances_ReturnsStoppedInstances(t *testing.T)
 
 	input := &ec2.DescribeInstancesInput{}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.DescribeStoppedInstances", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.DescribeStoppedInstances", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var output ec2.DescribeInstancesOutput
@@ -1151,9 +1156,10 @@ func TestHandleEC2DescribeStoppedInstances_WithFilter(t *testing.T) {
 	// Write two stopped instances
 	for _, id := range []string{"i-filter-001", "i-filter-002"} {
 		v := &vm.VM{
-			ID:       id,
-			Status:   vm.StateStopped,
-			LastNode: "node-1",
+			ID:        id,
+			Status:    vm.StateStopped,
+			LastNode:  "node-1",
+			AccountID: testAccountID,
 			Reservation: &ec2.Reservation{
 				ReservationId: aws.String("r-filter"),
 				OwnerId:       aws.String("123456789012"),
@@ -1175,7 +1181,7 @@ func TestHandleEC2DescribeStoppedInstances_WithFilter(t *testing.T) {
 		InstanceIds: []*string{aws.String("i-filter-001")},
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.DescribeStoppedInstances", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.DescribeStoppedInstances", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var output ec2.DescribeInstancesOutput
@@ -1252,6 +1258,7 @@ func TestHandleEC2TerminateStoppedInstance_NotStoppedState(t *testing.T) {
 		ID:           "i-term-running",
 		Status:       vm.StateRunning,
 		InstanceType: getTestInstanceType(),
+		AccountID:    testAccountID,
 	}
 	err := daemon.jsManager.WriteStoppedInstance(runningVM.ID, runningVM)
 	require.NoError(t, err)
@@ -1261,7 +1268,7 @@ func TestHandleEC2TerminateStoppedInstance_NotStoppedState(t *testing.T) {
 	defer sub.Unsubscribe()
 
 	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-running"})
-	reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var errResp map[string]any
@@ -1284,6 +1291,7 @@ func TestHandleEC2TerminateStoppedInstance_Success(t *testing.T) {
 		Status:       vm.StateStopped,
 		InstanceType: getTestInstanceType(),
 		LastNode:     "node-1",
+		AccountID:    testAccountID,
 		Reservation: &ec2.Reservation{
 			ReservationId: aws.String("r-term-001"),
 			OwnerId:       aws.String("123456789012"),
@@ -1301,7 +1309,7 @@ func TestHandleEC2TerminateStoppedInstance_Success(t *testing.T) {
 	defer sub.Unsubscribe()
 
 	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-stopped-001"})
-	reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var resp map[string]string
@@ -1330,8 +1338,9 @@ func TestHandleEC2GetConsoleOutput(t *testing.T) {
 	// Add an instance with console log path
 	daemon.Instances.Mu.Lock()
 	daemon.Instances.VMS[instanceID] = &vm.VM{
-		ID:     instanceID,
-		Status: vm.StateRunning,
+		ID:        instanceID,
+		Status:    vm.StateRunning,
+		AccountID: testAccountID,
 		Config: vm.Config{
 			ConsoleLogPath: logPath,
 		},
@@ -1347,7 +1356,7 @@ func TestHandleEC2GetConsoleOutput(t *testing.T) {
 		InstanceId: aws.String(instanceID),
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request(topic, reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, topic, reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var output ec2.GetConsoleOutputOutput
@@ -1373,8 +1382,9 @@ func TestHandleEC2GetConsoleOutput_EmptyLog(t *testing.T) {
 	// Instance exists but no log file yet
 	daemon.Instances.Mu.Lock()
 	daemon.Instances.VMS[instanceID] = &vm.VM{
-		ID:     instanceID,
-		Status: vm.StateRunning,
+		ID:        instanceID,
+		Status:    vm.StateRunning,
+		AccountID: testAccountID,
 		Config: vm.Config{
 			ConsoleLogPath: "/nonexistent/console.log",
 		},
@@ -1390,7 +1400,7 @@ func TestHandleEC2GetConsoleOutput_EmptyLog(t *testing.T) {
 		InstanceId: aws.String(instanceID),
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request(topic, reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, topic, reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var output ec2.GetConsoleOutputOutput
@@ -1439,6 +1449,7 @@ func TestAttachVolume_ZoneMismatch(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: getTestInstanceType(),
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
 	}
@@ -1483,7 +1494,7 @@ func TestAttachVolume_ZoneMismatch(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -1508,6 +1519,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
 		ID:           instanceID,
 		Status:       vm.StateStopped,
 		InstanceType: "t3.micro",
+		AccountID:    testAccountID,
 		Config:       vm.Config{InstanceType: "t3.micro"},
 		Instance: &ec2.Instance{
 			InstanceId:   aws.String(instanceID),
@@ -1523,7 +1535,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
 		InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
 	require.NoError(t, err)
 	assert.Equal(t, `{}`, string(reply.Data))
 
@@ -1549,6 +1561,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeUserData(t *testing.T) {
 		ID:           instanceID,
 		Status:       vm.StateStopped,
 		InstanceType: "t3.micro",
+		AccountID:    testAccountID,
 		UserData:     "old data",
 		RunInstancesInput: &ec2.RunInstancesInput{
 			UserData: aws.String("b2xkIGRhdGE="),
@@ -1569,7 +1582,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeUserData(t *testing.T) {
 		UserData:   &ec2.BlobAttributeValue{Value: []byte(newContent)},
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
 	require.NoError(t, err)
 	assert.Equal(t, `{}`, string(reply.Data))
 
@@ -1617,6 +1630,7 @@ func TestHandleEC2ModifyInstanceAttribute_NotStopped(t *testing.T) {
 		ID:           instanceID,
 		Status:       vm.StateRunning,
 		InstanceType: "t3.micro",
+		AccountID:    testAccountID,
 	}
 	err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
 	require.NoError(t, err)
@@ -1627,7 +1641,7 @@ func TestHandleEC2ModifyInstanceAttribute_NotStopped(t *testing.T) {
 		InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var errResp map[string]any
@@ -1650,6 +1664,7 @@ func TestHandleEC2ModifyInstanceAttribute_ClearsStateReason(t *testing.T) {
 		ID:           instanceID,
 		Status:       vm.StateStopped,
 		InstanceType: "m7i.small",
+		AccountID:    testAccountID,
 		Config:       vm.Config{InstanceType: "m7i.small"},
 		Instance: &ec2.Instance{
 			InstanceId:   aws.String(instanceID),
@@ -1669,7 +1684,7 @@ func TestHandleEC2ModifyInstanceAttribute_ClearsStateReason(t *testing.T) {
 		InstanceType: &ec2.AttributeValue{Value: aws.String("t3.micro")},
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
 	require.NoError(t, err)
 	assert.Equal(t, `{}`, string(reply.Data))
 
@@ -1696,6 +1711,7 @@ func TestHandleEC2ModifyInstanceAttribute_InvalidTypeAccepted(t *testing.T) {
 		ID:           instanceID,
 		Status:       vm.StateStopped,
 		InstanceType: "t3.micro",
+		AccountID:    testAccountID,
 		Config:       vm.Config{InstanceType: "t3.micro"},
 		Instance: &ec2.Instance{
 			InstanceId:   aws.String(instanceID),
@@ -1712,7 +1728,7 @@ func TestHandleEC2ModifyInstanceAttribute_InvalidTypeAccepted(t *testing.T) {
 		InstanceType: &ec2.AttributeValue{Value: aws.String("z99.mega")},
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
 	require.NoError(t, err)
 	assert.Equal(t, `{}`, string(reply.Data))
 
@@ -2518,6 +2534,7 @@ func TestHandleEC2TerminateStoppedInstance_WithVolumes(t *testing.T) {
 	stoppedVM := &vm.VM{
 		ID:           "i-term-vol-001",
 		Status:       vm.StateStopped,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		LastNode:     "node-1",
 		Reservation: &ec2.Reservation{
@@ -2545,7 +2562,7 @@ func TestHandleEC2TerminateStoppedInstance_WithVolumes(t *testing.T) {
 	defer sub.Unsubscribe()
 
 	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-vol-001"})
-	reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 10*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 10*time.Second)
 	require.NoError(t, err)
 
 	var resp map[string]string
@@ -2616,6 +2633,7 @@ func TestHandleEC2StartStoppedInstance_InstanceTypeNotAvailable(t *testing.T) {
 		ID:           "i-start-badtype-001",
 		Status:       vm.StateStopped,
 		InstanceType: "z99.nonexistent",
+		AccountID:    testAccountID,
 		Instance: &ec2.Instance{
 			InstanceId:   aws.String("i-start-badtype-001"),
 			InstanceType: aws.String("z99.nonexistent"),
@@ -2630,7 +2648,7 @@ func TestHandleEC2StartStoppedInstance_InstanceTypeNotAvailable(t *testing.T) {
 	defer sub.Unsubscribe()
 
 	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-start-badtype-001"})
-	reply, err := daemon.natsConn.Request("ec2.start", reqData, 5*time.Second)
+	reply, err := natsRequest(daemon.natsConn, "ec2.start", reqData, 5*time.Second)
 	require.NoError(t, err)
 
 	var errResp map[string]any
@@ -2672,6 +2690,7 @@ func TestHandleEC2CreateImage_RunningInstanceReachesService(t *testing.T) {
 		ID:           instanceID,
 		Status:       vm.StateRunning,
 		InstanceType: getTestInstanceType(),
+		AccountID:    testAccountID,
 		Instance: &ec2.Instance{
 			InstanceId: aws.String(instanceID),
 			ImageId:    aws.String(sourceImageID),
@@ -2699,7 +2718,7 @@ func TestHandleEC2CreateImage_RunningInstanceReachesService(t *testing.T) {
 		Name:       aws.String("test-image-snapshot"),
 	}
 	reqData, _ := json.Marshal(input)
-	reply, err := daemon.natsConn.Request(
+	reply, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.%s.CreateImage", instanceID),
 		reqData,
 		5*time.Second,
@@ -2720,6 +2739,7 @@ func TestAttachVolume_MissingVolumeData(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -2743,7 +2763,7 @@ func TestAttachVolume_MissingVolumeData(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -2759,6 +2779,7 @@ func TestAttachVolume_InstanceNotRunning(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateStopped,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -2783,7 +2804,7 @@ func TestAttachVolume_InstanceNotRunning(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -2799,6 +2820,7 @@ func TestAttachVolume_VolumeNotFound(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -2823,7 +2845,7 @@ func TestAttachVolume_VolumeNotFound(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -2841,6 +2863,7 @@ func TestAttachVolume_VolumeInUse(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -2884,7 +2907,7 @@ func TestAttachVolume_VolumeInUse(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -2902,6 +2925,7 @@ func TestDetachVolume_MissingVolumeData(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -2924,7 +2948,7 @@ func TestDetachVolume_MissingVolumeData(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -2940,6 +2964,7 @@ func TestDetachVolume_InstanceNotRunning(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateStopped,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -2964,7 +2989,7 @@ func TestDetachVolume_InstanceNotRunning(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -2980,6 +3005,7 @@ func TestDetachVolume_VolumeNotAttached(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -3004,7 +3030,7 @@ func TestDetachVolume_VolumeNotAttached(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -3020,6 +3046,7 @@ func TestDetachVolume_BootVolumeRejected(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -3047,7 +3074,7 @@ func TestDetachVolume_BootVolumeRejected(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,
@@ -3063,6 +3090,7 @@ func TestDetachVolume_DeviceMismatch(t *testing.T) {
 	instance := &vm.VM{
 		ID:           instanceID,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		InstanceType: getTestInstanceType(),
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{},
@@ -3091,7 +3119,7 @@ func TestDetachVolume_DeviceMismatch(t *testing.T) {
 	}
 	cmdData, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		cmdData,
 		5*time.Second,

hive/daemon/daemon_test.go

@@ -860,12 +860,14 @@ func TestDaemon_BootAllocation(t *testing.T) {
 			ID:           "i-running",
 			InstanceType: getTestInstanceType(),
 			Status:       vm.StateRunning,
+			AccountID:    testAccountID,
 			Attributes:   qmp.Attributes{StopInstance: false},
 		},
 		"i-stopped": {
 			ID:           "i-stopped",
 			InstanceType: getTestInstanceType(),
 			Status:       vm.StateStopped,
+			AccountID:    testAccountID,
 			Attributes:   qmp.Attributes{StopInstance: true},
 		},
 		"i-terminated": {
@@ -940,6 +942,7 @@ func TestStopInstance_Deallocation(t *testing.T) {
 		ID:           instanceId,
 		InstanceType: instanceTypeStr,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 	}
 
 	err := daemon.resourceMgr.allocate(instanceType)
@@ -1123,6 +1126,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 		daemon.Instances.VMS[instanceID] = &vm.VM{
 			ID:          instanceID,
 			Status:      vm.StateRunning,
+			AccountID:   testAccountID,
 			Reservation: reservation1,
 			Instance:    ec2Instance,
 		}
@@ -1142,6 +1146,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 		daemon.Instances.VMS[instanceID] = &vm.VM{
 			ID:          instanceID,
 			Status:      vm.StateRunning,
+			AccountID:   testAccountID,
 			Reservation: reservation2,
 			Instance:    ec2Instance,
 		}
@@ -1159,6 +1164,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 	daemon.Instances.VMS["i-single-001"] = &vm.VM{
 		ID:          "i-single-001",
 		Status:      vm.StateStopped,
+		AccountID:   testAccountID,
 		Reservation: reservation3,
 		Instance:    ec2Instance,
 	}
@@ -1172,7 +1178,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 		input := &ec2.DescribeInstancesInput{}
 		inputJSON, _ := json.Marshal(input)
 
-		resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
+		resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second)
 		require.NoError(t, err)
 
 		var output ec2.DescribeInstancesOutput
@@ -1202,7 +1208,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 		}
 		inputJSON, _ := json.Marshal(input)
 
-		resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
+		resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second)
 		require.NoError(t, err)
 
 		var output ec2.DescribeInstancesOutput
@@ -1226,7 +1232,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 		}
 		inputJSON, _ := json.Marshal(input)
 
-		resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
+		resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second)
 		require.NoError(t, err)
 
 		var output ec2.DescribeInstancesOutput
@@ -1250,7 +1256,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 		}
 		inputJSON, _ := json.Marshal(input)
 
-		resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
+		resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second)
 		require.NoError(t, err)
 
 		var output ec2.DescribeInstancesOutput
@@ -1268,7 +1274,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 		input := &ec2.DescribeInstancesInput{}
 		inputJSON, _ := json.Marshal(input)
 
-		resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
+		resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second)
 		require.NoError(t, err)
 
 		var output ec2.DescribeInstancesOutput
@@ -1752,6 +1758,7 @@ func TestStopInstance_DeleteOnTermination_VolumeDeletion(t *testing.T) {
 		ID:           "i-test-dot",
 		InstanceType: getTestInstanceType(),
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		QMPClient:    &qmp.QMPClient{}, // nil encoder/decoder => QMP will fail, which is fine
 		EBSRequests: config.EBSRequests{
 			Requests: []config.EBSRequest{
@@ -1849,6 +1856,7 @@ func TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion(t *testing.T
 		ID:           "i-test-no-delete",
 		InstanceType: getTestInstanceType(),
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		QMPClient:    &qmp.QMPClient{},
 		EBSRequests: config.EBSRequests{
 			Requests: []config.EBSRequest{
@@ -1935,6 +1943,7 @@ func TestStopInstance_NoDelete_OnStop(t *testing.T) {
 		ID:           "i-test-stop-only",
 		InstanceType: getTestInstanceType(),
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		QMPClient:    &qmp.QMPClient{},
 		EBSRequests: config.EBSRequests{
 			Requests: []config.EBSRequest{
@@ -1986,6 +1995,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: instanceType,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		Instance:     &ec2.Instance{},
 		QMPClient:    &qmp.QMPClient{}, // nil encoder/decoder
 	}
@@ -2009,7 +2019,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2037,7 +2047,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2065,7 +2075,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2114,6 +2124,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: instanceType,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		Instance: &ec2.Instance{
 			BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
 				{
@@ -2154,7 +2165,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2180,7 +2191,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2206,7 +2217,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2237,7 +2248,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2272,7 +2283,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2306,7 +2317,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2332,7 +2343,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2356,7 +2367,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -2450,6 +2461,7 @@ func TestDetachVolume_SuccessPath(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: instanceType,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		Instance: &ec2.Instance{
 			BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
 				{
@@ -2515,7 +2527,7 @@ func TestDetachVolume_SuccessPath(t *testing.T) {
 	}
 	data, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		data,
 		10*time.Second,
@@ -2603,6 +2615,7 @@ func TestDetachVolume_ForceFlag(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: instanceType,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		Instance: &ec2.Instance{
 			BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
 				{
@@ -2654,7 +2667,7 @@ func TestDetachVolume_ForceFlag(t *testing.T) {
 	}
 	data, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		data,
 		10*time.Second,
@@ -2716,6 +2729,7 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: instanceType,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		Instance: &ec2.Instance{
 			BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
 				{
@@ -2757,7 +2771,7 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) {
 	}
 	data, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		data,
 		10*time.Second,
@@ -2807,6 +2821,7 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: instanceType,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		Instance: &ec2.Instance{
 			BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
 				{
@@ -2857,7 +2872,7 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) {
 	}
 	data, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		data,
 		10*time.Second,
@@ -2891,6 +2906,7 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: instanceType,
 		Status:       vm.StateRunning,
+		AccountID:    testAccountID,
 		Instance:     &ec2.Instance{},
 		QMPClient:    qmpClient,
 		EBSRequests: config.EBSRequests{
@@ -2933,7 +2949,7 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) {
 	}
 	data, _ := json.Marshal(command)
 
-	resp, err := daemon.natsConn.Request(
+	resp, err := natsRequest(daemon.natsConn,
 		fmt.Sprintf("ec2.cmd.%s", instanceID),
 		data,
 		10*time.Second,
@@ -3236,6 +3252,7 @@ func TestMarkInstanceFailed(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: getTestInstanceType(),
 		Status:       vm.StatePending,
+		AccountID:    testAccountID,
 		Instance:     ec2Instance,
 		QMPClient:    &qmp.QMPClient{},
 	}
@@ -3267,6 +3284,7 @@ func TestMarkInstanceFailed_NilInstance(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: getTestInstanceType(),
 		Status:       vm.StatePending,
+		AccountID:    testAccountID,
 		Instance:     nil, // no ec2.Instance
 		QMPClient:    &qmp.QMPClient{},
 	}
@@ -3389,7 +3407,7 @@ func TestDescribeInstances_InvalidInstanceIDMalformed(t *testing.T) {
 		}
 		inputJSON, _ := json.Marshal(input)
 
-		resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
+		resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second)
 		require.NoError(t, err)
 		assert.Contains(t, string(resp.Data), "InvalidInstanceID.Malformed")
 	})
@@ -3400,7 +3418,7 @@ func TestDescribeInstances_InvalidInstanceIDMalformed(t *testing.T) {
 		}
 		inputJSON, _ := json.Marshal(input)
 
-		resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
+		resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second)
 		require.NoError(t, err)
 		// Should not contain a malformed error — returns empty results instead
 		assert.NotContains(t, string(resp.Data), "InvalidInstanceID.Malformed")
@@ -3419,6 +3437,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
 		ID:           instanceID,
 		InstanceType: getTestInstanceType(),
 		Status:       vm.StateStopped,
+		AccountID:    testAccountID,
 		Instance:     &ec2.Instance{},
 	}
 	daemon.Instances.VMS[instanceID] = instance
@@ -3443,7 +3462,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,
@@ -3466,7 +3485,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
 		}
 		data, _ := json.Marshal(command)
 
-		resp, err := daemon.natsConn.Request(
+		resp, err := natsRequest(daemon.natsConn,
 			fmt.Sprintf("ec2.cmd.%s", instanceID),
 			data,
 			5*time.Second,

hive/daemon/ownership.go

@@ -0,0 +1,45 @@
+package daemon
+
+import (
+	"log/slog"
+
+	"github.com/mulgadc/hive/hive/awserrors"
+	"github.com/mulgadc/hive/hive/utils"
+	"github.com/nats-io/nats.go"
+)
+
+// 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).
+func checkInstanceOwnership(msg *nats.Msg, instanceID, ownerAccountID string) bool {
+	callerAccountID := utils.AccountIDFromMsg(msg)
+
+	// Pre-Phase4 instance: only root can access
+	if ownerAccountID == "" {
+		if callerAccountID != utils.GlobalAccountID {
+			slog.Warn("Pre-Phase4 instance access denied (not root)",
+				"instanceId", instanceID, "callerAccount", callerAccountID)
+			respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+			return false
+		}
+		return true
+	}
+
+	// Normal ownership check
+	if callerAccountID != ownerAccountID {
+		slog.Warn("Account does not own instance",
+			"instanceId", instanceID, "callerAccount", callerAccountID, "ownerAccount", ownerAccountID)
+		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+		return false
+	}
+	return true
+}
+
+// isInstanceVisible checks if the caller can see this instance (for Describe operations).
+// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID).
+func isInstanceVisible(callerAccountID, ownerAccountID string) bool {
+	if ownerAccountID == "" {
+		return callerAccountID == utils.GlobalAccountID
+	}
+	return callerAccountID == ownerAccountID
+}

hive/gateway/auth_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/gofiber/fiber/v2"
 	"github.com/mulgadc/hive/hive/awserrors"
 	handlers_iam "github.com/mulgadc/hive/hive/handlers/iam"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/mulgadc/predastore/auth"
 )
 
@@ -1461,7 +1462,7 @@ func TestCheckPolicy_RootGlobalAccount_Bypasses(t *testing.T) {
 					AccessKeyID:     testAccessKey,
 					SecretAccessKey: encryptedSecret,
 					UserName:        "root",
-					AccountID:       handlers_iam.GlobalAccountID,
+					AccountID:       utils.GlobalAccountID,
 					Status:          "Active",
 				},
 			},

hive/gateway/ec2.go

@@ -98,7 +98,7 @@ var ec2Actions = map[string]EC2Handler{
 		return gateway_ec2_image.DescribeImages(input, gw.NATSConn, accountID)
 	}),
 	"CreateImage": ec2Handler(func(input *ec2.CreateImageInput, gw *GatewayConfig, accountID string) (any, error) {
-		return gateway_ec2_image.CreateImage(input, gw.NATSConn, accountID)
+		return gateway_ec2_image.CreateImage(input, gw.NATSConn, gw.DiscoverActiveNodes(), accountID)
 	}),
 	"DescribeRegions": ec2Handler(func(input *ec2.DescribeRegionsInput, gw *GatewayConfig, accountID string) (any, error) {
 		return gateway_ec2_zone.DescribeRegions(input, gw.Region)

hive/gateway/ec2/image/CreateImage.go

@@ -32,14 +32,14 @@ func ValidateCreateImageInput(input *ec2.CreateImageInput) error {
 }
 
 // CreateImage handles the EC2 CreateImage API call
-func CreateImage(input *ec2.CreateImageInput, natsConn *nats.Conn, accountID string) (ec2.CreateImageOutput, error) {
+func CreateImage(input *ec2.CreateImageInput, natsConn *nats.Conn, expectedNodes int, accountID string) (ec2.CreateImageOutput, error) {
 	var output ec2.CreateImageOutput
 
 	if err := ValidateCreateImageInput(input); err != nil {
 		return output, err
 	}
 
-	svc := handlers_ec2_image.NewNATSImageService(natsConn)
+	svc := handlers_ec2_image.NewNATSImageService(natsConn, expectedNodes)
 	result, err := svc.CreateImage(input, accountID)
 	if err != nil {
 		return output, err

hive/gateway/ec2/image/DescribeImages.go

@@ -37,7 +37,7 @@ func DescribeImages(input *ec2.DescribeImagesInput, natsConn *nats.Conn, account
 	}
 
 	// Create NATS service and call handler
-	imageService := handlers_ec2_image.NewNATSImageService(natsConn)
+	imageService := handlers_ec2_image.NewNATSImageService(natsConn, 0)
 	result, err := imageService.DescribeImages(input, accountID)
 
 	if err != nil {

hive/gateway/ec2/instance/RunInstances.go

@@ -90,7 +90,7 @@ func RunInstances(input *ec2.RunInstancesInput, natsConn *nats.Conn, accountID s
 // isKnownInstanceType checks whether any daemon recognizes the given instance type.
 func isKnownInstanceType(natsConn *nats.Conn, instanceType string) bool {
 	result, err := utils.NATSRequest[ec2.DescribeInstanceTypesOutput](
-		natsConn, "ec2.DescribeInstanceTypes", &ec2.DescribeInstanceTypesInput{}, 3*time.Second, "")
+		natsConn, "ec2.DescribeInstanceTypes", &ec2.DescribeInstanceTypesInput{}, 3*time.Second, utils.GlobalAccountID)
 	if err != nil || result == nil {
 		return false
 	}

hive/gateway/ec2/volume/AttachVolume.go

@@ -74,7 +74,7 @@ func AttachVolume(input *ec2.AttachVolumeInput, natsConn *nats.Conn, accountID s
 		slog.Error("AttachVolume: NATS request failed", "instanceId", instanceID, "volumeId", volumeID, "err", err)
 		if errors.Is(err, nats.ErrNoResponders) {
 			// Check shared KV for stopped instances before returning NotFound
-			if isStoppedInstance(instanceID, natsConn) {
+			if isStoppedInstance(instanceID, natsConn, accountID) {
 				return output, errors.New(awserrors.ErrorIncorrectInstanceState)
 			}
 			return output, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
@@ -98,7 +98,7 @@ func AttachVolume(input *ec2.AttachVolumeInput, natsConn *nats.Conn, accountID s
 
 // isStoppedInstance checks the shared KV (via ec2.DescribeStoppedInstances) to
 // determine whether instanceID exists as a stopped instance.
-func isStoppedInstance(instanceID string, natsConn *nats.Conn) bool {
+func isStoppedInstance(instanceID string, natsConn *nats.Conn, accountID string) bool {
 	input := &ec2.DescribeInstancesInput{
 		InstanceIds: []*string{aws.String(instanceID)},
 	}
@@ -107,7 +107,10 @@ func isStoppedInstance(instanceID string, natsConn *nats.Conn) bool {
 		return false
 	}
 
-	msg, err := natsConn.Request("ec2.DescribeStoppedInstances", reqData, 3*time.Second)
+	reqMsg := nats.NewMsg("ec2.DescribeStoppedInstances")
+	reqMsg.Data = reqData
+	reqMsg.Header.Set(utils.AccountIDHeader, accountID)
+	msg, err := natsConn.RequestMsg(reqMsg, 3*time.Second)
 	if err != nil {
 		return false
 	}

hive/gateway/gateway.go

@@ -19,6 +19,7 @@ import (
 	"github.com/mulgadc/hive/hive/awserrors"
 	"github.com/mulgadc/hive/hive/gateway/policy"
 	handlers_iam "github.com/mulgadc/hive/hive/handlers/iam"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats.go"
 )
 
@@ -199,7 +200,7 @@ func (gw *GatewayConfig) checkPolicy(ctx *fiber.Ctx, service, action string) err
 		return errors.New(awserrors.ErrorInternalError)
 	}
 
-	if identity == "" || (identity == "root" && accountID == handlers_iam.GlobalAccountID) {
+	if identity == "" || (identity == "root" && accountID == utils.GlobalAccountID) {
 		return nil
 	}
 

hive/handlers/ec2/account/service_impl.go

@@ -39,7 +39,7 @@ func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Co
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	settingsKV, err := getOrCreateKVBucket(js, KVBucketAccountSettings, 10)
+	settingsKV, err := utils.GetOrCreateKVBucket(js, KVBucketAccountSettings, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create account settings KV bucket: %w", err)
 	}
@@ -53,25 +53,11 @@ func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Co
 	}, nil
 }
 
-func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) {
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
-		Bucket:  bucketName,
-		History: utils.SafeIntToUint8(history),
-	})
-	if err != nil {
-		kv, err = js.KeyValue(bucketName)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return kv, nil
-}
-
 // settingsKey returns the per-account KV key for storing settings.
-// Falls back to "default" for pre-Phase-4 resources with no accountID.
+// Falls back to GlobalAccountID for pre-Phase-4 resources with no accountID.
 func settingsKey(accountID string) string {
 	if accountID == "" {
-		return "default"
+		return utils.GlobalAccountID
 	}
 	return accountID
 }

hive/handlers/ec2/account/service_impl_test.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
@@ -169,6 +170,6 @@ func TestSerialConsole_MultiAccountIsolation(t *testing.T) {
 }
 
 func TestSettingsKey_EmptyAccountIDFallback(t *testing.T) {
-	assert.Equal(t, "default", settingsKey(""))
+	assert.Equal(t, utils.GlobalAccountID, settingsKey(""))
 	assert.Equal(t, "123456789012", settingsKey("123456789012"))
 }

hive/handlers/ec2/eigw/service_impl.go

@@ -12,6 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/hive/hive/awserrors"
 	"github.com/mulgadc/hive/hive/config"
+	handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc"
 	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -20,7 +21,6 @@ import (
 var _ EgressOnlyIGWService = (*EgressOnlyIGWServiceImpl)(nil)
 
 const KVBucketEgressOnlyIGW = "hive-eigw"
-const kvBucketVPCs = "hive-vpc-vpcs"
 
 // EgressOnlyIGWRecord represents a stored Egress-only Internet Gateway
 type EgressOnlyIGWRecord struct {
@@ -38,13 +38,6 @@ type EgressOnlyIGWServiceImpl struct {
 	vpcKV  nats.KeyValue
 }
 
-// NewEgressOnlyIGWServiceImpl creates a new Egress-only Internet Gateway service implementation
-func NewEgressOnlyIGWServiceImpl(cfg *config.Config) *EgressOnlyIGWServiceImpl {
-	return &EgressOnlyIGWServiceImpl{
-		config: cfg,
-	}
-}
-
 // NewEgressOnlyIGWServiceImplWithNATS creates an Egress-only Internet Gateway service with NATS JetStream for persistence
 func NewEgressOnlyIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*EgressOnlyIGWServiceImpl, error) {
 	js, err := natsConn.JetStream()
@@ -52,15 +45,15 @@ func NewEgressOnlyIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	eigwKV, err := getOrCreateKVBucket(js, KVBucketEgressOnlyIGW, 10)
+	eigwKV, err := utils.GetOrCreateKVBucket(js, KVBucketEgressOnlyIGW, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEgressOnlyIGW, err)
 	}
 
-	// Get VPC KV bucket for cross-resource ownership validation
-	vpcKV, err := js.KeyValue(kvBucketVPCs)
+	// Get or create VPC KV bucket for cross-resource ownership validation
+	vpcKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketVPCs, 10)
 	if err != nil {
-		slog.Warn("EIGW service: VPC KV bucket not available, VPC ownership checks disabled", "error", err)
+		return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
 	}
 
 	slog.Info("Egress-only IGW service initialized with JetStream KV", "bucket", KVBucketEgressOnlyIGW)
@@ -72,36 +65,20 @@ func NewEgressOnlyIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn
 	}, nil
 }
 
-func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) {
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
-		Bucket:  bucketName,
-		History: utils.SafeIntToUint8(history),
-	})
-	if err != nil {
-		kv, err = js.KeyValue(bucketName)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return kv, nil
-}
-
-func accountKey(accountID, resourceID string) string {
-	return accountID + "." + resourceID
-}
-
 // CreateEgressOnlyInternetGateway creates a new Egress-only Internet Gateway
 func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(input *ec2.CreateEgressOnlyInternetGatewayInput, accountID string) (*ec2.CreateEgressOnlyInternetGatewayOutput, error) {
 	if input.VpcId == nil || *input.VpcId == "" {
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
 
-	// Verify the caller owns the target VPC
-	if s.vpcKV != nil {
-		if _, err := s.vpcKV.Get(accountKey(accountID, *input.VpcId)); err != nil {
-			slog.Warn("CreateEgressOnlyInternetGateway: VPC not found for account", "vpcId", *input.VpcId, "accountID", accountID)
-			return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound)
-		}
+	// Verify the caller owns the target VPC (fail-closed if KV unavailable)
+	if s.vpcKV == nil {
+		slog.Error("VPC KV unavailable, cannot verify VPC ownership")
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	if _, err := s.vpcKV.Get(utils.AccountKey(accountID, *input.VpcId)); err != nil {
+		slog.Warn("CreateEgressOnlyInternetGateway: VPC not found for account", "vpcId", *input.VpcId, "accountID", accountID)
+		return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound)
 	}
 
 	eigwID := utils.GenerateResourceID("eigw")
@@ -128,7 +105,7 @@ func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(input *ec2.Cr
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal Egress-only IGW record: %w", err)
 	}
-	if _, err := s.eigwKV.Put(accountKey(accountID, eigwID), data); err != nil {
+	if _, err := s.eigwKV.Put(utils.AccountKey(accountID, eigwID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -146,12 +123,15 @@ func (s *EgressOnlyIGWServiceImpl) DeleteEgressOnlyInternetGateway(input *ec2.De
 	}
 
 	eigwID := *input.EgressOnlyInternetGatewayId
-	key := accountKey(accountID, eigwID)
+	key := utils.AccountKey(accountID, eigwID)
 
+	// Verify the EIGW exists before deleting
 	if _, err := s.eigwKV.Get(key); err != nil {
-		return nil, errors.New(awserrors.ErrorInvalidEgressOnlyInternetGatewayIdNotFound)
+		if errors.Is(err, nats.ErrKeyNotFound) {
+			return nil, errors.New(awserrors.ErrorInvalidEgressOnlyInternetGatewayIdNotFound)
+		}
+		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
-
 	if err := s.eigwKV.Delete(key); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}

hive/handlers/ec2/eigw/service_impl_test.go

@@ -6,6 +6,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
@@ -34,6 +36,16 @@ func setupTestEIGWService(t *testing.T) *EgressOnlyIGWServiceImpl {
 	require.NoError(t, err)
 	t.Cleanup(func() { nc.Close() })
 
+	// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
+	js, err := nc.JetStream()
+	require.NoError(t, err)
+	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
+	require.NoError(t, err)
+	for _, vpcID := range []string{"vpc-test123", "vpc-tagged", "vpc-tagged2"} {
+		_, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`))
+		require.NoError(t, err)
+	}
+
 	svc, err := NewEgressOnlyIGWServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)
 	return svc
@@ -211,11 +223,11 @@ func TestCreateEgressOnlyInternetGateway_CrossAccountVPCRejected(t *testing.T) {
 	// Create VPC KV bucket with a VPC owned by testAccountID
 	js, err := nc.JetStream()
 	require.NoError(t, err)
-	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: kvBucketVPCs, History: 1})
+	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
 	require.NoError(t, err)
 
 	vpcID := "vpc-alpha456"
-	_, err = vpcKV.Put(accountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`))
+	_, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`))
 	require.NoError(t, err)
 
 	// Create EIGW service (VPC KV will be populated since we created it before EIGW init)

hive/handlers/ec2/igw/service_impl.go

@@ -12,6 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/hive/hive/awserrors"
 	"github.com/mulgadc/hive/hive/config"
+	handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc"
 	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -20,7 +21,6 @@ import (
 var _ IGWService = (*IGWServiceImpl)(nil)
 
 const KVBucketIGW = "hive-igw"
-const kvBucketVPCs = "hive-vpc-vpcs"
 
 // IGWRecord represents a stored Internet Gateway
 type IGWRecord struct {
@@ -39,13 +39,6 @@ type IGWServiceImpl struct {
 	natsConn *nats.Conn
 }
 
-// NewIGWServiceImpl creates a new Internet Gateway service without NATS persistence
-func NewIGWServiceImpl(cfg *config.Config) *IGWServiceImpl {
-	return &IGWServiceImpl{
-		config: cfg,
-	}
-}
-
 // NewIGWServiceImplWithNATS creates an Internet Gateway service with NATS JetStream for persistence
 func NewIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*IGWServiceImpl, error) {
 	js, err := natsConn.JetStream()
@@ -53,15 +46,15 @@ func NewIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*IGWSer
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	igwKV, err := getOrCreateKVBucket(js, KVBucketIGW, 10)
+	igwKV, err := utils.GetOrCreateKVBucket(js, KVBucketIGW, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketIGW, err)
 	}
 
-	// Get VPC KV bucket for cross-resource ownership validation
-	vpcKV, err := js.KeyValue(kvBucketVPCs)
+	// Get or create VPC KV bucket for cross-resource ownership validation
+	vpcKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketVPCs, 10)
 	if err != nil {
-		slog.Warn("IGW service: VPC KV bucket not available, VPC ownership checks disabled", "error", err)
+		return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
 	}
 
 	slog.Info("IGW service initialized with JetStream KV", "bucket", KVBucketIGW)
@@ -74,24 +67,6 @@ func NewIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*IGWSer
 	}, nil
 }
 
-func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) {
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
-		Bucket:  bucketName,
-		History: utils.SafeIntToUint8(history),
-	})
-	if err != nil {
-		kv, err = js.KeyValue(bucketName)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return kv, nil
-}
-
-func accountKey(accountID, resourceID string) string {
-	return accountID + "." + resourceID
-}
-
 // CreateInternetGateway creates a new Internet Gateway (initially detached)
 func (s *IGWServiceImpl) CreateInternetGateway(input *ec2.CreateInternetGatewayInput, accountID string) (*ec2.CreateInternetGatewayOutput, error) {
 	igwID := utils.GenerateResourceID("igw")
@@ -117,7 +92,7 @@ func (s *IGWServiceImpl) CreateInternetGateway(input *ec2.CreateInternetGatewayI
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal IGW record: %w", err)
 	}
-	if _, err := s.igwKV.Put(accountKey(accountID, igwID), data); err != nil {
+	if _, err := s.igwKV.Put(utils.AccountKey(accountID, igwID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -135,7 +110,7 @@ func (s *IGWServiceImpl) DeleteInternetGateway(input *ec2.DeleteInternetGatewayI
 	}
 
 	igwID := *input.InternetGatewayId
-	key := accountKey(accountID, igwID)
+	key := utils.AccountKey(accountID, igwID)
 
 	entry, err := s.igwKV.Get(key)
 	if err != nil {
@@ -231,7 +206,7 @@ func (s *IGWServiceImpl) AttachInternetGateway(input *ec2.AttachInternetGatewayI
 
 	igwID := *input.InternetGatewayId
 	vpcID := *input.VpcId
-	key := accountKey(accountID, igwID)
+	key := utils.AccountKey(accountID, igwID)
 
 	entry, err := s.igwKV.Get(key)
 	if err != nil {
@@ -247,12 +222,14 @@ func (s *IGWServiceImpl) AttachInternetGateway(input *ec2.AttachInternetGatewayI
 		return nil, errors.New(awserrors.ErrorResourceAlreadyAssociated)
 	}
 
-	// Verify the caller owns the target VPC
-	if s.vpcKV != nil {
-		if _, err := s.vpcKV.Get(accountKey(accountID, vpcID)); err != nil {
-			slog.Warn("AttachInternetGateway: VPC not found for account", "vpcId", vpcID, "accountID", accountID)
-			return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound)
-		}
+	// Verify the caller owns the target VPC (fail-closed if KV unavailable)
+	if s.vpcKV == nil {
+		slog.Error("VPC KV unavailable, cannot verify VPC ownership")
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	if _, err := s.vpcKV.Get(utils.AccountKey(accountID, vpcID)); err != nil {
+		slog.Warn("AttachInternetGateway: VPC not found for account", "vpcId", vpcID, "accountID", accountID)
+		return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound)
 	}
 
 	record.VpcId = vpcID
@@ -295,7 +272,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(input *ec2.DetachInternetGatewayI
 
 	igwID := *input.InternetGatewayId
 	vpcID := *input.VpcId
-	key := accountKey(accountID, igwID)
+	key := utils.AccountKey(accountID, igwID)
 
 	entry, err := s.igwKV.Get(key)
 	if err != nil {

hive/handlers/ec2/igw/service_impl_test.go

@@ -6,6 +6,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
@@ -32,6 +34,16 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) {
 	require.NoError(t, err)
 	t.Cleanup(func() { nc.Close() })
 
+	// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
+	js, err := nc.JetStream()
+	require.NoError(t, err)
+	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
+	require.NoError(t, err)
+	for _, vpcID := range []string{"vpc-test123", "vpc-other", "vpc-lifecycle", "vpc-event-test"} {
+		_, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`))
+		require.NoError(t, err)
+	}
+
 	svc, err := NewIGWServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)
 	return svc, nc
@@ -443,14 +455,6 @@ func TestCreateInternetGateway_PublishesNoEvent(t *testing.T) {
 	}
 }
 
-func TestNewIGWServiceImpl(t *testing.T) {
-	svc := NewIGWServiceImpl(nil)
-	require.NotNil(t, svc)
-	// Without KV, IGW operations will panic (nil KV), so just verify constructor
-	assert.Nil(t, svc.igwKV)
-	assert.Nil(t, svc.natsConn)
-}
-
 // TestAttachInternetGateway_CrossAccountVPCRejected tests that attaching an IGW to another account's VPC is rejected.
 func TestAttachInternetGateway_CrossAccountVPCRejected(t *testing.T) {
 	svc, nc := setupTestIGWService(t)
@@ -458,11 +462,11 @@ func TestAttachInternetGateway_CrossAccountVPCRejected(t *testing.T) {
 	// Create VPC KV bucket and add a VPC owned by testAccountID
 	js, err := nc.JetStream()
 	require.NoError(t, err)
-	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: kvBucketVPCs, History: 1})
+	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
 	require.NoError(t, err)
 
 	vpcID := "vpc-alpha123"
-	_, err = vpcKV.Put(accountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
+	_, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
 	require.NoError(t, err)
 
 	// Refresh service to pick up the VPC KV bucket

hive/handlers/ec2/igw/service_nats_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -44,7 +45,7 @@ func handleNATSMsg[In any, Out any](msg *nats.Msg, fn func(*In, string) (*Out, e
 		_ = msg.Respond([]byte(`{"error":"unmarshal"}`))
 		return
 	}
-	accountID := msg.Header.Get("X-Account-ID")
+	accountID := msg.Header.Get(utils.AccountIDHeader)
 	result, err := fn(&input, accountID)
 	if err != nil {
 		errResp, _ := json.Marshal(map[string]string{"error": err.Error()})

hive/handlers/ec2/image/service_impl.go

@@ -163,7 +163,7 @@ func (s *ImageServiceImpl) DescribeImages(input *ec2.DescribeImagesInput, accoun
 		// in ImageOwnerAlias. Pre-phase4 AMIs have non-account values like "self"
 		// or "hive" and are treated as system/public images visible to all.
 		amiOwner := amiMeta.ImageOwnerAlias
-		isSystemAMI := !isAccountID(amiOwner)
+		isSystemAMI := !utils.IsAccountID(amiOwner)
 
 		// Visibility check: callers can only see their own AMIs and system AMIs.
 		// This runs regardless of whether an owner filter is specified.
@@ -202,7 +202,7 @@ func (s *ImageServiceImpl) DescribeImages(input *ec2.DescribeImagesInput, accoun
 		// Resolve the OwnerId for the response. System AMIs use global account.
 		ownerID := amiOwner
 		if isSystemAMI {
-			ownerID = "000000000000"
+			ownerID = utils.GlobalAccountID
 		}
 
 		// Build EC2 Image from AMIMetadata
@@ -593,20 +593,6 @@ func (s *ImageServiceImpl) putSnapshotMetadata(snapshotID, volumeID string, volu
 	return err
 }
 
-// isAccountID returns true if the string looks like a valid Hive account ID
-// (12-digit zero-padded number). Pre-phase4 values like "self" or "hive" return false.
-func isAccountID(s string) bool {
-	if len(s) != 12 {
-		return false
-	}
-	for _, c := range s {
-		if c < '0' || c > '9' {
-			return false
-		}
-	}
-	return true
-}
-
 func (s *ImageServiceImpl) CopyImage(input *ec2.CopyImageInput, accountID string) (*ec2.CopyImageOutput, error) {
 	return nil, errors.New("CopyImage not yet implemented")
 }

hive/handlers/ec2/image/service_nats.go

@@ -11,12 +11,15 @@ import (
 
 // NATSImageService handles image operations via NATS messaging
 type NATSImageService struct {
-	natsConn *nats.Conn
+	natsConn      *nats.Conn
+	expectedNodes int
 }
 
-// NewNATSImageService creates a new NATS-based image service
-func NewNATSImageService(conn *nats.Conn) ImageService {
-	return &NATSImageService{natsConn: conn}
+// NewNATSImageService creates a new NATS-based image service.
+// expectedNodes is used by scatter-gather operations (e.g. CreateImage) to
+// enable early exit once all nodes have responded.
+func NewNATSImageService(conn *nats.Conn, expectedNodes int) ImageService {
+	return &NATSImageService{natsConn: conn, expectedNodes: expectedNodes}
 }
 
 func (s *NATSImageService) DescribeImages(input *ec2.DescribeImagesInput, accountID string) (*ec2.DescribeImagesOutput, error) {
@@ -24,7 +27,7 @@ func (s *NATSImageService) DescribeImages(input *ec2.DescribeImagesInput, accoun
 }
 
 func (s *NATSImageService) CreateImage(input *ec2.CreateImageInput, accountID string) (*ec2.CreateImageOutput, error) {
-	return utils.NATSRequest[ec2.CreateImageOutput](s.natsConn, "ec2.CreateImage", input, 120*time.Second, accountID)
+	return utils.NATSScatterGather[ec2.CreateImageOutput](s.natsConn, "ec2.CreateImage", input, 120*time.Second, s.expectedNodes, accountID)
 }
 
 func (s *NATSImageService) CopyImage(input *ec2.CopyImageInput, accountID string) (*ec2.CopyImageOutput, error) {

hive/handlers/ec2/snapshot/service_impl.go

@@ -65,7 +65,7 @@ func NewSnapshotServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*S
 		return nil, nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	kv, err := getOrCreateKVBucket(js, KVBucketVolumeSnapshots, 10)
+	kv, err := utils.GetOrCreateKVBucket(js, KVBucketVolumeSnapshots, 10)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVolumeSnapshots, err)
 	}
@@ -94,20 +94,6 @@ func NewSnapshotServiceImplWithStore(cfg *config.Config, store objectstore.Objec
 	return svc
 }
 
-func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) {
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
-		Bucket:  bucketName,
-		History: utils.SafeIntToUint8(history),
-	})
-	if err != nil {
-		kv, err = js.KeyValue(bucketName)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return kv, nil
-}
-
 // getSnapshotKey returns the S3 key for storing snapshot metadata.
 // Uses metadata.json to avoid conflicting with viperblock's config.json
 // which stores the SnapshotState (block map references, source volume, etc).

hive/handlers/ec2/vpc/eni.go

@@ -38,9 +38,6 @@ type ENIRecord struct {
 
 // CreateNetworkInterface creates a new ENI in the specified subnet
 func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	if input.SubnetId == nil || *input.SubnetId == "" {
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
@@ -48,7 +45,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfac
 	subnetId := *input.SubnetId
 
 	// Verify subnet exists and belongs to this account
-	subnetEntry, err := s.subnetKV.Get(accountKey(accountID, subnetId))
+	subnetEntry, err := s.subnetKV.Get(utils.AccountKey(accountID, subnetId))
 	if err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
 	}
@@ -108,7 +105,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfac
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal ENI record: %w", err)
 	}
-	if _, err := s.eniKV.Put(accountKey(accountID, eniId), data); err != nil {
+	if _, err := s.eniKV.Put(utils.AccountKey(accountID, eniId), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -124,15 +121,12 @@ func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfac
 
 // DeleteNetworkInterface deletes an ENI
 func (s *VPCServiceImpl) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInput, accountID string) (*ec2.DeleteNetworkInterfaceOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	if input.NetworkInterfaceId == nil || *input.NetworkInterfaceId == "" {
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
 
 	eniId := *input.NetworkInterfaceId
-	key := accountKey(accountID, eniId)
+	key := utils.AccountKey(accountID, eniId)
 
 	// Get the ENI record
 	entry, err := s.eniKV.Get(key)
@@ -169,9 +163,6 @@ func (s *VPCServiceImpl) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfac
 
 // DescribeNetworkInterfaces lists ENIs with optional filters
 func (s *VPCServiceImpl) DescribeNetworkInterfaces(input *ec2.DescribeNetworkInterfacesInput, accountID string) (*ec2.DescribeNetworkInterfacesOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	var enis []*ec2.NetworkInterface
 
 	eniIDs := make(map[string]bool)
@@ -264,10 +255,7 @@ func (s *VPCServiceImpl) DescribeNetworkInterfaces(input *ec2.DescribeNetworkInt
 // AttachENI marks an ENI as attached to an instance (internal use by RunInstances).
 // accountID scopes the lookup to the correct KV key.
 func (s *VPCServiceImpl) AttachENI(accountID, eniId, instanceId string, deviceIndex int64) (string, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return "", errors.New(awserrors.ErrorServerInternal)
-	}
-	key := accountKey(accountID, eniId)
+	key := utils.AccountKey(accountID, eniId)
 	entry, err := s.eniKV.Get(key)
 	if err != nil {
 		return "", errors.New(awserrors.ErrorInvalidNetworkInterfaceIDNotFound)
@@ -303,10 +291,7 @@ func (s *VPCServiceImpl) AttachENI(accountID, eniId, instanceId string, deviceIn
 // DetachENI marks an ENI as detached from an instance (internal use by TerminateInstances).
 // accountID scopes the lookup to the correct KV key.
 func (s *VPCServiceImpl) DetachENI(accountID, eniId string) error {
-	if err := s.ensureKVReady(); err != nil {
-		return errors.New(awserrors.ErrorServerInternal)
-	}
-	key := accountKey(accountID, eniId)
+	key := utils.AccountKey(accountID, eniId)
 	entry, err := s.eniKV.Get(key)
 	if err != nil {
 		return errors.New(awserrors.ErrorInvalidNetworkInterfaceIDNotFound)

hive/handlers/ec2/vpc/ipam.go

@@ -8,6 +8,7 @@ import (
 	"math/big"
 	"net"
 
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats.go"
 )
 
@@ -29,7 +30,7 @@ type IPAM struct {
 
 // NewIPAM creates a new IPAM instance backed by NATS JetStream KV.
 func NewIPAM(js nats.JetStreamContext) (*IPAM, error) {
-	kv, err := getOrCreateKVBucket(js, KVBucketIPAM, 5)
+	kv, err := utils.GetOrCreateKVBucket(js, KVBucketIPAM, 5)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create IPAM KV bucket: %w", err)
 	}

hive/handlers/ec2/vpc/service_impl.go

@@ -62,13 +62,6 @@ type VPCServiceImpl struct {
 	ipam     *IPAM
 }
 
-// NewVPCServiceImpl creates a new VPC service implementation (in-memory, no persistence)
-func NewVPCServiceImpl(cfg *config.Config) *VPCServiceImpl {
-	return &VPCServiceImpl{
-		config: cfg,
-	}
-}
-
 // NewVPCServiceImplWithNATS creates a VPC service with NATS JetStream for persistence
 func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCServiceImpl, error) {
 	js, err := natsConn.JetStream()
@@ -76,22 +69,22 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	vpcKV, err := getOrCreateKVBucket(js, KVBucketVPCs, 10)
+	vpcKV, err := utils.GetOrCreateKVBucket(js, KVBucketVPCs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVPCs, err)
 	}
 
-	subnetKV, err := getOrCreateKVBucket(js, KVBucketSubnets, 10)
+	subnetKV, err := utils.GetOrCreateKVBucket(js, KVBucketSubnets, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSubnets, err)
 	}
 
-	vniKV, err := getOrCreateKVBucket(js, KVBucketVNICounter, 10)
+	vniKV, err := utils.GetOrCreateKVBucket(js, KVBucketVNICounter, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVNICounter, err)
 	}
 
-	eniKV, err := getOrCreateKVBucket(js, KVBucketENIs, 10)
+	eniKV, err := utils.GetOrCreateKVBucket(js, KVBucketENIs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketENIs, err)
 	}
@@ -118,35 +111,6 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer
 	}, nil
 }
 
-func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) {
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
-		Bucket:  bucketName,
-		History: utils.SafeIntToUint8(history),
-	})
-	if err != nil {
-		kv, err = js.KeyValue(bucketName)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return kv, nil
-}
-
-// errKVNotReady is returned when the VPC service was initialized without NATS KV
-// (in-memory fallback mode). Callers receive an internal server error.
-var errKVNotReady = errors.New("VPC KV stores not initialized (in-memory fallback mode)")
-
-// ensureKVReady returns an error if any required KV store is nil.
-// This guards against nil pointer dereferences when the daemon fell back to
-// NewVPCServiceImpl (no NATS) but still receives VPC requests via queue group.
-func (s *VPCServiceImpl) ensureKVReady() error {
-	if s.vpcKV == nil || s.subnetKV == nil || s.vniKV == nil || s.eniKV == nil {
-		slog.Error("VPC service called without KV stores initialized")
-		return errKVNotReady
-	}
-	return nil
-}
-
 // nextVNI allocates the next VNI using atomic increment on the NATS KV counter
 func (s *VPCServiceImpl) nextVNI() (int64, error) {
 	entry, err := s.vniKV.Get(vniCounterKey)
@@ -177,16 +141,8 @@ func (s *VPCServiceImpl) nextVNI() (int64, error) {
 	return current, nil
 }
 
-// accountKey returns a KV key scoped to the given account: "{accountID}.{resourceID}".
-func accountKey(accountID, resourceID string) string {
-	return accountID + "." + resourceID
-}
-
 // CreateVpc creates a new VPC
 func (s *VPCServiceImpl) CreateVpc(input *ec2.CreateVpcInput, accountID string) (*ec2.CreateVpcOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	if input.CidrBlock == nil || *input.CidrBlock == "" {
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
@@ -235,7 +191,7 @@ func (s *VPCServiceImpl) CreateVpc(input *ec2.CreateVpcInput, accountID string)
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal VPC record: %w", err)
 	}
-	if _, err := s.vpcKV.Put(accountKey(accountID, vpcID), data); err != nil {
+	if _, err := s.vpcKV.Put(utils.AccountKey(accountID, vpcID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -251,15 +207,12 @@ func (s *VPCServiceImpl) CreateVpc(input *ec2.CreateVpcInput, accountID string)
 
 // DeleteVpc deletes a VPC
 func (s *VPCServiceImpl) DeleteVpc(input *ec2.DeleteVpcInput, accountID string) (*ec2.DeleteVpcOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	if input.VpcId == nil || *input.VpcId == "" {
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
 
 	vpcID := *input.VpcId
-	key := accountKey(accountID, vpcID)
+	key := utils.AccountKey(accountID, vpcID)
 
 	if _, err := s.vpcKV.Get(key); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound)
@@ -302,9 +255,6 @@ func (s *VPCServiceImpl) DeleteVpc(input *ec2.DeleteVpcInput, accountID string)
 
 // DescribeVpcs describes VPCs
 func (s *VPCServiceImpl) DescribeVpcs(input *ec2.DescribeVpcsInput, accountID string) (*ec2.DescribeVpcsOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	var vpcs []*ec2.Vpc
 
 	vpcIDs := make(map[string]bool)
@@ -344,6 +294,32 @@ func (s *VPCServiceImpl) DescribeVpcs(input *ec2.DescribeVpcsInput, accountID st
 		vpcs = append(vpcs, s.vpcRecordToEC2(&record, accountID))
 	}
 
+	// Include global default VPCs (created at startup) for non-root accounts.
+	// This matches AWS behavior where all accounts have a default VPC.
+	if accountID != utils.GlobalAccountID {
+		globalPrefix := utils.GlobalAccountID + "."
+		for _, key := range keys {
+			if !strings.HasPrefix(key, globalPrefix) {
+				continue
+			}
+			entry, err := s.vpcKV.Get(key)
+			if err != nil {
+				continue
+			}
+			var record VPCRecord
+			if err := json.Unmarshal(entry.Value(), &record); err != nil {
+				continue
+			}
+			if !record.IsDefault {
+				continue
+			}
+			if len(vpcIDs) > 0 && !vpcIDs[record.VpcId] {
+				continue
+			}
+			vpcs = append(vpcs, s.vpcRecordToEC2(&record, accountID))
+		}
+	}
+
 	// If specific VPC IDs were requested but not found, return error
 	if len(vpcIDs) > 0 {
 		found := make(map[string]bool)
@@ -368,9 +344,6 @@ func (s *VPCServiceImpl) DescribeVpcs(input *ec2.DescribeVpcsInput, accountID st
 
 // CreateSubnet creates a new subnet within a VPC
 func (s *VPCServiceImpl) CreateSubnet(input *ec2.CreateSubnetInput, accountID string) (*ec2.CreateSubnetOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	if input.VpcId == nil || *input.VpcId == "" {
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
@@ -381,7 +354,7 @@ func (s *VPCServiceImpl) CreateSubnet(input *ec2.CreateSubnetInput, accountID st
 	vpcID := *input.VpcId
 
 	// Verify VPC exists and belongs to this account
-	vpcEntry, err := s.vpcKV.Get(accountKey(accountID, vpcID))
+	vpcEntry, err := s.vpcKV.Get(utils.AccountKey(accountID, vpcID))
 	if err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound)
 	}
@@ -483,7 +456,7 @@ func (s *VPCServiceImpl) CreateSubnet(input *ec2.CreateSubnetInput, accountID st
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal subnet record: %w", err)
 	}
-	if _, err := s.subnetKV.Put(accountKey(accountID, subnetID), data); err != nil {
+	if _, err := s.subnetKV.Put(utils.AccountKey(accountID, subnetID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -499,15 +472,12 @@ func (s *VPCServiceImpl) CreateSubnet(input *ec2.CreateSubnetInput, accountID st
 
 // DeleteSubnet deletes a subnet
 func (s *VPCServiceImpl) DeleteSubnet(input *ec2.DeleteSubnetInput, accountID string) (*ec2.DeleteSubnetOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	if input.SubnetId == nil || *input.SubnetId == "" {
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
 
 	subnetID := *input.SubnetId
-	key := accountKey(accountID, subnetID)
+	key := utils.AccountKey(accountID, subnetID)
 
 	// Read subnet record before deletion (needed for vpcd event)
 	subnetEntry, err := s.subnetKV.Get(key)
@@ -531,9 +501,6 @@ func (s *VPCServiceImpl) DeleteSubnet(input *ec2.DeleteSubnetInput, accountID st
 
 // DescribeSubnets describes subnets
 func (s *VPCServiceImpl) DescribeSubnets(input *ec2.DescribeSubnetsInput, accountID string) (*ec2.DescribeSubnetsOutput, error) {
-	if err := s.ensureKVReady(); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
 	var subnets []*ec2.Subnet
 
 	subnetIDs := make(map[string]bool)
@@ -677,11 +644,6 @@ const (
 	DefaultSubnetCidr = "172.31.0.0/20"
 )
 
-// GlobalAccountID is used for resources created without an explicit account context
-// (e.g., default VPC at daemon startup). Pre-existing resources with no accountID
-// are treated as owned by this account.
-const GlobalAccountID = "000000000000"
-
 // EnsureDefaultVPC creates a default VPC and subnet if none exists for the given account.
 // This matches AWS behavior where a default VPC is present on account creation.
 // Safe to call multiple times — no-ops if a default VPC already exists.
@@ -736,7 +698,7 @@ func (s *VPCServiceImpl) EnsureDefaultVPC(accountID string) error {
 	if err != nil {
 		return fmt.Errorf("marshal default VPC: %w", err)
 	}
-	if _, err := s.vpcKV.Put(accountKey(accountID, vpcID), data); err != nil {
+	if _, err := s.vpcKV.Put(utils.AccountKey(accountID, vpcID), data); err != nil {
 		return fmt.Errorf("store default VPC: %w", err)
 	}
 
@@ -765,7 +727,7 @@ func (s *VPCServiceImpl) EnsureDefaultVPC(accountID string) error {
 	if err != nil {
 		return fmt.Errorf("marshal default subnet: %w", err)
 	}
-	if _, err := s.subnetKV.Put(accountKey(accountID, subnetID), data); err != nil {
+	if _, err := s.subnetKV.Put(utils.AccountKey(accountID, subnetID), data); err != nil {
 		return fmt.Errorf("store default subnet: %w", err)
 	}
 
@@ -784,10 +746,6 @@ func (s *VPCServiceImpl) EnsureDefaultVPC(accountID string) error {
 
 // GetDefaultSubnet returns the default subnet for RunInstances when no SubnetId is specified.
 func (s *VPCServiceImpl) GetDefaultSubnet(accountID string) (*SubnetRecord, error) {
-	if s.subnetKV == nil {
-		return nil, fmt.Errorf("VPC service not initialized")
-	}
-
 	prefix := accountID + "."
 	keys, err := s.subnetKV.Keys()
 	if err != nil {

hive/handlers/ec2/vpc/service_impl_test.go

@@ -642,74 +642,6 @@ func TestVpcCidrBlockAssociation(t *testing.T) {
 	assert.Equal(t, "associated", *out.Vpc.CidrBlockAssociationSet[0].CidrBlockState.State)
 }
 
-// --- KV nil guard tests (mulga-x83) ---
-
-func TestKVNilGuard_CreateVpc(t *testing.T) {
-	svc := NewVPCServiceImpl(nil) // in-memory fallback, all KV fields nil
-	_, err := svc.CreateVpc(&ec2.CreateVpcInput{CidrBlock: aws.String("10.0.0.0/16")}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
-func TestKVNilGuard_DeleteVpc(t *testing.T) {
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.DeleteVpc(&ec2.DeleteVpcInput{VpcId: aws.String("vpc-123")}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
-func TestKVNilGuard_DescribeVpcs(t *testing.T) {
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.DescribeVpcs(&ec2.DescribeVpcsInput{}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
-func TestKVNilGuard_CreateSubnet(t *testing.T) {
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.CreateSubnet(&ec2.CreateSubnetInput{
-		VpcId:     aws.String("vpc-123"),
-		CidrBlock: aws.String("10.0.1.0/24"),
-	}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
-func TestKVNilGuard_DeleteSubnet(t *testing.T) {
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.DeleteSubnet(&ec2.DeleteSubnetInput{SubnetId: aws.String("subnet-123")}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
-func TestKVNilGuard_DescribeSubnets(t *testing.T) {
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.DescribeSubnets(&ec2.DescribeSubnetsInput{}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
-func TestKVNilGuard_CreateNetworkInterface(t *testing.T) {
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.CreateNetworkInterface(&ec2.CreateNetworkInterfaceInput{SubnetId: aws.String("subnet-123")}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
-func TestKVNilGuard_DeleteNetworkInterface(t *testing.T) {
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{NetworkInterfaceId: aws.String("eni-123")}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
-func TestKVNilGuard_DescribeNetworkInterfaces(t *testing.T) {
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{}, testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "ServerInternal")
-}
-
 // --- Filter tests ---
 
 func TestDescribeVpcs_NilFields(t *testing.T) {
@@ -883,14 +815,6 @@ func TestEnsureDefaultVPC_WithConfigAZ(t *testing.T) {
 	assert.Equal(t, "us-west-2b", *subDesc.Subnets[0].AvailabilityZone)
 }
 
-func TestGetDefaultSubnet_NoKV(t *testing.T) {
-	// Service without KV should return error
-	svc := NewVPCServiceImpl(nil)
-	_, err := svc.GetDefaultSubnet(testAccountID)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "not initialized")
-}
-
 func TestCreateVpc_NormalizesNetworkCidr(t *testing.T) {
 	svc := setupTestVPCService(t)
 	// Pass a CIDR with host bits set — should be normalized to network address

hive/handlers/iam/account_test.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -229,28 +230,28 @@ func TestSeedRootUser_AccountScoped(t *testing.T) {
 	err = svc.SeedRootUser(&BootstrapData{
 		AccessKeyID:     "AKIAROOTEXAMPLE12345",
 		EncryptedSecret: encryptedSecret,
-		AccountID:       GlobalAccountID,
+		AccountID:       utils.GlobalAccountID,
 	})
 	require.NoError(t, err)
 
 	// Root user stored at 000000000000:root
-	out, err := svc.GetUser(GlobalAccountID, &iam.GetUserInput{
+	out, err := svc.GetUser(utils.GlobalAccountID, &iam.GetUserInput{
 		UserName: aws.String("root"),
 	})
 	require.NoError(t, err)
 	assert.Equal(t, "root", *out.User.UserName)
-	assert.Contains(t, *out.User.Arn, GlobalAccountID)
+	assert.Contains(t, *out.User.Arn, utils.GlobalAccountID)
 
 	// Access key has correct AccountID
 	ak, err := svc.LookupAccessKey("AKIAROOTEXAMPLE12345")
 	require.NoError(t, err)
-	assert.Equal(t, GlobalAccountID, ak.AccountID)
+	assert.Equal(t, utils.GlobalAccountID, ak.AccountID)
 	assert.Equal(t, "root", ak.UserName)
 
 	// Global account record was created
-	account, err := svc.GetAccount(GlobalAccountID)
+	account, err := svc.GetAccount(utils.GlobalAccountID)
 	require.NoError(t, err)
-	assert.Equal(t, GlobalAccountID, account.AccountID)
+	assert.Equal(t, utils.GlobalAccountID, account.AccountID)
 	assert.Equal(t, "Global", account.AccountName)
 	assert.Equal(t, "ACTIVE", account.Status)
 }

hive/handlers/iam/service_impl.go

@@ -16,6 +16,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/mulgadc/hive/hive/admin"
 	"github.com/mulgadc/hive/hive/awserrors"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats.go"
 )
 
@@ -26,8 +27,6 @@ const (
 	KVBucketAccounts       = "hive-accounts"
 	KVBucketAccountCounter = "hive-account-counter"
 
-	GlobalAccountID = "000000000000"
-
 	maxAccessKeysPerUser = 2
 )
 
@@ -512,7 +511,7 @@ func (s *IAMServiceImpl) DecryptSecret(ciphertext string) (string, error) {
 func (s *IAMServiceImpl) SeedRootUser(data *BootstrapData) error {
 	// Ensure the global account record exists
 	globalAccount := Account{
-		AccountID:   GlobalAccountID,
+		AccountID:   utils.GlobalAccountID,
 		AccountName: "Global",
 		Status:      AccountStatusActive,
 		CreatedAt:   time.Now().UTC().Format(time.RFC3339),
@@ -521,16 +520,16 @@ func (s *IAMServiceImpl) SeedRootUser(data *BootstrapData) error {
 	if err != nil {
 		return fmt.Errorf("marshal global account: %w", err)
 	}
-	if _, err := s.accountsBucket.Create(GlobalAccountID, accountData); err != nil && !errors.Is(err, nats.ErrKeyExists) {
+	if _, err := s.accountsBucket.Create(utils.GlobalAccountID, accountData); err != nil && !errors.Is(err, nats.ErrKeyExists) {
 		return fmt.Errorf("seed global account: %w", err)
 	}
 
-	kvKey := GlobalAccountID + ".root"
+	kvKey := utils.GlobalAccountID + ".root"
 	rootUser := User{
 		UserName:         "root",
 		UserID:           "AIDAAAAAAAAAAAAAAAAA",
-		AccountID:        GlobalAccountID,
-		ARN:              fmt.Sprintf("arn:aws:iam::%s:root", GlobalAccountID),
+		AccountID:        utils.GlobalAccountID,
+		ARN:              fmt.Sprintf("arn:aws:iam::%s:root", utils.GlobalAccountID),
 		Path:             "/",
 		CreatedAt:        time.Now().UTC().Format(time.RFC3339),
 		AccessKeys:       []string{data.AccessKeyID},
@@ -558,7 +557,7 @@ func (s *IAMServiceImpl) SeedRootUser(data *BootstrapData) error {
 		AccessKeyID:     data.AccessKeyID,
 		SecretAccessKey: data.EncryptedSecret,
 		UserName:        "root",
-		AccountID:       GlobalAccountID,
+		AccountID:       utils.GlobalAccountID,
 		Status:          AccessKeyStatusActive,
 		CreatedAt:       rootUser.CreatedAt,
 	}
@@ -577,7 +576,7 @@ func (s *IAMServiceImpl) SeedRootUser(data *BootstrapData) error {
 		return fmt.Errorf("seed root access key: %w", err)
 	}
 
-	slog.Info("Root IAM user seeded", "accountID", GlobalAccountID, "accessKeyID", data.AccessKeyID)
+	slog.Info("Root IAM user seeded", "accountID", utils.GlobalAccountID, "accessKeyID", data.AccessKeyID)
 	return nil
 }
 

hive/handlers/iam/service_impl_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/mulgadc/hive/hive/admin"
 	"github.com/mulgadc/hive/hive/awserrors"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
@@ -18,7 +19,7 @@ import (
 // Interface compliance check
 var _ IAMService = (*IAMServiceImpl)(nil)
 
-const testAccountID = GlobalAccountID
+const testAccountID = utils.GlobalAccountID
 
 func setupTestIAMService(t *testing.T) *IAMServiceImpl {
 	t.Helper()
@@ -570,24 +571,24 @@ func TestSeedRootUser(t *testing.T) {
 	err = svc.SeedRootUser(&BootstrapData{
 		AccessKeyID:     "AKIAEXAMPLE123456789",
 		EncryptedSecret: encryptedSecret,
-		AccountID:       GlobalAccountID,
+		AccountID:       utils.GlobalAccountID,
 	})
 	require.NoError(t, err)
 
 	// Verify root user exists at account-scoped key
-	out, err := svc.GetUser(GlobalAccountID, &iam.GetUserInput{
+	out, err := svc.GetUser(utils.GlobalAccountID, &iam.GetUserInput{
 		UserName: aws.String("root"),
 	})
 	require.NoError(t, err)
 	assert.Equal(t, "root", *out.User.UserName)
-	assert.Contains(t, *out.User.Arn, GlobalAccountID)
+	assert.Contains(t, *out.User.Arn, utils.GlobalAccountID)
 	assert.Contains(t, *out.User.Arn, "root")
 
 	// Verify access key exists with AccountID
 	ak, err := svc.LookupAccessKey("AKIAEXAMPLE123456789")
 	require.NoError(t, err)
 	assert.Equal(t, "root", ak.UserName)
-	assert.Equal(t, GlobalAccountID, ak.AccountID)
+	assert.Equal(t, utils.GlobalAccountID, ak.AccountID)
 	assert.Equal(t, "Active", ak.Status)
 
 	// Verify secret is decryptable
@@ -596,9 +597,9 @@ func TestSeedRootUser(t *testing.T) {
 	assert.Equal(t, "test-secret-key", decrypted)
 
 	// Verify global account record was created
-	account, err := svc.GetAccount(GlobalAccountID)
+	account, err := svc.GetAccount(utils.GlobalAccountID)
 	require.NoError(t, err)
-	assert.Equal(t, GlobalAccountID, account.AccountID)
+	assert.Equal(t, utils.GlobalAccountID, account.AccountID)
 	assert.Equal(t, "Global", account.AccountName)
 	assert.Equal(t, "ACTIVE", account.Status)
 }
@@ -610,7 +611,7 @@ func TestSeedRootUser_Idempotent(t *testing.T) {
 	data := &BootstrapData{
 		AccessKeyID:     "AKIAEXAMPLE123456789",
 		EncryptedSecret: encryptedSecret,
-		AccountID:       GlobalAccountID,
+		AccountID:       utils.GlobalAccountID,
 	}
 
 	// First call seeds
@@ -622,7 +623,7 @@ func TestSeedRootUser_Idempotent(t *testing.T) {
 	require.NoError(t, err)
 
 	// Root user should still exist with original data
-	out, err := svc.GetUser(GlobalAccountID, &iam.GetUserInput{
+	out, err := svc.GetUser(utils.GlobalAccountID, &iam.GetUserInput{
 		UserName: aws.String("root"),
 	})
 	require.NoError(t, err)

hive/services/vpcd/integration_test.go

@@ -9,6 +9,8 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	handlers_ec2_igw "github.com/mulgadc/hive/hive/handlers/ec2/igw"
+	handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc"
+	"github.com/mulgadc/hive/hive/utils"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 )
@@ -153,6 +155,19 @@ func TestIntegration_VPCLifecycle(t *testing.T) {
 	}
 
 	// === Phase 4: Create and Attach Internet Gateway ===
+	// Create VPC KV bucket and register test VPC so IGW ownership checks pass
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatalf("JetStream: %v", err)
+	}
+	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
+	if err != nil {
+		t.Fatalf("create VPC KV: %v", err)
+	}
+	if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, "vpc-integ1"), []byte(`{"vpc_id":"vpc-integ1","state":"available"}`)); err != nil {
+		t.Fatalf("register test VPC: %v", err)
+	}
+
 	igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc)
 	if err != nil {
 		t.Fatalf("create IGW service: %v", err)
@@ -373,6 +388,19 @@ func TestIntegration_MultiSubnetWithIGW(t *testing.T) {
 		t.Errorf("router ports = %d, want 3", len(router.Ports))
 	}
 
+	// Create VPC KV bucket and register test VPC so IGW ownership checks pass
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatalf("JetStream: %v", err)
+	}
+	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
+	if err != nil {
+		t.Fatalf("create VPC KV: %v", err)
+	}
+	if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, "vpc-multi"), []byte(`{"vpc_id":"vpc-multi","state":"available"}`)); err != nil {
+		t.Fatalf("register test VPC: %v", err)
+	}
+
 	// Attach IGW
 	igwSvc, _ := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc)
 	igwOut, _ := igwSvc.CreateInternetGateway(&ec2.CreateInternetGatewayInput{}, testIGWAccountID)
@@ -459,6 +487,21 @@ func TestIntegration_IGWErrorPaths(t *testing.T) {
 		}
 	}()
 
+	// Create VPC KV bucket and register test VPCs so IGW ownership checks pass
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatalf("JetStream: %v", err)
+	}
+	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
+	if err != nil {
+		t.Fatalf("create VPC KV: %v", err)
+	}
+	for _, vpcID := range []string{"vpc-err1", "vpc-err2"} {
+		if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`)); err != nil {
+			t.Fatalf("register test VPC: %v", err)
+		}
+	}
+
 	igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc)
 	if err != nil {
 		t.Fatalf("create IGW service: %v", err)

hive/utils/account.go

@@ -0,0 +1,39 @@
+package utils
+
+import "github.com/nats-io/nats.go"
+
+// GlobalAccountID is the root/system account. Pre-Phase4 resources are owned by this account.
+const GlobalAccountID = "000000000000"
+
+// AccountKey returns a KV key scoped to an account: "{accountID}.{resourceID}".
+func AccountKey(accountID, resourceID string) string {
+	return accountID + "." + resourceID
+}
+
+// IsAccountID checks if a string is a valid 12-digit AWS account ID.
+func IsAccountID(s string) bool {
+	if len(s) != 12 {
+		return false
+	}
+	for _, c := range s {
+		if c < '0' || c > '9' {
+			return false
+		}
+	}
+	return true
+}
+
+// GetOrCreateKVBucket creates or retrieves a NATS KV bucket.
+func GetOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) {
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
+		Bucket:  bucketName,
+		History: SafeIntToUint8(history),
+	})
+	if err != nil {
+		kv, err = js.KeyValue(bucketName)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return kv, nil
+}

hive/utils/account_test.go

@@ -0,0 +1,54 @@
+package utils
+
+import "testing"
+
+func TestAccountKey(t *testing.T) {
+	tests := []struct {
+		accountID  string
+		resourceID string
+		want       string
+	}{
+		{"000000000000", "vpc-123", "000000000000.vpc-123"},
+		{"123456789012", "igw-abc", "123456789012.igw-abc"},
+		{"", "vol-1", ".vol-1"},
+	}
+	for _, tt := range tests {
+		got := AccountKey(tt.accountID, tt.resourceID)
+		if got != tt.want {
+			t.Errorf("AccountKey(%q, %q) = %q, want %q", tt.accountID, tt.resourceID, got, tt.want)
+		}
+	}
+}
+
+func TestIsAccountID(t *testing.T) {
+	tests := []struct {
+		input string
+		want  bool
+	}{
+		{"000000000000", true},
+		{"123456789012", true},
+		{"999999999999", true},
+		{"self", false},
+		{"hive", false},
+		{"", false},
+		{"12345678901", false},   // 11 digits
+		{"1234567890123", false}, // 13 digits
+		{"12345678901a", false},  // non-digit
+		{"abcdefghijkl", false},
+	}
+	for _, tt := range tests {
+		got := IsAccountID(tt.input)
+		if got != tt.want {
+			t.Errorf("IsAccountID(%q) = %v, want %v", tt.input, got, tt.want)
+		}
+	}
+}
+
+func TestGlobalAccountID(t *testing.T) {
+	if GlobalAccountID != "000000000000" {
+		t.Errorf("GlobalAccountID = %q, want %q", GlobalAccountID, "000000000000")
+	}
+	if !IsAccountID(GlobalAccountID) {
+		t.Error("GlobalAccountID should be a valid account ID")
+	}
+}

hive/utils/nats.go

@@ -77,6 +77,80 @@ func NATSRequest[Out any](conn *nats.Conn, subject string, input any, timeout ti
 	return &output, nil
 }
 
+// NATSScatterGather publishes a fan-out request and collects responses from
+// multiple nodes. It returns the first successful (non-error) response. Error
+// payloads from individual nodes are skipped. If all responses are errors, the
+// last error is returned. If no responses arrive before the deadline, a timeout
+// error is returned. When expectedNodes > 0, collection exits early once that
+// many responses have been received.
+func NATSScatterGather[Out any](conn *nats.Conn, subject string, input any, timeout time.Duration, expectedNodes int, accountID string) (*Out, error) {
+	jsonData, err := json.Marshal(input)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal input: %w", err)
+	}
+
+	inbox := nats.NewInbox()
+	sub, err := conn.SubscribeSync(inbox)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create inbox: %w", err)
+	}
+	defer sub.Unsubscribe()
+
+	pubMsg := nats.NewMsg(subject)
+	pubMsg.Reply = inbox
+	pubMsg.Data = jsonData
+	pubMsg.Header.Set(AccountIDHeader, accountID)
+	if err := conn.PublishMsg(pubMsg); err != nil {
+		return nil, fmt.Errorf("failed to publish request: %w", err)
+	}
+
+	deadline := time.Now().Add(timeout)
+	responsesReceived := 0
+	var lastErr error
+
+	for time.Now().Before(deadline) {
+		if expectedNodes > 0 && responsesReceived >= expectedNodes {
+			break
+		}
+
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			break
+		}
+
+		msg, err := sub.NextMsg(remaining)
+		if err != nil {
+			if err == nats.ErrTimeout || err == nats.ErrNoResponders {
+				break
+			}
+			return nil, fmt.Errorf("scatter-gather receive error: %w", err)
+		}
+
+		responsesReceived++
+
+		responseError, err := ValidateErrorPayload(msg.Data)
+		if err != nil {
+			slog.Debug("ScatterGather: skipping error response", "code", *responseError.Code, "subject", subject)
+			lastErr = errors.New(*responseError.Code)
+			continue
+		}
+
+		var output Out
+		if err := json.Unmarshal(msg.Data, &output); err != nil {
+			slog.Debug("ScatterGather: skipping malformed response", "subject", subject, "err", err)
+			lastErr = fmt.Errorf("failed to unmarshal response: %w", err)
+			continue
+		}
+
+		return &output, nil
+	}
+
+	if lastErr != nil {
+		return nil, lastErr
+	}
+	return nil, fmt.Errorf("scatter-gather timeout: no responses received for %s", subject)
+}
+
 // AccountIDFromMsg extracts the caller's account ID from a NATS message header.
 // Returns the account ID, or empty string if the header is not set.
 func AccountIDFromMsg(msg *nats.Msg) string {

hive/utils/nats_test.go

@@ -239,6 +239,145 @@ func TestNATSRequest_MarshalError(t *testing.T) {
 
 // --- AccountIDFromMsg tests ---
 
+// --- NATSScatterGather tests ---
+
+func TestNATSScatterGather_SuccessAmongErrors(t *testing.T) {
+	ns := startTestNATSServer(t)
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	type Resp struct {
+		Value string `json:"value"`
+	}
+
+	// Simulate 3 nodes: 2 return errors, 1 returns success
+	_, err = nc.Subscribe("test.scatter", func(msg *nats.Msg) {
+		// Simulate varying response times
+		resp := Resp{Value: "found"}
+		data, _ := json.Marshal(resp)
+		msg.Respond(data)
+	})
+	require.NoError(t, err)
+
+	// Two error responders (faster)
+	for range 2 {
+		_, err = nc.Subscribe("test.scatter", func(msg *nats.Msg) {
+			errPayload := GenerateErrorPayload("InvalidInstanceID.NotFound")
+			msg.Respond(errPayload)
+		})
+		require.NoError(t, err)
+	}
+
+	result, err := NATSScatterGather[Resp](nc, "test.scatter", struct{}{}, 3*time.Second, 3, "")
+	require.NoError(t, err)
+	assert.Equal(t, "found", result.Value)
+}
+
+func TestNATSScatterGather_AllErrors(t *testing.T) {
+	ns := startTestNATSServer(t)
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	type Resp struct{}
+
+	// All 3 nodes return errors
+	for range 3 {
+		_, err = nc.Subscribe("test.scatter.allerr", func(msg *nats.Msg) {
+			errPayload := GenerateErrorPayload("InvalidInstanceID.NotFound")
+			msg.Respond(errPayload)
+		})
+		require.NoError(t, err)
+	}
+
+	_, err = NATSScatterGather[Resp](nc, "test.scatter.allerr", struct{}{}, 2*time.Second, 3, "")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "InvalidInstanceID.NotFound")
+}
+
+func TestNATSScatterGather_Timeout(t *testing.T) {
+	ns := startTestNATSServer(t)
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	type Resp struct{}
+
+	// No subscribers — should return a timeout error
+	_, err = NATSScatterGather[Resp](nc, "test.scatter.timeout", struct{}{}, 200*time.Millisecond, 0, "")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "no responses received")
+}
+
+func TestNATSScatterGather_SingleSuccess(t *testing.T) {
+	ns := startTestNATSServer(t)
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	type Resp struct {
+		ID string `json:"id"`
+	}
+
+	_, err = nc.Subscribe("test.scatter.single", func(msg *nats.Msg) {
+		resp := Resp{ID: "ami-123"}
+		data, _ := json.Marshal(resp)
+		msg.Respond(data)
+	})
+	require.NoError(t, err)
+
+	result, err := NATSScatterGather[Resp](nc, "test.scatter.single", struct{}{}, 2*time.Second, 1, "acct-123")
+	require.NoError(t, err)
+	assert.Equal(t, "ami-123", result.ID)
+}
+
+func TestNATSScatterGather_MarshalError(t *testing.T) {
+	ns := startTestNATSServer(t)
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	type Resp struct{}
+	_, err = NATSScatterGather[Resp](nc, "test.scatter.marshal", make(chan int), 2*time.Second, 0, "")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "failed to marshal input")
+}
+
+func TestNATSScatterGather_EarlyExitOnSuccess(t *testing.T) {
+	ns := startTestNATSServer(t)
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	type Resp struct {
+		Value string `json:"value"`
+	}
+
+	// One fast success responder — should return immediately without waiting
+	_, err = nc.Subscribe("test.scatter.early", func(msg *nats.Msg) {
+		resp := Resp{Value: "quick"}
+		data, _ := json.Marshal(resp)
+		msg.Respond(data)
+	})
+	require.NoError(t, err)
+
+	start := time.Now()
+	result, err := NATSScatterGather[Resp](nc, "test.scatter.early", struct{}{}, 5*time.Second, 0, "")
+	elapsed := time.Since(start)
+
+	require.NoError(t, err)
+	assert.Equal(t, "quick", result.Value)
+	// Should return well before the 5s timeout
+	assert.Less(t, elapsed, 2*time.Second)
+}
+
 func TestAccountIDFromMsg(t *testing.T) {
 	msg := nats.NewMsg("test")
 	msg.Header.Set(AccountIDHeader, "444455556666")