feat(ecs): ecs-agent skeleton + containerd image pull (Sprint 4c) (#414)

* feat(ecs): ecs-agent skeleton + containerd image pull (Sprint 4c)

Add the ecs-agent binary that runs inside an ECS container instance: it
resolves host identity from IMDS, registers on the Layer-2 NATS bus, and
heartbeats while alive. Image pulls go through a containerd-backed
ImagePuller behind an interface, with an ECR resolver that mints
GetAuthorizationToken creds from the instance role.

Extract the IMDS-credential and ECR-auth flow out of
ecr-credential-provider into internal/imdscreds and internal/ecrauth so
the agent and the kubelet provider share one implementation.

Add spinifex/handlers/ecs/bus as the single source of truth for Layer-2
subject builders and wire payloads, shared by the agent now and the
scheduler in Sprint 4e.

* build: pin genproto via replace to fix containerd/v2 ambiguous import

go.opencensus.io -> grpc@v1.33.2 drags the pre-split monolithic genproto
into the module graph; it still ships googleapis/rpc/status, which collides
with the split genproto/googleapis/rpc module that containerd/ttrpc needs.

The main module never imports the monolith, so a plain require bump is
dropped by 'go mod tidy'. Pin it with a replace to a post-split version so
the fix survives tidy and contributors can run 'go mod tidy' freely.

* fix(ecs): agent boot failsafes from runtime verify

- Build the ECR gateway HTTP client lazily on first image pull instead of
  at boot. A missing/malformed gateway CA no longer kills the agent before
  it can register + heartbeat, matching the best-effort containerd path.
- Actively probe containerd with IsServing in runtime.New. The client dials
  lazily, so a dead socket previously slipped past the boot-time
  'containerd unavailable' warn and only surfaced on the first pull.
- Make the heartbeat interval configurable via ECS_HEARTBEAT_INTERVAL
  (default 30s) for test/verify and slow-link tuning.

Makefile

@@ -68,6 +68,10 @@ build-lb-agent:
 	@echo -e "\n....Building lb-agent (static)"
 	CGO_ENABLED=0 GOFIPS140=v1.0.0 go build -ldflags "-s -w" -o ./bin/lb-agent cmd/lb-agent/main.go
 
+build-ecs-agent: ## Build the ecs-agent (ships in the ECS-AMI guest; not in host `build`)
+	@echo -e "\n....Building ecs-agent (static)"
+	CGO_ENABLED=0 GOFIPS140=v1.0.0 go build -ldflags "-s -w -X main.version=$(VERSION)" -o ./bin/ecs-agent ./cmd/ecs-agent
+
 build-system-image: ## Build a system image from manifest (use IMAGE=lb or IMAGE=eks-node)
 ifndef IMAGE
 	$(error IMAGE is required. Usage: make build-system-image IMAGE=lb)
@@ -287,7 +291,7 @@ distro-arm64:
 distro-clean:
 	rm -rf dist/
 
-.PHONY: build build-ui build-installer build-lb-agent build-system-image build-eks-node-image import-eks-node-image publish-eks-node-image build-microvm-image install-microvm go_build preflight test test-cover test-race diff-coverage bench test-actions test-harness manifest-check manifest-lint manifest-lint-update \
+.PHONY: build build-ui build-installer build-lb-agent build-ecs-agent build-system-image build-eks-node-image import-eks-node-image publish-eks-node-image build-microvm-image install-microvm go_build preflight test test-cover test-race diff-coverage bench test-actions test-harness manifest-check manifest-lint manifest-lint-update \
 	deploy reinstall clean \
 	install-system install-go install-aws quickinstall \
 	lint fix govulncheck \

cmd/ecr-credential-provider/main.go

@@ -13,9 +13,6 @@ package main
 
 import (
 	"bufio"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -25,11 +22,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/credentials"
-	"github.com/aws/aws-sdk-go/aws/session"
-	"github.com/aws/aws-sdk-go/service/ecr"
-	"github.com/mulgadc/spinifex/internal/tlsconfig"
+	"github.com/mulgadc/spinifex/internal/ecrauth"
+	"github.com/mulgadc/spinifex/internal/imdscreds"
 
 	_ "github.com/mulgadc/spinifex/internal/fipsboot"
 )
@@ -44,7 +38,6 @@ const (
 	defaultEnvFile   = "/etc/spinifex-eks/agent.env"
 	defaultGatewayCA = "/etc/spinifex-eks/gateway-ca.pem"
 	defaultIMDSBase  = "http://169.254.169.254/latest"
-	imdsTokenTTL     = "21600"
 )
 
 // CredentialProviderRequest is the kubelet exec-provider request on STDIN.
@@ -80,7 +73,7 @@ type config struct {
 func main() {
 	cfg := loadConfig(defaultEnvFile)
 
-	httpClient, err := gatewayHTTPClient(cfg.GatewayCA)
+	httpClient, err := ecrauth.GatewayHTTPClient(cfg.GatewayCA)
 	if err != nil {
 		slog.Error("ecr-credential-provider: build gateway client", "err", err)
 		emitEmpty(os.Stdout)
@@ -106,155 +99,35 @@ func run(in io.Reader, out io.Writer, cfg config, httpClient *http.Client) error
 	if req.Image == "" {
 		return fmt.Errorf("request image is empty")
 	}
-	imageHost := hostFromImage(req.Image)
+	imageHost := ecrauth.HostFromImage(req.Image)
 
-	creds, err := fetchIMDSCredentials(httpClient, cfg.IMDSBase)
+	creds, err := imdscreds.Fetch(httpClient, cfg.IMDSBase)
 	if err != nil {
 		return fmt.Errorf("fetch IMDS credentials: %w", err)
 	}
 
-	username, password, proxyHost, expiresAt, err := getAuthToken(cfg, httpClient, creds)
+	tok, err := ecrauth.GetAuthorizationToken(cfg.Region, cfg.GatewayURL, httpClient,
+		creds.AccessKeyID, creds.SecretAccessKey, creds.SessionToken)
 	if err != nil {
 		return fmt.Errorf("get authorization token: %w", err)
 	}
 
-	auth := AuthConfig{Username: username, Password: password}
+	auth := AuthConfig{Username: tok.Username, Password: tok.Password}
 	authMap := map[string]AuthConfig{imageHost: auth}
-	if proxyHost != "" && proxyHost != imageHost {
-		authMap[proxyHost] = auth
+	if tok.ProxyHost != "" && tok.ProxyHost != imageHost {
+		authMap[tok.ProxyHost] = auth
 	}
 
 	resp := CredentialProviderResponse{
 		APIVersion:    credProviderAPIVersion,
 		Kind:          "CredentialProviderResponse",
 		CacheKeyType:  "Registry",
-		CacheDuration: cacheDurationFrom(expiresAt),
+		CacheDuration: cacheDurationFrom(tok.ExpiresAt),
 		Auth:          authMap,
 	}
 	return json.NewEncoder(out).Encode(&resp)
 }
 
-// imdsCreds mirrors the IMDS security-credentials JSON shape.
-type imdsCreds struct {
-	Code            string `json:"Code"`
-	AccessKeyID     string `json:"AccessKeyId"`
-	SecretAccessKey string `json:"SecretAccessKey"`
-	Token           string `json:"Token"`
-	Expiration      string `json:"Expiration"`
-}
-
-// fetchIMDSCredentials resolves the node role name then its credentials via
-// IMDSv2 (PUT token, then GET with the token header).
-func fetchIMDSCredentials(client *http.Client, base string) (imdsCreds, error) {
-	base = strings.TrimRight(base, "/")
-	token := imdsV2Token(client, base)
-
-	roleName, err := imdsGet(client, base+"/meta-data/iam/security-credentials/", token)
-	if err != nil {
-		return imdsCreds{}, fmt.Errorf("fetch role name: %w", err)
-	}
-	roleName = strings.TrimSpace(roleName)
-	if roleName == "" {
-		return imdsCreds{}, fmt.Errorf("IMDS returned empty role name")
-	}
-
-	body, err := imdsGet(client, base+"/meta-data/iam/security-credentials/"+roleName, token)
-	if err != nil {
-		return imdsCreds{}, fmt.Errorf("fetch role credentials: %w", err)
-	}
-	var creds imdsCreds
-	if err := json.Unmarshal([]byte(body), &creds); err != nil {
-		return imdsCreds{}, fmt.Errorf("decode role credentials: %w", err)
-	}
-	if creds.AccessKeyID == "" || creds.SecretAccessKey == "" {
-		return imdsCreds{}, fmt.Errorf("IMDS credentials missing access/secret key")
-	}
-	return creds, nil
-}
-
-// imdsV2Token requests an IMDSv2 session token; returns "" if the service does
-// not enforce v2 (tokenless v1 then applies).
-func imdsV2Token(client *http.Client, base string) string {
-	req, err := http.NewRequest(http.MethodPut, base+"/api/token", nil)
-	if err != nil {
-		return ""
-	}
-	req.Header.Set("X-Aws-Ec2-Metadata-Token-Ttl-Seconds", imdsTokenTTL)
-	resp, err := client.Do(req)
-	if err != nil {
-		return ""
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusOK {
-		return ""
-	}
-	body, _ := io.ReadAll(resp.Body)
-	return strings.TrimSpace(string(body))
-}
-
-func imdsGet(client *http.Client, url, token string) (string, error) {
-	req, err := http.NewRequest(http.MethodGet, url, nil)
-	if err != nil {
-		return "", err
-	}
-	if token != "" {
-		req.Header.Set("X-Aws-Ec2-Metadata-Token", token)
-	}
-	resp, err := client.Do(req)
-	if err != nil {
-		return "", err
-	}
-	defer resp.Body.Close()
-	body, _ := io.ReadAll(resp.Body)
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return "", fmt.Errorf("IMDS %s returned %d: %s", url, resp.StatusCode, strings.TrimSpace(string(body)))
-	}
-	return string(body), nil
-}
-
-// getAuthToken builds an aws-sdk ECR client pointed at the gateway and calls
-// GetAuthorizationToken, decoding the AWS:<jwt> token into Basic-auth parts. It
-// returns the username, password, the ProxyEndpoint host, and the token expiry.
-func getAuthToken(cfg config, httpClient *http.Client, creds imdsCreds) (username, password, proxyHost string, expiresAt time.Time, err error) {
-	region := cfg.Region
-	if region == "" {
-		region = "us-east-1"
-	}
-	sess, serr := session.NewSession(&aws.Config{
-		Region:      aws.String(region),
-		Endpoint:    aws.String(cfg.GatewayURL),
-		Credentials: credentials.NewStaticCredentials(creds.AccessKeyID, creds.SecretAccessKey, creds.Token),
-		HTTPClient:  httpClient,
-		DisableSSL:  aws.Bool(false),
-	})
-	if serr != nil {
-		return "", "", "", time.Time{}, fmt.Errorf("new session: %w", serr)
-	}
-
-	out, gerr := ecr.New(sess).GetAuthorizationToken(&ecr.GetAuthorizationTokenInput{})
-	if gerr != nil {
-		return "", "", "", time.Time{}, fmt.Errorf("GetAuthorizationToken: %w", gerr)
-	}
-	if len(out.AuthorizationData) == 0 || out.AuthorizationData[0].AuthorizationToken == nil {
-		return "", "", "", time.Time{}, fmt.Errorf("GetAuthorizationToken returned no authorization data")
-	}
-	data := out.AuthorizationData[0]
-
-	raw, derr := base64.StdEncoding.DecodeString(aws.StringValue(data.AuthorizationToken))
-	if derr != nil {
-		return "", "", "", time.Time{}, fmt.Errorf("decode authorization token: %w", derr)
-	}
-	user, pass, ok := strings.Cut(string(raw), ":")
-	if !ok {
-		return "", "", "", time.Time{}, fmt.Errorf("authorization token not in user:password form")
-	}
-
-	if data.ExpiresAt != nil {
-		expiresAt = aws.TimeValue(data.ExpiresAt)
-	}
-	return user, pass, hostFromImage(aws.StringValue(data.ProxyEndpoint)), expiresAt, nil
-}
-
 // cacheDurationFrom derives a Go-duration string from the token expiry minus a
 // safety margin, falling back to the default when expiry is unknown or too near.
 func cacheDurationFrom(expiresAt time.Time) string {
@@ -268,30 +141,6 @@ func cacheDurationFrom(expiresAt time.Time) string {
 	return d.Round(time.Second).String()
 }
 
-// gatewayHTTPClient builds an HTTP client trusting the gateway CA at caPath.
-// When caPath is empty it relies on the system trust store.
-func gatewayHTTPClient(caPath string) (*http.Client, error) {
-	tlsCfg := &tls.Config{
-		MinVersion:       tls.VersionTLS13,
-		CurvePreferences: tlsconfig.Curves,
-	}
-	if caPath != "" {
-		pem, err := os.ReadFile(caPath)
-		if err != nil {
-			return nil, fmt.Errorf("read gateway CA %q: %w", caPath, err)
-		}
-		pool := x509.NewCertPool()
-		if !pool.AppendCertsFromPEM(pem) {
-			return nil, fmt.Errorf("gateway CA %q has no usable certificates", caPath)
-		}
-		tlsCfg.RootCAs = pool
-	}
-	return &http.Client{
-		Timeout:   15 * time.Second,
-		Transport: &http.Transport{TLSClientConfig: tlsCfg, MaxIdleConns: 2, IdleConnTimeout: 30 * time.Second},
-	}, nil
-}
-
 // loadConfig reads the cloud-init env file then lets real env vars override. The
 // IMDS base is fixed (not in the env file) but overridable for tests.
 func loadConfig(envFile string) config {
@@ -341,18 +190,6 @@ func parseEnvFile(path string) map[string]string {
 	return out
 }
 
-// hostFromImage returns the host[:port] portion of an image ref or URL. It strips
-// any scheme and trims at the first path separator.
-func hostFromImage(image string) string {
-	if i := strings.Index(image, "://"); i >= 0 {
-		image = image[i+3:]
-	}
-	if i := strings.IndexByte(image, '/'); i >= 0 {
-		image = image[:i]
-	}
-	return image
-}
-
 // emitEmpty writes a valid CredentialProviderResponse with no auth entries.
 func emitEmpty(out io.Writer) {
 	resp := CredentialProviderResponse{

cmd/ecr-credential-provider/main_test.go

@@ -122,17 +122,3 @@ func TestRun_ReturnsBasicAuth(t *testing.T) {
 		t.Errorf("password = %q, want testjwt", auth.Password)
 	}
 }
-
-func TestHostFromImage(t *testing.T) {
-	cases := map[string]string{
-		"acct.dkr.ecr.us-east-1.spinifex.internal/demo:latest": "acct.dkr.ecr.us-east-1.spinifex.internal",
-		"https://acct.dkr.ecr.us-east-1.spinifex.internal":     "acct.dkr.ecr.us-east-1.spinifex.internal",
-		"host:9999/repo:tag": "host:9999",
-		"plainhost":          "plainhost",
-	}
-	for in, want := range cases {
-		if got := hostFromImage(in); got != want {
-			t.Errorf("hostFromImage(%q) = %q, want %q", in, got, want)
-		}
-	}
-}

cmd/ecs-agent/agent.go

@@ -0,0 +1,167 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	goruntime "runtime"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/mulgadc/spinifex/cmd/ecs-agent/credentials"
+	ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+// version is the agent build version, reported in the register message.
+// Overridable via -ldflags "-X main.version=...".
+var version = "dev"
+
+// Agent wires the ecs-agent's runtime seams: a NATS publisher for the Layer-2
+// bus, a container runtime, and an ECR resolver. It registers the host as a
+// container instance at boot and heartbeats while alive. Task assignment and
+// container lifecycle land in Sprint 4e.
+type Agent struct {
+	cfg      config
+	id       identity
+	puller   ctrruntime.ImagePuller
+	resolver ctrruntime.Resolver
+
+	reg *registrar
+	hb  *heartbeater
+
+	nc      *nats.Conn
+	closers []func() error
+}
+
+// newAgent assembles an Agent from already-built seams. Tests use this directly
+// with fakes; New builds the production seams and delegates here.
+func newAgent(cfg config, id identity, pub publisher, puller ctrruntime.ImagePuller, resolver ctrruntime.Resolver) *Agent {
+	return &Agent{
+		cfg:      cfg,
+		id:       id,
+		puller:   puller,
+		resolver: resolver,
+		reg:      newRegistrar(pub, id),
+		hb:       newHeartbeater(pub, id, cfg.Heartbeat, nil),
+	}
+}
+
+// New builds an Agent from config: it resolves the host identity from IMDS,
+// connects to NATS, builds the ECR resolver and (best-effort) the containerd
+// runtime. A containerd connect failure is logged, not fatal — registration and
+// heartbeat still run so the instance is visible while the runtime recovers.
+// The ECR gateway client is built lazily on first image pull (not here), so a
+// missing or malformed gateway CA does not stop the agent from registering.
+func New(cfg config) (*Agent, error) {
+	imdsClient := &http.Client{Timeout: 5 * time.Second}
+
+	meta, err := fetchInstanceMetadata(imdsClient, cfg.IMDSBase)
+	if err != nil {
+		return nil, fmt.Errorf("instance metadata: %w", err)
+	}
+	host, _ := os.Hostname()
+	id := identity{
+		AccountID:    meta.AccountID,
+		ClusterName:  cfg.ClusterName,
+		InstanceID:   meta.InstanceID,
+		AZ:           meta.AZ,
+		Hostname:     host,
+		Capacity:     detectCapacity(),
+		AgentVersion: version,
+	}
+
+	creds := credentials.NewIMDSProvider(imdsClient, cfg.IMDSBase)
+	resolver := newLazyECRResolver(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA)
+
+	var puller ctrruntime.ImagePuller
+	if p, perr := ctrruntime.New(cfg.ContainerdSocket); perr != nil {
+		slog.Warn("ecs-agent: containerd unavailable at boot, image pulls disabled", "err", perr)
+	} else {
+		puller = p
+	}
+
+	nc, err := nats.Connect(cfg.NATSURL, nats.Name("ecs-agent/"+id.InstanceID))
+	if err != nil {
+		return nil, fmt.Errorf("connect nats %s: %w", cfg.NATSURL, err)
+	}
+
+	a := newAgent(cfg, id, nc, puller, resolver)
+	a.nc = nc
+	if puller != nil {
+		a.closers = append(a.closers, puller.Close)
+	}
+	return a, nil
+}
+
+// Run registers the instance, starts the heartbeat loop, and blocks until ctx is
+// cancelled, then tears down.
+func (a *Agent) Run(ctx context.Context) error {
+	if err := a.reg.Register(); err != nil {
+		slog.Error("ecs-agent: initial registration failed", "err", err)
+	} else {
+		slog.Info("ecs-agent: registered",
+			"cluster", a.id.ClusterName, "instance", a.id.InstanceID,
+			"subject", bus.RegisterSubject(a.id.AccountID, a.id.ClusterName, a.id.InstanceID))
+	}
+
+	go a.hb.Run(ctx)
+
+	<-ctx.Done()
+	return a.Stop()
+}
+
+// Stop drains NATS and closes the runtime. Safe to call once.
+func (a *Agent) Stop() error {
+	var firstErr error
+	for _, c := range a.closers {
+		if err := c(); err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+	if a.nc != nil {
+		_ = a.nc.Drain()
+		a.nc.Close()
+	}
+	return firstErr
+}
+
+// detectCapacity reports the host's total CPU units (1 vCPU = 1024) and memory.
+func detectCapacity() bus.InstanceCapacity {
+	return bus.InstanceCapacity{
+		CPU:       goruntime.NumCPU() * 1024,
+		MemoryMiB: memTotalMiB(),
+	}
+}
+
+// memTotalMiB reads MemTotal from /proc/meminfo, returning 0 if unavailable.
+func memTotalMiB() int {
+	f, err := os.Open("/proc/meminfo")
+	if err != nil {
+		return 0
+	}
+	defer f.Close()
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		line := scanner.Text()
+		if !strings.HasPrefix(line, "MemTotal:") {
+			continue
+		}
+		fields := strings.Fields(line)
+		if len(fields) < 2 {
+			return 0
+		}
+		kb, err := strconv.Atoi(fields[1])
+		if err != nil {
+			return 0
+		}
+		return kb / 1024
+	}
+	return 0
+}

cmd/ecs-agent/agent_test.go

@@ -0,0 +1,59 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+func TestAgent_RunRegistersThenStopsOnContext(t *testing.T) {
+	pub := newCapturePub()
+	cfg := config{Heartbeat: 5 * time.Millisecond}
+	id := testIdentity()
+	puller := &ctrruntime.FakePuller{}
+	a := newAgent(cfg, id, pub, puller, nil)
+	a.closers = append(a.closers, puller.Close)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- a.Run(ctx) }()
+
+	// Give it time to register + emit a heartbeat or two.
+	time.Sleep(30 * time.Millisecond)
+	cancel()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Run returned error: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Run did not return after cancel")
+	}
+
+	regSubj := bus.RegisterSubject(id.AccountID, id.ClusterName, id.InstanceID)
+	if _, ok := pub.msgs[regSubj]; !ok {
+		t.Errorf("no register message on %s", regSubj)
+	}
+	hbSubj := bus.HeartbeatSubject(id.AccountID, id.ClusterName, id.InstanceID)
+	if _, ok := pub.msgs[hbSubj]; !ok {
+		t.Errorf("no heartbeat message on %s", hbSubj)
+	}
+	if !puller.Closed {
+		t.Error("Stop did not close the puller")
+	}
+}
+
+func TestDetectCapacity_Positive(t *testing.T) {
+	c := detectCapacity()
+	if c.CPU <= 0 {
+		t.Errorf("CPU = %d, want > 0", c.CPU)
+	}
+	// MemoryMiB may be 0 on platforms without /proc/meminfo; just assert non-negative.
+	if c.MemoryMiB < 0 {
+		t.Errorf("MemoryMiB = %d, want >= 0", c.MemoryMiB)
+	}
+}

cmd/ecs-agent/config.go

@@ -0,0 +1,97 @@
+package main
+
+import (
+	"bufio"
+	"os"
+	"strings"
+	"time"
+)
+
+const (
+	defaultEnvFile          = "/etc/spinifex-ecs/agent.env"
+	defaultGatewayCA        = "/etc/spinifex-ecs/gateway-ca.pem"
+	defaultIMDSBase         = "http://169.254.169.254/latest"
+	defaultContainerdSocket = "/run/containerd/containerd.sock"
+	defaultNATSURL          = "nats://127.0.0.1:4222"
+	defaultHeartbeat        = 30 * time.Second
+)
+
+// config holds the static settings the agent reads at boot. Cluster identity
+// (account ID, instance ID) is discovered at runtime from IMDS, not configured;
+// only the cluster *name* the instance was launched into is static.
+type config struct {
+	GatewayURL       string
+	GatewayCA        string
+	Region           string
+	ClusterName      string
+	NATSURL          string
+	IMDSBase         string
+	ContainerdSocket string
+	Heartbeat        time.Duration
+}
+
+// loadConfig reads the cloud-init env file then lets real env vars override.
+func loadConfig(envFile string) config {
+	env := parseEnvFile(envFile)
+	get := func(key string) string {
+		if v := os.Getenv(key); v != "" {
+			return v
+		}
+		return env[key]
+	}
+
+	cfg := config{
+		GatewayURL:       get("ECS_GATEWAY_URL"),
+		GatewayCA:        get("ECS_GATEWAY_CA"),
+		Region:           get("ECS_REGION"),
+		ClusterName:      get("ECS_CLUSTER"),
+		NATSURL:          get("ECS_NATS_URL"),
+		IMDSBase:         get("ECS_IMDS_BASE"),
+		ContainerdSocket: get("ECS_CONTAINERD_SOCKET"),
+	}
+	if cfg.GatewayCA == "" {
+		cfg.GatewayCA = defaultGatewayCA
+	}
+	if cfg.IMDSBase == "" {
+		cfg.IMDSBase = defaultIMDSBase
+	}
+	if cfg.ContainerdSocket == "" {
+		cfg.ContainerdSocket = defaultContainerdSocket
+	}
+	if cfg.NATSURL == "" {
+		cfg.NATSURL = defaultNATSURL
+	}
+	if cfg.ClusterName == "" {
+		cfg.ClusterName = "default"
+	}
+	cfg.Heartbeat = defaultHeartbeat
+	if v := get("ECS_HEARTBEAT_INTERVAL"); v != "" {
+		if d, err := time.ParseDuration(v); err == nil && d > 0 {
+			cfg.Heartbeat = d
+		}
+	}
+	return cfg
+}
+
+// parseEnvFile reads a simple KEY=value file; missing files yield an empty map.
+func parseEnvFile(path string) map[string]string {
+	out := map[string]string{}
+	f, err := os.Open(path)
+	if err != nil {
+		return out
+	}
+	defer f.Close()
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" || strings.HasPrefix(line, "#") {
+			continue
+		}
+		key, val, ok := strings.Cut(line, "=")
+		if !ok {
+			continue
+		}
+		out[strings.TrimSpace(key)] = strings.TrimSpace(val)
+	}
+	return out
+}

cmd/ecs-agent/config_test.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestLoadConfig_Defaults(t *testing.T) {
+	// Ensure env vars don't leak in from the host.
+	for _, k := range []string{"ECS_GATEWAY_URL", "ECS_GATEWAY_CA", "ECS_REGION",
+		"ECS_CLUSTER", "ECS_NATS_URL", "ECS_IMDS_BASE", "ECS_CONTAINERD_SOCKET",
+		"ECS_HEARTBEAT_INTERVAL"} {
+		t.Setenv(k, "")
+	}
+	cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env"))
+	if cfg.GatewayCA != defaultGatewayCA {
+		t.Errorf("GatewayCA = %q, want default", cfg.GatewayCA)
+	}
+	if cfg.IMDSBase != defaultIMDSBase {
+		t.Errorf("IMDSBase = %q, want default", cfg.IMDSBase)
+	}
+	if cfg.ContainerdSocket != defaultContainerdSocket {
+		t.Errorf("ContainerdSocket = %q, want default", cfg.ContainerdSocket)
+	}
+	if cfg.NATSURL != defaultNATSURL {
+		t.Errorf("NATSURL = %q, want default", cfg.NATSURL)
+	}
+	if cfg.ClusterName != "default" {
+		t.Errorf("ClusterName = %q, want default", cfg.ClusterName)
+	}
+	if cfg.Heartbeat != defaultHeartbeat {
+		t.Errorf("Heartbeat = %v, want %v", cfg.Heartbeat, defaultHeartbeat)
+	}
+}
+
+func TestLoadConfig_FileThenEnvOverride(t *testing.T) {
+	dir := t.TempDir()
+	envFile := filepath.Join(dir, "agent.env")
+	body := "# comment\nECS_GATEWAY_URL=https://gw.file\nECS_REGION=us-west-2\nECS_CLUSTER=prod\n"
+	if err := os.WriteFile(envFile, []byte(body), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("ECS_GATEWAY_URL", "https://gw.env")
+	t.Setenv("ECS_NATS_URL", "nats://10.0.0.1:4222")
+
+	cfg := loadConfig(envFile)
+	if cfg.GatewayURL != "https://gw.env" {
+		t.Errorf("GatewayURL = %q, want env override", cfg.GatewayURL)
+	}
+	if cfg.Region != "us-west-2" {
+		t.Errorf("Region = %q, want from file", cfg.Region)
+	}
+	if cfg.ClusterName != "prod" {
+		t.Errorf("ClusterName = %q, want from file", cfg.ClusterName)
+	}
+	if cfg.NATSURL != "nats://10.0.0.1:4222" {
+		t.Errorf("NATSURL = %q, want env", cfg.NATSURL)
+	}
+}
+
+func TestLoadConfig_HeartbeatOverride(t *testing.T) {
+	t.Setenv("ECS_HEARTBEAT_INTERVAL", "5s")
+	if cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env")); cfg.Heartbeat != 5*time.Second {
+		t.Errorf("Heartbeat = %v, want 5s", cfg.Heartbeat)
+	}
+	// Garbage value falls back to the default, not zero.
+	t.Setenv("ECS_HEARTBEAT_INTERVAL", "not-a-duration")
+	if cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env")); cfg.Heartbeat != defaultHeartbeat {
+		t.Errorf("Heartbeat = %v, want default on bad value", cfg.Heartbeat)
+	}
+}
+
+func TestParseEnvFile_SkipsBlankAndComments(t *testing.T) {
+	dir := t.TempDir()
+	p := filepath.Join(dir, "f.env")
+	if err := os.WriteFile(p, []byte("\n# c\nA=1\nnokeyval\nB = 2 \n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	m := parseEnvFile(p)
+	if m["A"] != "1" || m["B"] != "2" {
+		t.Errorf("parse mismatch: %#v", m)
+	}
+	if _, ok := m["nokeyval"]; ok {
+		t.Errorf("line without = should be skipped")
+	}
+}

cmd/ecs-agent/credentials/imds.go

@@ -0,0 +1,59 @@
+package credentials
+
+import (
+	"context"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/imdscreds"
+)
+
+// refreshMargin is how long before expiry the cache is considered stale.
+const refreshMargin = 5 * time.Minute
+
+// IMDSProvider fetches instance-role credentials from IMDS and caches them until
+// they near expiry. Safe for concurrent use.
+type IMDSProvider struct {
+	client *http.Client
+	base   string
+
+	mu     sync.Mutex
+	cached Credentials
+}
+
+var _ CredentialsProvider = (*IMDSProvider)(nil)
+
+// NewIMDSProvider builds a provider against the given IMDS base URL (e.g.
+// "http://169.254.169.254/latest"). A nil client uses http.DefaultClient.
+func NewIMDSProvider(client *http.Client, base string) *IMDSProvider {
+	if client == nil {
+		client = http.DefaultClient
+	}
+	return &IMDSProvider{client: client, base: base}
+}
+
+// Retrieve returns cached credentials when still valid, else fetches fresh ones.
+func (p *IMDSProvider) Retrieve(ctx context.Context) (Credentials, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.cached.Valid(refreshMargin) {
+		return p.cached, nil
+	}
+	if err := ctx.Err(); err != nil {
+		return Credentials{}, err
+	}
+
+	raw, err := imdscreds.Fetch(p.client, p.base)
+	if err != nil {
+		return Credentials{}, err
+	}
+	p.cached = Credentials{
+		AccessKeyID:     raw.AccessKeyID,
+		SecretAccessKey: raw.SecretAccessKey,
+		SessionToken:    raw.SessionToken,
+		Expiration:      raw.Expiration,
+	}
+	return p.cached, nil
+}

cmd/ecs-agent/credentials/imds_test.go

@@ -0,0 +1,104 @@
+package credentials
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func imdsStub(t *testing.T, hits *int32, creds map[string]string) *httptest.Server {
+	t.Helper()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/token", func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("v2-token"))
+	})
+	mux.HandleFunc("/meta-data/iam/security-credentials/", func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/meta-data/iam/security-credentials/" {
+			_, _ = w.Write([]byte("node-role"))
+			return
+		}
+		if hits != nil {
+			atomic.AddInt32(hits, 1)
+		}
+		_ = json.NewEncoder(w).Encode(creds)
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func TestRetrieve_FetchesAndCaches(t *testing.T) {
+	var hits int32
+	exp := time.Now().Add(6 * time.Hour).UTC().Round(time.Second)
+	srv := imdsStub(t, &hits, map[string]string{
+		"AccessKeyId":     "AKIA",
+		"SecretAccessKey": "secret",
+		"Token":           "session",
+		"Expiration":      exp.Format(time.RFC3339),
+	})
+
+	p := NewIMDSProvider(srv.Client(), srv.URL)
+	got, err := p.Retrieve(context.Background())
+	if err != nil {
+		t.Fatalf("Retrieve: %v", err)
+	}
+	if got.AccessKeyID != "AKIA" || !got.Expiration.Equal(exp) {
+		t.Errorf("creds mismatch: %+v", got)
+	}
+
+	// Second call within validity window must not re-hit IMDS.
+	if _, err := p.Retrieve(context.Background()); err != nil {
+		t.Fatalf("Retrieve 2: %v", err)
+	}
+	if hits != 1 {
+		t.Errorf("IMDS credential hits = %d, want 1 (cached)", hits)
+	}
+}
+
+func TestRetrieve_RefetchesWhenStale(t *testing.T) {
+	var hits int32
+	// Expiry inside the refresh margin → never cached, always refetched.
+	srv := imdsStub(t, &hits, map[string]string{
+		"AccessKeyId":     "AKIA",
+		"SecretAccessKey": "secret",
+		"Expiration":      time.Now().Add(time.Minute).UTC().Format(time.RFC3339),
+	})
+	p := NewIMDSProvider(srv.Client(), srv.URL)
+	if _, err := p.Retrieve(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := p.Retrieve(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if hits != 2 {
+		t.Errorf("hits = %d, want 2 (stale refetch)", hits)
+	}
+}
+
+func TestRetrieve_CancelledContext(t *testing.T) {
+	srv := imdsStub(t, nil, map[string]string{"AccessKeyId": "A", "SecretAccessKey": "B"})
+	p := NewIMDSProvider(srv.Client(), srv.URL)
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err := p.Retrieve(ctx); err == nil {
+		t.Fatal("expected error from cancelled context")
+	}
+}
+
+func TestCredentials_Valid(t *testing.T) {
+	if (Credentials{}).Valid(0) {
+		t.Error("empty creds should be invalid")
+	}
+	noExp := Credentials{AccessKeyID: "a", SecretAccessKey: "b"}
+	if !noExp.Valid(time.Hour) {
+		t.Error("zero expiry should be treated valid")
+	}
+	soon := Credentials{AccessKeyID: "a", SecretAccessKey: "b", Expiration: time.Now().Add(time.Minute)}
+	if soon.Valid(5 * time.Minute) {
+		t.Error("expiry inside margin should be invalid")
+	}
+}

cmd/ecs-agent/credentials/provider.go

@@ -0,0 +1,36 @@
+// Package credentials supplies AWS SigV4 credentials to the ecs-agent. The agent
+// signs gateway calls (ECR GetAuthorizationToken, instance register/heartbeat)
+// with the host's instance-role credentials, fetched from IMDS and cached until
+// near expiry.
+package credentials
+
+import (
+	"context"
+	"time"
+)
+
+// Credentials is a SigV4 credential set with an expiry.
+type Credentials struct {
+	AccessKeyID     string
+	SecretAccessKey string
+	SessionToken    string
+	Expiration      time.Time
+}
+
+// Valid reports whether the credentials are usable now, leaving a margin so a
+// caller does not start a request with a token about to expire mid-flight.
+func (c Credentials) Valid(margin time.Duration) bool {
+	if c.AccessKeyID == "" || c.SecretAccessKey == "" {
+		return false
+	}
+	if c.Expiration.IsZero() {
+		return true
+	}
+	return time.Now().Add(margin).Before(c.Expiration)
+}
+
+// CredentialsProvider yields current credentials, refreshing as needed. Retrieve
+// must be safe for concurrent use.
+type CredentialsProvider interface {
+	Retrieve(ctx context.Context) (Credentials, error)
+}

cmd/ecs-agent/ecrresolver.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"context"
+	"net/http"
+	"sync"
+
+	"github.com/mulgadc/spinifex/cmd/ecs-agent/credentials"
+	"github.com/mulgadc/spinifex/cmd/ecs-agent/runtime"
+	"github.com/mulgadc/spinifex/internal/ecrauth"
+)
+
+// ecrResolver implements runtime.Resolver by minting an ECR authorization token
+// per pull: it fetches the host's IMDS credentials, SigV4-signs
+// GetAuthorizationToken against the gateway, and returns the AWS:<jwt> as a
+// user/password pair for containerd.
+//
+// The gateway HTTP client is built lazily on first Authorize, not at agent boot:
+// a missing or malformed gateway CA only matters when an image is actually
+// pulled, so the agent can still register + heartbeat on a host with absent or
+// wrong ECR config.
+type ecrResolver struct {
+	creds      credentials.CredentialsProvider
+	region     string
+	gatewayURL string
+	caPath     string
+
+	mu         sync.Mutex
+	httpClient *http.Client
+}
+
+var _ runtime.Resolver = (*ecrResolver)(nil)
+
+// newECRResolver builds a resolver with an injected HTTP client (tests). The
+// client is used as-is and never rebuilt from caPath.
+func newECRResolver(creds credentials.CredentialsProvider, region, gatewayURL string, httpClient *http.Client) *ecrResolver {
+	return &ecrResolver{creds: creds, region: region, gatewayURL: gatewayURL, httpClient: httpClient}
+}
+
+// newLazyECRResolver builds a resolver that constructs its gateway HTTP client
+// from caPath on first use (production path).
+func newLazyECRResolver(creds credentials.CredentialsProvider, region, gatewayURL, caPath string) *ecrResolver {
+	return &ecrResolver{creds: creds, region: region, gatewayURL: gatewayURL, caPath: caPath}
+}
+
+// client returns the gateway HTTP client, building it from caPath on first call
+// and caching the result. An injected client (newECRResolver) is returned as-is.
+func (e *ecrResolver) client() (*http.Client, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.httpClient != nil {
+		return e.httpClient, nil
+	}
+	c, err := ecrauth.GatewayHTTPClient(e.caPath)
+	if err != nil {
+		return nil, err
+	}
+	e.httpClient = c
+	return c, nil
+}
+
+// Authorize returns the ECR token's user/password and proxy endpoint for ref.
+func (e *ecrResolver) Authorize(ctx context.Context, ref string) (user, pass, endpoint string, err error) {
+	c, err := e.creds.Retrieve(ctx)
+	if err != nil {
+		return "", "", "", err
+	}
+	httpClient, err := e.client()
+	if err != nil {
+		return "", "", "", err
+	}
+	tok, err := ecrauth.GetAuthorizationToken(e.region, e.gatewayURL, httpClient,
+		c.AccessKeyID, c.SecretAccessKey, c.SessionToken)
+	if err != nil {
+		return "", "", "", err
+	}
+	return tok.Username, tok.Password, tok.ProxyHost, nil
+}

cmd/ecs-agent/ecrresolver_test.go

@@ -0,0 +1,102 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/cmd/ecs-agent/credentials"
+)
+
+// stubCreds is a credentials.CredentialsProvider returning fixed creds or error.
+type stubCreds struct {
+	c   credentials.Credentials
+	err error
+}
+
+func (s stubCreds) Retrieve(context.Context) (credentials.Credentials, error) {
+	return s.c, s.err
+}
+
+func gatewayStub(t *testing.T, token, proxy string, status int) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.Contains(r.Header.Get("X-Amz-Target"), "GetAuthorizationToken") {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		if status != http.StatusOK {
+			w.WriteHeader(status)
+			return
+		}
+		w.Header().Set("Content-Type", "application/x-amz-json-1.1")
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"authorizationData": []map[string]any{{
+				"authorizationToken": base64.StdEncoding.EncodeToString([]byte(token)),
+				"proxyEndpoint":      proxy,
+			}},
+		})
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func trusting(srv *httptest.Server) *http.Client {
+	pool := x509.NewCertPool()
+	pool.AddCert(srv.Certificate())
+	return &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}}}
+}
+
+func TestECRResolver_Authorize(t *testing.T) {
+	srv := gatewayStub(t, "AWS:jwt", "https://123456789012.dkr.ecr.us-east-1.spinifex.internal", http.StatusOK)
+	creds := stubCreds{c: credentials.Credentials{
+		AccessKeyID: "AKIA", SecretAccessKey: "secret", SessionToken: "sess",
+		Expiration: time.Now().Add(time.Hour),
+	}}
+	r := newECRResolver(creds, "us-east-1", srv.URL, trusting(srv))
+
+	user, pass, endpoint, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1")
+	if err != nil {
+		t.Fatalf("Authorize: %v", err)
+	}
+	if user != "AWS" || pass != "jwt" {
+		t.Errorf("got %q:%q, want AWS:jwt", user, pass)
+	}
+	if endpoint != "123456789012.dkr.ecr.us-east-1.spinifex.internal" {
+		t.Errorf("endpoint = %q", endpoint)
+	}
+}
+
+func TestECRResolver_CredsError(t *testing.T) {
+	r := newECRResolver(stubCreds{err: errors.New("no imds")}, "us-east-1", "https://x", http.DefaultClient)
+	if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil {
+		t.Fatal("expected creds error to propagate")
+	}
+}
+
+func TestECRResolver_GatewayError(t *testing.T) {
+	srv := gatewayStub(t, "", "", http.StatusInternalServerError)
+	creds := stubCreds{c: credentials.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}}
+	r := newECRResolver(creds, "us-east-1", srv.URL, trusting(srv))
+	if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil {
+		t.Fatal("expected gateway error")
+	}
+}
+
+func TestECRResolver_LazyClient_BadCADeferredToAuthorize(t *testing.T) {
+	creds := stubCreds{c: credentials.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}}
+	// Construction with a bogus CA path must not fail — only Authorize does.
+	r := newLazyECRResolver(creds, "us-east-1", "https://gw", filepath.Join(t.TempDir(), "absent.pem"))
+	if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil {
+		t.Fatal("expected lazy gateway-client build to fail on missing CA")
+	}
+}

cmd/ecs-agent/heartbeat.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+// heartbeater periodically emits instance-heartbeat messages so the scheduler
+// can tell live instances from dead ones.
+type heartbeater struct {
+	pub      publisher
+	id       identity
+	interval time.Duration
+	// runningTasks reports the current running-task count; nil means zero.
+	runningTasks func() int
+}
+
+func newHeartbeater(pub publisher, id identity, interval time.Duration, running func() int) *heartbeater {
+	if interval <= 0 {
+		interval = defaultHeartbeat
+	}
+	return &heartbeater{pub: pub, id: id, interval: interval, runningTasks: running}
+}
+
+// beat publishes a single Heartbeat message.
+func (h *heartbeater) beat() error {
+	n := 0
+	if h.runningTasks != nil {
+		n = h.runningTasks()
+	}
+	msg := bus.Heartbeat{
+		AccountID:    h.id.AccountID,
+		ClusterName:  h.id.ClusterName,
+		InstanceID:   h.id.InstanceID,
+		Status:       bus.StatusActive,
+		RunningTasks: n,
+		SentAt:       time.Now().UTC(),
+	}
+	data, err := json.Marshal(&msg)
+	if err != nil {
+		return fmt.Errorf("marshal heartbeat: %w", err)
+	}
+	subj := bus.HeartbeatSubject(h.id.AccountID, h.id.ClusterName, h.id.InstanceID)
+	if err := h.pub.Publish(subj, data); err != nil {
+		return fmt.Errorf("publish heartbeat %s: %w", subj, err)
+	}
+	return nil
+}
+
+// Run beats every interval until ctx is cancelled. A failed beat is logged and
+// retried on the next tick rather than killing the loop.
+func (h *heartbeater) Run(ctx context.Context) {
+	ticker := time.NewTicker(h.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := h.beat(); err != nil {
+				slog.Warn("ecs-agent: heartbeat failed", "err", err)
+			}
+		}
+	}
+}

cmd/ecs-agent/main.go

@@ -0,0 +1,39 @@
+// ecs-agent runs inside a Spinifex ECS container instance (a guest VM booted from
+// the ECS-AMI, with containerd baked in). It registers the host with the ECS
+// scheduler over the Layer-2 NATS bus, heartbeats while alive, and pulls task
+// images from the internal ECR registry through containerd. Task assignment and
+// container lifecycle land in Sprint 4e.
+//
+// Static config (gateway URL, CA, region, cluster, NATS, containerd socket) is
+// read from the cloud-init env file /etc/spinifex-ecs/agent.env (KEY=value);
+// real env vars override it. Host identity (account, instance, AZ) comes from
+// IMDS at boot.
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+
+	_ "github.com/mulgadc/spinifex/internal/fipsboot"
+)
+
+func main() {
+	cfg := loadConfig(defaultEnvFile)
+
+	agent, err := New(cfg)
+	if err != nil {
+		slog.Error("ecs-agent: startup failed", "err", err)
+		os.Exit(1)
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	if err := agent.Run(ctx); err != nil {
+		slog.Error("ecs-agent: run failed", "err", err)
+		os.Exit(1)
+	}
+}

cmd/ecs-agent/metadata.go

@@ -0,0 +1,97 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+)
+
+// instanceMetadata is the IMDS-derived identity of the host the agent runs on.
+type instanceMetadata struct {
+	AccountID  string
+	InstanceID string
+	AZ         string
+}
+
+// fetchInstanceMetadata reads the instance-id, AZ and account-id from IMDSv2.
+// base is the IMDS root, e.g. "http://169.254.169.254/latest".
+func fetchInstanceMetadata(client *http.Client, base string) (instanceMetadata, error) {
+	token, err := imdsToken(client, base)
+	if err != nil {
+		return instanceMetadata{}, err
+	}
+
+	instanceID, err := imdsGet(client, base+"/meta-data/instance-id", token)
+	if err != nil {
+		return instanceMetadata{}, fmt.Errorf("instance-id: %w", err)
+	}
+	az, err := imdsGet(client, base+"/meta-data/placement/availability-zone", token)
+	if err != nil {
+		return instanceMetadata{}, fmt.Errorf("availability-zone: %w", err)
+	}
+
+	doc, err := imdsGet(client, base+"/dynamic/instance-identity/document", token)
+	if err != nil {
+		return instanceMetadata{}, fmt.Errorf("identity document: %w", err)
+	}
+	var ident struct {
+		AccountID string `json:"accountId"`
+	}
+	if err := json.Unmarshal([]byte(doc), &ident); err != nil {
+		return instanceMetadata{}, fmt.Errorf("parse identity document: %w", err)
+	}
+
+	return instanceMetadata{
+		AccountID:  ident.AccountID,
+		InstanceID: strings.TrimSpace(instanceID),
+		AZ:         strings.TrimSpace(az),
+	}, nil
+}
+
+// imdsToken fetches an IMDSv2 session token.
+func imdsToken(client *http.Client, base string) (string, error) {
+	req, err := http.NewRequest(http.MethodPut, base+"/api/token", nil)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("X-Aws-Ec2-Metadata-Token-Ttl-Seconds", "21600")
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("imds token: %w", err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("imds token: status %d", resp.StatusCode)
+	}
+	return string(body), nil
+}
+
+// imdsGet performs a token-authenticated IMDS GET.
+func imdsGet(client *http.Client, url, token string) (string, error) {
+	req, err := http.NewRequest(http.MethodGet, url, nil)
+	if err != nil {
+		return "", err
+	}
+	if token != "" {
+		req.Header.Set("X-Aws-Ec2-Metadata-Token", token)
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("status %d", resp.StatusCode)
+	}
+	return string(body), nil
+}

cmd/ecs-agent/metadata_test.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func metadataStub(t *testing.T, enforceV2 bool) *httptest.Server {
+	t.Helper()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/token", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPut {
+			w.WriteHeader(http.StatusMethodNotAllowed)
+			return
+		}
+		_, _ = w.Write([]byte("v2-token"))
+	})
+	guard := func(w http.ResponseWriter, r *http.Request) bool {
+		if enforceV2 && r.Header.Get("X-Aws-Ec2-Metadata-Token") != "v2-token" {
+			w.WriteHeader(http.StatusUnauthorized)
+			return false
+		}
+		return true
+	}
+	mux.HandleFunc("/meta-data/instance-id", func(w http.ResponseWriter, r *http.Request) {
+		if guard(w, r) {
+			_, _ = w.Write([]byte("i-0abc123"))
+		}
+	})
+	mux.HandleFunc("/meta-data/placement/availability-zone", func(w http.ResponseWriter, r *http.Request) {
+		if guard(w, r) {
+			_, _ = w.Write([]byte("us-east-1a"))
+		}
+	})
+	mux.HandleFunc("/dynamic/instance-identity/document", func(w http.ResponseWriter, r *http.Request) {
+		if guard(w, r) {
+			_, _ = w.Write([]byte(`{"accountId":"123456789012","region":"us-east-1"}`))
+		}
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func TestFetchInstanceMetadata_OK(t *testing.T) {
+	srv := metadataStub(t, true)
+	got, err := fetchInstanceMetadata(srv.Client(), srv.URL)
+	if err != nil {
+		t.Fatalf("fetch: %v", err)
+	}
+	if got.InstanceID != "i-0abc123" || got.AZ != "us-east-1a" || got.AccountID != "123456789012" {
+		t.Errorf("metadata mismatch: %+v", got)
+	}
+}
+
+func TestFetchInstanceMetadata_TokenRequired(t *testing.T) {
+	// Server returns 405 for non-PUT token; a GET-only client would 401 downstream.
+	srv := metadataStub(t, true)
+	if _, err := fetchInstanceMetadata(srv.Client(), srv.URL); err != nil {
+		t.Fatalf("v2 path should succeed: %v", err)
+	}
+}
+
+func TestFetchInstanceMetadata_BadDocument(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/token", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("t")) })
+	mux.HandleFunc("/meta-data/instance-id", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("i-1")) })
+	mux.HandleFunc("/meta-data/placement/availability-zone", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("z")) })
+	mux.HandleFunc("/dynamic/instance-identity/document", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("not json")) })
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	if _, err := fetchInstanceMetadata(srv.Client(), srv.URL); err == nil {
+		t.Fatal("expected parse error on bad identity document")
+	}
+}

cmd/ecs-agent/registration.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+// identity is the container instance's stable identity, assembled at boot from
+// IMDS (account, instance, AZ) and config (cluster). It is the join key the
+// scheduler uses to track this instance.
+type identity struct {
+	AccountID    string
+	ClusterName  string
+	InstanceID   string
+	AZ           string
+	Hostname     string
+	Capacity     bus.InstanceCapacity
+	AgentVersion string
+}
+
+// publisher is the subset of *nats.Conn the agent uses to emit bus messages.
+// Narrowed to an interface so tests can drive it without a live server.
+type publisher interface {
+	Publish(subject string, data []byte) error
+}
+
+var _ publisher = (*nats.Conn)(nil)
+
+// registrar emits the boot-time instance-register message.
+type registrar struct {
+	pub publisher
+	id  identity
+}
+
+func newRegistrar(pub publisher, id identity) *registrar {
+	return &registrar{pub: pub, id: id}
+}
+
+// Register publishes the RegisterInstance message on the cluster's register
+// subject. The scheduler records the container instance on receipt (Sprint 4e).
+func (r *registrar) Register() error {
+	msg := bus.RegisterInstance{
+		AccountID:    r.id.AccountID,
+		ClusterName:  r.id.ClusterName,
+		InstanceID:   r.id.InstanceID,
+		AZ:           r.id.AZ,
+		Hostname:     r.id.Hostname,
+		Capacity:     r.id.Capacity,
+		AgentVersion: r.id.AgentVersion,
+		RegisteredAt: time.Now().UTC(),
+	}
+	data, err := json.Marshal(&msg)
+	if err != nil {
+		return fmt.Errorf("marshal register: %w", err)
+	}
+	subj := bus.RegisterSubject(r.id.AccountID, r.id.ClusterName, r.id.InstanceID)
+	if err := r.pub.Publish(subj, data); err != nil {
+		return fmt.Errorf("publish register %s: %w", subj, err)
+	}
+	return nil
+}

cmd/ecs-agent/registration_test.go

@@ -0,0 +1,132 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+)
+
+// capturePub records published messages; ErrOn forces Publish to fail.
+type capturePub struct {
+	mu    sync.Mutex
+	msgs  map[string][]byte
+	count int
+	errOn bool
+}
+
+func newCapturePub() *capturePub { return &capturePub{msgs: map[string][]byte{}} }
+
+func (c *capturePub) Publish(subject string, data []byte) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.errOn {
+		return errors.New("publish boom")
+	}
+	c.count++
+	cp := make([]byte, len(data))
+	copy(cp, data)
+	c.msgs[subject] = cp
+	return nil
+}
+
+func (c *capturePub) calls() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.count
+}
+
+func testIdentity() identity {
+	return identity{
+		AccountID:    "123456789012",
+		ClusterName:  "default",
+		InstanceID:   "i-abc123",
+		AZ:           "us-east-1a",
+		Hostname:     "ecs-host",
+		Capacity:     bus.InstanceCapacity{CPU: 2048, MemoryMiB: 4096},
+		AgentVersion: "test",
+	}
+}
+
+func TestRegistrar_Register(t *testing.T) {
+	pub := newCapturePub()
+	id := testIdentity()
+	if err := newRegistrar(pub, id).Register(); err != nil {
+		t.Fatalf("Register: %v", err)
+	}
+	subj := bus.RegisterSubject(id.AccountID, id.ClusterName, id.InstanceID)
+	raw, ok := pub.msgs[subj]
+	if !ok {
+		t.Fatalf("no message on %s; got %v", subj, pub.msgs)
+	}
+	var got bus.RegisterInstance
+	if err := json.Unmarshal(raw, &got); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if got.InstanceID != id.InstanceID || got.Capacity.CPU != 2048 {
+		t.Errorf("payload mismatch: %+v", got)
+	}
+	if got.RegisteredAt.IsZero() {
+		t.Error("RegisteredAt not set")
+	}
+}
+
+func TestRegistrar_PublishError(t *testing.T) {
+	pub := newCapturePub()
+	pub.errOn = true
+	if err := newRegistrar(pub, testIdentity()).Register(); err == nil {
+		t.Fatal("expected publish error")
+	}
+}
+
+func TestHeartbeater_Beat(t *testing.T) {
+	pub := newCapturePub()
+	id := testIdentity()
+	h := newHeartbeater(pub, id, time.Second, func() int { return 3 })
+	if err := h.beat(); err != nil {
+		t.Fatalf("beat: %v", err)
+	}
+	subj := bus.HeartbeatSubject(id.AccountID, id.ClusterName, id.InstanceID)
+	var got bus.Heartbeat
+	if err := json.Unmarshal(pub.msgs[subj], &got); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if got.Status != bus.StatusActive || got.RunningTasks != 3 {
+		t.Errorf("payload mismatch: %+v", got)
+	}
+}
+
+func TestHeartbeater_NilRunningTasksIsZero(t *testing.T) {
+	pub := newCapturePub()
+	h := newHeartbeater(pub, testIdentity(), 0, nil)
+	if h.interval != defaultHeartbeat {
+		t.Errorf("interval = %v, want default", h.interval)
+	}
+	if err := h.beat(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestHeartbeater_RunStopsOnContext(t *testing.T) {
+	pub := newCapturePub()
+	h := newHeartbeater(pub, testIdentity(), 5*time.Millisecond, nil)
+	ctx, cancel := context.WithCancel(context.Background())
+
+	done := make(chan struct{})
+	go func() { h.Run(ctx); close(done) }()
+
+	time.Sleep(25 * time.Millisecond)
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Run did not stop on context cancel")
+	}
+	if pub.calls() == 0 {
+		t.Error("expected at least one heartbeat before cancel")
+	}
+}

cmd/ecs-agent/runtime/containerd.go

@@ -0,0 +1,100 @@
+package runtime
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	containerd "github.com/containerd/containerd/v2/client"
+	"github.com/containerd/containerd/v2/core/remotes"
+	"github.com/containerd/containerd/v2/core/remotes/docker"
+	"github.com/containerd/containerd/v2/pkg/namespaces"
+)
+
+// ecsNamespace isolates ecs-agent images/containers in containerd, mirroring the
+// "moby"/"k8s.io" convention other runtimes use.
+const ecsNamespace = "ecs"
+
+// servingProbeTimeout bounds the boot-time daemon liveness check.
+const servingProbeTimeout = 3 * time.Second
+
+// containerdPuller drives a real containerd daemon over its unix socket.
+type containerdPuller struct {
+	client *containerd.Client
+}
+
+var _ ImagePuller = (*containerdPuller)(nil)
+
+// New connects to the containerd daemon at socket (e.g.
+// "/run/containerd/containerd.sock") and returns an ImagePuller. The containerd
+// client dials lazily, so New actively probes the daemon with IsServing and
+// returns an error if it is unreachable — otherwise a dead socket would only
+// surface on the first pull, and the agent's boot-time "containerd unavailable"
+// log would never fire.
+func New(socket string) (ImagePuller, error) {
+	client, err := containerd.New(socket, containerd.WithDefaultNamespace(ecsNamespace))
+	if err != nil {
+		return nil, fmt.Errorf("connect containerd %s: %w", socket, err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), servingProbeTimeout)
+	defer cancel()
+	serving, err := client.IsServing(ctx)
+	if err != nil || !serving {
+		_ = client.Close()
+		return nil, fmt.Errorf("containerd %s not serving: %w", socket, err)
+	}
+	return &containerdPuller{client: client}, nil
+}
+
+// Pull resolves credentials via r, then pulls and unpacks the image.
+func (p *containerdPuller) Pull(ctx context.Context, spec PullSpec, r Resolver) (Image, error) {
+	ctx = namespaces.WithNamespace(ctx, ecsNamespace)
+
+	resolver, err := newDockerResolver(ctx, spec.Ref, r)
+	if err != nil {
+		return Image{}, err
+	}
+
+	img, err := p.client.Pull(ctx, spec.Ref,
+		containerd.WithPullUnpack,
+		containerd.WithResolver(resolver),
+	)
+	if err != nil {
+		return Image{}, fmt.Errorf("pull %s: %w", spec.Ref, err)
+	}
+
+	target := img.Target()
+	return Image{
+		Ref:    spec.Ref,
+		Digest: target.Digest.String(),
+		Size:   target.Size,
+	}, nil
+}
+
+// Close releases the containerd client connection.
+func (p *containerdPuller) Close() error {
+	if p.client == nil {
+		return nil
+	}
+	return p.client.Close()
+}
+
+// newDockerResolver builds a containerd remote resolver whose per-host
+// credentials come from r (the ECR token resolver). Credentials are fetched once
+// for ref and reused for every host containerd asks about during the pull.
+func newDockerResolver(ctx context.Context, ref string, r Resolver) (remotes.Resolver, error) {
+	creds := func(string) (string, string, error) { return "", "", nil }
+	if r != nil {
+		user, pass, _, err := r.Authorize(ctx, ref)
+		if err != nil {
+			return nil, fmt.Errorf("authorize %s: %w", ref, err)
+		}
+		creds = func(string) (string, string, error) { return user, pass, nil }
+	}
+
+	authorizer := docker.NewDockerAuthorizer(docker.WithAuthCreds(creds))
+	return docker.NewResolver(docker.ResolverOptions{
+		Hosts: docker.ConfigureDefaultRegistries(docker.WithAuthorizer(authorizer)),
+	}), nil
+}

cmd/ecs-agent/runtime/runtime.go

@@ -0,0 +1,36 @@
+// Package runtime is the ecs-agent's container-runtime seam. The agent pulls
+// task images and (from Sprint 4e) runs containers through an ImagePuller; the
+// real implementation drives containerd, but the interface lets tests and the
+// scheduler-only build run against a fake.
+package runtime
+
+import "context"
+
+// PullSpec describes one image to pull.
+type PullSpec struct {
+	// Ref is the full image reference, e.g.
+	// "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:latest".
+	Ref string
+}
+
+// Image identifies a pulled image in the runtime's content store.
+type Image struct {
+	Ref    string
+	Digest string
+	Size   int64
+}
+
+// Resolver supplies registry credentials for a given image reference. The
+// ecs-agent backs this with ECR GetAuthorizationToken (see ecrresolver.go).
+type Resolver interface {
+	// Authorize returns the username, password and registry endpoint to use for
+	// ref. An empty user/pass means anonymous pull.
+	Authorize(ctx context.Context, ref string) (user, pass, endpoint string, err error)
+}
+
+// ImagePuller pulls images into the local runtime. Close releases the runtime
+// client connection.
+type ImagePuller interface {
+	Pull(ctx context.Context, spec PullSpec, r Resolver) (Image, error)
+	Close() error
+}

cmd/ecs-agent/runtime/runtime_fake.go

@@ -0,0 +1,59 @@
+package runtime
+
+import (
+	"context"
+	"fmt"
+	"sync"
+)
+
+// FakePuller is an in-memory ImagePuller for tests and the scheduler-only build.
+// It records pull requests and replays a programmed result or error.
+type FakePuller struct {
+	mu     sync.Mutex
+	Pulls  []PullSpec
+	Authzd []string
+	Result Image
+	Err    error
+	Closed bool
+	OnPull func(PullSpec) (Image, error)
+}
+
+var _ ImagePuller = (*FakePuller)(nil)
+
+// Pull records the spec, exercises the resolver, and returns the programmed
+// result. If OnPull is set it takes precedence over Result/Err.
+func (f *FakePuller) Pull(ctx context.Context, spec PullSpec, r Resolver) (Image, error) {
+	f.mu.Lock()
+	f.Pulls = append(f.Pulls, spec)
+	f.mu.Unlock()
+
+	if r != nil {
+		user, _, _, err := r.Authorize(ctx, spec.Ref)
+		if err != nil {
+			return Image{}, fmt.Errorf("authorize %s: %w", spec.Ref, err)
+		}
+		f.mu.Lock()
+		f.Authzd = append(f.Authzd, user)
+		f.mu.Unlock()
+	}
+
+	if f.OnPull != nil {
+		return f.OnPull(spec)
+	}
+	if f.Err != nil {
+		return Image{}, f.Err
+	}
+	res := f.Result
+	if res.Ref == "" {
+		res.Ref = spec.Ref
+	}
+	return res, nil
+}
+
+// Close marks the puller closed.
+func (f *FakePuller) Close() error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.Closed = true
+	return nil
+}

cmd/ecs-agent/runtime/runtime_test.go

@@ -0,0 +1,67 @@
+package runtime
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+// staticResolver is a test Resolver returning fixed creds or an error.
+type staticResolver struct {
+	user, pass, endpoint string
+	err                  error
+}
+
+func (s staticResolver) Authorize(_ context.Context, _ string) (string, string, string, error) {
+	return s.user, s.pass, s.endpoint, s.err
+}
+
+func TestFakePuller_RecordsAndDefaultsRef(t *testing.T) {
+	f := &FakePuller{}
+	img, err := f.Pull(context.Background(), PullSpec{Ref: "repo/app:1"}, staticResolver{user: "AWS"})
+	if err != nil {
+		t.Fatalf("Pull: %v", err)
+	}
+	if img.Ref != "repo/app:1" {
+		t.Errorf("Ref defaulted wrong: %q", img.Ref)
+	}
+	if len(f.Pulls) != 1 || f.Pulls[0].Ref != "repo/app:1" {
+		t.Errorf("Pulls = %+v", f.Pulls)
+	}
+	if len(f.Authzd) != 1 || f.Authzd[0] != "AWS" {
+		t.Errorf("Authzd = %+v", f.Authzd)
+	}
+}
+
+func TestFakePuller_ResolverError(t *testing.T) {
+	f := &FakePuller{}
+	_, err := f.Pull(context.Background(), PullSpec{Ref: "r"}, staticResolver{err: errors.New("boom")})
+	if err == nil {
+		t.Fatal("expected resolver error to propagate")
+	}
+}
+
+func TestFakePuller_ProgrammedError(t *testing.T) {
+	f := &FakePuller{Err: errors.New("pull failed")}
+	if _, err := f.Pull(context.Background(), PullSpec{Ref: "r"}, nil); err == nil {
+		t.Fatal("expected programmed error")
+	}
+}
+
+func TestFakePuller_OnPullOverrides(t *testing.T) {
+	want := Image{Ref: "x", Digest: "sha256:abc", Size: 42}
+	f := &FakePuller{Err: errors.New("ignored"), OnPull: func(PullSpec) (Image, error) {
+		return want, nil
+	}}
+	got, err := f.Pull(context.Background(), PullSpec{Ref: "r"}, nil)
+	if err != nil || got != want {
+		t.Fatalf("got %+v, %v; want %+v", got, err, want)
+	}
+}
+
+func TestFakePuller_Close(t *testing.T) {
+	f := &FakePuller{}
+	if err := f.Close(); err != nil || !f.Closed {
+		t.Fatalf("Close: err=%v closed=%v", err, f.Closed)
+	}
+}

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/charmbracelet/bubbles v1.0.0
 	github.com/charmbracelet/bubbletea v1.3.10
 	github.com/charmbracelet/lipgloss v1.1.0
+	github.com/containerd/containerd/v2 v2.3.2
 	github.com/go-chi/chi/v5 v5.3.0
 	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/google/uuid v1.6.0
@@ -35,6 +36,8 @@ require (
 	atomicgo.dev/cursor v0.2.0 // indirect
 	atomicgo.dev/keyboard v0.2.9 // indirect
 	atomicgo.dev/schedule v0.1.0 // indirect
+	github.com/Microsoft/go-winio v0.6.3-0.20251027160822-ad3df93bed29 // indirect
+	github.com/Microsoft/hcsshim v0.15.0-rc.1 // indirect
 	github.com/antithesishq/antithesis-sdk-go v0.7.0 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/atotto/clipboard v0.1.4 // indirect
@@ -55,13 +58,26 @@ require (
 	github.com/clipperhouse/displaywidth v0.9.0 // indirect
 	github.com/clipperhouse/stringish v0.1.1 // indirect
 	github.com/clipperhouse/uax29/v2 v2.7.0 // indirect
+	github.com/containerd/cgroups/v3 v3.1.3 // indirect
 	github.com/containerd/console v1.0.5 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/containerd/containerd/api v1.11.1 // indirect
+	github.com/containerd/continuity v0.5.0 // indirect
+	github.com/containerd/errdefs v1.0.0 // indirect
+	github.com/containerd/errdefs/pkg v0.3.0 // indirect
+	github.com/containerd/fifo v1.1.0 // indirect
+	github.com/containerd/log v0.1.0 // indirect
+	github.com/containerd/platforms v1.0.0-rc.4 // indirect
+	github.com/containerd/plugin v1.1.0 // indirect
+	github.com/containerd/ttrpc v1.2.8 // indirect
+	github.com/containerd/typeurl/v2 v2.2.3 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgraph-io/badger/v4 v4.9.2 // indirect
 	github.com/dgraph-io/ristretto/v2 v2.4.0 // indirect
+	github.com/distribution/reference v0.6.0 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect
 	github.com/fatih/color v1.19.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.13 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
@@ -70,6 +86,8 @@ require (
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.30.1 // indirect
 	github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect
 	github.com/google/flatbuffers v25.12.19+incompatible // indirect
 	github.com/google/go-tpm v0.9.8 // indirect
 	github.com/gookit/color v1.6.0 // indirect
@@ -94,9 +112,15 @@ require (
 	github.com/mattn/go-localereader v0.0.1 // indirect
 	github.com/mattn/go-runewidth v0.0.21 // indirect
 	github.com/mdlayher/packet v1.1.2 // indirect
-	github.com/mdlayher/socket v0.4.1 // indirect
+	github.com/mdlayher/socket v0.5.1 // indirect
 	github.com/minio/crc64nvme v1.1.1 // indirect
 	github.com/minio/highwayhash v1.0.4 // indirect
+	github.com/moby/locker v1.0.1 // indirect
+	github.com/moby/sys/mountinfo v0.7.2 // indirect
+	github.com/moby/sys/sequential v0.6.0 // indirect
+	github.com/moby/sys/signal v0.7.1 // indirect
+	github.com/moby/sys/user v0.4.0 // indirect
+	github.com/moby/sys/userns v0.1.0 // indirect
 	github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 // indirect
 	github.com/muesli/cancelreader v0.2.2 // indirect
 	github.com/muesli/termenv v0.16.0 // indirect
@@ -104,8 +128,12 @@ require (
 	github.com/nats-io/jwt/v2 v2.8.2 // indirect
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
-	github.com/pierrec/lz4/v4 v4.1.14 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.1 // indirect
+	github.com/opencontainers/runtime-spec v1.3.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.18 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_golang v1.23.2 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.67.5 // indirect
@@ -113,6 +141,7 @@ require (
 	github.com/quic-go/quic-go v0.60.0 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
 	github.com/sagikazarmark/locafero v0.12.0 // indirect
+	github.com/sirupsen/logrus v1.9.4 // indirect
 	github.com/spf13/afero v1.15.0 // indirect
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect
@@ -120,10 +149,12 @@ require (
 	github.com/u-root/uio v0.0.0-20230220225925-ffce2a382923 // indirect
 	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
 	go.etcd.io/bbolt v1.4.3 // indirect
+	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
-	go.opentelemetry.io/otel v1.42.0 // indirect
-	go.opentelemetry.io/otel/metric v1.42.0 // indirect
-	go.opentelemetry.io/otel/trace v1.42.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect
+	go.opentelemetry.io/otel v1.44.0 // indirect
+	go.opentelemetry.io/otel/metric v1.44.0 // indirect
+	go.opentelemetry.io/otel/trace v1.44.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/mod v0.36.0 // indirect
@@ -134,7 +165,18 @@ require (
 	golang.org/x/time v0.15.0 // indirect
 	golang.org/x/tools v0.45.0 // indirect
 	golang.org/x/vuln v1.1.4 // indirect
-	google.golang.org/protobuf v1.36.11 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260618152121-87f3d3e198d3 // indirect
+	google.golang.org/grpc v1.81.1 // indirect
+	google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect
 )
 
+// go.opencensus.io -> grpc@v1.33.2 drags in the pre-split monolithic genproto,
+// which still ships google.golang.org/genproto/googleapis/rpc/status and so
+// collides with the split genproto/googleapis/rpc module that containerd/ttrpc
+// needs ("ambiguous import"). The main module never imports the monolith, so a
+// plain require is dropped by `go mod tidy`; pin it via replace to a post-split
+// version (no rpc/status) so the fix survives tidy. Remove if grpc/opencensus
+// stop requiring the monolith.
+replace google.golang.org/genproto => google.golang.org/genproto v0.0.0-20260622175928-b703f567277d
+
 tool golang.org/x/vuln/cmd/govulncheck

internal/ecrauth/ecrauth.go

@@ -0,0 +1,119 @@
+// Package ecrauth mints ECR registry credentials by SigV4-signing
+// GetAuthorizationToken against the Spinifex AWS gateway's ECR endpoint and
+// decoding the returned AWS:<jwt> token into Basic-auth parts. It is shared by
+// the in-guest binaries that pull images from the internal registry: the EKS
+// ecr-credential-provider and the ECS ecs-agent.
+package ecrauth
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/base64"
+	"fmt"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/ecr"
+	"github.com/mulgadc/spinifex/internal/tlsconfig"
+)
+
+// defaultRegion is used when the caller supplies no region; SigV4 still needs a
+// scope and the gateway does not enforce a specific value.
+const defaultRegion = "us-east-1"
+
+// Token is the decoded result of GetAuthorizationToken: Basic-auth parts plus
+// the registry proxy host and token expiry.
+type Token struct {
+	Username  string
+	Password  string
+	ProxyHost string
+	ExpiresAt time.Time
+}
+
+// GetAuthorizationToken builds an aws-sdk ECR client pointed at the gateway and
+// calls GetAuthorizationToken with the supplied static credentials, decoding the
+// AWS:<jwt> token into Basic-auth parts. httpClient must trust the gateway CA.
+func GetAuthorizationToken(region, gatewayURL string, httpClient *http.Client, akid, secret, sessionToken string) (Token, error) {
+	if region == "" {
+		region = defaultRegion
+	}
+	sess, err := session.NewSession(&aws.Config{
+		Region:      aws.String(region),
+		Endpoint:    aws.String(gatewayURL),
+		Credentials: credentials.NewStaticCredentials(akid, secret, sessionToken),
+		HTTPClient:  httpClient,
+		DisableSSL:  aws.Bool(false),
+	})
+	if err != nil {
+		return Token{}, fmt.Errorf("new session: %w", err)
+	}
+
+	out, err := ecr.New(sess).GetAuthorizationToken(&ecr.GetAuthorizationTokenInput{})
+	if err != nil {
+		return Token{}, fmt.Errorf("GetAuthorizationToken: %w", err)
+	}
+	if len(out.AuthorizationData) == 0 || out.AuthorizationData[0].AuthorizationToken == nil {
+		return Token{}, fmt.Errorf("GetAuthorizationToken returned no authorization data")
+	}
+	data := out.AuthorizationData[0]
+
+	raw, err := base64.StdEncoding.DecodeString(aws.StringValue(data.AuthorizationToken))
+	if err != nil {
+		return Token{}, fmt.Errorf("decode authorization token: %w", err)
+	}
+	user, pass, ok := strings.Cut(string(raw), ":")
+	if !ok {
+		return Token{}, fmt.Errorf("authorization token not in user:password form")
+	}
+
+	tok := Token{
+		Username:  user,
+		Password:  pass,
+		ProxyHost: HostFromImage(aws.StringValue(data.ProxyEndpoint)),
+	}
+	if data.ExpiresAt != nil {
+		tok.ExpiresAt = aws.TimeValue(data.ExpiresAt)
+	}
+	return tok, nil
+}
+
+// GatewayHTTPClient builds an HTTP client trusting the gateway CA at caPath.
+// When caPath is empty it relies on the system trust store.
+func GatewayHTTPClient(caPath string) (*http.Client, error) {
+	tlsCfg := &tls.Config{
+		MinVersion:       tls.VersionTLS13,
+		CurvePreferences: tlsconfig.Curves,
+	}
+	if caPath != "" {
+		pem, err := os.ReadFile(caPath)
+		if err != nil {
+			return nil, fmt.Errorf("read gateway CA %q: %w", caPath, err)
+		}
+		pool := x509.NewCertPool()
+		if !pool.AppendCertsFromPEM(pem) {
+			return nil, fmt.Errorf("gateway CA %q has no usable certificates", caPath)
+		}
+		tlsCfg.RootCAs = pool
+	}
+	return &http.Client{
+		Timeout:   15 * time.Second,
+		Transport: &http.Transport{TLSClientConfig: tlsCfg, MaxIdleConns: 2, IdleConnTimeout: 30 * time.Second},
+	}, nil
+}
+
+// HostFromImage returns the host[:port] portion of an image ref or URL. It strips
+// any scheme and trims at the first path separator.
+func HostFromImage(image string) string {
+	if i := strings.Index(image, "://"); i >= 0 {
+		image = image[i+3:]
+	}
+	if i := strings.IndexByte(image, '/'); i >= 0 {
+		image = image[:i]
+	}
+	return image
+}

internal/ecrauth/ecrauth_test.go

@@ -0,0 +1,107 @@
+package ecrauth
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestHostFromImage(t *testing.T) {
+	cases := map[string]string{
+		"acct.dkr.ecr.us-east-1.spinifex.internal/demo:latest": "acct.dkr.ecr.us-east-1.spinifex.internal",
+		"https://acct.dkr.ecr.us-east-1.spinifex.internal":     "acct.dkr.ecr.us-east-1.spinifex.internal",
+		"host:9999/repo:tag": "host:9999",
+		"plainhost":          "plainhost",
+	}
+	for in, want := range cases {
+		if got := HostFromImage(in); got != want {
+			t.Errorf("HostFromImage(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+// gatewayStub serves the ECR GetAuthorizationToken X-Amz-Target dispatch,
+// returning a base64 AWS:<jwt> token like the real gateway.
+func gatewayStub(t *testing.T, token, proxy string, status int) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.Contains(r.Header.Get("X-Amz-Target"), "GetAuthorizationToken") {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		if status != http.StatusOK {
+			w.WriteHeader(status)
+			return
+		}
+		w.Header().Set("Content-Type", "application/x-amz-json-1.1")
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"authorizationData": []map[string]any{{
+				"authorizationToken": base64.StdEncoding.EncodeToString([]byte(token)),
+				"proxyEndpoint":      proxy,
+			}},
+		})
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func clientTrusting(srv *httptest.Server) *http.Client {
+	pool := x509.NewCertPool()
+	pool.AddCert(srv.Certificate())
+	return &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}}}
+}
+
+func TestGetAuthorizationToken_OK(t *testing.T) {
+	srv := gatewayStub(t, "AWS:testjwt", "https://123456789012.dkr.ecr.us-east-1.spinifex.internal", http.StatusOK)
+
+	tok, err := GetAuthorizationToken("us-east-1", srv.URL, clientTrusting(srv), "AKIA", "secret", "session")
+	if err != nil {
+		t.Fatalf("GetAuthorizationToken: %v", err)
+	}
+	if tok.Username != "AWS" || tok.Password != "testjwt" {
+		t.Errorf("got %q:%q, want AWS:testjwt", tok.Username, tok.Password)
+	}
+	if tok.ProxyHost != "123456789012.dkr.ecr.us-east-1.spinifex.internal" {
+		t.Errorf("ProxyHost = %q", tok.ProxyHost)
+	}
+}
+
+func TestGetAuthorizationToken_BadTokenForm(t *testing.T) {
+	srv := gatewayStub(t, "no-colon-token", "https://x", http.StatusOK)
+	if _, err := GetAuthorizationToken("us-east-1", srv.URL, clientTrusting(srv), "AKIA", "secret", ""); err == nil {
+		t.Fatal("expected error for token not in user:password form")
+	}
+}
+
+func TestGetAuthorizationToken_ServerError(t *testing.T) {
+	srv := gatewayStub(t, "", "", http.StatusInternalServerError)
+	if _, err := GetAuthorizationToken("us-east-1", srv.URL, clientTrusting(srv), "AKIA", "secret", ""); err == nil {
+		t.Fatal("expected error on gateway 500")
+	}
+}
+
+func TestGatewayHTTPClient(t *testing.T) {
+	// Empty CA path: system trust store, no error.
+	if _, err := GatewayHTTPClient(""); err != nil {
+		t.Fatalf("empty CA path: %v", err)
+	}
+	// Missing file: error.
+	if _, err := GatewayHTTPClient(filepath.Join(t.TempDir(), "nope.pem")); err == nil {
+		t.Fatal("expected error for missing CA file")
+	}
+	// Garbage PEM: error.
+	bad := filepath.Join(t.TempDir(), "bad.pem")
+	if err := os.WriteFile(bad, []byte("not a cert"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := GatewayHTTPClient(bad); err == nil {
+		t.Fatal("expected error for CA file with no certificates")
+	}
+}

internal/imdscreds/imdscreds.go

@@ -0,0 +1,118 @@
+// Package imdscreds fetches AWS instance-role credentials from the in-guest
+// IMDS endpoint (169.254.169.254) using IMDSv2. It is shared by the binaries
+// that run inside a Spinifex guest and need the node/instance role credentials:
+// the EKS ecr-credential-provider and the ECS ecs-agent.
+package imdscreds
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// imdsTokenTTL is the IMDSv2 session-token lifetime requested on the PUT.
+const imdsTokenTTL = "21600"
+
+// Credentials is a resolved instance-role credential set with parsed expiry.
+type Credentials struct {
+	AccessKeyID     string
+	SecretAccessKey string
+	SessionToken    string
+	Expiration      time.Time
+}
+
+// imdsCreds mirrors the IMDS security-credentials JSON shape.
+type imdsCreds struct {
+	Code            string `json:"Code"`
+	AccessKeyID     string `json:"AccessKeyId"`
+	SecretAccessKey string `json:"SecretAccessKey"`
+	Token           string `json:"Token"`
+	Expiration      string `json:"Expiration"`
+}
+
+// Fetch resolves the instance role name then its credentials via IMDSv2
+// (PUT token, then GET with the token header). base is the IMDS root, e.g.
+// http://169.254.169.254/latest.
+func Fetch(client *http.Client, base string) (Credentials, error) {
+	base = strings.TrimRight(base, "/")
+	token := v2Token(client, base)
+
+	roleName, err := get(client, base+"/meta-data/iam/security-credentials/", token)
+	if err != nil {
+		return Credentials{}, fmt.Errorf("fetch role name: %w", err)
+	}
+	roleName = strings.TrimSpace(roleName)
+	if roleName == "" {
+		return Credentials{}, fmt.Errorf("IMDS returned empty role name")
+	}
+
+	body, err := get(client, base+"/meta-data/iam/security-credentials/"+roleName, token)
+	if err != nil {
+		return Credentials{}, fmt.Errorf("fetch role credentials: %w", err)
+	}
+	var raw imdsCreds
+	if err := json.Unmarshal([]byte(body), &raw); err != nil {
+		return Credentials{}, fmt.Errorf("decode role credentials: %w", err)
+	}
+	if raw.AccessKeyID == "" || raw.SecretAccessKey == "" {
+		return Credentials{}, fmt.Errorf("IMDS credentials missing access/secret key")
+	}
+
+	creds := Credentials{
+		AccessKeyID:     raw.AccessKeyID,
+		SecretAccessKey: raw.SecretAccessKey,
+		SessionToken:    raw.Token,
+	}
+	if raw.Expiration != "" {
+		// IMDS emits RFC3339; a parse failure leaves a zero expiry (treated as
+		// unknown by callers) rather than failing the whole fetch.
+		if exp, perr := time.Parse(time.RFC3339, raw.Expiration); perr == nil {
+			creds.Expiration = exp
+		}
+	}
+	return creds, nil
+}
+
+// v2Token requests an IMDSv2 session token; returns "" if the service does not
+// enforce v2 (tokenless v1 then applies).
+func v2Token(client *http.Client, base string) string {
+	req, err := http.NewRequest(http.MethodPut, base+"/api/token", nil)
+	if err != nil {
+		return ""
+	}
+	req.Header.Set("X-Aws-Ec2-Metadata-Token-Ttl-Seconds", imdsTokenTTL)
+	resp, err := client.Do(req)
+	if err != nil {
+		return ""
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return ""
+	}
+	body, _ := io.ReadAll(resp.Body)
+	return strings.TrimSpace(string(body))
+}
+
+// get issues an IMDS GET, attaching the IMDSv2 token header when present.
+func get(client *http.Client, url, token string) (string, error) {
+	req, err := http.NewRequest(http.MethodGet, url, nil)
+	if err != nil {
+		return "", err
+	}
+	if token != "" {
+		req.Header.Set("X-Aws-Ec2-Metadata-Token", token)
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return "", fmt.Errorf("IMDS %s returned %d: %s", url, resp.StatusCode, strings.TrimSpace(string(body)))
+	}
+	return string(body), nil
+}

internal/imdscreds/imdscreds_test.go

@@ -0,0 +1,80 @@
+package imdscreds
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+// imdsStub serves the IMDSv2 token + role + credentials endpoints.
+func imdsStub(t *testing.T, creds map[string]string, enforceV2 bool) *httptest.Server {
+	t.Helper()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/token", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPut {
+			w.WriteHeader(http.StatusMethodNotAllowed)
+			return
+		}
+		_, _ = w.Write([]byte("v2-token"))
+	})
+	mux.HandleFunc("/meta-data/iam/security-credentials/", func(w http.ResponseWriter, r *http.Request) {
+		if enforceV2 && r.Header.Get("X-Aws-Ec2-Metadata-Token") != "v2-token" {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		if r.URL.Path == "/meta-data/iam/security-credentials/" {
+			_, _ = w.Write([]byte("node-role"))
+			return
+		}
+		_ = json.NewEncoder(w).Encode(creds)
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func TestFetch_OK(t *testing.T) {
+	exp := time.Now().Add(6 * time.Hour).UTC().Round(time.Second)
+	srv := imdsStub(t, map[string]string{
+		"Code":            "Success",
+		"AccessKeyId":     "AKIATEST",
+		"SecretAccessKey": "secrettest",
+		"Token":           "sessiontoken",
+		"Expiration":      exp.Format(time.RFC3339),
+	}, true)
+
+	got, err := Fetch(srv.Client(), srv.URL)
+	if err != nil {
+		t.Fatalf("Fetch: %v", err)
+	}
+	if got.AccessKeyID != "AKIATEST" || got.SecretAccessKey != "secrettest" || got.SessionToken != "sessiontoken" {
+		t.Errorf("creds mismatch: %+v", got)
+	}
+	if !got.Expiration.Equal(exp) {
+		t.Errorf("Expiration = %v, want %v", got.Expiration, exp)
+	}
+}
+
+func TestFetch_MissingKeys(t *testing.T) {
+	srv := imdsStub(t, map[string]string{"Code": "Success"}, false)
+	if _, err := Fetch(srv.Client(), srv.URL); err == nil {
+		t.Fatal("expected error when access/secret key absent")
+	}
+}
+
+func TestFetch_BadExpirationIsZero(t *testing.T) {
+	srv := imdsStub(t, map[string]string{
+		"AccessKeyId":     "AKIA",
+		"SecretAccessKey": "secret",
+		"Expiration":      "not-a-time",
+	}, false)
+	got, err := Fetch(srv.Client(), srv.URL)
+	if err != nil {
+		t.Fatalf("Fetch: %v", err)
+	}
+	if !got.Expiration.IsZero() {
+		t.Errorf("Expiration = %v, want zero on parse failure", got.Expiration)
+	}
+}

spinifex/handlers/NATS_HANDLER_NAMING.md

@@ -266,8 +266,12 @@ ecs.bus.<accountID>.<clusterName>.instance-register.<instanceID>  # agent → sc
 ecs.bus.<accountID>.<clusterName>.service-reconcile               # scheduler-internal tick
 ```
 
-This layer lands with the scheduler + agent bodies. The Phase 5 per-AZ NATS
-cutover inserts `{azID}` after `ecs.` → `ecs.<azID>.bus.<accountID>.<clusterName>.*`.
+Sprint 4c lands the agent-side publishers for `instance-register` (at boot) and
+`instance-heartbeat` (every 30s); the scheduler-side subscribers, `assign`, and
+`task-state` land in Sprint 4e. Subject builders and wire payloads live in
+`spinifex/handlers/ecs/bus` so both the agent and the scheduler share one source
+of truth. The Phase 5 per-AZ NATS cutover inserts `{azID}` after `ecs.` →
+`ecs.<azID>.bus.<accountID>.<clusterName>.*`.
 
 ### Cross-service calls
 

spinifex/handlers/ecs/bus/messages.go

@@ -0,0 +1,41 @@
+package bus
+
+import "time"
+
+// InstanceCapacity is a container instance's total registrable resources.
+type InstanceCapacity struct {
+	CPU       int `json:"cpu"`       // total vCPU units (1 vCPU = 1024)
+	MemoryMiB int `json:"memoryMiB"` // total memory in MiB
+}
+
+// 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).
+type RegisterInstance struct {
+	AccountID    string           `json:"accountId"`
+	ClusterName  string           `json:"clusterName"`
+	InstanceID   string           `json:"instanceId"`
+	AZ           string           `json:"availabilityZone"`
+	Hostname     string           `json:"hostname"`
+	Capacity     InstanceCapacity `json:"capacity"`
+	AgentVersion string           `json:"agentVersion"`
+	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).
+type Heartbeat struct {
+	AccountID    string    `json:"accountId"`
+	ClusterName  string    `json:"clusterName"`
+	InstanceID   string    `json:"instanceId"`
+	Status       string    `json:"status"` // ACTIVE | DRAINING
+	RunningTasks int       `json:"runningTasks"`
+	SentAt       time.Time `json:"sentAt"`
+}
+
+// Instance status values reported in Heartbeat.Status.
+const (
+	StatusActive   = "ACTIVE"
+	StatusDraining = "DRAINING"
+)

spinifex/handlers/ecs/bus/subjects.go

@@ -0,0 +1,43 @@
+// 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.
+//
+// Cluster identity is "<accountID>.<clusterName>" (matching the KV layout), not a
+// UUID. The Phase 5 per-AZ NATS cutover inserts "{azID}" after "ecs." — keep
+// these builders the one place that knows the shape so that change stays local.
+package bus
+
+import "fmt"
+
+// Prefix returns "ecs.bus.<accountID>.<clusterName>".
+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).
+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).
+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.
+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.
+func TaskStateSubject(accountID, clusterName, taskID string) string {
+	return fmt.Sprintf("%s.task-state.%s", Prefix(accountID, clusterName), taskID)
+}
+
+// ServiceReconcileSubject is the scheduler-internal reconcile tick.
+func ServiceReconcileSubject(accountID, clusterName string) string {
+	return fmt.Sprintf("%s.service-reconcile", Prefix(accountID, clusterName))
+}

spinifex/handlers/ecs/bus/subjects_test.go

@@ -0,0 +1,25 @@
+package bus
+
+import "testing"
+
+func TestSubjectBuilders(t *testing.T) {
+	const (
+		acct    = "123456789012"
+		cluster = "default"
+		inst    = "i-abc123"
+		task    = "t-deadbeef"
+	)
+	cases := map[string]struct{ got, want string }{
+		"prefix":    {Prefix(acct, cluster), "ecs.bus.123456789012.default"},
+		"register":  {RegisterSubject(acct, cluster, inst), "ecs.bus.123456789012.default.instance-register.i-abc123"},
+		"heartbeat": {HeartbeatSubject(acct, cluster, inst), "ecs.bus.123456789012.default.instance-heartbeat.i-abc123"},
+		"assign":    {AssignSubject(acct, cluster, inst), "ecs.bus.123456789012.default.assign.i-abc123"},
+		"taskstate": {TaskStateSubject(acct, cluster, task), "ecs.bus.123456789012.default.task-state.t-deadbeef"},
+		"reconcile": {ServiceReconcileSubject(acct, cluster), "ecs.bus.123456789012.default.service-reconcile"},
+	}
+	for name, c := range cases {
+		if c.got != c.want {
+			t.Errorf("%s = %q, want %q", name, c.got, c.want)
+		}
+	}
+}