feat: ECS scheduler, RunTask placement, and agent task execution (#416)

Add the ECS v1 scheduler control loop and end-to-end RunTask flow:

- handlers/ecs: per-account KV-backed Service (clusters, task definitions,
  container instances, tasks), bin-pack/spread/random placement with CAS
  capacity reservation, and a leader-elected Scheduler that owns the Layer-2
  bus subscriptions (register/heartbeat/task-state) plus a heartbeat reaper.
- gateway/ecs: wire CreateCluster, Describe/ListClusters,
  Register/Describe/ListTaskDefinitions, Register/Describe/ListContainerInstances,
  RunTask, Describe/ListTasks onto the daemon over NATS.
- daemon: construct the Service, start the Scheduler goroutine, subscribe the
  ECS action subjects.
- ecs-agent: subscribe the per-instance assign subject, pull and run each
  container via containerd, and report task state back on the bus.

cmd/ecs-agent/agent.go

@@ -25,12 +25,14 @@ var version = "dev"
 
 // Agent wires the ecs-agent's runtime seams: a NATS publisher for the Layer-2
 // bus, a container runtime, and an ECR resolver. It registers the host as a
-// container instance at boot and heartbeats while alive. Task assignment and
-// container lifecycle land in Sprint 4e.
+// container instance at boot, heartbeats while alive, and (Sprint 4e) runs
+// assigned tasks through containerd, reporting their state back on the bus.
 type Agent struct {
 	cfg      config
 	id       identity
+	pub      publisher
 	puller   ctrruntime.ImagePuller
+	runner   ctrruntime.Runner
 	resolver ctrruntime.Resolver
 
 	reg *registrar
@@ -41,12 +43,16 @@ type Agent struct {
 }
 
 // newAgent assembles an Agent from already-built seams. Tests use this directly
-// with fakes; New builds the production seams and delegates here.
-func newAgent(cfg config, id identity, pub publisher, puller ctrruntime.ImagePuller, resolver ctrruntime.Resolver) *Agent {
+// with fakes; New builds the production seams and delegates here. runner may be
+// nil when containerd is unavailable; the assign path then reports the task
+// STOPPED rather than crashing.
+func newAgent(cfg config, id identity, pub publisher, puller ctrruntime.ImagePuller, runner ctrruntime.Runner, resolver ctrruntime.Resolver) *Agent {
 	return &Agent{
 		cfg:      cfg,
 		id:       id,
+		pub:      pub,
 		puller:   puller,
+		runner:   runner,
 		resolver: resolver,
 		reg:      newRegistrar(pub, id),
 		hb:       newHeartbeater(pub, id, cfg.Heartbeat, nil),
@@ -81,10 +87,12 @@ func New(cfg config) (*Agent, error) {
 	resolver := newLazyECRResolver(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA)
 
 	var puller ctrruntime.ImagePuller
-	if p, perr := ctrruntime.New(cfg.ContainerdSocket); perr != nil {
+	var runner ctrruntime.Runner
+	if rt, perr := ctrruntime.New(cfg.ContainerdSocket); perr != nil {
 		slog.Warn("ecs-agent: containerd unavailable at boot, image pulls disabled", "err", perr)
 	} else {
-		puller = p
+		puller = rt
+		runner = rt
 	}
 
 	nc, err := nats.Connect(cfg.NATSURL, nats.Name("ecs-agent/"+id.InstanceID))
@@ -92,7 +100,7 @@ func New(cfg config) (*Agent, error) {
 		return nil, fmt.Errorf("connect nats %s: %w", cfg.NATSURL, err)
 	}
 
-	a := newAgent(cfg, id, nc, puller, resolver)
+	a := newAgent(cfg, id, nc, puller, runner, resolver)
 	a.nc = nc
 	if puller != nil {
 		a.closers = append(a.closers, puller.Close)
@@ -113,6 +121,10 @@ func (a *Agent) Run(ctx context.Context) error {
 
 	go a.hb.Run(ctx)
 
+	if err := a.subscribeAssign(ctx); err != nil {
+		slog.Error("ecs-agent: assign subscription failed; task assignment disabled", "err", err)
+	}
+
 	<-ctx.Done()
 	return a.Stop()
 }

cmd/ecs-agent/agent_test.go

@@ -14,7 +14,7 @@ func TestAgent_RunRegistersThenStopsOnContext(t *testing.T) {
 	cfg := config{Heartbeat: 5 * time.Millisecond}
 	id := testIdentity()
 	puller := &ctrruntime.FakePuller{}
-	a := newAgent(cfg, id, pub, puller, nil)
+	a := newAgent(cfg, id, pub, puller, puller, nil)
 	a.closers = append(a.closers, puller.Close)
 
 	ctx, cancel := context.WithCancel(context.Background())

cmd/ecs-agent/runtime/containerd.go

@@ -31,7 +31,7 @@ var _ ImagePuller = (*containerdPuller)(nil)
 // returns an error if it is unreachable — otherwise a dead socket would only
 // surface on the first pull, and the agent's boot-time "containerd unavailable"
 // log would never fire.
-func New(socket string) (ImagePuller, error) {
+func New(socket string) (Runtime, error) {
 	client, err := containerd.New(socket, containerd.WithDefaultNamespace(ecsNamespace))
 	if err != nil {
 		return nil, fmt.Errorf("connect containerd %s: %w", socket, err)

cmd/ecs-agent/runtime/containerd_run.go

@@ -0,0 +1,105 @@
+package runtime
+
+import (
+	"context"
+	"fmt"
+	"sort"
+
+	containerd "github.com/containerd/containerd/v2/client"
+	"github.com/containerd/containerd/v2/pkg/cio"
+	"github.com/containerd/containerd/v2/pkg/namespaces"
+	"github.com/containerd/containerd/v2/pkg/oci"
+	specs "github.com/opencontainers/runtime-spec/specs-go"
+)
+
+var _ Runner = (*containerdPuller)(nil)
+
+// Run creates a container from an already-pulled image and starts its task. The
+// container joins the host network namespace (bridge/CNI task networking is a
+// later sprint) and carries spec.Labels for the reboot reconciler. id must be
+// unique within the "ecs" namespace.
+func (p *containerdPuller) Run(ctx context.Context, id string, spec RunSpec) (string, error) {
+	ctx = namespaces.WithNamespace(ctx, ecsNamespace)
+
+	image, err := p.client.GetImage(ctx, spec.Image)
+	if err != nil {
+		return "", fmt.Errorf("get image %s: %w", spec.Image, err)
+	}
+
+	specOpts := []oci.SpecOpts{
+		oci.WithImageConfig(image),
+		oci.WithHostNamespace(specs.NetworkNamespace),
+		oci.WithHostHostsFile,
+		oci.WithHostResolvconf,
+	}
+	if len(spec.Command) > 0 {
+		specOpts = append(specOpts, oci.WithProcessArgs(spec.Command...))
+	}
+	if len(spec.Env) > 0 {
+		specOpts = append(specOpts, oci.WithEnv(envSlice(spec.Env)))
+	}
+
+	container, err := p.client.NewContainer(ctx, id,
+		containerd.WithNewSnapshot(id+"-snapshot", image),
+		containerd.WithNewSpec(specOpts...),
+		containerd.WithContainerLabels(spec.Labels),
+	)
+	if err != nil {
+		return "", fmt.Errorf("create container %s: %w", id, err)
+	}
+
+	task, err := container.NewTask(ctx, cio.NullIO)
+	if err != nil {
+		_ = container.Delete(ctx, containerd.WithSnapshotCleanup)
+		return "", fmt.Errorf("create task %s: %w", id, err)
+	}
+	if err := task.Start(ctx); err != nil {
+		_, _ = task.Delete(ctx)
+		_ = container.Delete(ctx, containerd.WithSnapshotCleanup)
+		return "", fmt.Errorf("start task %s: %w", id, err)
+	}
+	return id, nil
+}
+
+// Wait blocks until the container's task exits and returns its exit code.
+func (p *containerdPuller) Wait(ctx context.Context, containerID string) (RunStatus, error) {
+	ctx = namespaces.WithNamespace(ctx, ecsNamespace)
+	container, err := p.client.LoadContainer(ctx, containerID)
+	if err != nil {
+		return RunStatus{}, fmt.Errorf("load container %s: %w", containerID, err)
+	}
+	task, err := container.Task(ctx, nil)
+	if err != nil {
+		return RunStatus{}, fmt.Errorf("attach task %s: %w", containerID, err)
+	}
+	statusC, err := task.Wait(ctx)
+	if err != nil {
+		return RunStatus{}, fmt.Errorf("wait task %s: %w", containerID, err)
+	}
+	status := <-statusC
+	code, _, err := status.Result()
+	return RunStatus{ExitCode: int(code)}, err
+}
+
+// Remove kills and deletes the container's task, then the container + snapshot.
+func (p *containerdPuller) Remove(ctx context.Context, containerID string) error {
+	ctx = namespaces.WithNamespace(ctx, ecsNamespace)
+	container, err := p.client.LoadContainer(ctx, containerID)
+	if err != nil {
+		return nil //nolint:nilerr // load failure means the container is already gone
+	}
+	if task, terr := container.Task(ctx, nil); terr == nil {
+		_, _ = task.Delete(ctx, containerd.WithProcessKill)
+	}
+	return container.Delete(ctx, containerd.WithSnapshotCleanup)
+}
+
+// envSlice renders an env map as sorted "K=V" entries for deterministic specs.
+func envSlice(env map[string]string) []string {
+	out := make([]string, 0, len(env))
+	for k, v := range env {
+		out = append(out, k+"="+v)
+	}
+	sort.Strings(out)
+	return out
+}

cmd/ecs-agent/runtime/runtime.go

@@ -34,3 +34,34 @@ type ImagePuller interface {
 	Pull(ctx context.Context, spec PullSpec, r Resolver) (Image, error)
 	Close() error
 }
+
+// RunSpec describes a single container the agent must create and start. v1 runs
+// the container in the host network namespace (bridge/CNI task networking lands
+// in a later sprint); Labels carry the mulga.ecs.* task identity for the reboot
+// reconciler.
+type RunSpec struct {
+	Image   string
+	Command []string
+	Env     map[string]string
+	Labels  map[string]string
+}
+
+// RunStatus is a finished container's outcome.
+type RunStatus struct {
+	ExitCode int
+}
+
+// Runner creates and starts containers from already-pulled images. id is a
+// caller-unique container ID; Wait blocks until the container exits; Remove
+// tears down the container + its task.
+type Runner interface {
+	Run(ctx context.Context, id string, spec RunSpec) (containerID string, err error)
+	Wait(ctx context.Context, containerID string) (RunStatus, error)
+	Remove(ctx context.Context, containerID string) error
+}
+
+// Runtime is the full container runtime the ecs-agent drives: pull + run.
+type Runtime interface {
+	ImagePuller
+	Runner
+}

cmd/ecs-agent/runtime/runtime_fake.go

@@ -16,9 +16,20 @@ type FakePuller struct {
 	Err    error
 	Closed bool
 	OnPull func(PullSpec) (Image, error)
+
+	// Run bookkeeping.
+	Runs     []RunSpec
+	RunErr   error
+	WaitCode int
+	WaitErr  error
+	Removed  []string
+	OnRun    func(string, RunSpec) (string, error)
 }
 
-var _ ImagePuller = (*FakePuller)(nil)
+var (
+	_ ImagePuller = (*FakePuller)(nil)
+	_ Runner      = (*FakePuller)(nil)
+)
 
 // Pull records the spec, exercises the resolver, and returns the programmed
 // result. If OnPull is set it takes precedence over Result/Err.
@@ -57,3 +68,30 @@ func (f *FakePuller) Close() error {
 	f.Closed = true
 	return nil
 }
+
+// Run records the spec and returns id (or the programmed result/error).
+func (f *FakePuller) Run(_ context.Context, id string, spec RunSpec) (string, error) {
+	f.mu.Lock()
+	f.Runs = append(f.Runs, spec)
+	f.mu.Unlock()
+	if f.OnRun != nil {
+		return f.OnRun(id, spec)
+	}
+	if f.RunErr != nil {
+		return "", f.RunErr
+	}
+	return id, nil
+}
+
+// Wait returns the programmed exit code/error.
+func (f *FakePuller) Wait(_ context.Context, _ string) (RunStatus, error) {
+	return RunStatus{ExitCode: f.WaitCode}, f.WaitErr
+}
+
+// Remove records the container ID.
+func (f *FakePuller) Remove(_ context.Context, containerID string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.Removed = append(f.Removed, containerID)
+	return nil
+}

cmd/ecs-agent/taskrun.go

@@ -0,0 +1,132 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+// subscribeAssign wires the per-instance assign subject to the task runner. It is
+// a no-op when the agent has no live NATS connection (unit tests drive runTask
+// directly).
+func (a *Agent) subscribeAssign(ctx context.Context) error {
+	if a.nc == nil {
+		return nil
+	}
+	subj := bus.AssignSubject(a.id.AccountID, a.id.ClusterName, a.id.InstanceID)
+	sub, err := a.nc.Subscribe(subj, func(msg *nats.Msg) {
+		var as bus.Assign
+		if err := json.Unmarshal(msg.Data, &as); err != nil {
+			slog.Error("ecs-agent: bad assign payload", "err", err)
+			return
+		}
+		slog.Info("ecs-agent: task assigned", "task", as.TaskID, "containers", len(as.Containers))
+		go a.runTask(ctx, &as)
+	})
+	if err != nil {
+		return fmt.Errorf("subscribe %s: %w", subj, err)
+	}
+	a.closers = append(a.closers, sub.Unsubscribe)
+	slog.Info("ecs-agent: assign subscription active", "subject", subj)
+	return nil
+}
+
+// runTask pulls each container image, starts the containers, and reports task
+// state on the bus. A missing runtime or any per-container failure reports the
+// task STOPPED with a reason; success reports RUNNING and waits for exit.
+func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
+	if a.puller == nil || a.runner == nil {
+		a.reportTaskState(as, bus.TaskStatusStopped, "containerd unavailable on agent", nil)
+		return
+	}
+
+	statuses := make([]bus.ContainerStatus, 0, len(as.Containers))
+	for _, c := range as.Containers {
+		if _, err := a.puller.Pull(ctx, ctrruntime.PullSpec{Ref: c.Image}, a.resolver); err != nil {
+			slog.Error("ecs-agent: pull failed", "task", as.TaskID, "image", c.Image, "err", err)
+			a.reportTaskState(as, bus.TaskStatusStopped, "image pull failed: "+err.Error(), statuses)
+			return
+		}
+
+		cid := containerID(as.TaskID, c.Name)
+		spec := ctrruntime.RunSpec{
+			Image:   c.Image,
+			Command: c.Command,
+			Env:     c.Environment,
+			Labels:  taskLabels(as, c.Name),
+		}
+		id, err := a.runner.Run(ctx, cid, spec)
+		if err != nil {
+			slog.Error("ecs-agent: run failed", "task", as.TaskID, "container", c.Name, "err", err)
+			a.reportTaskState(as, bus.TaskStatusStopped, "container start failed: "+err.Error(), statuses)
+			return
+		}
+		statuses = append(statuses, bus.ContainerStatus{Name: c.Name, Status: bus.TaskStatusRunning, ContainerID: id})
+		go a.waitContainer(ctx, as, c.Name, id)
+	}
+
+	a.reportTaskState(as, bus.TaskStatusRunning, "", statuses)
+	slog.Info("ecs-agent: task running", "task", as.TaskID, "containers", len(statuses))
+}
+
+// waitContainer blocks until a container exits, then reports the task STOPPED
+// with the exit code. v1 stops the whole task when its first container exits.
+func (a *Agent) waitContainer(ctx context.Context, as *bus.Assign, name, containerID string) {
+	status, err := a.runner.Wait(ctx, containerID)
+	if err != nil {
+		slog.Warn("ecs-agent: wait container failed", "task", as.TaskID, "container", name, "err", err)
+		return
+	}
+	exit := status.ExitCode
+	statuses := []bus.ContainerStatus{{Name: name, Status: bus.TaskStatusStopped, ContainerID: containerID, ExitCode: &exit}}
+	a.reportTaskState(as, bus.TaskStatusStopped, fmt.Sprintf("container %s exited (%d)", name, exit), statuses)
+	slog.Info("ecs-agent: task stopped", "task", as.TaskID, "container", name, "exitCode", exit)
+}
+
+// reportTaskState publishes a TaskState message on the task's state subject.
+func (a *Agent) reportTaskState(as *bus.Assign, status, reason string, containers []bus.ContainerStatus) {
+	if a.pub == nil {
+		return
+	}
+	msg := bus.TaskState{
+		AccountID:   a.id.AccountID,
+		ClusterName: a.id.ClusterName,
+		InstanceID:  a.id.InstanceID,
+		TaskID:      as.TaskID,
+		LastStatus:  status,
+		Containers:  containers,
+		Reason:      reason,
+		ReportedAt:  time.Now().UTC(),
+	}
+	data, err := json.Marshal(&msg)
+	if err != nil {
+		slog.Error("ecs-agent: marshal task-state failed", "task", as.TaskID, "err", err)
+		return
+	}
+	subj := bus.TaskStateSubject(a.id.AccountID, a.id.ClusterName, as.TaskID)
+	if err := a.pub.Publish(subj, data); err != nil {
+		slog.Error("ecs-agent: publish task-state failed", "task", as.TaskID, "subject", subj, "err", err)
+	}
+}
+
+// containerID composes a containerd-valid container ID from a task + container.
+func containerID(taskID, name string) string {
+	return fmt.Sprintf("%s-%s", taskID, name)
+}
+
+// taskLabels are the mulga.ecs.* labels stamped on every container so the reboot
+// reconciler can re-associate running containers with their task on restart.
+func taskLabels(as *bus.Assign, name string) map[string]string {
+	return map[string]string{
+		"mulga.ecs.taskID":        as.TaskID,
+		"mulga.ecs.containerName": name,
+		"mulga.ecs.clusterName":   as.ClusterName,
+	}
+}

cmd/ecs-agent/taskrun_test.go

@@ -0,0 +1,162 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+// historyPub records every published message in order (capturePub keeps only the
+// last per subject, which loses the RUNNING→STOPPED transition on one subject).
+type historyPub struct {
+	mu   sync.Mutex
+	msgs [][]byte
+}
+
+func (h *historyPub) Publish(_ string, data []byte) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	cp := make([]byte, len(data))
+	copy(cp, data)
+	h.msgs = append(h.msgs, cp)
+	return nil
+}
+
+func (h *historyPub) states() []bus.TaskState {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	out := make([]bus.TaskState, 0, len(h.msgs))
+	for _, raw := range h.msgs {
+		var ts bus.TaskState
+		if err := json.Unmarshal(raw, &ts); err == nil {
+			out = append(out, ts)
+		}
+	}
+	return out
+}
+
+func testAssign() *bus.Assign {
+	return &bus.Assign{
+		AccountID:   "123456789012",
+		ClusterName: "default",
+		InstanceID:  "i-abc123",
+		TaskID:      "t-001",
+		Containers: []bus.AssignContainer{
+			{Name: "web", Image: "registry/web:1", Command: []string{"/bin/true"},
+				Environment: map[string]string{"FOO": "bar"}},
+		},
+	}
+}
+
+func newRunAgent(pub publisher, rt *ctrruntime.FakePuller) *Agent {
+	return newAgent(config{}, testIdentity(), pub, rt, rt, nil)
+}
+
+// runTask with no runtime reports the task STOPPED instead of crashing.
+func TestRunTask_NoRuntimeReportsStopped(t *testing.T) {
+	pub := &historyPub{}
+	a := newAgent(config{}, testIdentity(), pub, nil, nil, nil)
+	a.runTask(context.Background(), testAssign())
+
+	st := pub.states()
+	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+		t.Fatalf("want one STOPPED state, got %+v", st)
+	}
+	if st[0].Reason == "" {
+		t.Error("STOPPED state missing reason")
+	}
+}
+
+// Happy path: pull + run reports RUNNING with the container ID; Wait blocking
+// (forced error) means no STOPPED overwrite, so we can assert the RUNNING frame.
+func TestRunTask_PullRunReportsRunning(t *testing.T) {
+	pub := &historyPub{}
+	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
+	a := newRunAgent(pub, rt)
+	as := testAssign()
+	a.runTask(context.Background(), as)
+
+	if len(rt.Pulls) != 1 || rt.Pulls[0].Ref != "registry/web:1" {
+		t.Fatalf("expected one pull of registry/web:1, got %+v", rt.Pulls)
+	}
+	if len(rt.Runs) != 1 || rt.Runs[0].Image != "registry/web:1" {
+		t.Fatalf("expected one run, got %+v", rt.Runs)
+	}
+	if rt.Runs[0].Labels["mulga.ecs.taskID"] != "t-001" {
+		t.Errorf("missing task label: %+v", rt.Runs[0].Labels)
+	}
+
+	st := pub.states()
+	if len(st) == 0 || st[len(st)-1].LastStatus != bus.TaskStatusRunning {
+		t.Fatalf("want RUNNING final state, got %+v", st)
+	}
+	if len(st[len(st)-1].Containers) != 1 || st[len(st)-1].Containers[0].ContainerID != "t-001-web" {
+		t.Errorf("RUNNING state container wrong: %+v", st[len(st)-1].Containers)
+	}
+}
+
+// A pull failure reports STOPPED and never starts the container.
+func TestRunTask_PullFailureReportsStopped(t *testing.T) {
+	pub := &historyPub{}
+	rt := &ctrruntime.FakePuller{Err: errors.New("pull denied")}
+	a := newRunAgent(pub, rt)
+	a.runTask(context.Background(), testAssign())
+
+	if len(rt.Runs) != 0 {
+		t.Errorf("container should not run after pull failure, got %+v", rt.Runs)
+	}
+	st := pub.states()
+	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+		t.Fatalf("want STOPPED, got %+v", st)
+	}
+}
+
+// A run failure reports STOPPED.
+func TestRunTask_RunFailureReportsStopped(t *testing.T) {
+	pub := &historyPub{}
+	rt := &ctrruntime.FakePuller{RunErr: errors.New("start boom")}
+	a := newRunAgent(pub, rt)
+	a.runTask(context.Background(), testAssign())
+
+	st := pub.states()
+	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+		t.Fatalf("want STOPPED, got %+v", st)
+	}
+}
+
+// On container exit, waitContainer reports STOPPED with the exit code.
+func TestRunTask_ExitReportsStoppedWithCode(t *testing.T) {
+	pub := &historyPub{}
+	rt := &ctrruntime.FakePuller{WaitCode: 7}
+	a := newRunAgent(pub, rt)
+	a.runTask(context.Background(), testAssign())
+
+	deadline := time.After(time.Second)
+	for {
+		st := pub.states()
+		if len(st) > 0 && st[len(st)-1].LastStatus == bus.TaskStatusStopped {
+			last := st[len(st)-1]
+			if len(last.Containers) != 1 || last.Containers[0].ExitCode == nil || *last.Containers[0].ExitCode != 7 {
+				t.Fatalf("want exit code 7, got %+v", last.Containers)
+			}
+			return
+		}
+		select {
+		case <-deadline:
+			t.Fatalf("no STOPPED state after exit; got %+v", pub.states())
+		case <-time.After(2 * time.Millisecond):
+		}
+	}
+}
+
+func TestContainerID(t *testing.T) {
+	if got := containerID("t-1", "web"); got != "t-1-web" {
+		t.Errorf("containerID = %q, want t-1-web", got)
+	}
+}

spinifex/awserrors/awserrors.go

@@ -371,9 +371,13 @@ var (
 	ErrorResourceCountLimitExceeded                           = "ResourceCountLimitExceeded"
 	ErrorResourceLimitExceeded                                = "ResourceLimitExceeded"
 	// ErrorResourceNotFound is the ACM not-found code. EKS must use ErrorEKSResourceNotFound — do not cross-wire.
-	ErrorResourceNotFound                               = "ResourceNotFound"
-	ErrorEKSResourceInUse                               = "ResourceInUseException"
-	ErrorEKSResourceNotFound                            = "ResourceNotFoundException"
+	ErrorResourceNotFound    = "ResourceNotFound"
+	ErrorEKSResourceInUse    = "ResourceInUseException"
+	ErrorEKSResourceNotFound = "ResourceNotFoundException"
+	// ECS JSON-1.1 exception codes. ECS clients key on the "Exception"-suffixed
+	// __type; do not cross-wire with the EC2/ACM not-found codes above.
+	ErrorECSClusterNotFound                             = "ClusterNotFoundException"
+	ErrorECSInvalidParameter                            = "InvalidParameterException"
 	ErrorRetryableError                                 = "RetryableError"
 	ErrorRouteAlreadyExists                             = "RouteAlreadyExists"
 	ErrorRouteLimitExceeded                             = "RouteLimitExceeded"
@@ -894,6 +898,8 @@ var ErrorLookup = map[string]ErrorMessage{
 	ErrorResourceNotFound:                                      {HTTPCode: 404, Message: "The specified resource was not found."},
 	ErrorEKSResourceInUse:                                      {HTTPCode: 409, Message: "A cluster with this name already exists in this account and Region. Delete the existing cluster, or choose a different name."},
 	ErrorEKSResourceNotFound:                                   {HTTPCode: 404, Message: "The specified cluster could not be found. You can view your available clusters with ListClusters. Amazon EKS clusters are Region specific."},
+	ErrorECSClusterNotFound:                                    {HTTPCode: 400, Message: "The specified cluster was not found. You can view your available clusters with ListClusters. Amazon ECS clusters are Region specific."},
+	ErrorECSInvalidParameter:                                   {HTTPCode: 400, Message: "The specified parameter is not valid. Review the available parameters for the API request."},
 	ErrorRetryableError:                                        {HTTPCode: 400, Message: "A request submitted by an AWS service on your behalf could not be completed. The requesting service might automatically retry the request."},
 	ErrorRouteAlreadyExists:                                    {HTTPCode: 409, Message: "A route for the specified CIDR block already exists in this route table."},
 	ErrorRouteLimitExceeded:                                    {HTTPCode: 400, Message: "You've reached the limit on the number of routes that you can add to a route table."},

spinifex/daemon/daemon.go

@@ -49,6 +49,7 @@ import (
 	handlers_ec2_volume "github.com/mulgadc/spinifex/spinifex/handlers/ec2/volume"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
 	handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr"
+	handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs"
 	handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks"
 	handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2"
 	"github.com/mulgadc/spinifex/spinifex/instancetypes"
@@ -149,6 +150,8 @@ type Daemon struct {
 	eipService            *handlers_ec2_eip.EIPServiceImpl
 	elbv2Service          *handlers_elbv2.ELBv2ServiceImpl
 	eksService            *handlers_eks.EKSServiceImpl
+	ecsService            *handlers_ecs.Service
+	ecsScheduler          *handlers_ecs.Scheduler
 	acmService            *handlers_acm.ACMServiceImpl
 	ecrMetaService        *handlers_ecr.MetaServiceImpl
 	routeTableService     *handlers_ec2_routetable.RouteTableServiceImpl
@@ -947,6 +950,24 @@ func (d *Daemon) subscribeAll() error {
 		)
 	}
 
+	// ECS gateway → daemon subscriptions (control plane; per-account KV).
+	if d.ecsService != nil {
+		subs = append(subs,
+			natsSub{"ecs.CreateCluster", d.handleECSCreateCluster, "spinifex-workers"},
+			natsSub{"ecs.DescribeClusters", d.handleECSDescribeClusters, "spinifex-workers"},
+			natsSub{"ecs.ListClusters", d.handleECSListClusters, "spinifex-workers"},
+			natsSub{"ecs.RegisterTaskDefinition", d.handleECSRegisterTaskDefinition, "spinifex-workers"},
+			natsSub{"ecs.DescribeTaskDefinition", d.handleECSDescribeTaskDefinition, "spinifex-workers"},
+			natsSub{"ecs.ListTaskDefinitions", d.handleECSListTaskDefinitions, "spinifex-workers"},
+			natsSub{"ecs.RegisterContainerInstance", d.handleECSRegisterContainerInstance, "spinifex-workers"},
+			natsSub{"ecs.DescribeContainerInstances", d.handleECSDescribeContainerInstances, "spinifex-workers"},
+			natsSub{"ecs.ListContainerInstances", d.handleECSListContainerInstances, "spinifex-workers"},
+			natsSub{"ecs.RunTask", d.handleECSRunTask, "spinifex-workers"},
+			natsSub{"ecs.DescribeTasks", d.handleECSDescribeTasks, "spinifex-workers"},
+			natsSub{"ecs.ListTasks", d.handleECSListTasks, "spinifex-workers"},
+		)
+	}
+
 	// ACM gateway → daemon subscriptions (minimal certificate store).
 	if d.acmService != nil {
 		subs = append(subs,
@@ -1377,6 +1398,27 @@ func (d *Daemon) startCluster() error {
 		return fmt.Errorf("failed to initialize EKS service: %w", err)
 	}
 
+	// ECS control plane: per-account KV-backed handlers + a leader-elected
+	// scheduler goroutine that owns the Layer-2 bus subscriptions and heartbeat
+	// reaper. The scheduler is disabled (handlers still serve) when JetStream is
+	// unavailable.
+	d.ecsService = handlers_ecs.NewService(d.natsConn, d.config.Region, d.clusterConfig.AWS.InternalSuffix)
+	if js, jsErr := d.natsConn.JetStream(); jsErr != nil {
+		slog.Warn("ECS scheduler disabled: JetStream unavailable", "err", jsErr)
+	} else if _, lbErr := handlers_ecs.InitLeaderBucket(js); lbErr != nil {
+		slog.Warn("ECS scheduler disabled: leader bucket init failed", "err", lbErr)
+	} else {
+		d.ecsScheduler = handlers_ecs.NewScheduler(d.natsConn, d.ecsService, d.node)
+		d.shutdownWg.Go(func() {
+			defer func() {
+				if r := recover(); r != nil {
+					slog.Error("ECS scheduler goroutine panicked", "recover", r)
+				}
+			}()
+			d.ecsScheduler.Run(d.ctx)
+		})
+	}
+
 	d.acmService, err = initServiceWithRetry("ACM service", func() (*handlers_acm.ACMServiceImpl, error) {
 		return handlers_acm.NewACMServiceImplWithNATS(d.config, d.natsConn)
 	})

spinifex/daemon/daemon_handlers_ecs.go

@@ -0,0 +1,59 @@
+package daemon
+
+import "github.com/nats-io/nats.go"
+
+// --- Cluster ---
+
+func (d *Daemon) handleECSCreateCluster(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.CreateCluster)
+}
+
+func (d *Daemon) handleECSDescribeClusters(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.DescribeClusters)
+}
+
+func (d *Daemon) handleECSListClusters(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.ListClusters)
+}
+
+// --- Task definition ---
+
+func (d *Daemon) handleECSRegisterTaskDefinition(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.RegisterTaskDefinition)
+}
+
+func (d *Daemon) handleECSDescribeTaskDefinition(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.DescribeTaskDefinition)
+}
+
+func (d *Daemon) handleECSListTaskDefinitions(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.ListTaskDefinitions)
+}
+
+// --- Container instance ---
+
+func (d *Daemon) handleECSRegisterContainerInstance(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.RegisterContainerInstance)
+}
+
+func (d *Daemon) handleECSDescribeContainerInstances(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.DescribeContainerInstances)
+}
+
+func (d *Daemon) handleECSListContainerInstances(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.ListContainerInstances)
+}
+
+// --- Task ---
+
+func (d *Daemon) handleECSRunTask(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.RunTask)
+}
+
+func (d *Daemon) handleECSDescribeTasks(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.DescribeTasks)
+}
+
+func (d *Daemon) handleECSListTasks(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.ListTasks)
+}

spinifex/gateway/ecs/actions.go

@@ -0,0 +1,121 @@
+package gateway_ecs
+
+import (
+	"encoding/json"
+
+	"github.com/aws/aws-sdk-go/service/ecs"
+	handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs"
+	"github.com/nats-io/nats.go"
+)
+
+// unmarshalIfBody decodes body into out only when non-empty.
+func unmarshalIfBody(body []byte, out any) error {
+	if len(body) == 0 {
+		return nil
+	}
+	return json.Unmarshal(body, out)
+}
+
+// --- Cluster ---
+
+func CreateCluster(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.CreateClusterInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).CreateCluster(input, accountID)
+}
+
+func DescribeClusters(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.DescribeClustersInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).DescribeClusters(input, accountID)
+}
+
+func ListClusters(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.ListClustersInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).ListClusters(input, accountID)
+}
+
+// --- Task definition ---
+
+func RegisterTaskDefinition(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.RegisterTaskDefinitionInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).RegisterTaskDefinition(input, accountID)
+}
+
+func DescribeTaskDefinition(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.DescribeTaskDefinitionInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).DescribeTaskDefinition(input, accountID)
+}
+
+func ListTaskDefinitions(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.ListTaskDefinitionsInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).ListTaskDefinitions(input, accountID)
+}
+
+// --- Container instance ---
+
+func RegisterContainerInstance(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.RegisterContainerInstanceInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).RegisterContainerInstance(input, accountID)
+}
+
+func DescribeContainerInstances(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.DescribeContainerInstancesInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).DescribeContainerInstances(input, accountID)
+}
+
+func ListContainerInstances(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.ListContainerInstancesInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).ListContainerInstances(input, accountID)
+}
+
+// --- Task ---
+
+func RunTask(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.RunTaskInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).RunTask(input, accountID)
+}
+
+func DescribeTasks(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.DescribeTasksInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).DescribeTasks(input, accountID)
+}
+
+func ListTasks(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.ListTasksInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).ListTasks(input, accountID)
+}

spinifex/gateway/ecs/handler.go

@@ -3,9 +3,9 @@
 // (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.
+// The full v1 action namespace (ecs-v1.md §1) is registered as the API contract.
+// Cluster, task-definition, container-instance and task actions dispatch to the
+// daemon over NATS; the remaining actions resolve to the NotImplemented stub.
 package gateway_ecs
 
 import (
@@ -37,30 +37,30 @@ func NotImplemented(_ *nats.Conn, _ string, _ []byte) (any, error) {
 }
 
 // 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.
+// 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).
-	"CreateCluster":               NotImplemented,
-	"DescribeClusters":            NotImplemented,
-	"ListClusters":                NotImplemented,
+	"CreateCluster":               CreateCluster,
+	"DescribeClusters":            DescribeClusters,
+	"ListClusters":                ListClusters,
 	"DeleteCluster":               NotImplemented,
 	"UpdateCluster":               NotImplemented,
 	"PutClusterCapacityProviders": NotImplemented,
 
 	// Task definition.
-	"RegisterTaskDefinition":     NotImplemented,
+	"RegisterTaskDefinition":     RegisterTaskDefinition,
 	"DeregisterTaskDefinition":   NotImplemented,
-	"DescribeTaskDefinition":     NotImplemented,
-	"ListTaskDefinitions":        NotImplemented,
+	"DescribeTaskDefinition":     DescribeTaskDefinition,
+	"ListTaskDefinitions":        ListTaskDefinitions,
 	"ListTaskDefinitionFamilies": NotImplemented,
 
-	// Task.
-	"RunTask":       NotImplemented,
+	// Task. StartTask/StopTask land with the 4f service controller + reaper.
+	"RunTask":       RunTask,
 	"StartTask":     NotImplemented,
 	"StopTask":      NotImplemented,
-	"DescribeTasks": NotImplemented,
-	"ListTasks":     NotImplemented,
+	"DescribeTasks": DescribeTasks,
+	"ListTasks":     ListTasks,
 
 	// Service. ListServicesByNamespace is a no-op stub in v1.
 	"CreateService":           NotImplemented,
@@ -71,10 +71,10 @@ var Actions = map[string]Handler{
 	"ListServicesByNamespace": NotImplemented,
 
 	// Container instance.
-	"RegisterContainerInstance":     NotImplemented,
+	"RegisterContainerInstance":     RegisterContainerInstance,
 	"DeregisterContainerInstance":   NotImplemented,
-	"DescribeContainerInstances":    NotImplemented,
-	"ListContainerInstances":        NotImplemented,
+	"DescribeContainerInstances":    DescribeContainerInstances,
+	"ListContainerInstances":        ListContainerInstances,
 	"UpdateContainerInstancesState": NotImplemented,
 
 	// Account settings (passthrough; no enforcement v1).

spinifex/gateway/ecs/handler_test.go

@@ -17,13 +17,26 @@ func TestNotImplemented(t *testing.T) {
 	assert.Equal(t, awserrors.ErrorNotImplemented, err.Error())
 }
 
-// Every registered action is the NotImplemented stub in this sprint.
-func TestActions_AllNotImplemented(t *testing.T) {
+// wiredActions are dispatched to the daemon over NATS; the rest are still the
+// NotImplemented stub. Wired actions error here because the nil NATS conn cannot
+// reach the daemon — that they no longer return NotImplemented is the assertion.
+var wiredActions = map[string]bool{
+	"CreateCluster": true, "DescribeClusters": true, "ListClusters": true,
+	"RegisterTaskDefinition": true, "DescribeTaskDefinition": true, "ListTaskDefinitions": true,
+	"RegisterContainerInstance": true, "DescribeContainerInstances": true, "ListContainerInstances": true,
+	"RunTask": true, "DescribeTasks": true, "ListTasks": true,
+}
+
+func TestActions_StubsReturnNotImplemented(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)
+		if wiredActions[action] {
+			assert.NotEqual(t, awserrors.ErrorNotImplemented, err.Error(), "wired action %q should not be a stub", action)
+			continue
+		}
+		assert.Equal(t, awserrors.ErrorNotImplemented, err.Error(), "stub action %q", action)
 	}
 }
 

spinifex/gateway/ecs_test.go

@@ -32,7 +32,7 @@ func TestECSActionFromTarget(t *testing.T) {
 
 // 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.
+// surfaces must be registered, whether wired to the daemon or a 501 stub.
 func TestECSActionsMap_NamespaceRegistered(t *testing.T) {
 	namespace := []string{
 		// Cluster
@@ -76,12 +76,11 @@ func TestECSRequest_UnknownAction(t *testing.T) {
 	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.
+// An action whose real handler has not yet landed still resolves to the 501 stub.
 func TestECSRequest_KnownActionNotImplemented(t *testing.T) {
 	gw := &GatewayConfig{DisableLogging: true}
 	err := gw.ECS_Request(httptest.NewRecorder(),
-		setupECSRequest("AmazonEC2ContainerServiceV20141113.RunTask", "{}"))
+		setupECSRequest("AmazonEC2ContainerServiceV20141113.DeleteCluster", "{}"))
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorNotImplemented, err.Error())
 }

spinifex/gateway/gateway.go

@@ -229,8 +229,8 @@ const clusterUnavailableMsg = "cluster unavailable: NATS disconnected — check
 func (gw *GatewayConfig) writeClusterUnavailable(w http.ResponseWriter, _ *http.Request, svc string) {
 	requestID := uuid.NewString()
 
-	// EKS uses AWS REST-JSON 1.1.
-	if svc == "eks" {
+	// EKS and ECS use AWS JSON 1.1.
+	if svc == "eks" || svc == "ecs" {
 		body := GenerateEKSErrorResponse(awserrors.ErrorServiceUnavailable, clusterUnavailableMsg, requestID)
 		w.Header().Set("Content-Type", eksJSONContentType)
 		w.WriteHeader(http.StatusServiceUnavailable)
@@ -281,8 +281,8 @@ func (gw *GatewayConfig) writeThrottleError(w http.ResponseWriter, r *http.Reque
 	}
 	errorMsg := awserrors.ErrorLookup[errorCode]
 
-	// EKS uses AWS REST-JSON 1.1.
-	if svc == "eks" {
+	// EKS and ECS use AWS JSON 1.1.
+	if svc == "eks" || svc == "ecs" {
 		body := GenerateEKSErrorResponse(errorCode, errorMsg.Message, requestID)
 		w.Header().Set("Content-Type", eksJSONContentType)
 		w.WriteHeader(errorMsg.HTTPCode)
@@ -497,8 +497,8 @@ func (gw *GatewayConfig) ErrorHandler(w http.ResponseWriter, r *http.Request, er
 		errorMsg.HTTPCode = 500
 	}
 
-	// EKS, ECR, ACM, and tagging use AWS JSON 1.1; query/XML services fall through.
-	if svc == "eks" || svc == "ecr" || svc == "acm" || svc == "tagging" {
+	// EKS, ECR, ACM, ECS, and tagging use AWS JSON 1.1; query/XML services fall through.
+	if svc == "eks" || svc == "ecr" || svc == "acm" || svc == "ecs" || svc == "tagging" {
 		body := GenerateEKSErrorResponse(err.Error(), errorMsg.Message, requestId)
 		slog.Debug("Generated JSON error response", "service", svc, "error", err.Error(), "json", string(body), "requestId", requestId)
 		w.Header().Set("Content-Type", eksJSONContentType)

spinifex/handlers/ecs/bus/messages.go

@@ -39,3 +39,71 @@ const (
 	StatusActive   = "ACTIVE"
 	StatusDraining = "DRAINING"
 )
+
+// Task lifecycle status values, matching the AWS ECS task status enum, reported
+// in TaskState.LastStatus and per-container in ContainerStatus.Status.
+const (
+	TaskStatusPending = "PENDING"
+	TaskStatusRunning = "RUNNING"
+	TaskStatusStopped = "STOPPED"
+)
+
+// PortMapping is a container port exposed on the host (bridge mode v1).
+type PortMapping struct {
+	ContainerPort int    `json:"containerPort"`
+	HostPort      int    `json:"hostPort,omitempty"`
+	Protocol      string `json:"protocol,omitempty"` // tcp | udp (default tcp)
+}
+
+// AssignContainer is one container the agent must pull and run for a task.
+type AssignContainer struct {
+	Name         string            `json:"name"`
+	Image        string            `json:"image"`
+	CPU          int               `json:"cpu,omitempty"`
+	MemoryMiB    int               `json:"memoryMiB,omitempty"`
+	Essential    bool              `json:"essential"`
+	Command      []string          `json:"command,omitempty"`
+	Environment  map[string]string `json:"environment,omitempty"`
+	PortMappings []PortMapping     `json:"portMappings,omitempty"`
+}
+
+// Assign is published on AssignSubject when the scheduler places a task on this
+// instance (scheduler → agent). The agent pulls each container image, creates and
+// starts the containers via containerd, then reports progress on TaskStateSubject.
+type Assign struct {
+	AccountID       string            `json:"accountId"`
+	ClusterName     string            `json:"clusterName"`
+	InstanceID      string            `json:"instanceId"`
+	TaskID          string            `json:"taskId"`
+	TaskARN         string            `json:"taskArn"`
+	TaskDefFamily   string            `json:"taskDefFamily"`
+	TaskDefRevision int               `json:"taskDefRevision"`
+	Containers      []AssignContainer `json:"containers"`
+	// CredID + TaskRoleARN are placeholders for the Sprint 4g IAM credential
+	// endpoint; the agent ignores them in 4e.
+	CredID      string    `json:"credId,omitempty"`
+	TaskRoleARN string    `json:"taskRoleArn,omitempty"`
+	AssignedAt  time.Time `json:"assignedAt"`
+}
+
+// ContainerStatus is a single container's reported lifecycle state.
+type ContainerStatus struct {
+	Name        string `json:"name"`
+	Status      string `json:"status"` // PENDING | RUNNING | STOPPED
+	ContainerID string `json:"containerId,omitempty"`
+	ExitCode    *int   `json:"exitCode,omitempty"`
+}
+
+// TaskState is published on TaskStateSubject as the agent drives a task through
+// its lifecycle (agent → scheduler). The scheduler updates the task record and
+// recomputes the instance's remaining capacity.
+type TaskState struct {
+	AccountID   string            `json:"accountId"`
+	ClusterName string            `json:"clusterName"`
+	InstanceID  string            `json:"instanceId"`
+	TaskID      string            `json:"taskId"`
+	LastStatus  string            `json:"lastStatus"` // PENDING | RUNNING | STOPPED
+	Containers  []ContainerStatus `json:"containers,omitempty"`
+	Reason      string            `json:"reason,omitempty"`
+	ReportedAt  time.Time         `json:"reportedAt"`
+}

spinifex/handlers/ecs/conv.go

@@ -0,0 +1,124 @@
+package handlers_ecs
+
+import (
+	"strings"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+// awsStringSlice dereferences a []*string, dropping nils/empties.
+func awsStringSlice(in []*string) []string {
+	out := make([]string, 0, len(in))
+	for _, p := range in {
+		if v := aws.StringValue(p); v != "" {
+			out = append(out, v)
+		}
+	}
+	return out
+}
+
+// clusterShortName extracts the cluster name from a name or a cluster ARN.
+func clusterShortName(ref string) string {
+	ref = strings.TrimSpace(ref)
+	if ref == "" {
+		return defaultCluster
+	}
+	if i := strings.LastIndex(ref, "cluster/"); i >= 0 {
+		return ref[i+len("cluster/"):]
+	}
+	return ref
+}
+
+// containerInstanceShortID extracts the container-instance UUID from an ID or ARN.
+func containerInstanceShortID(ref string) string {
+	if i := strings.LastIndexByte(ref, '/'); i >= 0 {
+		return ref[i+1:]
+	}
+	return ref
+}
+
+// containerDefsFromAWS maps SDK container definitions to the persisted subset.
+func containerDefsFromAWS(in []*ecs.ContainerDefinition) []ContainerDef {
+	out := make([]ContainerDef, 0, len(in))
+	for _, c := range in {
+		if c == nil {
+			continue
+		}
+		def := ContainerDef{
+			Name:      aws.StringValue(c.Name),
+			Image:     aws.StringValue(c.Image),
+			CPU:       int(aws.Int64Value(c.Cpu)),
+			MemoryMiB: int(aws.Int64Value(c.Memory)),
+			Essential: aws.BoolValue(c.Essential),
+			Command:   awsStringSlice(c.Command),
+		}
+		for _, e := range c.Environment {
+			if e == nil {
+				continue
+			}
+			if def.Environment == nil {
+				def.Environment = map[string]string{}
+			}
+			def.Environment[aws.StringValue(e.Name)] = aws.StringValue(e.Value)
+		}
+		for _, p := range c.PortMappings {
+			if p == nil {
+				continue
+			}
+			def.PortMappings = append(def.PortMappings, bus.PortMapping{
+				ContainerPort: int(aws.Int64Value(p.ContainerPort)),
+				HostPort:      int(aws.Int64Value(p.HostPort)),
+				Protocol:      aws.StringValue(p.Protocol),
+			})
+		}
+		out = append(out, def)
+	}
+	return out
+}
+
+func (c ContainerDef) toAWS() *ecs.ContainerDefinition {
+	cd := &ecs.ContainerDefinition{
+		Name:      aws.String(c.Name),
+		Image:     aws.String(c.Image),
+		Essential: aws.Bool(c.Essential),
+	}
+	if c.CPU > 0 {
+		cd.Cpu = aws.Int64(int64(c.CPU))
+	}
+	if c.MemoryMiB > 0 {
+		cd.Memory = aws.Int64(int64(c.MemoryMiB))
+	}
+	for _, cmd := range c.Command {
+		cd.Command = append(cd.Command, aws.String(cmd))
+	}
+	for k, v := range c.Environment {
+		cd.Environment = append(cd.Environment, &ecs.KeyValuePair{Name: aws.String(k), Value: aws.String(v)})
+	}
+	for _, p := range c.PortMappings {
+		pm := &ecs.PortMapping{ContainerPort: aws.Int64(int64(p.ContainerPort))}
+		if p.HostPort > 0 {
+			pm.HostPort = aws.Int64(int64(p.HostPort))
+		}
+		if p.Protocol != "" {
+			pm.Protocol = aws.String(p.Protocol)
+		}
+		cd.PortMappings = append(cd.PortMappings, pm)
+	}
+	return cd
+}
+
+// toAssignContainer maps a persisted container def to its bus assign payload.
+func (c ContainerDef) toAssignContainer() bus.AssignContainer {
+	return bus.AssignContainer{
+		Name:         c.Name,
+		Image:        c.Image,
+		CPU:          c.CPU,
+		MemoryMiB:    c.MemoryMiB,
+		Essential:    c.Essential,
+		Command:      c.Command,
+		Environment:  c.Environment,
+		PortMappings: c.PortMappings,
+	}
+}

spinifex/handlers/ecs/conv_test.go

@@ -0,0 +1,83 @@
+package handlers_ecs
+
+import (
+	"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 TestARNBuilders(t *testing.T) {
+	assert.Equal(t, "arn:aws:ecs:ap-southeast-2:123456789012:cluster/web",
+		ClusterARN("ap-southeast-2", "123456789012", "web"))
+	assert.Equal(t, "arn:aws:ecs:ap-southeast-2:123456789012:task-definition/nginx:3",
+		TaskDefARN("ap-southeast-2", "123456789012", "nginx", 3))
+	assert.Equal(t, "arn:aws:ecs:ap-southeast-2:123456789012:task/web/t-1",
+		TaskARN("ap-southeast-2", "123456789012", "web", "t-1"))
+	assert.Equal(t, "arn:aws:ecs:ap-southeast-2:123456789012:container-instance/web/i-1",
+		ContainerInstanceARN("ap-southeast-2", "123456789012", "web", "i-1"))
+}
+
+func TestTaskDefReservedSums(t *testing.T) {
+	td := &TaskDefRecord{Containers: []ContainerDef{
+		{CPU: 128, MemoryMiB: 256},
+		{CPU: 64, MemoryMiB: 128},
+	}}
+	assert.Equal(t, 192, td.reservedCPU())
+	assert.Equal(t, 384, td.reservedMemory())
+}
+
+func TestClusterShortName(t *testing.T) {
+	assert.Equal(t, defaultCluster, clusterShortName(""))
+	assert.Equal(t, "web", clusterShortName("web"))
+	assert.Equal(t, "web", clusterShortName("arn:aws:ecs:r:a:cluster/web"))
+}
+
+func TestContainerInstanceShortID(t *testing.T) {
+	assert.Equal(t, "i-1", containerInstanceShortID("i-1"))
+	assert.Equal(t, "i-1", containerInstanceShortID("arn:aws:ecs:r:a:container-instance/web/i-1"))
+}
+
+func TestAWSStringSlice(t *testing.T) {
+	got := awsStringSlice([]*string{aws.String("a"), nil, aws.String(""), aws.String("b")})
+	assert.Equal(t, []string{"a", "b"}, got)
+}
+
+// containerDefsFromAWS → toAWS / toAssignContainer must preserve image, command,
+// env, and port mappings (the fields the agent actually runs on).
+func TestContainerDefRoundTrip(t *testing.T) {
+	in := []*ecs.ContainerDefinition{{
+		Name:         aws.String("web"),
+		Image:        aws.String("registry/web:1"),
+		Cpu:          aws.Int64(128),
+		Memory:       aws.Int64(256),
+		Essential:    aws.Bool(true),
+		Command:      []*string{aws.String("/bin/sh"), aws.String("-c")},
+		Environment:  []*ecs.KeyValuePair{{Name: aws.String("FOO"), Value: aws.String("bar")}},
+		PortMappings: []*ecs.PortMapping{{ContainerPort: aws.Int64(80), HostPort: aws.Int64(8080), Protocol: aws.String("tcp")}},
+	}}
+	defs := containerDefsFromAWS(in)
+	require.Len(t, defs, 1)
+	d := defs[0]
+	assert.Equal(t, "registry/web:1", d.Image)
+	assert.Equal(t, 128, d.CPU)
+	assert.Equal(t, []string{"/bin/sh", "-c"}, d.Command)
+	assert.Equal(t, "bar", d.Environment["FOO"])
+	require.Len(t, d.PortMappings, 1)
+	assert.Equal(t, 8080, d.PortMappings[0].HostPort)
+
+	back := d.toAWS()
+	assert.Equal(t, "registry/web:1", aws.StringValue(back.Image))
+	assert.Equal(t, int64(80), aws.Int64Value(back.PortMappings[0].ContainerPort))
+
+	ac := d.toAssignContainer()
+	assert.Equal(t, "web", ac.Name)
+	assert.Equal(t, "registry/web:1", ac.Image)
+	assert.Equal(t, "bar", ac.Environment["FOO"])
+}
+
+func TestContainerDefsFromAWS_SkipsNil(t *testing.T) {
+	assert.Empty(t, containerDefsFromAWS([]*ecs.ContainerDefinition{nil}))
+}

spinifex/handlers/ecs/instances.go

@@ -0,0 +1,274 @@
+package handlers_ecs
+
+import (
+	"encoding/json"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	"github.com/nats-io/nats.go"
+)
+
+// listInstanceRecords returns all container-instance records in a cluster.
+func (s *Service) listInstanceRecords(kv nats.KeyValue, cluster string) ([]InstanceRecord, error) {
+	keys, err := keysWithPrefix(kv, InstancesPrefix(cluster))
+	if err != nil {
+		return nil, err
+	}
+	out := make([]InstanceRecord, 0, len(keys))
+	for _, k := range keys {
+		var rec InstanceRecord
+		found, err := getJSON(kv, k, &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out = append(out, rec)
+		}
+	}
+	return out, nil
+}
+
+// RegisterContainerInstance is the AWS-API registration path. In 4e the agent
+// normally registers over the Layer-2 bus; this keeps API parity by writing the
+// same record shape from an explicit call.
+func (s *Service) RegisterContainerInstance(input *ecs.RegisterContainerInstanceInput, accountID string) (*ecs.RegisterContainerInstanceOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	instanceID := aws.StringValue(input.InstanceIdentityDocument)
+	if instanceID == "" {
+		instanceID = "ci-" + time.Now().UTC().Format("20060102150405")
+	}
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	rec, err := s.upsertInstance(kv, accountID, cluster, instanceID, func(r *InstanceRecord) {
+		for _, res := range input.TotalResources {
+			switch aws.StringValue(res.Name) {
+			case "CPU":
+				r.TotalCPU = int(aws.Int64Value(res.IntegerValue))
+			case "MEMORY":
+				r.TotalMemoryMiB = int(aws.Int64Value(res.IntegerValue))
+			}
+		}
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &ecs.RegisterContainerInstanceOutput{ContainerInstance: s.instanceToAWS(rec)}, nil
+}
+
+// DescribeContainerInstances returns records for the named container instances.
+func (s *Service) DescribeContainerInstances(input *ecs.DescribeContainerInstancesInput, accountID string) (*ecs.DescribeContainerInstancesOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.DescribeContainerInstancesOutput{}
+	for _, ref := range awsStringSlice(input.ContainerInstances) {
+		id := containerInstanceShortID(ref)
+		var rec InstanceRecord
+		found, err := getJSON(kv, InstanceKey(cluster, id), &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out.ContainerInstances = append(out.ContainerInstances, s.instanceToAWS(&rec))
+		} else {
+			out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")})
+		}
+	}
+	return out, nil
+}
+
+// ListContainerInstances returns the ARNs of all container instances in a cluster.
+func (s *Service) ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	recs, err := s.listInstanceRecords(kv, cluster)
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.ListContainerInstancesOutput{}
+	for i := range recs {
+		out.ContainerInstanceArns = append(out.ContainerInstanceArns, aws.String(recs[i].ARN))
+	}
+	return out, nil
+}
+
+// upsertInstance reads-or-creates the instance record, applies mutate, and writes
+// it back. Used by both the AWS register path and the bus register handler.
+func (s *Service) upsertInstance(kv nats.KeyValue, accountID, cluster, instanceID string, mutate func(*InstanceRecord)) (*InstanceRecord, error) {
+	var rec InstanceRecord
+	found, err := getJSON(kv, InstanceKey(cluster, instanceID), &rec)
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now().UTC()
+	if !found {
+		rec = InstanceRecord{
+			InstanceID:   instanceID,
+			ARN:          ContainerInstanceARN(s.region, accountID, cluster, instanceID),
+			Cluster:      cluster,
+			Status:       InstanceStatusActive,
+			RegisteredAt: now,
+		}
+	}
+	rec.LastSeen = now
+	if mutate != nil {
+		mutate(&rec)
+	}
+	if err := putJSON(kv, InstanceKey(cluster, instanceID), &rec); err != nil {
+		return nil, err
+	}
+	return &rec, nil
+}
+
+func (s *Service) instanceToAWS(r *InstanceRecord) *ecs.ContainerInstance {
+	registered := []*ecs.Resource{
+		{Name: aws.String("CPU"), Type: aws.String("INTEGER"), IntegerValue: aws.Int64(int64(r.TotalCPU))},
+		{Name: aws.String("MEMORY"), Type: aws.String("INTEGER"), IntegerValue: aws.Int64(int64(r.TotalMemoryMiB))},
+	}
+	remaining := []*ecs.Resource{
+		{Name: aws.String("CPU"), Type: aws.String("INTEGER"), IntegerValue: aws.Int64(int64(r.TotalCPU - r.ReservedCPU))},
+		{Name: aws.String("MEMORY"), Type: aws.String("INTEGER"), IntegerValue: aws.Int64(int64(r.TotalMemoryMiB - r.ReservedMemoryMiB))},
+	}
+	return &ecs.ContainerInstance{
+		ContainerInstanceArn: aws.String(r.ARN),
+		Ec2InstanceId:        aws.String(r.InstanceID),
+		Status:               aws.String(r.Status),
+		AgentConnected:       aws.Bool(r.Status == InstanceStatusActive),
+		RunningTasksCount:    aws.Int64(int64(len(r.PlacedTasks))),
+		RegisteredResources:  registered,
+		RemainingResources:   remaining,
+		VersionInfo:          &ecs.VersionInfo{AgentVersion: aws.String(r.AgentVersion)},
+	}
+}
+
+// --- Layer-2 bus event handlers (called by the scheduler) ---
+
+// recordRegister upserts a container-instance record from a bus RegisterInstance.
+func (s *Service) recordRegister(msg *bus.RegisterInstance) error {
+	kv, err := s.bucket(msg.AccountID)
+	if err != nil {
+		return err
+	}
+	_, err = s.upsertInstance(kv, msg.AccountID, msg.ClusterName, msg.InstanceID, func(r *InstanceRecord) {
+		r.AZ = msg.AZ
+		r.Hostname = msg.Hostname
+		r.AgentVersion = msg.AgentVersion
+		r.TotalCPU = msg.Capacity.CPU
+		r.TotalMemoryMiB = msg.Capacity.MemoryMiB
+		r.Status = InstanceStatusActive
+	})
+	return err
+}
+
+// recordHeartbeat refreshes an instance's LastSeen and status from a bus beat.
+func (s *Service) recordHeartbeat(msg *bus.Heartbeat) error {
+	kv, err := s.bucket(msg.AccountID)
+	if err != nil {
+		return err
+	}
+	var rec InstanceRecord
+	found, err := getJSON(kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
+	if err != nil || !found {
+		return err
+	}
+	rec.LastSeen = time.Now().UTC()
+	if msg.Status != "" {
+		rec.Status = msg.Status
+	}
+	return putJSON(kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
+}
+
+// recordTaskState applies an agent task-state report: it updates the task record
+// and, on STOPPED, releases the reserved capacity back to the instance.
+func (s *Service) recordTaskState(msg *bus.TaskState) error {
+	kv, err := s.bucket(msg.AccountID)
+	if err != nil {
+		return err
+	}
+	var task TaskRecord
+	found, err := getJSON(kv, TaskKey(msg.ClusterName, msg.TaskID), &task)
+	if err != nil || !found {
+		return err
+	}
+
+	prev := task.LastStatus
+	task.LastStatus = msg.LastStatus
+	if len(msg.Containers) > 0 {
+		task.Containers = task.Containers[:0]
+		for _, c := range msg.Containers {
+			task.Containers = append(task.Containers, ContainerState{
+				Name: c.Name, Status: c.Status, ContainerID: c.ContainerID, ExitCode: c.ExitCode,
+			})
+		}
+	}
+	now := time.Now().UTC()
+	if msg.LastStatus == TaskStatusRunning && task.StartedAt.IsZero() {
+		task.StartedAt = now
+	}
+	if msg.LastStatus == TaskStatusStopped {
+		task.StoppedAt = now
+		if msg.Reason != "" {
+			task.StoppedReason = msg.Reason
+		}
+	}
+	if err := putJSON(kv, TaskKey(msg.ClusterName, msg.TaskID), &task); err != nil {
+		return err
+	}
+
+	// Release capacity once, on the transition into STOPPED.
+	if msg.LastStatus == TaskStatusStopped && prev != TaskStatusStopped {
+		return s.releaseReservation(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID, task.ReservedCPU, task.ReservedMemoryMiB)
+	}
+	return nil
+}
+
+// releaseReservation returns a stopped task's capacity to its instance under CAS.
+func (s *Service) releaseReservation(kv nats.KeyValue, cluster, instanceID, taskID string, cpu, mem int) error {
+	for range reservePlacementRetries {
+		entry, err := kv.Get(InstanceKey(cluster, instanceID))
+		if err != nil {
+			return nil //nolint:nilerr // instance gone; nothing to release
+		}
+		var rec InstanceRecord
+		if uerr := json.Unmarshal(entry.Value(), &rec); uerr != nil {
+			return uerr
+		}
+		rec.ReservedCPU = maxZero(rec.ReservedCPU - cpu)
+		rec.ReservedMemoryMiB = maxZero(rec.ReservedMemoryMiB - mem)
+		rec.PlacedTasks = removeString(rec.PlacedTasks, taskID)
+		data, merr := json.Marshal(&rec)
+		if merr != nil {
+			return merr
+		}
+		if _, uerr := kv.Update(InstanceKey(cluster, instanceID), data, entry.Revision()); uerr == nil {
+			return nil
+		}
+	}
+	return nil
+}
+
+func maxZero(v int) int {
+	if v < 0 {
+		return 0
+	}
+	return v
+}
+
+func removeString(in []string, s string) []string {
+	out := in[:0]
+	for _, v := range in {
+		if v != s {
+			out = append(out, v)
+		}
+	}
+	return out
+}

spinifex/handlers/ecs/placement.go

@@ -0,0 +1,86 @@
+package handlers_ecs
+
+import (
+	"errors"
+	"sort"
+	"strings"
+)
+
+// Placement strategy identifiers (ecs-v1.md Q15). Default is binpack:memory.
+const (
+	StrategyBinpack = "binpack"
+	StrategySpread  = "spread"
+	StrategyRandom  = "random"
+)
+
+// ErrNoCapacity is returned when no ACTIVE instance can fit the task.
+var ErrNoCapacity = errors.New("no container instance has capacity for the task")
+
+// remainingCPU/remainingMemory report an instance's unreserved capacity.
+func (r *InstanceRecord) remainingCPU() int    { return r.TotalCPU - r.ReservedCPU }
+func (r *InstanceRecord) remainingMemory() int { return r.TotalMemoryMiB - r.ReservedMemoryMiB }
+
+// fits reports whether the instance is ACTIVE and has room for the reservation.
+func (r *InstanceRecord) fits(cpu, mem int) bool {
+	if r.Status != InstanceStatusActive {
+		return false
+	}
+	return r.remainingCPU() >= cpu && r.remainingMemory() >= mem
+}
+
+// placeTask selects a container instance for a task reserving (cpu, mem) using
+// the requested strategy. Candidates are filtered to ACTIVE instances that fit;
+// ties broken by instance ID for determinism. strategy "" defaults to binpack.
+//
+// binpack: pick the instance with the LEAST remaining memory that still fits
+// (tightest pack). spread: pick the MOST remaining memory (widest spread).
+// random: caller-stable first fit by instance ID.
+func placeTask(instances []InstanceRecord, cpu, mem int, strategy string) (*InstanceRecord, error) {
+	candidates := make([]InstanceRecord, 0, len(instances))
+	for _, inst := range instances {
+		if inst.fits(cpu, mem) {
+			candidates = append(candidates, inst)
+		}
+	}
+	if len(candidates) == 0 {
+		return nil, ErrNoCapacity
+	}
+
+	switch normalizeStrategy(strategy) {
+	case StrategySpread:
+		sort.Slice(candidates, func(i, j int) bool {
+			if candidates[i].remainingMemory() != candidates[j].remainingMemory() {
+				return candidates[i].remainingMemory() > candidates[j].remainingMemory()
+			}
+			return candidates[i].InstanceID < candidates[j].InstanceID
+		})
+	case StrategyRandom:
+		sort.Slice(candidates, func(i, j int) bool {
+			return candidates[i].InstanceID < candidates[j].InstanceID
+		})
+	default: // binpack
+		sort.Slice(candidates, func(i, j int) bool {
+			if candidates[i].remainingMemory() != candidates[j].remainingMemory() {
+				return candidates[i].remainingMemory() < candidates[j].remainingMemory()
+			}
+			return candidates[i].InstanceID < candidates[j].InstanceID
+		})
+	}
+
+	chosen := candidates[0]
+	return &chosen, nil
+}
+
+// normalizeStrategy maps "binpack:memory"/"binpack:cpu" → "binpack" and lowercases.
+func normalizeStrategy(s string) string {
+	s = strings.ToLower(strings.TrimSpace(s))
+	if i := strings.IndexByte(s, ':'); i >= 0 {
+		s = s[:i]
+	}
+	switch s {
+	case StrategySpread, StrategyRandom:
+		return s
+	default:
+		return StrategyBinpack
+	}
+}

spinifex/handlers/ecs/placement_test.go

@@ -0,0 +1,79 @@
+package handlers_ecs
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func inst(id string, total, reserved int) InstanceRecord {
+	return InstanceRecord{
+		InstanceID:        id,
+		Status:            InstanceStatusActive,
+		TotalCPU:          10000,
+		ReservedCPU:       0,
+		TotalMemoryMiB:    total,
+		ReservedMemoryMiB: reserved,
+	}
+}
+
+func TestPlaceTask_Binpack_PicksTightest(t *testing.T) {
+	// remaining mem: a=900, b=300, c=600 → binpack picks b (least remaining).
+	instances := []InstanceRecord{
+		inst("a", 1000, 100),
+		inst("b", 1000, 700),
+		inst("c", 1000, 400),
+	}
+	got, err := placeTask(instances, 100, 200, "binpack:memory")
+	require.NoError(t, err)
+	assert.Equal(t, "b", got.InstanceID)
+}
+
+func TestPlaceTask_Spread_PicksWidest(t *testing.T) {
+	instances := []InstanceRecord{
+		inst("a", 1000, 100), // 900
+		inst("b", 1000, 700), // 300
+		inst("c", 1000, 400), // 600
+	}
+	got, err := placeTask(instances, 100, 200, StrategySpread)
+	require.NoError(t, err)
+	assert.Equal(t, "a", got.InstanceID)
+}
+
+func TestPlaceTask_Random_StableByID(t *testing.T) {
+	instances := []InstanceRecord{inst("z", 1000, 0), inst("a", 1000, 0), inst("m", 1000, 0)}
+	got, err := placeTask(instances, 100, 100, StrategyRandom)
+	require.NoError(t, err)
+	assert.Equal(t, "a", got.InstanceID)
+}
+
+func TestPlaceTask_NoCapacity(t *testing.T) {
+	instances := []InstanceRecord{inst("a", 100, 90)}
+	_, err := placeTask(instances, 0, 50, StrategyBinpack)
+	assert.ErrorIs(t, err, ErrNoCapacity)
+}
+
+func TestPlaceTask_SkipsDraining(t *testing.T) {
+	d := inst("drain", 1000, 0)
+	d.Status = InstanceStatusDraining
+	_, err := placeTask([]InstanceRecord{d}, 0, 100, StrategyBinpack)
+	assert.ErrorIs(t, err, ErrNoCapacity)
+}
+
+func TestNormalizeStrategy(t *testing.T) {
+	assert.Equal(t, StrategyBinpack, normalizeStrategy(""))
+	assert.Equal(t, StrategyBinpack, normalizeStrategy("binpack:memory"))
+	assert.Equal(t, StrategyBinpack, normalizeStrategy("BinPack:cpu"))
+	assert.Equal(t, StrategySpread, normalizeStrategy("spread"))
+	assert.Equal(t, StrategyRandom, normalizeStrategy("RANDOM"))
+	assert.Equal(t, StrategyBinpack, normalizeStrategy("bogus"))
+}
+
+func TestInstanceRecord_Fits(t *testing.T) {
+	r := inst("a", 1000, 600) // remaining mem 400, cpu 10000
+	assert.True(t, r.fits(100, 400))
+	assert.False(t, r.fits(100, 401))
+	r.Status = InstanceStatusDraining
+	assert.False(t, r.fits(0, 0))
+}

spinifex/handlers/ecs/records.go

@@ -0,0 +1,149 @@
+package handlers_ecs
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+// Lifecycle status constants matching the AWS ECS enums. Task statuses are
+// re-exported from the bus package so the scheduler and the wire payloads agree.
+const (
+	ClusterStatusActive = "ACTIVE"
+
+	InstanceStatusActive   = "ACTIVE"
+	InstanceStatusDraining = "DRAINING"
+
+	TaskDefStatusActive = "ACTIVE"
+
+	TaskStatusPending = bus.TaskStatusPending
+	TaskStatusRunning = bus.TaskStatusRunning
+	TaskStatusStopped = bus.TaskStatusStopped
+)
+
+// ARN builders for the ECS resource shapes (ecs-v1.md §1). Region + accountID
+// scope every ARN; the partition is fixed to "aws" to match the rest of the
+// stack.
+func ClusterARN(region, accountID, name string) string {
+	return fmt.Sprintf("arn:aws:ecs:%s:%s:cluster/%s", region, accountID, name)
+}
+
+func TaskDefARN(region, accountID, family string, rev int) string {
+	return fmt.Sprintf("arn:aws:ecs:%s:%s:task-definition/%s:%d", region, accountID, family, rev)
+}
+
+func TaskARN(region, accountID, cluster, taskID string) string {
+	return fmt.Sprintf("arn:aws:ecs:%s:%s:task/%s/%s", region, accountID, cluster, taskID)
+}
+
+func ContainerInstanceARN(region, accountID, cluster, ciID string) string {
+	return fmt.Sprintf("arn:aws:ecs:%s:%s:container-instance/%s/%s", region, accountID, cluster, ciID)
+}
+
+// ClusterRecord is the persisted cluster meta at ClusterMetaKey.
+type ClusterRecord struct {
+	Name      string            `json:"name"`
+	ARN       string            `json:"arn"`
+	Status    string            `json:"status"`
+	Tags      map[string]string `json:"tags,omitempty"`
+	CreatedAt time.Time         `json:"createdAt"`
+}
+
+// ContainerDef is the persisted subset of an ecs.ContainerDefinition needed to
+// pull and run a container (bridge mode v1).
+type ContainerDef struct {
+	Name         string            `json:"name"`
+	Image        string            `json:"image"`
+	CPU          int               `json:"cpu,omitempty"`
+	MemoryMiB    int               `json:"memoryMiB,omitempty"`
+	Essential    bool              `json:"essential"`
+	Command      []string          `json:"command,omitempty"`
+	Environment  map[string]string `json:"environment,omitempty"`
+	PortMappings []bus.PortMapping `json:"portMappings,omitempty"`
+}
+
+// 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"`
+	Containers   []ContainerDef `json:"containers"`
+	Status       string         `json:"status"`
+	RegisteredAt time.Time      `json:"registeredAt"`
+}
+
+// reservedCPU/reservedMemory sum the task definition's per-container reservations
+// used for bin-pack placement. A taskdef-level cpu/memory is not modelled in v1;
+// placement uses the container sums.
+func (t *TaskDefRecord) reservedCPU() int {
+	total := 0
+	for _, c := range t.Containers {
+		total += c.CPU
+	}
+	return total
+}
+
+func (t *TaskDefRecord) reservedMemory() int {
+	total := 0
+	for _, c := range t.Containers {
+		total += c.MemoryMiB
+	}
+	return total
+}
+
+// InstanceRecord is the persisted container-instance state at InstanceKey. The
+// 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"`
+	// 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.
+	ReservedCPU       int       `json:"reservedCpu"`
+	ReservedMemoryMiB int       `json:"reservedMemoryMiB"`
+	AgentVersion      string    `json:"agentVersion,omitempty"`
+	PlacedTasks       []string  `json:"placedTasks,omitempty"`
+	RegisteredAt      time.Time `json:"registeredAt"`
+	LastSeen          time.Time `json:"lastSeen"`
+}
+
+// ContainerState is a container's last-reported status within a task record.
+type ContainerState struct {
+	Name        string `json:"name"`
+	Status      string `json:"status"`
+	ContainerID string `json:"containerId,omitempty"`
+	ExitCode    *int   `json:"exitCode,omitempty"`
+}
+
+// TaskRecord is the persisted task state at TaskKey; source of truth for
+// DescribeTasks and the placement/capacity accounting.
+type TaskRecord struct {
+	TaskID               string           `json:"taskId"`
+	ARN                  string           `json:"arn"`
+	Cluster              string           `json:"cluster"`
+	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"`
+	CreatedAt            time.Time        `json:"createdAt"`
+	StartedAt            time.Time        `json:"startedAt,omitzero"`
+	StoppedAt            time.Time        `json:"stoppedAt,omitzero"`
+}

spinifex/handlers/ecs/scheduler.go

@@ -0,0 +1,288 @@
+package handlers_ecs
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	"github.com/nats-io/nats.go"
+)
+
+// Scheduler timing. The leader lease lives in spinifex-ecs-leader (60s TTL); the
+// holder refreshes well inside that. The reaper marks an instance DRAINING after
+// three missed 30s heartbeats.
+const (
+	schedulerLeaderKey  = "scheduler"
+	leaseRefresh        = 20 * time.Second
+	reaperInterval      = 30 * time.Second
+	heartbeatTimeout    = 90 * time.Second
+	stoppedReasonReaped = "ContainerInstance disconnected"
+)
+
+// Scheduler is the per-daemon ECS control loop. A single leader (elected via the
+// shared leader bucket) owns the Layer-2 bus subscriptions and the heartbeat
+// reaper; losers idle and retry. Placement itself happens synchronously in the
+// RunTask handler, so a brief leaderless gap never blocks RunTask.
+type Scheduler struct {
+	nc     *nats.Conn
+	svc    *Service
+	holder string
+
+	mu     sync.Mutex
+	leader bool
+	subs   []*nats.Subscription
+}
+
+// NewScheduler constructs a Scheduler. holder identifies this daemon in the lease.
+func NewScheduler(nc *nats.Conn, svc *Service, holder string) *Scheduler {
+	return &Scheduler{nc: nc, svc: svc, holder: holder}
+}
+
+// Run drives the leadership + reaper loop until ctx is cancelled. It is intended
+// to run as a daemon-boot goroutine; panics are the caller's recover concern.
+func (sc *Scheduler) Run(ctx context.Context) {
+	leaseTicker := time.NewTicker(leaseRefresh)
+	reaperTicker := time.NewTicker(reaperInterval)
+	defer leaseTicker.Stop()
+	defer reaperTicker.Stop()
+
+	sc.evaluateLeadership(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			sc.relinquish()
+			return
+		case <-leaseTicker.C:
+			sc.evaluateLeadership(ctx)
+		case <-reaperTicker.C:
+			if sc.isLeader() {
+				sc.reap()
+			}
+		}
+	}
+}
+
+func (sc *Scheduler) isLeader() bool {
+	sc.mu.Lock()
+	defer sc.mu.Unlock()
+	return sc.leader
+}
+
+// evaluateLeadership acquires or refreshes the lease, wiring up (or tearing down)
+// the bus subscriptions as leadership changes.
+func (sc *Scheduler) evaluateLeadership(ctx context.Context) {
+	won := sc.acquireOrRefresh()
+	sc.mu.Lock()
+	was := sc.leader
+	sc.leader = won
+	sc.mu.Unlock()
+
+	switch {
+	case won && !was:
+		if err := sc.subscribeBus(ctx); err != nil {
+			slog.Error("ECS scheduler: bus subscribe failed", "holder", sc.holder, "err", err)
+		} else {
+			slog.Info("ECS scheduler: elected leader, bus subscriptions active", "holder", sc.holder)
+		}
+	case !won && was:
+		sc.unsubscribeBus()
+		slog.Info("ECS scheduler: lost leadership, bus subscriptions dropped", "holder", sc.holder)
+	}
+}
+
+// acquireOrRefresh tries to claim the scheduler lease, refreshing it (resetting
+// the TTL) when we already hold it. Returns true when we are the leader.
+func (sc *Scheduler) acquireOrRefresh() bool {
+	js, err := sc.nc.JetStream()
+	if err != nil {
+		return false
+	}
+	kv, err := InitLeaderBucket(js)
+	if err != nil {
+		return false
+	}
+	if _, err := kv.Create(schedulerLeaderKey, []byte(sc.holder)); err == nil {
+		return true
+	}
+	entry, err := kv.Get(schedulerLeaderKey)
+	if err != nil {
+		return false
+	}
+	if string(entry.Value()) != sc.holder {
+		return false // someone else holds it
+	}
+	// We hold it — refresh to reset the TTL.
+	if _, err := kv.Put(schedulerLeaderKey, []byte(sc.holder)); err != nil {
+		return false
+	}
+	return true
+}
+
+// relinquish drops subscriptions and deletes the lease key on shutdown.
+func (sc *Scheduler) relinquish() {
+	sc.unsubscribeBus()
+	js, err := sc.nc.JetStream()
+	if err != nil {
+		return
+	}
+	kv, err := InitLeaderBucket(js)
+	if err != nil {
+		return
+	}
+	if entry, gerr := kv.Get(schedulerLeaderKey); gerr == nil && string(entry.Value()) == sc.holder {
+		_ = kv.Delete(schedulerLeaderKey)
+	}
+}
+
+// subscribeBus wires the wildcard Layer-2 subscriptions onto the service KV
+// writers. All clusters/accounts fan into the leader.
+func (sc *Scheduler) subscribeBus(_ context.Context) error {
+	register, err := sc.nc.Subscribe("ecs.bus.*.*.instance-register.*", sc.onRegister)
+	if err != nil {
+		return err
+	}
+	heartbeat, err := sc.nc.Subscribe("ecs.bus.*.*.instance-heartbeat.*", sc.onHeartbeat)
+	if err != nil {
+		_ = register.Unsubscribe()
+		return err
+	}
+	taskState, err := sc.nc.Subscribe("ecs.bus.*.*.task-state.*", sc.onTaskState)
+	if err != nil {
+		_ = register.Unsubscribe()
+		_ = heartbeat.Unsubscribe()
+		return err
+	}
+	sc.mu.Lock()
+	sc.subs = []*nats.Subscription{register, heartbeat, taskState}
+	sc.mu.Unlock()
+	return nil
+}
+
+func (sc *Scheduler) unsubscribeBus() {
+	sc.mu.Lock()
+	subs := sc.subs
+	sc.subs = nil
+	sc.mu.Unlock()
+	for _, s := range subs {
+		_ = s.Unsubscribe()
+	}
+}
+
+func (sc *Scheduler) onRegister(msg *nats.Msg) {
+	var m bus.RegisterInstance
+	if err := json.Unmarshal(msg.Data, &m); err != nil {
+		slog.Warn("ECS scheduler: bad register payload", "err", err)
+		return
+	}
+	if err := sc.svc.recordRegister(&m); err != nil {
+		slog.Error("ECS scheduler: record register failed", "instance", m.InstanceID, "err", err)
+		return
+	}
+	slog.Info("ECS scheduler: container instance registered", "cluster", m.ClusterName, "instance", m.InstanceID)
+}
+
+func (sc *Scheduler) onHeartbeat(msg *nats.Msg) {
+	var m bus.Heartbeat
+	if err := json.Unmarshal(msg.Data, &m); err != nil {
+		return
+	}
+	if err := sc.svc.recordHeartbeat(&m); err != nil {
+		slog.Debug("ECS scheduler: record heartbeat failed", "instance", m.InstanceID, "err", err)
+	}
+}
+
+func (sc *Scheduler) onTaskState(msg *nats.Msg) {
+	var m bus.TaskState
+	if err := json.Unmarshal(msg.Data, &m); err != nil {
+		return
+	}
+	if err := sc.svc.recordTaskState(&m); err != nil {
+		slog.Error("ECS scheduler: record task-state failed", "task", m.TaskID, "err", err)
+		return
+	}
+	slog.Info("ECS scheduler: task state", "task", m.TaskID, "status", m.LastStatus)
+}
+
+// reap marks instances that have missed their heartbeat window DRAINING and stops
+// their tasks, releasing capacity. Iterates every ECS account bucket.
+func (sc *Scheduler) reap() {
+	js, err := sc.nc.JetStream()
+	if err != nil {
+		return
+	}
+	now := time.Now().UTC()
+	for bucket := range js.KeyValueStoreNames() {
+		accountID, ok := accountIDFromBucket(bucket)
+		if !ok {
+			continue
+		}
+		kv, err := js.KeyValue(bucket)
+		if err != nil {
+			continue
+		}
+		sc.reapBucket(kv, accountID, now)
+	}
+}
+
+func (sc *Scheduler) reapBucket(kv nats.KeyValue, _ string, now time.Time) {
+	keys, err := keysWithPrefix(kv, "clusters/")
+	if err != nil {
+		return
+	}
+	for _, k := range keys {
+		if !strings.Contains(k, "/instances/") {
+			continue
+		}
+		var inst InstanceRecord
+		found, err := getJSON(kv, k, &inst)
+		if err != nil || !found {
+			continue
+		}
+		if inst.Status != InstanceStatusActive || now.Sub(inst.LastSeen) < heartbeatTimeout {
+			continue
+		}
+		slog.Warn("ECS scheduler: reaping disconnected instance", "cluster", inst.Cluster, "instance", inst.InstanceID,
+			"lastSeen", inst.LastSeen.Format(time.RFC3339))
+		inst.Status = InstanceStatusDraining
+		if perr := putJSON(kv, k, &inst); perr != nil {
+			continue
+		}
+		sc.stopInstanceTasks(kv, inst.Cluster, inst.InstanceID)
+	}
+}
+
+// stopInstanceTasks transitions a reaped instance's non-stopped tasks to STOPPED.
+func (sc *Scheduler) stopInstanceTasks(kv nats.KeyValue, cluster, instanceID string) {
+	keys, err := keysWithPrefix(kv, TasksPrefix(cluster))
+	if err != nil {
+		return
+	}
+	for _, k := range keys {
+		var task TaskRecord
+		found, err := getJSON(kv, k, &task)
+		if err != nil || !found {
+			continue
+		}
+		if task.ContainerInstanceID != instanceID || task.LastStatus == TaskStatusStopped {
+			continue
+		}
+		task.LastStatus = TaskStatusStopped
+		task.StoppedReason = stoppedReasonReaped
+		task.StoppedAt = time.Now().UTC()
+		if perr := putJSON(kv, k, &task); perr != nil {
+			continue
+		}
+	}
+}
+
+// accountIDFromBucket extracts the account ID from an ECS per-account bucket name.
+func accountIDFromBucket(bucket string) (string, bool) {
+	if !strings.HasPrefix(bucket, KVBucketECSAccountPrefix) {
+		return "", false
+	}
+	return strings.TrimPrefix(bucket, KVBucketECSAccountPrefix), true
+}

spinifex/handlers/ecs/service.go

@@ -0,0 +1,384 @@
+package handlers_ecs
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"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"
+)
+
+// ECSService is the ECS control-plane contract implemented by the daemon and
+// adapted onto NATS by NATSECSService. One method per wired AWS ECS action; the
+// remaining actions stay NotImplemented at the gateway.
+type ECSService interface {
+	CreateCluster(input *ecs.CreateClusterInput, accountID string) (*ecs.CreateClusterOutput, error)
+	DescribeClusters(input *ecs.DescribeClustersInput, accountID string) (*ecs.DescribeClustersOutput, error)
+	ListClusters(input *ecs.ListClustersInput, accountID string) (*ecs.ListClustersOutput, error)
+
+	RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, accountID string) (*ecs.RegisterTaskDefinitionOutput, error)
+	DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error)
+	ListTaskDefinitions(input *ecs.ListTaskDefinitionsInput, accountID string) (*ecs.ListTaskDefinitionsOutput, error)
+
+	RegisterContainerInstance(input *ecs.RegisterContainerInstanceInput, accountID string) (*ecs.RegisterContainerInstanceOutput, error)
+	DescribeContainerInstances(input *ecs.DescribeContainerInstancesInput, accountID string) (*ecs.DescribeContainerInstancesOutput, error)
+	ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error)
+
+	RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTaskOutput, error)
+	DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error)
+	ListTasks(input *ecs.ListTasksInput, accountID string) (*ecs.ListTasksOutput, error)
+}
+
+// Service is the daemon-side ECS control-plane implementation, backed by the
+// per-account JetStream KV bucket. It publishes task assignments on the Layer-2
+// bus; the scheduler goroutine consumes instance/task-state events.
+type Service struct {
+	nc     *nats.Conn
+	region string
+	suffix string
+}
+
+var _ ECSService = (*Service)(nil)
+
+// NewService constructs a Service bound to a NATS connection. region scopes the
+// ARNs it mints; suffix is the AWS-parity internal DNS suffix (reserved for ECR
+// endpoint composition).
+func NewService(nc *nats.Conn, region, suffix string) *Service {
+	return &Service{nc: nc, region: region, suffix: suffix}
+}
+
+// defaultCluster is the implicit cluster name when a request omits one.
+const defaultCluster = "default"
+
+func (s *Service) js() (nats.JetStreamContext, error) {
+	if s.nc == nil {
+		return nil, errors.New("ecs service: nil nats connection")
+	}
+	return s.nc.JetStream()
+}
+
+// bucket returns the per-account KV handle, creating it on first use.
+func (s *Service) bucket(accountID string) (nats.KeyValue, error) {
+	js, err := s.js()
+	if err != nil {
+		return nil, err
+	}
+	return GetOrCreateAccountBucket(js, accountID)
+}
+
+// getJSON reads key into out. Returns (false, nil) when the key is absent.
+func getJSON(kv nats.KeyValue, key string, out any) (bool, error) {
+	entry, err := kv.Get(key)
+	if errors.Is(err, nats.ErrKeyNotFound) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	if err := json.Unmarshal(entry.Value(), out); err != nil {
+		return false, fmt.Errorf("unmarshal %s: %w", key, err)
+	}
+	return true, nil
+}
+
+// putJSON marshals v and writes it at key.
+func putJSON(kv nats.KeyValue, key string, v any) error {
+	data, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	if _, err := kv.Put(key, data); err != nil {
+		return fmt.Errorf("put %s: %w", key, err)
+	}
+	return nil
+}
+
+// keysWithPrefix returns all KV keys under prefix.
+func keysWithPrefix(kv nats.KeyValue, prefix string) ([]string, error) {
+	keys, err := kv.Keys()
+	if errors.Is(err, nats.ErrNoKeysFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	out := make([]string, 0, len(keys))
+	for _, k := range keys {
+		if strings.HasPrefix(k, prefix) {
+			out = append(out, k)
+		}
+	}
+	sort.Strings(out)
+	return out, nil
+}
+
+// --- Cluster ---
+
+// CreateCluster persists a cluster meta record. Idempotent: re-creating an
+// existing cluster returns the existing ACTIVE record.
+func (s *Service) CreateCluster(input *ecs.CreateClusterInput, accountID string) (*ecs.CreateClusterOutput, error) {
+	name := aws.StringValue(input.ClusterName)
+	if name == "" {
+		name = defaultCluster
+	}
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	var rec ClusterRecord
+	found, err := getJSON(kv, ClusterMetaKey(name), &rec)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		rec = ClusterRecord{
+			Name:      name,
+			ARN:       ClusterARN(s.region, accountID, name),
+			Status:    ClusterStatusActive,
+			CreatedAt: time.Now().UTC(),
+		}
+		if len(input.Tags) > 0 {
+			rec.Tags = map[string]string{}
+			for _, t := range input.Tags {
+				rec.Tags[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
+			}
+		}
+		if err := putJSON(kv, ClusterMetaKey(name), &rec); err != nil {
+			return nil, err
+		}
+	}
+	return &ecs.CreateClusterOutput{Cluster: rec.toAWS()}, nil
+}
+
+// DescribeClusters returns meta for the named clusters; unknown names are
+// silently skipped (AWS returns them under "failures", omitted here in v1).
+func (s *Service) DescribeClusters(input *ecs.DescribeClustersInput, accountID string) (*ecs.DescribeClustersOutput, error) {
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	names := awsStringSlice(input.Clusters)
+	if len(names) == 0 {
+		names = []string{defaultCluster}
+	}
+	out := &ecs.DescribeClustersOutput{}
+	for _, name := range names {
+		name = clusterShortName(name)
+		var rec ClusterRecord
+		found, err := getJSON(kv, ClusterMetaKey(name), &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out.Clusters = append(out.Clusters, rec.toAWS())
+		}
+	}
+	return out, nil
+}
+
+// ListClusters returns the ARNs of all clusters in the account.
+func (s *Service) ListClusters(_ *ecs.ListClustersInput, accountID string) (*ecs.ListClustersOutput, error) {
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	keys, err := keysWithPrefix(kv, "clusters/")
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.ListClustersOutput{}
+	for _, k := range keys {
+		if !strings.HasSuffix(k, "/meta") {
+			continue
+		}
+		var rec ClusterRecord
+		found, err := getJSON(kv, k, &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out.ClusterArns = append(out.ClusterArns, aws.String(rec.ARN))
+		}
+	}
+	return out, nil
+}
+
+func (r *ClusterRecord) toAWS() *ecs.Cluster {
+	return &ecs.Cluster{
+		ClusterName: aws.String(r.Name),
+		ClusterArn:  aws.String(r.ARN),
+		Status:      aws.String(r.Status),
+	}
+}
+
+// --- Task definition ---
+
+// RegisterTaskDefinition stores a new revision of a family, bumping latest-rev.
+func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, accountID string) (*ecs.RegisterTaskDefinitionOutput, error) {
+	family := aws.StringValue(input.Family)
+	if family == "" {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	rev, err := s.nextRevision(kv, family)
+	if err != nil {
+		return nil, err
+	}
+
+	rec := TaskDefRecord{
+		Family:       family,
+		Revision:     rev,
+		ARN:          TaskDefARN(s.region, accountID, family, rev),
+		NetworkMode:  aws.StringValue(input.NetworkMode),
+		CPU:          aws.StringValue(input.Cpu),
+		Memory:       aws.StringValue(input.Memory),
+		Status:       TaskDefStatusActive,
+		RegisteredAt: time.Now().UTC(),
+		Containers:   containerDefsFromAWS(input.ContainerDefinitions),
+	}
+	if err := putJSON(kv, TaskDefRevKey(family, rev), &rec); err != nil {
+		return nil, err
+	}
+	if err := putJSON(kv, TaskDefLatestRevKey(family), rev); err != nil {
+		return nil, err
+	}
+	return &ecs.RegisterTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil
+}
+
+// nextRevision reads the family's latest-rev and returns latest+1 (1 if absent).
+func (s *Service) nextRevision(kv nats.KeyValue, family string) (int, error) {
+	var latest int
+	found, err := getJSON(kv, TaskDefLatestRevKey(family), &latest)
+	if err != nil {
+		return 0, err
+	}
+	if !found {
+		return 1, nil
+	}
+	return latest + 1, nil
+}
+
+// DescribeTaskDefinition resolves "family", "family:rev" or an ARN to a revision.
+func (s *Service) DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error) {
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	rec, err := s.resolveTaskDef(kv, aws.StringValue(input.TaskDefinition))
+	if err != nil {
+		return nil, err
+	}
+	return &ecs.DescribeTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil
+}
+
+// ListTaskDefinitions returns all revision ARNs, optionally filtered by family.
+func (s *Service) ListTaskDefinitions(input *ecs.ListTaskDefinitionsInput, accountID string) (*ecs.ListTaskDefinitionsOutput, error) {
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	prefix := TaskDefFamiliesPrefix()
+	if fam := aws.StringValue(input.FamilyPrefix); fam != "" {
+		prefix = TaskDefRevsPrefix(fam)
+	}
+	keys, err := keysWithPrefix(kv, prefix)
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.ListTaskDefinitionsOutput{}
+	for _, k := range keys {
+		if !strings.Contains(k, "/revs/") {
+			continue
+		}
+		var rec TaskDefRecord
+		found, err := getJSON(kv, k, &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out.TaskDefinitionArns = append(out.TaskDefinitionArns, aws.String(rec.ARN))
+		}
+	}
+	return out, nil
+}
+
+// resolveTaskDef loads the TaskDefRecord named by ref ("family", "family:rev",
+// or a task-definition ARN). A bare family resolves to its latest revision.
+func (s *Service) resolveTaskDef(kv nats.KeyValue, ref string) (*TaskDefRecord, error) {
+	family, rev := parseTaskDefRef(ref)
+	if family == "" {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	if rev == 0 {
+		var latest int
+		found, err := getJSON(kv, TaskDefLatestRevKey(family), &latest)
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		rev = latest
+	}
+	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, nil
+}
+
+// parseTaskDefRef splits "family", "family:rev" or an ARN into (family, rev).
+// rev is 0 when unspecified (caller resolves to latest).
+func parseTaskDefRef(ref string) (string, int) {
+	ref = strings.TrimSpace(ref)
+	if i := strings.LastIndex(ref, "task-definition/"); i >= 0 {
+		ref = ref[i+len("task-definition/"):]
+	}
+	family := ref
+	rev := 0
+	if i := strings.LastIndexByte(ref, ':'); i >= 0 {
+		family = ref[:i]
+		if n, err := strconv.Atoi(ref[i+1:]); err == nil {
+			rev = n
+		}
+	}
+	return family, rev
+}
+
+func (r *TaskDefRecord) toAWS() *ecs.TaskDefinition {
+	td := &ecs.TaskDefinition{
+		Family:            aws.String(r.Family),
+		Revision:          aws.Int64(int64(r.Revision)),
+		TaskDefinitionArn: aws.String(r.ARN),
+		Status:            aws.String(r.Status),
+	}
+	if r.NetworkMode != "" {
+		td.NetworkMode = aws.String(r.NetworkMode)
+	}
+	if r.CPU != "" {
+		td.Cpu = aws.String(r.CPU)
+	}
+	if r.Memory != "" {
+		td.Memory = aws.String(r.Memory)
+	}
+	for _, c := range r.Containers {
+		td.ContainerDefinitions = append(td.ContainerDefinitions, c.toAWS())
+	}
+	return td
+}

spinifex/handlers/ecs/service_nats.go

@@ -0,0 +1,80 @@
+package handlers_ecs
+
+import (
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+)
+
+const defaultTimeout = 30 * time.Second
+
+// NATSECSService is the gateway-side adapter that forwards every ECSService
+// method as a NATS request to the daemon's matching subscriber.
+type NATSECSService struct {
+	natsConn *nats.Conn
+}
+
+var _ ECSService = (*NATSECSService)(nil)
+
+// NewNATSECSService returns an ECSService that uses NATS request-response.
+func NewNATSECSService(conn *nats.Conn) ECSService {
+	return &NATSECSService{natsConn: conn}
+}
+
+// --- Cluster ---
+
+func (s *NATSECSService) CreateCluster(input *ecs.CreateClusterInput, accountID string) (*ecs.CreateClusterOutput, error) {
+	return utils.NATSRequest[ecs.CreateClusterOutput](s.natsConn, "ecs.CreateCluster", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) DescribeClusters(input *ecs.DescribeClustersInput, accountID string) (*ecs.DescribeClustersOutput, error) {
+	return utils.NATSRequest[ecs.DescribeClustersOutput](s.natsConn, "ecs.DescribeClusters", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) ListClusters(input *ecs.ListClustersInput, accountID string) (*ecs.ListClustersOutput, error) {
+	return utils.NATSRequest[ecs.ListClustersOutput](s.natsConn, "ecs.ListClusters", input, defaultTimeout, accountID)
+}
+
+// --- Task definition ---
+
+func (s *NATSECSService) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, accountID string) (*ecs.RegisterTaskDefinitionOutput, error) {
+	return utils.NATSRequest[ecs.RegisterTaskDefinitionOutput](s.natsConn, "ecs.RegisterTaskDefinition", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error) {
+	return utils.NATSRequest[ecs.DescribeTaskDefinitionOutput](s.natsConn, "ecs.DescribeTaskDefinition", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) ListTaskDefinitions(input *ecs.ListTaskDefinitionsInput, accountID string) (*ecs.ListTaskDefinitionsOutput, error) {
+	return utils.NATSRequest[ecs.ListTaskDefinitionsOutput](s.natsConn, "ecs.ListTaskDefinitions", input, defaultTimeout, accountID)
+}
+
+// --- Container instance ---
+
+func (s *NATSECSService) RegisterContainerInstance(input *ecs.RegisterContainerInstanceInput, accountID string) (*ecs.RegisterContainerInstanceOutput, error) {
+	return utils.NATSRequest[ecs.RegisterContainerInstanceOutput](s.natsConn, "ecs.RegisterContainerInstance", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) DescribeContainerInstances(input *ecs.DescribeContainerInstancesInput, accountID string) (*ecs.DescribeContainerInstancesOutput, error) {
+	return utils.NATSRequest[ecs.DescribeContainerInstancesOutput](s.natsConn, "ecs.DescribeContainerInstances", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error) {
+	return utils.NATSRequest[ecs.ListContainerInstancesOutput](s.natsConn, "ecs.ListContainerInstances", input, defaultTimeout, accountID)
+}
+
+// --- Task ---
+
+func (s *NATSECSService) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTaskOutput, error) {
+	return utils.NATSRequest[ecs.RunTaskOutput](s.natsConn, "ecs.RunTask", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error) {
+	return utils.NATSRequest[ecs.DescribeTasksOutput](s.natsConn, "ecs.DescribeTasks", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) ListTasks(input *ecs.ListTasksInput, accountID string) (*ecs.ListTasksOutput, error) {
+	return utils.NATSRequest[ecs.ListTasksOutput](s.natsConn, "ecs.ListTasks", input, defaultTimeout, accountID)
+}

spinifex/handlers/ecs/service_test.go

@@ -0,0 +1,292 @@
+package handlers_ecs
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const testRegion = "ap-southeast-2"
+
+func newTestService(t *testing.T) (*Service, *nats.Conn) {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	return NewService(nc, testRegion, "internal"), nc
+}
+
+func TestService_CreateCluster_Idempotent(t *testing.T) {
+	svc, _ := newTestService(t)
+	out, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "ACTIVE", aws.StringValue(out.Cluster.Status))
+	assert.Equal(t, ClusterARN(testRegion, testAccountID, "web"), aws.StringValue(out.Cluster.ClusterArn))
+
+	// Recreate returns the same record, not a duplicate.
+	_, err = svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+
+	list, err := svc.ListClusters(&ecs.ListClustersInput{}, testAccountID)
+	require.NoError(t, err)
+	assert.Len(t, list.ClusterArns, 1)
+}
+
+func TestService_DescribeClusters_DefaultAndMissing(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(&ecs.CreateClusterInput{}, testAccountID) // implicit "default"
+	require.NoError(t, err)
+
+	out, err := svc.DescribeClusters(&ecs.DescribeClustersInput{}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, out.Clusters, 1)
+	assert.Equal(t, defaultCluster, aws.StringValue(out.Clusters[0].ClusterName))
+
+	miss, err := svc.DescribeClusters(&ecs.DescribeClustersInput{Clusters: []*string{aws.String("nope")}}, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, miss.Clusters)
+}
+
+func registerTaskDef(t *testing.T, svc *Service, family string, cpu, mem int) *ecs.RegisterTaskDefinitionOutput {
+	t.Helper()
+	out, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{
+		Family: aws.String(family),
+		ContainerDefinitions: []*ecs.ContainerDefinition{{
+			Name:      aws.String("app"),
+			Image:     aws.String("registry/app:1"),
+			Cpu:       aws.Int64(int64(cpu)),
+			Memory:    aws.Int64(int64(mem)),
+			Essential: aws.Bool(true),
+		}},
+	}, testAccountID)
+	require.NoError(t, err)
+	return out
+}
+
+func TestService_RegisterTaskDefinition_RevisionBump(t *testing.T) {
+	svc, _ := newTestService(t)
+	r1 := registerTaskDef(t, svc, "nginx", 128, 256)
+	assert.Equal(t, int64(1), aws.Int64Value(r1.TaskDefinition.Revision))
+	r2 := registerTaskDef(t, svc, "nginx", 128, 256)
+	assert.Equal(t, int64(2), aws.Int64Value(r2.TaskDefinition.Revision))
+
+	// Bare family resolves to latest.
+	d, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("nginx")}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), aws.Int64Value(d.TaskDefinition.Revision))
+
+	// family:rev resolves the pinned revision.
+	d1, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("nginx:1")}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, int64(1), aws.Int64Value(d1.TaskDefinition.Revision))
+
+	list, err := svc.ListTaskDefinitions(&ecs.ListTaskDefinitionsInput{}, testAccountID)
+	require.NoError(t, err)
+	assert.Len(t, list.TaskDefinitionArns, 2)
+}
+
+func TestService_RegisterTaskDefinition_NoFamily(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{}, testAccountID)
+	require.Error(t, err)
+}
+
+func TestService_DescribeTaskDefinition_Unknown(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("ghost")}, testAccountID)
+	require.Error(t, err)
+}
+
+// registerInstance seeds an ACTIVE container instance with capacity via the bus
+// register path (the way the agent registers in production).
+func registerInstance(t *testing.T, svc *Service, cluster, id string, cpu, mem int) {
+	t.Helper()
+	require.NoError(t, svc.recordRegister(&bus.RegisterInstance{
+		AccountID:   testAccountID,
+		ClusterName: cluster,
+		InstanceID:  id,
+		Capacity:    bus.InstanceCapacity{CPU: cpu, MemoryMiB: mem},
+	}))
+}
+
+func TestService_RunTask_PlacesAndAssigns(t *testing.T) {
+	svc, nc := newTestService(t)
+	_, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerTaskDef(t, svc, "app", 128, 256)
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+
+	sub, err := nc.SubscribeSync(bus.AssignSubject(testAccountID, "web", "i-1"))
+	require.NoError(t, err)
+
+	out, err := svc.RunTask(&ecs.RunTaskInput{
+		Cluster:        aws.String("web"),
+		TaskDefinition: aws.String("app"),
+		Count:          aws.Int64(1),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, out.Tasks, 1)
+	assert.Empty(t, out.Failures)
+	assert.Equal(t, "PENDING", aws.StringValue(out.Tasks[0].LastStatus))
+
+	// Assign published to the instance's agent.
+	msg, err := sub.NextMsg(2 * time.Second)
+	require.NoError(t, err)
+	var as bus.Assign
+	require.NoError(t, json.Unmarshal(msg.Data, &as))
+	assert.Equal(t, "i-1", as.InstanceID)
+	require.Len(t, as.Containers, 1)
+	assert.Equal(t, "registry/app:1", as.Containers[0].Image)
+
+	// Capacity reserved on the instance.
+	di, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{
+		Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, di.ContainerInstances, 1)
+	assert.Equal(t, int64(1), aws.Int64Value(di.ContainerInstances[0].RunningTasksCount))
+
+	// Task visible via Describe/List.
+	lt, err := svc.ListTasks(&ecs.ListTasksInput{Cluster: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	assert.Len(t, lt.TaskArns, 1)
+	assert.Equal(t, as.TaskID, containerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn)))
+}
+
+func TestService_RunTask_ClusterNotFound(t *testing.T) {
+	svc, _ := newTestService(t)
+	registerTaskDef(t, svc, "app", 1, 1)
+	_, err := svc.RunTask(&ecs.RunTaskInput{Cluster: aws.String("ghost"), TaskDefinition: aws.String("app")}, testAccountID)
+	require.Error(t, err)
+}
+
+func TestService_RunTask_NoCapacityFails(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerTaskDef(t, svc, "big", 100, 100000) // 100GB memory, won't fit
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+
+	out, err := svc.RunTask(&ecs.RunTaskInput{
+		Cluster: aws.String("web"), TaskDefinition: aws.String("big"), Count: aws.Int64(1),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, out.Tasks)
+	require.Len(t, out.Failures, 1)
+}
+
+// Task-state RUNNING then STOPPED updates the task and releases capacity.
+func TestService_RecordTaskState_ReleasesCapacityOnStop(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerTaskDef(t, svc, "app", 128, 256)
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+	out, err := svc.RunTask(&ecs.RunTaskInput{Cluster: aws.String("web"), TaskDefinition: aws.String("app")}, testAccountID)
+	require.NoError(t, err)
+	taskID := containerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn))
+
+	require.NoError(t, svc.recordTaskState(&bus.TaskState{
+		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID,
+		LastStatus: bus.TaskStatusRunning,
+	}))
+	exit := 0
+	require.NoError(t, svc.recordTaskState(&bus.TaskState{
+		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID,
+		LastStatus: bus.TaskStatusStopped, Reason: "exited",
+		Containers: []bus.ContainerStatus{{Name: "app", Status: bus.TaskStatusStopped, ExitCode: &exit}},
+	}))
+
+	dt, err := svc.DescribeTasks(&ecs.DescribeTasksInput{Cluster: aws.String("web"), Tasks: []*string{aws.String(taskID)}}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, dt.Tasks, 1)
+	assert.Equal(t, "STOPPED", aws.StringValue(dt.Tasks[0].LastStatus))
+
+	// Capacity fully released: remaining == total.
+	di, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{
+		Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, int64(0), aws.Int64Value(di.ContainerInstances[0].RunningTasksCount))
+	for _, r := range di.ContainerInstances[0].RemainingResources {
+		switch aws.StringValue(r.Name) {
+		case "CPU":
+			assert.Equal(t, int64(1024), aws.Int64Value(r.IntegerValue))
+		case "MEMORY":
+			assert.Equal(t, int64(2048), aws.Int64Value(r.IntegerValue))
+		}
+	}
+}
+
+func TestService_RecordHeartbeat_UpdatesStatus(t *testing.T) {
+	svc, _ := newTestService(t)
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+	require.NoError(t, svc.recordHeartbeat(&bus.Heartbeat{
+		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", Status: bus.StatusDraining,
+	}))
+	di, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{
+		Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "DRAINING", aws.StringValue(di.ContainerInstances[0].Status))
+}
+
+// The reaper marks a stale instance DRAINING and stops its tasks.
+func TestScheduler_ReapBucket_StopsStaleInstanceTasks(t *testing.T) {
+	svc, nc := newTestService(t)
+	_, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerTaskDef(t, svc, "app", 128, 256)
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+	out, err := svc.RunTask(&ecs.RunTaskInput{Cluster: aws.String("web"), TaskDefinition: aws.String("app")}, testAccountID)
+	require.NoError(t, err)
+	taskID := containerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn))
+
+	// Backdate LastSeen beyond the heartbeat timeout.
+	kv, err := svc.bucket(testAccountID)
+	require.NoError(t, err)
+	var rec InstanceRecord
+	_, err = getJSON(kv, InstanceKey("web", "i-1"), &rec)
+	require.NoError(t, err)
+	rec.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
+	require.NoError(t, putJSON(kv, InstanceKey("web", "i-1"), &rec))
+
+	sc := NewScheduler(nc, svc, "test-holder")
+	sc.reapBucket(kv, testAccountID, time.Now().UTC())
+
+	di, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{
+		Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "DRAINING", aws.StringValue(di.ContainerInstances[0].Status))
+
+	dt, err := svc.DescribeTasks(&ecs.DescribeTasksInput{Cluster: aws.String("web"), Tasks: []*string{aws.String(taskID)}}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "STOPPED", aws.StringValue(dt.Tasks[0].LastStatus))
+	assert.Equal(t, stoppedReasonReaped, aws.StringValue(dt.Tasks[0].StoppedReason))
+}
+
+// Leader election: first holder wins Create; a second holder is rejected.
+func TestScheduler_AcquireLease_SingleLeader(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	svc := NewService(nc, testRegion, "")
+	a := NewScheduler(nc, svc, "holder-a")
+	b := NewScheduler(nc, svc, "holder-b")
+	assert.True(t, a.acquireOrRefresh())
+	assert.False(t, b.acquireOrRefresh())
+	assert.True(t, a.acquireOrRefresh()) // refresh keeps leadership
+}
+
+func TestAccountIDFromBucket(t *testing.T) {
+	id, ok := accountIDFromBucket(AccountBucketName(testAccountID))
+	assert.True(t, ok)
+	assert.Equal(t, testAccountID, id)
+	_, ok = accountIDFromBucket("some-other-bucket")
+	assert.False(t, ok)
+}

spinifex/handlers/ecs/tasks.go

@@ -0,0 +1,244 @@
+package handlers_ecs
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/google/uuid"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	"github.com/nats-io/nats.go"
+)
+
+// reservePlacementRetries bounds the CAS retry loop when concurrent RunTask calls
+// contend for the same instance's capacity.
+const reservePlacementRetries = 5
+
+// RunTask places input.Count tasks of a definition onto container instances via
+// bin-pack, reserves their capacity in KV, writes PENDING task records, and
+// publishes an assign on the Layer-2 bus for each. Placement failures for a task
+// are returned as RunTask failures; already-placed tasks in the same call stay.
+func (s *Service) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTaskOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	var clusterRec ClusterRecord
+	found, err := getJSON(kv, ClusterMetaKey(cluster), &clusterRec)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		return nil, errors.New(awserrors.ErrorECSClusterNotFound)
+	}
+
+	taskDef, err := s.resolveTaskDef(kv, aws.StringValue(input.TaskDefinition))
+	if err != nil {
+		return nil, err
+	}
+
+	count := int(aws.Int64Value(input.Count))
+	if count <= 0 {
+		count = 1
+	}
+	strategy := placementStrategyFromAWS(input.PlacementStrategy)
+	cpu, mem := taskDef.reservedCPU(), taskDef.reservedMemory()
+
+	out := &ecs.RunTaskOutput{}
+	for i := 0; i < count; i++ {
+		taskID := uuid.NewString()
+		inst, err := s.reservePlacement(kv, cluster, taskID, cpu, mem, strategy)
+		if err != nil {
+			out.Failures = append(out.Failures, &ecs.Failure{
+				Reason: aws.String("RESOURCE:placement"),
+				Detail: aws.String(err.Error()),
+			})
+			continue
+		}
+
+		rec := s.newTaskRecord(accountID, cluster, taskID, taskDef, inst, cpu, mem)
+		if err := putJSON(kv, TaskKey(cluster, taskID), rec); err != nil {
+			return nil, err
+		}
+		if err := s.publishAssign(accountID, cluster, inst.InstanceID, rec, taskDef); err != nil {
+			slog.Error("ECS RunTask: failed to publish assign", "task", taskID, "instance", inst.InstanceID, "err", err)
+		}
+		out.Tasks = append(out.Tasks, s.taskToAWS(accountID, rec))
+	}
+	return out, nil
+}
+
+// reservePlacement bin-packs a task onto an ACTIVE instance and commits the
+// reservation under a KV CAS, retrying on contention.
+func (s *Service) reservePlacement(kv nats.KeyValue, cluster, taskID string, cpu, mem int, strategy string) (*InstanceRecord, error) {
+	for range reservePlacementRetries {
+		instances, err := s.listInstanceRecords(kv, cluster)
+		if err != nil {
+			return nil, err
+		}
+		chosen, err := placeTask(instances, cpu, mem, strategy)
+		if err != nil {
+			return nil, err
+		}
+
+		// Re-read the chosen instance for its current revision, then CAS-update.
+		entry, err := kv.Get(InstanceKey(cluster, chosen.InstanceID))
+		if err != nil {
+			continue
+		}
+		var live InstanceRecord
+		if uerr := json.Unmarshal(entry.Value(), &live); uerr != nil {
+			return nil, uerr
+		}
+		if !live.fits(cpu, mem) {
+			continue
+		}
+		live.ReservedCPU += cpu
+		live.ReservedMemoryMiB += mem
+		live.PlacedTasks = append(live.PlacedTasks, taskID)
+		data, merr := json.Marshal(&live)
+		if merr != nil {
+			return nil, merr
+		}
+		if _, uerr := kv.Update(InstanceKey(cluster, live.InstanceID), data, entry.Revision()); uerr != nil {
+			continue // lost the CAS race; retry placement
+		}
+		return &live, nil
+	}
+	return nil, fmt.Errorf("placement contended after %d attempts", reservePlacementRetries)
+}
+
+// newTaskRecord builds a PENDING task record for a placed task.
+func (s *Service) newTaskRecord(accountID, cluster, taskID string, td *TaskDefRecord, inst *InstanceRecord, cpu, mem int) *TaskRecord {
+	now := time.Now().UTC()
+	rec := &TaskRecord{
+		TaskID:               taskID,
+		ARN:                  TaskARN(s.region, accountID, cluster, taskID),
+		Cluster:              cluster,
+		TaskDefFamily:        td.Family,
+		TaskDefRevision:      td.Revision,
+		TaskDefARN:           td.ARN,
+		ContainerInstanceID:  inst.InstanceID,
+		ContainerInstanceARN: inst.ARN,
+		DesiredStatus:        TaskStatusRunning,
+		LastStatus:           TaskStatusPending,
+		ReservedCPU:          cpu,
+		ReservedMemoryMiB:    mem,
+		CreatedAt:            now,
+	}
+	for _, c := range td.Containers {
+		rec.Containers = append(rec.Containers, ContainerState{Name: c.Name, Status: TaskStatusPending})
+	}
+	return rec
+}
+
+// publishAssign sends the task assignment to the instance's agent over the bus.
+func (s *Service) publishAssign(accountID, cluster, instanceID string, rec *TaskRecord, td *TaskDefRecord) error {
+	msg := bus.Assign{
+		AccountID:       accountID,
+		ClusterName:     cluster,
+		InstanceID:      instanceID,
+		TaskID:          rec.TaskID,
+		TaskARN:         rec.ARN,
+		TaskDefFamily:   td.Family,
+		TaskDefRevision: td.Revision,
+		AssignedAt:      time.Now().UTC(),
+	}
+	for _, c := range td.Containers {
+		msg.Containers = append(msg.Containers, c.toAssignContainer())
+	}
+	data, err := json.Marshal(&msg)
+	if err != nil {
+		return err
+	}
+	subj := bus.AssignSubject(accountID, cluster, instanceID)
+	return s.nc.Publish(subj, data)
+}
+
+// DescribeTasks returns task records for the named tasks in a cluster.
+func (s *Service) DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.DescribeTasksOutput{}
+	for _, ref := range awsStringSlice(input.Tasks) {
+		taskID := containerInstanceShortID(ref)
+		var rec TaskRecord
+		found, err := getJSON(kv, TaskKey(cluster, taskID), &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out.Tasks = append(out.Tasks, s.taskToAWS(accountID, &rec))
+		} else {
+			out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")})
+		}
+	}
+	return out, nil
+}
+
+// ListTasks returns the ARNs of all tasks in a cluster.
+func (s *Service) ListTasks(input *ecs.ListTasksInput, accountID string) (*ecs.ListTasksOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	keys, err := keysWithPrefix(kv, TasksPrefix(cluster))
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.ListTasksOutput{}
+	for _, k := range keys {
+		var rec TaskRecord
+		found, err := getJSON(kv, k, &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out.TaskArns = append(out.TaskArns, aws.String(rec.ARN))
+		}
+	}
+	return out, nil
+}
+
+func (s *Service) taskToAWS(accountID string, r *TaskRecord) *ecs.Task {
+	t := &ecs.Task{
+		TaskArn:              aws.String(r.ARN),
+		ClusterArn:           aws.String(ClusterARN(s.region, accountID, r.Cluster)),
+		TaskDefinitionArn:    aws.String(r.TaskDefARN),
+		DesiredStatus:        aws.String(r.DesiredStatus),
+		LastStatus:           aws.String(r.LastStatus),
+		ContainerInstanceArn: aws.String(r.ContainerInstanceARN),
+	}
+	if r.StoppedReason != "" {
+		t.StoppedReason = aws.String(r.StoppedReason)
+	}
+	for _, c := range r.Containers {
+		ctr := &ecs.Container{Name: aws.String(c.Name), LastStatus: aws.String(c.Status)}
+		if c.ExitCode != nil {
+			ctr.ExitCode = aws.Int64(int64(*c.ExitCode))
+		}
+		t.Containers = append(t.Containers, ctr)
+	}
+	return t
+}
+
+// placementStrategyFromAWS extracts the first strategy type, defaulting binpack.
+func placementStrategyFromAWS(in []*ecs.PlacementStrategy) string {
+	for _, p := range in {
+		if p != nil {
+			return aws.StringValue(p.Type)
+		}
+	}
+	return StrategyBinpack
+}