feat(ecs): service controller + ELBv2 target-group registration (Sprint 4f) (#420)

* feat(ecsgw): on-VM SigV4 client for the ECS agent→gateway channel

Adds internal/ecsgw, the twin of internal/eksgw: a TLS + SigV4 (service "ecs")
client the ecs-agent will use to reach the AWS gateway over HTTPS instead of
connecting to host NATS directly. Calls are AWS JSON 1.1 with an X-Amz-Target
action; the gateway relays onto ecs.* host-side, keeping NATS cluster-internal.

* feat(ecs): SubmitTaskStateChange gateway action → recordTaskState

Wires the AWS-API SubmitTaskStateChange action end to end (gateway action →
ecs.SubmitTaskStateChange NATS → daemon → Service.SubmitTaskStateChange) so a
gateway-routed agent reports task state without touching the NATS bus. The
handler maps the SDK input onto the same bus.TaskState shape and converges on
recordTaskState; the account is authoritative from the SigV4 caller, so an
instance cannot report state for another account's task.

* feat(ecs): durable assign KV inbox + PollAssignments gateway action

Scheduler writes task assigns into a per-instance KV inbox
(clusters/{cluster}/assignments/{instance}/{task}) instead of a
fire-and-forget core-NATS publish, and a new PollAssignments gateway
action lets a gateway-routed agent drain its inbox over TLS+SigV4.

At-least-once: an unacked assign is re-delivered on the next poll;
the agent acks accepted assigns (AckTaskIDs) and STOPPED transitions
reclaim stale inbox entries. The Poll response carries bus.Assign with
an RFC3339 time, so it is encoded with encoding/json (RawJSONActions)
rather than the jsonutil epoch-second marshaler.

* feat(ecs-agent): rewire control-plane off NATS onto the AWS gateway

The agent no longer connects to NATS. It registers, heartbeats, reports
task state, and receives assignments entirely through the AWS gateway
over TLS+SigV4, using seeded IAM credentials — keeping the NATS bus
host-internal and closing the trust-boundary gap (a tenant container
instance must not hold a shared bus token).

- controlPlane seam over ecsgw.Client replaces the NATS publisher:
  Register (RegisterContainerInstance), SubmitTaskState
  (SubmitTaskStateChange), PollAssignments.
- Heartbeat folds into a periodic idempotent re-register (refreshes
  LastSeen); bus.Heartbeat is dropped from the agent.
- subscribeAssign (NATS subscribe) becomes a poll loop: drain the
  inbox, dispatch each assign once, ack accepted taskIDs on the next
  poll (at-least-once).
- config: ECS_ACCESS_KEY / ECS_SECRET_KEY / ECS_POLL_INTERVAL added,
  ECS_NATS_URL removed; cloud-init initd seeds the IAM creds.

Tests move from the NATS publisher fake to a controlPlane fake.

* feat(ecs): cloud-init cred-seed + ecs-node bring-up script

ECS container instances are plain EC2 booted from the spinifex-ecs-node
AMI (AWS-faithful: no ECS-specific launch API). This renders the
cloud-init user-data that seeds /etc/spinifex-ecs/agent.env (gateway
URL/CA/region/cluster + IAM creds) and the gateway CA, then runs the
instance. The agent registers, heartbeats, polls assignments, and
reports task state entirely through the AWS gateway over TLS+SigV4 —
never NATS.

The guest dials the customer-facing gateway address (network advertise
IP), not the mgmt bridge, mirroring the EKS GatewayURL/AddonGatewayURL
split. Seeds the system key for now; scoped per-instance creds are a
follow-up (4g task-role endpoint).

* fix(ecs): reconcile dev merge — restore #418 ENI release + netns wiring

The 'Merge branch dev' resolution dropped code where Sprint 4d (#418,
awsvpc ENI-per-task) and the gateway control-plane rewrite (#419) touch
the same paths:

- recordTaskState STOPPED branch: reclaimTaskENI was clobbered by
  reclaimAssignInbox; both are needed (release the task ENI AND drain the
  KV inbox). Restores TestRecordTaskState_Awsvpc_ReleasesENIOnStopOnce.
- ecs-agent Agent: the netns field, its init, and the setupTaskNetns/
  teardownTaskNetns wrappers were lost when the controlPlane Agent struct
  won the merge, leaving taskrun.go calling undefined methods.
- eni_test assign assertion read the assign over NATS; #419 delivers it via
  the durable KV inbox, so it now polls via PollAssignments.
- taskrun_test netns cases used the removed historyPub fake; switched to
  fakeCP to match the controlPlane seam.

* feat(ecs): service controller + ELBv2 target-group registration (Sprint 4f)

Land the ECS Service controller on top of the 4e scheduler/RunTask flow:

- CreateService/UpdateService/DeleteService/DescribeServices/ListServices.
  REPLICA only; DAEMON and serviceRegistries rejected in v1 (Q15).
- StartTask (explicit placement) and StopTask.
- Leader-driven replica reconciliation: a 10s scheduler tick converges each
  service's running task count on desiredCount, launching the shortfall via
  RunTask (tasks tagged group=service:{name}) and force-stopping the surplus.
- ELBv2 target-group registration: the scheduler registers a task's awsvpc ENI
  IP on the RUNNING transition and deregisters on STOPPED via the existing
  elbv2.RegisterTargets/DeregisterTargets NATS handlers (Q8, single-writer).
- forceStopTask consolidates the control-plane stop (STOPPED state, capacity +
  ENI release, inbox drain, TG deregister); reused by StopTask, scale-in, and
  the heartbeat reaper (the reaper now deregisters targets per Q13).
- Wire the 7 new actions through the gateway action map, NATS adapter, and
  daemon subscriptions; add ServiceNotFoundException.

* fix(ecs-agent): anonymous pull for non-ECR images + normalize image refs

The agent minted an ECR authorization token via IMDS for every image, so
public docker.io refs failed with 'IMDS returned empty role name' on hosts
without an instance role. Gate ECR auth on .dkr.ecr. refs and pull all other
registries anonymously.

Bare image strings (nginx:alpine) were handed to containerd unmodified, which
parsed them as dummy://nginx:alpine and rejected the pull. Normalize refs to
their fully-qualified form (docker.io/library/nginx:alpine, :latest default)
before Pull and GetImage.

Adds StartTask/StopTask service-controller coverage.

* fix: address ECS 4f live-verification findings

ecs-node-bringup.sh defaulted GATEWAY_IP to the mgmt bridge (10.15.8.1), which
a VPC guest cannot route to — the agent could not reach the gateway. Default to
the host's WAN-bridge IP, which guests reach by NATing out through the WAN.

A target group not forwarded to by any listener reported its targets stuck in
'initial' forever (no health checker advances a detached TG). DescribeTargetHealth
now reports such targets as 'unused' with reason Target.NotInUse, matching AWS.

cmd/ecs-agent/ecrresolver.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"net/http"
+	"strings"
 	"sync"
 
 	"github.com/mulgadc/spinifex/cmd/ecs-agent/credentials"
@@ -59,8 +60,24 @@ func (e *ecrResolver) client() (*http.Client, error) {
 	return c, nil
 }
 
+// isECRRef reports whether ref targets an ECR registry, i.e. its registry host
+// is "<acct>.dkr.ecr.<region>.<domain>". Only ECR pulls need a minted token;
+// public registries (docker.io and friends) pull anonymously, so refs without
+// an ECR host must not drag in IMDS host credentials.
+func isECRRef(ref string) bool {
+	host, _, _ := strings.Cut(ref, "/")
+	if !strings.Contains(host, ".") && !strings.Contains(host, ":") {
+		return false // no registry host -> docker.io default
+	}
+	return strings.Contains(host, ".dkr.ecr.")
+}
+
 // Authorize returns the ECR token's user/password and proxy endpoint for ref.
+// Non-ECR refs resolve to anonymous pull (empty credentials).
 func (e *ecrResolver) Authorize(ctx context.Context, ref string) (user, pass, endpoint string, err error) {
+	if !isECRRef(ref) {
+		return "", "", "", nil
+	}
 	c, err := e.creds.Retrieve(ctx)
 	if err != nil {
 		return "", "", "", err

cmd/ecs-agent/ecrresolver_test.go

@@ -78,7 +78,7 @@ func TestECRResolver_Authorize(t *testing.T) {
 
 func TestECRResolver_CredsError(t *testing.T) {
 	r := newECRResolver(stubCreds{err: errors.New("no imds")}, "us-east-1", "https://x", http.DefaultClient)
-	if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil {
+	if _, _, _, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1"); err == nil {
 		t.Fatal("expected creds error to propagate")
 	}
 }
@@ -87,7 +87,7 @@ func TestECRResolver_GatewayError(t *testing.T) {
 	srv := gatewayStub(t, "", "", http.StatusInternalServerError)
 	creds := stubCreds{c: credentials.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}}
 	r := newECRResolver(creds, "us-east-1", srv.URL, trusting(srv))
-	if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil {
+	if _, _, _, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1"); err == nil {
 		t.Fatal("expected gateway error")
 	}
 }
@@ -96,7 +96,36 @@ func TestECRResolver_LazyClient_BadCADeferredToAuthorize(t *testing.T) {
 	creds := stubCreds{c: credentials.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}}
 	// Construction with a bogus CA path must not fail — only Authorize does.
 	r := newLazyECRResolver(creds, "us-east-1", "https://gw", filepath.Join(t.TempDir(), "absent.pem"))
-	if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil {
+	if _, _, _, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1"); err == nil {
 		t.Fatal("expected lazy gateway-client build to fail on missing CA")
 	}
 }
+
+// TestECRResolver_NonECRAnonymous asserts that a public-registry ref pulls
+// anonymously: Authorize returns empty creds without ever touching IMDS, even
+// when the credentials provider would error.
+func TestECRResolver_NonECRAnonymous(t *testing.T) {
+	r := newECRResolver(stubCreds{err: errors.New("imds must not be called")}, "us-east-1", "https://x", http.DefaultClient)
+	for _, ref := range []string{"nginx:alpine", "docker.io/library/nginx:alpine", "ghcr.io/org/app:1"} {
+		user, pass, endpoint, err := r.Authorize(context.Background(), ref)
+		if err != nil || user != "" || pass != "" || endpoint != "" {
+			t.Errorf("ref %q: got (%q,%q,%q,%v), want anonymous", ref, user, pass, endpoint, err)
+		}
+	}
+}
+
+func TestIsECRRef(t *testing.T) {
+	cases := map[string]bool{
+		"nginx:alpine":            false,
+		"docker.io/library/nginx": false,
+		"ghcr.io/org/app:1":       false,
+		"registry:5000/app:1":     false,
+		"123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1": true,
+		"123456789012.dkr.ecr.ap-southeast-2.amazonaws.com/x":    true,
+	}
+	for ref, want := range cases {
+		if got := isECRRef(ref); got != want {
+			t.Errorf("isECRRef(%q) = %v, want %v", ref, got, want)
+		}
+	}
+}

cmd/ecs-agent/runtime/containerd.go

@@ -9,6 +9,7 @@ import (
 	"github.com/containerd/containerd/v2/core/remotes"
 	"github.com/containerd/containerd/v2/core/remotes/docker"
 	"github.com/containerd/containerd/v2/pkg/namespaces"
+	"github.com/distribution/reference"
 )
 
 // ecsNamespace isolates ecs-agent images/containers in containerd, mirroring the
@@ -51,27 +52,44 @@ func New(socket string) (Runtime, error) {
 func (p *containerdPuller) Pull(ctx context.Context, spec PullSpec, r Resolver) (Image, error) {
 	ctx = namespaces.WithNamespace(ctx, ecsNamespace)
 
-	resolver, err := newDockerResolver(ctx, spec.Ref, r)
+	ref, err := normalizeRef(spec.Ref)
 	if err != nil {
 		return Image{}, err
 	}
 
-	img, err := p.client.Pull(ctx, spec.Ref,
+	resolver, err := newDockerResolver(ctx, ref, r)
+	if err != nil {
+		return Image{}, err
+	}
+
+	img, err := p.client.Pull(ctx, ref,
 		containerd.WithPullUnpack,
 		containerd.WithResolver(resolver),
 	)
 	if err != nil {
-		return Image{}, fmt.Errorf("pull %s: %w", spec.Ref, err)
+		return Image{}, fmt.Errorf("pull %s: %w", ref, err)
 	}
 
 	target := img.Target()
 	return Image{
-		Ref:    spec.Ref,
+		Ref:    ref,
 		Digest: target.Digest.String(),
 		Size:   target.Size,
 	}, nil
 }
 
+// normalizeRef expands a task-definition image string into the fully-qualified
+// reference containerd's resolver requires: a bare "nginx:alpine" becomes
+// "docker.io/library/nginx:alpine" and an untagged ref gets ":latest". Without
+// this, containerd parses the bare ref as "dummy://nginx:alpine" and rejects it.
+func normalizeRef(ref string) (string, error) {
+	named, err := reference.ParseNormalizedNamed(ref)
+	if err != nil {
+		return "", fmt.Errorf("parse image ref %q: %w", ref, err)
+	}
+	return reference.TagNameOnly(named).String(), nil
+}
+
 // Close releases the containerd client connection.
 func (p *containerdPuller) Close() error {
 	if p.client == nil {

cmd/ecs-agent/runtime/containerd_run.go

@@ -21,9 +21,13 @@ var _ Runner = (*containerdPuller)(nil)
 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)
+	ref, err := normalizeRef(spec.Image)
 	if err != nil {
-		return "", fmt.Errorf("get image %s: %w", spec.Image, err)
+		return "", err
+	}
+	image, err := p.client.GetImage(ctx, ref)
+	if err != nil {
+		return "", fmt.Errorf("get image %s: %w", ref, err)
 	}
 
 	specOpts := []oci.SpecOpts{

cmd/ecs-agent/runtime/runtime_test.go

@@ -65,3 +65,25 @@ func TestFakePuller_Close(t *testing.T) {
 		t.Fatalf("Close: err=%v closed=%v", err, f.Closed)
 	}
 }
+
+func TestNormalizeRef(t *testing.T) {
+	cases := map[string]string{
+		"nginx:alpine":      "docker.io/library/nginx:alpine",
+		"nginx":             "docker.io/library/nginx:latest",
+		"ghcr.io/org/app:1": "ghcr.io/org/app:1",
+		"123456789012.dkr.ecr.ap-southeast-2.spinifex.internal/app:1": "123456789012.dkr.ecr.ap-southeast-2.spinifex.internal/app:1",
+	}
+	for in, want := range cases {
+		got, err := normalizeRef(in)
+		if err != nil {
+			t.Errorf("normalizeRef(%q): %v", in, err)
+			continue
+		}
+		if got != want {
+			t.Errorf("normalizeRef(%q) = %q, want %q", in, got, want)
+		}
+	}
+	if _, err := normalizeRef("Bad::Ref"); err == nil {
+		t.Error("normalizeRef accepted invalid ref")
+	}
+}

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/charmbracelet/bubbletea v1.3.10
 	github.com/charmbracelet/lipgloss v1.1.0
 	github.com/containerd/containerd/v2 v2.3.2
+	github.com/distribution/reference v0.6.0
 	github.com/go-chi/chi/v5 v5.3.0
 	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/google/uuid v1.6.0
@@ -73,7 +74,6 @@ require (
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgraph-io/badger/v4 v4.9.2 // indirect
 	github.com/dgraph-io/ristretto/v2 v2.4.0 // indirect
-	github.com/distribution/reference v0.6.0 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect
 	github.com/fatih/color v1.19.0 // indirect

scripts/ecs-node-bringup.sh

@@ -14,7 +14,7 @@
 #   AWS_PROFILE=spinifex scripts/ecs-node-bringup.sh <cluster-name>
 #
 # Env overrides:
-#   GATEWAY_IP   gateway/mgmt IP the guest dials   (default 10.15.8.1)
+#   GATEWAY_IP   gateway IP the guest dials        (default: host WAN-bridge IP)
 #   GATEWAY_CA   host path to the gateway CA PEM   (default /etc/spinifex/ca.pem)
 #   AMI_ID       ECS-node AMI                      (default: latest spinifex-ecs-node)
 #   SUBNET_ID / SG_ID / KEY_NAME / INSTANCE_TYPE   (default: first available / t3.small)
@@ -23,6 +23,14 @@ set -euo pipefail
 CLUSTER="${1:?usage: ecs-node-bringup.sh <cluster-name>}"
 PROFILE="${AWS_PROFILE:-spinifex}"
 REGION="$(aws configure get region --profile "$PROFILE")"
+
+# VPC guests NAT out through the WAN and reach host-run services on the host's
+# WAN-bridge IP; the mgmt bridge (10.15.8.x) is not routable from a guest netns.
+# Default to the br-wan address, falling back to the mgmt IP if it can't be read.
+default_gateway_ip() {
+  ip -4 addr show br-wan 2>/dev/null | awk '/inet /{print $2; exit}' | cut -d/ -f1
+}
+GATEWAY_IP="${GATEWAY_IP:-$(default_gateway_ip)}"
 GATEWAY_IP="${GATEWAY_IP:-10.15.8.1}"
 GATEWAY_CA="${GATEWAY_CA:-/etc/spinifex/ca.pem}"
 INSTANCE_TYPE="${INSTANCE_TYPE:-t3.small}"

spinifex/awserrors/awserrors.go

@@ -378,6 +378,7 @@ var (
 	// __type; do not cross-wire with the EC2/ACM not-found codes above.
 	ErrorECSClusterNotFound                             = "ClusterNotFoundException"
 	ErrorECSInvalidParameter                            = "InvalidParameterException"
+	ErrorECSServiceNotFound                             = "ServiceNotFoundException"
 	ErrorRetryableError                                 = "RetryableError"
 	ErrorRouteAlreadyExists                             = "RouteAlreadyExists"
 	ErrorRouteLimitExceeded                             = "RouteLimitExceeded"
@@ -900,6 +901,7 @@ var ErrorLookup = map[string]ErrorMessage{
 	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."},
+	ErrorECSServiceNotFound:                                    {HTTPCode: 400, Message: "The specified service was not found. You can view your available services with ListServices. Amazon ECS services are cluster specific and Region specific."},
 	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

@@ -963,8 +963,15 @@ func (d *Daemon) subscribeAll() error {
 			natsSub{"ecs.DescribeContainerInstances", d.handleECSDescribeContainerInstances, "spinifex-workers"},
 			natsSub{"ecs.ListContainerInstances", d.handleECSListContainerInstances, "spinifex-workers"},
 			natsSub{"ecs.RunTask", d.handleECSRunTask, "spinifex-workers"},
+			natsSub{"ecs.StartTask", d.handleECSStartTask, "spinifex-workers"},
+			natsSub{"ecs.StopTask", d.handleECSStopTask, "spinifex-workers"},
 			natsSub{"ecs.DescribeTasks", d.handleECSDescribeTasks, "spinifex-workers"},
 			natsSub{"ecs.ListTasks", d.handleECSListTasks, "spinifex-workers"},
+			natsSub{"ecs.CreateService", d.handleECSCreateService, "spinifex-workers"},
+			natsSub{"ecs.UpdateService", d.handleECSUpdateService, "spinifex-workers"},
+			natsSub{"ecs.DeleteService", d.handleECSDeleteService, "spinifex-workers"},
+			natsSub{"ecs.DescribeServices", d.handleECSDescribeServices, "spinifex-workers"},
+			natsSub{"ecs.ListServices", d.handleECSListServices, "spinifex-workers"},
 			natsSub{"ecs.SubmitTaskStateChange", d.handleECSSubmitTaskStateChange, "spinifex-workers"},
 			natsSub{"ecs.PollAssignments", d.handleECSPollAssignments, "spinifex-workers"},
 		)

spinifex/daemon/daemon_handlers_ecs.go

@@ -50,10 +50,40 @@ func (d *Daemon) handleECSRunTask(msg *nats.Msg) {
 	handleNATSRequest(msg, d.ecsService.RunTask)
 }
 
+func (d *Daemon) handleECSStartTask(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.StartTask)
+}
+
+func (d *Daemon) handleECSStopTask(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.StopTask)
+}
+
 func (d *Daemon) handleECSDescribeTasks(msg *nats.Msg) {
 	handleNATSRequest(msg, d.ecsService.DescribeTasks)
 }
 
+// --- Service ---
+
+func (d *Daemon) handleECSCreateService(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.CreateService)
+}
+
+func (d *Daemon) handleECSUpdateService(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.UpdateService)
+}
+
+func (d *Daemon) handleECSDeleteService(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.DeleteService)
+}
+
+func (d *Daemon) handleECSDescribeServices(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.DescribeServices)
+}
+
+func (d *Daemon) handleECSListServices(msg *nats.Msg) {
+	handleNATSRequest(msg, d.ecsService.ListServices)
+}
+
 func (d *Daemon) handleECSSubmitTaskStateChange(msg *nats.Msg) {
 	handleNATSRequest(msg, d.ecsService.SubmitTaskStateChange)
 }

spinifex/gateway/ecs/actions.go

@@ -104,6 +104,22 @@ func RunTask(nc *nats.Conn, accountID string, body []byte) (any, error) {
 	return handlers_ecs.NewNATSECSService(nc).RunTask(input, accountID)
 }
 
+func StartTask(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.StartTaskInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).StartTask(input, accountID)
+}
+
+func StopTask(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.StopTaskInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).StopTask(input, accountID)
+}
+
 func DescribeTasks(nc *nats.Conn, accountID string, body []byte) (any, error) {
 	input := new(ecs.DescribeTasksInput)
 	if err := unmarshalIfBody(body, input); err != nil {
@@ -120,6 +136,48 @@ func ListTasks(nc *nats.Conn, accountID string, body []byte) (any, error) {
 	return handlers_ecs.NewNATSECSService(nc).ListTasks(input, accountID)
 }
 
+// --- Service ---
+
+func CreateService(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.CreateServiceInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).CreateService(input, accountID)
+}
+
+func UpdateService(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.UpdateServiceInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).UpdateService(input, accountID)
+}
+
+func DeleteService(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.DeleteServiceInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).DeleteService(input, accountID)
+}
+
+func DescribeServices(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.DescribeServicesInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).DescribeServices(input, accountID)
+}
+
+func ListServices(nc *nats.Conn, accountID string, body []byte) (any, error) {
+	input := new(ecs.ListServicesInput)
+	if err := unmarshalIfBody(body, input); err != nil {
+		return nil, err
+	}
+	return handlers_ecs.NewNATSECSService(nc).ListServices(input, accountID)
+}
+
 // SubmitTaskStateChange is the agent's task-state report path over the gateway
 // (replaces the Layer-2 bus publish). The account is the SigV4 caller, so an
 // instance cannot report state for another account's task.

spinifex/gateway/ecs/handler.go

@@ -56,10 +56,10 @@ var Actions = map[string]Handler{
 	"ListTaskDefinitions":        ListTaskDefinitions,
 	"ListTaskDefinitionFamilies": NotImplemented,
 
-	// Task. StartTask/StopTask land with the 4f service controller + reaper.
+	// Task.
 	"RunTask":       RunTask,
-	"StartTask":     NotImplemented,
-	"StopTask":      NotImplemented,
+	"StartTask":     StartTask,
+	"StopTask":      StopTask,
 	"DescribeTasks": DescribeTasks,
 	"ListTasks":     ListTasks,
 
@@ -70,11 +70,11 @@ var Actions = map[string]Handler{
 	"PollAssignments": PollAssignments,
 
 	// Service. ListServicesByNamespace is a no-op stub in v1.
-	"CreateService":           NotImplemented,
-	"UpdateService":           NotImplemented,
-	"DeleteService":           NotImplemented,
-	"DescribeServices":        NotImplemented,
-	"ListServices":            NotImplemented,
+	"CreateService":           CreateService,
+	"UpdateService":           UpdateService,
+	"DeleteService":           DeleteService,
+	"DescribeServices":        DescribeServices,
+	"ListServices":            ListServices,
 	"ListServicesByNamespace": NotImplemented,
 
 	// Container instance.

spinifex/gateway/ecs/handler_test.go

@@ -24,7 +24,9 @@ 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,
+	"RunTask": true, "StartTask": true, "StopTask": true, "DescribeTasks": true, "ListTasks": true,
+	"CreateService": true, "UpdateService": true, "DeleteService": true,
+	"DescribeServices": true, "ListServices": true,
 	"SubmitTaskStateChange": true,
 	"PollAssignments":       true,
 }

spinifex/handlers/ecs/instances.go

@@ -225,8 +225,15 @@ func (s *Service) recordTaskState(msg *bus.TaskState) error {
 		return err
 	}
 
-	// Release capacity + reclaim the task ENI once, on the transition into STOPPED.
+	// Register the task's ELBv2 targets on the transition into RUNNING (Q8).
+	if msg.LastStatus == TaskStatusRunning && prev != TaskStatusRunning {
+		s.registerServiceTargets(kv, msg.AccountID, &task)
+	}
+
+	// Deregister targets, release capacity + reclaim the task ENI once, on the
+	// transition into STOPPED.
 	if msg.LastStatus == TaskStatusStopped && prev != TaskStatusStopped {
+		s.deregisterServiceTargets(kv, msg.AccountID, &task)
 		s.reclaimAssignInbox(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID)
 		s.reclaimTaskENI(msg.AccountID, &task)
 		return s.releaseReservation(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID, task.ReservedCPU, task.ReservedMemoryMiB)

spinifex/handlers/ecs/records.go

@@ -20,6 +20,15 @@ const (
 	TaskStatusPending = bus.TaskStatusPending
 	TaskStatusRunning = bus.TaskStatusRunning
 	TaskStatusStopped = bus.TaskStatusStopped
+
+	ServiceStatusActive   = "ACTIVE"
+	ServiceStatusDraining = "DRAINING"
+	ServiceStatusInactive = "INACTIVE"
+
+	// SchedulingStrategyReplica is the only strategy supported in v1 (Q15);
+	// DAEMON is rejected at CreateService.
+	SchedulingStrategyReplica = "REPLICA"
+	SchedulingStrategyDaemon  = "DAEMON"
 )
 
 // ARN builders for the ECS resource shapes (ecs-v1.md §1). Region + accountID
@@ -41,6 +50,27 @@ func ContainerInstanceARN(region, accountID, cluster, ciID string) string {
 	return fmt.Sprintf("arn:aws:ecs:%s:%s:container-instance/%s/%s", region, accountID, cluster, ciID)
 }
 
+func ServiceARN(region, accountID, cluster, name string) string {
+	return fmt.Sprintf("arn:aws:ecs:%s:%s:service/%s/%s", region, accountID, cluster, name)
+}
+
+// serviceTaskGroup is the AWS task-group label a service stamps on its tasks
+// ("service:{name}"). The reconciler counts a service's tasks by this group and
+// the task-state hook resolves a task back to its owning service through it.
+func serviceTaskGroup(name string) string {
+	return "service:" + name
+}
+
+// serviceNameFromGroup returns the service name encoded in a task group, or ""
+// when the group is not a service group.
+func serviceNameFromGroup(group string) string {
+	const p = "service:"
+	if len(group) > len(p) && group[:len(p)] == p {
+		return group[len(p):]
+	}
+	return ""
+}
+
 // ClusterRecord is the persisted cluster meta at ClusterMetaKey.
 type ClusterRecord struct {
 	Name      string            `json:"name"`
@@ -129,9 +159,14 @@ type ContainerState struct {
 // 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"`
+	TaskID  string `json:"taskId"`
+	ARN     string `json:"arn"`
+	Cluster string `json:"cluster"`
+	// Group / StartedBy mirror the AWS task fields. A service's tasks carry
+	// Group="service:{name}" so the reconciler counts them and the task-state
+	// hook resolves a RUNNING/STOPPED task back to its owning service.
+	Group                string           `json:"group,omitempty"`
+	StartedBy            string           `json:"startedBy,omitempty"`
 	TaskDefFamily        string           `json:"taskDefFamily"`
 	TaskDefRevision      int              `json:"taskDefRevision"`
 	TaskDefARN           string           `json:"taskDefArn"`
@@ -157,3 +192,39 @@ type TaskRecord struct {
 	StartedAt       time.Time `json:"startedAt,omitzero"`
 	StoppedAt       time.Time `json:"stoppedAt,omitzero"`
 }
+
+// LoadBalancerTarget is one ELBv2 target-group attachment on a service. On task
+// RUNNING the scheduler registers the task's ENI IP at ContainerPort; on STOPPED
+// it deregisters (ecs-v1.md Q8, single-writer).
+type LoadBalancerTarget struct {
+	TargetGroupARN string `json:"targetGroupArn"`
+	ContainerName  string `json:"containerName,omitempty"`
+	ContainerPort  int    `json:"containerPort"`
+}
+
+// ServiceRecord is the persisted service state at ServiceKey. The captured
+// network config + placement strategy let the reconciler launch replacement
+// tasks identically to the original RunTask.
+type ServiceRecord struct {
+	Name               string               `json:"name"`
+	ARN                string               `json:"arn"`
+	Cluster            string               `json:"cluster"`
+	TaskDefFamily      string               `json:"taskDefFamily"`
+	TaskDefRevision    int                  `json:"taskDefRevision"`
+	TaskDefARN         string               `json:"taskDefArn"`
+	DesiredCount       int                  `json:"desiredCount"`
+	Status             string               `json:"status"`
+	SchedulingStrategy string               `json:"schedulingStrategy"`
+	LaunchType         string               `json:"launchType,omitempty"`
+	NetworkMode        string               `json:"networkMode,omitempty"`
+	Subnets            []string             `json:"subnets,omitempty"`
+	SecurityGroups     []string             `json:"securityGroups,omitempty"`
+	AssignPublicIP     string               `json:"assignPublicIp,omitempty"`
+	PlacementStrategy  string               `json:"placementStrategy,omitempty"`
+	LoadBalancers      []LoadBalancerTarget `json:"loadBalancers,omitempty"`
+	DeploymentID       string               `json:"deploymentId"`
+	RunningCount       int                  `json:"runningCount"`
+	PendingCount       int                  `json:"pendingCount"`
+	CreatedAt          time.Time            `json:"createdAt"`
+	UpdatedAt          time.Time            `json:"updatedAt"`
+}

spinifex/handlers/ecs/scheduler.go

@@ -19,6 +19,7 @@ const (
 	schedulerLeaderKey  = "scheduler"
 	leaseRefresh        = 20 * time.Second
 	reaperInterval      = 30 * time.Second
+	reconcileInterval   = 10 * time.Second
 	heartbeatTimeout    = 90 * time.Second
 	stoppedReasonReaped = "ContainerInstance disconnected"
 )
@@ -47,8 +48,10 @@ func NewScheduler(nc *nats.Conn, svc *Service, holder string) *Scheduler {
 func (sc *Scheduler) Run(ctx context.Context) {
 	leaseTicker := time.NewTicker(leaseRefresh)
 	reaperTicker := time.NewTicker(reaperInterval)
+	reconcileTicker := time.NewTicker(reconcileInterval)
 	defer leaseTicker.Stop()
 	defer reaperTicker.Stop()
+	defer reconcileTicker.Stop()
 
 	sc.evaluateLeadership(ctx)
 	for {
@@ -62,6 +65,10 @@ func (sc *Scheduler) Run(ctx context.Context) {
 			if sc.isLeader() {
 				sc.reap()
 			}
+		case <-reconcileTicker.C:
+			if sc.isLeader() {
+				sc.svc.reconcileAllServices()
+			}
 		}
 	}
 }
@@ -271,13 +278,7 @@ func (sc *Scheduler) stopInstanceTasks(kv nats.KeyValue, accountID, cluster, ins
 		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
-		}
-		sc.svc.reclaimTaskENI(accountID, &task)
+		sc.svc.forceStopTask(kv, accountID, &task, stoppedReasonReaped)
 	}
 }
 

spinifex/handlers/ecs/service.go

@@ -32,9 +32,17 @@ type ECSService interface {
 	ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error)
 
 	RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTaskOutput, error)
+	StartTask(input *ecs.StartTaskInput, accountID string) (*ecs.StartTaskOutput, error)
+	StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.StopTaskOutput, error)
 	DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error)
 	ListTasks(input *ecs.ListTasksInput, accountID string) (*ecs.ListTasksOutput, error)
 
+	CreateService(input *ecs.CreateServiceInput, accountID string) (*ecs.CreateServiceOutput, error)
+	UpdateService(input *ecs.UpdateServiceInput, accountID string) (*ecs.UpdateServiceOutput, error)
+	DeleteService(input *ecs.DeleteServiceInput, accountID string) (*ecs.DeleteServiceOutput, error)
+	DescribeServices(input *ecs.DescribeServicesInput, accountID string) (*ecs.DescribeServicesOutput, error)
+	ListServices(input *ecs.ListServicesInput, accountID string) (*ecs.ListServicesOutput, error)
+
 	SubmitTaskStateChange(input *ecs.SubmitTaskStateChangeInput, accountID string) (*ecs.SubmitTaskStateChangeOutput, error)
 
 	// PollAssignments drains an instance's task-assignment inbox (agent → gateway).
@@ -51,6 +59,9 @@ type Service struct {
 	// eni owns the awsvpc task-ENI control-plane (create/attach/detach/delete).
 	// Defaults to the NATS-backed controller; tests substitute a stub.
 	eni eniController
+	// targets registers/deregisters service tasks with ELBv2 target groups.
+	// Defaults to the NATS-backed registrar; tests substitute a stub.
+	targets targetRegistrar
 }
 
 var _ ECSService = (*Service)(nil)
@@ -59,7 +70,7 @@ var _ ECSService = (*Service)(nil)
 // 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, eni: newNATSENIController(nc)}
+	return &Service{nc: nc, region: region, suffix: suffix, eni: newNATSENIController(nc), targets: newNATSTargetRegistrar(nc)}
 }
 
 // defaultCluster is the implicit cluster name when a request omits one.

spinifex/handlers/ecs/service_nats.go

@@ -71,6 +71,14 @@ func (s *NATSECSService) RunTask(input *ecs.RunTaskInput, accountID string) (*ec
 	return utils.NATSRequest[ecs.RunTaskOutput](s.natsConn, "ecs.RunTask", input, defaultTimeout, accountID)
 }
 
+func (s *NATSECSService) StartTask(input *ecs.StartTaskInput, accountID string) (*ecs.StartTaskOutput, error) {
+	return utils.NATSRequest[ecs.StartTaskOutput](s.natsConn, "ecs.StartTask", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.StopTaskOutput, error) {
+	return utils.NATSRequest[ecs.StopTaskOutput](s.natsConn, "ecs.StopTask", 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)
 }
@@ -83,6 +91,28 @@ func (s *NATSECSService) SubmitTaskStateChange(input *ecs.SubmitTaskStateChangeI
 	return utils.NATSRequest[ecs.SubmitTaskStateChangeOutput](s.natsConn, "ecs.SubmitTaskStateChange", input, defaultTimeout, accountID)
 }
 
+// --- Service ---
+
+func (s *NATSECSService) CreateService(input *ecs.CreateServiceInput, accountID string) (*ecs.CreateServiceOutput, error) {
+	return utils.NATSRequest[ecs.CreateServiceOutput](s.natsConn, "ecs.CreateService", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) UpdateService(input *ecs.UpdateServiceInput, accountID string) (*ecs.UpdateServiceOutput, error) {
+	return utils.NATSRequest[ecs.UpdateServiceOutput](s.natsConn, "ecs.UpdateService", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) DeleteService(input *ecs.DeleteServiceInput, accountID string) (*ecs.DeleteServiceOutput, error) {
+	return utils.NATSRequest[ecs.DeleteServiceOutput](s.natsConn, "ecs.DeleteService", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) DescribeServices(input *ecs.DescribeServicesInput, accountID string) (*ecs.DescribeServicesOutput, error) {
+	return utils.NATSRequest[ecs.DescribeServicesOutput](s.natsConn, "ecs.DescribeServices", input, defaultTimeout, accountID)
+}
+
+func (s *NATSECSService) ListServices(input *ecs.ListServicesInput, accountID string) (*ecs.ListServicesOutput, error) {
+	return utils.NATSRequest[ecs.ListServicesOutput](s.natsConn, "ecs.ListServices", input, defaultTimeout, accountID)
+}
+
 func (s *NATSECSService) PollAssignments(input *PollAssignmentsInput, accountID string) (*PollAssignmentsOutput, error) {
 	return utils.NATSRequest[PollAssignmentsOutput](s.natsConn, "ecs.PollAssignments", input, defaultTimeout, accountID)
 }

spinifex/handlers/ecs/services.go

@@ -0,0 +1,520 @@
+package handlers_ecs
+
+import (
+	"errors"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/aws/aws-sdk-go/service/elbv2"
+	"github.com/google/uuid"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2"
+	"github.com/nats-io/nats.go"
+)
+
+// targetRegistrar registers/deregisters a service task's awsvpc ENI IP with an
+// ELBv2 target group. The scheduler is the single writer (ecs-v1.md Q8): it
+// registers on the RUNNING transition and deregisters on STOPPED.
+type targetRegistrar interface {
+	Register(accountID, tgARN, ip string, port int) error
+	Deregister(accountID, tgARN, ip string, port int) error
+}
+
+// natsTargetRegistrar drives the existing elbv2.RegisterTargets/DeregisterTargets
+// NATS handlers; no new ELBv2 surface is added for ECS (Q16).
+type natsTargetRegistrar struct {
+	nc *nats.Conn
+}
+
+var _ targetRegistrar = (*natsTargetRegistrar)(nil)
+
+func newNATSTargetRegistrar(nc *nats.Conn) *natsTargetRegistrar {
+	return &natsTargetRegistrar{nc: nc}
+}
+
+func (r *natsTargetRegistrar) Register(accountID, tgARN, ip string, port int) error {
+	_, err := handlers_elbv2.NewNATSELBv2Service(r.nc).RegisterTargets(&elbv2.RegisterTargetsInput{
+		TargetGroupArn: aws.String(tgARN),
+		Targets:        []*elbv2.TargetDescription{{Id: aws.String(ip), Port: aws.Int64(int64(port))}},
+	}, accountID)
+	return err
+}
+
+func (r *natsTargetRegistrar) Deregister(accountID, tgARN, ip string, port int) error {
+	_, err := handlers_elbv2.NewNATSELBv2Service(r.nc).DeregisterTargets(&elbv2.DeregisterTargetsInput{
+		TargetGroupArn: aws.String(tgARN),
+		Targets:        []*elbv2.TargetDescription{{Id: aws.String(ip), Port: aws.Int64(int64(port))}},
+	}, accountID)
+	return err
+}
+
+// --- Service API ---
+
+// CreateService persists a REPLICA service and reconciles up to desiredCount.
+// DAEMON scheduling and serviceRegistries are rejected in v1 (Q15). Re-creating
+// an existing ACTIVE service returns the existing record (idempotent).
+func (s *Service) CreateService(input *ecs.CreateServiceInput, accountID string) (*ecs.CreateServiceOutput, error) {
+	name := aws.StringValue(input.ServiceName)
+	if name == "" {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	if strategy := aws.StringValue(input.SchedulingStrategy); strategy != "" && strategy != SchedulingStrategyReplica {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter) // DAEMON unsupported (Q15)
+	}
+	if len(input.ServiceRegistries) > 0 {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter) // Route53 v0 not landed (Q15)
+	}
+
+	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)
+	}
+
+	var existing ServiceRecord
+	if ok, gerr := getJSON(kv, ServiceKey(cluster, name), &existing); gerr != nil {
+		return nil, gerr
+	} else if ok && existing.Status == ServiceStatusActive {
+		return &ecs.CreateServiceOutput{Service: s.serviceToAWS(accountID, &existing)}, nil
+	}
+
+	taskDef, err := s.resolveTaskDef(kv, aws.StringValue(input.TaskDefinition))
+	if err != nil {
+		return nil, err
+	}
+
+	now := time.Now().UTC()
+	rec := ServiceRecord{
+		Name:               name,
+		ARN:                ServiceARN(s.region, accountID, cluster, name),
+		Cluster:            cluster,
+		TaskDefFamily:      taskDef.Family,
+		TaskDefRevision:    taskDef.Revision,
+		TaskDefARN:         taskDef.ARN,
+		DesiredCount:       int(aws.Int64Value(input.DesiredCount)),
+		Status:             ServiceStatusActive,
+		SchedulingStrategy: SchedulingStrategyReplica,
+		LaunchType:         aws.StringValue(input.LaunchType),
+		NetworkMode:        resolveNetworkMode(taskDef),
+		PlacementStrategy:  placementStrategyFromAWS(input.PlacementStrategy),
+		LoadBalancers:      loadBalancersFromAWS(input.LoadBalancers),
+		DeploymentID:       uuid.NewString(),
+		CreatedAt:          now,
+		UpdatedAt:          now,
+	}
+	if input.NetworkConfiguration != nil && input.NetworkConfiguration.AwsvpcConfiguration != nil {
+		v := input.NetworkConfiguration.AwsvpcConfiguration
+		rec.Subnets = awsStringSlice(v.Subnets)
+		rec.SecurityGroups = awsStringSlice(v.SecurityGroups)
+		rec.AssignPublicIP = aws.StringValue(v.AssignPublicIp)
+	}
+	if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil {
+		return nil, err
+	}
+
+	if err := s.reconcileService(kv, accountID, &rec); err != nil {
+		slog.Error("ECS CreateService: initial reconcile failed", "service", name, "err", err)
+	}
+	return &ecs.CreateServiceOutput{Service: s.serviceToAWS(accountID, &rec)}, nil
+}
+
+// UpdateService mutates desiredCount and/or the task definition, then reconciles.
+func (s *Service) UpdateService(input *ecs.UpdateServiceInput, accountID string) (*ecs.UpdateServiceOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	name := serviceShortName(aws.StringValue(input.Service))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	var rec ServiceRecord
+	found, err := getJSON(kv, ServiceKey(cluster, name), &rec)
+	if err != nil {
+		return nil, err
+	}
+	if !found || rec.Status == ServiceStatusInactive {
+		return nil, errors.New(awserrors.ErrorECSServiceNotFound)
+	}
+
+	if input.DesiredCount != nil {
+		rec.DesiredCount = int(aws.Int64Value(input.DesiredCount))
+	}
+	if td := aws.StringValue(input.TaskDefinition); td != "" {
+		taskDef, terr := s.resolveTaskDef(kv, td)
+		if terr != nil {
+			return nil, terr
+		}
+		rec.TaskDefFamily = taskDef.Family
+		rec.TaskDefRevision = taskDef.Revision
+		rec.TaskDefARN = taskDef.ARN
+		rec.NetworkMode = resolveNetworkMode(taskDef)
+		rec.DeploymentID = uuid.NewString()
+	}
+	rec.UpdatedAt = time.Now().UTC()
+	if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil {
+		return nil, err
+	}
+	if err := s.reconcileService(kv, accountID, &rec); err != nil {
+		slog.Error("ECS UpdateService: reconcile failed", "service", name, "err", err)
+	}
+	return &ecs.UpdateServiceOutput{Service: s.serviceToAWS(accountID, &rec)}, nil
+}
+
+// DeleteService drains the service to zero, force-stops its tasks, and marks it
+// INACTIVE (kept describable, AWS parity).
+func (s *Service) DeleteService(input *ecs.DeleteServiceInput, accountID string) (*ecs.DeleteServiceOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	name := serviceShortName(aws.StringValue(input.Service))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	var rec ServiceRecord
+	found, err := getJSON(kv, ServiceKey(cluster, name), &rec)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		return nil, errors.New(awserrors.ErrorECSServiceNotFound)
+	}
+
+	rec.DesiredCount = 0
+	tasks, err := s.listServiceTasks(kv, cluster, name)
+	if err != nil {
+		return nil, err
+	}
+	for i := range tasks {
+		s.forceStopTask(kv, accountID, &tasks[i], "Service deleted")
+	}
+	rec.Status = ServiceStatusInactive
+	rec.RunningCount = 0
+	rec.PendingCount = 0
+	rec.UpdatedAt = time.Now().UTC()
+	if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil {
+		return nil, err
+	}
+	return &ecs.DeleteServiceOutput{Service: s.serviceToAWS(accountID, &rec)}, nil
+}
+
+// DescribeServices returns the named services; unknown names surface as failures.
+func (s *Service) DescribeServices(input *ecs.DescribeServicesInput, accountID string) (*ecs.DescribeServicesOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.DescribeServicesOutput{}
+	for _, ref := range awsStringSlice(input.Services) {
+		name := serviceShortName(ref)
+		var rec ServiceRecord
+		found, gerr := getJSON(kv, ServiceKey(cluster, name), &rec)
+		if gerr != nil {
+			return nil, gerr
+		}
+		if found {
+			out.Services = append(out.Services, s.serviceToAWS(accountID, &rec))
+		} else {
+			out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")})
+		}
+	}
+	return out, nil
+}
+
+// ListServices returns the ARNs of every service in a cluster.
+func (s *Service) ListServices(input *ecs.ListServicesInput, accountID string) (*ecs.ListServicesOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	recs, err := s.listServiceRecords(kv, cluster)
+	if err != nil {
+		return nil, err
+	}
+	out := &ecs.ListServicesOutput{}
+	for i := range recs {
+		out.ServiceArns = append(out.ServiceArns, aws.String(recs[i].ARN))
+	}
+	return out, nil
+}
+
+// --- Reconciliation ---
+
+// reconcileService converges a service's running task count on its desired count:
+// it launches the shortfall via RunTask (tagged with the service group so the
+// task counts toward the service) and force-stops any surplus. Counts on the
+// record are refreshed as a side effect.
+func (s *Service) reconcileService(kv nats.KeyValue, accountID string, svc *ServiceRecord) error {
+	if svc.Status != ServiceStatusActive {
+		return nil
+	}
+	tasks, err := s.listServiceTasks(kv, svc.Cluster, svc.Name)
+	if err != nil {
+		return err
+	}
+
+	running, pending := 0, 0
+	for i := range tasks {
+		switch tasks[i].LastStatus {
+		case TaskStatusRunning:
+			running++
+		case TaskStatusPending:
+			pending++
+		}
+	}
+
+	active := running + pending
+	switch {
+	case active < svc.DesiredCount:
+		s.launchServiceTasks(accountID, svc, svc.DesiredCount-active)
+	case active > svc.DesiredCount:
+		s.stopServiceSurplus(kv, accountID, tasks, active-svc.DesiredCount)
+	}
+
+	// Recount after launch/stop so the persisted counts reflect current state.
+	svc.RunningCount, svc.PendingCount = s.countServiceTasks(kv, svc.Cluster, svc.Name)
+	return putJSON(kv, ServiceKey(svc.Cluster, svc.Name), svc)
+}
+
+// countServiceTasks returns a service's current RUNNING and PENDING task counts.
+func (s *Service) countServiceTasks(kv nats.KeyValue, cluster, name string) (running, pending int) {
+	tasks, err := s.listServiceTasks(kv, cluster, name)
+	if err != nil {
+		return 0, 0
+	}
+	for i := range tasks {
+		switch tasks[i].LastStatus {
+		case TaskStatusRunning:
+			running++
+		case TaskStatusPending:
+			pending++
+		}
+	}
+	return running, pending
+}
+
+// reconcileAllServices is the scheduler-tick fan-out: every ACTIVE service in
+// every ECS account bucket is reconciled. Runs only on the scheduler leader.
+func (s *Service) reconcileAllServices() {
+	js, err := s.js()
+	if err != nil {
+		return
+	}
+	for bucket := range js.KeyValueStoreNames() {
+		accountID, ok := accountIDFromBucket(bucket)
+		if !ok {
+			continue
+		}
+		kv, err := js.KeyValue(bucket)
+		if err != nil {
+			continue
+		}
+		recs, err := s.allServiceRecords(kv)
+		if err != nil {
+			continue
+		}
+		for i := range recs {
+			if err := s.reconcileService(kv, accountID, &recs[i]); err != nil {
+				slog.Error("ECS reconcile: service failed", "service", recs[i].Name, "err", err)
+			}
+		}
+	}
+}
+
+// launchServiceTasks places n replacement tasks via the standard RunTask flow,
+// tagged with the service group + deployment so they count toward the service.
+func (s *Service) launchServiceTasks(accountID string, svc *ServiceRecord, n int) {
+	in := &ecs.RunTaskInput{
+		Cluster:        aws.String(svc.Cluster),
+		TaskDefinition: aws.String(svc.TaskDefARN),
+		Count:          aws.Int64(int64(n)),
+		Group:          aws.String(serviceTaskGroup(svc.Name)),
+		StartedBy:      aws.String("ecs-svc/" + svc.DeploymentID),
+	}
+	if svc.PlacementStrategy != "" {
+		in.PlacementStrategy = []*ecs.PlacementStrategy{{Type: aws.String(svc.PlacementStrategy)}}
+	}
+	if svc.NetworkMode == NetworkModeAwsvpc {
+		in.NetworkConfiguration = &ecs.NetworkConfiguration{
+			AwsvpcConfiguration: &ecs.AwsVpcConfiguration{
+				Subnets:        aws.StringSlice(svc.Subnets),
+				SecurityGroups: aws.StringSlice(svc.SecurityGroups),
+			},
+		}
+		if svc.AssignPublicIP != "" {
+			in.NetworkConfiguration.AwsvpcConfiguration.AssignPublicIp = aws.String(svc.AssignPublicIP)
+		}
+	}
+	out, err := s.RunTask(in, accountID)
+	if err != nil {
+		slog.Error("ECS reconcile: launch failed", "service", svc.Name, "err", err)
+		return
+	}
+	if len(out.Failures) > 0 {
+		slog.Warn("ECS reconcile: launch had placement failures", "service", svc.Name, "failures", len(out.Failures))
+	}
+}
+
+// stopServiceSurplus force-stops n of a service's tasks, preferring PENDING tasks
+// (least disruptive) before RUNNING ones.
+func (s *Service) stopServiceSurplus(kv nats.KeyValue, accountID string, tasks []TaskRecord, n int) {
+	order := make([]int, 0, len(tasks))
+	for i := range tasks {
+		if tasks[i].LastStatus == TaskStatusPending {
+			order = append(order, i)
+		}
+	}
+	for i := range tasks {
+		if tasks[i].LastStatus == TaskStatusRunning {
+			order = append(order, i)
+		}
+	}
+	for _, idx := range order {
+		if n <= 0 {
+			break
+		}
+		s.forceStopTask(kv, accountID, &tasks[idx], "Service scaled in")
+		n--
+	}
+}
+
+// --- Helpers ---
+
+// listServiceTasks returns a cluster's non-STOPPED tasks owned by the service.
+func (s *Service) listServiceTasks(kv nats.KeyValue, cluster, name string) ([]TaskRecord, error) {
+	group := serviceTaskGroup(name)
+	all, err := s.listTaskRecords(kv, cluster)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]TaskRecord, 0, len(all))
+	for _, t := range all {
+		if t.Group == group && t.LastStatus != TaskStatusStopped {
+			out = append(out, t)
+		}
+	}
+	return out, nil
+}
+
+// listTaskRecords returns every task record in a cluster.
+func (s *Service) listTaskRecords(kv nats.KeyValue, cluster string) ([]TaskRecord, error) {
+	keys, err := keysWithPrefix(kv, TasksPrefix(cluster))
+	if err != nil {
+		return nil, err
+	}
+	out := make([]TaskRecord, 0, len(keys))
+	for _, k := range keys {
+		var rec TaskRecord
+		found, err := getJSON(kv, k, &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out = append(out, rec)
+		}
+	}
+	return out, nil
+}
+
+// listServiceRecords returns every service record in a cluster.
+func (s *Service) listServiceRecords(kv nats.KeyValue, cluster string) ([]ServiceRecord, error) {
+	keys, err := keysWithPrefix(kv, ServicesPrefix(cluster))
+	if err != nil {
+		return nil, err
+	}
+	out := make([]ServiceRecord, 0, len(keys))
+	for _, k := range keys {
+		var rec ServiceRecord
+		found, err := getJSON(kv, k, &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out = append(out, rec)
+		}
+	}
+	return out, nil
+}
+
+// allServiceRecords returns every service record across all clusters in a bucket.
+func (s *Service) allServiceRecords(kv nats.KeyValue) ([]ServiceRecord, error) {
+	keys, err := keysWithPrefix(kv, "clusters/")
+	if err != nil {
+		return nil, err
+	}
+	out := make([]ServiceRecord, 0)
+	for _, k := range keys {
+		if !strings.Contains(k, "/services/") {
+			continue
+		}
+		var rec ServiceRecord
+		found, err := getJSON(kv, k, &rec)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out = append(out, rec)
+		}
+	}
+	return out, nil
+}
+
+// serviceShortName extracts the service name from a name or a service ARN.
+func serviceShortName(ref string) string {
+	ref = strings.TrimSpace(ref)
+	if i := strings.LastIndexByte(ref, '/'); i >= 0 {
+		return ref[i+1:]
+	}
+	return ref
+}
+
+// loadBalancersFromAWS maps the SDK loadBalancer list to the persisted subset.
+func loadBalancersFromAWS(in []*ecs.LoadBalancer) []LoadBalancerTarget {
+	out := make([]LoadBalancerTarget, 0, len(in))
+	for _, lb := range in {
+		if lb == nil || aws.StringValue(lb.TargetGroupArn) == "" {
+			continue
+		}
+		out = append(out, LoadBalancerTarget{
+			TargetGroupARN: aws.StringValue(lb.TargetGroupArn),
+			ContainerName:  aws.StringValue(lb.ContainerName),
+			ContainerPort:  int(aws.Int64Value(lb.ContainerPort)),
+		})
+	}
+	return out
+}
+
+func (s *Service) serviceToAWS(accountID string, r *ServiceRecord) *ecs.Service {
+	svc := &ecs.Service{
+		ServiceName:        aws.String(r.Name),
+		ServiceArn:         aws.String(r.ARN),
+		ClusterArn:         aws.String(ClusterARN(s.region, accountID, r.Cluster)),
+		Status:             aws.String(r.Status),
+		DesiredCount:       aws.Int64(int64(r.DesiredCount)),
+		RunningCount:       aws.Int64(int64(r.RunningCount)),
+		PendingCount:       aws.Int64(int64(r.PendingCount)),
+		SchedulingStrategy: aws.String(r.SchedulingStrategy),
+		TaskDefinition:     aws.String(r.TaskDefARN),
+	}
+	if r.LaunchType != "" {
+		svc.LaunchType = aws.String(r.LaunchType)
+	}
+	for _, lb := range r.LoadBalancers {
+		svc.LoadBalancers = append(svc.LoadBalancers, &ecs.LoadBalancer{
+			TargetGroupArn: aws.String(lb.TargetGroupARN),
+			ContainerName:  aws.String(lb.ContainerName),
+			ContainerPort:  aws.Int64(int64(lb.ContainerPort)),
+		})
+	}
+	return svc
+}

spinifex/handlers/ecs/services_test.go

@@ -0,0 +1,323 @@
+package handlers_ecs
+
+import (
+	"fmt"
+	"testing"
+
+	"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"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stubRegistrar records ELBv2 target calls so the RUNNING/STOPPED hooks can be
+// asserted without a live elbv2 daemon.
+type stubRegistrar struct {
+	registered   []string
+	deregistered []string
+}
+
+func key(tg, ip string, port int) string { return fmt.Sprintf("%s|%s|%d", tg, ip, port) }
+
+func (r *stubRegistrar) Register(_, tg, ip string, port int) error {
+	r.registered = append(r.registered, key(tg, ip, port))
+	return nil
+}
+
+func (r *stubRegistrar) Deregister(_, tg, ip string, port int) error {
+	r.deregistered = append(r.deregistered, key(tg, ip, port))
+	return nil
+}
+
+// serviceTestRig wires a Service with a stub registrar and a seeded cluster +
+// task definition + one fat container instance.
+func serviceTestRig(t *testing.T) (*Service, *stubRegistrar, nats.KeyValue) {
+	t.Helper()
+	svc, _ := newTestService(t)
+	reg := &stubRegistrar{}
+	svc.targets = reg
+	_, 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", 4096, 8192)
+	kv, err := svc.bucket(testAccountID)
+	require.NoError(t, err)
+	return svc, reg, kv
+}
+
+func TestService_CreateService_LaunchesReplicas(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+
+	out, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster:        aws.String("web"),
+		ServiceName:    aws.String("web"),
+		TaskDefinition: aws.String("app"),
+		DesiredCount:   aws.Int64(2),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, ServiceStatusActive, aws.StringValue(out.Service.Status))
+	assert.Equal(t, ServiceARN(testRegion, testAccountID, "web", "web"), aws.StringValue(out.Service.ServiceArn))
+	assert.Equal(t, int64(2), aws.Int64Value(out.Service.PendingCount))
+
+	tasks, err := svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	require.Len(t, tasks, 2)
+	for _, task := range tasks {
+		assert.Equal(t, serviceTaskGroup("web"), task.Group)
+		assert.Equal(t, TaskStatusPending, task.LastStatus)
+	}
+}
+
+func TestService_CreateService_RejectsDaemonAndRegistries(t *testing.T) {
+	svc, _, _ := serviceTestRig(t)
+
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("d"), TaskDefinition: aws.String("app"),
+		SchedulingStrategy: aws.String(SchedulingStrategyDaemon),
+	}, testAccountID)
+	require.Error(t, err)
+
+	_, err = svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("r"), TaskDefinition: aws.String("app"),
+		ServiceRegistries: []*ecs.ServiceRegistry{{RegistryArn: aws.String("arn:aws:servicediscovery:::service/x")}},
+	}, testAccountID)
+	require.Error(t, err)
+}
+
+func TestService_CreateService_UnknownCluster(t *testing.T) {
+	svc, _ := newTestService(t)
+	registerTaskDef(t, svc, "app", 1, 1)
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("ghost"), ServiceName: aws.String("s"), TaskDefinition: aws.String("app"),
+		DesiredCount: aws.Int64(1),
+	}, testAccountID)
+	require.Error(t, err)
+}
+
+// Reconcile relaunches the desired count after a task STOPs.
+func TestService_Reconcile_ReplacesStoppedTask(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"),
+		DesiredCount: aws.Int64(2),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	tasks, err := svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	require.Len(t, tasks, 2)
+
+	// One task stops (agent report).
+	require.NoError(t, svc.recordTaskState(&bus.TaskState{
+		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
+		TaskID: tasks[0].TaskID, LastStatus: bus.TaskStatusStopped,
+	}))
+	live, err := svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	require.Len(t, live, 1) // STOPPED task drops out of the active set
+
+	// Reconcile launches a replacement back to desired=2.
+	svc.reconcileAllServices()
+	live, err = svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	assert.Len(t, live, 2)
+}
+
+// UpdateService desiredCount=1 stops the surplus task.
+func TestService_UpdateService_ScalesIn(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"),
+		DesiredCount: aws.Int64(3),
+	}, testAccountID)
+	require.NoError(t, err)
+	live, _ := svc.listServiceTasks(kv, "web", "web")
+	require.Len(t, live, 3)
+
+	out, err := svc.UpdateService(&ecs.UpdateServiceInput{
+		Cluster: aws.String("web"), Service: aws.String("web"), DesiredCount: aws.Int64(1),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, int64(1), aws.Int64Value(out.Service.DesiredCount))
+
+	live, err = svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	assert.Len(t, live, 1)
+}
+
+func TestService_DeleteService_DrainsAndInactivates(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"),
+		DesiredCount: aws.Int64(2),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	out, err := svc.DeleteService(&ecs.DeleteServiceInput{
+		Cluster: aws.String("web"), Service: aws.String("web"),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, ServiceStatusInactive, aws.StringValue(out.Service.Status))
+
+	live, err := svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	assert.Empty(t, live)
+
+	// INACTIVE service still describes; reconcile leaves it alone.
+	desc, err := svc.DescribeServices(&ecs.DescribeServicesInput{
+		Cluster: aws.String("web"), Services: []*string{aws.String("web")},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, desc.Services, 1)
+	svc.reconcileAllServices()
+	live, _ = svc.listServiceTasks(kv, "web", "web")
+	assert.Empty(t, live)
+}
+
+func TestService_DescribeAndList(t *testing.T) {
+	svc, _, _ := serviceTestRig(t)
+	for _, name := range []string{"a", "b"} {
+		_, err := svc.CreateService(&ecs.CreateServiceInput{
+			Cluster: aws.String("web"), ServiceName: aws.String(name), TaskDefinition: aws.String("app"),
+			DesiredCount: aws.Int64(0),
+		}, testAccountID)
+		require.NoError(t, err)
+	}
+	list, err := svc.ListServices(&ecs.ListServicesInput{Cluster: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	assert.Len(t, list.ServiceArns, 2)
+
+	miss, err := svc.DescribeServices(&ecs.DescribeServicesInput{
+		Cluster: aws.String("web"), Services: []*string{aws.String("nope")},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, miss.Services)
+	assert.Len(t, miss.Failures, 1)
+}
+
+// StartTask places one task per named container instance and assigns it.
+func TestService_StartTask_PlacesPerInstance(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+
+	out, err := svc.StartTask(&ecs.StartTaskInput{
+		Cluster:            aws.String("web"),
+		TaskDefinition:     aws.String("app"),
+		ContainerInstances: []*string{aws.String("i-1")},
+		Group:              aws.String("custom"),
+		StartedBy:          aws.String("operator"),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, out.Tasks, 1)
+	assert.Empty(t, out.Failures)
+	assert.Equal(t, TaskStatusPending, aws.StringValue(out.Tasks[0].LastStatus))
+
+	tasks, err := svc.listTaskRecords(kv, "web")
+	require.NoError(t, err)
+	require.Len(t, tasks, 1)
+	assert.Equal(t, "custom", tasks[0].Group)
+	assert.Equal(t, "operator", tasks[0].StartedBy)
+	assert.Equal(t, "i-1", tasks[0].ContainerInstanceID)
+}
+
+// StartTask onto an unknown / too-small instance returns a placement Failure,
+// not a hard error, and onto an unknown cluster a ClusterNotFound error.
+func TestService_StartTask_Failures(t *testing.T) {
+	svc, _, _ := serviceTestRig(t)
+
+	out, err := svc.StartTask(&ecs.StartTaskInput{
+		Cluster: aws.String("web"), TaskDefinition: aws.String("app"),
+		ContainerInstances: []*string{aws.String("i-absent")},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, out.Tasks)
+	require.Len(t, out.Failures, 1)
+
+	_, err = svc.StartTask(&ecs.StartTaskInput{
+		Cluster: aws.String("ghost"), TaskDefinition: aws.String("app"),
+		ContainerInstances: []*string{aws.String("i-1")},
+	}, testAccountID)
+	require.Error(t, err)
+}
+
+// StopTask transitions a running service task to STOPPED, releases its
+// reservation, and deregisters its ELBv2 targets.
+func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
+	svc, reg, kv := serviceTestRig(t)
+	const tgARN = "arn:aws:elasticloadbalancing:ap-southeast-2:123456789012:targetgroup/web/abc"
+
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"),
+		DesiredCount: aws.Int64(0),
+		LoadBalancers: []*ecs.LoadBalancer{{
+			TargetGroupArn: aws.String(tgARN), ContainerName: aws.String("app"), ContainerPort: aws.Int64(80),
+		}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	task := &TaskRecord{
+		TaskID: "t-9", Cluster: "web", Group: serviceTaskGroup("web"),
+		ContainerInstanceID: "i-1", LastStatus: TaskStatusRunning,
+		NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.9",
+		ReservedCPU: 64, ReservedMemoryMiB: 64,
+	}
+	require.NoError(t, putJSON(kv, TaskKey("web", "t-9"), task))
+
+	out, err := svc.StopTask(&ecs.StopTaskInput{
+		Cluster: aws.String("web"), Task: aws.String("t-9"), Reason: aws.String("bye"),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.LastStatus))
+	assert.Equal(t, []string{key(tgARN, "10.0.1.9", 80)}, reg.deregistered)
+
+	// Idempotent: a second stop is a no-op (no extra deregister).
+	_, err = svc.StopTask(&ecs.StopTaskInput{Cluster: aws.String("web"), Task: aws.String("t-9")}, testAccountID)
+	require.NoError(t, err)
+	assert.Len(t, reg.deregistered, 1)
+}
+
+func TestService_StopTask_NotFound(t *testing.T) {
+	svc, _, _ := serviceTestRig(t)
+	_, err := svc.StopTask(&ecs.StopTaskInput{Cluster: aws.String("web"), Task: aws.String("ghost")}, testAccountID)
+	require.Error(t, err)
+}
+
+// A service with a target group registers each task's ENI IP on RUNNING and
+// deregisters it on STOPPED (ecs-v1.md Q8, single-writer).
+func TestService_TargetGroup_RegisterOnRunningDeregisterOnStopped(t *testing.T) {
+	svc, reg, kv := serviceTestRig(t)
+	const tgARN = "arn:aws:elasticloadbalancing:ap-southeast-2:123456789012:targetgroup/web/abc"
+
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"),
+		DesiredCount: aws.Int64(0),
+		LoadBalancers: []*ecs.LoadBalancer{{
+			TargetGroupArn: aws.String(tgARN), ContainerName: aws.String("app"), ContainerPort: aws.Int64(80),
+		}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	// Seed an awsvpc service task with an ENI IP (bypasses live ENI provisioning).
+	task := &TaskRecord{
+		TaskID: "t-1", Cluster: "web", Group: serviceTaskGroup("web"),
+		ContainerInstanceID: "i-1", LastStatus: TaskStatusPending,
+		NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.5",
+	}
+	require.NoError(t, putJSON(kv, TaskKey("web", "t-1"), task))
+
+	// RUNNING → register.
+	require.NoError(t, svc.recordTaskState(&bus.TaskState{
+		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
+		TaskID: "t-1", LastStatus: bus.TaskStatusRunning,
+	}))
+	require.Equal(t, []string{key(tgARN, "10.0.1.5", 80)}, reg.registered)
+	assert.Empty(t, reg.deregistered)
+
+	// STOPPED → deregister.
+	require.NoError(t, svc.recordTaskState(&bus.TaskState{
+		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
+		TaskID: "t-1", LastStatus: bus.TaskStatusStopped,
+	}))
+	assert.Equal(t, []string{key(tgARN, "10.0.1.5", 80)}, reg.deregistered)
+}

spinifex/handlers/ecs/task_stop.go

@@ -0,0 +1,211 @@
+package handlers_ecs
+
+import (
+	"encoding/json"
+	"errors"
+	"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/nats-io/nats.go"
+)
+
+// defaultStopReason is recorded when StopTask is called without a reason.
+const defaultStopReason = "Task stopped by user"
+
+// StopTask transitions a task to STOPPED via the control-plane forced-stop path
+// (capacity + ENI release, TG deregister). The live container is torn down by the
+// agent reconciler diffing KV against containerd (ecs-v1.md Q13).
+func (s *Service) StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.StopTaskOutput, error) {
+	cluster := clusterShortName(aws.StringValue(input.Cluster))
+	taskID := taskShortID(aws.StringValue(input.Task))
+	kv, err := s.bucket(accountID)
+	if err != nil {
+		return nil, err
+	}
+	var task TaskRecord
+	found, err := getJSON(kv, TaskKey(cluster, taskID), &task)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
+	reason := aws.StringValue(input.Reason)
+	if reason == "" {
+		reason = defaultStopReason
+	}
+	s.forceStopTask(kv, accountID, &task, reason)
+	return &ecs.StopTaskOutput{Task: s.taskToAWS(accountID, &task)}, nil
+}
+
+// StartTask places one task per named container instance (explicit placement, no
+// scheduler bin-pack). Mirrors RunTask's reserve → record → ENI → assign flow.
+func (s *Service) StartTask(input *ecs.StartTaskInput, accountID string) (*ecs.StartTaskOutput, 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
+	}
+
+	mode := resolveNetworkMode(taskDef)
+	netCfg := awsvpcConfigFromStartTask(input)
+	if mode == NetworkModeAwsvpc && netCfg.firstSubnet() == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	cpu, mem := taskDef.reservedCPU(), taskDef.reservedMemory()
+	group := aws.StringValue(input.Group)
+	startedBy := aws.StringValue(input.StartedBy)
+
+	out := &ecs.StartTaskOutput{}
+	for _, ref := range awsStringSlice(input.ContainerInstances) {
+		instanceID := containerInstanceShortID(ref)
+		taskID := uuid.NewString()
+		inst, rerr := s.reserveOnInstance(kv, cluster, instanceID, taskID, cpu, mem)
+		if rerr != nil {
+			out.Failures = append(out.Failures, &ecs.Failure{
+				Arn: aws.String(ref), Reason: aws.String("RESOURCE:placement"), Detail: aws.String(rerr.Error()),
+			})
+			continue
+		}
+		rec := s.newTaskRecord(accountID, cluster, taskID, taskDef, inst, cpu, mem)
+		rec.NetworkMode = mode
+		rec.Group = group
+		rec.StartedBy = startedBy
+		if mode == NetworkModeAwsvpc {
+			if failure := s.provisionTaskENI(kv, accountID, cluster, rec, netCfg); failure != nil {
+				out.Failures = append(out.Failures, failure)
+				continue
+			}
+		}
+		if err := putJSON(kv, TaskKey(cluster, taskID), rec); err != nil {
+			return nil, err
+		}
+		if err := s.publishAssign(kv, accountID, cluster, inst.InstanceID, rec, taskDef); err != nil {
+			slog.Error("ECS StartTask: failed to publish assign", "task", taskID, "instance", inst.InstanceID, "err", err)
+		}
+		out.Tasks = append(out.Tasks, s.taskToAWS(accountID, rec))
+	}
+	return out, nil
+}
+
+// reserveOnInstance commits a task's capacity reservation onto a specific
+// container instance under a KV CAS. Unlike reservePlacement it does not bin-pack;
+// the instance is fixed by the caller (StartTask).
+func (s *Service) reserveOnInstance(kv nats.KeyValue, cluster, instanceID, taskID string, cpu, mem int) (*InstanceRecord, error) {
+	for range reservePlacementRetries {
+		entry, err := kv.Get(InstanceKey(cluster, instanceID))
+		if errors.Is(err, nats.ErrKeyNotFound) {
+			return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+		if err != nil {
+			return nil, err
+		}
+		var live InstanceRecord
+		if uerr := json.Unmarshal(entry.Value(), &live); uerr != nil {
+			return nil, uerr
+		}
+		if live.Status != InstanceStatusActive || !live.fits(cpu, mem) {
+			return nil, errors.New("instance has insufficient capacity")
+		}
+		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, instanceID), data, entry.Revision()); uerr == nil {
+			return &live, nil
+		}
+	}
+	return nil, errors.New("reservation contended")
+}
+
+// forceStopTask is the control-plane forced-stop: it transitions a task to
+// STOPPED, releases its capacity + ENI, drains its assign inbox, and deregisters
+// its ELBv2 targets. Idempotent and reused by StopTask, service scale-in, and the
+// heartbeat reaper. The mutated record is reflected back into task.
+func (s *Service) forceStopTask(kv nats.KeyValue, accountID string, task *TaskRecord, reason string) {
+	if task.LastStatus == TaskStatusStopped {
+		return
+	}
+	now := time.Now().UTC()
+	task.LastStatus = TaskStatusStopped
+	task.DesiredStatus = TaskStatusStopped
+	task.StoppedReason = reason
+	task.StoppedAt = now
+	if perr := putJSON(kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
+		slog.Error("ECS forceStopTask: persist failed", "task", task.TaskID, "err", perr)
+		return
+	}
+	s.deregisterServiceTargets(kv, accountID, task)
+	s.reclaimAssignInbox(kv, task.Cluster, task.ContainerInstanceID, task.TaskID)
+	s.reclaimTaskENI(accountID, task)
+	if rerr := s.releaseReservation(kv, task.Cluster, task.ContainerInstanceID, task.TaskID, task.ReservedCPU, task.ReservedMemoryMiB); rerr != nil {
+		slog.Error("ECS forceStopTask: release reservation failed", "task", task.TaskID, "err", rerr)
+	}
+}
+
+// awsvpcConfigFromStartTask extracts the awsvpc subnet/SG selection from a
+// StartTask input (StartTask has no bin-pack, so no RunTaskInput to reuse).
+func awsvpcConfigFromStartTask(input *ecs.StartTaskInput) awsvpcConfig {
+	var cfg awsvpcConfig
+	if input != nil && input.NetworkConfiguration != nil && input.NetworkConfiguration.AwsvpcConfiguration != nil {
+		v := input.NetworkConfiguration.AwsvpcConfiguration
+		cfg.Subnets = awsStringSlice(v.Subnets)
+		cfg.SecurityGroups = awsStringSlice(v.SecurityGroups)
+	}
+	return cfg
+}
+
+// registerServiceTargets registers a task's awsvpc ENI IP with each of its
+// owning service's target groups. No-op when the task is not a service task,
+// the service has no load balancers, or the task has no ENI IP (bridge/host).
+func (s *Service) registerServiceTargets(kv nats.KeyValue, accountID string, task *TaskRecord) {
+	s.applyServiceTargets(kv, accountID, task, true)
+}
+
+// deregisterServiceTargets is the STOPPED-side inverse of registerServiceTargets.
+func (s *Service) deregisterServiceTargets(kv nats.KeyValue, accountID string, task *TaskRecord) {
+	s.applyServiceTargets(kv, accountID, task, false)
+}
+
+func (s *Service) applyServiceTargets(kv nats.KeyValue, accountID string, task *TaskRecord, register bool) {
+	name := serviceNameFromGroup(task.Group)
+	if name == "" || task.ENIPrivateIP == "" || s.targets == nil {
+		return
+	}
+	var svc ServiceRecord
+	found, err := getJSON(kv, ServiceKey(task.Cluster, name), &svc)
+	if err != nil || !found || len(svc.LoadBalancers) == 0 {
+		return
+	}
+	for _, lb := range svc.LoadBalancers {
+		var terr error
+		if register {
+			terr = s.targets.Register(accountID, lb.TargetGroupARN, task.ENIPrivateIP, lb.ContainerPort)
+		} else {
+			terr = s.targets.Deregister(accountID, lb.TargetGroupARN, task.ENIPrivateIP, lb.ContainerPort)
+		}
+		if terr != nil {
+			slog.Error("ECS target registration failed", "service", name, "tg", lb.TargetGroupARN,
+				"register", register, "err", terr)
+		}
+	}
+}

spinifex/handlers/ecs/tasks.go

@@ -71,6 +71,8 @@ func (s *Service) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTa
 
 		rec := s.newTaskRecord(accountID, cluster, taskID, taskDef, inst, cpu, mem)
 		rec.NetworkMode = mode
+		rec.Group = aws.StringValue(input.Group)
+		rec.StartedBy = aws.StringValue(input.StartedBy)
 		if mode == NetworkModeAwsvpc {
 			if failure := s.provisionTaskENI(kv, accountID, cluster, rec, netCfg); failure != nil {
 				out.Failures = append(out.Failures, failure)
@@ -267,6 +269,12 @@ func (s *Service) taskToAWS(accountID string, r *TaskRecord) *ecs.Task {
 		LastStatus:           aws.String(r.LastStatus),
 		ContainerInstanceArn: aws.String(r.ContainerInstanceARN),
 	}
+	if r.Group != "" {
+		t.Group = aws.String(r.Group)
+	}
+	if r.StartedBy != "" {
+		t.StartedBy = aws.String(r.StartedBy)
+	}
 	if r.StoppedReason != "" {
 		t.StoppedReason = aws.String(r.StoppedReason)
 	}

spinifex/handlers/elbv2/service_impl.go

@@ -1996,21 +1996,41 @@ func (s *ELBv2ServiceImpl) DescribeTargetHealth(input *elbv2.DescribeTargetHealt
 		}
 	}
 
+	// A target group not forwarded to by any listener serves no traffic; its
+	// targets report "unused" (AWS Target.NotInUse), not "initial".
+	inUse, err := s.store.TargetGroupInUse(tg.TargetGroupArn)
+	if err != nil {
+		slog.Error("DescribeTargetHealth: failed to check TG association", "arn", tg.TargetGroupArn, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
 	descriptions := make([]*elbv2.TargetHealthDescription, 0)
 	for _, t := range tg.Targets {
 		if len(targetFilter) > 0 && !targetFilter[t.Id] {
 			continue
 		}
 
+		state, healthDesc := t.HealthState, t.HealthDesc
+		var reason string
+		if !inUse && t.HealthState != TargetHealthDraining {
+			state = TargetHealthUnused
+			healthDesc = "Target group is not configured to receive traffic from a load balancer"
+			reason = "Target.NotInUse"
+		}
+
+		health := &elbv2.TargetHealth{
+			State:       aws.String(state),
+			Description: aws.String(healthDesc),
+		}
+		if reason != "" {
+			health.Reason = aws.String(reason)
+		}
 		desc := &elbv2.TargetHealthDescription{
 			Target: &elbv2.TargetDescription{
 				Id:   aws.String(t.Id),
 				Port: aws.Int64(t.Port),
 			},
-			TargetHealth: &elbv2.TargetHealth{
-				State:       aws.String(t.HealthState),
-				Description: aws.String(t.HealthDesc),
-			},
+			TargetHealth: health,
 		}
 		descriptions = append(descriptions, desc)
 	}

spinifex/handlers/elbv2/service_impl_health_test.go

@@ -206,3 +206,60 @@ func TestResetTargetHealthOnStartup_NilSafe(t *testing.T) {
 	svc2 := &ELBv2ServiceImpl{}
 	assert.NoError(t, svc2.ResetTargetHealthOnStartup(t.Context()))
 }
+
+// A target group not forwarded to by any listener serves no traffic, so its
+// targets report "unused" (Target.NotInUse) instead of stalling in "initial".
+func TestDescribeTargetHealth_UnusedWhenNoListener(t *testing.T) {
+	svc, _, _ := setupTestServiceWithInstance(t, "i-web001", "10.0.1.50")
+
+	tgOut, err := svc.CreateTargetGroup(&elbv2.CreateTargetGroupInput{
+		Name: aws.String("detached-tg"), Port: aws.Int64(80),
+	}, testAccountID)
+	require.NoError(t, err)
+	tgArn := tgOut.TargetGroups[0].TargetGroupArn
+
+	_, err = svc.RegisterTargets(&elbv2.RegisterTargetsInput{
+		TargetGroupArn: tgArn,
+		Targets:        []*elbv2.TargetDescription{{Id: aws.String("i-web001")}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	health, err := svc.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: tgArn}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, health.TargetHealthDescriptions, 1)
+	th := health.TargetHealthDescriptions[0].TargetHealth
+	assert.Equal(t, TargetHealthUnused, aws.StringValue(th.State))
+	assert.Equal(t, "Target.NotInUse", aws.StringValue(th.Reason))
+}
+
+// Once a listener forwards to the target group it is in use, so DescribeTargetHealth
+// reports the live health state ("initial" until a health report lands), not "unused".
+func TestDescribeTargetHealth_InUseWhenListenerForwards(t *testing.T) {
+	svc, _, _ := setupTestServiceWithInstance(t, "i-web001", "10.0.1.50")
+
+	tgOut, err := svc.CreateTargetGroup(&elbv2.CreateTargetGroupInput{
+		Name: aws.String("attached-tg"), Port: aws.Int64(80),
+	}, testAccountID)
+	require.NoError(t, err)
+	tgArn := tgOut.TargetGroups[0].TargetGroupArn
+
+	lbOut, err := svc.CreateLoadBalancer(&elbv2.CreateLoadBalancerInput{Name: aws.String("attached-lb")}, testAccountID)
+	require.NoError(t, err)
+	_, err = svc.CreateListener(&elbv2.CreateListenerInput{
+		LoadBalancerArn: lbOut.LoadBalancers[0].LoadBalancerArn,
+		Protocol:        aws.String("HTTP"), Port: aws.Int64(80),
+		DefaultActions: []*elbv2.Action{{Type: aws.String("forward"), TargetGroupArn: tgArn}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	_, err = svc.RegisterTargets(&elbv2.RegisterTargetsInput{
+		TargetGroupArn: tgArn,
+		Targets:        []*elbv2.TargetDescription{{Id: aws.String("i-web001")}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	health, err := svc.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: tgArn}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, health.TargetHealthDescriptions, 1)
+	assert.Equal(t, TargetHealthInitial, aws.StringValue(health.TargetHealthDescriptions[0].TargetHealth.State))
+}

spinifex/handlers/elbv2/service_impl_test.go

@@ -764,10 +764,11 @@ func TestRegisterTargets(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, health.TargetHealthDescriptions, 2)
 
-	// First target should use TG default port
+	// First target should use TG default port. With no listener forwarding to the
+	// TG it serves no traffic, so health is "unused" (Target.NotInUse).
 	assert.Equal(t, "i-aaa111", *health.TargetHealthDescriptions[0].Target.Id)
 	assert.Equal(t, int64(80), *health.TargetHealthDescriptions[0].Target.Port)
-	assert.Equal(t, "initial", *health.TargetHealthDescriptions[0].TargetHealth.State)
+	assert.Equal(t, "unused", *health.TargetHealthDescriptions[0].TargetHealth.State)
 
 	// Second target should use override port
 	assert.Equal(t, "i-bbb222", *health.TargetHealthDescriptions[1].Target.Id)

spinifex/handlers/elbv2/service_nats_e2e_test.go

@@ -134,7 +134,8 @@ func TestNATSE2E_FullWorkflow(t *testing.T) {
 	}, testAccountID)
 	require.NoError(t, err)
 	require.Len(t, healthOut.TargetHealthDescriptions, 2)
-	assert.Equal(t, "initial", *healthOut.TargetHealthDescriptions[0].TargetHealth.State)
+	// No listener forwards here yet, so the targets are "unused" (Target.NotInUse).
+	assert.Equal(t, "unused", *healthOut.TargetHealthDescriptions[0].TargetHealth.State)
 
 	// 4. Create load balancer
 	lbOut, err := client.CreateLoadBalancer(&elbv2.CreateLoadBalancerInput{
@@ -158,6 +159,11 @@ func TestNATSE2E_FullWorkflow(t *testing.T) {
 	require.Len(t, lstOut.Listeners, 1)
 	assert.Equal(t, int64(80), *lstOut.Listeners[0].Port)
 
+	// Now a listener forwards to the TG, so its targets are in use again ("initial").
+	healthOut, err = client.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: tgArn}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "initial", *healthOut.TargetHealthDescriptions[0].TargetHealth.State)
+
 	// 6. Describe listeners
 	lstDesc, err := client.DescribeListeners(&elbv2.DescribeListenersInput{
 		LoadBalancerArn: lbArn,

spinifex/handlers/elbv2/store.go

@@ -315,6 +315,39 @@ func (s *Store) TargetGroupsForLB(lbID string) ([]*TargetGroupRecord, error) {
 	return tgs, nil
 }
 
+// TargetGroupInUse reports whether any listener default action or rule action
+// forwards to tgArn. A target group not referenced by a load balancer serves no
+// traffic, so its targets are reported "unused" (AWS Target.NotInUse) instead of
+// staying stuck in "initial" with no health checker to advance them.
+func (s *Store) TargetGroupInUse(tgArn string) (bool, error) {
+	if tgArn == "" {
+		return false, nil
+	}
+	listeners, err := s.ListListeners()
+	if err != nil {
+		return false, err
+	}
+	for _, l := range listeners {
+		for _, a := range l.DefaultActions {
+			if a.TargetGroupArn == tgArn {
+				return true, nil
+			}
+		}
+	}
+	rules, err := s.ListRules()
+	if err != nil {
+		return false, err
+	}
+	for _, r := range rules {
+		for _, a := range r.Actions {
+			if a.TargetGroupArn == tgArn {
+				return true, nil
+			}
+		}
+	}
+	return false, nil
+}
+
 // GetTargetGroupByName finds a target group by name within a VPC.
 func (s *Store) GetTargetGroupByName(name, vpcID string) (*TargetGroupRecord, error) {
 	tgs, err := s.ListTargetGroups()