feat(ecs): Phase 4 gateway scaffolding + KV constants (Sprint 4a) (#413)

* fix: drop removed predastore s3.Config.Compaction from admin_test

The predastore submodule bump removed the Compaction field from s3.Config.
Parse the rendered [compaction] block into a local struct so the template
round-trip test no longer references the deleted field.

* feat(ecs): Phase 4 gateway scaffolding + KV constants (Sprint 4a)

ECS v1 control-plane skeleton. Every action is a 501 NotImplemented stub
until its handler lands in a later Phase 4 sprint; this sprint locks the
API contract + JetStream KV schema so sprints 4c-4h can build against it.

- gateway/ecs: X-Amz-Target JSON 1.1 dispatch (AmazonEC2ContainerServiceV20141113),
  full v1 action namespace mapped to NotImplemented (mirrors gateway/ecrapi)
- gateway: register ecs in supportedServices + Request dispatch switch
- handlers/ecs/store.go: per-account bucket ecs-account-{accountID} + shared
  leader bucket spinifex-ecs-leader + Q2 key-path helpers (mirrors handlers/eks)
- NATS_HANDLER_NAMING.md: document the two-layer ecs.* subject convention (Q14)

Ref: docs/development/feature/ecs-v1.md, ecs-4a-gateway-scaffolding.md

* test(ecs): cover store factories + gateway handler/error branches

Lift Sprint 4a diff coverage above the 60% gate (54.5% -> 78%):
- handlers/ecs: JetStream-backed GetOrCreateAccountBucket / InitLeaderBucket
  idempotency + NewStore success (testutil.StartTestJetStream)
- gateway/ecs: in-package NotImplemented + WriteJSONResponse coverage
- gateway: ECS_Request missing-account-ID branch

spinifex/gateway/ecs.go

@@ -0,0 +1,62 @@
+package gateway
+
+import (
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"strings"
+
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	gateway_ecs "github.com/mulgadc/spinifex/spinifex/gateway/ecs"
+)
+
+// ecsActionFromTarget extracts the action suffix from an X-Amz-Target header.
+// Any "<Prefix>.<Action>" or bare "<Action>" form is accepted.
+func ecsActionFromTarget(target string) string {
+	if i := strings.LastIndex(target, "."); i >= 0 {
+		return target[i+1:]
+	}
+	return target
+}
+
+// ECS_Request dispatches AWS JSON 1.1 ECS control-plane requests. The action
+// comes from X-Amz-Target; errors are returned as awserrors codes and rendered
+// by the shared ErrorHandler. Every action is a NotImplemented (501) stub until
+// its real handler lands in a later Phase 4 sprint.
+func (gw *GatewayConfig) ECS_Request(w http.ResponseWriter, r *http.Request) error {
+	action := ecsActionFromTarget(r.Header.Get("X-Amz-Target"))
+	if action == "" {
+		return errors.New(awserrors.ErrorMissingAction)
+	}
+
+	handler, ok := gateway_ecs.Actions[action]
+	if !ok {
+		slog.Debug("ECS: unknown action", "action", action)
+		return errors.New(awserrors.ErrorInvalidAction)
+	}
+
+	if err := gw.checkPolicy(r, "ecs", action); err != nil {
+		return err
+	}
+
+	accountID, _ := r.Context().Value(ctxAccountID).(string)
+	if accountID == "" {
+		slog.Error("ECS_Request: no account ID in auth context")
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		slog.Error("ECS_Request: failed to read body", "err", err)
+		return errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+
+	output, err := handler(gw.NATSConn, accountID, body)
+	if err != nil {
+		return err
+	}
+
+	gateway_ecs.WriteJSONResponse(w, output)
+	return nil
+}

spinifex/gateway/ecs/handler.go

@@ -0,0 +1,104 @@
+// Package gateway_ecs is the HTTP-side glue for the ECS control plane: the AWS
+// JSON 1.1 surface dispatched by X-Amz-Target
+// (AmazonEC2ContainerServiceV20141113.<Action>), matching aws-sdk-go's
+// service/ecs request shape.
+//
+// The full v1 action namespace (ecs-v1.md §1) is registered as the API contract;
+// every action resolves to the shared NotImplemented stub until its real handler
+// lands in a later Phase 4 sprint.
+package gateway_ecs
+
+import (
+	"errors"
+	"log/slog"
+	"net/http"
+
+	"github.com/aws/aws-sdk-go/private/protocol/json/jsonutil"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/nats-io/nats.go"
+)
+
+// TargetPrefix is the X-Amz-Target service prefix for ECS control-plane calls.
+const TargetPrefix = "AmazonEC2ContainerServiceV20141113"
+
+// JSONContentType is the AWS JSON 1.1 content type ECS clients expect.
+const JSONContentType = "application/x-amz-json-1.1"
+
+// Handler is the signature every ECS control-plane action implements. nc is the
+// gateway NATS connection (handlers relay onto ecs.* subjects); accountID is the
+// resolved caller account; body is the raw JSON 1.1 request payload.
+type Handler func(nc *nats.Conn, accountID string, body []byte) (any, error)
+
+// NotImplemented is the placeholder for every unimplemented ECS control-plane
+// action. It returns the AWS NotImplemented error, which the shared gateway
+// ErrorHandler renders as a 501 JSON 1.1 envelope.
+func NotImplemented(_ *nats.Conn, _ string, _ []byte) (any, error) {
+	return nil, errors.New(awserrors.ErrorNotImplemented)
+}
+
+// Actions is the authoritative ECS control-plane action namespace (ecs-v1.md §1).
+// Real implementations replace the NotImplemented value as each action lands; the
+// key set is the v1 API contract.
+var Actions = map[string]Handler{
+	// Cluster. PutClusterCapacityProviders is a no-op stub in v1 (Q15).
+	"CreateCluster":               NotImplemented,
+	"DescribeClusters":            NotImplemented,
+	"ListClusters":                NotImplemented,
+	"DeleteCluster":               NotImplemented,
+	"UpdateCluster":               NotImplemented,
+	"PutClusterCapacityProviders": NotImplemented,
+
+	// Task definition.
+	"RegisterTaskDefinition":     NotImplemented,
+	"DeregisterTaskDefinition":   NotImplemented,
+	"DescribeTaskDefinition":     NotImplemented,
+	"ListTaskDefinitions":        NotImplemented,
+	"ListTaskDefinitionFamilies": NotImplemented,
+
+	// Task.
+	"RunTask":       NotImplemented,
+	"StartTask":     NotImplemented,
+	"StopTask":      NotImplemented,
+	"DescribeTasks": NotImplemented,
+	"ListTasks":     NotImplemented,
+
+	// Service. ListServicesByNamespace is a no-op stub in v1.
+	"CreateService":           NotImplemented,
+	"UpdateService":           NotImplemented,
+	"DeleteService":           NotImplemented,
+	"DescribeServices":        NotImplemented,
+	"ListServices":            NotImplemented,
+	"ListServicesByNamespace": NotImplemented,
+
+	// Container instance.
+	"RegisterContainerInstance":     NotImplemented,
+	"DeregisterContainerInstance":   NotImplemented,
+	"DescribeContainerInstances":    NotImplemented,
+	"ListContainerInstances":        NotImplemented,
+	"UpdateContainerInstancesState": NotImplemented,
+
+	// Account settings (passthrough; no enforcement v1).
+	"PutAccountSetting":   NotImplemented,
+	"ListAccountSettings": NotImplemented,
+
+	// Tags.
+	"TagResource":         NotImplemented,
+	"UntagResource":       NotImplemented,
+	"ListTagsForResource": NotImplemented,
+}
+
+// WriteJSONResponse serialises obj as a 200 AWS JSON 1.1 response using the
+// SDK's jsonutil marshaler (epoch-second time encoding), matching ECR/ACM/EKS.
+func WriteJSONResponse(w http.ResponseWriter, obj any) {
+	body, err := jsonutil.BuildJSON(obj)
+	if err != nil {
+		slog.Error("ECS: failed to marshal response JSON", "err", err)
+		http.Error(w, awserrors.ErrorInternalError, http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Type", JSONContentType)
+	w.WriteHeader(http.StatusOK)
+	if _, err := w.Write(body); err != nil {
+		slog.Error("ECS: failed to write response", "err", err)
+	}
+}

spinifex/gateway/ecs/handler_test.go

@@ -0,0 +1,40 @@
+package gateway_ecs
+
+import (
+	"encoding/json"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNotImplemented(t *testing.T) {
+	out, err := NotImplemented(nil, "123456789012", []byte("{}"))
+	assert.Nil(t, out)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorNotImplemented, err.Error())
+}
+
+// Every registered action is the NotImplemented stub in this sprint.
+func TestActions_AllNotImplemented(t *testing.T) {
+	require.NotEmpty(t, Actions)
+	for action, h := range Actions {
+		_, err := h(nil, "123456789012", []byte("{}"))
+		require.Error(t, err, "action %q", action)
+		assert.Equal(t, awserrors.ErrorNotImplemented, err.Error(), "action %q", action)
+	}
+}
+
+func TestWriteJSONResponse(t *testing.T) {
+	w := httptest.NewRecorder()
+	WriteJSONResponse(w, map[string]string{"clusterArn": "arn:aws:ecs:::cluster/x"})
+
+	assert.Equal(t, 200, w.Code)
+	assert.Equal(t, JSONContentType, w.Header().Get("Content-Type"))
+
+	var got map[string]string
+	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &got))
+	assert.Equal(t, "arn:aws:ecs:::cluster/x", got["clusterArn"])
+}

spinifex/gateway/ecs_test.go

@@ -0,0 +1,100 @@
+package gateway
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	gateway_ecs "github.com/mulgadc/spinifex/spinifex/gateway/ecs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func setupECSRequest(target, body string) *http.Request {
+	req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
+	if target != "" {
+		req.Header.Set("X-Amz-Target", target)
+	}
+	ctx := context.WithValue(req.Context(), ctxService, "ecs")
+	ctx = context.WithValue(ctx, ctxAccountID, "123456789012")
+	return req.WithContext(ctx)
+}
+
+func TestECSActionFromTarget(t *testing.T) {
+	assert.Equal(t, "RunTask",
+		ecsActionFromTarget("AmazonEC2ContainerServiceV20141113.RunTask"))
+	assert.Equal(t, "CreateCluster", ecsActionFromTarget("CreateCluster"))
+	assert.Equal(t, "", ecsActionFromTarget(""))
+}
+
+// The Actions map is the v1 API contract (ecs-v1.md §1): every action across the
+// Cluster / TaskDefinition / Task / Service / ContainerInstance / Account / Tags
+// surfaces must be registered, even though each is a 501 stub in this sprint.
+func TestECSActionsMap_NamespaceRegistered(t *testing.T) {
+	namespace := []string{
+		// Cluster
+		"CreateCluster", "DescribeClusters", "ListClusters", "DeleteCluster",
+		"UpdateCluster", "PutClusterCapacityProviders",
+		// Task definition
+		"RegisterTaskDefinition", "DeregisterTaskDefinition", "DescribeTaskDefinition",
+		"ListTaskDefinitions", "ListTaskDefinitionFamilies",
+		// Task
+		"RunTask", "StartTask", "StopTask", "DescribeTasks", "ListTasks",
+		// Service
+		"CreateService", "UpdateService", "DeleteService", "DescribeServices",
+		"ListServices", "ListServicesByNamespace",
+		// Container instance
+		"RegisterContainerInstance", "DeregisterContainerInstance",
+		"DescribeContainerInstances", "ListContainerInstances",
+		"UpdateContainerInstancesState",
+		// Account
+		"PutAccountSetting", "ListAccountSettings",
+		// Tags
+		"TagResource", "UntagResource", "ListTagsForResource",
+	}
+	for _, action := range namespace {
+		_, ok := gateway_ecs.Actions[action]
+		assert.True(t, ok, "action %q should be registered", action)
+	}
+}
+
+func TestECSRequest_MissingTarget(t *testing.T) {
+	gw := &GatewayConfig{DisableLogging: true}
+	err := gw.ECS_Request(httptest.NewRecorder(), setupECSRequest("", ""))
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorMissingAction, err.Error())
+}
+
+func TestECSRequest_UnknownAction(t *testing.T) {
+	gw := &GatewayConfig{DisableLogging: true}
+	err := gw.ECS_Request(httptest.NewRecorder(),
+		setupECSRequest("AmazonEC2ContainerServiceV20141113.MadeUpAction", "{}"))
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidAction, err.Error())
+}
+
+// Every registered action resolves to the 501 stub until its real handler lands
+// in a later Phase 4 sprint.
+func TestECSRequest_KnownActionNotImplemented(t *testing.T) {
+	gw := &GatewayConfig{DisableLogging: true}
+	err := gw.ECS_Request(httptest.NewRecorder(),
+		setupECSRequest("AmazonEC2ContainerServiceV20141113.RunTask", "{}"))
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorNotImplemented, err.Error())
+}
+
+// A request that clears auth but carries no account ID in context is an internal
+// fault, not a client error.
+func TestECSRequest_MissingAccountID(t *testing.T) {
+	req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("{}"))
+	req.Header.Set("X-Amz-Target", "AmazonEC2ContainerServiceV20141113.ListClusters")
+	req = req.WithContext(context.WithValue(req.Context(), ctxService, "ecs"))
+
+	gw := &GatewayConfig{DisableLogging: true}
+	err := gw.ECS_Request(httptest.NewRecorder(), req)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+}

spinifex/gateway/gateway.go

@@ -110,6 +110,7 @@ var supportedServices = map[string]bool{
 	"account":              true,
 	"elasticloadbalancing": true,
 	"eks":                  true,
+	"ecs":                  true,
 	"ecr":                  true,
 	"acm":                  true,
 	"tagging":              true,
@@ -335,6 +336,8 @@ func (gw *GatewayConfig) Request(w http.ResponseWriter, r *http.Request) {
 		err = gw.ELBv2_Request(w, r)
 	case "eks":
 		err = gw.EKS_Request(w, r)
+	case "ecs":
+		err = gw.ECS_Request(w, r)
 	case "ecr":
 		err = gw.ECR_Request(w, r)
 	case "acm":

spinifex/handlers/NATS_HANDLER_NAMING.md

@@ -222,3 +222,56 @@ JetStream KV compare-and-swap, so a concurrent PATCH that loses the revision
 race is rejected rather than silently merged. Absent records and CAS conflicts
 travel back as response flags (`found`, `conflict`), not transport errors, so
 they round-trip a reply envelope that otherwise only carries AWS error codes.
+
+## ECS (Elastic Container Service)
+
+ECS uses two layers of NATS subjects (ecs-v1.md Q14). The `.bus.` segment
+distinguishes internal scheduler↔agent traffic from the AWS API surface.
+
+### Layer 1 — AWS API surface
+
+The gateway translates ECS JSON 1.1 requests (X-Amz-Target
+`AmazonEC2ContainerServiceV20141113.<Action>`) into `ecs.<AWSAction>` NATS
+requests, using the existing `spinifex-workers` queue group. Handler names
+follow the `handleECS<AWSAction>` convention. The full v1 namespace:
+
+```
+ecs.CreateCluster, ecs.DescribeClusters, ecs.ListClusters, ecs.DeleteCluster,
+ecs.UpdateCluster, ecs.PutClusterCapacityProviders
+ecs.RegisterTaskDefinition, ecs.DeregisterTaskDefinition,
+ecs.DescribeTaskDefinition, ecs.ListTaskDefinitions, ecs.ListTaskDefinitionFamilies
+ecs.RunTask, ecs.StartTask, ecs.StopTask, ecs.DescribeTasks, ecs.ListTasks
+ecs.CreateService, ecs.UpdateService, ecs.DeleteService, ecs.DescribeServices,
+ecs.ListServices, ecs.ListServicesByNamespace
+ecs.RegisterContainerInstance, ecs.DeregisterContainerInstance,
+ecs.DescribeContainerInstances, ecs.ListContainerInstances,
+ecs.UpdateContainerInstancesState
+ecs.PutAccountSetting, ecs.ListAccountSettings
+ecs.TagResource, ecs.UntagResource, ecs.ListTagsForResource
+```
+
+Today every action is a 501 stub; the subjects land with their handler bodies.
+
+### Layer 2 — internal scheduler↔agent bus
+
+The per-cluster scheduler and the on-VM ecs-agent communicate on
+`ecs.bus.<accountID>.<clusterName>.*`. Cluster identity is
+`<accountID>.<clusterName>` (matches the KV layout), not a UUID.
+
+```
+ecs.bus.<accountID>.<clusterName>.assign.<instanceID>             # scheduler → agent
+ecs.bus.<accountID>.<clusterName>.task-state.<taskID>             # agent → scheduler
+ecs.bus.<accountID>.<clusterName>.instance-heartbeat.<instanceID> # agent → scheduler
+ecs.bus.<accountID>.<clusterName>.instance-register.<instanceID>  # agent → scheduler
+ecs.bus.<accountID>.<clusterName>.service-reconcile               # scheduler-internal tick
+```
+
+This layer lands with the scheduler + agent bodies. The Phase 5 per-AZ NATS
+cutover inserts `{azID}` after `ecs.` → `ecs.<azID>.bus.<accountID>.<clusterName>.*`.
+
+### Cross-service calls
+
+The scheduler also publishes on existing namespaces (no `ecs.` prefix):
+`ec2.CreateNetworkInterface`, `ec2.AttachNetworkInterface`, `sts.AssumeRole`,
+`elbv2.RegisterTargets`, `elbv2.DeregisterTargets`,
+`route53.ChangeResourceRecordSets`.

spinifex/handlers/ecs/store.go

@@ -0,0 +1,172 @@
+package handlers_ecs
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/migrate"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+)
+
+// JetStream KV bucket and key-path constants for the ECS control plane.
+//
+// Per-account bucket "ecs-account-{accountID}" holds all customer-visible
+// cluster, task-definition, task, service, and container-instance state. It is
+// created lazily on first cluster touch (no daemon-boot pre-creation) so accounts
+// without any ECS clusters never grow a bucket (ecs-v1.md Q2).
+//
+// Shared leader bucket "spinifex-ecs-leader" holds 60s-TTL CAS locks keyed by
+// "{accountID}/{clusterName}" for the per-cluster scheduler leader election
+// (Q3). Created once at daemon boot.
+const (
+	KVBucketECSAccountPrefix  = "ecs-account-"
+	KVBucketECSAccountVersion = 1
+	KVBucketECSAccountHistory = 1
+
+	KVBucketECSLeader        = "spinifex-ecs-leader"
+	KVBucketECSLeaderVersion = 1
+	KVBucketECSLeaderTTL     = 60 * time.Second
+)
+
+// Key-path helpers for the per-account bucket. The layout matches ecs-v1.md Q2.
+// Task definitions are account-scoped (not cluster-scoped) per the AWS ARN shape:
+//
+//	clusters/{name}/meta
+//	clusters/{name}/instances/{instanceID}
+//	clusters/{name}/tasks/{taskID}
+//	clusters/{name}/services/{serviceName}
+//	taskdef-families/{family}/latest-rev
+//	taskdef-families/{family}/revs/{rev}
+
+// ClusterMetaKey returns the KV key for a cluster's meta record.
+func ClusterMetaKey(name string) string {
+	return fmt.Sprintf("clusters/%s/meta", name)
+}
+
+// InstancesPrefix returns the KV key prefix under which all of a cluster's
+// container-instance records live. Used by ListContainerInstances to enumerate.
+func InstancesPrefix(cluster string) string {
+	return fmt.Sprintf("clusters/%s/instances/", cluster)
+}
+
+// InstanceKey returns the KV key for a container-instance record under a cluster.
+func InstanceKey(cluster, instanceID string) string {
+	return InstancesPrefix(cluster) + instanceID
+}
+
+// TasksPrefix returns the KV key prefix under which all of a cluster's task
+// records live. Used by ListTasks and the agent reboot reconciler to enumerate.
+func TasksPrefix(cluster string) string {
+	return fmt.Sprintf("clusters/%s/tasks/", cluster)
+}
+
+// TaskKey returns the KV key for a task record under a cluster.
+func TaskKey(cluster, taskID string) string {
+	return TasksPrefix(cluster) + taskID
+}
+
+// ServicesPrefix returns the KV key prefix under which all of a cluster's service
+// records live. Used by ListServices to enumerate.
+func ServicesPrefix(cluster string) string {
+	return fmt.Sprintf("clusters/%s/services/", cluster)
+}
+
+// ServiceKey returns the KV key for a service record under a cluster.
+func ServiceKey(cluster, serviceName string) string {
+	return ServicesPrefix(cluster) + serviceName
+}
+
+// TaskDefFamiliesPrefix returns the KV key prefix under which all task-definition
+// families live. Task definitions are account-scoped, not cluster-scoped (Q2).
+func TaskDefFamiliesPrefix() string {
+	return "taskdef-families/"
+}
+
+// TaskDefLatestRevKey returns the KV key holding a family's latest revision
+// number. Read-modify-written on RegisterTaskDefinition.
+func TaskDefLatestRevKey(family string) string {
+	return fmt.Sprintf("taskdef-families/%s/latest-rev", family)
+}
+
+// TaskDefRevsPrefix returns the KV key prefix under which all revisions of a
+// task-definition family live. Used by ListTaskDefinitions to enumerate.
+func TaskDefRevsPrefix(family string) string {
+	return fmt.Sprintf("taskdef-families/%s/revs/", family)
+}
+
+// TaskDefRevKey returns the KV key for a specific revision of a task-definition
+// family.
+func TaskDefRevKey(family string, rev int) string {
+	return fmt.Sprintf("%s%d", TaskDefRevsPrefix(family), rev)
+}
+
+// LeaderLeaseKey returns the per-cluster scheduler leader-lease key in the shared
+// leader bucket (Q3).
+func LeaderLeaseKey(accountID, clusterName string) string {
+	return fmt.Sprintf("%s/%s", accountID, clusterName)
+}
+
+// Store is the per-daemon ECS KV handle. Per-account and leader buckets are
+// accessed via the package-level factories below.
+type Store struct {
+	nc *nats.Conn
+}
+
+// NewStore constructs a Store bound to the supplied NATS connection. It does not
+// touch JetStream — per-account buckets are created lazily by
+// GetOrCreateAccountBucket and the leader bucket by InitLeaderBucket.
+func NewStore(nc *nats.Conn) (*Store, error) {
+	if nc == nil {
+		return nil, errors.New("ecs store: nats connection is nil")
+	}
+	return &Store{nc: nc}, nil
+}
+
+// AccountBucketName returns the per-account JetStream KV bucket name for the
+// given AWS account ID.
+func AccountBucketName(accountID string) string {
+	return KVBucketECSAccountPrefix + accountID
+}
+
+// GetOrCreateAccountBucket returns the per-account KV bucket for accountID,
+// creating it on first use. Idempotent: subsequent calls with the same accountID
+// return the existing handle.
+func GetOrCreateAccountBucket(js nats.JetStreamContext, accountID string) (nats.KeyValue, error) {
+	bucket := AccountBucketName(accountID)
+	kv, err := utils.GetOrCreateKVBucket(js, bucket, KVBucketECSAccountHistory)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create ECS per-account KV bucket %s: %w", bucket, err)
+	}
+	if err := migrate.DefaultRegistry.RunKV(bucket, kv, KVBucketECSAccountVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", bucket, err)
+	}
+	return kv, nil
+}
+
+// InitLeaderBucket creates (or attaches to) the shared spinifex-ecs-leader bucket
+// used for per-cluster scheduler leader-lease CAS locks. The bucket is configured
+// with History=1 and a 60s TTL so stale leases expire on their own when a leader
+// dies mid-cycle. utils.GetOrCreateKVBucket doesn't expose a TTL knob, so this
+// takes the direct js.CreateKeyValue path and falls back to js.KeyValue on
+// already-exists.
+func InitLeaderBucket(js nats.JetStreamContext) (nats.KeyValue, error) {
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
+		Bucket:  KVBucketECSLeader,
+		History: 1,
+		TTL:     KVBucketECSLeaderTTL,
+	})
+	if err != nil {
+		kv, err = js.KeyValue(KVBucketECSLeader)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create or open ECS leader bucket %s: %w", KVBucketECSLeader, err)
+		}
+	}
+	if err := migrate.DefaultRegistry.RunKV(KVBucketECSLeader, kv, KVBucketECSLeaderVersion); err != nil {
+		return nil, fmt.Errorf("migrate %s: %w", KVBucketECSLeader, err)
+	}
+	slog.Info("ECS leader bucket initialized", "bucket", KVBucketECSLeader, "ttl", KVBucketECSLeaderTTL)
+	return kv, nil
+}

spinifex/handlers/ecs/store_test.go

@@ -0,0 +1,87 @@
+package handlers_ecs
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const testAccountID = "123456789012"
+
+func TestAccountBucketName(t *testing.T) {
+	assert.Equal(t, "ecs-account-123456789012", AccountBucketName(testAccountID))
+}
+
+func TestNewStore_NilConn(t *testing.T) {
+	_, err := NewStore(nil)
+	require.Error(t, err)
+}
+
+func TestNewStore_Valid(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	s, err := NewStore(nc)
+	require.NoError(t, err)
+	require.NotNil(t, s)
+}
+
+func TestGetOrCreateAccountBucket_Idempotent(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+
+	kv1, err := GetOrCreateAccountBucket(js, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, kv1)
+
+	kv2, err := GetOrCreateAccountBucket(js, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, kv2)
+
+	assert.Equal(t, AccountBucketName(testAccountID), kv1.Bucket())
+	assert.Equal(t, kv1.Bucket(), kv2.Bucket())
+}
+
+func TestInitLeaderBucket_Idempotent(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+
+	kv1, err := InitLeaderBucket(js)
+	require.NoError(t, err)
+	require.NotNil(t, kv1)
+	assert.Equal(t, KVBucketECSLeader, kv1.Bucket())
+
+	kv2, err := InitLeaderBucket(js)
+	require.NoError(t, err)
+	require.NotNil(t, kv2)
+	assert.Equal(t, KVBucketECSLeader, kv2.Bucket())
+}
+
+// Key-path helpers must produce the ecs-v1.md Q2 layout exactly: prefixes are
+// what the List* enumerations watch, so a drift here silently breaks listing.
+func TestKeyPaths(t *testing.T) {
+	assert.Equal(t, "clusters/web/meta", ClusterMetaKey("web"))
+
+	assert.Equal(t, "clusters/web/instances/", InstancesPrefix("web"))
+	assert.Equal(t, "clusters/web/instances/i-abc", InstanceKey("web", "i-abc"))
+
+	assert.Equal(t, "clusters/web/tasks/", TasksPrefix("web"))
+	assert.Equal(t, "clusters/web/tasks/t-abc", TaskKey("web", "t-abc"))
+
+	assert.Equal(t, "clusters/web/services/", ServicesPrefix("web"))
+	assert.Equal(t, "clusters/web/services/api", ServiceKey("web", "api"))
+
+	assert.Equal(t, "taskdef-families/", TaskDefFamiliesPrefix())
+	assert.Equal(t, "taskdef-families/nginx/latest-rev", TaskDefLatestRevKey("nginx"))
+	assert.Equal(t, "taskdef-families/nginx/revs/", TaskDefRevsPrefix("nginx"))
+	assert.Equal(t, "taskdef-families/nginx/revs/3", TaskDefRevKey("nginx", 3))
+
+	assert.Equal(t, "123456789012/web", LeaderLeaseKey("123456789012", "web"))
+}
+
+// Prefix helpers must be a true prefix of their per-record key so a KV
+// prefix-watch over the prefix sees the record key.
+func TestPrefixContainment(t *testing.T) {
+	assert.Contains(t, InstanceKey("c", "x"), InstancesPrefix("c"))
+	assert.Contains(t, TaskKey("c", "x"), TasksPrefix("c"))
+	assert.Contains(t, ServiceKey("c", "x"), ServicesPrefix("c"))
+	assert.Contains(t, TaskDefRevKey("f", 1), TaskDefRevsPrefix("f"))
+}