refactor(daemon): extract Describe instance handlers to service (1b-pre phase 1)

Collapse the Describe family in daemon_handlers_instance.go to thin Pattern A
wrappers over a transport-neutral InstanceServiceImpl. This brings the read
path in line with the rest of the codebase and is the prerequisite for 1b's
local HTTP API (mulga-siv-14, daemon-local-autonomy plan).

Extracted: DescribeInstances, DescribeInstanceTypes, DescribeInstanceAttribute,
DescribeStoppedInstances, DescribeTerminatedInstances. Filter helpers
(instanceMatchesFilters, matchTagKey/Value, DescribeInstancesValidFilters,
IsInstanceVisible) move to the instance package alongside the service methods.

InstanceService gains the new methods and two dep interfaces
(ResourceCapacityProvider, StoppedInstanceStore) so the daemon's ResourceManager
and JetStreamManager can plug in without leaking concrete types. NATSInstanceService
returns a clear error for the fan-out describes (DescribeInstances,
DescribeInstanceTypes) since cluster-wide aggregation lives in the gateway's
scatter-gather helpers, not a single-node NATS request.

Behaviour changes: malformed JSON on DescribeInstanceAttribute now responds
with ValidationError (matching every other Pattern A handler) instead of
ServerInternal. Test updated.

Mutators (RunInstances, ModifyInstanceAttribute, StartStoppedInstance,
TerminateStoppedInstance, sub-handlers) remain Pattern B for now; follow-up
phases of mulga-siv-14 will collapse them.

spinifex/daemon/daemon.go

@@ -578,7 +578,7 @@ func (d *Daemon) Start() error {
 
 	// Create services before loading/launching instances, since LaunchInstance depends on them
 	store := objectstore.NewS3ObjectStoreFromConfig(admin.DialTarget(d.config.Predastore.Host), d.config.Predastore.Region, d.config.Predastore.AccessKey, d.config.Predastore.SecretKey)
-	d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store)
+	d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store, d.resourceMgr, d.jsManager)
 	d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config)
 	d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config, d.natsConn)
 

spinifex/daemon/daemon_handlers_instance.go

@@ -11,7 +11,6 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
-	"github.com/mulgadc/spinifex/spinifex/filterutil"
 	handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
 	"github.com/mulgadc/spinifex/spinifex/qmp"
 	"github.com/mulgadc/spinifex/spinifex/types"
@@ -633,265 +632,14 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2I
 	}(instance, command.Attributes)
 }
 
-// describeInstancesValidFilters defines the set of filter names accepted by DescribeInstances.
-var describeInstancesValidFilters = map[string]bool{
-	"instance-state-name": true,
-	"instance-id":         true,
-	"instance-type":       true,
-	"vpc-id":              true,
-	"subnet-id":           true,
-	"tag-key":             true,
-	"tag-value":           true,
-}
-
-// instanceMatchesFilters checks whether a VM + its built ec2.Instance copy satisfy all parsed filters.
-func instanceMatchesFilters(inst *vm.VM, ic *ec2.Instance, filters map[string][]string) bool {
-	for name, values := range filters {
-		if strings.HasPrefix(name, "tag:") {
-			// tag:Key filters are handled after all field filters.
-			continue
-		}
-
-		var field string
-		switch name {
-		case "instance-state-name":
-			if ic.State != nil && ic.State.Name != nil {
-				field = *ic.State.Name
-			}
-		case "instance-id":
-			field = inst.ID
-		case "instance-type":
-			field = inst.InstanceType
-		case "vpc-id":
-			if ic.VpcId != nil {
-				field = *ic.VpcId
-			}
-		case "subnet-id":
-			if ic.SubnetId != nil {
-				field = *ic.SubnetId
-			}
-		case "tag-key":
-			if !matchTagKey(ic.Tags, values) {
-				return false
-			}
-			continue
-		case "tag-value":
-			if !matchTagValue(ic.Tags, values) {
-				return false
-			}
-			continue
-		default:
-			// Filter name passed ParseFilters but has no case — treat as non-match
-			// to avoid silently ignoring it.
-			return false
-		}
-
-		if !filterutil.MatchesAny(values, field) {
-			return false
-		}
-	}
-
-	// Check tag:Key filters via the instance's Tag slice.
-	tags := filterutil.EC2TagsToMap(ic.Tags)
-	return filterutil.MatchesTags(filters, tags)
-}
-
-// matchTagKey returns true if any tag key on the resource matches any of the filter values.
-func matchTagKey(tags []*ec2.Tag, values []string) bool {
-	for _, t := range tags {
-		if t.Key != nil && filterutil.MatchesAny(values, *t.Key) {
-			return true
-		}
-	}
-	return false
-}
-
-// matchTagValue returns true if any tag value on the resource matches any of the filter values.
-func matchTagValue(tags []*ec2.Tag, values []string) bool {
-	for _, t := range tags {
-		if t.Value != nil && filterutil.MatchesAny(values, *t.Value) {
-			return true
-		}
-	}
-	return false
-}
-
-// handleEC2DescribeInstances processes incoming EC2 DescribeInstances requests
-// This handler responds with instances running on this node owned by the caller's account
+// handleEC2DescribeInstances responds with instances running on this node visible to the caller.
 func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) {
-	slog.Debug("Received message", "subject", msg.Subject, "data", string(msg.Data))
-
-	// Extract account ID from NATS header for scoping
-	accountID := utils.AccountIDFromMsg(msg)
-
-	// Initialize describeInstancesInput before unmarshaling into it
-	describeInstancesInput := &ec2.DescribeInstancesInput{}
-	errResp := utils.UnmarshalJsonPayload(describeInstancesInput, msg.Data)
-
-	if errResp != nil {
-		if err := msg.Respond(errResp); err != nil {
-			slog.Error("Failed to respond to NATS request", "err", err)
-		}
-		slog.Error("Request does not match DescribeInstancesInput")
-		return
-	}
-
-	slog.Info("Processing DescribeInstances request from this node", "accountID", accountID)
-
-	d.Instances.Mu.Lock()
-	defer d.Instances.Mu.Unlock()
-
-	// Validate and filter instances if specific instance IDs were requested
-	instanceIDFilter := make(map[string]bool)
-	if len(describeInstancesInput.InstanceIds) > 0 {
-		for _, id := range describeInstancesInput.InstanceIds {
-			if id != nil && *id != "" {
-				if !strings.HasPrefix(*id, "i-") {
-					respondWithError(msg, awserrors.ErrorInvalidInstanceIDMalformed)
-					return
-				}
-				instanceIDFilter[*id] = true
-			}
-		}
-	}
-
-	// Parse filters (returns error for unknown filter names)
-	parsedFilters, err := filterutil.ParseFilters(describeInstancesInput.Filters, describeInstancesValidFilters)
-	if err != nil {
-		slog.Warn("DescribeInstances: invalid filter", "err", err)
-		respondWithError(msg, awserrors.ErrorInvalidParameterValue)
-		return
-	}
-
-	// Group instances by reservation ID (AWS returns instances grouped by reservation)
-	reservationMap := make(map[string]*ec2.Reservation)
-
-	// 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 only visible to root.
-		if !isInstanceVisible(accountID, instance.AccountID) {
-			continue
-		}
-
-		// Skip if filtering by instance IDs and this instance is not in the filter
-		if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] {
-			continue
-		}
-
-		// Use stored reservation metadata if available
-		if instance.Reservation != nil && instance.Instance != nil {
-			resID := ""
-			if instance.Reservation.ReservationId != nil {
-				resID = *instance.Reservation.ReservationId
-			}
-
-			// Create reservation entry if it doesn't exist
-			if _, exists := reservationMap[resID]; !exists {
-				reservation := &ec2.Reservation{}
-				reservation.SetReservationId(resID)
-				if instance.Reservation.OwnerId != nil {
-					reservation.SetOwnerId(*instance.Reservation.OwnerId)
-				}
-				reservation.Instances = []*ec2.Instance{}
-				reservationMap[resID] = reservation
-			}
-
-			// Update the instance state to current state
-			instanceCopy := *instance.Instance
-			instanceCopy.State = &ec2.InstanceState{}
-
-			// Populate PublicIpAddress from VM if stored
-			if instance.PublicIP != "" && instanceCopy.PublicIpAddress == nil {
-				instanceCopy.PublicIpAddress = aws.String(instance.PublicIP)
-			}
-
-			// Map internal status to EC2 state codes using the centralized mapping
-			if info, ok := vm.EC2StateCodes[instance.Status]; ok {
-				instanceCopy.State.SetCode(info.Code)
-				instanceCopy.State.SetName(info.Name)
-			} else {
-				slog.Warn("Instance has unmapped status, reporting as pending",
-					"instanceId", instance.ID, "status", string(instance.Status))
-				instanceCopy.State.SetCode(0)
-				instanceCopy.State.SetName("pending")
-			}
-
-			// Populate Placement if instance belongs to a placement group
-			if instance.PlacementGroupName != "" {
-				instanceCopy.Placement = &ec2.Placement{
-					GroupName:        aws.String(instance.PlacementGroupName),
-					AvailabilityZone: aws.String(d.config.AZ),
-				}
-			}
-
-			// Apply filters against the fully-built instance copy
-			if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) {
-				continue
-			}
-
-			// Add instance to its reservation
-			reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy)
-		}
-	}
-
-	// Convert map to slice
-	reservations := make([]*ec2.Reservation, 0, len(reservationMap))
-	for _, reservation := range reservationMap {
-		reservations = append(reservations, reservation)
-	}
-
-	// Create the response
-	output := &ec2.DescribeInstancesOutput{
-		Reservations: reservations,
-	}
-
-	respondWithJSON(msg, output)
-	slog.Info("handleEC2DescribeInstances completed", "count", len(reservations))
+	handleNATSRequest(msg, d.instanceService.DescribeInstances)
 }
 
-// handleEC2DescribeInstanceTypes processes incoming EC2 DescribeInstanceTypes requests
-// This handler responds with instance types that can currently be provisioned on this node
-// based on available resources (CPU and memory not already allocated to running instances)
+// handleEC2DescribeInstanceTypes responds with instance types provisionable on this node.
 func (d *Daemon) handleEC2DescribeInstanceTypes(msg *nats.Msg) {
-	slog.Debug("Received message", "subject", msg.Subject)
-
-	// Initialize input
-	describeInput := &ec2.DescribeInstanceTypesInput{}
-	errResp := utils.UnmarshalJsonPayload(describeInput, msg.Data)
-	if errResp != nil {
-		if err := msg.Respond(errResp); err != nil {
-			slog.Error("Failed to respond to NATS request", "err", err)
-		}
-		slog.Error("Request does not match DescribeInstanceTypesInput")
-		return
-	}
-
-	slog.Info("Processing DescribeInstanceTypes request from this node")
-
-	// Check if "capacity" filter is set to "true"
-	showCapacity := false
-	for _, f := range describeInput.Filters {
-		if f.Name != nil && *f.Name == "capacity" {
-			for _, v := range f.Values {
-				if v != nil && *v == "true" {
-					showCapacity = true
-					break
-				}
-			}
-		}
-	}
-
-	// Get instance types based on capacity and the showCapacity flag
-	filteredTypes := d.resourceMgr.GetAvailableInstanceTypeInfos(showCapacity)
-
-	// Create the response
-	output := &ec2.DescribeInstanceTypesOutput{
-		InstanceTypes: filteredTypes,
-	}
-
-	respondWithJSON(msg, output)
-	slog.Info("handleEC2DescribeInstanceTypes completed", "count", len(filteredTypes))
+	handleNATSRequest(msg, d.instanceService.DescribeInstanceTypes)
 }
 
 // startStoppedInstanceRequest is the payload for ec2.start topic
@@ -1153,109 +901,12 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
 
 // handleEC2DescribeStoppedInstances returns stopped instances from shared KV.
 func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) {
-	d.describeInstancesFromKV(msg, d.jsManager.ListStoppedInstances, 80, "stopped", "handleEC2DescribeStoppedInstances")
+	handleNATSRequest(msg, d.instanceService.DescribeStoppedInstances)
 }
 
 // handleEC2DescribeTerminatedInstances returns terminated instances from the terminated KV bucket.
 func (d *Daemon) handleEC2DescribeTerminatedInstances(msg *nats.Msg) {
-	d.describeInstancesFromKV(msg, d.jsManager.ListTerminatedInstances, 48, "terminated", "handleEC2DescribeTerminatedInstances")
-}
-
-// describeInstancesFromKV is a shared helper for DescribeStopped/TerminatedInstances handlers.
-// It lists instances from a KV bucket, filters by account/instance ID, and responds with reservations.
-func (d *Daemon) describeInstancesFromKV(msg *nats.Msg, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, handlerName string) {
-	if d.jsManager == nil {
-		respondWithError(msg, awserrors.ErrorServerInternal)
-		return
-	}
-
-	accountID := utils.AccountIDFromMsg(msg)
-
-	describeInput := &ec2.DescribeInstancesInput{}
-	if len(msg.Data) > 0 {
-		if errResp := utils.UnmarshalJsonPayload(describeInput, msg.Data); errResp != nil {
-			if err := msg.Respond(errResp); err != nil {
-				slog.Error("Failed to respond to NATS request", "err", err)
-			}
-			return
-		}
-	}
-
-	instanceIDFilter := make(map[string]bool)
-	for _, id := range describeInput.InstanceIds {
-		if id != nil {
-			instanceIDFilter[*id] = true
-		}
-	}
-
-	parsedFilters, filterErr := filterutil.ParseFilters(describeInput.Filters, describeInstancesValidFilters)
-	if filterErr != nil {
-		slog.Warn(handlerName+": invalid filter", "err", filterErr)
-		respondWithError(msg, awserrors.ErrorInvalidParameterValue)
-		return
-	}
-
-	instances, err := listFn()
-	if err != nil {
-		slog.Error(handlerName+": failed to list instances", "err", err)
-		respondWithError(msg, awserrors.ErrorServerInternal)
-		return
-	}
-
-	reservationMap := make(map[string]*ec2.Reservation)
-
-	for _, instance := range instances {
-		if !isInstanceVisible(accountID, instance.AccountID) {
-			continue
-		}
-		if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] {
-			continue
-		}
-		if instance.Reservation == nil || instance.Instance == nil {
-			slog.Warn(handlerName+": skipping instance with nil Reservation/Instance (data integrity issue)",
-				"instanceId", instance.ID)
-			continue
-		}
-
-		resID := ""
-		if instance.Reservation.ReservationId != nil {
-			resID = *instance.Reservation.ReservationId
-		}
-
-		if _, exists := reservationMap[resID]; !exists {
-			reservation := &ec2.Reservation{}
-			reservation.SetReservationId(resID)
-			if instance.Reservation.OwnerId != nil {
-				reservation.SetOwnerId(*instance.Reservation.OwnerId)
-			}
-			reservation.Instances = []*ec2.Instance{}
-			reservationMap[resID] = reservation
-		}
-
-		instanceCopy := *instance.Instance
-		instanceCopy.State = &ec2.InstanceState{}
-		if info, ok := vm.EC2StateCodes[instance.Status]; ok {
-			instanceCopy.State.SetCode(info.Code)
-			instanceCopy.State.SetName(info.Name)
-		} else {
-			instanceCopy.State.SetCode(fallbackCode)
-			instanceCopy.State.SetName(fallbackName)
-		}
-
-		if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) {
-			continue
-		}
-
-		reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy)
-	}
-
-	reservations := make([]*ec2.Reservation, 0, len(reservationMap))
-	for _, reservation := range reservationMap {
-		reservations = append(reservations, reservation)
-	}
-
-	respondWithJSON(msg, &ec2.DescribeInstancesOutput{Reservations: reservations})
-	slog.Info(handlerName+" completed", "count", len(reservations))
+	handleNATSRequest(msg, d.instanceService.DescribeTerminatedInstances)
 }
 
 // handleEC2ModifyInstanceAttribute modifies attributes of a stopped instance in shared KV.
@@ -1365,120 +1016,8 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) {
 }
 
 // handleEC2DescribeInstanceAttribute returns a single requested attribute for an instance.
-// It checks running instances first (in-memory), then falls back to stopped instances in KV.
 func (d *Daemon) handleEC2DescribeInstanceAttribute(msg *nats.Msg) {
-	var input ec2.DescribeInstanceAttributeInput
-	if err := json.Unmarshal(msg.Data, &input); err != nil {
-		slog.Error("handleEC2DescribeInstanceAttribute: failed to unmarshal request", "err", err)
-		respondWithError(msg, awserrors.ErrorServerInternal)
-		return
-	}
-
-	if input.InstanceId == nil || *input.InstanceId == "" {
-		slog.Error("handleEC2DescribeInstanceAttribute: missing instance_id")
-		respondWithError(msg, awserrors.ErrorMissingParameter)
-		return
-	}
-	if input.Attribute == nil || *input.Attribute == "" {
-		slog.Error("handleEC2DescribeInstanceAttribute: missing attribute")
-		respondWithError(msg, awserrors.ErrorMissingParameter)
-		return
-	}
-
-	instanceID := *input.InstanceId
-	attribute := *input.Attribute
-	accountID := utils.AccountIDFromMsg(msg)
-
-	// Look up instance: running first, then stopped KV.
-	var instance *vm.VM
-
-	d.Instances.Mu.Lock()
-	if running, ok := d.Instances.VMS[instanceID]; ok {
-		instance = running
-	}
-	d.Instances.Mu.Unlock()
-
-	if instance == nil {
-		if d.jsManager == nil {
-			slog.Error("handleEC2DescribeInstanceAttribute: JetStream not available")
-			respondWithError(msg, awserrors.ErrorServerInternal)
-			return
-		}
-		stopped, err := d.jsManager.LoadStoppedInstance(instanceID)
-		if err != nil {
-			slog.Error("handleEC2DescribeInstanceAttribute: failed to load stopped instance",
-				"instanceId", instanceID, "err", err)
-			respondWithError(msg, awserrors.ErrorServerInternal)
-			return
-		}
-		instance = stopped
-	}
-
-	if instance == nil {
-		slog.Warn("handleEC2DescribeInstanceAttribute: instance not found",
-			"instanceId", instanceID)
-		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
-		return
-	}
-
-	if !checkInstanceOwnership(msg, instanceID, instance.AccountID) {
-		return
-	}
-
-	output := &ec2.DescribeInstanceAttributeOutput{
-		InstanceId: &instanceID,
-	}
-
-	switch attribute {
-	case ec2.InstanceAttributeNameInstanceType:
-		val := instance.InstanceType
-		output.InstanceType = &ec2.AttributeValue{Value: &val}
-
-	case ec2.InstanceAttributeNameUserData:
-		val := instance.UserData
-		output.UserData = &ec2.AttributeValue{Value: &val}
-
-	case ec2.InstanceAttributeNameDisableApiTermination:
-		val := false
-		output.DisableApiTermination = &ec2.AttributeBooleanValue{Value: &val}
-
-	case ec2.InstanceAttributeNameDisableApiStop:
-		val := false
-		output.DisableApiStop = &ec2.AttributeBooleanValue{Value: &val}
-
-	case ec2.InstanceAttributeNameInstanceInitiatedShutdownBehavior:
-		val := ec2.ShutdownBehaviorStop
-		output.InstanceInitiatedShutdownBehavior = &ec2.AttributeValue{Value: &val}
-
-	case ec2.InstanceAttributeNameEbsOptimized:
-		val := false
-		output.EbsOptimized = &ec2.AttributeBooleanValue{Value: &val}
-
-	case ec2.InstanceAttributeNameEnaSupport:
-		val := true
-		output.EnaSupport = &ec2.AttributeBooleanValue{Value: &val}
-
-	case ec2.InstanceAttributeNameSourceDestCheck:
-		val := true
-		output.SourceDestCheck = &ec2.AttributeBooleanValue{Value: &val}
-
-	case ec2.InstanceAttributeNameGroupSet:
-		if instance.Instance != nil && len(instance.Instance.SecurityGroups) > 0 {
-			output.Groups = instance.Instance.SecurityGroups
-		} else {
-			output.Groups = []*ec2.GroupIdentifier{}
-		}
-
-	default:
-		slog.Warn("handleEC2DescribeInstanceAttribute: unsupported attribute",
-			"instanceId", instanceID, "attribute", attribute)
-		respondWithError(msg, awserrors.ErrorInvalidParameterValue)
-		return
-	}
-
-	slog.Info("handleEC2DescribeInstanceAttribute: completed",
-		"instanceId", instanceID, "attribute", attribute, "accountID", accountID)
-	respondWithJSON(msg, output)
+	handleNATSRequest(msg, d.instanceService.DescribeInstanceAttribute)
 }
 
 // publishNATEvent sends a NAT lifecycle event (vpc.add-nat or vpc.delete-nat) to NATS.

spinifex/daemon/daemon_handlers_test.go

@@ -87,6 +87,14 @@ func createFullTestDaemonWithJetStream(t *testing.T, natsURL string) *Daemon {
 	err = daemon.jsManager.InitTerminatedInstanceBucket()
 	require.NoError(t, err)
 
+	// Re-bind the instance service so describe-stopped/terminated handlers see
+	// the KV that was just initialised.
+	daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(
+		daemon.config, daemon.resourceMgr.instanceTypes, daemon.natsConn,
+		&daemon.Instances, objectstore.NewMemoryObjectStore(),
+		daemon.resourceMgr, daemon.jsManager,
+	)
+
 	return daemon
 }
 
@@ -512,6 +520,7 @@ func TestHandleEC2RunInstances_ServiceErrorPropagated(t *testing.T) {
 	daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(
 		daemon.config, emptyTypes, daemon.natsConn, &daemon.Instances,
 		objectstore.NewMemoryObjectStore(),
+		daemon.resourceMgr, nil,
 	)
 
 	sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "spinifex-workers", daemon.handleEC2RunInstances)
@@ -2306,7 +2315,9 @@ func TestHandleEC2DescribeInstanceAttribute_InvalidJSON(t *testing.T) {
 	var errResp map[string]any
 	err = json.Unmarshal(reply.Data, &errResp)
 	require.NoError(t, err)
-	assert.Equal(t, "ServerInternal", errResp["Code"])
+	// handleNATSRequest reports unmarshal failures as ValidationError, matching
+	// the rest of the Pattern A handlers.
+	assert.Equal(t, "ValidationError", errResp["Code"])
 }
 
 // --- Delegate handler round-trip tests (table-driven) ---

spinifex/daemon/daemon_test.go

@@ -82,7 +82,10 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon {
 	daemon.detachDelay = 0 // Skip sleep in tests
 
 	// Initialize services (needed for handler tests)
-	daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances, objectstore.NewMemoryObjectStore())
+	// jsManager is nil here; pass a nil literal to keep the StoppedInstanceStore
+	// interface itself nil (rather than a typed-nil pointer) so the service can
+	// short-circuit cleanly when no KV is available.
+	daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances, objectstore.NewMemoryObjectStore(), daemon.resourceMgr, nil)
 	daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImplWithStore(cfg, objectstore.NewMemoryObjectStore(), nc)
 
 	t.Cleanup(func() {

spinifex/handlers/ec2/instance/instance_filter_test.go

@@ -1,4 +1,4 @@
-package daemon
+package handlers_ec2_instance
 
 import (
 	"testing"
@@ -68,14 +68,12 @@ func TestInstanceMatchesFilters_MultipleFilters_ANDLogic(t *testing.T) {
 		VpcId:        aws.String("vpc-111"),
 		State:        &ec2.InstanceState{Name: aws.String("running")},
 	}
-	// Both must match
 	filters := map[string][]string{
 		"instance-state-name": {"running"},
 		"vpc-id":              {"vpc-111"},
 	}
 	assert.True(t, instanceMatchesFilters(inst, ic, filters))
 
-	// One doesn't match
 	filters["vpc-id"] = []string{"vpc-999"}
 	assert.False(t, instanceMatchesFilters(inst, ic, filters))
 }
@@ -189,7 +187,6 @@ func TestInstanceMatchesFilters_EmptyResults(t *testing.T) {
 		InstanceType: aws.String("t3.micro"),
 		State:        &ec2.InstanceState{Name: aws.String("running")},
 	}
-	// Filter that matches nothing
 	filters := map[string][]string{"instance-type": {"c5.4xlarge"}}
 	assert.False(t, instanceMatchesFilters(inst, ic, filters))
 }
@@ -198,7 +195,7 @@ func TestParseFilters_UnknownFilterName(t *testing.T) {
 	filters := []*ec2.Filter{
 		{Name: aws.String("bogus-filter"), Values: []*string{aws.String("val")}},
 	}
-	_, err := filterutil.ParseFilters(filters, describeInstancesValidFilters)
+	_, err := filterutil.ParseFilters(filters, DescribeInstancesValidFilters)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "InvalidParameterValue")
 }

spinifex/handlers/ec2/instance/service.go

@@ -1,8 +1,30 @@
 package handlers_ec2_instance
 
-import "github.com/aws/aws-sdk-go/service/ec2"
+import (
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/vm"
+)
 
 // InstanceService defines the interface for EC2 instance operations business logic
 type InstanceService interface {
 	RunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error)
+	DescribeInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error)
+	DescribeInstanceTypes(input *ec2.DescribeInstanceTypesInput, accountID string) (*ec2.DescribeInstanceTypesOutput, error)
+	DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error)
+	DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error)
+	DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error)
+}
+
+// ResourceCapacityProvider exposes per-node instance-type availability used by
+// DescribeInstanceTypes. Implemented by daemon.ResourceManager.
+type ResourceCapacityProvider interface {
+	GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo
+}
+
+// StoppedInstanceStore provides KV-backed read access to stopped/terminated
+// instances. Implemented by daemon.JetStreamManager.
+type StoppedInstanceStore interface {
+	LoadStoppedInstance(instanceID string) (*vm.VM, error)
+	ListStoppedInstances() ([]*vm.VM, error)
+	ListTerminatedInstances() ([]*vm.VM, error)
 }

spinifex/handlers/ec2/instance/service_impl.go

@@ -19,6 +19,7 @@ import (
 	"github.com/kdomanski/iso9660"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/config"
+	"github.com/mulgadc/spinifex/spinifex/filterutil"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	spxtypes "github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/utils"
@@ -233,16 +234,28 @@ type InstanceServiceImpl struct {
 	natsConn      *nats.Conn
 	instances     *vm.Instances
 	objectStore   objectstore.ObjectStore
+	resourceMgr   ResourceCapacityProvider
+	stoppedStore  StoppedInstanceStore
 }
 
 // NewInstanceServiceImpl creates a new instance service implementation for daemon use
-func NewInstanceServiceImpl(cfg *config.Config, instanceTypes map[string]*ec2.InstanceTypeInfo, nc *nats.Conn, instances *vm.Instances, store objectstore.ObjectStore) *InstanceServiceImpl {
+func NewInstanceServiceImpl(
+	cfg *config.Config,
+	instanceTypes map[string]*ec2.InstanceTypeInfo,
+	nc *nats.Conn,
+	instances *vm.Instances,
+	store objectstore.ObjectStore,
+	resourceMgr ResourceCapacityProvider,
+	stoppedStore StoppedInstanceStore,
+) *InstanceServiceImpl {
 	return &InstanceServiceImpl{
 		config:        cfg,
 		instanceTypes: instanceTypes,
 		natsConn:      nc,
 		instances:     instances,
 		objectStore:   store,
+		resourceMgr:   resourceMgr,
+		stoppedStore:  stoppedStore,
 	}
 }
 
@@ -844,3 +857,401 @@ func generateHostname(instanceID string) string {
 	}
 	return "spinifex-vm-unknown"
 }
+
+// DescribeInstancesValidFilters lists the filter names accepted by DescribeInstances
+// (and the stopped/terminated variants, which share the same filter shape).
+var DescribeInstancesValidFilters = map[string]bool{
+	"instance-state-name": true,
+	"instance-id":         true,
+	"instance-type":       true,
+	"vpc-id":              true,
+	"subnet-id":           true,
+	"tag-key":             true,
+	"tag-value":           true,
+}
+
+// IsInstanceVisible reports whether the caller can see this instance.
+// 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
+}
+
+// instanceMatchesFilters checks whether a VM + its built ec2.Instance copy satisfy all parsed filters.
+func instanceMatchesFilters(inst *vm.VM, ic *ec2.Instance, filters map[string][]string) bool {
+	for name, values := range filters {
+		if strings.HasPrefix(name, "tag:") {
+			continue
+		}
+
+		var field string
+		switch name {
+		case "instance-state-name":
+			if ic.State != nil && ic.State.Name != nil {
+				field = *ic.State.Name
+			}
+		case "instance-id":
+			field = inst.ID
+		case "instance-type":
+			field = inst.InstanceType
+		case "vpc-id":
+			if ic.VpcId != nil {
+				field = *ic.VpcId
+			}
+		case "subnet-id":
+			if ic.SubnetId != nil {
+				field = *ic.SubnetId
+			}
+		case "tag-key":
+			if !matchTagKey(ic.Tags, values) {
+				return false
+			}
+			continue
+		case "tag-value":
+			if !matchTagValue(ic.Tags, values) {
+				return false
+			}
+			continue
+		default:
+			return false
+		}
+
+		if !filterutil.MatchesAny(values, field) {
+			return false
+		}
+	}
+
+	tags := filterutil.EC2TagsToMap(ic.Tags)
+	return filterutil.MatchesTags(filters, tags)
+}
+
+func matchTagKey(tags []*ec2.Tag, values []string) bool {
+	for _, t := range tags {
+		if t.Key != nil && filterutil.MatchesAny(values, *t.Key) {
+			return true
+		}
+	}
+	return false
+}
+
+func matchTagValue(tags []*ec2.Tag, values []string) bool {
+	for _, t := range tags {
+		if t.Value != nil && filterutil.MatchesAny(values, *t.Value) {
+			return true
+		}
+	}
+	return false
+}
+
+// DescribeInstances returns instances on this node visible to the caller's account.
+func (s *InstanceServiceImpl) DescribeInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
+	slog.Info("Processing DescribeInstances request from this node", "accountID", accountID)
+
+	s.instances.Mu.Lock()
+	defer s.instances.Mu.Unlock()
+
+	instanceIDFilter := make(map[string]bool)
+	for _, id := range input.InstanceIds {
+		if id != nil && *id != "" {
+			if !strings.HasPrefix(*id, "i-") {
+				return nil, errors.New(awserrors.ErrorInvalidInstanceIDMalformed)
+			}
+			instanceIDFilter[*id] = true
+		}
+	}
+
+	parsedFilters, err := filterutil.ParseFilters(input.Filters, DescribeInstancesValidFilters)
+	if err != nil {
+		slog.Warn("DescribeInstances: invalid filter", "err", err)
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+
+	reservationMap := make(map[string]*ec2.Reservation)
+
+	for _, instance := range s.instances.VMS {
+		if !IsInstanceVisible(accountID, instance.AccountID) {
+			continue
+		}
+		if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] {
+			continue
+		}
+		if instance.Reservation == nil || instance.Instance == nil {
+			continue
+		}
+
+		resID := ""
+		if instance.Reservation.ReservationId != nil {
+			resID = *instance.Reservation.ReservationId
+		}
+
+		if _, exists := reservationMap[resID]; !exists {
+			reservation := &ec2.Reservation{}
+			reservation.SetReservationId(resID)
+			if instance.Reservation.OwnerId != nil {
+				reservation.SetOwnerId(*instance.Reservation.OwnerId)
+			}
+			reservation.Instances = []*ec2.Instance{}
+			reservationMap[resID] = reservation
+		}
+
+		instanceCopy := *instance.Instance
+		instanceCopy.State = &ec2.InstanceState{}
+
+		if instance.PublicIP != "" && instanceCopy.PublicIpAddress == nil {
+			instanceCopy.PublicIpAddress = aws.String(instance.PublicIP)
+		}
+
+		if info, ok := vm.EC2StateCodes[instance.Status]; ok {
+			instanceCopy.State.SetCode(info.Code)
+			instanceCopy.State.SetName(info.Name)
+		} else {
+			slog.Warn("Instance has unmapped status, reporting as pending",
+				"instanceId", instance.ID, "status", string(instance.Status))
+			instanceCopy.State.SetCode(0)
+			instanceCopy.State.SetName("pending")
+		}
+
+		if instance.PlacementGroupName != "" {
+			instanceCopy.Placement = &ec2.Placement{
+				GroupName:        aws.String(instance.PlacementGroupName),
+				AvailabilityZone: aws.String(s.config.AZ),
+			}
+		}
+
+		if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) {
+			continue
+		}
+
+		reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy)
+	}
+
+	reservations := make([]*ec2.Reservation, 0, len(reservationMap))
+	for _, reservation := range reservationMap {
+		reservations = append(reservations, reservation)
+	}
+
+	slog.Info("DescribeInstances completed", "count", len(reservations))
+	return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil
+}
+
+// DescribeInstanceTypes returns instance types provisionable on this node.
+// The "capacity" filter (when "true") expands each type to one entry per
+// available slot so callers can report cluster-wide capacity by aggregating.
+func (s *InstanceServiceImpl) DescribeInstanceTypes(input *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) {
+	slog.Info("Processing DescribeInstanceTypes request from this node")
+
+	if s.resourceMgr == nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	showCapacity := false
+	for _, f := range input.Filters {
+		if f.Name != nil && *f.Name == "capacity" {
+			for _, v := range f.Values {
+				if v != nil && *v == "true" {
+					showCapacity = true
+					break
+				}
+			}
+		}
+	}
+
+	filteredTypes := s.resourceMgr.GetAvailableInstanceTypeInfos(showCapacity)
+	slog.Info("DescribeInstanceTypes completed", "count", len(filteredTypes))
+	return &ec2.DescribeInstanceTypesOutput{InstanceTypes: filteredTypes}, nil
+}
+
+// DescribeStoppedInstances returns stopped instances from shared KV.
+func (s *InstanceServiceImpl) DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
+	if s.stoppedStore == nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	return s.describeInstancesFromKV(input, accountID, s.stoppedStore.ListStoppedInstances, 80, "stopped", "DescribeStoppedInstances")
+}
+
+// DescribeTerminatedInstances returns terminated instances from the terminated KV bucket.
+func (s *InstanceServiceImpl) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
+	if s.stoppedStore == nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	return s.describeInstancesFromKV(input, accountID, s.stoppedStore.ListTerminatedInstances, 48, "terminated", "DescribeTerminatedInstances")
+}
+
+func (s *InstanceServiceImpl) describeInstancesFromKV(input *ec2.DescribeInstancesInput, accountID string, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, opName string) (*ec2.DescribeInstancesOutput, error) {
+	instanceIDFilter := make(map[string]bool)
+	for _, id := range input.InstanceIds {
+		if id != nil {
+			instanceIDFilter[*id] = true
+		}
+	}
+
+	parsedFilters, filterErr := filterutil.ParseFilters(input.Filters, DescribeInstancesValidFilters)
+	if filterErr != nil {
+		slog.Warn(opName+": invalid filter", "err", filterErr)
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+
+	instances, err := listFn()
+	if err != nil {
+		slog.Error(opName+": failed to list instances", "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	reservationMap := make(map[string]*ec2.Reservation)
+
+	for _, instance := range instances {
+		if !IsInstanceVisible(accountID, instance.AccountID) {
+			continue
+		}
+		if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] {
+			continue
+		}
+		if instance.Reservation == nil || instance.Instance == nil {
+			slog.Warn(opName+": skipping instance with nil Reservation/Instance (data integrity issue)",
+				"instanceId", instance.ID)
+			continue
+		}
+
+		resID := ""
+		if instance.Reservation.ReservationId != nil {
+			resID = *instance.Reservation.ReservationId
+		}
+
+		if _, exists := reservationMap[resID]; !exists {
+			reservation := &ec2.Reservation{}
+			reservation.SetReservationId(resID)
+			if instance.Reservation.OwnerId != nil {
+				reservation.SetOwnerId(*instance.Reservation.OwnerId)
+			}
+			reservation.Instances = []*ec2.Instance{}
+			reservationMap[resID] = reservation
+		}
+
+		instanceCopy := *instance.Instance
+		instanceCopy.State = &ec2.InstanceState{}
+		if info, ok := vm.EC2StateCodes[instance.Status]; ok {
+			instanceCopy.State.SetCode(info.Code)
+			instanceCopy.State.SetName(info.Name)
+		} else {
+			instanceCopy.State.SetCode(fallbackCode)
+			instanceCopy.State.SetName(fallbackName)
+		}
+
+		if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) {
+			continue
+		}
+
+		reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy)
+	}
+
+	reservations := make([]*ec2.Reservation, 0, len(reservationMap))
+	for _, reservation := range reservationMap {
+		reservations = append(reservations, reservation)
+	}
+
+	slog.Info(opName+" completed", "count", len(reservations))
+	return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil
+}
+
+// DescribeInstanceAttribute returns a single requested attribute for an instance.
+// Checks running instances first (in-memory), then falls back to stopped instances in KV.
+func (s *InstanceServiceImpl) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) {
+	if input.InstanceId == nil || *input.InstanceId == "" {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+	if input.Attribute == nil || *input.Attribute == "" {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+
+	instanceID := *input.InstanceId
+	attribute := *input.Attribute
+
+	var instance *vm.VM
+	s.instances.Mu.Lock()
+	if running, ok := s.instances.VMS[instanceID]; ok {
+		instance = running
+	}
+	s.instances.Mu.Unlock()
+
+	if instance == nil {
+		if s.stoppedStore == nil {
+			slog.Error("DescribeInstanceAttribute: stopped store not available")
+			return nil, errors.New(awserrors.ErrorServerInternal)
+		}
+		stopped, err := s.stoppedStore.LoadStoppedInstance(instanceID)
+		if err != nil {
+			slog.Error("DescribeInstanceAttribute: failed to load stopped instance",
+				"instanceId", instanceID, "err", err)
+			return nil, errors.New(awserrors.ErrorServerInternal)
+		}
+		instance = stopped
+	}
+
+	if instance == nil {
+		slog.Warn("DescribeInstanceAttribute: instance not found",
+			"instanceId", instanceID)
+		return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+	}
+
+	if !IsInstanceVisible(accountID, instance.AccountID) {
+		slog.Warn("DescribeInstanceAttribute: instance not visible",
+			"instanceId", instanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID)
+		return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+	}
+
+	output := &ec2.DescribeInstanceAttributeOutput{
+		InstanceId: &instanceID,
+	}
+
+	switch attribute {
+	case ec2.InstanceAttributeNameInstanceType:
+		val := instance.InstanceType
+		output.InstanceType = &ec2.AttributeValue{Value: &val}
+
+	case ec2.InstanceAttributeNameUserData:
+		val := instance.UserData
+		output.UserData = &ec2.AttributeValue{Value: &val}
+
+	case ec2.InstanceAttributeNameDisableApiTermination:
+		val := false
+		output.DisableApiTermination = &ec2.AttributeBooleanValue{Value: &val}
+
+	case ec2.InstanceAttributeNameDisableApiStop:
+		val := false
+		output.DisableApiStop = &ec2.AttributeBooleanValue{Value: &val}
+
+	case ec2.InstanceAttributeNameInstanceInitiatedShutdownBehavior:
+		val := ec2.ShutdownBehaviorStop
+		output.InstanceInitiatedShutdownBehavior = &ec2.AttributeValue{Value: &val}
+
+	case ec2.InstanceAttributeNameEbsOptimized:
+		val := false
+		output.EbsOptimized = &ec2.AttributeBooleanValue{Value: &val}
+
+	case ec2.InstanceAttributeNameEnaSupport:
+		val := true
+		output.EnaSupport = &ec2.AttributeBooleanValue{Value: &val}
+
+	case ec2.InstanceAttributeNameSourceDestCheck:
+		val := true
+		output.SourceDestCheck = &ec2.AttributeBooleanValue{Value: &val}
+
+	case ec2.InstanceAttributeNameGroupSet:
+		if instance.Instance != nil && len(instance.Instance.SecurityGroups) > 0 {
+			output.Groups = instance.Instance.SecurityGroups
+		} else {
+			output.Groups = []*ec2.GroupIdentifier{}
+		}
+
+	default:
+		slog.Warn("DescribeInstanceAttribute: unsupported attribute",
+			"instanceId", instanceID, "attribute", attribute)
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+
+	slog.Info("DescribeInstanceAttribute: completed",
+		"instanceId", instanceID, "attribute", attribute, "accountID", accountID)
+	return output, nil
+}

spinifex/handlers/ec2/instance/service_impl_test.go

@@ -26,7 +26,7 @@ func TestNewInstanceServiceImpl(t *testing.T) {
 	store := objectstore.NewMemoryObjectStore()
 	instances := &vm.Instances{VMS: make(map[string]*vm.VM)}
 
-	svc := NewInstanceServiceImpl(cfg, instanceTypes, nil, instances, store)
+	svc := NewInstanceServiceImpl(cfg, instanceTypes, nil, instances, store, nil, nil)
 
 	require.NotNil(t, svc)
 	assert.Equal(t, cfg, svc.config)

spinifex/handlers/ec2/instance/service_mock.go

@@ -33,3 +33,23 @@ func (s *MockInstanceService) RunInstances(input *ec2.RunInstancesInput, account
 
 	return reservation, nil
 }
+
+func (s *MockInstanceService) DescribeInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) {
+	return &ec2.DescribeInstancesOutput{}, nil
+}
+
+func (s *MockInstanceService) DescribeInstanceTypes(_ *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) {
+	return &ec2.DescribeInstanceTypesOutput{}, nil
+}
+
+func (s *MockInstanceService) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, _ string) (*ec2.DescribeInstanceAttributeOutput, error) {
+	return &ec2.DescribeInstanceAttributeOutput{InstanceId: input.InstanceId}, nil
+}
+
+func (s *MockInstanceService) DescribeStoppedInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) {
+	return &ec2.DescribeInstancesOutput{}, nil
+}
+
+func (s *MockInstanceService) DescribeTerminatedInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) {
+	return &ec2.DescribeInstancesOutput{}, nil
+}

spinifex/handlers/ec2/instance/service_nats.go

@@ -10,7 +10,14 @@ import (
 	"github.com/nats-io/nats.go"
 )
 
-// NATSInstanceService handles instance operations via NATS messaging
+// NATSInstanceService handles instance operations via NATS messaging.
+//
+// Note: DescribeInstances and DescribeInstanceTypes subscriptions are
+// fan-out (no queue group); a single NATSRequest only captures one
+// responder. Cluster-wide aggregation is the gateway's job (see
+// spinifex/gateway/ec2/instance). The methods below are kept for symmetry
+// with the InstanceService interface and intentionally fail rather than
+// silently returning a one-node view.
 type NATSInstanceService struct {
 	natsConn *nats.Conn
 }
@@ -27,3 +34,27 @@ func (s *NATSInstanceService) RunInstances(input *ec2.RunInstancesInput, account
 	topic := fmt.Sprintf("ec2.RunInstances.%s", aws.StringValue(input.InstanceType))
 	return utils.NATSRequest[ec2.Reservation](s.natsConn, topic, input, 5*time.Minute, accountID)
 }
+
+// DescribeInstances is a fan-out operation; use the gateway's scatter-gather
+// helper instead of this single-node client.
+func (s *NATSInstanceService) DescribeInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) {
+	return nil, fmt.Errorf("DescribeInstances is fan-out; use gateway scatter-gather, not NATSInstanceService")
+}
+
+// DescribeInstanceTypes is a fan-out operation; use the gateway's scatter-gather
+// helper instead of this single-node client.
+func (s *NATSInstanceService) DescribeInstanceTypes(_ *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) {
+	return nil, fmt.Errorf("DescribeInstanceTypes is fan-out; use gateway scatter-gather, not NATSInstanceService")
+}
+
+func (s *NATSInstanceService) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) {
+	return utils.NATSRequest[ec2.DescribeInstanceAttributeOutput](s.natsConn, "ec2.DescribeInstanceAttribute", input, 30*time.Second, accountID)
+}
+
+func (s *NATSInstanceService) DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
+	return utils.NATSRequest[ec2.DescribeInstancesOutput](s.natsConn, "ec2.DescribeStoppedInstances", input, 30*time.Second, accountID)
+}
+
+func (s *NATSInstanceService) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
+	return utils.NATSRequest[ec2.DescribeInstancesOutput](s.natsConn, "ec2.DescribeTerminatedInstances", input, 30*time.Second, accountID)
+}