refactor: move the ECS agent wire vocabulary to contracts/ecs/v1

Move handlers/ecs/bus to contracts/ecs/v1 (package ecsv1) with no legacy
shim. Subject strings, JSON fields and types are byte-identical; the old-agent
compatibility test moves unchanged and golden pins cover the remaining
message types. The README route table records that only service-reconcile
has a live producer today; the agent reuses these payload types through the
gateway actions. service-interfaces.yaml lists ecs.bus.* under the daemon.

architecture/layering/allowlist_test.go

@@ -24,7 +24,6 @@ var allowlist = []allowance{
 	{From: "accountteardown", To: "domains/ec2/awsapi/vpc", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 
 	{From: "agents/ecs/agent", To: "handlers/ecs", Debt: "agents/ecs/agent", Reason: "assignment/state wire types from the ECS implementation; needs a guest/controller contract"},
-	{From: "agents/ecs/agent", To: "handlers/ecs/bus", Debt: "agents/ecs/agent", Reason: "assignment/state wire types from the ECS implementation; needs a guest/controller contract"},
 
 	{From: "agents/eks/tokenwebhook", To: "handlers/eks", Debt: "agents/eks/tokenwebhook", Reason: "WebhookTokenReviewResult from the EKS implementation; needs an EKS contract or projection"},
 

architecture/layering/legacy_test.go

@@ -24,7 +24,6 @@ var frozenLegacyPackages = []string{
 	"gateway/sts",
 	"gateway/tagging",
 	"handlers/ecs",
-	"handlers/ecs/bus",
 	"handlers/eks",
 	"handlers/elbv2",
 	"handlers/iam",

contracts/ecs/v1/README.md

@@ -0,0 +1,59 @@
+# ECS agent/scheduler bus contract, version 1
+
+This package owns the deployed wire vocabulary for the `ecs.bus.*`
+scheduler-to-agent boundary: subjects, payload JSON and the Layer-2 message
+types. It is a characterization of what is deployed today, not a redesign; it
+does not move the scheduler, the agent, or perform any NATS I/O itself.
+
+## Route table
+
+| Subject | Payload | Reply | Delivery | Queue group | Producer | Consumer |
+|---|---|---|---|---|---|---|
+| `ecs.bus.<account>.<cluster>.instance-register.<instanceId>` | `RegisterInstance` | none | fire-and-forget | none | none today | `spinifex_daemon` leader, `handlers/ecs/scheduler.go` `onRegister` |
+| `ecs.bus.<account>.<cluster>.instance-heartbeat.<instanceId>` | `Heartbeat` | none | fire-and-forget | none | none today | `spinifex_daemon` leader, `handlers/ecs/scheduler.go` `onHeartbeat` |
+| `ecs.bus.<account>.<cluster>.assign.<instanceId>` | `Assign` | none | fire-and-forget | none | none today | none today |
+| `ecs.bus.<account>.<cluster>.task-state.<taskId>` | `TaskState` | none | fire-and-forget | none | none today | `spinifex_daemon` leader, `handlers/ecs/scheduler.go` `onTaskState` |
+| `ecs.bus.<account>.<cluster>.service-reconcile` | none | none | fire-and-forget | none | `spinifex_daemon`, `handlers/ecs/instances.go` `SubmitTaskStateChange` | `spinifex_daemon` leader, `handlers/ecs/scheduler.go` `onReconcileWake` |
+
+The agent today registers, heartbeats and reports task state through the AWS
+gateway's `RegisterContainerInstance`, `PollAssignments` and
+`SubmitTaskStateChange` NATS actions instead of publishing on this bus; those
+actions are a separate, pre-existing subject family and carry some of the
+same message types (`Assign`, `TaskState`, `StopDirective`) as their wire
+body. Four of the five subject builders above, every route except
+`service-reconcile`, therefore have no current publisher. The scheduler still
+subscribes to all four wildcard patterns; only its election leader holds the
+subscriptions, which is why no queue group is needed. `service-reconcile` is
+the one route with a live producer and consumer: the worker that recorded a
+task-state change is rarely the leader, so it publishes a cross-node wake the
+leader's own subscription picks up.
+
+`StopDirective`, `ContainerStatus`, `InstanceCapacity`, `PortMapping`,
+`AssignContainer` and `SystemControl` are nested payload types carried inside
+`Assign` and `TaskState`; they travel on the same routes as their parent type
+and are not separately addressed.
+
+## Ownership
+
+The ECS domain owns the meaning of every subject and type in this package.
+The ecs-agent binary baked into the ECS node AMI is a deployed consumer: it
+decodes `Assign` and `StopDirective` and encodes `TaskState`, all carried
+today by the AWS-gateway actions that reuse these types.
+It does not encode `Heartbeat` or `RegisterInstance`.
+
+## Compatibility rules
+
+This is the deployed v1 contract. Changes follow N/N-1: a subject, JSON field
+name, JSON field type, or message semantics must not be renamed, retyped, or
+reinterpreted to mean something else. The AMI-baked agent cannot be upgraded
+in lockstep with the daemon, so it must keep decoding the shapes it already
+knows; `messages_test.go` pins the older agent's `AssignContainer` shape
+against the current one for exactly this reason. A breaking change needs a
+new, explicitly versioned successor package, not an edit here.
+
+## What this package does not do
+
+It does not move the scheduler, the agent, or any subscriber out of its
+current package, and it changes no ECS lifecycle behaviour: every subject
+string, JSON field and message type here is identical to the one the
+`handlers/ecs/bus` package declared before this move.

contracts/ecs/v1/messages.go

@@ -1,4 +1,4 @@
-package bus
+package ecsv1
 
 import "time"
 
@@ -13,9 +13,9 @@ type InstanceCapacity struct {
 	GPUIDs []string `json:"gpuIds,omitempty"`
 }
 
-// RegisterInstance is published on RegisterSubject when an ecs-agent boots and
-// joins its cluster. The scheduler records a container-instance entry keyed by
-// InstanceID (handlers_ecs.InstanceKey).
+// RegisterInstance is the registration payload the scheduler decodes from
+// RegisterSubject, which has no current publisher. It records a
+// container-instance entry keyed by InstanceID.
 type RegisterInstance struct {
 	AccountID    string           `json:"accountId"`
 	ClusterName  string           `json:"clusterName"`
@@ -27,9 +27,9 @@ type RegisterInstance struct {
 	RegisteredAt time.Time        `json:"registeredAt"`
 }
 
-// Heartbeat is published on HeartbeatSubject every ~30s while the agent is
-// healthy. A missed heartbeat past the scheduler's TTL marks the instance
-// DISCONNECTED (wired in Sprint 4e).
+// Heartbeat is the liveness payload the scheduler decodes from HeartbeatSubject,
+// which has no current publisher; the agent's registration action carries
+// liveness instead.
 type Heartbeat struct {
 	AccountID    string    `json:"accountId"`
 	ClusterName  string    `json:"clusterName"`
@@ -78,7 +78,7 @@ type AssignContainer struct {
 	CPU       int    `json:"cpu,omitempty"`
 	MemoryMiB int    `json:"memoryMiB,omitempty"`
 	// GPU is the whole-GPU count requested via resourceRequirements type=GPU.
-	// Device pinning/CDI injection is not wired yet (later Epic C tasks).
+	// Device pinning/CDI injection is not wired yet.
 	GPU          int               `json:"gpu,omitempty"`
 	Essential    bool              `json:"essential"`
 	Command      []string          `json:"command,omitempty"`
@@ -106,9 +106,9 @@ type AssignContainer struct {
 	StopTimeout  *int64 `json:"stopTimeout,omitempty"`
 }
 
-// Assign is published on AssignSubject when the scheduler places a task on this
-// instance (scheduler → agent). The agent pulls each container image, creates and
-// starts the containers via containerd, then reports progress on TaskStateSubject.
+// Assign places a task on an instance (scheduler → agent). It is the payload of
+// the gateway's PollAssignments action; AssignSubject carries nothing today.
+// The agent pulls each image, starts the containers, then reports TaskState.
 type Assign struct {
 	AccountID       string            `json:"accountId"`
 	ClusterName     string            `json:"clusterName"`
@@ -157,9 +157,9 @@ type ContainerStatus struct {
 	GPUIDs []string `json:"gpuIds,omitempty"`
 }
 
-// TaskState is published on TaskStateSubject as the agent drives a task through
-// its lifecycle (agent → scheduler). The scheduler updates the task record and
-// recomputes the instance's remaining capacity.
+// TaskState reports a task's lifecycle (agent → scheduler). It is the payload of
+// the gateway's SubmitTaskStateChange action; TaskStateSubject carries nothing
+// today. The scheduler updates the task record and the instance's capacity.
 type TaskState struct {
 	AccountID   string            `json:"accountId"`
 	ClusterName string            `json:"clusterName"`

contracts/ecs/v1/messages_pin_test.go

@@ -0,0 +1,193 @@
+package ecsv1_test
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var pinTime = time.Date(2026, 1, 15, 10, 30, 0, 0, time.UTC)
+
+func TestRegisterInstance_JSON(t *testing.T) {
+	golden := `{"accountId":"123456789012","clusterName":"default","instanceId":"i-abc123",` +
+		`"availabilityZone":"az-1","hostname":"host-1",` +
+		`"capacity":{"cpu":2048,"memoryMiB":4096},"agentVersion":"1.0.0",` +
+		`"registeredAt":"2026-01-15T10:30:00Z"}`
+	msg := ecsv1.RegisterInstance{
+		AccountID: "123456789012", ClusterName: "default", InstanceID: "i-abc123",
+		AZ: "az-1", Hostname: "host-1",
+		Capacity:     ecsv1.InstanceCapacity{CPU: 2048, MemoryMiB: 4096},
+		AgentVersion: "1.0.0", RegisteredAt: pinTime,
+	}
+
+	data, err := json.Marshal(msg)
+	require.NoError(t, err)
+	assert.JSONEq(t, golden, string(data))
+
+	var decoded ecsv1.RegisterInstance
+	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
+	assert.Equal(t, msg, decoded)
+}
+
+func TestInstanceCapacity_GPUFields(t *testing.T) {
+	golden := `{"cpu":4096,"memoryMiB":8192,"gpu":2,"gpuIds":["GPU-1","GPU-2"]}`
+	capacity := ecsv1.InstanceCapacity{CPU: 4096, MemoryMiB: 8192, GPU: 2, GPUIDs: []string{"GPU-1", "GPU-2"}}
+
+	data, err := json.Marshal(capacity)
+	require.NoError(t, err)
+	assert.JSONEq(t, golden, string(data))
+
+	var decoded ecsv1.InstanceCapacity
+	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
+	assert.Equal(t, capacity, decoded)
+}
+
+func TestHeartbeat_JSON(t *testing.T) {
+	golden := `{"accountId":"123456789012","clusterName":"default","instanceId":"i-abc123",` +
+		`"status":"ACTIVE","runningTasks":3,"sentAt":"2026-01-15T10:30:00Z"}`
+	msg := ecsv1.Heartbeat{
+		AccountID: "123456789012", ClusterName: "default", InstanceID: "i-abc123",
+		Status: ecsv1.StatusActive, RunningTasks: 3, SentAt: pinTime,
+	}
+
+	data, err := json.Marshal(msg)
+	require.NoError(t, err)
+	assert.JSONEq(t, golden, string(data))
+
+	var decoded ecsv1.Heartbeat
+	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
+	assert.Equal(t, msg, decoded)
+}
+
+func TestStopDirective_JSON(t *testing.T) {
+	golden := `{"taskId":"t-deadbeef","reason":"drain"}`
+	msg := ecsv1.StopDirective{TaskID: "t-deadbeef", Reason: "drain"}
+
+	data, err := json.Marshal(msg)
+	require.NoError(t, err)
+	assert.JSONEq(t, golden, string(data))
+
+	var decoded ecsv1.StopDirective
+	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
+	assert.Equal(t, msg, decoded)
+}
+
+func TestStopDirective_OmitsEmptyReason(t *testing.T) {
+	data, err := json.Marshal(ecsv1.StopDirective{TaskID: "t-deadbeef"})
+	require.NoError(t, err)
+	assert.JSONEq(t, `{"taskId":"t-deadbeef"}`, string(data))
+}
+
+func TestAssign_JSON(t *testing.T) {
+	ro := true
+	start := int64(30)
+	golden := `{"accountId":"123456789012","clusterName":"default","instanceId":"i-abc123",` +
+		`"taskId":"t-deadbeef","taskArn":"arn:aws:ecs:ap-southeast-2:123456789012:task/default/t-deadbeef",` +
+		`"taskDefFamily":"web","taskDefRevision":2,` +
+		`"containers":[{"name":"app","image":"registry/app:1","essential":true,` +
+		`"readonlyRootFilesystem":true,"startTimeout":30}],` +
+		`"eniId":"eni-1","eniMac":"02:00:00:00:00:01","eniPrivateIp":"10.0.1.5","eniSubnetId":"subnet-1",` +
+		`"credId":"c-1","taskRoleArn":"arn:aws:iam::123456789012:role/task",` +
+		`"executionRoleArn":"arn:aws:iam::123456789012:role/exec",` +
+		`"assignedAt":"2026-01-15T10:30:00Z"}`
+	msg := ecsv1.Assign{
+		AccountID: "123456789012", ClusterName: "default", InstanceID: "i-abc123",
+		TaskID: "t-deadbeef", TaskARN: "arn:aws:ecs:ap-southeast-2:123456789012:task/default/t-deadbeef",
+		TaskDefFamily: "web", TaskDefRevision: 2,
+		Containers: []ecsv1.AssignContainer{{
+			Name: "app", Image: "registry/app:1", Essential: true,
+			ReadonlyRootFilesystem: &ro, StartTimeout: &start,
+		}},
+		ENIID: "eni-1", ENIMacAddress: "02:00:00:00:00:01", ENIPrivateIP: "10.0.1.5", ENISubnetID: "subnet-1",
+		CredID: "c-1", TaskRoleARN: "arn:aws:iam::123456789012:role/task",
+		ExecutionRoleARN: "arn:aws:iam::123456789012:role/exec",
+		AssignedAt:       pinTime,
+	}
+
+	data, err := json.Marshal(msg)
+	require.NoError(t, err)
+	assert.JSONEq(t, golden, string(data))
+
+	var decoded ecsv1.Assign
+	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
+	assert.Equal(t, msg, decoded)
+}
+
+func TestAssignContainer_FullFields(t *testing.T) {
+	ro, priv, pty, interactive := true, true, false, true
+	start, stop := int64(30), int64(10)
+	golden := `{"name":"app","image":"registry/app:1","cpu":256,"memoryMiB":512,"gpu":1,` +
+		`"essential":true,"command":["/bin/app"],"environment":{"ENV":"prod"},` +
+		`"portMappings":[{"containerPort":80,"hostPort":8080,"protocol":"tcp","name":"http"}],` +
+		`"logDriver":"json-file","user":"1000","readonlyRootFilesystem":true,"privileged":true,` +
+		`"pseudoTerminal":false,"interactive":true,` +
+		`"systemControls":[{"namespace":"net.core.somaxconn","value":"1024"}],` +
+		`"capAdd":["SYS_PTRACE"],"capDrop":["NET_RAW"],"startTimeout":30,"stopTimeout":10}`
+	msg := ecsv1.AssignContainer{
+		Name: "app", Image: "registry/app:1", CPU: 256, MemoryMiB: 512, GPU: 1,
+		Essential: true, Command: []string{"/bin/app"}, Environment: map[string]string{"ENV": "prod"},
+		PortMappings:           []ecsv1.PortMapping{{ContainerPort: 80, HostPort: 8080, Protocol: "tcp", Name: "http"}},
+		LogDriver:              "json-file",
+		User:                   "1000",
+		ReadonlyRootFilesystem: &ro,
+		Privileged:             &priv,
+		PseudoTerminal:         &pty,
+		Interactive:            &interactive,
+		SystemControls:         []ecsv1.SystemControl{{Namespace: "net.core.somaxconn", Value: "1024"}},
+		CapAdd:                 []string{"SYS_PTRACE"},
+		CapDrop:                []string{"NET_RAW"},
+		StartTimeout:           &start,
+		StopTimeout:            &stop,
+	}
+
+	data, err := json.Marshal(msg)
+	require.NoError(t, err)
+	assert.JSONEq(t, golden, string(data))
+
+	var decoded ecsv1.AssignContainer
+	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
+	assert.Equal(t, msg, decoded)
+}
+
+func TestAssignContainer_OmitsEmptyFields(t *testing.T) {
+	data, err := json.Marshal(ecsv1.AssignContainer{Name: "app", Image: "registry/app:1"})
+	require.NoError(t, err)
+	assert.JSONEq(t, `{"name":"app","image":"registry/app:1","essential":false}`, string(data))
+}
+
+func TestTaskState_JSON(t *testing.T) {
+	exit := 0
+	golden := `{"accountId":"123456789012","clusterName":"default","instanceId":"i-abc123",` +
+		`"taskId":"t-deadbeef","lastStatus":"STOPPED",` +
+		`"containers":[{"name":"app","status":"STOPPED","containerId":"c-1","exitCode":0,"gpuIds":["GPU-1"]}],` +
+		`"reason":"done","reportedAt":"2026-01-15T10:30:00Z"}`
+	msg := ecsv1.TaskState{
+		AccountID: "123456789012", ClusterName: "default", InstanceID: "i-abc123",
+		TaskID: "t-deadbeef", LastStatus: ecsv1.TaskStatusStopped,
+		Containers: []ecsv1.ContainerStatus{{
+			Name: "app", Status: ecsv1.TaskStatusStopped, ContainerID: "c-1",
+			ExitCode: &exit, GPUIDs: []string{"GPU-1"},
+		}},
+		Reason: "done", ReportedAt: pinTime,
+	}
+
+	data, err := json.Marshal(msg)
+	require.NoError(t, err)
+	assert.JSONEq(t, golden, string(data))
+
+	var decoded ecsv1.TaskState
+	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
+	assert.Equal(t, msg, decoded)
+}
+
+func TestLifecycleStatusConstants(t *testing.T) {
+	assert.Equal(t, "ACTIVE", ecsv1.StatusActive)
+	assert.Equal(t, "DRAINING", ecsv1.StatusDraining)
+	assert.Equal(t, "PENDING", ecsv1.TaskStatusPending)
+	assert.Equal(t, "RUNNING", ecsv1.TaskStatusRunning)
+	assert.Equal(t, "STOPPED", ecsv1.TaskStatusStopped)
+}

contracts/ecs/v1/messages_test.go

@@ -1,26 +1,26 @@
-package bus_test
+package ecsv1_test
 
 import (
 	"encoding/json"
 	"testing"
 
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 )
 
 // oldAssignContainer models an ecs-agent built before the container-runtime
 // fields were added: the wire shape AssignContainer had before user,
 // readonlyRootFilesystem, privileged and the rest. The AMI-baked decode side.
 type oldAssignContainer struct {
-	Name         string            `json:"name"`
-	Image        string            `json:"image"`
-	CPU          int               `json:"cpu,omitempty"`
-	MemoryMiB    int               `json:"memoryMiB,omitempty"`
-	GPU          int               `json:"gpu,omitempty"`
-	Essential    bool              `json:"essential"`
-	Command      []string          `json:"command,omitempty"`
-	Environment  map[string]string `json:"environment,omitempty"`
-	PortMappings []bus.PortMapping `json:"portMappings,omitempty"`
-	LogDriver    string            `json:"logDriver,omitempty"`
+	Name         string              `json:"name"`
+	Image        string              `json:"image"`
+	CPU          int                 `json:"cpu,omitempty"`
+	MemoryMiB    int                 `json:"memoryMiB,omitempty"`
+	GPU          int                 `json:"gpu,omitempty"`
+	Essential    bool                `json:"essential"`
+	Command      []string            `json:"command,omitempty"`
+	Environment  map[string]string   `json:"environment,omitempty"`
+	PortMappings []ecsv1.PortMapping `json:"portMappings,omitempty"`
+	LogDriver    string              `json:"logDriver,omitempty"`
 }
 
 // An agent baked into the ECS AMI must survive a daemon upgrade that adds
@@ -30,12 +30,12 @@ func TestAssignContainer_OlderAgentIgnoresUnknownFields(t *testing.T) {
 	ro := true
 	priv := true
 	start := int64(30)
-	current := bus.AssignContainer{
+	current := ecsv1.AssignContainer{
 		Name: "app", Image: "registry/app:1", Essential: true,
 		User:                   "1000",
 		ReadonlyRootFilesystem: &ro,
 		Privileged:             &priv,
-		SystemControls:         []bus.SystemControl{{Namespace: "net.core.somaxconn", Value: "1024"}},
+		SystemControls:         []ecsv1.SystemControl{{Namespace: "net.core.somaxconn", Value: "1024"}},
 		CapAdd:                 []string{"SYS_PTRACE"},
 		StartTimeout:           &start,
 	}

contracts/ecs/v1/subjects.go

@@ -1,12 +1,10 @@
-// Package bus defines the Layer-2 internal scheduler↔agent NATS subjects and
-// wire payloads for the ECS control plane (ecs-v1.md Q14). Both the ecs-agent
-// (publisher) and the spinifex scheduler (subscriber, Sprint 4e) import it so the
-// subject shape and message schema have a single source of truth.
+// Package ecsv1 defines version 1 of the ECS scheduler-agent NATS subjects and
+// wire payloads. The ecs-agent and the spinifex scheduler both consume this
+// shared vocabulary; the README's route table records which routes are live.
 //
 // Cluster identity is "<accountID>.<clusterName>" (matching the KV layout), not a
-// UUID. The per-AZ NATS routing inserts "{azID}" after "ecs." — keep
-// these builders the one place that knows the shape so that change stays local.
-package bus
+// UUID; these builders are the one place that knows the subject shape.
+package ecsv1
 
 import "fmt"
 
@@ -15,24 +13,27 @@ func Prefix(accountID, clusterName string) string {
 	return fmt.Sprintf("ecs.bus.%s.%s", accountID, clusterName)
 }
 
-// RegisterSubject is published by the agent at boot (agent → scheduler).
+// RegisterSubject is the per-instance registration subject (agent → scheduler).
+// The scheduler subscribes to it; nothing publishes on it today.
 func RegisterSubject(accountID, clusterName, instanceID string) string {
 	return fmt.Sprintf("%s.instance-register.%s", Prefix(accountID, clusterName), instanceID)
 }
 
-// HeartbeatSubject is published by the agent every 30s (agent → scheduler).
+// HeartbeatSubject is the per-instance liveness subject (agent → scheduler).
+// The scheduler subscribes to it; nothing publishes on it today.
 func HeartbeatSubject(accountID, clusterName, instanceID string) string {
 	return fmt.Sprintf("%s.instance-heartbeat.%s", Prefix(accountID, clusterName), instanceID)
 }
 
 // AssignSubject is the per-instance task-assignment subject (scheduler → agent).
-// The agent does not subscribe to it until Sprint 4e.
+// No production code subscribes to it today; assignments are delivered through
+// the AWS-API gateway poll path instead.
 func AssignSubject(accountID, clusterName, instanceID string) string {
 	return fmt.Sprintf("%s.assign.%s", Prefix(accountID, clusterName), instanceID)
 }
 
-// TaskStateSubject is the per-task state-report subject (agent → scheduler),
-// published from Sprint 4e.
+// TaskStateSubject is the per-task state-report subject (agent → scheduler).
+// The scheduler subscribes to it; nothing publishes on it today.
 func TaskStateSubject(accountID, clusterName, taskID string) string {
 	return fmt.Sprintf("%s.task-state.%s", Prefix(accountID, clusterName), taskID)
 }

contracts/ecs/v1/subjects_test.go

@@ -1,4 +1,4 @@
-package bus
+package ecsv1
 
 import "testing"
 

docs/service-interfaces.yaml

@@ -85,6 +85,8 @@ services:
       - spinifex/domains/eks/addon
       # Guest add-on wire contract (subject, report, staged-manifest body).
       - contracts/eks/v1
+      # ECS agent/scheduler Layer-2 bus wire contract (subjects, message types).
+      - contracts/ecs/v1
       - spinifex/domains/ochre/vector
     subscribes:
       # EC2 instance lifecycle.
@@ -162,6 +164,10 @@ services:
       # {account}.{db-instance}, so the subject is built at runtime and only the
       # prefix is declarable.
       - "rds.bus.*"
+      # ECS Layer-2 scheduler/agent bus (contracts/ecs/v1): account, cluster and
+      # instance/task IDs are runtime-built, so only the prefix is declarable.
+      # Only the leader holds these subscriptions, so there is no queue group.
+      - "ecs.bus.*"
       # Ochre vector store tenant surface (domains/ochre/vector/api.go),
       # gated behind config.OchreVectorConfig.Enabled (default off).
       - "ochre.vector.createIndex"
@@ -235,6 +241,10 @@ services:
       # → the in-guest rds-agent: set-password and the like, published on the
       # per-instance command subject and answered on its reply subject.
       - "rds.bus.*"
+      # ECS service-reconcile self-wake (contracts/ecs/v1): the worker that
+      # recorded a task-state change is rarely the leader, so this fans the
+      # wake out and the leader's own "ecs.bus.*" subscription picks it up.
+      - "ecs.bus.*"
     depends_on:
       - "vpc.*"
       - "ebs.*"

spinifex/agents/ecs/agent/agent.go

@@ -12,9 +12,9 @@ import (
 	"strings"
 	"time"
 
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/spinifex/agents/ecs/credentials"
 	ctrruntime "github.com/mulgadc/spinifex/spinifex/agents/ecs/runtime"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 )
 
 // Agent wires the ecs-agent's runtime seams: a gateway control-plane client, a
@@ -174,8 +174,8 @@ func (a *Agent) Stop() error {
 // detectCapacity reports the host's total CPU units (1 vCPU = 1024), memory,
 // and GPU count/UUIDs from the caller's nvidia-smi discovery (nil on a
 // non-GPU host).
-func detectCapacity(gpuUUIDs []string) bus.InstanceCapacity {
-	return bus.InstanceCapacity{
+func detectCapacity(gpuUUIDs []string) ecsv1.InstanceCapacity {
+	return ecsv1.InstanceCapacity{
 		CPU:       goruntime.NumCPU() * 1024,
 		MemoryMiB: memTotalMiB(),
 		GPU:       len(gpuUUIDs),

spinifex/agents/ecs/agent/controlplane.go

@@ -9,10 +9,10 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
 
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/internal/ecsgw"
 	"github.com/mulgadc/spinifex/spinifex/agents/ecs/credentials"
 	handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 )
 
 // controlPlane is the agent's channel to the AWS gateway. It replaces the direct
@@ -21,14 +21,14 @@ import (
 type controlPlane interface {
 	// Register (re-)registers the container instance. Called at boot and on every
 	// heartbeat tick — the gateway's RegisterContainerInstance is idempotent and
-	// refreshes LastSeen, folding liveness into registration (no bus.Heartbeat).
+	// refreshes LastSeen, folding liveness into registration (no ecsv1.Heartbeat).
 	Register(id identity) error
 	// SubmitTaskState reports a task's RUNNING/STOPPED transition.
-	SubmitTaskState(st bus.TaskState) error
+	SubmitTaskState(st ecsv1.TaskState) 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)
+	PollAssignments(cluster, instance string, ackAssigns, ackStops []string) ([]ecsv1.Assign, []ecsv1.StopDirective, error)
 	// ReportTaskGPU reports the local ledger's pinned device UUIDs for a task's
 	// containers, so the control plane can populate DescribeTasks gpuIds.
 	ReportTaskGPU(cluster, task string, containers []handlers_ecs.ContainerGPUReport) error
@@ -96,7 +96,7 @@ func (g *gatewayControlPlane) Register(id identity) error {
 	return nil
 }
 
-func (g *gatewayControlPlane) SubmitTaskState(st bus.TaskState) error {
+func (g *gatewayControlPlane) SubmitTaskState(st ecsv1.TaskState) error {
 	in := &ecs.SubmitTaskStateChangeInput{
 		Cluster: aws.String(st.ClusterName),
 		Task:    aws.String(st.TaskID),
@@ -124,7 +124,7 @@ func (g *gatewayControlPlane) SubmitTaskState(st bus.TaskState) error {
 	return nil
 }
 
-func (g *gatewayControlPlane) PollAssignments(cluster, instance string, ackAssigns, ackStops []string) ([]bus.Assign, []bus.StopDirective, error) {
+func (g *gatewayControlPlane) PollAssignments(cluster, instance string, ackAssigns, ackStops []string) ([]ecsv1.Assign, []ecsv1.StopDirective, error) {
 	in := &handlers_ecs.PollAssignmentsInput{
 		Cluster:           cluster,
 		ContainerInstance: instance,

spinifex/agents/ecs/agent/credinject_test.go

@@ -3,18 +3,18 @@ package agent
 import (
 	"testing"
 
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 )
 
 func TestTaskCredID(t *testing.T) {
 	cases := []struct {
 		name string
-		as   bus.Assign
+		as   ecsv1.Assign
 		want string
 	}{
-		{"no role", bus.Assign{TaskID: "t1"}, ""},
-		{"role, no credID -> taskID", bus.Assign{TaskID: "t1", TaskRoleARN: "arn:role"}, "t1"},
-		{"role + credID", bus.Assign{TaskID: "t1", CredID: "c9", TaskRoleARN: "arn:role"}, "c9"},
+		{"no role", ecsv1.Assign{TaskID: "t1"}, ""},
+		{"role, no credID -> taskID", ecsv1.Assign{TaskID: "t1", TaskRoleARN: "arn:role"}, "t1"},
+		{"role + credID", ecsv1.Assign{TaskID: "t1", CredID: "c9", TaskRoleARN: "arn:role"}, "c9"},
 	}
 	for _, tc := range cases {
 		if got := taskCredID(&tc.as); got != tc.want {

spinifex/agents/ecs/agent/heartbeat.go

@@ -9,7 +9,7 @@ import (
 // heartbeater keeps the container instance alive by periodically re-registering
 // it through the gateway. Real ECS folds liveness into ACS; v1 folds it into an
 // idempotent RegisterContainerInstance, which refreshes the instance's LastSeen
-// so the scheduler can tell live instances from dead ones (no bus.Heartbeat).
+// so the scheduler can tell live instances from dead ones (no ecsv1.Heartbeat).
 type heartbeater struct {
 	cp       controlPlane
 	id       identity

spinifex/agents/ecs/agent/reconcile.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"log/slog"
 
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	ctrruntime "github.com/mulgadc/spinifex/spinifex/agents/ecs/runtime"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 )
 
 // reconcile re-syncs the gateway with what this instance is actually running
@@ -38,7 +38,7 @@ func (a *Agent) reconcile(ctx context.Context) map[string]bool {
 	// Assign is reconstructed from labels — enough to re-register creds and tear
 	// down netns.
 	type task struct {
-		as      *bus.Assign
+		as      *ecsv1.Assign
 		running []ctrruntime.Container
 		dead    []ctrruntime.Container
 	}
@@ -53,7 +53,7 @@ func (a *Agent) reconcile(ctx context.Context) map[string]bool {
 		}
 		t := tasks[taskID]
 		if t == nil {
-			t = &task{as: &bus.Assign{
+			t = &task{as: &ecsv1.Assign{
 				TaskID:        taskID,
 				ClusterName:   a.id.ClusterName,
 				CredID:        c.Labels[labelCredID],
@@ -80,13 +80,13 @@ func (a *Agent) reconcile(ctx context.Context) map[string]bool {
 		}
 
 		a.registerTaskCreds(t.as)
-		statuses := make([]bus.ContainerStatus, 0, len(t.running))
+		statuses := make([]ecsv1.ContainerStatus, 0, len(t.running))
 		for _, c := range t.running {
 			name := c.Labels[labelContainerName]
 			go a.waitContainer(ctx, t.as, name, c.ID)
-			statuses = append(statuses, bus.ContainerStatus{Name: name, Status: bus.TaskStatusRunning, ContainerID: c.ID})
+			statuses = append(statuses, ecsv1.ContainerStatus{Name: name, Status: ecsv1.TaskStatusRunning, ContainerID: c.ID})
 		}
-		a.reportTaskState(t.as, bus.TaskStatusRunning, "", statuses)
+		a.reportTaskState(t.as, ecsv1.TaskStatusRunning, "", statuses)
 		adopted[taskID] = true
 		slog.Info("ecs-agent: re-adopted running task", "task", taskID, "containers", len(t.running))
 	}
@@ -96,9 +96,9 @@ func (a *Agent) reconcile(ctx context.Context) map[string]bool {
 // reportStoppedTask clears a task's leftover containers and tells the gateway it
 // is STOPPED, so the scheduler stops counting a task this instance is not
 // running and its service can place a replacement.
-func (a *Agent) reportStoppedTask(ctx context.Context, as *bus.Assign, dead []ctrruntime.Container) {
+func (a *Agent) reportStoppedTask(ctx context.Context, as *ecsv1.Assign, dead []ctrruntime.Container) {
 	runner := a.rt.get()
-	statuses := make([]bus.ContainerStatus, 0, len(dead))
+	statuses := make([]ecsv1.ContainerStatus, 0, len(dead))
 	for _, c := range dead {
 		if runner == nil {
 			break
@@ -106,10 +106,10 @@ func (a *Agent) reportStoppedTask(ctx context.Context, as *bus.Assign, dead []ct
 		if err := runner.Remove(ctx, c.ID); err != nil {
 			slog.Warn("ecs-agent: reconcile remove failed", "task", as.TaskID, "container", c.ID, "err", err)
 		}
-		statuses = append(statuses, bus.ContainerStatus{
-			Name: c.Labels[labelContainerName], Status: bus.TaskStatusStopped, ContainerID: c.ID,
+		statuses = append(statuses, ecsv1.ContainerStatus{
+			Name: c.Labels[labelContainerName], Status: ecsv1.TaskStatusStopped, ContainerID: c.ID,
 		})
 	}
-	a.reportTaskState(as, bus.TaskStatusStopped, "Container exited while the agent was not running", statuses)
+	a.reportTaskState(as, ecsv1.TaskStatusStopped, "Container exited while the agent was not running", statuses)
 	slog.Info("ecs-agent: reported stopped task", "task", as.TaskID, "containers", len(dead))
 }

spinifex/agents/ecs/agent/reconcile_test.go

@@ -7,8 +7,8 @@ import (
 	"testing"
 	"time"
 
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	ctrruntime "github.com/mulgadc/spinifex/spinifex/agents/ecs/runtime"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 )
 
 const testTaskRole = "arn:aws:iam::111122223333:role/task"
@@ -66,7 +66,7 @@ func TestReconcile_AdoptsRunningLabeledContainer(t *testing.T) {
 	if got := a.cred.roles["t-001"]; got != testTaskRole {
 		t.Errorf("cred not re-registered: roles[t-001] = %q, want %q", got, testTaskRole)
 	}
-	if n := countStatus(cp.taskStates(), "t-001", bus.TaskStatusRunning); n != 1 {
+	if n := countStatus(cp.taskStates(), "t-001", ecsv1.TaskStatusRunning); n != 1 {
 		t.Errorf("RUNNING reports for t-001 = %d, want 1", n)
 	}
 	waitedFor(t, rt, containerID("t-001", "web"))
@@ -89,10 +89,10 @@ func TestReconcile_ReportsTaskWhoseContainerDied(t *testing.T) {
 
 	adopted := a.reconcile(context.Background())
 
-	if n := countStatus(cp.taskStates(), "t-dead", bus.TaskStatusStopped); n != 1 {
+	if n := countStatus(cp.taskStates(), "t-dead", ecsv1.TaskStatusStopped); n != 1 {
 		t.Errorf("STOPPED reports for t-dead = %d, want 1", n)
 	}
-	if n := countStatus(cp.taskStates(), "t-dead", bus.TaskStatusRunning); n != 0 {
+	if n := countStatus(cp.taskStates(), "t-dead", ecsv1.TaskStatusRunning); n != 0 {
 		t.Errorf("t-dead must not be reported RUNNING, got %d such reports", n)
 	}
 	// Seeded so a re-delivered assignment for the dead task is acked, not re-run.
@@ -121,10 +121,10 @@ func TestReconcile_PartiallyDeadTaskStaysRunning(t *testing.T) {
 
 	a.reconcile(context.Background())
 
-	if n := countStatus(cp.taskStates(), "t-mixed", bus.TaskStatusRunning); n != 1 {
+	if n := countStatus(cp.taskStates(), "t-mixed", ecsv1.TaskStatusRunning); n != 1 {
 		t.Errorf("RUNNING reports for t-mixed = %d, want 1", n)
 	}
-	if n := countStatus(cp.taskStates(), "t-mixed", bus.TaskStatusStopped); n != 0 {
+	if n := countStatus(cp.taskStates(), "t-mixed", ecsv1.TaskStatusStopped); n != 0 {
 		t.Errorf("t-mixed must not be reported STOPPED while a container still runs, got %d", n)
 	}
 }
@@ -132,14 +132,14 @@ func TestReconcile_PartiallyDeadTaskStaysRunning(t *testing.T) {
 // pollAssignments seeded with an adopted task acks its re-delivered assignment
 // but does not re-run it, while a genuinely new task still dispatches.
 func TestPollAssignments_SeededTaskNotRerun(t *testing.T) {
-	mk := func(id string) bus.Assign {
-		return bus.Assign{
+	mk := func(id string) ecsv1.Assign {
+		return ecsv1.Assign{
 			AccountID: "123456789012", ClusterName: "default", InstanceID: "i-abc123", TaskID: id,
-			Containers: []bus.AssignContainer{{Name: "web", Image: "registry/web:1", Command: []string{"/bin/true"}}},
+			Containers: []ecsv1.AssignContainer{{Name: "web", Image: "registry/web:1", Command: []string{"/bin/true"}}},
 		}
 	}
 	adopted, fresh := mk("t-001"), mk("t-002")
-	cp := &fakeCP{pollReplies: [][]bus.Assign{
+	cp := &fakeCP{pollReplies: [][]ecsv1.Assign{
 		{adopted, fresh},
 		{adopted, fresh},
 		nil,
@@ -151,7 +151,7 @@ func TestPollAssignments_SeededTaskNotRerun(t *testing.T) {
 	go a.pollAssignments(ctx, map[string]bool{"t-001": true})
 
 	deadline := time.After(time.Second)
-	for countStatus(cp.taskStates(), "t-002", bus.TaskStatusRunning) < 1 {
+	for countStatus(cp.taskStates(), "t-002", ecsv1.TaskStatusRunning) < 1 {
 		select {
 		case <-deadline:
 			t.Fatal("fresh task t-002 was never dispatched")
@@ -161,10 +161,10 @@ func TestPollAssignments_SeededTaskNotRerun(t *testing.T) {
 	time.Sleep(20 * time.Millisecond)
 	cancel()
 
-	if n := countStatus(cp.taskStates(), "t-001", bus.TaskStatusRunning); n != 0 {
+	if n := countStatus(cp.taskStates(), "t-001", ecsv1.TaskStatusRunning); n != 0 {
 		t.Errorf("adopted task t-001 was re-run %d times, want 0", n)
 	}
-	if n := countStatus(cp.taskStates(), "t-002", bus.TaskStatusRunning); n != 1 {
+	if n := countStatus(cp.taskStates(), "t-002", ecsv1.TaskStatusRunning); n != 1 {
 		t.Errorf("fresh task t-002 dispatched %d times, want 1", n)
 	}
 	var t1Acked bool
@@ -205,7 +205,7 @@ func TestReconcile_IgnoresOtherClusterContainers(t *testing.T) {
 		t.Fatalf("adopted = %v, want only t-ok", adopted)
 	}
 	for _, id := range []string{"t-other", "t-other-dead"} {
-		if n := countStatus(cp.taskStates(), id, bus.TaskStatusStopped); n != 0 {
+		if n := countStatus(cp.taskStates(), id, ecsv1.TaskStatusStopped); n != 0 {
 			t.Errorf("reported %d STOPPED states for another cluster's task %s, want 0", n, id)
 		}
 	}
@@ -227,7 +227,7 @@ func TestReconcile_DegradesGracefully(t *testing.T) {
 
 // taskLabels emits the optional cred/role/MAC labels only when set.
 func TestTaskLabels_OmitsEmptyOptionalLabels(t *testing.T) {
-	full := taskLabels(&bus.Assign{
+	full := taskLabels(&ecsv1.Assign{
 		TaskID: "t1", ClusterName: "c", CredID: "x", TaskRoleARN: testTaskRole, ENIMacAddress: "mac",
 	}, "web")
 	for k, want := range map[string]string{
@@ -238,7 +238,7 @@ func TestTaskLabels_OmitsEmptyOptionalLabels(t *testing.T) {
 		}
 	}
 
-	bare := taskLabels(&bus.Assign{TaskID: "t1", ClusterName: "c"}, "web")
+	bare := taskLabels(&ecsv1.Assign{TaskID: "t1", ClusterName: "c"}, "web")
 	if len(bare) != 3 {
 		t.Errorf("bare labels = %v, want only taskID/containerName/clusterName", bare)
 	}

spinifex/agents/ecs/agent/registration.go

@@ -5,8 +5,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 )
 
 // identity is the container instance's stable identity, assembled at boot from
@@ -19,7 +19,7 @@ type identity struct {
 	AZ           string
 	InstanceType string
 	Hostname     string
-	Capacity     bus.InstanceCapacity
+	Capacity     ecsv1.InstanceCapacity
 	AgentVersion string
 }
 
@@ -48,7 +48,7 @@ func advertisedIdentity(id identity, runtimeReady bool) identity {
 	if runtimeReady {
 		return id
 	}
-	id.Capacity = bus.InstanceCapacity{}
+	id.Capacity = ecsv1.InstanceCapacity{}
 	return id
 }
 

spinifex/agents/ecs/agent/registration_test.go

@@ -8,8 +8,8 @@ import (
 	"testing"
 	"time"
 
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 )
 
 // fakeCP is a test double for the gateway control-plane. It records registers,
@@ -20,13 +20,13 @@ type fakeCP struct {
 
 	registers      int
 	lastRegistered identity
-	states         []bus.TaskState
+	states         []ecsv1.TaskState
 	gpuReports     []gpuReport
 	pollAcks       [][]string
 	pollStopAcks   [][]string
-	pollReplies    [][]bus.Assign
-	stopReplies    [][]bus.StopDirective
-	stopPending    []bus.StopDirective
+	pollReplies    [][]ecsv1.Assign
+	stopReplies    [][]ecsv1.StopDirective
+	stopPending    []ecsv1.StopDirective
 	pollCalls      int
 
 	registerErr bool
@@ -60,7 +60,7 @@ func (f *fakeCP) registered() identity {
 	return f.lastRegistered
 }
 
-func (f *fakeCP) SubmitTaskState(st bus.TaskState) error {
+func (f *fakeCP) SubmitTaskState(st ecsv1.TaskState) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	if f.submitErr {
@@ -70,22 +70,22 @@ func (f *fakeCP) SubmitTaskState(st bus.TaskState) error {
 	return nil
 }
 
-func (f *fakeCP) PollAssignments(_, _ string, ackAssigns, ackStops []string) ([]bus.Assign, []bus.StopDirective, error) {
+func (f *fakeCP) PollAssignments(_, _ string, ackAssigns, ackStops []string) ([]ecsv1.Assign, []ecsv1.StopDirective, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.pollAcks = append(f.pollAcks, append([]string(nil), ackAssigns...))
 	f.pollStopAcks = append(f.pollStopAcks, append([]string(nil), ackStops...))
-	var out []bus.Assign
+	var out []ecsv1.Assign
 	if f.pollCalls < len(f.pollReplies) {
 		out = f.pollReplies[f.pollCalls]
 	}
-	var stops []bus.StopDirective
+	var stops []ecsv1.StopDirective
 	if f.pollCalls < len(f.stopReplies) {
 		stops = f.stopReplies[f.pollCalls]
 	}
 	// stopPending is the durable inbox the gateway keeps: an acked directive is
 	// deleted, and everything still in it is redelivered on every poll.
-	f.stopPending = slices.DeleteFunc(f.stopPending, func(sd bus.StopDirective) bool {
+	f.stopPending = slices.DeleteFunc(f.stopPending, func(sd ecsv1.StopDirective) bool {
 		return slices.Contains(ackStops, sd.TaskID)
 	})
 	stops = slices.Concat(stops, f.stopPending)
@@ -118,10 +118,10 @@ func (f *fakeCP) registerCount() int {
 	return f.registers
 }
 
-func (f *fakeCP) taskStates() []bus.TaskState {
+func (f *fakeCP) taskStates() []ecsv1.TaskState {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	return append([]bus.TaskState(nil), f.states...)
+	return append([]ecsv1.TaskState(nil), f.states...)
 }
 
 func (f *fakeCP) acks() [][]string {
@@ -143,7 +143,7 @@ func testIdentity() identity {
 		InstanceID:   "i-abc123",
 		AZ:           "us-east-1a",
 		Hostname:     "ecs-host",
-		Capacity:     bus.InstanceCapacity{CPU: 2048, MemoryMiB: 4096},
+		Capacity:     ecsv1.InstanceCapacity{CPU: 2048, MemoryMiB: 4096},
 		AgentVersion: "test",
 	}
 }

spinifex/agents/ecs/agent/taskrun.go

@@ -9,9 +9,9 @@ import (
 	"sync"
 	"time"
 
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	ctrruntime "github.com/mulgadc/spinifex/spinifex/agents/ecs/runtime"
 	handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 )
 
 // defaultStopTimeout is applied when a container carries no explicit
@@ -127,7 +127,7 @@ func (l *stopLedger) seen(taskID string) bool {
 // Idempotent — a task with no live containers still reports STOPPED so the
 // scheduler releases its capacity. Reports whether it got as far as saying
 // STOPPED: a caller may only ack the directive when it did.
-func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool {
+func (a *Agent) stopTask(ctx context.Context, sd ecsv1.StopDirective) bool {
 	runner := a.rt.get()
 	if runner == nil {
 		slog.Warn("ecs-agent: stop deferred, no container runtime", "task", sd.TaskID)
@@ -138,7 +138,7 @@ func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool {
 		slog.Warn("ecs-agent: stop list failed", "task", sd.TaskID, "err", err)
 		return false
 	}
-	statuses := make([]bus.ContainerStatus, 0)
+	statuses := make([]ecsv1.ContainerStatus, 0)
 	for _, c := range containers {
 		if c.Labels[labelTaskID] != sd.TaskID {
 			continue
@@ -146,15 +146,15 @@ func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool {
 		if rerr := runner.Stop(ctx, c.ID, containerStopTimeout(c.Labels)); rerr != nil {
 			slog.Warn("ecs-agent: stop failed", "task", sd.TaskID, "container", c.ID, "err", rerr)
 		}
-		statuses = append(statuses, bus.ContainerStatus{
-			Name: c.Labels[labelContainerName], Status: bus.TaskStatusStopped, ContainerID: c.ID,
+		statuses = append(statuses, ecsv1.ContainerStatus{
+			Name: c.Labels[labelContainerName], Status: ecsv1.TaskStatusStopped, ContainerID: c.ID,
 		})
 	}
 	reason := sd.Reason
 	if reason == "" {
 		reason = "Task stopped"
 	}
-	a.reportTaskState(&bus.Assign{TaskID: sd.TaskID}, bus.TaskStatusStopped, reason, statuses)
+	a.reportTaskState(&ecsv1.Assign{TaskID: sd.TaskID}, ecsv1.TaskStatusStopped, reason, statuses)
 	slog.Info("ecs-agent: task reaped", "task", sd.TaskID, "containers", len(statuses))
 	return true
 }
@@ -162,10 +162,10 @@ func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool {
 // runTask pulls each container image, starts the containers, and reports task
 // state on the bus. A missing runtime or any per-container failure reports the
 // task STOPPED with a reason; success reports RUNNING and waits for exit.
-func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
+func (a *Agent) runTask(ctx context.Context, as *ecsv1.Assign) {
 	rt := a.rt.get()
 	if rt == nil {
-		a.reportTaskState(as, bus.TaskStatusStopped, "containerd unavailable on agent", nil)
+		a.reportTaskState(as, ecsv1.TaskStatusStopped, "containerd unavailable on agent", nil)
 		return
 	}
 
@@ -173,14 +173,14 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
 	netnsPath, err := a.setupTaskNetns(as)
 	if err != nil {
 		slog.Error("ecs-agent: task netns setup failed", "task", as.TaskID, "err", err)
-		a.reportTaskState(as, bus.TaskStatusStopped, "network setup failed: "+err.Error(), nil)
+		a.reportTaskState(as, ecsv1.TaskStatusStopped, "network setup failed: "+err.Error(), nil)
 		return
 	}
 
 	credID := a.registerTaskCreds(as)
 	resolver := a.pullResolver(as)
 
-	statuses := make([]bus.ContainerStatus, 0, len(as.Containers))
+	statuses := make([]ecsv1.ContainerStatus, 0, len(as.Containers))
 	for _, c := range as.Containers {
 		// startTimeout bounds pull-through-start as a unit: real ECS measures a
 		// container's time-to-RUNNING from the same starting point.
@@ -190,7 +190,7 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
 			cancel()
 			slog.Error("ecs-agent: pull failed", "task", as.TaskID, "image", c.Image, "err", err)
 			a.teardownTaskNetns(as)
-			a.reportTaskState(as, bus.TaskStatusStopped, "image pull failed: "+err.Error(), statuses)
+			a.reportTaskState(as, ecsv1.TaskStatusStopped, "image pull failed: "+err.Error(), statuses)
 			return
 		}
 
@@ -222,22 +222,22 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
 		if err != nil {
 			slog.Error("ecs-agent: run failed", "task", as.TaskID, "container", c.Name, "err", err)
 			a.teardownTaskNetns(as)
-			a.reportTaskState(as, bus.TaskStatusStopped, "container start failed: "+err.Error(), statuses)
+			a.reportTaskState(as, ecsv1.TaskStatusStopped, "container start failed: "+err.Error(), statuses)
 			return
 		}
-		statuses = append(statuses, bus.ContainerStatus{
-			Name: c.Name, Status: bus.TaskStatusRunning, ContainerID: id, GPUIDs: gpuIDs,
+		statuses = append(statuses, ecsv1.ContainerStatus{
+			Name: c.Name, Status: ecsv1.TaskStatusRunning, ContainerID: id, GPUIDs: gpuIDs,
 		})
 		go a.waitContainer(ctx, as, c.Name, id)
 	}
 
-	a.reportTaskState(as, bus.TaskStatusRunning, "", statuses)
+	a.reportTaskState(as, ecsv1.TaskStatusRunning, "", statuses)
 	slog.Info("ecs-agent: task running", "task", as.TaskID, "containers", len(statuses))
 }
 
 // waitContainer blocks until a container exits, then reports the task STOPPED
 // with the exit code. v1 stops the whole task when its first container exits.
-func (a *Agent) waitContainer(ctx context.Context, as *bus.Assign, name, containerID string) {
+func (a *Agent) waitContainer(ctx context.Context, as *ecsv1.Assign, name, containerID string) {
 	runner := a.rt.get()
 	if runner == nil {
 		slog.Warn("ecs-agent: cannot wait on container, no container runtime", "task", as.TaskID, "container", name)
@@ -250,14 +250,14 @@ func (a *Agent) waitContainer(ctx context.Context, as *bus.Assign, name, contain
 	}
 	exit := status.ExitCode
 	a.teardownTaskNetns(as)
-	statuses := []bus.ContainerStatus{{Name: name, Status: bus.TaskStatusStopped, ContainerID: containerID, ExitCode: &exit}}
-	a.reportTaskState(as, bus.TaskStatusStopped, fmt.Sprintf("container %s exited (%d)", name, exit), statuses)
+	statuses := []ecsv1.ContainerStatus{{Name: name, Status: ecsv1.TaskStatusStopped, ContainerID: containerID, ExitCode: &exit}}
+	a.reportTaskState(as, ecsv1.TaskStatusStopped, fmt.Sprintf("container %s exited (%d)", name, exit), statuses)
 	slog.Info("ecs-agent: task stopped", "task", as.TaskID, "container", name, "exitCode", exit)
 }
 
 // setupTaskNetns builds the awsvpc task netns from the hot-plugged ENI. Bridge/
 // host tasks (no ENI MAC) and a nil controller are no-ops returning an empty path.
-func (a *Agent) setupTaskNetns(as *bus.Assign) (string, error) {
+func (a *Agent) setupTaskNetns(as *ecsv1.Assign) (string, error) {
 	if as.ENIMacAddress == "" || a.netns == nil {
 		return "", nil
 	}
@@ -267,7 +267,7 @@ func (a *Agent) setupTaskNetns(as *bus.Assign) (string, error) {
 // teardownTaskNetns releases the task's IAM credential registration, then removes
 // the awsvpc task netns (the latter is a no-op for bridge/host tasks). It runs at
 // every task-stop path so credentials never outlive the task that owns them.
-func (a *Agent) teardownTaskNetns(as *bus.Assign) {
+func (a *Agent) teardownTaskNetns(as *ecsv1.Assign) {
 	if a.cred != nil {
 		a.cred.Deregister(taskCredID(as))
 	}
@@ -283,7 +283,7 @@ func (a *Agent) teardownTaskNetns(as *bus.Assign) {
 // assign carries an execution role, it authorizes pulls as that role (assumed
 // over the gateway); an empty role or a nil credential endpoint (unit tests)
 // falls back to the instance-role resolver, so a pull is never worse off.
-func (a *Agent) pullResolver(as *bus.Assign) ctrruntime.Resolver {
+func (a *Agent) pullResolver(as *ecsv1.Assign) ctrruntime.Resolver {
 	if as.ExecutionRoleARN == "" || a.cred == nil {
 		return a.resolver
 	}
@@ -293,7 +293,7 @@ func (a *Agent) pullResolver(as *bus.Assign) ctrruntime.Resolver {
 
 // registerTaskCreds registers the task's credID -> role mapping with the
 // credential endpoint and returns the credID (empty when the task has no role).
-func (a *Agent) registerTaskCreds(as *bus.Assign) string {
+func (a *Agent) registerTaskCreds(as *ecsv1.Assign) string {
 	credID := taskCredID(as)
 	if credID == "" || a.cred == nil {
 		return credID
@@ -305,7 +305,7 @@ func (a *Agent) registerTaskCreds(as *bus.Assign) string {
 // taskCredID is the credential ID a task's containers fetch credentials under:
 // the scheduler-assigned CredID, falling back to the taskID. Empty when the task
 // carries no IAM role.
-func taskCredID(as *bus.Assign) string {
+func taskCredID(as *ecsv1.Assign) string {
 	if as.TaskRoleARN == "" {
 		return ""
 	}
@@ -353,9 +353,9 @@ func containerStopTimeout(labels map[string]string) time.Duration {
 }
 
 // toRuntimeSystemControls maps the assign's sysctl pairs to the runtime
-// package's own type, keeping agents/ecs/runtime decoupled from the ecs
-// handlers' bus package.
-func toRuntimeSystemControls(in []bus.SystemControl) []ctrruntime.SystemControl {
+// package's own type, keeping agents/ecs/runtime decoupled from
+// contracts/ecs/v1.
+func toRuntimeSystemControls(in []ecsv1.SystemControl) []ctrruntime.SystemControl {
 	if len(in) == 0 {
 		return nil
 	}
@@ -386,11 +386,11 @@ func (a *Agent) pinContainerGPUs(taskID, containerName string, n int) []string {
 // reportTaskState reports a task transition through the gateway's
 // SubmitTaskStateChange action, then (RUNNING with pinned GPUs) reports the
 // per-container device UUIDs, and (STOPPED) releases them back to the ledger.
-func (a *Agent) reportTaskState(as *bus.Assign, status, reason string, containers []bus.ContainerStatus) {
+func (a *Agent) reportTaskState(as *ecsv1.Assign, status, reason string, containers []ecsv1.ContainerStatus) {
 	if a.cp == nil {
 		return
 	}
-	msg := bus.TaskState{
+	msg := ecsv1.TaskState{
 		AccountID:   a.id.AccountID,
 		ClusterName: a.id.ClusterName,
 		InstanceID:  a.id.InstanceID,
@@ -404,7 +404,7 @@ func (a *Agent) reportTaskState(as *bus.Assign, status, reason string, container
 		slog.Error("ecs-agent: report task-state failed", "task", as.TaskID, "err", err)
 	}
 	a.reportTaskGPUs(as.TaskID, containers)
-	if status == bus.TaskStatusStopped && a.gpu != nil {
+	if status == ecsv1.TaskStatusStopped && a.gpu != nil {
 		a.gpu.ReleaseTask(as.TaskID)
 	}
 }
@@ -412,7 +412,7 @@ func (a *Agent) reportTaskState(as *bus.Assign, status, reason string, container
 // reportTaskGPUs sends the control plane the pinned device UUIDs for any
 // container in the report that has them, so DescribeTasks can surface gpuIds.
 // A no-op when nothing was pinned (the common non-GPU-task case).
-func (a *Agent) reportTaskGPUs(taskID string, containers []bus.ContainerStatus) {
+func (a *Agent) reportTaskGPUs(taskID string, containers []ecsv1.ContainerStatus) {
 	reports := make([]handlers_ecs.ContainerGPUReport, 0, len(containers))
 	for _, c := range containers {
 		if len(c.GPUIDs) > 0 {
@@ -450,7 +450,7 @@ const (
 
 // taskLabels are the mulga.ecs.* labels stamped on a container. The cred/role/MAC
 // labels are omitted when empty so a container carries only what it needs.
-func taskLabels(as *bus.Assign, name string) map[string]string {
+func taskLabels(as *ecsv1.Assign, name string) map[string]string {
 	labels := map[string]string{
 		labelTaskID:        as.TaskID,
 		labelContainerName: name,

spinifex/agents/ecs/agent/taskrun_test.go

@@ -7,17 +7,17 @@ import (
 	"testing"
 	"time"
 
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	ctrruntime "github.com/mulgadc/spinifex/spinifex/agents/ecs/runtime"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 )
 
-func testAssign() *bus.Assign {
-	return &bus.Assign{
+func testAssign() *ecsv1.Assign {
+	return &ecsv1.Assign{
 		AccountID:   "123456789012",
 		ClusterName: "default",
 		InstanceID:  "i-abc123",
 		TaskID:      "t-001",
-		Containers: []bus.AssignContainer{
+		Containers: []ecsv1.AssignContainer{
 			{Name: "web", Image: "registry/web:1", Command: []string{"/bin/true"},
 				Environment: map[string]string{"FOO": "bar"}},
 		},
@@ -35,7 +35,7 @@ func TestRunTask_NoRuntimeReportsStopped(t *testing.T) {
 	a.runTask(context.Background(), testAssign())
 
 	st := cp.taskStates()
-	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+	if len(st) != 1 || st[0].LastStatus != ecsv1.TaskStatusStopped {
 		t.Fatalf("want one STOPPED state, got %+v", st)
 	}
 	if st[0].Reason == "" {
@@ -63,7 +63,7 @@ func TestRunTask_PullRunReportsRunning(t *testing.T) {
 	}
 
 	st := cp.taskStates()
-	if len(st) == 0 || st[len(st)-1].LastStatus != bus.TaskStatusRunning {
+	if len(st) == 0 || st[len(st)-1].LastStatus != ecsv1.TaskStatusRunning {
 		t.Fatalf("want RUNNING final state, got %+v", st)
 	}
 	if len(st[len(st)-1].Containers) != 1 || st[len(st)-1].Containers[0].ContainerID != "t-001-web" {
@@ -87,7 +87,7 @@ func TestRunTask_RuntimeFieldsCarriedToRunSpec(t *testing.T) {
 	as.Containers[0].Privileged = &priv
 	as.Containers[0].PseudoTerminal = &tty
 	as.Containers[0].Interactive = &interactive
-	as.Containers[0].SystemControls = []bus.SystemControl{{Namespace: "net.core.somaxconn", Value: "1024"}}
+	as.Containers[0].SystemControls = []ecsv1.SystemControl{{Namespace: "net.core.somaxconn", Value: "1024"}}
 	as.Containers[0].CapAdd = []string{"SYS_PTRACE"}
 	as.Containers[0].CapDrop = []string{"NET_RAW"}
 	a.runTask(context.Background(), as)
@@ -194,7 +194,7 @@ func TestStopTask_UsesContainerStopTimeout(t *testing.T) {
 	}}}
 	a := newRunAgent(cp, rt)
 
-	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "bye"})
+	a.stopTask(context.Background(), ecsv1.StopDirective{TaskID: "t-001", Reason: "bye"})
 
 	if len(rt.Stopped) != 1 || rt.Stopped[0] != "t-001-web" {
 		t.Fatalf("want t-001-web stopped, got %+v", rt.Stopped)
@@ -215,7 +215,7 @@ func TestStopTask_DefaultStopTimeoutWhenUnset(t *testing.T) {
 	}}}
 	a := newRunAgent(cp, rt)
 
-	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "bye"})
+	a.stopTask(context.Background(), ecsv1.StopDirective{TaskID: "t-001", Reason: "bye"})
 
 	if len(rt.StopTimeouts) != 1 || rt.StopTimeouts[0] != defaultStopTimeout {
 		t.Fatalf("want default stopTimeout %v, got %+v", defaultStopTimeout, rt.StopTimeouts)
@@ -230,7 +230,7 @@ func TestRunTask_GPUCarriedToRunSpec(t *testing.T) {
 	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
 	a := newRunAgent(cp, rt)
 	as := testAssign()
-	as.Containers = append(as.Containers, bus.AssignContainer{
+	as.Containers = append(as.Containers, ecsv1.AssignContainer{
 		Name: "trainer", Image: "registry/trainer:1", GPU: 2,
 	})
 	a.runTask(context.Background(), as)
@@ -254,17 +254,17 @@ func TestRunTask_GPUPinsAndReportsUUIDs(t *testing.T) {
 	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
 	a := newAgent(Config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, nil)
 	as := testAssign()
-	as.Containers = append(as.Containers, bus.AssignContainer{
+	as.Containers = append(as.Containers, ecsv1.AssignContainer{
 		Name: "trainer", Image: "registry/trainer:1", GPU: 1,
 	})
 	a.runTask(context.Background(), as)
 
 	st := cp.taskStates()
-	if len(st) == 0 || st[len(st)-1].LastStatus != bus.TaskStatusRunning {
+	if len(st) == 0 || st[len(st)-1].LastStatus != ecsv1.TaskStatusRunning {
 		t.Fatalf("want RUNNING final state, got %+v", st)
 	}
 	last := st[len(st)-1]
-	var trainer *bus.ContainerStatus
+	var trainer *ecsv1.ContainerStatus
 	for i := range last.Containers {
 		if last.Containers[i].Name == "trainer" {
 			trainer = &last.Containers[i]
@@ -298,7 +298,7 @@ func TestRunTask_GPUIDsReachRunSpec(t *testing.T) {
 	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
 	a := newAgent(Config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, nil)
 	as := testAssign()
-	as.Containers = append(as.Containers, bus.AssignContainer{
+	as.Containers = append(as.Containers, ecsv1.AssignContainer{
 		Name: "trainer", Image: "registry/trainer:1", GPU: 1,
 	})
 	a.runTask(context.Background(), as)
@@ -339,7 +339,7 @@ func TestRunTask_GPUShortLedgerStillRuns(t *testing.T) {
 	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
 	a := newRunAgent(cp, rt) // testIdentity() has no discovered GPU UUIDs
 	as := testAssign()
-	as.Containers = append(as.Containers, bus.AssignContainer{
+	as.Containers = append(as.Containers, ecsv1.AssignContainer{
 		Name: "trainer", Image: "registry/trainer:1", GPU: 2,
 	})
 	a.runTask(context.Background(), as)
@@ -369,7 +369,7 @@ func TestStopTask_ReleasesGPUsBackToLedger(t *testing.T) {
 		t.Fatalf("setup: want ledger fully pinned, free = %v", a.gpu.free)
 	}
 
-	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "bye"})
+	a.stopTask(context.Background(), ecsv1.StopDirective{TaskID: "t-001", Reason: "bye"})
 
 	if len(a.gpu.free) != 1 {
 		t.Errorf("want the GPU released back to the ledger, free = %v", a.gpu.free)
@@ -387,7 +387,7 @@ func TestRunTask_PullFailureReportsStopped(t *testing.T) {
 		t.Errorf("container should not run after pull failure, got %+v", rt.Runs)
 	}
 	st := cp.taskStates()
-	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+	if len(st) != 1 || st[0].LastStatus != ecsv1.TaskStatusStopped {
 		t.Fatalf("want STOPPED, got %+v", st)
 	}
 }
@@ -400,7 +400,7 @@ func TestRunTask_RunFailureReportsStopped(t *testing.T) {
 	a.runTask(context.Background(), testAssign())
 
 	st := cp.taskStates()
-	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+	if len(st) != 1 || st[0].LastStatus != ecsv1.TaskStatusStopped {
 		t.Fatalf("want STOPPED, got %+v", st)
 	}
 }
@@ -415,7 +415,7 @@ func TestRunTask_ExitReportsStoppedWithCode(t *testing.T) {
 	deadline := time.After(time.Second)
 	for {
 		st := cp.taskStates()
-		if len(st) > 0 && st[len(st)-1].LastStatus == bus.TaskStatusStopped {
+		if len(st) > 0 && st[len(st)-1].LastStatus == ecsv1.TaskStatusStopped {
 			last := st[len(st)-1]
 			if len(last.Containers) != 1 || last.Containers[0].ExitCode == nil || *last.Containers[0].ExitCode != 7 {
 				t.Fatalf("want exit code 7, got %+v", last.Containers)
@@ -433,7 +433,7 @@ func TestRunTask_ExitReportsStoppedWithCode(t *testing.T) {
 // pollAssignments dispatches each assign once and acks it on the next poll.
 func TestPollAssignments_DispatchesOnceAndAcks(t *testing.T) {
 	as := testAssign()
-	cp := &fakeCP{pollReplies: [][]bus.Assign{
+	cp := &fakeCP{pollReplies: [][]ecsv1.Assign{
 		{*as}, // poll 1: one pending assign
 		{*as}, // poll 2: still present (not yet acked when poll 2 was issued)
 		nil,   // poll 3: gateway dropped it after the ack
@@ -449,7 +449,7 @@ func TestPollAssignments_DispatchesOnceAndAcks(t *testing.T) {
 	// exactly one RUNNING report.
 	deadline := time.After(time.Second)
 	for {
-		if running := countStatus(cp.taskStates(), as.TaskID, bus.TaskStatusRunning); running >= 1 {
+		if running := countStatus(cp.taskStates(), as.TaskID, ecsv1.TaskStatusRunning); running >= 1 {
 			break
 		}
 		select {
@@ -461,7 +461,7 @@ func TestPollAssignments_DispatchesOnceAndAcks(t *testing.T) {
 	time.Sleep(20 * time.Millisecond)
 	cancel()
 
-	if running := countStatus(cp.taskStates(), as.TaskID, bus.TaskStatusRunning); running != 1 {
+	if running := countStatus(cp.taskStates(), as.TaskID, ecsv1.TaskStatusRunning); running != 1 {
 		t.Errorf("task dispatched %d times, want exactly 1 (dedup on taskID)", running)
 	}
 	// The taskID must appear in a later poll's ack list.
@@ -492,19 +492,19 @@ func TestStopTask_ReapsLabeledContainersAndReportsStopped(t *testing.T) {
 	}}
 	a := newRunAgent(cp, rt)
 
-	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "bye"})
+	a.stopTask(context.Background(), ecsv1.StopDirective{TaskID: "t-001", Reason: "bye"})
 
 	if len(rt.Stopped) != 1 || rt.Stopped[0] != "t-001-web" {
 		t.Fatalf("want only t-001-web stopped, got %+v", rt.Stopped)
 	}
 	st := cp.taskStates()
-	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+	if len(st) != 1 || st[0].LastStatus != ecsv1.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 {
+	if len(st[0].Containers) != 1 || st[0].Containers[0].Status != ecsv1.TaskStatusStopped {
 		t.Errorf("want one STOPPED container, got %+v", st[0].Containers)
 	}
 }
@@ -516,13 +516,13 @@ func TestStopTask_NoContainersStillReportsStopped(t *testing.T) {
 	rt := &ctrruntime.FakePuller{}
 	a := newRunAgent(cp, rt)
 
-	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "gone"})
+	a.stopTask(context.Background(), ecsv1.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 {
+	if len(st) != 1 || st[0].LastStatus != ecsv1.TaskStatusStopped {
 		t.Fatalf("want one STOPPED report, got %+v", st)
 	}
 }
@@ -532,8 +532,8 @@ func TestStopTask_NoContainersStillReportsStopped(t *testing.T) {
 func TestPollAssignments_StopReapsAndSuppressesAssign(t *testing.T) {
 	as := testAssign()
 	cp := &fakeCP{
-		pollReplies: [][]bus.Assign{{*as}, {*as}, nil},
-		stopPending: []bus.StopDirective{{TaskID: as.TaskID, Reason: "bye"}},
+		pollReplies: [][]ecsv1.Assign{{*as}, {*as}, nil},
+		stopPending: []ecsv1.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"}},
@@ -545,7 +545,7 @@ func TestPollAssignments_StopReapsAndSuppressesAssign(t *testing.T) {
 
 	deadline := time.After(time.Second)
 	for {
-		if stopped := countStatus(cp.taskStates(), as.TaskID, bus.TaskStatusStopped); stopped >= 1 {
+		if stopped := countStatus(cp.taskStates(), as.TaskID, ecsv1.TaskStatusStopped); stopped >= 1 {
 			break
 		}
 		select {
@@ -561,7 +561,7 @@ func TestPollAssignments_StopReapsAndSuppressesAssign(t *testing.T) {
 	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 {
+	if running := countStatus(cp.taskStates(), as.TaskID, ecsv1.TaskStatusRunning); running != 0 {
 		t.Errorf("stopping task should never report RUNNING, got %d", running)
 	}
 	acked := false
@@ -591,7 +591,7 @@ func countStopAcks(acks [][]string, taskID string) int {
 // A reap that cannot run is never acked, so the directive stays in the gateway's
 // inbox and is redelivered and retried until the runtime comes back.
 func TestPollAssignments_UnperformableStopIsNotAckedAndRetried(t *testing.T) {
-	cp := &fakeCP{stopPending: []bus.StopDirective{{TaskID: "t-001", Reason: "bye"}}}
+	cp := &fakeCP{stopPending: []ecsv1.StopDirective{{TaskID: "t-001", Reason: "bye"}}}
 	rt := &ctrruntime.FakePuller{ListErr: errors.New("containerd is down")}
 	a := newAgent(Config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
 
@@ -612,7 +612,7 @@ func TestPollAssignments_UnperformableStopIsNotAckedAndRetried(t *testing.T) {
 	if n := countStopAcks(cp.stopAcks(), "t-001"); n != 0 {
 		t.Errorf("a reap that never ran was acked %d times; stopAcks=%v", n, cp.stopAcks())
 	}
-	if stopped := countStatus(cp.taskStates(), "t-001", bus.TaskStatusStopped); stopped != 0 {
+	if stopped := countStatus(cp.taskStates(), "t-001", ecsv1.TaskStatusStopped); stopped != 0 {
 		t.Errorf("a reap that never ran reported STOPPED %d times", stopped)
 	}
 }
@@ -620,7 +620,7 @@ func TestPollAssignments_UnperformableStopIsNotAckedAndRetried(t *testing.T) {
 // Once the runtime recovers, the redelivered directive is reaped, reported
 // STOPPED and acked exactly once, which deletes it from the inbox.
 func TestPollAssignments_RecoveredStopIsAckedExactlyOnce(t *testing.T) {
-	cp := &fakeCP{stopPending: []bus.StopDirective{{TaskID: "t-001", Reason: "bye"}}}
+	cp := &fakeCP{stopPending: []ecsv1.StopDirective{{TaskID: "t-001", Reason: "bye"}}}
 	rt := &ctrruntime.FakePuller{
 		ListErr:    errors.New("containerd is down"),
 		ListErrFor: 1,
@@ -648,7 +648,7 @@ func TestPollAssignments_RecoveredStopIsAckedExactlyOnce(t *testing.T) {
 	if n := countStopAcks(cp.stopAcks(), "t-001"); n != 1 {
 		t.Errorf("stop acked %d times, want exactly 1; stopAcks=%v", n, cp.stopAcks())
 	}
-	if stopped := countStatus(cp.taskStates(), "t-001", bus.TaskStatusStopped); stopped != 1 {
+	if stopped := countStatus(cp.taskStates(), "t-001", ecsv1.TaskStatusStopped); stopped != 1 {
 		t.Errorf("task reported STOPPED %d times, want exactly 1", stopped)
 	}
 	if reaped := rt.Reaped(); len(reaped) != 1 || reaped[0] != "t-001-web" {
@@ -661,8 +661,8 @@ func TestPollAssignments_RecoveredStopIsAckedExactlyOnce(t *testing.T) {
 func TestPollAssignments_StoppedTaskIsNotRerunByRedeliveredAssign(t *testing.T) {
 	as := testAssign()
 	cp := &fakeCP{
-		pollReplies: [][]bus.Assign{{*as}, {*as}, {*as}, {*as}, {*as}, {*as}},
-		stopPending: []bus.StopDirective{{TaskID: as.TaskID, Reason: "bye"}},
+		pollReplies: [][]ecsv1.Assign{{*as}, {*as}, {*as}, {*as}, {*as}, {*as}},
+		stopPending: []ecsv1.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"}},
@@ -688,7 +688,7 @@ func TestPollAssignments_StoppedTaskIsNotRerunByRedeliveredAssign(t *testing.T)
 	if runs := rt.RunCalls(); len(runs) != 0 {
 		t.Errorf("stopped task was started by a redelivered assign: %+v", runs)
 	}
-	if running := countStatus(cp.taskStates(), as.TaskID, bus.TaskStatusRunning); running != 0 {
+	if running := countStatus(cp.taskStates(), as.TaskID, ecsv1.TaskStatusRunning); running != 0 {
 		t.Errorf("stopped task reported RUNNING %d times", running)
 	}
 }
@@ -697,7 +697,7 @@ func TestPollAssignments_StoppedTaskIsNotRerunByRedeliveredAssign(t *testing.T)
 func TestStopTask_NilRunnerNoOp(t *testing.T) {
 	cp := &fakeCP{}
 	a := newAgent(Config{}, testIdentity(), cp, nil, nil)
-	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "x"})
+	a.stopTask(context.Background(), ecsv1.StopDirective{TaskID: "t-001", Reason: "x"})
 	if len(cp.taskStates()) != 0 {
 		t.Fatalf("nil runner should report no state, got %+v", cp.taskStates())
 	}
@@ -708,7 +708,7 @@ 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"})
+	a.stopTask(context.Background(), ecsv1.StopDirective{TaskID: "t-001", Reason: "x"})
 	if len(cp.taskStates()) != 0 {
 		t.Fatalf("list error should report no state, got %+v", cp.taskStates())
 	}
@@ -725,13 +725,13 @@ func TestStopTask_EmptyReasonReapsWithDefault(t *testing.T) {
 		},
 	}}}
 	a := newRunAgent(cp, rt)
-	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001"})
+	a.stopTask(context.Background(), ecsv1.StopDirective{TaskID: "t-001"})
 
 	if len(rt.Stopped) != 1 || rt.Stopped[0] != "t-001-web" {
 		t.Fatalf("want container t-001-web reaped, got %+v", rt.Stopped)
 	}
 	st := cp.taskStates()
-	if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped {
+	if len(st) != 1 || st[0].LastStatus != ecsv1.TaskStatusStopped {
 		t.Fatalf("want one STOPPED report, got %+v", st)
 	}
 	if st[0].Reason != "Task stopped" {
@@ -739,7 +739,7 @@ func TestStopTask_EmptyReasonReapsWithDefault(t *testing.T) {
 	}
 }
 
-func countStatus(states []bus.TaskState, taskID, status string) int {
+func countStatus(states []ecsv1.TaskState, taskID, status string) int {
 	n := 0
 	for _, st := range states {
 		if st.TaskID == taskID && st.LastStatus == status {
@@ -809,10 +809,10 @@ func TestPullResolver(t *testing.T) {
 	fake := sentinelResolver{}
 	a := newAgent(Config{Region: "us-east-1"}, testIdentity(), &fakeCP{}, nil, fake)
 
-	if got := a.pullResolver(&bus.Assign{TaskID: "t1"}); got != fake {
+	if got := a.pullResolver(&ecsv1.Assign{TaskID: "t1"}); got != fake {
 		t.Fatalf("no execution role: want instance resolver fallback")
 	}
-	execAssign := &bus.Assign{TaskID: "t1", ExecutionRoleARN: "arn:aws:iam::1:role/e"}
+	execAssign := &ecsv1.Assign{TaskID: "t1", ExecutionRoleARN: "arn:aws:iam::1:role/e"}
 	if got := a.pullResolver(execAssign); got != fake {
 		t.Fatalf("nil cred endpoint: want instance resolver fallback")
 	}

spinifex/gateway/ecs/actions.go

@@ -420,7 +420,7 @@ func SubmitTaskStateChange(ctx context.Context, nc *nats.Conn, accountID string,
 
 // PollAssignments drains the calling instance's assignment inbox (replaces the
 // Layer-2 assign subscribe). Internal agent↔gateway action, not an AWS SDK
-// shape; the response carries bus.Assign with an RFC3339 time, so the gateway
+// shape; the response carries ecsv1.Assign with an RFC3339 time, so the gateway
 // encodes it with encoding/json (RawJSONActions), not the jsonutil marshaler.
 func PollAssignments(ctx context.Context, nc *nats.Conn, accountID string, body []byte, passRoleCheck PassRoleChecker) (any, error) {
 	input := new(handlers_ecs.PollAssignmentsInput)

spinifex/handlers/ecs/conv.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 )
 
 // awsStringSlice dereferences a []*string, dropping nils/empties.
@@ -266,8 +266,8 @@ func containerToAWS(c taskdefinition.Container) *ecs.ContainerDefinition {
 }
 
 // containerToAssign maps a persisted container def to its bus assign payload.
-func containerToAssign(c taskdefinition.Container) bus.AssignContainer {
-	return bus.AssignContainer{
+func containerToAssign(c taskdefinition.Container) ecsv1.AssignContainer {
+	return ecsv1.AssignContainer{
 		Name:                   c.Name,
 		Image:                  c.Image,
 		CPU:                    c.CPU,
@@ -291,24 +291,24 @@ func containerToAssign(c taskdefinition.Container) bus.AssignContainer {
 	}
 }
 
-func portMappingsToAssign(in []taskdefinition.PortMapping) []bus.PortMapping {
+func portMappingsToAssign(in []taskdefinition.PortMapping) []ecsv1.PortMapping {
 	if in == nil {
 		return nil
 	}
-	out := make([]bus.PortMapping, len(in))
+	out := make([]ecsv1.PortMapping, len(in))
 	for i, p := range in {
-		out[i] = bus.PortMapping{ContainerPort: p.ContainerPort, HostPort: p.HostPort, Protocol: p.Protocol, Name: p.Name}
+		out[i] = ecsv1.PortMapping{ContainerPort: p.ContainerPort, HostPort: p.HostPort, Protocol: p.Protocol, Name: p.Name}
 	}
 	return out
 }
 
-func systemControlsToAssign(in []taskdefinition.SystemControl) []bus.SystemControl {
+func systemControlsToAssign(in []taskdefinition.SystemControl) []ecsv1.SystemControl {
 	if in == nil {
 		return nil
 	}
-	out := make([]bus.SystemControl, len(in))
+	out := make([]ecsv1.SystemControl, len(in))
 	for i, sc := range in {
-		out[i] = bus.SystemControl{Namespace: sc.Namespace, Value: sc.Value}
+		out[i] = ecsv1.SystemControl{Namespace: sc.Namespace, Value: sc.Value}
 	}
 	return out
 }

spinifex/handlers/ecs/deployments_test.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -34,7 +34,7 @@ func driveRunning(t *testing.T, svc *Service, kv jetstream.KeyValue, cluster, na
 		if tasks[i].LastStatus != TaskStatusPending {
 			continue
 		}
-		require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+		require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 			AccountID: testAccountID, ClusterName: cluster, TaskID: tasks[i].TaskID,
 			LastStatus: TaskStatusRunning,
 		}))
@@ -51,7 +51,7 @@ func failPending(t *testing.T, svc *Service, kv jetstream.KeyValue, cluster, nam
 		if tasks[i].LastStatus != TaskStatusPending {
 			continue
 		}
-		require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+		require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 			AccountID: testAccountID, ClusterName: cluster, TaskID: tasks[i].TaskID,
 			LastStatus: TaskStatusStopped, Reason: "image pull failed",
 		}))

spinifex/handlers/ecs/eni_placement_test.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -170,7 +170,7 @@ func TestTaskStop_ReleasesENIReservation(t *testing.T) {
 
 	// The agent reporting STOPPED is what releases a reservation; a StopTask only
 	// asks for it, and the release still waits on the report.
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID:   testAccountID,
 		ClusterName: "web",
 		TaskID:      ContainerInstanceShortID(aws.StringValue(first.Tasks[0].TaskArn)),

spinifex/handlers/ecs/eni_test.go

@@ -13,8 +13,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/ecs"
 	"github.com/mulgadc/spinifex/contracts/ec2/v1"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/internal/testkit"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -291,9 +291,9 @@ func TestRecordTaskState_Awsvpc_ReleasesENIOnStopOnce(t *testing.T) {
 	require.NoError(t, err)
 	taskID := ContainerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn))
 
-	stop := &bus.TaskState{
+	stop := &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID,
-		LastStatus: bus.TaskStatusStopped, Reason: "exited",
+		LastStatus: ecsv1.TaskStatusStopped, Reason: "exited",
 	}
 	require.NoError(t, svc.recordTaskState(context.Background(), stop))
 	require.NoError(t, svc.recordTaskState(context.Background(), stop)) // re-report STOPPED

spinifex/handlers/ecs/export_test.go

@@ -3,7 +3,7 @@ package handlers_ecs
 import (
 	"context"
 
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -30,6 +30,6 @@ func (s *Service) UpsertInstance(ctx context.Context, kv jetstream.KeyValue, acc
 	return s.upsertInstance(ctx, kv, accountID, cluster, instanceID, mutate)
 }
 
-func (s *Service) RecordTaskState(ctx context.Context, msg *bus.TaskState) error {
+func (s *Service) RecordTaskState(ctx context.Context, msg *ecsv1.TaskState) error {
 	return s.recordTaskState(ctx, msg)
 }

spinifex/handlers/ecs/gpu_report_test.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -92,10 +92,10 @@ func TestRecordTaskState_PreservesGPUIDsAcrossStoppedReport(t *testing.T) {
 	}))
 
 	// RUNNING report pins the GPU UUIDs onto the container.
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: "t-1",
-		LastStatus: bus.TaskStatusRunning,
-		Containers: []bus.ContainerStatus{{Name: "trainer", Status: bus.TaskStatusRunning, GPUIDs: []string{"GPU-aaa", "GPU-bbb"}}},
+		LastStatus: ecsv1.TaskStatusRunning,
+		Containers: []ecsv1.ContainerStatus{{Name: "trainer", Status: ecsv1.TaskStatusRunning, GPUIDs: []string{"GPU-aaa", "GPU-bbb"}}},
 	}))
 
 	var rec TaskRecord
@@ -108,10 +108,10 @@ func TestRecordTaskState_PreservesGPUIDsAcrossStoppedReport(t *testing.T) {
 	// STOPPED report arrives without GPUIDs (the agent doesn't re-send them) —
 	// the previously pinned UUIDs must survive the transition.
 	exit := 0
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: "t-1",
-		LastStatus: bus.TaskStatusStopped, Reason: "exited",
-		Containers: []bus.ContainerStatus{{Name: "trainer", Status: bus.TaskStatusStopped, ExitCode: &exit}},
+		LastStatus: ecsv1.TaskStatusStopped, Reason: "exited",
+		Containers: []ecsv1.ContainerStatus{{Name: "trainer", Status: ecsv1.TaskStatusStopped, ExitCode: &exit}},
 	}))
 
 	found, err = getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
@@ -143,10 +143,10 @@ func TestRecordTaskState_ReportedGPUIDsOverwritePinned(t *testing.T) {
 		Containers:          []ContainerState{{Name: "trainer", Status: "RUNNING", GPUIDs: []string{"GPU-aaa"}}},
 	}))
 
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: "t-1",
-		LastStatus: bus.TaskStatusRunning,
-		Containers: []bus.ContainerStatus{{Name: "trainer", Status: bus.TaskStatusRunning, GPUIDs: []string{"GPU-ccc"}}},
+		LastStatus: ecsv1.TaskStatusRunning,
+		Containers: []ecsv1.ContainerStatus{{Name: "trainer", Status: ecsv1.TaskStatusRunning, GPUIDs: []string{"GPU-ccc"}}},
 	}))
 
 	var rec TaskRecord
@@ -169,17 +169,17 @@ func TestRecordTaskState_NonGPUContainerStaysEmpty(t *testing.T) {
 		Containers:          []ContainerState{{Name: "web", Status: "PENDING"}},
 	}))
 
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: "t-1",
-		LastStatus: bus.TaskStatusRunning,
-		Containers: []bus.ContainerStatus{{Name: "web", Status: bus.TaskStatusRunning}},
+		LastStatus: ecsv1.TaskStatusRunning,
+		Containers: []ecsv1.ContainerStatus{{Name: "web", Status: ecsv1.TaskStatusRunning}},
 	}))
 
 	exit := 0
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: "t-1",
-		LastStatus: bus.TaskStatusStopped,
-		Containers: []bus.ContainerStatus{{Name: "web", Status: bus.TaskStatusStopped, ExitCode: &exit}},
+		LastStatus: ecsv1.TaskStatusStopped,
+		Containers: []ecsv1.ContainerStatus{{Name: "web", Status: ecsv1.TaskStatusStopped, ExitCode: &exit}},
 	}))
 
 	var rec TaskRecord

spinifex/handlers/ecs/instances.go

@@ -10,8 +10,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -322,7 +322,7 @@ func instanceAttributes(r *InstanceRecord) []*ecs.Attribute {
 // --- Layer-2 bus event handlers (called by the scheduler) ---
 
 // recordRegister upserts a container-instance record from a bus RegisterInstance.
-func (s *Service) recordRegister(ctx context.Context, msg *bus.RegisterInstance) error {
+func (s *Service) recordRegister(ctx context.Context, msg *ecsv1.RegisterInstance) error {
 	kv, err := s.bucket(ctx, msg.AccountID)
 	if err != nil {
 		return err
@@ -339,7 +339,7 @@ func (s *Service) recordRegister(ctx context.Context, msg *bus.RegisterInstance)
 }
 
 // recordHeartbeat refreshes an instance's LastSeen and status from a bus beat.
-func (s *Service) recordHeartbeat(ctx context.Context, msg *bus.Heartbeat) error {
+func (s *Service) recordHeartbeat(ctx context.Context, msg *ecsv1.Heartbeat) error {
 	kv, err := s.bucket(ctx, msg.AccountID)
 	if err != nil {
 		return err
@@ -358,7 +358,7 @@ func (s *Service) recordHeartbeat(ctx context.Context, msg *bus.Heartbeat) error
 
 // recordTaskState applies an agent task-state report: it updates the task record
 // and, on STOPPED, releases the reserved capacity back to the instance.
-func (s *Service) recordTaskState(ctx context.Context, msg *bus.TaskState) error {
+func (s *Service) recordTaskState(ctx context.Context, msg *ecsv1.TaskState) error {
 	kv, err := s.bucket(ctx, msg.AccountID)
 	if err != nil {
 		return err
@@ -439,11 +439,11 @@ func (s *Service) recordTaskState(ctx context.Context, msg *bus.TaskState) error
 }
 
 // SubmitTaskStateChange is the AWS-API task-state path (agent → gateway → here).
-// It maps the SDK input onto the same bus.TaskState shape the Layer-2 bus
+// It maps the SDK input onto the same ecsv1.TaskState shape the Layer-2 bus
 // delivers and converges on recordTaskState, so a gateway-routed agent reports
 // state without touching NATS. The account is authoritative from accountID.
 func (s *Service) SubmitTaskStateChange(ctx context.Context, input *ecs.SubmitTaskStateChangeInput, accountID string) (*ecs.SubmitTaskStateChangeOutput, error) {
-	msg := bus.TaskState{
+	msg := ecsv1.TaskState{
 		AccountID:   accountID,
 		ClusterName: ClusterShortName(aws.StringValue(input.Cluster)),
 		TaskID:      TaskShortID(aws.StringValue(input.Task)),
@@ -452,7 +452,7 @@ func (s *Service) SubmitTaskStateChange(ctx context.Context, input *ecs.SubmitTa
 		ReportedAt:  time.Now().UTC(),
 	}
 	for _, c := range input.Containers {
-		cs := bus.ContainerStatus{
+		cs := ecsv1.ContainerStatus{
 			Name:        aws.StringValue(c.ContainerName),
 			Status:      aws.StringValue(c.Status),
 			ContainerID: aws.StringValue(c.RuntimeId),
@@ -469,7 +469,7 @@ func (s *Service) SubmitTaskStateChange(ctx context.Context, input *ecs.SubmitTa
 	// A task changing state is what moves a service's running count, so the
 	// scheduler is woken here rather than left to find out on its own deadline.
 	// The worker serving this call is rarely the leader, hence a published wake.
-	if err := s.nc.Publish(bus.ServiceReconcileSubject(accountID, msg.ClusterName), nil); err != nil {
+	if err := s.nc.Publish(ecsv1.ServiceReconcileSubject(accountID, msg.ClusterName), nil); err != nil {
 		slog.Debug("ECS: reconcile wake failed", "cluster", msg.ClusterName, "err", err)
 	}
 	return &ecs.SubmitTaskStateChangeOutput{Acknowledgment: aws.String("OK")}, nil

spinifex/handlers/ecs/poll.go

@@ -3,7 +3,7 @@ package handlers_ecs
 import (
 	"context"
 
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -25,8 +25,8 @@ type PollAssignmentsInput struct {
 // PollAssignmentsOutput carries the instance's currently pending assignments and
 // stop directives.
 type PollAssignmentsOutput struct {
-	Assignments []bus.Assign        `json:"assignments,omitempty"`
-	Stops       []bus.StopDirective `json:"stops,omitempty"`
+	Assignments []ecsv1.Assign        `json:"assignments,omitempty"`
+	Stops       []ecsv1.StopDirective `json:"stops,omitempty"`
 }
 
 // PollAssignments acks any completed assignments (deletes their inbox keys) then
@@ -61,7 +61,7 @@ func (s *Service) PollAssignments(ctx context.Context, input *PollAssignmentsInp
 	}
 	out := &PollAssignmentsOutput{}
 	for _, key := range keys {
-		var as bus.Assign
+		var as ecsv1.Assign
 		found, err := getJSON(ctx, kv, key, &as)
 		if err != nil {
 			return nil, err
@@ -76,7 +76,7 @@ func (s *Service) PollAssignments(ctx context.Context, input *PollAssignmentsInp
 		return nil, err
 	}
 	for _, key := range stopKeys {
-		var sd bus.StopDirective
+		var sd ecsv1.StopDirective
 		found, err := getJSON(ctx, kv, key, &sd)
 		if err != nil {
 			return nil, err

spinifex/handlers/ecs/poll_test.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"testing"
 
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -17,9 +17,9 @@ func TestPollAssignments_DrainsStopsAndAcks(t *testing.T) {
 	svc, _, kv := serviceTestRig(t)
 
 	require.NoError(t, putJSON(t.Context(), kv, AssignmentKey("web", "i-1", "t-1"),
-		&bus.Assign{TaskID: "t-1", ClusterName: "web", InstanceID: "i-1"}))
+		&ecsv1.Assign{TaskID: "t-1", ClusterName: "web", InstanceID: "i-1"}))
 	require.NoError(t, putJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"),
-		&bus.StopDirective{TaskID: "t-2", Reason: "drain"}))
+		&ecsv1.StopDirective{TaskID: "t-2", Reason: "drain"}))
 
 	out, err := svc.PollAssignments(context.Background(), &PollAssignmentsInput{
 		Cluster: "web", ContainerInstance: "i-1",
@@ -41,7 +41,7 @@ func TestPollAssignments_DrainsStopsAndAcks(t *testing.T) {
 	assert.Empty(t, out.Assignments)
 	assert.Empty(t, out.Stops)
 
-	found, err := getJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"), &bus.StopDirective{})
+	found, err := getJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"), &ecsv1.StopDirective{})
 	require.NoError(t, err)
 	assert.False(t, found)
 }

spinifex/handlers/ecs/records.go

@@ -6,11 +6,11 @@ import (
 	"time"
 	"uuid"
 
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 )
 
 // Lifecycle status constants matching the AWS ECS enums. Task statuses are
-// re-exported from the bus package so the scheduler and the wire payloads agree.
+// re-exported from contracts/ecs/v1 so the scheduler and the wire payloads agree.
 const (
 	ClusterStatusActive   = "ACTIVE"
 	ClusterStatusInactive = "INACTIVE"
@@ -18,9 +18,9 @@ const (
 	InstanceStatusActive   = "ACTIVE"
 	InstanceStatusDraining = "DRAINING"
 
-	TaskStatusPending = bus.TaskStatusPending
-	TaskStatusRunning = bus.TaskStatusRunning
-	TaskStatusStopped = bus.TaskStatusStopped
+	TaskStatusPending = ecsv1.TaskStatusPending
+	TaskStatusRunning = ecsv1.TaskStatusRunning
+	TaskStatusStopped = ecsv1.TaskStatusStopped
 
 	ServiceStatusActive   = "ACTIVE"
 	ServiceStatusDraining = "DRAINING"

spinifex/handlers/ecs/revisit_test.go

@@ -12,8 +12,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/reconciler"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
@@ -74,12 +74,12 @@ func TestScheduler_ATaskStateTriggersAPassAndAHeartbeatDoesNot(t *testing.T) {
 	sc, svc, kv := schedulerRig(t)
 	taskID := runOneTask(t, svc, kv)
 
-	sc.onHeartbeat(busMsg(t, &bus.Heartbeat{
+	sc.onHeartbeat(busMsg(t, &ecsv1.Heartbeat{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
 	}))
 	assert.False(t, woke(sc), "a heartbeat says only that an instance is alive: the reaper reads it, the reconcile does not")
 
-	sc.onTaskState(busMsg(t, &bus.TaskState{
+	sc.onTaskState(busMsg(t, &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", TaskID: taskID, LastStatus: TaskStatusRunning,
 	}))
 	assert.True(t, woke(sc), "a task changing state moves a service's running count, so it must wake the reconcile")
@@ -90,7 +90,7 @@ func TestScheduler_ABurstOfTaskStatesLeavesOneSignal(t *testing.T) {
 	taskID := runOneTask(t, svc, kv)
 
 	for range 5 {
-		sc.onTaskState(busMsg(t, &bus.TaskState{
+		sc.onTaskState(busMsg(t, &ecsv1.TaskState{
 			AccountID: testAccountID, ClusterName: "web", TaskID: taskID, LastStatus: TaskStatusRunning,
 		}))
 	}
@@ -238,7 +238,7 @@ func TestScheduler_AnInFlightServiceAsksToBeRevisited(t *testing.T) {
 	found, err := getJSON(t.Context(), kv, keys[0], &task)
 	require.NoError(t, err)
 	require.True(t, found)
-	require.NoError(t, svc.recordTaskState(t.Context(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(t.Context(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", TaskID: task.TaskID, LastStatus: TaskStatusRunning,
 	}))
 

spinifex/handlers/ecs/scheduler.go

@@ -10,9 +10,9 @@ import (
 	"sync"
 	"time"
 
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/reconciler"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -278,7 +278,7 @@ func (sc *Scheduler) unsubscribeBus() {
 }
 
 func (sc *Scheduler) onRegister(msg *nats.Msg) {
-	var m bus.RegisterInstance
+	var m ecsv1.RegisterInstance
 	if err := json.Unmarshal(msg.Data, &m); err != nil {
 		slog.Warn("ECS scheduler: bad register payload", "err", err)
 		return
@@ -293,7 +293,7 @@ func (sc *Scheduler) onRegister(msg *nats.Msg) {
 }
 
 func (sc *Scheduler) onHeartbeat(msg *nats.Msg) {
-	var m bus.Heartbeat
+	var m ecsv1.Heartbeat
 	if err := json.Unmarshal(msg.Data, &m); err != nil {
 		return
 	}
@@ -303,7 +303,7 @@ func (sc *Scheduler) onHeartbeat(msg *nats.Msg) {
 }
 
 func (sc *Scheduler) onTaskState(msg *nats.Msg) {
-	var m bus.TaskState
+	var m ecsv1.TaskState
 	if err := json.Unmarshal(msg.Data, &m); err != nil {
 		return
 	}

spinifex/handlers/ecs/service_test.go

@@ -9,8 +9,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/internal/testkit"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -175,11 +175,11 @@ func TestService_DescribeTaskDefinition_Unknown(t *testing.T) {
 // register path (the way the agent registers in production).
 func registerInstance(t *testing.T, svc *Service, cluster, id string, cpu, mem int) {
 	t.Helper()
-	require.NoError(t, svc.recordRegister(t.Context(), &bus.RegisterInstance{
+	require.NoError(t, svc.recordRegister(t.Context(), &ecsv1.RegisterInstance{
 		AccountID:   testAccountID,
 		ClusterName: cluster,
 		InstanceID:  id,
-		Capacity:    bus.InstanceCapacity{CPU: cpu, MemoryMiB: mem},
+		Capacity:    ecsv1.InstanceCapacity{CPU: cpu, MemoryMiB: mem},
 	}))
 }
 
@@ -290,9 +290,9 @@ func TestService_PollAssignments_AckAndReclaim(t *testing.T) {
 	out2, err := svc.RunTask(context.Background(), &ecs.RunTaskInput{Cluster: aws.String("web"), TaskDefinition: aws.String("app")}, testAccountID)
 	require.NoError(t, err)
 	task2 := ContainerInstanceShortID(aws.StringValue(out2.Tasks[0].TaskArn))
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: task2,
-		LastStatus: bus.TaskStatusStopped,
+		LastStatus: ecsv1.TaskStatusStopped,
 	}))
 	p4, err := svc.PollAssignments(context.Background(), &PollAssignmentsInput{Cluster: "web", ContainerInstance: "i-1"}, testAccountID)
 	require.NoError(t, err)
@@ -332,15 +332,15 @@ func TestService_RecordTaskState_ReleasesCapacityOnStop(t *testing.T) {
 	require.NoError(t, err)
 	taskID := ContainerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn))
 
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID,
-		LastStatus: bus.TaskStatusRunning,
+		LastStatus: ecsv1.TaskStatusRunning,
 	}))
 	exit := 0
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID,
-		LastStatus: bus.TaskStatusStopped, Reason: "exited",
-		Containers: []bus.ContainerStatus{{Name: "app", Status: bus.TaskStatusStopped, ExitCode: &exit}},
+		LastStatus: ecsv1.TaskStatusStopped, Reason: "exited",
+		Containers: []ecsv1.ContainerStatus{{Name: "app", Status: ecsv1.TaskStatusStopped, ExitCode: &exit}},
 	}))
 
 	dt, err := svc.DescribeTasks(context.Background(), &ecs.DescribeTasksInput{Cluster: aws.String("web"), Tasks: []*string{aws.String(taskID)}}, testAccountID)
@@ -379,9 +379,9 @@ func TestService_SubmitTaskStateChange_StopsTask(t *testing.T) {
 	exit := int64(0)
 	ack, err := svc.SubmitTaskStateChange(context.Background(), &ecs.SubmitTaskStateChangeInput{
 		Cluster: aws.String("web"), Task: aws.String(taskARN),
-		Status: aws.String(bus.TaskStatusStopped), Reason: aws.String("exited"),
+		Status: aws.String(ecsv1.TaskStatusStopped), Reason: aws.String("exited"),
 		Containers: []*ecs.ContainerStateChange{
-			{ContainerName: aws.String("app"), Status: aws.String(bus.TaskStatusStopped),
+			{ContainerName: aws.String("app"), Status: aws.String(ecsv1.TaskStatusStopped),
 				ExitCode: &exit, RuntimeId: aws.String("ctr-1")},
 		},
 	}, testAccountID)
@@ -405,8 +405,8 @@ func TestService_SubmitTaskStateChange_StopsTask(t *testing.T) {
 func TestService_RecordHeartbeat_UpdatesStatus(t *testing.T) {
 	svc, _ := newTestService(t)
 	registerInstance(t, svc, "web", "i-1", 1024, 2048)
-	require.NoError(t, svc.recordHeartbeat(t.Context(), &bus.Heartbeat{
-		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", Status: bus.StatusDraining,
+	require.NoError(t, svc.recordHeartbeat(t.Context(), &ecsv1.Heartbeat{
+		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", Status: ecsv1.StatusDraining,
 	}))
 	di, err := svc.DescribeContainerInstances(context.Background(), &ecs.DescribeContainerInstancesInput{
 		Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")},
@@ -540,11 +540,11 @@ func registerTaskDefGPU(t *testing.T, svc *Service, family string, cpu, mem, gpu
 // capacity via the bus register path.
 func registerInstanceGPU(t *testing.T, svc *Service, cluster, id string, cpu, mem, gpu int) {
 	t.Helper()
-	require.NoError(t, svc.recordRegister(t.Context(), &bus.RegisterInstance{
+	require.NoError(t, svc.recordRegister(t.Context(), &ecsv1.RegisterInstance{
 		AccountID:   testAccountID,
 		ClusterName: cluster,
 		InstanceID:  id,
-		Capacity:    bus.InstanceCapacity{CPU: cpu, MemoryMiB: mem, GPU: gpu},
+		Capacity:    ecsv1.InstanceCapacity{CPU: cpu, MemoryMiB: mem, GPU: gpu},
 	}))
 }
 
@@ -600,9 +600,9 @@ func TestService_RunTask_GPU_ReservesOnPlacement(t *testing.T) {
 	assert.Equal(t, 1, rec.ReservedGPU)
 
 	// Stopping the task releases the GPU reservation.
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID,
-		LastStatus: bus.TaskStatusStopped, Reason: "exited",
+		LastStatus: ecsv1.TaskStatusStopped, Reason: "exited",
 	}))
 	found, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
 	require.NoError(t, err)
@@ -688,9 +688,9 @@ func TestRegisterContainerInstance_GPU_StringSetResource(t *testing.T) {
 // path (Epic C3 register report-back).
 func TestRecordRegister_GPU_CarriesDeviceUUIDs(t *testing.T) {
 	svc, _ := newTestService(t)
-	require.NoError(t, svc.recordRegister(t.Context(), &bus.RegisterInstance{
+	require.NoError(t, svc.recordRegister(t.Context(), &ecsv1.RegisterInstance{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
-		Capacity: bus.InstanceCapacity{CPU: 1024, MemoryMiB: 2048, GPU: 2, GPUIDs: []string{"GPU-aaa", "GPU-bbb"}},
+		Capacity: ecsv1.InstanceCapacity{CPU: 1024, MemoryMiB: 2048, GPU: 2, GPUIDs: []string{"GPU-aaa", "GPU-bbb"}},
 	}))
 
 	rec := instanceStatus(t, svc, "web", "i-1")

spinifex/handlers/ecs/services_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/private/protocol/json/jsonutil"
 	"github.com/aws/aws-sdk-go/service/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -113,9 +113,9 @@ func TestService_Reconcile_ReplacesStoppedTask(t *testing.T) {
 	require.Len(t, tasks, 2)
 
 	// One task stops (agent report).
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
-		TaskID: tasks[0].TaskID, LastStatus: bus.TaskStatusStopped,
+		TaskID: tasks[0].TaskID, LastStatus: ecsv1.TaskStatusStopped,
 	}))
 	live, err := svc.listServiceTasks(t.Context(), kv, "web", "web")
 	require.NoError(t, err)
@@ -482,16 +482,16 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
 	assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.DesiredStatus))
 	assert.Empty(t, reg.deregistered)
 
-	var sd bus.StopDirective
+	var sd ecsv1.StopDirective
 	found, err := getJSON(t.Context(), 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(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
-		TaskID: "t-9", LastStatus: bus.TaskStatusStopped,
+		TaskID: "t-9", LastStatus: ecsv1.TaskStatusStopped,
 	}))
 	assert.Equal(t, []string{key(tgARN, "10.0.1.9", 80)}, reg.deregistered)
 	found, err = getJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
@@ -553,17 +553,17 @@ func TestService_TargetGroup_RegisterOnRunningDeregisterOnStopped(t *testing.T)
 	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
 
 	// RUNNING → register.
-	require.NoError(t, svc.recordTaskState(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
-		TaskID: "t-1", LastStatus: bus.TaskStatusRunning,
+		TaskID: "t-1", LastStatus: ecsv1.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(context.Background(), &bus.TaskState{
+	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1",
-		TaskID: "t-1", LastStatus: bus.TaskStatusStopped,
+		TaskID: "t-1", LastStatus: ecsv1.TaskStatusStopped,
 	}))
 	assert.Equal(t, []string{key(tgARN, "10.0.1.5", 80)}, reg.deregistered)
 }

spinifex/handlers/ecs/task_accounting_test.go

@@ -5,8 +5,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -114,7 +114,7 @@ func TestRecordTaskState_SelfExitMovesDesiredStatusToStopped(t *testing.T) {
 		&handlers_ecs.TaskRecord{TaskID: "t-1", DesiredStatus: handlers_ecs.TaskStatusRunning, LastStatus: handlers_ecs.TaskStatusRunning},
 	)
 
-	require.NoError(t, svc.RecordTaskState(t.Context(), &bus.TaskState{
+	require.NoError(t, svc.RecordTaskState(t.Context(), &ecsv1.TaskState{
 		AccountID: handlers_ecs.TestAccountID, ClusterName: "web", TaskID: "t-1",
 		LastStatus: handlers_ecs.TaskStatusStopped, Reason: "container exited (1)",
 	}))

spinifex/handlers/ecs/task_stop.go

@@ -11,8 +11,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -172,7 +172,7 @@ func (s *Service) requestStopTask(ctx context.Context, kv jetstream.KeyValue, ac
 		slog.ErrorContext(ctx, "ECS requestStopTask: persist failed", "task", task.TaskID, "err", perr)
 		return
 	}
-	sd := bus.StopDirective{TaskID: task.TaskID, Reason: reason}
+	sd := ecsv1.StopDirective{TaskID: task.TaskID, Reason: reason}
 	if perr := putJSON(ctx, kv, StopKey(task.Cluster, task.ContainerInstanceID, task.TaskID), &sd); perr != nil {
 		slog.ErrorContext(ctx, "ECS requestStopTask: post stop directive failed", "task", task.TaskID, "err", perr)
 	}

spinifex/handlers/ecs/taskdef_validation_test.go

@@ -2,7 +2,7 @@ package handlers_ecs
 
 import (
 	"context"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -192,8 +192,8 @@ func TestRunTask_AssignCarriesPortMappingsAndSystemControls(t *testing.T) {
 	require.Len(t, poll.Assignments, 1)
 	require.Len(t, poll.Assignments[0].Containers, 1)
 	c := poll.Assignments[0].Containers[0]
-	assert.Equal(t, []bus.PortMapping{{ContainerPort: 80, HostPort: 8080, Protocol: "udp", Name: "web"}}, c.PortMappings)
-	assert.Equal(t, []bus.SystemControl{{Namespace: "net.core.somaxconn", Value: "1024"}}, c.SystemControls)
+	assert.Equal(t, []ecsv1.PortMapping{{ContainerPort: 80, HostPort: 8080, Protocol: "udp", Name: "web"}}, c.PortMappings)
+	assert.Equal(t, []ecsv1.SystemControl{{Namespace: "net.core.somaxconn", Value: "1024"}}, c.SystemControls)
 }
 
 // TestRunTask_AssignCarriesGPU verifies that a
@@ -531,7 +531,7 @@ func TestRegisterTaskDefinition_RejectsSharedMemoryAndTmpfs(t *testing.T) {
 }
 
 // TestRunTask_AssignCarriesRuntimeFields verifies the enforced runtime fields
-// reach bus.AssignContainer, the daemon-to-agent wire contract — the third of
+// reach ecsv1.AssignContainer, the daemon-to-agent wire contract — the third of
 // the four drop points this fix closes (stored, sent, applied, returned).
 func TestRunTask_AssignCarriesRuntimeFields(t *testing.T) {
 	svc, _ := newTestService(t)

spinifex/handlers/ecs/tasks.go

@@ -13,8 +13,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -248,7 +248,7 @@ func (s *Service) newTaskRecord(accountID, cluster, taskID string, td *taskdefin
 // subscribing to NATS, so the bus stays host-internal. Durable + restart-safe:
 // an unacked assign survives an agent crash and is re-delivered on the next poll.
 func (s *Service) publishAssign(ctx context.Context, kv jetstream.KeyValue, accountID, cluster, instanceID string, rec *TaskRecord, td *taskdefinition.Record) error {
-	msg := bus.Assign{
+	msg := ecsv1.Assign{
 		AccountID:        accountID,
 		ClusterName:      cluster,
 		InstanceID:       instanceID,