feat: EC2 instance tagging (#473)

spinifex/daemon/daemon.go

@@ -1369,7 +1369,7 @@ func (d *Daemon) startCluster() error {
 		}
 	}
 
-	d.instanceService.SetTerminationDeps(d.volumeService, d.vpcService, d.externalIPAM)
+	d.instanceService.SetTerminationDeps(d.volumeService, d.vpcService, d.externalIPAM, d.tagsService)
 	d.instanceService.SetRunInstancesDeps(d.imageService, d.keyService, &daemonENICreator{d: d}, d.externalIPAM)
 
 	if d.gpuManager != nil {

spinifex/daemon/daemon_handlers.go

@@ -117,6 +117,8 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
 		d.handleAssociateIamInstanceProfile(msg, command, instance)
 	case command.Attributes.SetSpotLineage:
 		d.handleSetSpotLineage(msg, command)
+	case command.Attributes.SetInstanceTags, command.Attributes.RemoveInstanceTags:
+		d.handleSetInstanceTags(msg, command, instance)
 	case command.Attributes.StartInstance:
 		if err := d.instanceService.StartInstance(instance, command); err != nil {
 			respondWithError(msg, awserrors.ValidErrorCode(err.Error()))

spinifex/daemon/daemon_handlers_instance.go

@@ -12,6 +12,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
+	handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
+	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go"
@@ -25,6 +27,61 @@ import (
 // the deterministic tap name (TUNSETIFF: Device or resource busy).
 const startStoppedForwardTimeout = 30 * time.Second
 
+// handleSetInstanceTags applies a create-tags/delete-tags mutation to a running
+// instance: central store first, then the record under the manager lock, so a
+// failed S3 write leaves both stores untouched, matching the stopped path.
+// Ownership is checked by checkInstanceOwnership before dispatch.
+func (d *Daemon) handleSetInstanceTags(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) {
+	remove := command.Attributes.RemoveInstanceTags
+	data := command.InstanceTagsData
+	if data == nil || (!remove && len(data.Tags) == 0) {
+		respondWithError(msg, awserrors.ErrorMissingParameter)
+		return
+	}
+
+	var newTags []*ec2.Tag
+	missingRecord := false
+	d.vmMgr.Inspect(instance, func(v *vm.VM) {
+		if v.Instance == nil {
+			missingRecord = true
+			return
+		}
+		newTags = handlers_ec2_instance.ApplyInstanceTagMutation(v.Instance.Tags, data, remove)
+	})
+	if missingRecord {
+		respondWithError(msg, awserrors.ErrorServerInternal)
+		return
+	}
+
+	accountID := utils.AccountIDFromMsg(msg)
+	if err := d.tagsService.PutResourceTags(accountID, instance.ID, handlers_ec2_instance.TagsToMap(newTags)); err != nil {
+		slog.Error("SetInstanceTags: central tag store write failed",
+			"instanceId", instance.ID, "err", err)
+		respondWithError(msg, awserrors.ErrorServerInternal)
+		return
+	}
+
+	found, err := d.vmMgr.UpdateAndPersist(instance.ID, func(v *vm.VM) bool {
+		if v.Instance == nil {
+			return false
+		}
+		v.Instance.Tags = handlers_ec2_instance.ApplyInstanceTagMutation(v.Instance.Tags, data, remove)
+		return true
+	})
+	if err != nil {
+		respondWithError(msg, awserrors.ValidErrorCode(err.Error()))
+		return
+	}
+	if !found {
+		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+		return
+	}
+
+	if err := msg.Respond([]byte(`{}`)); err != nil {
+		slog.Error("Failed to respond to NATS request", "err", err)
+	}
+}
+
 // handleEC2RunInstances orchestrates the RunInstances flow across
 // InstanceService.PrepareRunInstances (validation + ENI creation),
 // daemon-local Insert/WriteState/per-instance subscribe, and
@@ -102,6 +159,20 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
 	}
 	slog.Info("Instances added to state with pending status", "count", len(instances))
 
+	// Project launch tags into the central tag store so describe-tags agrees
+	// with the record from birth. Best-effort: the reservation is already
+	// returned, so a failed write is logged rather than failing the launch.
+	for _, instance := range instances {
+		if instance.Instance == nil || len(instance.Instance.Tags) == 0 {
+			continue
+		}
+		if err := d.tagsService.PutResourceTags(accountID, instance.ID,
+			handlers_ec2_instance.TagsToMap(instance.Instance.Tags)); err != nil {
+			slog.Error("handleEC2RunInstances: launch tag central store write failed",
+				"instanceId", instance.ID, "err", err)
+		}
+	}
+
 	// Subscribe per-instance NATS topics so terminate/stop reach this daemon
 	// while volumes prepare. LaunchInstance replaces these on success.
 	d.mu.Lock()

spinifex/daemon/daemon_handlers_instance_tags_test.go

@@ -0,0 +1,243 @@
+package daemon
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
+	handlers_ec2_tags "github.com/mulgadc/spinifex/spinifex/handlers/ec2/tags"
+	"github.com/mulgadc/spinifex/spinifex/objectstore"
+	"github.com/mulgadc/spinifex/spinifex/types"
+	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// memStoppedStore is a map-backed StoppedInstanceStore standing in for the
+// shared JetStream KV in tag fallback tests.
+type memStoppedStore struct {
+	instances map[string]*vm.VM
+}
+
+var _ handlers_ec2_instance.StoppedInstanceStore = (*memStoppedStore)(nil)
+
+func (m *memStoppedStore) LoadStoppedInstance(id string) (*vm.VM, error) {
+	return m.instances[id], nil
+}
+
+func (m *memStoppedStore) ListStoppedInstances() ([]*vm.VM, error) {
+	var out []*vm.VM
+	for _, v := range m.instances {
+		out = append(out, v)
+	}
+	return out, nil
+}
+
+func (m *memStoppedStore) ListTerminatedInstances() ([]*vm.VM, error) { return nil, nil }
+
+func (m *memStoppedStore) WriteStoppedInstance(id string, instance *vm.VM) error {
+	if m.instances == nil {
+		m.instances = make(map[string]*vm.VM)
+	}
+	m.instances[id] = instance
+	return nil
+}
+
+func (m *memStoppedStore) DeleteStoppedInstance(id string) error {
+	delete(m.instances, id)
+	return nil
+}
+
+func (m *memStoppedStore) WriteTerminatedInstance(string, *vm.VM) error { return nil }
+
+// tagTestDaemon returns a test daemon with an in-memory central tag store, an
+// empty stopped store, and a running VM carrying the given initial tags.
+func tagTestDaemon(t *testing.T, instanceID string, initial map[string]string) *Daemon {
+	d, _ := tagTestDaemonWithStopped(t, instanceID, initial)
+	return d
+}
+
+// tagTestDaemonWithStopped is tagTestDaemon exposing the stopped store, so
+// tests can seed stopped instances for the no-running-owner fallback.
+func tagTestDaemonWithStopped(t *testing.T, instanceID string, initial map[string]string) (*Daemon, *memStoppedStore) {
+	t.Helper()
+	d := createTestDaemon(t, sharedNATSURL)
+	d.tagsService = handlers_ec2_tags.NewTagsServiceImplWithStore(d.config, objectstore.NewMemoryObjectStore())
+
+	stopped := &memStoppedStore{instances: map[string]*vm.VM{}}
+	d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(
+		d.config, d.resourceMgr.instanceTypes, d.natsConn,
+		objectstore.NewMemoryObjectStore(), d.vmMgr, d.resourceMgr, stopped)
+
+	var tags []*ec2.Tag
+	for k, v := range initial {
+		tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
+	}
+	d.vmMgr.Insert(&vm.VM{
+		ID:        instanceID,
+		Status:    vm.StateRunning,
+		AccountID: testAccountID,
+		Instance:  &ec2.Instance{InstanceId: aws.String(instanceID), Tags: tags},
+	})
+	return d, stopped
+}
+
+// centralTags reads a resource's tags back out of the central store via
+// DescribeTags, as the given account.
+func centralTags(t *testing.T, d *Daemon, accountID, resourceID string) map[string]string {
+	t.Helper()
+	out, err := d.tagsService.DescribeTags(&ec2.DescribeTagsInput{
+		Filters: []*ec2.Filter{{Name: aws.String("resource-id"), Values: []*string{aws.String(resourceID)}}},
+	}, accountID)
+	require.NoError(t, err)
+	tags := make(map[string]string, len(out.Tags))
+	for _, td := range out.Tags {
+		tags[aws.StringValue(td.Key)] = aws.StringValue(td.Value)
+	}
+	return tags
+}
+
+func recordTags(t *testing.T, d *Daemon, instanceID string) map[string]string {
+	t.Helper()
+	got, ok := d.vmMgr.Get(instanceID)
+	require.True(t, ok)
+	return tagsAsMap(got.Instance.Tags)
+}
+
+func tagsAsMap(tags []*ec2.Tag) map[string]string {
+	out := make(map[string]string, len(tags))
+	for _, tag := range tags {
+		out[aws.StringValue(tag.Key)] = aws.StringValue(tag.Value)
+	}
+	return out
+}
+
+func tagCommand(instanceID string, attrs types.EC2CommandAttributes, data *types.InstanceTagsData) []byte {
+	body, _ := json.Marshal(types.EC2InstanceCommand{ID: instanceID, Attributes: attrs, InstanceTagsData: data})
+	return body
+}
+
+// SetInstanceTags merges the upsert set into the record and projects the full
+// tag set into the central store in the same call.
+func TestHandleSetInstanceTags_MergesAndWritesCentral(t *testing.T) {
+	const id = "i-tag-set"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web", "env": "dev"})
+
+	body := tagCommand(id, types.EC2CommandAttributes{SetInstanceTags: true},
+		&types.InstanceTagsData{Tags: map[string]string{"env": "prod", "team": "infra"}})
+	reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, body)
+	assert.JSONEq(t, `{}`, string(reply.Data))
+
+	want := map[string]string{"Name": "web", "env": "prod", "team": "infra"}
+	assert.Equal(t, want, recordTags(t, d, id))
+	assert.Equal(t, want, centralTags(t, d, testAccountID, id))
+}
+
+// RemoveInstanceTags removes named keys unconditionally, value-matched keys
+// only on match, and both stores reflect the result.
+func TestHandleSetInstanceTags_RemoveWritesBothStores(t *testing.T) {
+	const id = "i-tag-remove"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web", "env": "dev", "team": "infra"})
+	require.NoError(t, d.tagsService.PutResourceTags(testAccountID, id,
+		map[string]string{"Name": "web", "env": "dev", "team": "infra"}))
+
+	body := tagCommand(id, types.EC2CommandAttributes{RemoveInstanceTags: true},
+		&types.InstanceTagsData{TagKeys: []string{"team"}, Tags: map[string]string{"env": "prod", "Name": "web"}})
+	reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, body)
+	assert.JSONEq(t, `{}`, string(reply.Data))
+
+	want := map[string]string{"env": "dev"}
+	assert.Equal(t, want, recordTags(t, d, id))
+	assert.Equal(t, want, centralTags(t, d, testAccountID, id))
+}
+
+// RemoveInstanceTags with empty Tags and TagKeys clears every tag.
+func TestHandleSetInstanceTags_RemoveClearAll(t *testing.T) {
+	const id = "i-tag-clear"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web", "env": "dev"})
+	require.NoError(t, d.tagsService.PutResourceTags(testAccountID, id,
+		map[string]string{"Name": "web", "env": "dev"}))
+
+	body := tagCommand(id, types.EC2CommandAttributes{RemoveInstanceTags: true}, &types.InstanceTagsData{})
+	reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, body)
+	assert.JSONEq(t, `{}`, string(reply.Data))
+
+	assert.Empty(t, recordTags(t, d, id))
+	assert.Empty(t, centralTags(t, d, testAccountID, id))
+}
+
+// A cross-account caller is rejected with InvalidID.NotFound before dispatch
+// and neither the record nor either central namespace is written.
+func TestHandleSetInstanceTags_CrossAccountRejected(t *testing.T) {
+	const id = "i-tag-cross"
+	const attacker = "999999999999"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web"})
+
+	body := tagCommand(id, types.EC2CommandAttributes{SetInstanceTags: true},
+		&types.InstanceTagsData{Tags: map[string]string{"stolen": "yes"}})
+	reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, attacker, body)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, decodeError(t, reply.Data)["Code"])
+
+	assert.Equal(t, map[string]string{"Name": "web"}, recordTags(t, d, id))
+	assert.Empty(t, centralTags(t, d, testAccountID, id))
+	assert.Empty(t, centralTags(t, d, attacker, id))
+}
+
+// RunInstances with instance-scoped TagSpecifications projects the launch tags
+// into the central tag store, so describe-tags sees them from birth. The write
+// happens after the reservation reply, hence the Eventually.
+func TestHandleEC2RunInstances_LaunchTagsWriteCentralStore(t *testing.T) {
+	daemon, memStore := createFullTestDaemonWithStore(t, sharedNATSURL)
+	seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-launchtags")
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances.launchtags", "spinifex-workers", daemon.handleEC2RunInstances)
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	input := &ec2.RunInstancesInput{
+		ImageId:      aws.String("ami-launchtags"),
+		InstanceType: aws.String(getTestInstanceType(t)),
+		MinCount:     aws.Int64(1),
+		MaxCount:     aws.Int64(1),
+		TagSpecifications: []*ec2.TagSpecification{{
+			ResourceType: aws.String("instance"),
+			Tags: []*ec2.Tag{
+				{Key: aws.String("Name"), Value: aws.String("web")},
+				{Key: aws.String("env"), Value: aws.String("dev")},
+			},
+		}},
+	}
+	reply, err := natsRequest(daemon.natsConn, "ec2.RunInstances.launchtags", mustMarshal(t, input), 5*time.Second)
+	require.NoError(t, err)
+
+	var reservation ec2.Reservation
+	require.NoError(t, json.Unmarshal(reply.Data, &reservation))
+	require.Len(t, reservation.Instances, 1, "launch must succeed and return the instance: %s", reply.Data)
+	id := aws.StringValue(reservation.Instances[0].InstanceId)
+
+	want := map[string]string{"Name": "web", "env": "dev"}
+	assert.Eventually(t, func() bool {
+		return assert.ObjectsAreEqual(want, centralTags(t, daemon, testAccountID, id))
+	}, 5*time.Second, 50*time.Millisecond, "central tag store must receive launch tags")
+	assert.Equal(t, want, recordTags(t, daemon, id))
+}
+
+// Missing InstanceTagsData, and a set with no tags, are rejected with
+// MissingParameter and leave both stores untouched.
+func TestHandleSetInstanceTags_RejectsMissingData(t *testing.T) {
+	const id = "i-tag-nodata"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web"})
+
+	for _, data := range []*types.InstanceTagsData{nil, {}} {
+		body := tagCommand(id, types.EC2CommandAttributes{SetInstanceTags: true}, data)
+		reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, body)
+		assert.Equal(t, awserrors.ErrorMissingParameter, decodeError(t, reply.Data)["Code"])
+	}
+
+	assert.Equal(t, map[string]string{"Name": "web"}, recordTags(t, d, id))
+	assert.Empty(t, centralTags(t, d, testAccountID, id))
+}

spinifex/daemon/daemon_handlers_tag.go

@@ -1,10 +1,25 @@
 package daemon
 
 import (
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"strings"
+	"sync"
+	"time"
+
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/mulgadc/spinifex/spinifex/types"
+	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
 
+// instanceTagCommandTimeout bounds each ec2.cmd owner request. A stopped or
+// absent instance has no subscriber and fails fast with no responders, so
+// only a partitioned-but-subscribed owner pays the full timeout.
+const instanceTagCommandTimeout = 5 * time.Second
+
 func (d *Daemon) handleEC2CreateTags(msg *nats.Msg) {
 	handleNATSRequest(msg, d.createTags)
 }
@@ -62,32 +77,163 @@ func (d *Daemon) recordTagMirrors() []recordTagMirror {
 	return mirrors
 }
 
-// createTags writes the generic tag store, then mirrors tags into the owning
-// records so tag-filtered describes observe them.
-func (d *Daemon) createTags(input *ec2.CreateTagsInput, accountID string) (*ec2.CreateTagsOutput, error) {
-	out, err := d.tagsService.CreateTags(input, accountID)
-	if err != nil {
-		return nil, err
-	}
-	for _, m := range d.recordTagMirrors() {
-		if err := m.ApplyRecordTags(input, accountID); err != nil {
-			return nil, err
+// splitInstanceResources separates instance IDs from the other resource IDs.
+// Instances never take the generic central write: their central entry is only
+// written by the ownership-gated path co-located with the record write.
+func splitInstanceResources(resources []*string) (instanceIDs []string, rest []*string) {
+	for _, r := range resources {
+		if r == nil {
+			continue
+		}
+		if strings.HasPrefix(*r, "i-") {
+			instanceIDs = append(instanceIDs, *r)
+		} else {
+			rest = append(rest, r)
 		}
 	}
-	return out, nil
+	return instanceIDs, rest
 }
 
-// deleteTags removes from the generic tag store, then mirrors the removal
-// into the owning records.
-func (d *Daemon) deleteTags(input *ec2.DeleteTagsInput, accountID string) (*ec2.DeleteTagsOutput, error) {
-	out, err := d.tagsService.DeleteTags(input, accountID)
-	if err != nil {
-		return nil, err
+// createTags routes instance IDs to their owning daemon, which writes the
+// record and the central store together, and writes everything else to the
+// generic tag store plus the owning record mirrors.
+func (d *Daemon) createTags(input *ec2.CreateTagsInput, accountID string) (*ec2.CreateTagsOutput, error) {
+	if input == nil {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
 	}
-	for _, m := range d.recordTagMirrors() {
-		if err := m.RemoveRecordTags(input, accountID); err != nil {
+	if len(input.Resources) == 0 || len(input.Tags) == 0 {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+
+	instanceIDs, rest := splitInstanceResources(input.Resources)
+	if len(rest) > 0 {
+		restInput := *input
+		restInput.Resources = rest
+		if _, err := d.tagsService.CreateTags(&restInput, accountID); err != nil {
 			return nil, err
 		}
+		for _, m := range d.recordTagMirrors() {
+			if err := m.ApplyRecordTags(&restInput, accountID); err != nil {
+				return nil, err
+			}
+		}
 	}
-	return out, nil
+
+	data := &types.InstanceTagsData{Tags: make(map[string]string, len(input.Tags))}
+	for _, tag := range input.Tags {
+		if tag != nil && tag.Key != nil && tag.Value != nil {
+			data.Tags[*tag.Key] = *tag.Value
+		}
+	}
+	if err := d.tagInstances(instanceIDs, data, false, accountID); err != nil {
+		return nil, err
+	}
+	return &ec2.CreateTagsOutput{}, nil
+}
+
+// deleteTags is createTags' removal counterpart: instances go via their owner,
+// the rest via the generic tag store plus record mirrors.
+func (d *Daemon) deleteTags(input *ec2.DeleteTagsInput, accountID string) (*ec2.DeleteTagsOutput, error) {
+	if input == nil {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	if len(input.Resources) == 0 {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+
+	instanceIDs, rest := splitInstanceResources(input.Resources)
+	if len(rest) > 0 {
+		restInput := *input
+		restInput.Resources = rest
+		if _, err := d.tagsService.DeleteTags(&restInput, accountID); err != nil {
+			return nil, err
+		}
+		for _, m := range d.recordTagMirrors() {
+			if err := m.RemoveRecordTags(&restInput, accountID); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	// Keys without a value delete unconditionally; keys with a value delete
+	// only on match; an empty input.Tags clears every tag.
+	data := &types.InstanceTagsData{}
+	for _, tag := range input.Tags {
+		if tag == nil || tag.Key == nil {
+			continue
+		}
+		if tag.Value == nil {
+			data.TagKeys = append(data.TagKeys, *tag.Key)
+			continue
+		}
+		if data.Tags == nil {
+			data.Tags = make(map[string]string)
+		}
+		data.Tags[*tag.Key] = *tag.Value
+	}
+	if err := d.tagInstances(instanceIDs, data, true, accountID); err != nil {
+		return nil, err
+	}
+	return &ec2.DeleteTagsOutput{}, nil
+}
+
+// tagInstances sends the tag mutation to each instance's owner concurrently,
+// so a many-instance call costs one owner timeout rather than one per
+// instance. The first failure in input order is returned.
+func (d *Daemon) tagInstances(instanceIDs []string, data *types.InstanceTagsData, remove bool, accountID string) error {
+	errs := make([]error, len(instanceIDs))
+	var wg sync.WaitGroup
+	for i, id := range instanceIDs {
+		wg.Go(func() {
+			errs[i] = d.tagInstance(id, data, remove, accountID)
+		})
+	}
+	wg.Wait()
+	for _, err := range errs {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// tagInstance asks the owning daemon over ec2.cmd.<id> to apply the mutation
+// to the record and central store together; only the owner subscribes there.
+// No responders means the instance isn't running, so the mutation falls back
+// to the shared stopped store; a timeout (partitioned-but-subscribed owner)
+// surfaces InvalidID.NotFound and writes nothing.
+func (d *Daemon) tagInstance(instanceID string, data *types.InstanceTagsData, remove bool, accountID string) error {
+	command := types.EC2InstanceCommand{
+		ID: instanceID,
+		Attributes: types.EC2CommandAttributes{
+			SetInstanceTags:    !remove,
+			RemoveInstanceTags: remove,
+		},
+		InstanceTagsData: data,
+	}
+	body, err := json.Marshal(command)
+	if err != nil {
+		slog.Error("tagInstance: failed to marshal command", "instanceId", instanceID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	reqMsg := nats.NewMsg("ec2.cmd." + instanceID)
+	reqMsg.Data = body
+	reqMsg.Header.Set(utils.AccountIDHeader, accountID)
+
+	msg, err := d.natsConn.RequestMsg(reqMsg, instanceTagCommandTimeout)
+	switch {
+	case errors.Is(err, nats.ErrNoResponders):
+		return d.instanceService.TagStoppedInstance(instanceID, data, remove, d.tagsService, accountID)
+	case errors.Is(err, nats.ErrTimeout):
+		return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+	case err != nil:
+		slog.Error("tagInstance: owner request failed", "instanceId", instanceID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	if responseError, parseErr := utils.ValidateErrorPayload(msg.Data); parseErr != nil {
+		return errors.New(*responseError.Code)
+	}
+	return nil
 }

spinifex/daemon/daemon_handlers_tag_test.go

@@ -0,0 +1,206 @@
+package daemon
+
+import (
+	"testing"
+
+	"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/vm"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// subscribeOwner registers this daemon as the instance's owner on ec2.cmd.<id>,
+// the way daemon startup does for every VM it runs.
+func subscribeOwner(t *testing.T, d *Daemon, instanceID string) {
+	t.Helper()
+	sub, err := d.natsConn.Subscribe("ec2.cmd."+instanceID, d.handleEC2Events)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+// A mixed create-tags writes the instance via the owner-gated co-located path
+// and the non-instance resource via the generic central write.
+func TestCreateTags_MixedResourceDispatch(t *testing.T) {
+	const id = "i-tag-mixed"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web"})
+	subscribeOwner(t, d, id)
+
+	out, err := d.createTags(&ec2.CreateTagsInput{
+		Resources: []*string{aws.String(id), aws.String("vol-tag-mixed")},
+		Tags:      []*ec2.Tag{{Key: aws.String("env"), Value: aws.String("prod")}},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, out)
+
+	want := map[string]string{"Name": "web", "env": "prod"}
+	assert.Equal(t, want, recordTags(t, d, id))
+	assert.Equal(t, want, centralTags(t, d, testAccountID, id))
+	assert.Equal(t, map[string]string{"env": "prod"}, centralTags(t, d, testAccountID, "vol-tag-mixed"))
+}
+
+// delete-tags routes instance removals via the owner: bare keys delete
+// unconditionally, valued keys only on match, and both stores agree after.
+func TestDeleteTags_InstanceRoutedViaOwner(t *testing.T) {
+	const id = "i-tag-del"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web", "env": "dev", "team": "infra"})
+	subscribeOwner(t, d, id)
+	require.NoError(t, d.tagsService.PutResourceTags(testAccountID, id,
+		map[string]string{"Name": "web", "env": "dev", "team": "infra"}))
+
+	_, err := d.deleteTags(&ec2.DeleteTagsInput{
+		Resources: []*string{aws.String(id)},
+		Tags: []*ec2.Tag{
+			{Key: aws.String("team")},
+			{Key: aws.String("env"), Value: aws.String("prod")},
+			{Key: aws.String("Name"), Value: aws.String("web")},
+		},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	want := map[string]string{"env": "dev"}
+	assert.Equal(t, want, recordTags(t, d, id))
+	assert.Equal(t, want, centralTags(t, d, testAccountID, id))
+}
+
+// delete-tags with no Tags clears every tag on the instance in both stores.
+func TestDeleteTags_InstanceClearAll(t *testing.T) {
+	const id = "i-tag-delall"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web", "env": "dev"})
+	subscribeOwner(t, d, id)
+	require.NoError(t, d.tagsService.PutResourceTags(testAccountID, id,
+		map[string]string{"Name": "web", "env": "dev"}))
+
+	_, err := d.deleteTags(&ec2.DeleteTagsInput{
+		Resources: []*string{aws.String(id)},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	assert.Empty(t, recordTags(t, d, id))
+	assert.Empty(t, centralTags(t, d, testAccountID, id))
+}
+
+// An instance with no responding owner returns InvalidID.NotFound and no
+// central entry is created for it, while co-listed resources still land.
+func TestCreateTags_NoOwnerNotFound(t *testing.T) {
+	const absent = "i-tag-absent"
+	d := tagTestDaemon(t, "i-tag-unrelated", nil)
+
+	_, err := d.createTags(&ec2.CreateTagsInput{
+		Resources: []*string{aws.String("vol-tag-still"), aws.String(absent)},
+		Tags:      []*ec2.Tag{{Key: aws.String("env"), Value: aws.String("prod")}},
+	}, testAccountID)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+
+	assert.Empty(t, centralTags(t, d, testAccountID, absent))
+	assert.Equal(t, map[string]string{"env": "prod"}, centralTags(t, d, testAccountID, "vol-tag-still"))
+}
+
+// A many-instance call fans out concurrently: absent owners all fail fast
+// with NotFound rather than serialising owner requests.
+func TestDeleteTags_NoOwnerNotFound(t *testing.T) {
+	d := tagTestDaemon(t, "i-tag-unrelated2", nil)
+
+	_, err := d.deleteTags(&ec2.DeleteTagsInput{
+		Resources: []*string{aws.String("i-tag-gone1"), aws.String("i-tag-gone2")},
+	}, testAccountID)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+}
+
+// With no running owner, create-tags falls back to the shared stopped store
+// and the mutation lands on both the stopped record and the central store.
+func TestCreateTags_StoppedFallback(t *testing.T) {
+	const id = "i-tag-stopfall"
+	d, stopped := tagTestDaemonWithStopped(t, "i-tag-unrelated3", nil)
+	stopped.instances[id] = &vm.VM{
+		ID:        id,
+		Status:    vm.StateStopped,
+		AccountID: testAccountID,
+		Instance: &ec2.Instance{InstanceId: aws.String(id),
+			Tags: []*ec2.Tag{{Key: aws.String("Name"), Value: aws.String("web")}}},
+	}
+
+	_, err := d.createTags(&ec2.CreateTagsInput{
+		Resources: []*string{aws.String(id)},
+		Tags:      []*ec2.Tag{{Key: aws.String("env"), Value: aws.String("prod")}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	want := map[string]string{"Name": "web", "env": "prod"}
+	assert.Equal(t, want, tagsAsMap(stopped.instances[id].Instance.Tags))
+	assert.Equal(t, want, centralTags(t, d, testAccountID, id))
+}
+
+// A stopped instance owned by another account is rejected via the fallback's
+// owner check and nothing is written to either store.
+func TestDeleteTags_StoppedCrossAccountRejected(t *testing.T) {
+	const id = "i-tag-stopcross"
+	const attacker = "999999999999"
+	d, stopped := tagTestDaemonWithStopped(t, "i-tag-unrelated4", nil)
+	stopped.instances[id] = &vm.VM{
+		ID:        id,
+		Status:    vm.StateStopped,
+		AccountID: testAccountID,
+		Instance: &ec2.Instance{InstanceId: aws.String(id),
+			Tags: []*ec2.Tag{{Key: aws.String("Name"), Value: aws.String("web")}}},
+	}
+
+	_, err := d.deleteTags(&ec2.DeleteTagsInput{
+		Resources: []*string{aws.String(id)},
+	}, attacker)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+
+	assert.Equal(t, map[string]string{"Name": "web"}, tagsAsMap(stopped.instances[id].Instance.Tags))
+	assert.Empty(t, centralTags(t, d, attacker, id))
+}
+
+// A daemon-side error reply from the owner (cross-account NotFound) is
+// relayed to the caller and nothing lands in either central namespace.
+func TestCreateTags_OwnerErrorRelayed(t *testing.T) {
+	const id = "i-tag-crossacct"
+	const attacker = "999999999999"
+	d := tagTestDaemon(t, id, map[string]string{"Name": "web"})
+	subscribeOwner(t, d, id)
+
+	_, err := d.createTags(&ec2.CreateTagsInput{
+		Resources: []*string{aws.String(id)},
+		Tags:      []*ec2.Tag{{Key: aws.String("stolen"), Value: aws.String("yes")}},
+	}, attacker)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+
+	assert.Equal(t, map[string]string{"Name": "web"}, recordTags(t, d, id))
+	assert.Empty(t, centralTags(t, d, testAccountID, id))
+	assert.Empty(t, centralTags(t, d, attacker, id))
+}
+
+func TestCreateTags_Validation(t *testing.T) {
+	d := tagTestDaemon(t, "i-tag-valid", nil)
+
+	_, err := d.createTags(nil, testAccountID)
+	assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
+
+	_, err = d.createTags(&ec2.CreateTagsInput{
+		Tags: []*ec2.Tag{{Key: aws.String("k"), Value: aws.String("v")}},
+	}, testAccountID)
+	assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
+
+	_, err = d.createTags(&ec2.CreateTagsInput{
+		Resources: []*string{aws.String("i-tag-valid")},
+	}, testAccountID)
+	assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
+}
+
+func TestDeleteTags_Validation(t *testing.T) {
+	d := tagTestDaemon(t, "i-tag-valid2", nil)
+
+	_, err := d.deleteTags(nil, testAccountID)
+	assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
+
+	_, err = d.deleteTags(&ec2.DeleteTagsInput{}, testAccountID)
+	assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
+}

spinifex/daemon/daemon_handlers_test.go

@@ -1921,6 +1921,14 @@ func TestHandleEC2DescribeInstanceAttribute_InvalidJSON(t *testing.T) {
 func TestDelegateHandlers_RoundTrip(t *testing.T) {
 	daemon := createFullTestDaemon(t, sharedNATSURL)
 
+	// DeleteTags' no-owner path falls back to the shared stopped store; give
+	// the service an empty one so an absent instance resolves to NotFound.
+	daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(
+		daemon.config, daemon.resourceMgr.instanceTypes, daemon.natsConn,
+		objectstore.NewMemoryObjectStore(), daemon.vmMgr, daemon.resourceMgr,
+		&memStoppedStore{})
+	daemon.instanceService.SetRunInstancesDeps(daemon.imageService, daemon.keyService, nil, nil)
+
 	tests := []struct {
 		name         string
 		topic        string
@@ -1994,9 +2002,10 @@ func TestDelegateHandlers_RoundTrip(t *testing.T) {
 			name:    "DeleteTags",
 			topic:   "ec2.test.DeleteTags",
 			handler: daemon.handleEC2DeleteTags,
-			input:   &ec2.DeleteTagsInput{Resources: []*string{aws.String("i-12345678")}},
-			// DeleteTags returns `{}` on success.
-			allowEmpty: true,
+			// Instance IDs route to the owning daemon; with no owner
+			// subscribed the mutation is rejected rather than written blind.
+			input:        &ec2.DeleteTagsInput{Resources: []*string{aws.String("i-12345678")}},
+			expectedCode: awserrors.ErrorInvalidInstanceIDNotFound,
 		},
 		{
 			name:    "DescribeTags",

spinifex/handlers/ec2/instance/instance_tags_test.go

@@ -0,0 +1,218 @@
+package handlers_ec2_instance
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	spxtypes "github.com/mulgadc/spinifex/spinifex/types"
+	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func tagList(pairs ...string) []*ec2.Tag {
+	var tags []*ec2.Tag
+	for i := 0; i+1 < len(pairs); i += 2 {
+		tags = append(tags, &ec2.Tag{Key: aws.String(pairs[i]), Value: aws.String(pairs[i+1])})
+	}
+	return tags
+}
+
+func tagsAsMap(tags []*ec2.Tag) map[string]string {
+	out := make(map[string]string, len(tags))
+	for _, t := range tags {
+		out[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
+	}
+	return out
+}
+
+func TestApplyInstanceTagMutation_MergeUpsert(t *testing.T) {
+	existing := tagList("Name", "web", "env", "dev")
+	out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod", "team": "infra"},
+	}, false)
+	assert.Equal(t, map[string]string{"Name": "web", "env": "prod", "team": "infra"}, tagsAsMap(out))
+}
+
+func TestApplyInstanceTagMutation_RemoveUnconditional(t *testing.T) {
+	existing := tagList("Name", "web", "env", "dev")
+	out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{
+		TagKeys: []string{"env", "missing"},
+	}, true)
+	assert.Equal(t, map[string]string{"Name": "web"}, tagsAsMap(out))
+}
+
+func TestApplyInstanceTagMutation_RemoveValueMatch(t *testing.T) {
+	existing := tagList("Name", "web", "env", "dev")
+	out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod", "Name": "web"},
+	}, true)
+	assert.Equal(t, map[string]string{"env": "dev"}, tagsAsMap(out))
+}
+
+func TestApplyInstanceTagMutation_RemoveClearAll(t *testing.T) {
+	existing := tagList("Name", "web", "env", "dev")
+	out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{}, true)
+	assert.Empty(t, out)
+}
+
+func TestApplyInstanceTagMutation_SortedAndNilSafe(t *testing.T) {
+	existing := []*ec2.Tag{nil, {Key: aws.String("b"), Value: aws.String("2")}, {Key: nil}}
+	out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"a": "1"},
+	}, false)
+	require.Len(t, out, 2)
+	assert.Equal(t, "a", aws.StringValue(out[0].Key))
+	assert.Equal(t, "b", aws.StringValue(out[1].Key))
+}
+
+type fakeTagWriter struct {
+	accountID   string
+	resourceID  string
+	tags        map[string]string
+	err         error
+	calls       int
+	deleteCalls int
+}
+
+func (f *fakeTagWriter) PutResourceTags(accountID, resourceID string, tags map[string]string) error {
+	f.calls++
+	f.accountID = accountID
+	f.resourceID = resourceID
+	f.tags = tags
+	return f.err
+}
+
+func (f *fakeTagWriter) DeleteAllTags(accountID, resourceID string) error {
+	f.deleteCalls++
+	f.accountID = accountID
+	f.resourceID = resourceID
+	return f.err
+}
+
+func TestWriteInstanceTags_WritesRecordAndCentral(t *testing.T) {
+	instance := &vm.VM{ID: "i-123", Instance: &ec2.Instance{Tags: tagList("Name", "web")}}
+	writer := &fakeTagWriter{}
+
+	err := WriteInstanceTags(instance, &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod"},
+	}, false, writer, "111122223333")
+	require.NoError(t, err)
+
+	want := map[string]string{"Name": "web", "env": "prod"}
+	assert.Equal(t, want, tagsAsMap(instance.Instance.Tags))
+	assert.Equal(t, 1, writer.calls)
+	assert.Equal(t, "111122223333", writer.accountID)
+	assert.Equal(t, "i-123", writer.resourceID)
+	assert.Equal(t, want, writer.tags)
+}
+
+func TestWriteInstanceTags_CentralWriteErrorPropagates(t *testing.T) {
+	instance := &vm.VM{ID: "i-123", Instance: &ec2.Instance{}}
+	writer := &fakeTagWriter{err: errors.New("s3 down")}
+
+	err := WriteInstanceTags(instance, &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod"},
+	}, false, writer, "111122223333")
+	assert.Error(t, err)
+}
+
+func TestWriteInstanceTags_NilRecordRejected(t *testing.T) {
+	writer := &fakeTagWriter{}
+	err := WriteInstanceTags(nil, nil, false, writer, "111122223333")
+	assert.Error(t, err)
+	err = WriteInstanceTags(&vm.VM{ID: "i-123"}, nil, false, writer, "111122223333")
+	assert.Error(t, err)
+	assert.Zero(t, writer.calls)
+}
+
+func stoppedTagInstance(id, accountID string, tags []*ec2.Tag) *vm.VM {
+	return &vm.VM{
+		ID:        id,
+		AccountID: accountID,
+		Status:    vm.StateStopped,
+		Instance:  &ec2.Instance{InstanceId: aws.String(id), Tags: tags},
+	}
+}
+
+func TestTagStoppedInstance_WritesRecordAndCentral(t *testing.T) {
+	stored := stoppedTagInstance("i-123", "111122223333", tagList("Name", "web"))
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{"i-123": stored}}
+	writer := &fakeTagWriter{}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	err := svc.TagStoppedInstance("i-123", &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod"},
+	}, false, writer, "111122223333")
+	require.NoError(t, err)
+
+	want := map[string]string{"Name": "web", "env": "prod"}
+	written := store.wroteStopped["i-123"]
+	require.NotNil(t, written)
+	assert.Equal(t, want, tagsAsMap(written.Instance.Tags))
+	assert.Equal(t, want, writer.tags)
+	assert.Equal(t, "i-123", writer.resourceID)
+	assert.Equal(t, "111122223333", writer.accountID)
+}
+
+func TestTagStoppedInstance_RemoveKeys(t *testing.T) {
+	stored := stoppedTagInstance("i-123", "111122223333", tagList("Name", "web", "env", "dev"))
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{"i-123": stored}}
+	writer := &fakeTagWriter{}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	err := svc.TagStoppedInstance("i-123", &spxtypes.InstanceTagsData{
+		TagKeys: []string{"env"},
+	}, true, writer, "111122223333")
+	require.NoError(t, err)
+
+	want := map[string]string{"Name": "web"}
+	assert.Equal(t, want, tagsAsMap(store.wroteStopped["i-123"].Instance.Tags))
+	assert.Equal(t, want, writer.tags)
+}
+
+func TestTagStoppedInstance_NotFound(t *testing.T) {
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{}}
+	writer := &fakeTagWriter{}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	err := svc.TagStoppedInstance("i-missing", &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod"},
+	}, false, writer, "111122223333")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.Zero(t, writer.calls)
+	assert.Empty(t, store.wroteStopped)
+}
+
+func TestTagStoppedInstance_CrossAccountRejected(t *testing.T) {
+	stored := stoppedTagInstance("i-123", "999988887777", tagList("Name", "web"))
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{"i-123": stored}}
+	writer := &fakeTagWriter{}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	err := svc.TagStoppedInstance("i-123", &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod"},
+	}, false, writer, "111122223333")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.Zero(t, writer.calls)
+	assert.Empty(t, store.wroteStopped)
+}
+
+func TestTagStoppedInstance_CentralWriteErrorSkipsRecordWrite(t *testing.T) {
+	stored := stoppedTagInstance("i-123", "111122223333", tagList("Name", "web"))
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{"i-123": stored}}
+	writer := &fakeTagWriter{err: errors.New("s3 down")}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	err := svc.TagStoppedInstance("i-123", &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod"},
+	}, false, writer, "111122223333")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+	assert.Empty(t, store.wroteStopped)
+}

spinifex/handlers/ec2/instance/service.go

@@ -86,6 +86,14 @@ type StoppedInstanceStore interface {
 	WriteTerminatedInstance(instanceID string, instance *vm.VM) error
 }
 
+// InstanceTagWriter projects an instance record's full tag set into the
+// central tag store, and removes it on terminate. Implemented by
+// handlers/ec2/tags.TagsServiceImpl.
+type InstanceTagWriter interface {
+	PutResourceTags(accountID, resourceID string, tags map[string]string) error
+	DeleteAllTags(accountID, resourceID string) error
+}
+
 // VolumeDeleter deletes EBS volumes. Implemented by handlers/ec2/volume's
 // VolumeServiceImpl (used by the daemon).
 type VolumeDeleter interface {

spinifex/handlers/ec2/instance/service_impl.go

@@ -6,7 +6,9 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
 	"os"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -127,6 +129,7 @@ type InstanceServiceImpl struct {
 	volumeDeleter VolumeDeleter
 	eniDeleter    ENIDeleter
 	ipReleaser    PublicIPReleaser
+	tagWriter     InstanceTagWriter
 	gpuClaimer    GPUClaimer
 	amiLoader     AMIMetaLoader
 	keyValidator  KeyPairValidator
@@ -155,13 +158,14 @@ func NewInstanceServiceImpl(
 	}
 }
 
-// SetTerminationDeps wires the dependencies required by TerminateStoppedInstance.
+// SetTerminationDeps wires the dependencies required by the terminate paths.
 // Kept separate from the main constructor so handlers needing only read or
 // modify paths can construct a service without dragging the VPC/volume stack in.
-func (s *InstanceServiceImpl) SetTerminationDeps(vd VolumeDeleter, ed ENIDeleter, pr PublicIPReleaser) {
+func (s *InstanceServiceImpl) SetTerminationDeps(vd VolumeDeleter, ed ENIDeleter, pr PublicIPReleaser, tw InstanceTagWriter) {
 	s.volumeDeleter = vd
 	s.eniDeleter = ed
 	s.ipReleaser = pr
+	s.tagWriter = tw
 }
 
 // SetGPUClaimer wires the GPU passthrough claim/release dependency used by
@@ -275,6 +279,112 @@ func instanceTagsFromSpec(specs []*ec2.TagSpecification) []*ec2.Tag {
 	return tags
 }
 
+// ApplyInstanceTagMutation applies a set or remove tag mutation to a tag list,
+// mirroring the central tag store's merge, value-match delete, and clear-all
+// semantics, and returns the resulting list sorted by key.
+func ApplyInstanceTagMutation(existing []*ec2.Tag, data *spxtypes.InstanceTagsData, remove bool) []*ec2.Tag {
+	tags := make(map[string]string, len(existing))
+	for _, t := range existing {
+		if t == nil || t.Key == nil {
+			continue
+		}
+		tags[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
+	}
+
+	switch {
+	case remove && data == nil:
+		utils.ApplyTagRemovals(tags, nil, nil)
+	case remove:
+		utils.ApplyTagRemovals(tags, data.TagKeys, data.Tags)
+	case data != nil:
+		maps.Copy(tags, data.Tags)
+	}
+
+	out := make([]*ec2.Tag, 0, len(tags))
+	for key, value := range tags {
+		out = append(out, &ec2.Tag{Key: aws.String(key), Value: aws.String(value)})
+	}
+	sort.Slice(out, func(i, j int) bool { return aws.StringValue(out[i].Key) < aws.StringValue(out[j].Key) })
+	return out
+}
+
+// TagsToMap converts a record tag list to the central tag store's map form,
+// skipping nil entries and nil keys.
+func TagsToMap(tags []*ec2.Tag) map[string]string {
+	out := make(map[string]string, len(tags))
+	for _, t := range tags {
+		if t == nil || t.Key == nil {
+			continue
+		}
+		out[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
+	}
+	return out
+}
+
+// WriteInstanceTags applies the tag mutation to the instance record (the source
+// of truth) and writes the resulting full tag set to the central tag store, so
+// both stores move together. Callers own record persistence and must not hold
+// the vm.Manager lock, since the central write is an S3 round-trip.
+func WriteInstanceTags(instance *vm.VM, data *spxtypes.InstanceTagsData, remove bool, central InstanceTagWriter, accountID string) error {
+	if instance == nil || instance.Instance == nil || central == nil {
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+	instance.Instance.Tags = ApplyInstanceTagMutation(instance.Instance.Tags, data, remove)
+	return central.PutResourceTags(accountID, instance.ID, TagsToMap(instance.Instance.Tags))
+}
+
+// TagStoppedInstance applies a create-tags/delete-tags mutation to a stopped
+// instance's shared KV record and the central tag store together. A missing or
+// cross-account record returns InvalidID.NotFound and writes nothing.
+func (s *InstanceServiceImpl) TagStoppedInstance(instanceID string, data *spxtypes.InstanceTagsData, remove bool, central InstanceTagWriter, accountID string) error {
+	if s.stoppedStore == nil {
+		slog.Error("TagStoppedInstance: stopped store not available")
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	instance, err := s.stoppedStore.LoadStoppedInstance(instanceID)
+	if err != nil {
+		slog.Error("TagStoppedInstance: failed to load stopped instance", "instanceId", instanceID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+	if instance == nil {
+		return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+	}
+	if !IsInstanceVisible(accountID, instance.AccountID) {
+		slog.Warn("TagStoppedInstance: instance not visible",
+			"instanceId", instanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID)
+		return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+	}
+	if instance.Instance == nil {
+		slog.Error("TagStoppedInstance: instance.Instance is nil, data integrity issue", "instanceId", instanceID)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	if err := WriteInstanceTags(instance, data, remove, central, accountID); err != nil {
+		slog.Error("TagStoppedInstance: central tag store write failed", "instanceId", instanceID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+	if err := s.stoppedStore.WriteStoppedInstance(instanceID, instance); err != nil {
+		slog.Error("TagStoppedInstance: failed to write stopped instance", "instanceId", instanceID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+	return nil
+}
+
+// deleteCentralInstanceTags removes a terminated instance's central tag store
+// entry so describe-tags stops reporting it, while the terminated record keeps
+// its tags until TTL. Best-effort: the terminate already succeeded, so a
+// failed delete is logged rather than surfaced.
+func (s *InstanceServiceImpl) deleteCentralInstanceTags(instanceID, ownerAccountID string) {
+	if s.tagWriter == nil || ownerAccountID == "" {
+		return
+	}
+	if err := s.tagWriter.DeleteAllTags(ownerAccountID, instanceID); err != nil {
+		slog.Error("deleteCentralInstanceTags: central tag store delete failed",
+			"instanceId", instanceID, "err", err)
+	}
+}
+
 // PrepareRunInstances validates input, allocates capacity, creates VM metadata,
 // auto-creates the primary ENI, and auto-assigns a public IP when needed.
 // Does NOT touch vmMgr or NATS — callers insert VMs then call LaunchRunInstances.
@@ -927,7 +1037,7 @@ func (s *InstanceServiceImpl) StopOrTerminateInstance(instance *vm.VM, command s
 		return errors.New(awserrors.ErrorIncorrectInstanceState)
 	}
 
-	go func(id string) {
+	go func(id, ownerAccountID string) {
 		var err error
 		if isTerminate {
 			err = s.vmMgr.Terminate(id)
@@ -941,8 +1051,12 @@ func (s *InstanceServiceImpl) StopOrTerminateInstance(instance *vm.VM, command s
 				return
 			}
 			slog.Error("StopOrTerminateInstance: failed to "+strings.ToLower(action), "err", err, "id", id)
+			return
 		}
-	}(instance.ID)
+		if isTerminate {
+			s.deleteCentralInstanceTags(id, ownerAccountID)
+		}
+	}(instance.ID, instance.AccountID)
 
 	return nil
 }
@@ -2175,6 +2289,8 @@ func (s *InstanceServiceImpl) TerminateStoppedInstance(input *TerminateStoppedIn
 		}
 	}
 
+	s.deleteCentralInstanceTags(input.InstanceID, instance.AccountID)
+
 	slog.Info("Terminated stopped instance from shared KV", "instanceId", input.InstanceID)
 	return &TerminateStoppedInstanceOutput{Status: "terminated", InstanceID: input.InstanceID}, nil
 }

spinifex/handlers/ec2/instance/service_impl_test.go

@@ -1617,6 +1617,52 @@ func TestTerminateStoppedInstance_HappyPath(t *testing.T) {
 	assert.Contains(t, store.deletedStopped, id)
 }
 
+func TestTerminateStoppedInstance_CentralTagsDeleted(t *testing.T) {
+	id := "i-term-tags"
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
+		id: {ID: id, Status: vm.StateStopped, AccountID: "acc"},
+	}}
+	tw := &fakeTagWriter{}
+	svc := &InstanceServiceImpl{stoppedStore: store, tagWriter: tw}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+
+	assert.Equal(t, 1, tw.deleteCalls)
+	assert.Equal(t, "acc", tw.accountID, "central delete must target the owner's namespace")
+	assert.Equal(t, id, tw.resourceID)
+}
+
+func TestTerminateStoppedInstance_CentralTagDeleteFailureIsBestEffort(t *testing.T) {
+	id := "i-term-tagerr"
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
+		id: {ID: id, Status: vm.StateStopped, AccountID: "acc"},
+	}}
+	tw := &fakeTagWriter{err: errors.New("s3 down")}
+	svc := &InstanceServiceImpl{stoppedStore: store, tagWriter: tw}
+
+	out, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err, "terminate already succeeded; central delete failure must not surface")
+	assert.Equal(t, "terminated", out.Status)
+	assert.Contains(t, store.deletedStopped, id)
+}
+
+func TestTerminateStoppedInstance_ProtectedSkipsCentralTagDelete(t *testing.T) {
+	id := "i-prot-tags"
+	tw := &fakeTagWriter{}
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
+		id: {
+			ID: id, Status: vm.StateStopped, AccountID: "acc",
+			RunInstancesInput: &ec2.RunInstancesInput{DisableApiTermination: aws.Bool(true)},
+		},
+	}}
+	svc := &InstanceServiceImpl{stoppedStore: store, tagWriter: tw}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.Error(t, err)
+	assert.Zero(t, tw.deleteCalls, "central tags must survive a rejected terminate")
+}
+
 func TestTerminateStoppedInstance_WriteTerminatedError_Aborts(t *testing.T) {
 	id := "i-werr"
 	store := &fakeStoppedStore{

spinifex/handlers/ec2/tags/service_impl.go

@@ -14,12 +14,17 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/config"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
+	handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
+	"github.com/mulgadc/spinifex/spinifex/utils"
 )
 
 // Ensure TagsServiceImpl implements TagsService
 var _ TagsService = (*TagsServiceImpl)(nil)
 
+// Ensure TagsServiceImpl can project instance record tags into the store
+var _ handlers_ec2_instance.InstanceTagWriter = (*TagsServiceImpl)(nil)
+
 // TagsServiceImpl implements TagsService with S3-backed storage.
 // Tags are stored per-account in S3 (tags/{accountID}/{resourceID}.json),
 // so account scoping is enforced at the storage layer.
@@ -155,6 +160,33 @@ func (s *TagsServiceImpl) putResourceTags(accountID, resourceID string, tags map
 	return err
 }
 
+// PutResourceTags overwrites the stored tag set for a resource. Used to
+// project an instance record's tags (the source of truth) into the central
+// store so describe-tags agrees with describe-instances.
+func (s *TagsServiceImpl) PutResourceTags(accountID, resourceID string, tags map[string]string) error {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	return s.putResourceTags(accountID, resourceID, tags)
+}
+
+// DeleteAllTags removes the stored tag object for a resource. Used on
+// instance terminate so describe-tags stops reporting the instance while the
+// terminated record keeps its tags until TTL. Idempotent: a missing object
+// is not an error.
+func (s *TagsServiceImpl) DeleteAllTags(accountID, resourceID string) error {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	_, err := s.store.DeleteObject(&s3.DeleteObjectInput{
+		Bucket: aws.String(s.config.Predastore.Bucket),
+		Key:    aws.String(getTagsKey(accountID, resourceID)),
+	})
+	if err != nil && !objectstore.IsNoSuchKeyError(err) {
+		return err
+	}
+	return nil
+}
+
 // CreateTags adds or overwrites tags for the specified resources
 func (s *TagsServiceImpl) CreateTags(input *ec2.CreateTagsInput, accountID string) (*ec2.CreateTagsOutput, error) {
 	if input == nil {
@@ -317,27 +349,7 @@ func (s *TagsServiceImpl) DeleteTags(input *ec2.DeleteTagsInput, accountID strin
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
 
-		if len(input.Tags) == 0 {
-			// Delete all tags if no specific tags provided
-			existingTags = make(map[string]string)
-		} else {
-			// Delete specified tags — per AWS API, when Value is specified
-			// the tag is only deleted if the stored value matches
-			for _, tag := range input.Tags {
-				if tag.Key == nil {
-					continue
-				}
-				if tag.Value == nil {
-					// No value specified: delete unconditionally
-					delete(existingTags, *tag.Key)
-				} else {
-					// Value specified: only delete if current value matches
-					if current, exists := existingTags[*tag.Key]; exists && current == *tag.Value {
-						delete(existingTags, *tag.Key)
-					}
-				}
-			}
-		}
+		utils.RemoveTagsMut(input)(existingTags)
 
 		// Save updated tags
 		if err := s.putResourceTags(accountID, *resourceID, existingTags); err != nil {

spinifex/handlers/ec2/tags/service_impl_test.go

@@ -578,3 +578,36 @@ func TestAccountIsolation_DeleteDoesNotAffectOtherAccount(t *testing.T) {
 	assert.Len(t, resultB.Tags, 1)
 	assert.Equal(t, "staging", *resultB.Tags[0].Value)
 }
+
+// TestDeleteAllTags tests that the resource's stored tag object is removed
+// entirely and other accounts' entries for the same resource ID survive
+func TestDeleteAllTags(t *testing.T) {
+	svc, _ := setupTestTagsService(t)
+	accountA := "111111111111"
+	accountB := "222222222222"
+
+	for _, acct := range []string{accountA, accountB} {
+		_, err := svc.CreateTags(&ec2.CreateTagsInput{
+			Resources: []*string{aws.String("i-gone")},
+			Tags:      []*ec2.Tag{{Key: aws.String("Env"), Value: aws.String("prod")}},
+		}, acct)
+		require.NoError(t, err)
+	}
+
+	require.NoError(t, svc.DeleteAllTags(accountA, "i-gone"))
+
+	resultA, err := svc.DescribeTags(&ec2.DescribeTagsInput{}, accountA)
+	require.NoError(t, err)
+	assert.Empty(t, resultA.Tags)
+
+	resultB, err := svc.DescribeTags(&ec2.DescribeTagsInput{}, accountB)
+	require.NoError(t, err)
+	assert.Len(t, resultB.Tags, 1)
+}
+
+// TestDeleteAllTags_MissingResourceIsNoError tests idempotency for a resource
+// that never had tags
+func TestDeleteAllTags_MissingResourceIsNoError(t *testing.T) {
+	svc, _ := setupTestTagsService(t)
+	require.NoError(t, svc.DeleteAllTags("111111111111", "i-never-tagged"))
+}

spinifex/types/ec2.go

@@ -11,6 +11,7 @@ type EC2InstanceCommand struct {
 	DetachENIData             *DetachENIData             `json:"detach_eni_data,omitempty"`
 	IamProfileAssociationData *IamProfileAssociationData `json:"iam_profile_association_data,omitempty"`
 	SpotLineageData           *SpotLineageData           `json:"spot_lineage_data,omitempty"`
+	InstanceTagsData          *InstanceTagsData          `json:"instance_tags_data,omitempty"`
 }
 
 // EC2CommandAttributes indicates which action the daemon should perform.
@@ -25,6 +26,8 @@ type EC2CommandAttributes struct {
 	DetachENI                   bool `json:"detach_eni"`
 	AssociateIamInstanceProfile bool `json:"associate_iam_instance_profile,omitempty"`
 	SetSpotLineage              bool `json:"set_spot_lineage,omitempty"`
+	SetInstanceTags             bool `json:"set_instance_tags,omitempty"`
+	RemoveInstanceTags          bool `json:"remove_instance_tags,omitempty"`
 }
 
 // AttachVolumeData carries parameters for an attach-volume command.
@@ -60,6 +63,14 @@ type IamProfileAssociationData struct {
 	InstanceProfileArn string `json:"instance_profile_arn"`
 }
 
+// InstanceTagsData carries parameters for a set/remove-instance-tags command.
+// For create-tags, Tags is the upsert set. For delete-tags, TagKeys removes
+// keys unconditionally, Tags removes only on value match, both empty clears all.
+type InstanceTagsData struct {
+	Tags    map[string]string `json:"tags,omitempty"`
+	TagKeys []string          `json:"tag_keys,omitempty"`
+}
+
 // SpotLineageData carries the SIR id stamped onto a spot-launched instance.
 // Lifecycle is always "spot" for this command, so only the SIR id travels.
 type SpotLineageData struct {

spinifex/utils/tagmirror.go

@@ -23,27 +23,53 @@ func MergeTagsMut(input *ec2.CreateTagsInput) func(map[string]string) {
 	}
 }
 
+// ApplyTagRemovals applies AWS-faithful DeleteTags semantics to a tag map:
+// keys delete unconditionally, valueMatch entries delete only when the stored
+// value matches, and both empty clears every tag.
+func ApplyTagRemovals(tags map[string]string, keys []string, valueMatch map[string]string) {
+	if len(keys) == 0 && len(valueMatch) == 0 {
+		clear(tags)
+		return
+	}
+	for _, k := range keys {
+		delete(tags, k)
+	}
+	for k, v := range valueMatch {
+		if cur, ok := tags[k]; ok && cur == v {
+			delete(tags, k)
+		}
+	}
+}
+
 // RemoveTagsMut returns a mutator with AWS-faithful DeleteTags semantics:
 // empty input.Tags clears all tags; a tag with a value deletes only on a
 // value match, a nil value deletes unconditionally.
 func RemoveTagsMut(input *ec2.DeleteTagsInput) func(map[string]string) {
 	return func(tags map[string]string) {
 		if len(input.Tags) == 0 {
-			for k := range tags {
-				delete(tags, k)
-			}
+			ApplyTagRemovals(tags, nil, nil)
 			return
 		}
+		var keys []string
+		var valueMatch map[string]string
 		for _, t := range input.Tags {
 			if t.Key == nil {
 				continue
 			}
 			if t.Value == nil {
-				delete(tags, *t.Key)
-			} else if cur, ok := tags[*t.Key]; ok && cur == *t.Value {
-				delete(tags, *t.Key)
+				keys = append(keys, *t.Key)
+			} else {
+				if valueMatch == nil {
+					valueMatch = make(map[string]string)
+				}
+				valueMatch[*t.Key] = *t.Value
 			}
 		}
+		// All entries had nil keys: no removals requested, not a clear-all.
+		if len(keys) == 0 && len(valueMatch) == 0 {
+			return
+		}
+		ApplyTagRemovals(tags, keys, valueMatch)
 	}
 }