fix: ECS StopTask reaps container before releasing capacity (siv-484) (#470)

* fix: ECS StopTask reaps container before releasing capacity

StopTask marked a task lastStatus=STOPPED and released its capacity/ENI in
the control plane, but nothing ever told the ecs-agent to kill the container.
The container kept holding its host ports, so DescribeTasks reported a STOPPED
task whose container was still RUNNING, and a replacement task placed on the
freed capacity that rebound the same host port hit EADDRINUSE.

Wire the agent-driven cooperative stop the design intended (ecs-v1.md:165):

- Add a per-instance stop inbox (StopsPrefix/StopKey) and bus.StopDirective.
- requestStopTask marks DesiredStatus=STOPPED and posts a stop directive,
  leaving lastStatus/capacity/ENI/targets untouched. StopTask, DeleteService,
  and service scale-in use it; the heartbeat reaper and cluster/instance
  teardown keep the immediate forceStopTask since no live agent can reap.
- PollAssignments returns pending stops; the agent lists containers by
  mulga.ecs.taskID label, Removes them (kill+delete, freeing ports/netns),
  then reports STOPPED. recordTaskState performs the single capacity release
  on the agent's STOPPED report and reclaims the stop inbox.
- listServiceTasks excludes DesiredStatus=STOPPED tasks so a cooperatively
  stopped task drops out of the service's running total immediately.

DescribeTasks now keeps the task RUNNING with desiredStatus=STOPPED until the
container is actually reaped, matching AWS ordering and eliminating the port
race.

* test: cover StopTask reap paths — poll stop-inbox drain, unplaced force-stop, agent stopTask edges

* ci: retrigger PR checks

cmd/ecs-agent/controlplane.go

@@ -25,9 +25,10 @@ type controlPlane interface {
 	Register(id identity) error
 	// SubmitTaskState reports a task's RUNNING/STOPPED transition.
 	SubmitTaskState(st bus.TaskState) error
-	// PollAssignments drains this instance's assignment inbox, acking the taskIDs
-	// accepted on the previous poll. Returns the still-pending assignments.
-	PollAssignments(cluster, instance string, ack []string) ([]bus.Assign, error)
+	// PollAssignments drains this instance's assignment + stop inboxes, acking the
+	// assign taskIDs and stop taskIDs accepted on the previous poll. Returns the
+	// still-pending assignments and stop directives.
+	PollAssignments(cluster, instance string, ackAssigns, ackStops []string) ([]bus.Assign, []bus.StopDirective, error)
 }
 
 // gatewayControlPlane implements controlPlane over the SigV4 ECS gateway client.
@@ -111,19 +112,24 @@ func (g *gatewayControlPlane) SubmitTaskState(st bus.TaskState) error {
 	return nil
 }
 
-func (g *gatewayControlPlane) PollAssignments(cluster, instance string, ack []string) ([]bus.Assign, error) {
-	in := &handlers_ecs.PollAssignmentsInput{Cluster: cluster, ContainerInstance: instance, AckTaskIDs: ack}
+func (g *gatewayControlPlane) PollAssignments(cluster, instance string, ackAssigns, ackStops []string) ([]bus.Assign, []bus.StopDirective, error) {
+	in := &handlers_ecs.PollAssignmentsInput{
+		Cluster:           cluster,
+		ContainerInstance: instance,
+		AckTaskIDs:        ackAssigns,
+		AckStopIDs:        ackStops,
+	}
 	body, err := json.Marshal(in)
 	if err != nil {
-		return nil, fmt.Errorf("marshal poll: %w", err)
+		return nil, nil, fmt.Errorf("marshal poll: %w", err)
 	}
 	resp, err := g.client.Call("PollAssignments", body)
 	if err != nil {
-		return nil, fmt.Errorf("poll: %w", err)
+		return nil, nil, fmt.Errorf("poll: %w", err)
 	}
 	var out handlers_ecs.PollAssignmentsOutput
 	if err := json.Unmarshal(resp, &out); err != nil {
-		return nil, fmt.Errorf("decode poll: %w", err)
+		return nil, nil, fmt.Errorf("decode poll: %w", err)
 	}
-	return out.Assignments, nil
+	return out.Assignments, out.Stops, nil
 }

cmd/ecs-agent/registration_test.go

@@ -15,11 +15,13 @@ import (
 type fakeCP struct {
 	mu sync.Mutex
 
-	registers   int
-	states      []bus.TaskState
-	pollAcks    [][]string
-	pollReplies [][]bus.Assign
-	pollCalls   int
+	registers    int
+	states       []bus.TaskState
+	pollAcks     [][]string
+	pollStopAcks [][]string
+	pollReplies  [][]bus.Assign
+	stopReplies  [][]bus.StopDirective
+	pollCalls    int
 
 	registerErr bool
 	submitErr   bool
@@ -47,17 +49,21 @@ func (f *fakeCP) SubmitTaskState(st bus.TaskState) error {
 	return nil
 }
 
-func (f *fakeCP) PollAssignments(_, _ string, ack []string) ([]bus.Assign, error) {
+func (f *fakeCP) PollAssignments(_, _ string, ackAssigns, ackStops []string) ([]bus.Assign, []bus.StopDirective, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	cp := append([]string(nil), ack...)
-	f.pollAcks = append(f.pollAcks, cp)
+	f.pollAcks = append(f.pollAcks, append([]string(nil), ackAssigns...))
+	f.pollStopAcks = append(f.pollStopAcks, append([]string(nil), ackStops...))
 	var out []bus.Assign
 	if f.pollCalls < len(f.pollReplies) {
 		out = f.pollReplies[f.pollCalls]
 	}
+	var stops []bus.StopDirective
+	if f.pollCalls < len(f.stopReplies) {
+		stops = f.stopReplies[f.pollCalls]
+	}
 	f.pollCalls++
-	return out, nil
+	return out, stops, nil
 }
 
 func (f *fakeCP) registerCount() int {
@@ -78,6 +84,12 @@ func (f *fakeCP) acks() [][]string {
 	return append([][]string(nil), f.pollAcks...)
 }
 
+func (f *fakeCP) stopAcks() [][]string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([][]string(nil), f.pollStopAcks...)
+}
+
 func testIdentity() identity {
 	return identity{
 		AccountID:    "123456789012",

cmd/ecs-agent/taskrun.go

@@ -22,31 +22,78 @@ func (a *Agent) pollAssignments(ctx context.Context, dispatched map[string]bool)
 	ticker := time.NewTicker(a.cfg.PollInterval)
 	defer ticker.Stop()
 
-	var ackNext []string
+	stopping := map[string]bool{}
+	var ackAssigns, ackStops []string
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			assigns, err := a.cp.PollAssignments(a.id.ClusterName, a.id.InstanceID, ackNext)
+			assigns, stops, err := a.cp.PollAssignments(a.id.ClusterName, a.id.InstanceID, ackAssigns, ackStops)
 			if err != nil {
 				slog.Warn("ecs-agent: poll assignments failed", "err", err)
 				continue
 			}
-			ackNext = ackNext[:0]
+			// Dispatch stops first so a task stopped in the same poll it was assigned
+			// is never started.
+			ackStops = ackStops[:0]
+			for i := range stops {
+				sd := stops[i]
+				if !stopping[sd.TaskID] {
+					stopping[sd.TaskID] = true
+					slog.Info("ecs-agent: task stop requested", "task", sd.TaskID, "reason", sd.Reason)
+					go a.stopTask(ctx, sd)
+				}
+				ackStops = append(ackStops, sd.TaskID)
+			}
+			ackAssigns = ackAssigns[:0]
 			for i := range assigns {
 				as := assigns[i]
-				if !dispatched[as.TaskID] {
+				if !dispatched[as.TaskID] && !stopping[as.TaskID] {
 					dispatched[as.TaskID] = true
 					slog.Info("ecs-agent: task assigned", "task", as.TaskID, "containers", len(as.Containers))
 					go a.runTask(ctx, &as)
 				}
-				ackNext = append(ackNext, as.TaskID)
+				ackAssigns = append(ackAssigns, as.TaskID)
 			}
 		}
 	}
 }
 
+// stopTask reaps a task's containers on a control-plane stop directive: it lists
+// the runtime's containers, removes the ones labeled with this task (kill +
+// delete, releasing host ports/netns), then reports the task STOPPED with the
+// directive's reason. Idempotent — a task with no live containers still reports
+// STOPPED so the scheduler releases its capacity.
+func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) {
+	if a.runner == nil {
+		return
+	}
+	containers, err := a.runner.List(ctx)
+	if err != nil {
+		slog.Warn("ecs-agent: stop list failed", "task", sd.TaskID, "err", err)
+		return
+	}
+	statuses := make([]bus.ContainerStatus, 0)
+	for _, c := range containers {
+		if c.Labels[labelTaskID] != sd.TaskID {
+			continue
+		}
+		if rerr := a.runner.Remove(ctx, c.ID); rerr != nil {
+			slog.Warn("ecs-agent: stop remove failed", "task", sd.TaskID, "container", c.ID, "err", rerr)
+		}
+		statuses = append(statuses, bus.ContainerStatus{
+			Name: c.Labels[labelContainerName], Status: bus.TaskStatusStopped, ContainerID: c.ID,
+		})
+	}
+	reason := sd.Reason
+	if reason == "" {
+		reason = "Task stopped"
+	}
+	a.reportTaskState(&bus.Assign{TaskID: sd.TaskID}, bus.TaskStatusStopped, reason, statuses)
+	slog.Info("ecs-agent: task reaped", "task", sd.TaskID, "containers", len(statuses))
+}
+
 // 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.

cmd/ecs-agent/taskrun_test.go

@@ -172,6 +172,151 @@ func TestPollAssignments_DispatchesOnceAndAcks(t *testing.T) {
 	}
 }
 
+// stopTask reaps the task's labeled containers and reports STOPPED with the
+// directive reason and container status.
+func TestStopTask_ReapsLabeledContainersAndReportsStopped(t *testing.T) {
+	cp := &fakeCP{}
+	rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{
+		{ID: "t-001-web", Running: true, Labels: map[string]string{
+			"mulga.ecs.taskID": "t-001", "mulga.ecs.containerName": "web",
+		}},
+		{ID: "t-999-other", Running: true, Labels: map[string]string{
+			"mulga.ecs.taskID": "t-999", "mulga.ecs.containerName": "web",
+		}},
+	}}
+	a := newRunAgent(cp, rt)
+
+	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "bye"})
+
+	if len(rt.Removed) != 1 || rt.Removed[0] != "t-001-web" {
+		t.Fatalf("want only t-001-web removed, got %+v", rt.Removed)
+	}
+	st := cp.taskStates()
+	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+		t.Fatalf("want one STOPPED report, got %+v", st)
+	}
+	if st[0].Reason != "bye" {
+		t.Errorf("want reason %q, got %q", "bye", st[0].Reason)
+	}
+	if len(st[0].Containers) != 1 || st[0].Containers[0].Status != bus.TaskStatusStopped {
+		t.Errorf("want one STOPPED container, got %+v", st[0].Containers)
+	}
+}
+
+// A stop for a task with no live containers still reports STOPPED so the
+// scheduler releases its capacity.
+func TestStopTask_NoContainersStillReportsStopped(t *testing.T) {
+	cp := &fakeCP{}
+	rt := &ctrruntime.FakePuller{}
+	a := newRunAgent(cp, rt)
+
+	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "gone"})
+
+	if len(rt.Removed) != 0 {
+		t.Errorf("nothing to remove, got %+v", rt.Removed)
+	}
+	st := cp.taskStates()
+	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+		t.Fatalf("want one STOPPED report, got %+v", st)
+	}
+}
+
+// A stop delivered by poll is dispatched once, acked in the stop-ack list, and
+// suppresses a same-poll assign for the same task.
+func TestPollAssignments_StopReapsAndSuppressesAssign(t *testing.T) {
+	as := testAssign()
+	cp := &fakeCP{
+		pollReplies: [][]bus.Assign{{*as}, {*as}, nil},
+		stopReplies: [][]bus.StopDirective{{{TaskID: as.TaskID, Reason: "bye"}}},
+	}
+	rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{
+		{ID: "t-001-web", Running: true, Labels: map[string]string{"mulga.ecs.taskID": "t-001"}},
+	}}
+	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, rt, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go a.pollAssignments(ctx, map[string]bool{})
+
+	deadline := time.After(time.Second)
+	for {
+		if stopped := countStatus(cp.taskStates(), as.TaskID, bus.TaskStatusStopped); stopped >= 1 {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatalf("stop never reaped; states=%+v", cp.taskStates())
+		case <-time.After(2 * time.Millisecond):
+		}
+	}
+	time.Sleep(20 * time.Millisecond)
+	cancel()
+
+	// The assign for the stopping task must never have started a container.
+	if len(rt.Runs) != 0 {
+		t.Errorf("assign for a stopping task should not run, got %+v", rt.Runs)
+	}
+	if running := countStatus(cp.taskStates(), as.TaskID, bus.TaskStatusRunning); running != 0 {
+		t.Errorf("stopping task should never report RUNNING, got %d", running)
+	}
+	acked := false
+	for _, ack := range cp.stopAcks() {
+		for _, id := range ack {
+			if id == as.TaskID {
+				acked = true
+			}
+		}
+	}
+	if !acked {
+		t.Errorf("stop %s was never acked; stopAcks=%v", as.TaskID, cp.stopAcks())
+	}
+}
+
+// stopTask with no runtime is a no-op: nothing to reap, no state reported.
+func TestStopTask_NilRunnerNoOp(t *testing.T) {
+	cp := &fakeCP{}
+	a := newAgent(config{}, testIdentity(), cp, nil, nil, nil)
+	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "x"})
+	if len(cp.taskStates()) != 0 {
+		t.Fatalf("nil runner should report no state, got %+v", cp.taskStates())
+	}
+}
+
+// A List failure aborts the reap without reporting a (false) STOPPED state.
+func TestStopTask_ListErrorNoReport(t *testing.T) {
+	cp := &fakeCP{}
+	rt := &ctrruntime.FakePuller{ListErr: errors.New("list boom")}
+	a := newRunAgent(cp, rt)
+	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "x"})
+	if len(cp.taskStates()) != 0 {
+		t.Fatalf("list error should report no state, got %+v", cp.taskStates())
+	}
+}
+
+// A directive with no reason reaps the matching container and reports STOPPED with
+// the default reason.
+func TestStopTask_EmptyReasonReapsWithDefault(t *testing.T) {
+	cp := &fakeCP{}
+	rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{{
+		ID: "t-001-web",
+		Labels: map[string]string{
+			"mulga.ecs.taskID": "t-001", "mulga.ecs.containerName": "web",
+		},
+	}}}
+	a := newRunAgent(cp, rt)
+	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001"})
+
+	if len(rt.Removed) != 1 || rt.Removed[0] != "t-001-web" {
+		t.Fatalf("want container t-001-web reaped, got %+v", rt.Removed)
+	}
+	st := cp.taskStates()
+	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+		t.Fatalf("want one STOPPED report, got %+v", st)
+	}
+	if st[0].Reason != "Task stopped" {
+		t.Errorf("want default reason %q, got %q", "Task stopped", st[0].Reason)
+	}
+}
+
 func countStatus(states []bus.TaskState, taskID, status string) int {
 	n := 0
 	for _, st := range states {

spinifex/handlers/ecs/bus/messages.go

@@ -101,6 +101,15 @@ type Assign struct {
 	AssignedAt       time.Time `json:"assignedAt"`
 }
 
+// StopDirective is one pending task-stop the scheduler places in an instance's
+// stop inbox (scheduler → agent). The agent kills + reaps the task's containers,
+// then reports STOPPED with Reason. Delivery is idempotent; the STOPPED
+// transition deletes the inbox entry.
+type StopDirective struct {
+	TaskID string `json:"taskId"`
+	Reason string `json:"reason,omitempty"`
+}
+
 // ContainerStatus is a single container's reported lifecycle state.
 type ContainerStatus struct {
 	Name        string `json:"name"`

spinifex/handlers/ecs/instances.go

@@ -251,6 +251,7 @@ func (s *Service) recordTaskState(msg *bus.TaskState) error {
 		s.deregisterServiceTargets(kv, msg.AccountID, &task)
 		s.releaseTaskPublicIP(msg.AccountID, &task)
 		s.reclaimAssignInbox(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID)
+		s.reclaimStopInbox(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID)
 		s.reclaimTaskENI(msg.AccountID, &task)
 		if perr := putJSON(kv, TaskKey(msg.ClusterName, msg.TaskID), &task); perr != nil {
 			slog.Error("ECS task STOPPED: persist after EIP release failed", "task", msg.TaskID, "err", perr)

spinifex/handlers/ecs/poll.go

@@ -14,11 +14,17 @@ type PollAssignmentsInput struct {
 	Cluster           string   `json:"cluster"`
 	ContainerInstance string   `json:"containerInstance"`
 	AckTaskIDs        []string `json:"ackTaskIds,omitempty"`
+	// AckStopIDs are stop directives the agent reaped on a prior poll; they may now
+	// be removed. Optional — stop delivery is idempotent and the STOPPED transition
+	// deletes the entry, so a dropped ack only re-delivers a no-op reap.
+	AckStopIDs []string `json:"ackStopIds,omitempty"`
 }
 
-// PollAssignmentsOutput carries the instance's currently pending assignments.
+// PollAssignmentsOutput carries the instance's currently pending assignments and
+// stop directives.
 type PollAssignmentsOutput struct {
-	Assignments []bus.Assign `json:"assignments,omitempty"`
+	Assignments []bus.Assign        `json:"assignments,omitempty"`
+	Stops       []bus.StopDirective `json:"stops,omitempty"`
 }
 
 // PollAssignments acks any completed assignments (deletes their inbox keys) then
@@ -40,6 +46,12 @@ func (s *Service) PollAssignments(input *PollAssignmentsInput, accountID string)
 		}
 		_ = kv.Delete(AssignmentKey(cluster, instanceID, taskShortID(taskID)))
 	}
+	for _, taskID := range input.AckStopIDs {
+		if taskID == "" {
+			continue
+		}
+		_ = kv.Delete(StopKey(cluster, instanceID, taskShortID(taskID)))
+	}
 
 	keys, err := keysWithPrefix(kv, AssignmentsPrefix(cluster, instanceID))
 	if err != nil {
@@ -56,6 +68,21 @@ func (s *Service) PollAssignments(input *PollAssignmentsInput, accountID string)
 			out.Assignments = append(out.Assignments, as)
 		}
 	}
+
+	stopKeys, err := keysWithPrefix(kv, StopsPrefix(cluster, instanceID))
+	if err != nil {
+		return nil, err
+	}
+	for _, key := range stopKeys {
+		var sd bus.StopDirective
+		found, err := getJSON(kv, key, &sd)
+		if err != nil {
+			return nil, err
+		}
+		if found {
+			out.Stops = append(out.Stops, sd)
+		}
+	}
 	return out, nil
 }
 
@@ -67,3 +94,12 @@ func (s *Service) reclaimAssignInbox(kv nats.KeyValue, cluster, instanceID, task
 	}
 	_ = kv.Delete(AssignmentKey(cluster, instanceID, taskID))
 }
+
+// reclaimStopInbox removes a task's stop-inbox entry on the STOPPED transition so
+// a reaped task is never re-delivered as a stop. Best-effort.
+func (s *Service) reclaimStopInbox(kv nats.KeyValue, cluster, instanceID, taskID string) {
+	if instanceID == "" || taskID == "" {
+		return
+	}
+	_ = kv.Delete(StopKey(cluster, instanceID, taskID))
+}

spinifex/handlers/ecs/poll_test.go

@@ -0,0 +1,46 @@
+package handlers_ecs
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestPollAssignments_DrainsStopsAndAcks covers the agent poll path: a poll
+// returns both the pending assignments and the stop directives for the instance,
+// and acking their IDs on a follow-up poll deletes the inbox entries so nothing is
+// re-delivered. Empty ack IDs are skipped.
+func TestPollAssignments_DrainsStopsAndAcks(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+
+	require.NoError(t, putJSON(kv, AssignmentKey("web", "i-1", "t-1"),
+		&bus.Assign{TaskID: "t-1", ClusterName: "web", InstanceID: "i-1"}))
+	require.NoError(t, putJSON(kv, StopKey("web", "i-1", "t-2"),
+		&bus.StopDirective{TaskID: "t-2", Reason: "drain"}))
+
+	out, err := svc.PollAssignments(&PollAssignmentsInput{
+		Cluster: "web", ContainerInstance: "i-1",
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, out.Assignments, 1)
+	assert.Equal(t, "t-1", out.Assignments[0].TaskID)
+	require.Len(t, out.Stops, 1)
+	assert.Equal(t, "t-2", out.Stops[0].TaskID)
+	assert.Equal(t, "drain", out.Stops[0].Reason)
+
+	// Ack both (with a blank ID that must be skipped) → entries removed, next poll
+	// drains empty.
+	out, err = svc.PollAssignments(&PollAssignmentsInput{
+		Cluster: "web", ContainerInstance: "i-1",
+		AckTaskIDs: []string{"t-1", ""}, AckStopIDs: []string{"t-2", ""},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, out.Assignments)
+	assert.Empty(t, out.Stops)
+
+	found, err := getJSON(kv, StopKey("web", "i-1", "t-2"), &bus.StopDirective{})
+	require.NoError(t, err)
+	assert.False(t, found)
+}

spinifex/handlers/ecs/services.go

@@ -265,7 +265,7 @@ func (s *Service) DeleteService(input *ecs.DeleteServiceInput, accountID string)
 		return nil, err
 	}
 	for i := range tasks {
-		s.forceStopTask(kv, accountID, &tasks[i], "Service deleted")
+		s.requestStopTask(kv, accountID, &tasks[i], "Service deleted")
 	}
 	rec.Status = ServiceStatusInactive
 	rec.RunningCount = 0
@@ -459,7 +459,9 @@ func (s *Service) stopSurplusTasks(kv nats.KeyValue, accountID string, tasks []T
 	}
 	primarySurplus := max(len(primaryPending)+len(primaryRunning)-desired, 0)
 	runBudget := max(runningTotal-minCount, 0)
-	stop := func(i int, reason string) { s.forceStopTask(kv, accountID, &tasks[i], reason) }
+	// Cooperative stop: the agent reaps the container, then recordTaskState performs
+	// the single capacity release. A never-placed task force-stops internally.
+	stop := func(i int, reason string) { s.requestStopTask(kv, accountID, &tasks[i], reason) }
 
 	for _, i := range oldPending {
 		stop(i, deploymentSupersededReason)
@@ -490,7 +492,10 @@ func (s *Service) stopSurplusTasks(kv nats.KeyValue, accountID string, tasks []T
 
 // --- Helpers ---
 
-// listServiceTasks returns a cluster's non-STOPPED tasks owned by the service.
+// listServiceTasks returns a cluster's live tasks owned by the service: neither
+// STOPPED nor already requested to stop (DesiredStatus=STOPPED). A cooperatively
+// stopped task lingers RUNNING until the agent reaps it, but it is on its way out,
+// so it must not count toward the service's running total for scaling decisions.
 func (s *Service) listServiceTasks(kv nats.KeyValue, cluster, name string) ([]TaskRecord, error) {
 	group := serviceTaskGroup(name)
 	all, err := s.listTaskRecords(kv, cluster)
@@ -499,7 +504,7 @@ func (s *Service) listServiceTasks(kv nats.KeyValue, cluster, name string) ([]Ta
 	}
 	out := make([]TaskRecord, 0, len(all))
 	for _, t := range all {
-		if t.Group == group && t.LastStatus != TaskStatusStopped {
+		if t.Group == group && t.LastStatus != TaskStatusStopped && t.DesiredStatus != TaskStatusStopped {
 			out = append(out, t)
 		}
 	}

spinifex/handlers/ecs/services_test.go

@@ -258,20 +258,40 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
 
 	task := &TaskRecord{
 		TaskID: "t-9", Cluster: "web", Group: serviceTaskGroup("web"),
-		ContainerInstanceID: "i-1", LastStatus: TaskStatusRunning,
+		ContainerInstanceID: "i-1", LastStatus: TaskStatusRunning, DesiredStatus: TaskStatusRunning,
 		NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.9",
 		ReservedCPU: 64, ReservedMemoryMiB: 64,
 	}
 	require.NoError(t, putJSON(kv, TaskKey("web", "t-9"), task))
 
+	// Cooperative stop: the task stays RUNNING with desiredStatus=STOPPED and a
+	// stop directive lands in the instance inbox; nothing is deregistered until the
+	// agent reaps the container and reports STOPPED.
 	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)
+	assert.Equal(t, TaskStatusRunning, aws.StringValue(out.Task.LastStatus))
+	assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.DesiredStatus))
+	assert.Empty(t, reg.deregistered)
 
-	// Idempotent: a second stop is a no-op (no extra deregister).
+	var sd bus.StopDirective
+	found, err := getJSON(kv, StopKey("web", "i-1", "t-9"), &sd)
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.Equal(t, "bye", sd.Reason)
+
+	// Agent reports STOPPED → deregister + stop-inbox reclaimed.
+	require.NoError(t, svc.recordTaskState(&bus.TaskState{
+		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
+		TaskID: "t-9", LastStatus: bus.TaskStatusStopped,
+	}))
+	assert.Equal(t, []string{key(tgARN, "10.0.1.9", 80)}, reg.deregistered)
+	found, err = getJSON(kv, StopKey("web", "i-1", "t-9"), &sd)
+	require.NoError(t, err)
+	assert.False(t, found)
+
+	// Idempotent: a second stop is a no-op (already STOPPED).
 	_, 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)
@@ -283,6 +303,25 @@ func TestService_StopTask_NotFound(t *testing.T) {
 	require.Error(t, err)
 }
 
+// StopTask on a never-placed task (no container instance) force-stops it directly:
+// no agent can reap it, so the control plane transitions it to STOPPED immediately.
+func TestService_StopTask_UnplacedForceStops(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	task := &TaskRecord{
+		TaskID: "t-np", Cluster: "web",
+		LastStatus: TaskStatusPending, DesiredStatus: TaskStatusRunning,
+	}
+	require.NoError(t, putJSON(kv, TaskKey("web", "t-np"), task))
+
+	out, err := svc.StopTask(&ecs.StopTaskInput{
+		Cluster: aws.String("web"), Task: aws.String("t-np"), Reason: aws.String("cancel"),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.LastStatus))
+	assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.DesiredStatus))
+	assert.Equal(t, "cancel", aws.StringValue(out.Task.StoppedReason))
+}
+
 // 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) {

spinifex/handlers/ecs/store.go

@@ -81,6 +81,19 @@ func AssignmentKey(cluster, instanceID, taskID string) string {
 	return AssignmentsPrefix(cluster, instanceID) + taskID
 }
 
+// StopsPrefix returns the KV key prefix under which a container instance's pending
+// task-stop directives (its "stop inbox") live. A sibling of assignments/ so it is
+// never picked up when listing tasks or assignments. The agent drains it by
+// polling the gateway; the STOPPED transition removes an entry.
+func StopsPrefix(cluster, instanceID string) string {
+	return fmt.Sprintf("clusters/%s/stops/%s/", cluster, instanceID)
+}
+
+// StopKey returns the KV key for one task's stop directive in an instance inbox.
+func StopKey(cluster, instanceID, taskID string) string {
+	return StopsPrefix(cluster, instanceID) + taskID
+}
+
 // ServicesPrefix returns the KV key prefix under which all of a cluster's service
 // records live. Used by ListServices to enumerate.
 func ServicesPrefix(cluster string) string {

spinifex/handlers/ecs/task_stop.go

@@ -11,15 +11,19 @@ import (
 	"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"
 )
 
 // 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).
+// StopTask requests a cooperative stop: it marks the task DesiredStatus=STOPPED
+// and posts a directive to the instance's stop inbox. The agent reaps the
+// container and reports STOPPED, which drives the capacity + ENI + TG release in
+// recordTaskState. LastStatus stays RUNNING until the container is actually reaped
+// so a freed port is never rebound while the old container still holds it
+// (ecs-v1.md:165).
 func (s *Service) StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.StopTaskOutput, error) {
 	cluster := clusterShortName(aws.StringValue(input.Cluster))
 	taskID := taskShortID(aws.StringValue(input.Task))
@@ -39,7 +43,7 @@ func (s *Service) StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.Sto
 	if reason == "" {
 		reason = defaultStopReason
 	}
-	s.forceStopTask(kv, accountID, &task, reason)
+	s.requestStopTask(kv, accountID, &task, reason)
 	return &ecs.StopTaskOutput{Task: s.taskToAWS(accountID, &task)}, nil
 }
 
@@ -138,10 +142,37 @@ func (s *Service) reserveOnInstance(kv nats.KeyValue, cluster, instanceID, taskI
 	return nil, errors.New("reservation contended")
 }
 
+// requestStopTask is the cooperative stop used while the agent is alive: it marks
+// the task DesiredStatus=STOPPED and posts a stop directive to the instance's stop
+// inbox, but leaves LastStatus, capacity, ENI, targets, and the assign inbox
+// untouched. The agent reaps the container and reports STOPPED; recordTaskState
+// then performs the single release. No-op once the task is already STOPPED. A task
+// with no container instance (never placed) is forced-stopped directly since no
+// agent can report it.
+func (s *Service) requestStopTask(kv nats.KeyValue, accountID string, task *TaskRecord, reason string) {
+	if task.LastStatus == TaskStatusStopped {
+		return
+	}
+	if task.ContainerInstanceID == "" {
+		s.forceStopTask(kv, accountID, task, reason)
+		return
+	}
+	task.DesiredStatus = TaskStatusStopped
+	if perr := putJSON(kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
+		slog.Error("ECS requestStopTask: persist failed", "task", task.TaskID, "err", perr)
+		return
+	}
+	sd := bus.StopDirective{TaskID: task.TaskID, Reason: reason}
+	if perr := putJSON(kv, StopKey(task.Cluster, task.ContainerInstanceID, task.TaskID), &sd); perr != nil {
+		slog.Error("ECS requestStopTask: post stop directive failed", "task", task.TaskID, "err", perr)
+	}
+}
+
 // 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.
+// its ELBv2 targets. Used where no live agent can reap the container (heartbeat
+// reaper, cluster/instance teardown). 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
@@ -160,6 +191,7 @@ func (s *Service) forceStopTask(kv nats.KeyValue, accountID string, task *TaskRe
 	}
 	s.deregisterServiceTargets(kv, accountID, task)
 	s.reclaimAssignInbox(kv, task.Cluster, task.ContainerInstanceID, task.TaskID)
+	s.reclaimStopInbox(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)