feat(ecs): tags API + capacity-provider CRUD scaffolding (#499)

* feat(ecs): tags API (TagResource/UntagResource/ListTagsForResource)

* feat(ecs): capacity-provider CRUD scaffolding

spinifex/daemon/daemon.go

@@ -1014,6 +1014,13 @@ func (d *Daemon) subscribeAll() error {
 			natsSub{"ecs.SubmitTaskStateChange", d.handleECSSubmitTaskStateChange, "spinifex-workers"},
 			natsSub{"ecs.PollAssignments", d.handleECSPollAssignments, "spinifex-workers"},
 			natsSub{"ecs.ProvisionCapacity", d.handleECSProvisionCapacity, "spinifex-workers"},
+			natsSub{"ecs.TagResource", d.handleECSTagResource, "spinifex-workers"},
+			natsSub{"ecs.UntagResource", d.handleECSUntagResource, "spinifex-workers"},
+			natsSub{"ecs.ListTagsForResource", d.handleECSListTagsForResource, "spinifex-workers"},
+			natsSub{"ecs.PutClusterCapacityProviders", d.handleECSPutClusterCapacityProviders, "spinifex-workers"},
+			natsSub{"ecs.CreateCapacityProvider", d.handleECSCreateCapacityProvider, "spinifex-workers"},
+			natsSub{"ecs.DescribeCapacityProviders", d.handleECSDescribeCapacityProviders, "spinifex-workers"},
+			natsSub{"ecs.DeleteCapacityProvider", d.handleECSDeleteCapacityProvider, "spinifex-workers"},
 		)
 	}
 

spinifex/daemon/daemon_handlers_ecs.go

@@ -117,3 +117,35 @@ func (d *Daemon) handleECSPollAssignments(msg *nats.Msg) {
 func (d *Daemon) handleECSProvisionCapacity(msg *nats.Msg) {
 	handleNATSRequest(msg, d.ecsService.ProvisionCapacity)
 }
+
+// --- Tags ---
+
+func (d *Daemon) handleECSTagResource(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.TagResource)
+}
+
+func (d *Daemon) handleECSUntagResource(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.UntagResource)
+}
+
+func (d *Daemon) handleECSListTagsForResource(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.ListTagsForResource)
+}
+
+// --- Capacity providers ---
+
+func (d *Daemon) handleECSPutClusterCapacityProviders(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.PutClusterCapacityProviders)
+}
+
+func (d *Daemon) handleECSCreateCapacityProvider(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.CreateCapacityProvider)
+}
+
+func (d *Daemon) handleECSDescribeCapacityProviders(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.DescribeCapacityProviders)
+}
+
+func (d *Daemon) handleECSDeleteCapacityProvider(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.DeleteCapacityProvider)
+}

spinifex/gateway/ecs/actions.go

@@ -243,3 +243,63 @@ func PollAssignments(ctx context.Context, nc *nats.Conn, accountID string, body
 	}
 	return handlers_ecs.NewNATSECSService(nc).PollAssignments(ctx, input, accountID)
 }
+
+// --- Tags ---
+
+func TagResource(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.TagResourceInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).TagResource(ctx, input, accountID)
+}
+
+func UntagResource(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.UntagResourceInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).UntagResource(ctx, input, accountID)
+}
+
+func ListTagsForResource(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.ListTagsForResourceInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).ListTagsForResource(ctx, input, accountID)
+}
+
+// --- Capacity providers ---
+
+func PutClusterCapacityProviders(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.PutClusterCapacityProvidersInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).PutClusterCapacityProviders(ctx, input, accountID)
+}
+
+func CreateCapacityProvider(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.CreateCapacityProviderInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).CreateCapacityProvider(ctx, input, accountID)
+}
+
+func DescribeCapacityProviders(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.DescribeCapacityProvidersInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).DescribeCapacityProviders(ctx, input, accountID)
+}
+
+func DeleteCapacityProvider(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.DeleteCapacityProviderInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).DeleteCapacityProvider(ctx, input, accountID)
+}

spinifex/gateway/ecs/handler.go

@@ -42,13 +42,19 @@ func NotImplemented(_ context.Context, _ *nats.Conn, _ string, _ []byte) (any, e
 // Wired actions point at their handler; unimplemented actions keep the
 // NotImplemented value. The key set is the v1 API contract.
 var Actions = map[string]Handler{
-	// Cluster. PutClusterCapacityProviders is a no-op stub in v1 (Q15).
+	// Cluster.
 	"CreateCluster":               CreateCluster,
 	"DescribeClusters":            DescribeClusters,
 	"ListClusters":                ListClusters,
 	"DeleteCluster":               DeleteCluster,
 	"UpdateCluster":               NotImplemented,
-	"PutClusterCapacityProviders": NotImplemented,
+	"PutClusterCapacityProviders": PutClusterCapacityProviders,
+
+	// Capacity providers. Strategy is accepted + persisted but inert in v1
+	// (no scheduler coupling, no scale loop).
+	"CreateCapacityProvider":    CreateCapacityProvider,
+	"DescribeCapacityProviders": DescribeCapacityProviders,
+	"DeleteCapacityProvider":    DeleteCapacityProvider,
 
 	// Task definition.
 	"RegisterTaskDefinition":     RegisterTaskDefinition,
@@ -91,9 +97,9 @@ var Actions = map[string]Handler{
 	"ListAccountSettings": NotImplemented,
 
 	// Tags.
-	"TagResource":         NotImplemented,
-	"UntagResource":       NotImplemented,
-	"ListTagsForResource": NotImplemented,
+	"TagResource":         TagResource,
+	"UntagResource":       UntagResource,
+	"ListTagsForResource": ListTagsForResource,
 }
 
 // RawJSONActions encode their response with encoding/json instead of jsonutil.

spinifex/gateway/ecs/handler_test.go

@@ -33,6 +33,9 @@ var wiredActions = map[string]bool{
 	"DescribeServices": true, "ListServices": true,
 	"SubmitTaskStateChange": true,
 	"PollAssignments":       true,
+	"TagResource":           true, "UntagResource": true, "ListTagsForResource": true,
+	"PutClusterCapacityProviders": true,
+	"CreateCapacityProvider":      true, "DescribeCapacityProviders": true, "DeleteCapacityProvider": true,
 }
 
 func TestActions_StubsReturnNotImplemented(t *testing.T) {

spinifex/handlers/ecs/capacity_providers.go

@@ -0,0 +1,233 @@
+package handlers_ecs
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+)
+
+// PutClusterCapacityProviders persists the cluster's capacity-provider names
+// and default strategy. Accepted and stored only: no scheduler coupling, no
+// scale loop (a separate follow-on binds this to an ASG primitive).
+func (s *Service) PutClusterCapacityProviders(_ context.Context, input *ecs.PutClusterCapacityProvidersInput, accountID string) (*ecs.PutClusterCapacityProvidersOutput, error) {
+	if input == nil {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	var rec ClusterRecord
+	found, err := getJSON(kv, ClusterMetaKey(cluster), &rec)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		return nil, errors.New(awserrors.ErrorECSClusterNotFound)
+	}
+
+	rec.CapacityProviders = awsStringSlice(input.CapacityProviders)
+	rec.DefaultCapacityProviderStrategy = strategyFromAWS(input.DefaultCapacityProviderStrategy)
+	if err := putJSON(kv, ClusterMetaKey(cluster), &rec); err != nil {
+		return nil, err
+	}
+	return &ecs.PutClusterCapacityProvidersOutput{Cluster: rec.toAWS()}, nil
+}
+
+// CreateCapacityProvider persists a new capacity provider. Idempotent:
+// re-creating an existing name returns the existing record.
+func (s *Service) CreateCapacityProvider(_ context.Context, input *ecs.CreateCapacityProviderInput, accountID string) (*ecs.CreateCapacityProviderOutput, error) {
+	if input == nil || aws.StringValue(input.Name) == "" || input.AutoScalingGroupProvider == nil ||
+		aws.StringValue(input.AutoScalingGroupProvider.AutoScalingGroupArn) == "" {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	name := aws.StringValue(input.Name)
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	var existing CapacityProviderRecord
+	found, err := getJSON(kv, CapacityProviderKey(name), &existing)
+	if err != nil {
+		return nil, err
+	}
+	if found {
+		return &ecs.CreateCapacityProviderOutput{CapacityProvider: existing.toAWS()}, nil
+	}
+
+	rec := CapacityProviderRecord{
+		Name:                     name,
+		ARN:                      CapacityProviderARN(s.region, accountID, name),
+		Status:                   CapacityProviderStatusActive,
+		AutoScalingGroupProvider: autoScalingGroupProviderFromAWS(input.AutoScalingGroupProvider),
+		Tags:                     tagsToMap(input.Tags),
+		CreatedAt:                time.Now().UTC(),
+	}
+	if err := putJSON(kv, CapacityProviderKey(name), &rec); err != nil {
+		return nil, err
+	}
+	return &ecs.CreateCapacityProviderOutput{CapacityProvider: rec.toAWS()}, nil
+}
+
+// DescribeCapacityProviders returns the named providers, or every provider in
+// the account when none are named. Unknown names surface as failures.
+func (s *Service) DescribeCapacityProviders(_ context.Context, input *ecs.DescribeCapacityProvidersInput, accountID string) (*ecs.DescribeCapacityProvidersOutput, error) {
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.DescribeCapacityProvidersOutput{}
+
+	names := awsStringSlice(input.CapacityProviders)
+	if len(names) == 0 {
+		keys, kerr := keysWithPrefix(kv, CapacityProvidersPrefix())
+		if kerr != nil {
+			return nil, kerr
+		}
+		for _, k := range keys {
+			var rec CapacityProviderRecord
+			ok, gerr := getJSON(kv, k, &rec)
+			if gerr != nil {
+				return nil, gerr
+			}
+			if ok {
+				out.CapacityProviders = append(out.CapacityProviders, rec.toAWS())
+			}
+		}
+		return out, nil
+	}
+
+	for _, ref := range names {
+		name := capacityProviderShortName(ref)
+		var rec CapacityProviderRecord
+		ok, gerr := getJSON(kv, CapacityProviderKey(name), &rec)
+		if gerr != nil {
+			return nil, gerr
+		}
+		if ok {
+			out.CapacityProviders = append(out.CapacityProviders, rec.toAWS())
+		} else {
+			out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")})
+		}
+	}
+	return out, nil
+}
+
+// DeleteCapacityProvider removes a capacity provider record. Clusters that
+// still reference the name by CapacityProviders are left untouched (v1 does
+// not enforce referential integrity here).
+func (s *Service) DeleteCapacityProvider(_ context.Context, input *ecs.DeleteCapacityProviderInput, accountID string) (*ecs.DeleteCapacityProviderOutput, error) {
+	if input == nil || aws.StringValue(input.CapacityProvider) == "" {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	name := capacityProviderShortName(aws.StringValue(input.CapacityProvider))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	var rec CapacityProviderRecord
+	found, err := getJSON(kv, CapacityProviderKey(name), &rec)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	if err := kv.Delete(CapacityProviderKey(name)); err != nil {
+		return nil, err
+	}
+	return &ecs.DeleteCapacityProviderOutput{CapacityProvider: rec.toAWS()}, nil
+}
+
+// --- Conversions ---
+
+func (r *CapacityProviderRecord) toAWS() *ecs.CapacityProvider {
+	return &ecs.CapacityProvider{
+		Name:                     aws.String(r.Name),
+		CapacityProviderArn:      aws.String(r.ARN),
+		Status:                   aws.String(r.Status),
+		AutoScalingGroupProvider: r.AutoScalingGroupProvider.toAWS(),
+		Tags:                     tagsToAWS(r.Tags),
+	}
+}
+
+func autoScalingGroupProviderFromAWS(in *ecs.AutoScalingGroupProvider) AutoScalingGroupProviderRecord {
+	if in == nil {
+		return AutoScalingGroupProviderRecord{}
+	}
+	rec := AutoScalingGroupProviderRecord{
+		AutoScalingGroupARN:          aws.StringValue(in.AutoScalingGroupArn),
+		ManagedTerminationProtection: aws.StringValue(in.ManagedTerminationProtection),
+	}
+	if in.ManagedScaling != nil {
+		rec.ManagedScaling = ManagedScalingRecord{
+			Status:                 aws.StringValue(in.ManagedScaling.Status),
+			TargetCapacity:         int(aws.Int64Value(in.ManagedScaling.TargetCapacity)),
+			MinimumScalingStepSize: int(aws.Int64Value(in.ManagedScaling.MinimumScalingStepSize)),
+			MaximumScalingStepSize: int(aws.Int64Value(in.ManagedScaling.MaximumScalingStepSize)),
+			InstanceWarmupPeriod:   int(aws.Int64Value(in.ManagedScaling.InstanceWarmupPeriod)),
+		}
+	}
+	return rec
+}
+
+func (r AutoScalingGroupProviderRecord) toAWS() *ecs.AutoScalingGroupProvider {
+	out := &ecs.AutoScalingGroupProvider{AutoScalingGroupArn: aws.String(r.AutoScalingGroupARN)}
+	if r.ManagedTerminationProtection != "" {
+		out.ManagedTerminationProtection = aws.String(r.ManagedTerminationProtection)
+	}
+	if (r.ManagedScaling != ManagedScalingRecord{}) {
+		ms := &ecs.ManagedScaling{}
+		if r.ManagedScaling.Status != "" {
+			ms.Status = aws.String(r.ManagedScaling.Status)
+		}
+		if r.ManagedScaling.TargetCapacity > 0 {
+			ms.TargetCapacity = aws.Int64(int64(r.ManagedScaling.TargetCapacity))
+		}
+		if r.ManagedScaling.MinimumScalingStepSize > 0 {
+			ms.MinimumScalingStepSize = aws.Int64(int64(r.ManagedScaling.MinimumScalingStepSize))
+		}
+		if r.ManagedScaling.MaximumScalingStepSize > 0 {
+			ms.MaximumScalingStepSize = aws.Int64(int64(r.ManagedScaling.MaximumScalingStepSize))
+		}
+		if r.ManagedScaling.InstanceWarmupPeriod > 0 {
+			ms.InstanceWarmupPeriod = aws.Int64(int64(r.ManagedScaling.InstanceWarmupPeriod))
+		}
+		out.ManagedScaling = ms
+	}
+	return out
+}
+
+// strategyFromAWS maps the SDK capacity-provider strategy list to the
+// persisted subset, dropping nil entries.
+func strategyFromAWS(in []*ecs.CapacityProviderStrategyItem) []CapacityProviderStrategyItem {
+	out := make([]CapacityProviderStrategyItem, 0, len(in))
+	for _, item := range in {
+		if item == nil {
+			continue
+		}
+		out = append(out, CapacityProviderStrategyItem{
+			Provider: aws.StringValue(item.CapacityProvider),
+			Weight:   int(aws.Int64Value(item.Weight)),
+			Base:     int(aws.Int64Value(item.Base)),
+		})
+	}
+	return out
+}
+
+func (item CapacityProviderStrategyItem) toAWS() *ecs.CapacityProviderStrategyItem {
+	out := &ecs.CapacityProviderStrategyItem{CapacityProvider: aws.String(item.Provider)}
+	if item.Weight > 0 {
+		out.Weight = aws.Int64(int64(item.Weight))
+	}
+	if item.Base > 0 {
+		out.Base = aws.Int64(int64(item.Base))
+	}
+	return out
+}

spinifex/handlers/ecs/capacity_providers_test.go

@@ -0,0 +1,146 @@
+package handlers_ecs
+
+import (
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestService_CreateCapacityProvider_Idempotent(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	in := &ecs.CreateCapacityProviderInput{
+		Name: aws.String("cp-1"),
+		AutoScalingGroupProvider: &ecs.AutoScalingGroupProvider{
+			AutoScalingGroupArn: aws.String("arn:aws:autoscaling:r:a:autoScalingGroup:x:autoScalingGroupName/asg-1"),
+			ManagedScaling: &ecs.ManagedScaling{
+				Status:                 aws.String(ecs.ManagedScalingStatusEnabled),
+				TargetCapacity:         aws.Int64(90),
+				MinimumScalingStepSize: aws.Int64(1),
+				MaximumScalingStepSize: aws.Int64(10),
+			},
+		},
+		Tags: []*ecs.Tag{{Key: aws.String("team"), Value: aws.String("infra")}},
+	}
+	out, err := svc.CreateCapacityProvider(ctx, in, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, out.CapacityProvider)
+	assert.Equal(t, "cp-1", aws.StringValue(out.CapacityProvider.Name))
+	assert.Equal(t, CapacityProviderStatusActive, aws.StringValue(out.CapacityProvider.Status))
+	assert.Equal(t, CapacityProviderARN(testRegion, testAccountID, "cp-1"), aws.StringValue(out.CapacityProvider.CapacityProviderArn))
+	require.NotNil(t, out.CapacityProvider.AutoScalingGroupProvider)
+	require.NotNil(t, out.CapacityProvider.AutoScalingGroupProvider.ManagedScaling)
+	assert.Equal(t, int64(90), aws.Int64Value(out.CapacityProvider.AutoScalingGroupProvider.ManagedScaling.TargetCapacity))
+	require.Len(t, out.CapacityProvider.Tags, 1)
+
+	// Re-creating the same name returns the existing record, not a duplicate.
+	again, err := svc.CreateCapacityProvider(ctx, in, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, aws.StringValue(out.CapacityProvider.CapacityProviderArn), aws.StringValue(again.CapacityProvider.CapacityProviderArn))
+
+	list, err := svc.DescribeCapacityProviders(ctx, &ecs.DescribeCapacityProvidersInput{}, testAccountID)
+	require.NoError(t, err)
+	assert.Len(t, list.CapacityProviders, 1)
+}
+
+func TestService_CreateCapacityProvider_InvalidParameter(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	_, err := svc.CreateCapacityProvider(ctx, &ecs.CreateCapacityProviderInput{}, testAccountID)
+	require.Error(t, err)
+
+	_, err = svc.CreateCapacityProvider(ctx, &ecs.CreateCapacityProviderInput{Name: aws.String("cp")}, testAccountID)
+	require.Error(t, err)
+}
+
+func TestService_DescribeCapacityProviders_NamedAndMissing(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	_, err := svc.CreateCapacityProvider(ctx, &ecs.CreateCapacityProviderInput{
+		Name: aws.String("cp-1"),
+		AutoScalingGroupProvider: &ecs.AutoScalingGroupProvider{
+			AutoScalingGroupArn: aws.String("arn:aws:autoscaling:r:a:autoScalingGroup:x:autoScalingGroupName/asg-1"),
+		},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	out, err := svc.DescribeCapacityProviders(ctx, &ecs.DescribeCapacityProvidersInput{
+		CapacityProviders: []*string{aws.String("cp-1"), aws.String("cp-missing")},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, out.CapacityProviders, 1)
+	require.Len(t, out.Failures, 1)
+	assert.Equal(t, "cp-missing", aws.StringValue(out.Failures[0].Arn))
+}
+
+func TestService_DeleteCapacityProvider(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	_, err := svc.CreateCapacityProvider(ctx, &ecs.CreateCapacityProviderInput{
+		Name: aws.String("cp-1"),
+		AutoScalingGroupProvider: &ecs.AutoScalingGroupProvider{
+			AutoScalingGroupArn: aws.String("arn:aws:autoscaling:r:a:autoScalingGroup:x:autoScalingGroupName/asg-1"),
+		},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	out, err := svc.DeleteCapacityProvider(ctx, &ecs.DeleteCapacityProviderInput{CapacityProvider: aws.String("cp-1")}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "cp-1", aws.StringValue(out.CapacityProvider.Name))
+
+	list, err := svc.DescribeCapacityProviders(ctx, &ecs.DescribeCapacityProvidersInput{}, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, list.CapacityProviders)
+
+	_, err = svc.DeleteCapacityProvider(ctx, &ecs.DeleteCapacityProviderInput{CapacityProvider: aws.String("cp-1")}, testAccountID)
+	require.Error(t, err)
+}
+
+// TestService_PutClusterCapacityProviders_PersistsStrategy asserts the
+// strategy round-trips onto the cluster record and is returned by
+// DescribeClusters, without any scheduler/placement coupling (inert in v1).
+func TestService_PutClusterCapacityProviders_PersistsStrategy(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	_, err := svc.CreateCluster(ctx, &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+
+	out, err := svc.PutClusterCapacityProviders(ctx, &ecs.PutClusterCapacityProvidersInput{
+		Cluster:           aws.String("web"),
+		CapacityProviders: []*string{aws.String("cp-1"), aws.String("cp-2")},
+		DefaultCapacityProviderStrategy: []*ecs.CapacityProviderStrategyItem{
+			{CapacityProvider: aws.String("cp-1"), Weight: aws.Int64(1), Base: aws.Int64(2)},
+			{CapacityProvider: aws.String("cp-2"), Weight: aws.Int64(3)},
+		},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, out.Cluster.CapacityProviders, 2)
+	require.Len(t, out.Cluster.DefaultCapacityProviderStrategy, 2)
+	assert.Equal(t, "cp-1", aws.StringValue(out.Cluster.DefaultCapacityProviderStrategy[0].CapacityProvider))
+	assert.Equal(t, int64(2), aws.Int64Value(out.Cluster.DefaultCapacityProviderStrategy[0].Base))
+
+	// Persisted onto the cluster record itself, visible via DescribeClusters.
+	described, err := svc.DescribeClusters(ctx, &ecs.DescribeClustersInput{Clusters: []*string{aws.String("web")}}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, described.Clusters, 1)
+	require.Len(t, described.Clusters[0].CapacityProviders, 2)
+	require.Len(t, described.Clusters[0].DefaultCapacityProviderStrategy, 2)
+}
+
+func TestService_PutClusterCapacityProviders_UnknownCluster(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.PutClusterCapacityProviders(context.Background(), &ecs.PutClusterCapacityProvidersInput{
+		Cluster:           aws.String("ghost"),
+		CapacityProviders: []*string{aws.String("cp-1")},
+	}, testAccountID)
+	require.Error(t, err)
+}

spinifex/handlers/ecs/conv.go

@@ -48,6 +48,16 @@ func taskShortID(ref string) string {
 	return ref
 }
 
+// capacityProviderShortName extracts the capacity-provider name from a name
+// or a capacity-provider ARN.
+func capacityProviderShortName(ref string) string {
+	ref = strings.TrimSpace(ref)
+	if i := strings.LastIndex(ref, "capacity-provider/"); i >= 0 {
+		return ref[i+len("capacity-provider/"):]
+	}
+	return ref
+}
+
 // containerDefsFromAWS maps SDK container definitions to the persisted subset.
 func containerDefsFromAWS(in []*ecs.ContainerDefinition) []ContainerDef {
 	out := make([]ContainerDef, 0, len(in))

spinifex/handlers/ecs/instances.go

@@ -55,6 +55,9 @@ func (s *Service) RegisterContainerInstance(_ context.Context, input *ecs.Regist
 				r.TotalMemoryMiB = int(aws.Int64Value(res.IntegerValue))
 			}
 		}
+		if len(input.Tags) > 0 {
+			r.Tags = tagsToMap(input.Tags)
+		}
 		// The agent heartbeats by re-registering. A re-register from a reaped
 		// (involuntarily drained) instance proves the agent is back, so restore
 		// it to ACTIVE. An operator drain (Reaped=false) is left untouched.
@@ -157,6 +160,7 @@ func (s *Service) instanceToAWS(r *InstanceRecord) *ecs.ContainerInstance {
 		RegisteredResources:  registered,
 		RemainingResources:   remaining,
 		VersionInfo:          &ecs.VersionInfo{AgentVersion: aws.String(r.AgentVersion)},
+		Tags:                 tagsToAWS(r.Tags),
 	}
 }
 

spinifex/handlers/ecs/records.go

@@ -50,6 +50,9 @@ const (
 	// circuitBreakerFailureThreshold trips the breaker once this many of the
 	// primary deployment's task launches stop before ever reaching RUNNING.
 	circuitBreakerFailureThreshold = 3
+
+	CapacityProviderStatusActive   = "ACTIVE"
+	CapacityProviderStatusInactive = "INACTIVE"
 )
 
 // Deployment is one rollout of a service's task definition. A service has exactly
@@ -94,6 +97,10 @@ func ServiceARN(region, accountID, cluster, name string) string {
 	return fmt.Sprintf("arn:aws:ecs:%s:%s:service/%s/%s", region, accountID, cluster, name)
 }
 
+func CapacityProviderARN(region, accountID, name string) string {
+	return fmt.Sprintf("arn:aws:ecs:%s:%s:capacity-provider/%s", region, accountID, name)
+}
+
 // serviceTaskGroup is the AWS task-group label a service stamps on its tasks
 // ("service:{name}"). The reconciler counts a service's tasks by this group and
 // the task-state hook resolves a task back to its owning service through it.
@@ -118,6 +125,52 @@ type ClusterRecord struct {
 	Status    string            `json:"status"`
 	Tags      map[string]string `json:"tags,omitempty"`
 	CreatedAt time.Time         `json:"createdAt"`
+	// CapacityProviders / DefaultCapacityProviderStrategy are accepted and
+	// persisted by PutClusterCapacityProviders but are otherwise inert in v1:
+	// no scheduler coupling, no scale loop (a separate follow-on binds them to
+	// an ASG primitive).
+	CapacityProviders               []string                       `json:"capacityProviders,omitempty"`
+	DefaultCapacityProviderStrategy []CapacityProviderStrategyItem `json:"defaultCapacityProviderStrategy,omitempty"`
+}
+
+// CapacityProviderStrategyItem is one entry of a capacity-provider strategy:
+// how many tasks (Base) and what share of the remainder (Weight) a named
+// capacity provider takes. Persisted verbatim; not consulted by placement.
+type CapacityProviderStrategyItem struct {
+	Provider string `json:"provider"`
+	Weight   int    `json:"weight,omitempty"`
+	Base     int    `json:"base,omitempty"`
+}
+
+// ManagedScalingRecord is the persisted subset of an AutoScalingGroupProvider's
+// managed-scaling configuration. Accepted and stored; no scale loop reads it.
+type ManagedScalingRecord struct {
+	Status                 string `json:"status,omitempty"`
+	TargetCapacity         int    `json:"targetCapacity,omitempty"`
+	MinimumScalingStepSize int    `json:"minimumScalingStepSize,omitempty"`
+	MaximumScalingStepSize int    `json:"maximumScalingStepSize,omitempty"`
+	InstanceWarmupPeriod   int    `json:"instanceWarmupPeriod,omitempty"`
+}
+
+// AutoScalingGroupProviderRecord is the persisted ASG binding for a capacity
+// provider. Spinifex has no ASG primitive in v1, so this is stored for API
+// parity only; nothing reads AutoScalingGroupARN to launch/terminate capacity.
+type AutoScalingGroupProviderRecord struct {
+	AutoScalingGroupARN          string               `json:"autoScalingGroupArn"`
+	ManagedScaling               ManagedScalingRecord `json:"managedScaling,omitzero"`
+	ManagedTerminationProtection string               `json:"managedTerminationProtection,omitempty"`
+}
+
+// CapacityProviderRecord is the persisted capacity provider at
+// CapacityProviderKey. Account-scoped (not cluster-scoped), matching the AWS
+// ARN shape; a cluster references providers by name via CapacityProviders.
+type CapacityProviderRecord struct {
+	Name                     string                         `json:"name"`
+	ARN                      string                         `json:"arn"`
+	Status                   string                         `json:"status"`
+	AutoScalingGroupProvider AutoScalingGroupProviderRecord `json:"autoScalingGroupProvider"`
+	Tags                     map[string]string              `json:"tags,omitempty"`
+	CreatedAt                time.Time                      `json:"createdAt"`
 }
 
 // ContainerDef is the persisted subset of an ecs.ContainerDefinition needed to
@@ -144,17 +197,18 @@ const LogDriverJSONFile = "json-file"
 
 // TaskDefRecord is the persisted task definition revision at TaskDefRevKey.
 type TaskDefRecord struct {
-	Family           string         `json:"family"`
-	Revision         int            `json:"revision"`
-	ARN              string         `json:"arn"`
-	NetworkMode      string         `json:"networkMode,omitempty"`
-	CPU              string         `json:"cpu,omitempty"`
-	Memory           string         `json:"memory,omitempty"`
-	TaskRoleArn      string         `json:"taskRoleArn,omitempty"`
-	ExecutionRoleArn string         `json:"executionRoleArn,omitempty"`
-	Containers       []ContainerDef `json:"containers"`
-	Status           string         `json:"status"`
-	RegisteredAt     time.Time      `json:"registeredAt"`
+	Family           string            `json:"family"`
+	Revision         int               `json:"revision"`
+	ARN              string            `json:"arn"`
+	NetworkMode      string            `json:"networkMode,omitempty"`
+	CPU              string            `json:"cpu,omitempty"`
+	Memory           string            `json:"memory,omitempty"`
+	TaskRoleArn      string            `json:"taskRoleArn,omitempty"`
+	ExecutionRoleArn string            `json:"executionRoleArn,omitempty"`
+	Containers       []ContainerDef    `json:"containers"`
+	Status           string            `json:"status"`
+	Tags             map[string]string `json:"tags,omitempty"`
+	RegisteredAt     time.Time         `json:"registeredAt"`
 }
 
 // reservedCPU/reservedMemory sum the task definition's per-container reservations
@@ -180,14 +234,15 @@ func (t *TaskDefRecord) reservedMemory() int {
 // scheduler writes it from the Layer-2 bus (register/heartbeat) and reserves
 // capacity by appending placed task IDs.
 type InstanceRecord struct {
-	InstanceID     string `json:"instanceId"`
-	ARN            string `json:"arn"`
-	Cluster        string `json:"cluster"`
-	AZ             string `json:"availabilityZone,omitempty"`
-	Hostname       string `json:"hostname,omitempty"`
-	Status         string `json:"status"`
-	TotalCPU       int    `json:"totalCpu"`
-	TotalMemoryMiB int    `json:"totalMemoryMiB"`
+	InstanceID     string            `json:"instanceId"`
+	ARN            string            `json:"arn"`
+	Cluster        string            `json:"cluster"`
+	AZ             string            `json:"availabilityZone,omitempty"`
+	Hostname       string            `json:"hostname,omitempty"`
+	Status         string            `json:"status"`
+	Tags           map[string]string `json:"tags,omitempty"`
+	TotalCPU       int               `json:"totalCpu"`
+	TotalMemoryMiB int               `json:"totalMemoryMiB"`
 	// ReservedCPU/ReservedMemoryMiB track capacity committed to placed tasks;
 	// placement increments them under a KV CAS and the task-state STOPPED path
 	// releases them. Remaining = Total - Reserved.
@@ -221,19 +276,20 @@ type TaskRecord struct {
 	// Group / StartedBy mirror the AWS task fields. A service's tasks carry
 	// Group="service:{name}" so the reconciler counts them and the task-state
 	// hook resolves a RUNNING/STOPPED task back to its owning service.
-	Group                string           `json:"group,omitempty"`
-	StartedBy            string           `json:"startedBy,omitempty"`
-	TaskDefFamily        string           `json:"taskDefFamily"`
-	TaskDefRevision      int              `json:"taskDefRevision"`
-	TaskDefARN           string           `json:"taskDefArn"`
-	ContainerInstanceID  string           `json:"containerInstanceId,omitempty"`
-	ContainerInstanceARN string           `json:"containerInstanceArn,omitempty"`
-	DesiredStatus        string           `json:"desiredStatus"`
-	LastStatus           string           `json:"lastStatus"`
-	StoppedReason        string           `json:"stoppedReason,omitempty"`
-	ReservedCPU          int              `json:"reservedCpu"`
-	ReservedMemoryMiB    int              `json:"reservedMemoryMiB"`
-	Containers           []ContainerState `json:"containers,omitempty"`
+	Group                string            `json:"group,omitempty"`
+	StartedBy            string            `json:"startedBy,omitempty"`
+	TaskDefFamily        string            `json:"taskDefFamily"`
+	TaskDefRevision      int               `json:"taskDefRevision"`
+	TaskDefARN           string            `json:"taskDefArn"`
+	ContainerInstanceID  string            `json:"containerInstanceId,omitempty"`
+	ContainerInstanceARN string            `json:"containerInstanceArn,omitempty"`
+	DesiredStatus        string            `json:"desiredStatus"`
+	LastStatus           string            `json:"lastStatus"`
+	StoppedReason        string            `json:"stoppedReason,omitempty"`
+	ReservedCPU          int               `json:"reservedCpu"`
+	ReservedMemoryMiB    int               `json:"reservedMemoryMiB"`
+	Tags                 map[string]string `json:"tags,omitempty"`
+	Containers           []ContainerState  `json:"containers,omitempty"`
 	// NetworkMode is the resolved task network mode (awsvpc|bridge|host). The
 	// STOPPED path consults it to decide whether an ENI must be reclaimed.
 	NetworkMode string `json:"networkMode,omitempty"`
@@ -286,6 +342,7 @@ type ServiceRecord struct {
 	DeploymentID       string               `json:"deploymentId"`
 	RunningCount       int                  `json:"runningCount"`
 	PendingCount       int                  `json:"pendingCount"`
+	Tags               map[string]string    `json:"tags,omitempty"`
 	// Rolling-update configuration (deploymentConfiguration) and its live state.
 	// MinimumHealthyPercent / MaximumPercent gate the rollout; the circuit breaker
 	// trips a failing deployment and optionally rolls back to LastGoodTaskDefARN.

spinifex/handlers/ecs/service.go

@@ -59,6 +59,20 @@ type ECSService interface {
 
 	// ProvisionCapacity launches container-instance EC2 capacity into a cluster.
 	ProvisionCapacity(ctx context.Context, input *ProvisionCapacityInput, accountID string) (*ProvisionCapacityOutput, error)
+
+	// Tags. Dispatched on the resourceArn shape (ecs-v1.md §1); the ACM inline-tags
+	// pattern (map stored on the record, mutated in place).
+	ListTagsForResource(ctx context.Context, input *ecs.ListTagsForResourceInput, accountID string) (*ecs.ListTagsForResourceOutput, error)
+	TagResource(ctx context.Context, input *ecs.TagResourceInput, accountID string) (*ecs.TagResourceOutput, error)
+	UntagResource(ctx context.Context, input *ecs.UntagResourceInput, accountID string) (*ecs.UntagResourceOutput, error)
+
+	// Capacity providers. Strategy is accepted and persisted but inert: no
+	// scheduler coupling, no scale loop (a separate follow-on binds this to an
+	// ASG primitive).
+	PutClusterCapacityProviders(ctx context.Context, input *ecs.PutClusterCapacityProvidersInput, accountID string) (*ecs.PutClusterCapacityProvidersOutput, error)
+	CreateCapacityProvider(ctx context.Context, input *ecs.CreateCapacityProviderInput, accountID string) (*ecs.CreateCapacityProviderOutput, error)
+	DescribeCapacityProviders(ctx context.Context, input *ecs.DescribeCapacityProvidersInput, accountID string) (*ecs.DescribeCapacityProvidersOutput, error)
+	DeleteCapacityProvider(ctx context.Context, input *ecs.DeleteCapacityProviderInput, accountID string) (*ecs.DeleteCapacityProviderOutput, error)
 }
 
 // ecsImageResolver is the narrow AMI surface for resolving the spinifex-ecs-node
@@ -306,12 +320,19 @@ func (s *Service) ListClusters(_ context.Context, _ *ecs.ListClustersInput, acco
 }
 
 func (r *ClusterRecord) toAWS() *ecs.Cluster {
-	return &ecs.Cluster{
+	c := &ecs.Cluster{
 		ClusterName: aws.String(r.Name),
 		ClusterArn:  aws.String(r.ARN),
 		Status:      aws.String(r.Status),
 		Tags:        tagsToAWS(r.Tags),
 	}
+	if len(r.CapacityProviders) > 0 {
+		c.CapacityProviders = aws.StringSlice(r.CapacityProviders)
+	}
+	for _, item := range r.DefaultCapacityProviderStrategy {
+		c.DefaultCapacityProviderStrategy = append(c.DefaultCapacityProviderStrategy, item.toAWS())
+	}
+	return c
 }
 
 // tagsToAWS converts a stored tag map into the AWS list form with a stable
@@ -328,6 +349,25 @@ func tagsToAWS(tags map[string]string) []*ecs.Tag {
 	return out
 }
 
+// tagsToMap is the inverse of tagsToAWS: it converts the AWS tag list form
+// into a stored map, dropping nil entries and empty keys.
+func tagsToMap(tags []*ecs.Tag) map[string]string {
+	if len(tags) == 0 {
+		return nil
+	}
+	out := make(map[string]string, len(tags))
+	for _, t := range tags {
+		if t == nil || aws.StringValue(t.Key) == "" {
+			continue
+		}
+		out[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
+	}
+	if len(out) == 0 {
+		return nil
+	}
+	return out
+}
+
 // --- Task definition ---
 
 // RegisterTaskDefinition stores a new revision of a family, bumping latest-rev.
@@ -360,6 +400,7 @@ func (s *Service) RegisterTaskDefinition(ctx context.Context, input *ecs.Registe
 		TaskRoleArn:      aws.StringValue(input.TaskRoleArn),
 		ExecutionRoleArn: aws.StringValue(input.ExecutionRoleArn),
 		Status:           TaskDefStatusActive,
+		Tags:             tagsToMap(input.Tags),
 		RegisteredAt:     time.Now().UTC(),
 		Containers:       containerDefsFromAWS(input.ContainerDefinitions),
 	}
@@ -369,7 +410,7 @@ func (s *Service) RegisterTaskDefinition(ctx context.Context, input *ecs.Registe
 	if err := putJSON(kv, TaskDefLatestRevKey(family), rev); err != nil {
 		return nil, err
 	}
-	return &ecs.RegisterTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil
+	return &ecs.RegisterTaskDefinitionOutput{TaskDefinition: rec.toAWS(), Tags: tagsToAWS(rec.Tags)}, nil
 }
 
 // validateContainerDefs hard-rejects taskdef features the data plane cannot honor.
@@ -451,7 +492,7 @@ func (s *Service) DescribeTaskDefinition(_ context.Context, input *ecs.DescribeT
 	if err != nil {
 		return nil, err
 	}
-	return &ecs.DescribeTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil
+	return &ecs.DescribeTaskDefinitionOutput{TaskDefinition: rec.toAWS(), Tags: tagsToAWS(rec.Tags)}, nil
 }
 
 // ListTaskDefinitions returns revision ARNs, optionally filtered by family and

spinifex/handlers/ecs/service_nats.go

@@ -139,3 +139,35 @@ func (s *NATSECSService) PollAssignments(ctx context.Context, input *PollAssignm
 func (s *NATSECSService) ProvisionCapacity(ctx context.Context, input *ProvisionCapacityInput, accountID string) (*ProvisionCapacityOutput, error) {
 	return utils.NATSRequest[ProvisionCapacityOutput](ctx, s.natsConn, "ecs.ProvisionCapacity", input, defaultTimeout, accountID)
 }
+
+// --- Tags ---
+
+func (s *NATSECSService) ListTagsForResource(ctx context.Context, input *ecs.ListTagsForResourceInput, accountID string) (*ecs.ListTagsForResourceOutput, error) {
+	return utils.NATSRequest[ecs.ListTagsForResourceOutput](ctx, s.natsConn, "ecs.ListTagsForResource", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) TagResource(ctx context.Context, input *ecs.TagResourceInput, accountID string) (*ecs.TagResourceOutput, error) {
+	return utils.NATSRequest[ecs.TagResourceOutput](ctx, s.natsConn, "ecs.TagResource", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) UntagResource(ctx context.Context, input *ecs.UntagResourceInput, accountID string) (*ecs.UntagResourceOutput, error) {
+	return utils.NATSRequest[ecs.UntagResourceOutput](ctx, s.natsConn, "ecs.UntagResource", input, defaultTimeout, accountID)
+}
+
+// --- Capacity providers ---
+
+func (s *NATSECSService) PutClusterCapacityProviders(ctx context.Context, input *ecs.PutClusterCapacityProvidersInput, accountID string) (*ecs.PutClusterCapacityProvidersOutput, error) {
+	return utils.NATSRequest[ecs.PutClusterCapacityProvidersOutput](ctx, s.natsConn, "ecs.PutClusterCapacityProviders", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) CreateCapacityProvider(ctx context.Context, input *ecs.CreateCapacityProviderInput, accountID string) (*ecs.CreateCapacityProviderOutput, error) {
+	return utils.NATSRequest[ecs.CreateCapacityProviderOutput](ctx, s.natsConn, "ecs.CreateCapacityProvider", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) DescribeCapacityProviders(ctx context.Context, input *ecs.DescribeCapacityProvidersInput, accountID string) (*ecs.DescribeCapacityProvidersOutput, error) {
+	return utils.NATSRequest[ecs.DescribeCapacityProvidersOutput](ctx, s.natsConn, "ecs.DescribeCapacityProviders", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) DeleteCapacityProvider(ctx context.Context, input *ecs.DeleteCapacityProviderInput, accountID string) (*ecs.DeleteCapacityProviderOutput, error) {
+	return utils.NATSRequest[ecs.DeleteCapacityProviderOutput](ctx, s.natsConn, "ecs.DeleteCapacityProvider", input, defaultTimeout, accountID)
+}

spinifex/handlers/ecs/services.go

@@ -175,6 +175,7 @@ func (s *Service) CreateService(ctx context.Context, input *ecs.CreateServiceInp
 		PlacementStrategy:  placementStrategyFromAWS(input.PlacementStrategy),
 		LoadBalancers:      loadBalancersFromAWS(input.LoadBalancers),
 		DeploymentID:       uuid.NewString(),
+		Tags:               tagsToMap(input.Tags),
 		CreatedAt:          now,
 		UpdatedAt:          now,
 	}
@@ -612,6 +613,7 @@ func (s *Service) serviceToAWS(accountID string, r *ServiceRecord) *ecs.Service
 		PendingCount:       aws.Int64(int64(r.PendingCount)),
 		SchedulingStrategy: aws.String(r.SchedulingStrategy),
 		TaskDefinition:     aws.String(r.TaskDefARN),
+		Tags:               tagsToAWS(r.Tags),
 	}
 	if r.LaunchType != "" {
 		svc.LaunchType = aws.String(r.LaunchType)

spinifex/handlers/ecs/store.go

@@ -135,6 +135,18 @@ func LeaderLeaseKey(accountID, clusterName string) string {
 	return fmt.Sprintf("%s/%s", accountID, clusterName)
 }
 
+// CapacityProvidersPrefix returns the KV key prefix under which all
+// capacity-provider records live. Account-scoped (not cluster-scoped),
+// matching the AWS ARN shape; a cluster references providers by name.
+func CapacityProvidersPrefix() string {
+	return "capacity-providers/"
+}
+
+// CapacityProviderKey returns the KV key for a named capacity-provider record.
+func CapacityProviderKey(name string) string {
+	return CapacityProvidersPrefix() + name
+}
+
 // Store is the per-daemon ECS KV handle. Per-account and leader buckets are
 // accessed via the package-level factories below.
 type Store struct {

spinifex/handlers/ecs/tags.go

@@ -0,0 +1,268 @@
+package handlers_ecs
+
+import (
+	"context"
+	"errors"
+	"maps"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/nats-io/nats.go"
+)
+
+// ecsResourceKind identifies which stored record kind an ARN refers to.
+type ecsResourceKind int
+
+const (
+	ecsResourceCluster ecsResourceKind = iota
+	ecsResourceTaskDefinition
+	ecsResourceService
+	ecsResourceTask
+	ecsResourceContainerInstance
+)
+
+// parseResourceARN splits an ECS resource ARN into its kind, owning cluster
+// (empty for cluster/task-definition, which are flat), and short id/name.
+// Cluster and task-definition ARNs are flat (.../cluster/{name},
+// .../task-definition/{family}:{rev}); service/task/container-instance ARNs
+// embed the owning cluster (.../service/{cluster}/{name}).
+func parseResourceARN(resourceARN string) (kind ecsResourceKind, cluster, id string, err error) {
+	// SplitN(..., 6) keeps the resource segment intact even though a
+	// task-definition id ("family:1") itself contains a colon.
+	parts := strings.SplitN(resourceARN, ":", 6)
+	if len(parts) != 6 || parts[0] != "arn" {
+		return 0, "", "", errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	rtype, rest, ok := strings.Cut(parts[5], "/")
+	if !ok || rtype == "" || rest == "" {
+		return 0, "", "", errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+
+	switch rtype {
+	case "cluster":
+		return ecsResourceCluster, rest, rest, nil
+	case "task-definition":
+		return ecsResourceTaskDefinition, "", rest, nil
+	case "service", "task", "container-instance":
+		clusterName, name, ok := strings.Cut(rest, "/")
+		if !ok || clusterName == "" || name == "" {
+			return 0, "", "", errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		switch rtype {
+		case "service":
+			return ecsResourceService, clusterName, name, nil
+		case "task":
+			return ecsResourceTask, clusterName, name, nil
+		default:
+			return ecsResourceContainerInstance, clusterName, name, nil
+		}
+	default:
+		return 0, "", "", errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+}
+
+// resourceTags reads the tag map stored on the ARN-identified resource.
+func (s *Service) resourceTags(kv nats.KeyValue, kind ecsResourceKind, cluster, id string) (map[string]string, error) {
+	switch kind {
+	case ecsResourceCluster:
+		var rec ClusterRecord
+		found, err := getJSON(kv, ClusterMetaKey(id), &rec)
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			return nil, errors.New(awserrors.ErrorECSClusterNotFound)
+		}
+		return rec.Tags, nil
+	case ecsResourceTaskDefinition:
+		family, rev := parseTaskDefRef(id)
+		if family == "" || rev == 0 {
+			return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		var rec TaskDefRecord
+		found, err := getJSON(kv, TaskDefRevKey(family, rev), &rec)
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		return rec.Tags, nil
+	case ecsResourceService:
+		var rec ServiceRecord
+		found, err := getJSON(kv, ServiceKey(cluster, id), &rec)
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			return nil, errors.New(awserrors.ErrorECSServiceNotFound)
+		}
+		return rec.Tags, nil
+	case ecsResourceTask:
+		var rec TaskRecord
+		found, err := getJSON(kv, TaskKey(cluster, id), &rec)
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		return rec.Tags, nil
+	case ecsResourceContainerInstance:
+		var rec InstanceRecord
+		found, err := getJSON(kv, InstanceKey(cluster, id), &rec)
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		return rec.Tags, nil
+	default:
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+}
+
+// mutateResourceTags reads the ARN-identified resource, applies mutate to its
+// tag map, and re-puts the record. Used by TagResource (merge) and
+// UntagResource (delete).
+func (s *Service) mutateResourceTags(kv nats.KeyValue, kind ecsResourceKind, cluster, id string, mutate func(map[string]string) map[string]string) error {
+	switch kind {
+	case ecsResourceCluster:
+		var rec ClusterRecord
+		found, err := getJSON(kv, ClusterMetaKey(id), &rec)
+		if err != nil {
+			return err
+		}
+		if !found {
+			return errors.New(awserrors.ErrorECSClusterNotFound)
+		}
+		rec.Tags = mutate(rec.Tags)
+		return putJSON(kv, ClusterMetaKey(id), &rec)
+	case ecsResourceTaskDefinition:
+		family, rev := parseTaskDefRef(id)
+		if family == "" || rev == 0 {
+			return errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		var rec TaskDefRecord
+		found, err := getJSON(kv, TaskDefRevKey(family, rev), &rec)
+		if err != nil {
+			return err
+		}
+		if !found {
+			return errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		rec.Tags = mutate(rec.Tags)
+		return putJSON(kv, TaskDefRevKey(family, rev), &rec)
+	case ecsResourceService:
+		var rec ServiceRecord
+		found, err := getJSON(kv, ServiceKey(cluster, id), &rec)
+		if err != nil {
+			return err
+		}
+		if !found {
+			return errors.New(awserrors.ErrorECSServiceNotFound)
+		}
+		rec.Tags = mutate(rec.Tags)
+		return putJSON(kv, ServiceKey(cluster, id), &rec)
+	case ecsResourceTask:
+		var rec TaskRecord
+		found, err := getJSON(kv, TaskKey(cluster, id), &rec)
+		if err != nil {
+			return err
+		}
+		if !found {
+			return errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		rec.Tags = mutate(rec.Tags)
+		return putJSON(kv, TaskKey(cluster, id), &rec)
+	case ecsResourceContainerInstance:
+		var rec InstanceRecord
+		found, err := getJSON(kv, InstanceKey(cluster, id), &rec)
+		if err != nil {
+			return err
+		}
+		if !found {
+			return errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		rec.Tags = mutate(rec.Tags)
+		return putJSON(kv, InstanceKey(cluster, id), &rec)
+	default:
+		return errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+}
+
+// ListTagsForResource returns the tags stored on the ARN-identified resource.
+func (s *Service) ListTagsForResource(_ context.Context, input *ecs.ListTagsForResourceInput, accountID string) (*ecs.ListTagsForResourceOutput, error) {
+	if input == nil || aws.StringValue(input.ResourceArn) == "" {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	kind, cluster, id, err := parseResourceARN(aws.StringValue(input.ResourceArn))
+	if err != nil {
+		return nil, err
+	}
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	tags, err := s.resourceTags(kv, kind, cluster, id)
+	if err != nil {
+		return nil, err
+	}
+	return &ecs.ListTagsForResourceOutput{Tags: tagsToAWS(tags)}, nil
+}
+
+// TagResource merges the supplied tags onto the ARN-identified resource.
+func (s *Service) TagResource(_ context.Context, input *ecs.TagResourceInput, accountID string) (*ecs.TagResourceOutput, error) {
+	if input == nil || aws.StringValue(input.ResourceArn) == "" {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	kind, cluster, id, err := parseResourceARN(aws.StringValue(input.ResourceArn))
+	if err != nil {
+		return nil, err
+	}
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	add := tagsToMap(input.Tags)
+	err = s.mutateResourceTags(kv, kind, cluster, id, func(existing map[string]string) map[string]string {
+		if existing == nil {
+			existing = map[string]string{}
+		}
+		maps.Copy(existing, add)
+		return existing
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &ecs.TagResourceOutput{}, nil
+}
+
+// UntagResource deletes the named tag keys from the ARN-identified resource.
+func (s *Service) UntagResource(_ context.Context, input *ecs.UntagResourceInput, accountID string) (*ecs.UntagResourceOutput, error) {
+	if input == nil || aws.StringValue(input.ResourceArn) == "" {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	kind, cluster, id, err := parseResourceARN(aws.StringValue(input.ResourceArn))
+	if err != nil {
+		return nil, err
+	}
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	keys := awsStringSlice(input.TagKeys)
+	err = s.mutateResourceTags(kv, kind, cluster, id, func(existing map[string]string) map[string]string {
+		for _, k := range keys {
+			delete(existing, k)
+		}
+		return existing
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &ecs.UntagResourceOutput{}, nil
+}

spinifex/handlers/ecs/tags_test.go

@@ -0,0 +1,229 @@
+package handlers_ecs
+
+import (
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseResourceARN(t *testing.T) {
+	cases := []struct {
+		name        string
+		arn         string
+		wantKind    ecsResourceKind
+		wantCluster string
+		wantID      string
+	}{
+		{"cluster", "arn:aws:ecs:ap-southeast-2:123456789012:cluster/web", ecsResourceCluster, "web", "web"},
+		{"task-definition", "arn:aws:ecs:ap-southeast-2:123456789012:task-definition/nginx:3", ecsResourceTaskDefinition, "", "nginx:3"},
+		{"service", "arn:aws:ecs:ap-southeast-2:123456789012:service/web/api", ecsResourceService, "web", "api"},
+		{"task", "arn:aws:ecs:ap-southeast-2:123456789012:task/web/t-1", ecsResourceTask, "web", "t-1"},
+		{"container-instance", "arn:aws:ecs:ap-southeast-2:123456789012:container-instance/web/i-1", ecsResourceContainerInstance, "web", "i-1"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			kind, cluster, id, err := parseResourceARN(tc.arn)
+			require.NoError(t, err)
+			assert.Equal(t, tc.wantKind, kind)
+			assert.Equal(t, tc.wantCluster, cluster)
+			assert.Equal(t, tc.wantID, id)
+		})
+	}
+}
+
+func TestParseResourceARN_Invalid(t *testing.T) {
+	cases := []string{
+		"",
+		"not-an-arn",
+		"arn:aws:ecs:r:a:cluster",         // no "/" after resource type
+		"arn:aws:ecs:r:a:service/onlyone", // service missing embedded name
+		"arn:aws:ecs:r:a:unknown/x",       // unrecognized resource type
+	}
+	for _, arnStr := range cases {
+		t.Run(arnStr, func(t *testing.T) {
+			_, _, _, err := parseResourceARN(arnStr)
+			assert.EqualError(t, err, "InvalidParameterException")
+		})
+	}
+}
+
+// TestService_Tags_ClusterRoundTrip covers the flat ARN shape end to end via
+// the wired TagResource/ListTagsForResource/UntagResource actions.
+func TestService_Tags_ClusterRoundTrip(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	created, err := svc.CreateCluster(ctx, &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	arn := aws.StringValue(created.Cluster.ClusterArn)
+
+	_, err = svc.TagResource(ctx, &ecs.TagResourceInput{
+		ResourceArn: aws.String(arn),
+		Tags: []*ecs.Tag{
+			{Key: aws.String("team"), Value: aws.String("infra")},
+			{Key: aws.String("env"), Value: aws.String("prod")},
+		},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	listed, err := svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, listed.Tags, 2)
+
+	// Merge in an extra tag alongside the existing pair.
+	_, err = svc.TagResource(ctx, &ecs.TagResourceInput{
+		ResourceArn: aws.String(arn),
+		Tags:        []*ecs.Tag{{Key: aws.String("owner"), Value: aws.String("platform")}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	listed, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, listed.Tags, 3)
+
+	_, err = svc.UntagResource(ctx, &ecs.UntagResourceInput{
+		ResourceArn: aws.String(arn),
+		TagKeys:     []*string{aws.String("env")},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	listed, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, listed.Tags, 2)
+	for _, tag := range listed.Tags {
+		assert.NotEqual(t, "env", aws.StringValue(tag.Key))
+	}
+}
+
+// TestService_Tags_ServiceRoundTrip covers the cluster-embedded ARN shape.
+func TestService_Tags_ServiceRoundTrip(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	_, err := svc.CreateCluster(ctx, &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerTaskDef(t, svc, "nginx", 128, 256)
+
+	created, err := svc.CreateService(ctx, &ecs.CreateServiceInput{
+		Cluster:        aws.String("web"),
+		ServiceName:    aws.String("api"),
+		TaskDefinition: aws.String("nginx"),
+		DesiredCount:   aws.Int64(0),
+	}, testAccountID)
+	require.NoError(t, err)
+	arn := aws.StringValue(created.Service.ServiceArn)
+
+	_, err = svc.TagResource(ctx, &ecs.TagResourceInput{
+		ResourceArn: aws.String(arn),
+		Tags:        []*ecs.Tag{{Key: aws.String("team"), Value: aws.String("infra")}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	listed, err := svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, listed.Tags, 1)
+	assert.Equal(t, "team", aws.StringValue(listed.Tags[0].Key))
+
+	_, err = svc.UntagResource(ctx, &ecs.UntagResourceInput{
+		ResourceArn: aws.String(arn),
+		TagKeys:     []*string{aws.String("team")},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	listed, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, listed.Tags)
+}
+
+// TestService_Tags_TaskDefinitionRoundTrip covers the flat family:rev ARN shape.
+func TestService_Tags_TaskDefinitionRoundTrip(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	out := registerTaskDef(t, svc, "nginx", 128, 256)
+	arn := aws.StringValue(out.TaskDefinition.TaskDefinitionArn)
+
+	_, err := svc.TagResource(ctx, &ecs.TagResourceInput{
+		ResourceArn: aws.String(arn),
+		Tags:        []*ecs.Tag{{Key: aws.String("env"), Value: aws.String("prod")}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	listed, err := svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, listed.Tags, 1)
+
+	described, err := svc.DescribeTaskDefinition(ctx, &ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String(arn)}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, described.Tags, 1)
+	assert.Equal(t, "env", aws.StringValue(described.Tags[0].Key))
+}
+
+// TestService_Tags_MissingResource asserts the not-found mapping the plan
+// specifies: cluster/service resolve to their named NotFound errors; task and
+// container-instance resolve to the generic InvalidParameter.
+func TestService_Tags_MissingResource(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	_, err := svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{
+		ResourceArn: aws.String("arn:aws:ecs:ap-southeast-2:123456789012:cluster/nope"),
+	}, testAccountID)
+	assert.EqualError(t, err, "ClusterNotFoundException")
+
+	_, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{
+		ResourceArn: aws.String("arn:aws:ecs:ap-southeast-2:123456789012:service/web/nope"),
+	}, testAccountID)
+	assert.EqualError(t, err, "ServiceNotFoundException")
+
+	_, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{
+		ResourceArn: aws.String("arn:aws:ecs:ap-southeast-2:123456789012:task/web/nope"),
+	}, testAccountID)
+	assert.EqualError(t, err, "InvalidParameterException")
+
+	_, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{
+		ResourceArn: aws.String("arn:aws:ecs:ap-southeast-2:123456789012:container-instance/web/nope"),
+	}, testAccountID)
+	assert.EqualError(t, err, "InvalidParameterException")
+}
+
+// TestService_RunTask_TagsPersisted confirms RunTask plumbs input.Tags onto
+// the persisted task record and DescribeTasks reflects them back.
+func TestService_RunTask_TagsPersisted(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	_, err := svc.CreateCluster(ctx, &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerTaskDef(t, svc, "nginx", 128, 256)
+	_, err = svc.RegisterContainerInstance(ctx, &ecs.RegisterContainerInstanceInput{
+		Cluster: aws.String("web"),
+		TotalResources: []*ecs.Resource{
+			{Name: aws.String("CPU"), IntegerValue: aws.Int64(1024)},
+			{Name: aws.String("MEMORY"), IntegerValue: aws.Int64(2048)},
+		},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	out, err := svc.RunTask(ctx, &ecs.RunTaskInput{
+		Cluster:        aws.String("web"),
+		TaskDefinition: aws.String("nginx"),
+		Tags:           []*ecs.Tag{{Key: aws.String("env"), Value: aws.String("prod")}},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, out.Tasks, 1)
+	require.Len(t, out.Tasks[0].Tags, 1)
+	assert.Equal(t, "env", aws.StringValue(out.Tasks[0].Tags[0].Key))
+
+	described, err := svc.DescribeTasks(ctx, &ecs.DescribeTasksInput{
+		Cluster: aws.String("web"),
+		Tasks:   []*string{out.Tasks[0].TaskArn},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, described.Tasks, 1)
+	require.Len(t, described.Tasks[0].Tags, 1)
+}

spinifex/handlers/ecs/task_stop.go

@@ -93,6 +93,7 @@ func (s *Service) StartTask(ctx context.Context, input *ecs.StartTaskInput, acco
 		rec.NetworkMode = mode
 		rec.Group = group
 		rec.StartedBy = startedBy
+		rec.Tags = tagsToMap(input.Tags)
 		if mode == NetworkModeAwsvpc {
 			if failure := s.provisionTaskENI(ctx, kv, accountID, cluster, rec, netCfg); failure != nil {
 				out.Failures = append(out.Failures, failure)

spinifex/handlers/ecs/tasks.go

@@ -74,6 +74,7 @@ func (s *Service) RunTask(ctx context.Context, input *ecs.RunTaskInput, accountI
 		rec.NetworkMode = mode
 		rec.Group = aws.StringValue(input.Group)
 		rec.StartedBy = aws.StringValue(input.StartedBy)
+		rec.Tags = tagsToMap(input.Tags)
 		if mode == NetworkModeAwsvpc {
 			if failure := s.provisionTaskENI(ctx, kv, accountID, cluster, rec, netCfg); failure != nil {
 				out.Failures = append(out.Failures, failure)
@@ -271,6 +272,7 @@ func (s *Service) taskToAWS(accountID string, r *TaskRecord) *ecs.Task {
 		DesiredStatus:        aws.String(r.DesiredStatus),
 		LastStatus:           aws.String(r.LastStatus),
 		ContainerInstanceArn: aws.String(r.ContainerInstanceARN),
+		Tags:                 tagsToAWS(r.Tags),
 	}
 	if r.Group != "" {
 		t.Group = aws.String(r.Group)