refactor(ecr): delete the in-memory MetaStore and run its tests against JetStream KV (#1474)

* feat(admin): non-finalizing DeleteAccount exercises the quiesced EC2 platform kinds

A non-finalizing DeleteAccount quiesces the account, records the S9 evaluation, cleans only the key-pair, placement-group and launch-template kinds it recorded as quiesced, then re-reads the owner inventory and accounts for what remains. The job ends EXERCISED with the account still TERMINATING; a later ordinary delete takes the same operation over.

Only an exercise records a quiesced scope, so ordinary deletes still record none. The cleanup capability gains a kinds selector, awsgw binds the in-process owners, and the instance reference reader fails closed.

* refactor(admin): name the takeover of a finished exercise exercise_finished

cmd/lambda-agent/package_testlink.go

@@ -0,0 +1,276 @@
+//go:build lambdatestlink
+
+package main
+
+import (
+	"archive/zip"
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"os/exec"
+	"path"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// Dev-only code delivery for the test path: the host attaches the ZIP as a
+// read-only disk and passes its exact size and SHA-256 through fw_cfg. This
+// is not the production delivery, and these bounds are not AWS limits.
+const (
+	entryPackageSize   = "test-package-size"
+	entryPackageSHA256 = "test-package-sha256"
+
+	devMaxZipBytes    = 16 << 20
+	devMaxUnzipBytes  = 64 << 20
+	devMaxEntries     = 1024
+	packageDevice     = "/dev/vda"
+	taskTmpfsSizeOpts = "mode=0755,size=72m"
+)
+
+type packageSpec struct {
+	size   int64
+	sha256 [sha256.Size]byte
+}
+
+type unpackLimits struct {
+	entries int
+	bytes   int64
+}
+
+var devLimits = unpackLimits{entries: devMaxEntries, bytes: devMaxUnzipBytes}
+
+// readPackageSpec reads the expected size and SHA-256 from fw_cfg; it must
+// run before fw_cfg is sealed.
+func readPackageSpec(fwCfgDir string) (packageSpec, error) {
+	var spec packageSpec
+	read := func(name string) (string, error) {
+		b, err := os.ReadFile(filepath.Join(fwCfgDir, name, "raw"))
+		if err != nil {
+			return "", fmt.Errorf("read %s: %w", name, err)
+		}
+		return strings.TrimSpace(string(b)), nil
+	}
+	s, err := read(entryPackageSize)
+	if err != nil {
+		return spec, err
+	}
+	spec.size, err = strconv.ParseInt(s, 10, 64)
+	if err != nil || spec.size <= 0 || spec.size > devMaxZipBytes {
+		return spec, fmt.Errorf("package size %q outside 1..%d bytes", s, devMaxZipBytes)
+	}
+	h, err := read(entryPackageSHA256)
+	if err != nil {
+		return spec, err
+	}
+	b, err := hex.DecodeString(h)
+	if err != nil || len(b) != sha256.Size {
+		return spec, errors.New("package sha256 is not 64 hex digits")
+	}
+	copy(spec.sha256[:], b)
+	return spec, nil
+}
+
+// taskPackage puts the package under dir: a fresh nosuid,nodev tmpfs, made
+// read-only once the files are in place.
+type taskPackage struct {
+	device     string
+	dir        string
+	loadDriver func() error
+	mountDir   func(dir string) error
+	sealDir    func(dir string) error
+}
+
+func devTaskPackage() taskPackage {
+	return taskPackage{device: packageDevice, dir: taskRoot, loadDriver: loadBlockDriver, mountDir: mountTaskDir, sealDir: sealTaskDir}
+}
+
+func (p taskPackage) install(spec packageSpec) error {
+	if err := p.loadDriver(); err != nil {
+		return fmt.Errorf("load block driver: %w", err)
+	}
+	dev, err := os.Open(p.device)
+	if err != nil {
+		return fmt.Errorf("open package disk: %w", err)
+	}
+	defer dev.Close()
+	if err := p.mountDir(p.dir); err != nil {
+		return fmt.Errorf("mount %s: %w", p.dir, err)
+	}
+	if err := unpack(dev, spec, p.dir, devLimits); err != nil {
+		return err
+	}
+	if err := p.sealDir(p.dir); err != nil {
+		return fmt.Errorf("make %s read-only: %w", p.dir, err)
+	}
+	return nil
+}
+
+// unpack reads exactly spec.size bytes once, checks their SHA-256, then
+// validates every entry before writing any of them under dir.
+func unpack(r io.ReaderAt, spec packageSpec, dir string, lim unpackLimits) error {
+	buf := make([]byte, spec.size)
+	if _, err := io.ReadFull(io.NewSectionReader(r, 0, spec.size), buf); err != nil {
+		return fmt.Errorf("read package: %w", err)
+	}
+	if sha256.Sum256(buf) != spec.sha256 {
+		return errors.New("package sha256 mismatch")
+	}
+	zr, err := zip.NewReader(bytes.NewReader(buf), spec.size)
+	if err != nil {
+		return fmt.Errorf("open package: %w", err)
+	}
+	if len(zr.File) > lim.entries {
+		return fmt.Errorf("package has %d entries, limit %d", len(zr.File), lim.entries)
+	}
+	names := make([]string, len(zr.File))
+	var declared int64
+	for i, f := range zr.File {
+		name, err := entryName(f.Name)
+		if err != nil {
+			return err
+		}
+		names[i] = name
+		switch mode := f.Mode(); {
+		case mode.IsDir():
+		case mode.IsRegular():
+			// Compared against what remains, so a huge size cannot wrap the sum.
+			if f.UncompressedSize64 > uint64(lim.bytes-declared) {
+				return fmt.Errorf("%q: unpacked size over the %d-byte limit", f.Name, lim.bytes)
+			}
+			declared += int64(f.UncompressedSize64)
+		default:
+			return fmt.Errorf("%q: %s entries are refused", f.Name, mode.Type())
+		}
+	}
+
+	root, err := os.OpenRoot(dir)
+	if err != nil {
+		return err
+	}
+	defer root.Close()
+	var written int64
+	for i, f := range zr.File {
+		n, err := extract(root, names[i], f, lim.bytes-written)
+		written += n
+		if err != nil {
+			return fmt.Errorf("%q: %w", f.Name, err)
+		}
+	}
+	return nil
+}
+
+// extract writes one entry, refusing it if its actual bytes exceed budget
+// whatever its header claims.
+func extract(root *os.Root, name string, f *zip.File, budget int64) (int64, error) {
+	if f.Mode().IsDir() {
+		return 0, mkdirs(root, name)
+	}
+	if d := path.Dir(name); d != "." {
+		if err := mkdirs(root, d); err != nil {
+			return 0, err
+		}
+	}
+	perm := os.FileMode(0o644)
+	if f.Mode()&0o111 != 0 {
+		perm = 0o755
+	}
+	in, err := f.Open()
+	if err != nil {
+		return 0, err
+	}
+	defer in.Close()
+	out, err := root.OpenFile(name, os.O_CREATE|os.O_EXCL|os.O_WRONLY, perm)
+	if err != nil {
+		return 0, err
+	}
+	// The umask applies at create; fchmod does not, so the mode is exact.
+	if err := out.Chmod(perm); err != nil {
+		out.Close()
+		return 0, err
+	}
+	n, err := io.Copy(out, io.LimitReader(in, budget+1))
+	if cerr := out.Close(); err == nil {
+		err = cerr
+	}
+	if err == nil && n > budget {
+		err = errors.New("unpacked bytes over the limit")
+	}
+	return n, err
+}
+
+// mkdirs creates each component of dir as mode 0755 whatever the umask. An
+// existing component must be a real directory, never a symlink.
+func mkdirs(root *os.Root, dir string) error {
+	cur := ""
+	for _, part := range strings.Split(dir, "/") {
+		cur = path.Join(cur, part)
+		if err := root.Mkdir(cur, 0o755); err != nil {
+			if !errors.Is(err, fs.ErrExist) {
+				return err
+			}
+			fi, err := root.Lstat(cur)
+			if err != nil {
+				return err
+			}
+			if !fi.IsDir() {
+				return fmt.Errorf("%q exists and is not a directory", cur)
+			}
+		}
+		if err := root.Chmod(cur, 0o755); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// entryName returns a clean relative path, refusing anything that could name
+// a location outside the task root.
+func entryName(name string) (string, error) {
+	if name == "" || strings.ContainsAny(name, "\\\x00") || path.IsAbs(name) {
+		return "", fmt.Errorf("%q: entry name refused", name)
+	}
+	c := path.Clean(name)
+	if c == "." || c == ".." || strings.HasPrefix(c, "../") {
+		return "", fmt.Errorf("%q: entry name refused", name)
+	}
+	return c, nil
+}
+
+// loadBlockDriver loads virtio_blk, which the image ships but init does not
+// load, then waits briefly for the package disk to appear.
+func loadBlockDriver() error {
+	if err := exec.Command("/sbin/modprobe", "virtio_blk").Run(); err != nil {
+		var uts unix.Utsname
+		if uerr := unix.Uname(&uts); uerr != nil {
+			return uerr
+		}
+		ko := "/lib/modules/" + unix.ByteSliceToString(uts.Release[:]) + "/kernel/drivers/block/virtio_blk.ko"
+		if err := exec.Command("/sbin/insmod", ko).Run(); err != nil {
+			return fmt.Errorf("modprobe and insmod virtio_blk failed: %w", err)
+		}
+	}
+	for range 50 {
+		if _, err := os.Stat(packageDevice); err == nil {
+			return nil
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	return fmt.Errorf("%s did not appear", packageDevice)
+}
+
+func mountTaskDir(dir string) error {
+	return unix.Mount("tmpfs", dir, "tmpfs", unix.MS_NOSUID|unix.MS_NODEV, taskTmpfsSizeOpts)
+}
+
+func sealTaskDir(dir string) error {
+	return unix.Mount("", dir, "", unix.MS_REMOUNT|unix.MS_RDONLY|unix.MS_NOSUID|unix.MS_NODEV, "")
+}

cmd/lambda-agent/package_testlink_test.go

@@ -0,0 +1,209 @@
+//go:build lambdatestlink
+
+package main
+
+import (
+	"archive/zip"
+	"bytes"
+	"crypto/sha256"
+	"hash/crc32"
+	"io/fs"
+	"math"
+	"os"
+	"path/filepath"
+	"sort"
+	"syscall"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// testZip builds a ZIP; an entry named bootstrap is marked executable.
+func testZip(t *testing.T, files map[string]string) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	zw := zip.NewWriter(&buf)
+	names := make([]string, 0, len(files))
+	for n := range files {
+		names = append(names, n)
+	}
+	sort.Strings(names)
+	for _, n := range names {
+		h := &zip.FileHeader{Name: n, Method: zip.Deflate}
+		h.SetMode(0o644)
+		if n == "bootstrap" {
+			h.SetMode(0o755)
+		}
+		w, err := zw.CreateHeader(h)
+		require.NoError(t, err)
+		_, err = w.Write([]byte(files[n]))
+		require.NoError(t, err)
+	}
+	require.NoError(t, zw.Close())
+	return buf.Bytes()
+}
+
+// rawZip writes entries with headers set by the caller, so a test can claim
+// sizes or modes the data does not have.
+func rawZip(t *testing.T, entries ...*zip.FileHeader) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	zw := zip.NewWriter(&buf)
+	for _, h := range entries {
+		w, err := zw.CreateRaw(h)
+		require.NoError(t, err)
+		_, err = w.Write([]byte("x"))
+		require.NoError(t, err)
+	}
+	require.NoError(t, zw.Close())
+	return buf.Bytes()
+}
+
+func specFor(b []byte) packageSpec {
+	return packageSpec{size: int64(len(b)), sha256: sha256.Sum256(b)}
+}
+
+// Not parallel: the umask is process-wide, so the test sets a restrictive one
+// and restores it.
+func TestUnpackInstallsFilesWithFixedModes(t *testing.T) {
+	old := syscall.Umask(0o077)
+	t.Cleanup(func() { syscall.Umask(old) })
+	b := testZip(t, map[string]string{"bootstrap": "#!/bin/sh\n", "lib/data.txt": "data"})
+	dir := t.TempDir()
+	require.NoError(t, unpack(bytes.NewReader(b), specFor(b), dir, devLimits))
+
+	pfi, err := os.Stat(filepath.Join(dir, "lib"))
+	require.NoError(t, err)
+	assert.Equal(t, fs.FileMode(0o755), pfi.Mode().Perm(), "implicit parent dir")
+
+	fi, err := os.Stat(filepath.Join(dir, "bootstrap"))
+	require.NoError(t, err)
+	assert.Equal(t, fs.FileMode(0o755), fi.Mode().Perm(), "executable")
+	fi, err = os.Stat(filepath.Join(dir, "lib", "data.txt"))
+	require.NoError(t, err)
+	assert.Equal(t, fs.FileMode(0o644), fi.Mode().Perm())
+	got, err := os.ReadFile(filepath.Join(dir, "lib", "data.txt"))
+	require.NoError(t, err)
+	assert.Equal(t, "data", string(got))
+}
+
+func TestUnpackRefusesBeforeWritingAnything(t *testing.T) {
+	symlink := &zip.FileHeader{Name: "link", Method: zip.Store, CRC32: crc32.ChecksumIEEE([]byte("x")), CompressedSize64: 1, UncompressedSize64: 1}
+	symlink.SetMode(fs.ModeSymlink | 0o777)
+	fifo := &zip.FileHeader{Name: "pipe", Method: zip.Store, CRC32: crc32.ChecksumIEEE([]byte("x")), CompressedSize64: 1, UncompressedSize64: 1}
+	fifo.SetMode(fs.ModeNamedPipe | 0o644)
+	// The second size would wrap a plain running sum back under the limit.
+	big := func(name string, n uint64) *zip.FileHeader {
+		return &zip.FileHeader{Name: name, Method: zip.Store, CRC32: crc32.ChecksumIEEE([]byte("x")), CompressedSize64: 1, UncompressedSize64: n}
+	}
+	cases := map[string]struct {
+		zip  []byte
+		want string
+	}{
+		"parent traversal": {testZip(t, map[string]string{"a": "ok", "../escape": "x"}), "entry name refused"},
+		"nested traversal": {testZip(t, map[string]string{"a": "ok", "lib/../../escape": "x"}), "entry name refused"},
+		"absolute path":    {testZip(t, map[string]string{"a": "ok", "/etc/escape": "x"}), "entry name refused"},
+		"backslash":        {testZip(t, map[string]string{"a": "ok", `..\escape`: "x"}), "entry name refused"},
+		"symlink":          {rawZip(t, big("a", 1), symlink), "entries are refused"},
+		"named pipe":       {rawZip(t, big("a", 1), fifo), "entries are refused"},
+		"declared size":    {rawZip(t, big("a", devMaxUnzipBytes+1)), "over the"},
+		"wrapping sizes":   {rawZip(t, big("a", devMaxUnzipBytes), big("b", math.MaxUint64)), "over the"},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			dir := t.TempDir()
+			err := unpack(bytes.NewReader(tc.zip), specFor(tc.zip), dir, devLimits)
+			assert.ErrorContains(t, err, tc.want)
+			left, rerr := os.ReadDir(dir)
+			require.NoError(t, rerr)
+			assert.Empty(t, left, "files written before the refusal")
+		})
+	}
+}
+
+func TestUnpackRefusesTooManyEntries(t *testing.T) {
+	b := testZip(t, map[string]string{"a": "1", "b": "2", "c": "3"})
+	err := unpack(bytes.NewReader(b), specFor(b), t.TempDir(), unpackLimits{entries: 2, bytes: devMaxUnzipBytes})
+	assert.ErrorContains(t, err, "limit 2")
+}
+
+// The header claims a size within the limit, but the data is longer: the
+// actual bytes are refused, whatever the header says.
+func TestUnpackRefusesBytesBeyondTheLimit(t *testing.T) {
+	data := bytes.Repeat([]byte("a"), 64)
+	var buf bytes.Buffer
+	zw := zip.NewWriter(&buf)
+	h := &zip.FileHeader{Name: "a", Method: zip.Store, CRC32: crc32.ChecksumIEEE(data), CompressedSize64: uint64(len(data)), UncompressedSize64: 8}
+	w, err := zw.CreateRaw(h)
+	require.NoError(t, err)
+	_, err = w.Write(data)
+	require.NoError(t, err)
+	require.NoError(t, zw.Close())
+	b := buf.Bytes()
+
+	dir := t.TempDir()
+	err = unpack(bytes.NewReader(b), specFor(b), dir, unpackLimits{entries: 4, bytes: 16})
+	require.Error(t, err)
+	fi, serr := os.Stat(filepath.Join(dir, "a"))
+	if serr == nil {
+		assert.LessOrEqual(t, fi.Size(), int64(17))
+	}
+}
+
+func TestUnpackChecksSizeAndHash(t *testing.T) {
+	b := testZip(t, map[string]string{"bootstrap": "x"})
+	wrongHash := specFor(b)
+	wrongHash.sha256[0] ^= 1
+	assert.ErrorContains(t, unpack(bytes.NewReader(b), wrongHash, t.TempDir(), devLimits), "sha256 mismatch")
+
+	longer := specFor(b)
+	longer.size++
+	assert.ErrorContains(t, unpack(bytes.NewReader(b), longer, t.TempDir(), devLimits), "read package")
+}
+
+func TestReadPackageSpecBounds(t *testing.T) {
+	sum := "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+	cases := map[string]struct {
+		size, hash string
+		want       string
+	}{
+		"valid":     {"10", sum, ""},
+		"zero":      {"0", sum, "outside"},
+		"too large": {"16777217", sum, "outside"},
+		"not hex":   {"10", "zz", "64 hex digits"},
+		"short":     {"10", sum[:62], "64 hex digits"},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			dir := t.TempDir()
+			for n, v := range map[string]string{entryPackageSize: tc.size, entryPackageSHA256: tc.hash} {
+				require.NoError(t, os.MkdirAll(filepath.Join(dir, n), 0o700))
+				require.NoError(t, os.WriteFile(filepath.Join(dir, n, "raw"), []byte(v), 0o600))
+			}
+			spec, err := readPackageSpec(dir)
+			if tc.want == "" {
+				require.NoError(t, err)
+				assert.Equal(t, int64(10), spec.size)
+				return
+			}
+			assert.ErrorContains(t, err, tc.want)
+		})
+	}
+}
+
+// A pre-existing symlink where a directory belongs is refused, not followed
+// or chmodded.
+func TestMkdirsRefusesSymlinkComponent(t *testing.T) {
+	dir := t.TempDir()
+	outside := t.TempDir()
+	require.NoError(t, os.Chmod(outside, 0o700))
+	require.NoError(t, os.Symlink(outside, filepath.Join(dir, "lib")))
+	root, err := os.OpenRoot(dir)
+	require.NoError(t, err)
+	defer root.Close()
+	assert.ErrorContains(t, mkdirs(root, "lib/sub"), "not a directory")
+	fi, err := os.Stat(outside)
+	require.NoError(t, err)
+	assert.Equal(t, fs.FileMode(0o700), fi.Mode().Perm(), "symlink target was chmodded")
+}

cmd/lambda-agent/start_testlink.go

@@ -30,6 +30,7 @@ var handler = flag.String("handler", os.Getenv("_HANDLER"), "function handler pa
 // connection; the carrier is not the guest protocol.
 type testlink struct {
 	startup
+	pkg            taskPackage
 	runtimeAddr    string
 	startBootstrap func(ctx context.Context, env []string) (*exec.Cmd, error)
 }
@@ -37,6 +38,7 @@ type testlink struct {
 func start(ctx context.Context) int {
 	t := testlink{
 		startup:     defaultStartup(),
+		pkg:         devTaskPackage(),
 		runtimeAddr: runtimeAddr,
 		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
 			return agent.StartBootstrap(ctx, taskRoot+"/bootstrap", env, os.Stdout, os.Stderr)
@@ -50,11 +52,16 @@ func start(ctx context.Context) int {
 }
 
 func (t testlink) run(parent context.Context) (int, error) {
+	// The package values are fw_cfg entries, so read them before connect seals it.
+	spec, specErr := readPackageSpec(t.fwCfgDir)
 	conn, code, err := t.connect(parent)
 	if err != nil {
-		return code, err
+		return code, errors.Join(err, specErr)
 	}
 	defer conn.Close()
+	if specErr != nil {
+		return exitRefused, specErr
+	}
 	link := hostlink.NewTestStream(conn)
 
 	// The handshake can return before the host refuses the guest, so nothing
@@ -69,6 +76,9 @@ func (t testlink) run(parent context.Context) (int, error) {
 	if parent.Err() != nil {
 		return exitLink, context.Cause(parent)
 	}
+	if err := t.pkg.install(spec); err != nil {
+		return exitRefused, fmt.Errorf("install package: %w", err)
+	}
 
 	srv := runtimeapi.New()
 	ln, err := net.Listen("tcp", t.runtimeAddr)

cmd/lambda-agent/start_testlink_test.go

@@ -5,10 +5,14 @@ package main
 import (
 	"bytes"
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"io"
 	"net/http"
 	"os"
 	"os/exec"
+	"path/filepath"
+	"strconv"
 	"testing"
 	"time"
 
@@ -43,12 +47,31 @@ func TestHelperBootstrap(t *testing.T) {
 type bootstrapStarts struct {
 	n       int
 	started chan struct{}
+	task    string
+	sealed  bool
 }
 
-func (b *bootstrapStarts) testlink(g *guest) testlink {
+// testlink builds a guest whose fw_cfg also carries a controlled package,
+// installed into a temp dir without mounting anything.
+func (b *bootstrapStarts) testlink(t *testing.T, entries map[string][]byte) testlink {
+	t.Helper()
+	zipBytes := testZip(t, map[string]string{"bootstrap": "#!/bin/sh\n", "lib/data.txt": "data"})
+	disk := filepath.Join(t.TempDir(), "disk")
+	require.NoError(t, os.WriteFile(disk, zipBytes, 0o600))
+	sum := sha256.Sum256(zipBytes)
+	entries[entryPackageSize] = []byte(strconv.Itoa(len(zipBytes)))
+	entries[entryPackageSHA256] = []byte(hex.EncodeToString(sum[:]))
+	g := newGuest(t, entries)
 	b.started = make(chan struct{}, 1)
+	b.task = t.TempDir()
 	return testlink{
-		startup:     g.startup(),
+		startup: g.startup(),
+		pkg: taskPackage{
+			device: disk, dir: b.task,
+			loadDriver: func() error { return nil },
+			mountDir:   func(string) error { return nil },
+			sealDir:    func(string) error { b.sealed = true; return nil },
+		},
 		runtimeAddr: "127.0.0.1:0",
 		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
 			b.n++
@@ -80,7 +103,7 @@ func TestTestlinkServesOneInvocationThenStops(t *testing.T) {
 	h := newHost(t)
 	h.accept(ctx)
 	var b bootstrapStarts
-	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+	done := runTestlinkAsync(ctx, b.testlink(t, h.entries()))
 
 	conn := h.guest(t)
 	defer conn.Close()
@@ -109,6 +132,8 @@ func TestTestlinkServesOneInvocationThenStops(t *testing.T) {
 		assert.Error(t, err, "agent answered a second invocation")
 	}
 	assert.Equal(t, 1, b.n)
+	assert.True(t, b.sealed, "task dir not made read-only")
+	assert.FileExists(t, filepath.Join(b.task, "lib", "data.txt"))
 }
 
 func TestTestlinkShutdownBeforeInvocationNeverStartsBootstrap(t *testing.T) {
@@ -117,7 +142,7 @@ func TestTestlinkShutdownBeforeInvocationNeverStartsBootstrap(t *testing.T) {
 	h := newHost(t)
 	h.accept(ctx)
 	var b bootstrapStarts
-	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+	done := runTestlinkAsync(ctx, b.testlink(t, h.entries()))
 
 	conn := h.guest(t)
 	defer conn.Close()
@@ -142,7 +167,7 @@ func TestTestlinkHostRefusalNeverStartsBootstrap(t *testing.T) {
 	entries[lambdav1.EntryGuestKey], entries[lambdav1.EntryGuestCert] = wrong[lambdav1.EntryGuestKey], wrong[lambdav1.EntryGuestCert]
 	var b bootstrapStarts
 
-	res := wait(t, runTestlinkAsync(ctx, b.testlink(newGuest(t, entries))), "agent to stop")
+	res := wait(t, runTestlinkAsync(ctx, b.testlink(t, entries)), "agent to stop")
 	assert.Equal(t, exitLink, res.code)
 	require.Error(t, res.err)
 	assert.Contains(t, res.err.Error(), "bad certificate")
@@ -155,7 +180,7 @@ func TestTestlinkHostCloseBeforeFrameNeverStartsBootstrap(t *testing.T) {
 	h := newHost(t)
 	h.accept(ctx)
 	var b bootstrapStarts
-	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+	done := runTestlinkAsync(ctx, b.testlink(t, h.entries()))
 
 	require.NoError(t, h.guest(t).Close())
 	res := wait(t, done, "agent to stop")
@@ -177,3 +202,45 @@ func TestOneInvocationHonoursCancelledContext(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, "r-1", req.RequestID)
 }
+
+// A package that fails its check is refused after authentication, and the
+// bootstrap never starts.
+func TestTestlinkBadPackageNeverStartsBootstrap(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	var b bootstrapStarts
+	tl := b.testlink(t, h.entries())
+	tl.pkg.device = filepath.Join(t.TempDir(), "other")
+	require.NoError(t, os.WriteFile(tl.pkg.device, testZip(t, map[string]string{"bootstrap": "x"}), 0o600))
+	done := runTestlinkAsync(ctx, tl)
+
+	conn := h.guest(t)
+	defer conn.Close()
+	require.NoError(t, hostlink.NewTestStream(conn).HostSend(hostlink.Request{RequestID: "r-1"}))
+	res := wait(t, done, "agent to stop")
+	assert.Equal(t, exitRefused, res.code)
+	assert.ErrorContains(t, res.err, "install package")
+	assert.Zero(t, b.n)
+	assert.False(t, b.sealed)
+}
+
+// Missing package values still seal fw_cfg, and the agent refuses before
+// waiting for any frame.
+func TestTestlinkMissingPackageSpecRefuses(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	var b bootstrapStarts
+	tl := b.testlink(t, h.entries())
+	require.NoError(t, os.RemoveAll(filepath.Join(tl.fwCfgDir, entryPackageSHA256)))
+
+	res := wait(t, runTestlinkAsync(ctx, tl), "agent to stop")
+	assert.Equal(t, exitRefused, res.code)
+	assert.ErrorContains(t, res.err, "read "+entryPackageSHA256)
+	assert.Zero(t, b.n)
+	_, err := os.Stat(tl.fwCfgRoot)
+	assert.ErrorIs(t, err, os.ErrNotExist, "fw_cfg not sealed")
+}

contracts/operator/v1/admin.go

@@ -25,6 +25,10 @@ const (
 	DeletionStateRunning   = "RUNNING"
 	DeletionStateCompleted = "COMPLETED"
 	DeletionStateFailed    = "FAILED"
+
+	// DeletionStateExercised ends a non-finalizing run: the account is left
+	// TERMINATING and a later DeleteAccount takes the same operation over.
+	DeletionStateExercised = "EXERCISED"
 )
 
 // CreateAccountRequest is the /admin/CreateAccount request body.
@@ -52,6 +56,10 @@ type DeleteAccountRequest struct {
 	ClientToken string `json:"clientToken"`
 	Force       bool   `json:"force,omitempty"`
 	DryRun      bool   `json:"dryRun,omitempty"`
+
+	// NonFinalizing quiesces the account, cleans the quiesced EC2 platform
+	// kinds and stops. No legacy stage runs and the account record stays.
+	NonFinalizing bool `json:"nonFinalizing,omitempty"`
 }
 
 // DescribeAccountDeletionRequest is the /admin/DescribeAccountDeletion body.
@@ -76,8 +84,8 @@ type ListAccountsResponse struct {
 }
 
 // IAMQuiescence is the coordinator's recorded view of IAM quiescence for one
-// fence. Recording only: it admits nothing, and QuiescedScope stays empty until
+// fence. Recording only: it admits nothing. QuiescedScope is filled only for a
-// a final inventory re-read exists to catch a late write.
+// non-finalizing run, and holds as of EvaluatedAt: a write after that is not caught.
 type IAMQuiescence struct {
 	OperationID   string            `json:"operationId"`
 	Fence         uint64            `json:"fence"`
@@ -136,3 +144,67 @@ const (
 	ReasonOwnerInventoryIncomplete     = "owner-inventory-incomplete"
 	ReasonFinalInventoryRereadNotWired = "final-inventory-reread-not-wired"
 )
+
+// ExerciseResult is what a non-finalizing run did after IAM quiescence.
+type ExerciseResult struct {
+	// Skipped is the fixed reason no cleanup ran; Cleanup is then absent.
+	Skipped    string           `json:"skipped,omitempty"`
+	Cleanup    *PlatformCleanup `json:"cleanup,omitempty"`
+	Accounting []KindAccounting `json:"accounting,omitempty"`
+}
+
+// PlatformCleanup is the EC2 platform cleanup result. It carries resource IDs,
+// fixed reasons and counts only, never names, tags, key material or data.
+type PlatformCleanup struct {
+	Kinds  []CleanupKind  `json:"kinds"`
+	Items  []CleanupItem  `json:"items,omitempty"`
+	Counts []CleanupCount `json:"counts,omitempty"`
+}
+
+// CleanupKind is one kind's scan status after the cleanup's own re-scan.
+type CleanupKind struct {
+	Kind              string   `json:"kind"`
+	Status            string   `json:"status"`
+	IncompleteReasons []string `json:"incompleteReasons,omitempty"`
+}
+
+// CleanupItem is one resource by its ID with what the cleanup did to it.
+type CleanupItem struct {
+	ResourceID  string `json:"resourceId"`
+	Kind        string `json:"kind"`
+	Disposition string `json:"disposition"`
+	Reason      string `json:"reason,omitempty"`
+	MemberCount int    `json:"memberCount,omitempty"`
+}
+
+// CleanupCount is the number of records of one class that have no public ID.
+type CleanupCount struct {
+	Kind        string `json:"kind"`
+	Class       string `json:"class"`
+	Disposition string `json:"disposition"`
+	Reason      string `json:"reason,omitempty"`
+	Count       int    `json:"count"`
+}
+
+// KindAccounting is the final inventory re-read held against the cleanup result
+// for one kind. Quiescence is not rewritten from it.
+type KindAccounting struct {
+	Kind      string `json:"kind"`
+	Status    string `json:"status"`
+	Reason    string `json:"reason,omitempty"`
+	Remaining int    `json:"remaining"`
+}
+
+// Accounting statuses and fixed reasons of a KindAccounting.
+const (
+	AccountingAccounted   = "accounted"
+	AccountingUnaccounted = "unaccounted"
+
+	AccountingReasonRereadIncomplete = "reread-incomplete"
+	AccountingReasonUnnamedRemainder = "unnamed-remainder"
+)
+
+// Fixed reasons an exercise ran no cleanup.
+const (
+	ExerciseSkippedNoQuiescedKind = "no-quiesced-kind"
+)

contracts/operator/v1/admin_test.go

@@ -43,6 +43,27 @@ func TestAdminWire_DeletionStates(t *testing.T) {
 	require.Equal(t, "RUNNING", DeletionStateRunning)
 	require.Equal(t, "COMPLETED", DeletionStateCompleted)
 	require.Equal(t, "FAILED", DeletionStateFailed)
+	require.Equal(t, "EXERCISED", DeletionStateExercised)
+}
+
+func TestAdminWire_NonFinalizingDeleteRequest(t *testing.T) {
+	goldenJSON(t, DeleteAccountRequest{AccountID: "000000000042", AccountName: "acme", ClientToken: "tok-3", NonFinalizing: true},
+		`{"accountId":"000000000042","accountName":"acme","clientToken":"tok-3","nonFinalizing":true}`)
+}
+
+func TestAdminWire_ExerciseResult(t *testing.T) {
+	goldenJSON(t, ExerciseResult{
+		Cleanup: &PlatformCleanup{
+			Kinds:  []CleanupKind{{Kind: "key-pair", Status: "observed-clean-at-scan-time"}},
+			Items:  []CleanupItem{{ResourceID: "pg-1", Kind: "placement-group", Disposition: "blocked", Reason: "instance-members", MemberCount: 1}},
+			Counts: []CleanupCount{{Kind: "key-pair", Class: "public-key-object-without-metadata", Disposition: "deleted", Count: 2}},
+		},
+		Accounting: []KindAccounting{{Kind: "placement-group", Status: AccountingAccounted, Remaining: 1}},
+	}, `{"cleanup":{"kinds":[{"kind":"key-pair","status":"observed-clean-at-scan-time"}],
+		"items":[{"resourceId":"pg-1","kind":"placement-group","disposition":"blocked","reason":"instance-members","memberCount":1}],
+		"counts":[{"kind":"key-pair","class":"public-key-object-without-metadata","disposition":"deleted","count":2}]},
+		"accounting":[{"kind":"placement-group","status":"accounted","remaining":1}]}`)
+	goldenJSON(t, ExerciseResult{Skipped: ExerciseSkippedNoQuiescedKind}, `{"skipped":"no-quiesced-kind"}`)
 }
 
 func TestAdminWire_ListAccountsResponse(t *testing.T) {

docs/DESIGN.md

@@ -65,9 +65,10 @@ The EC2 registration (`spinifex/domains/ec2/awsapi`, `NewRegistration`) parses t
 The gateway communicates with daemons via NATS request/response. Most calls go through `natsmsg.NATSRequest`, which marshals the input, attaches the account ID as a NATS header, and unmarshals the typed response:
 
 ```go
-func (s *NATSInstanceService) RunInstances(ctx context.Context, input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) {
+func runIntoReservation(ctx context.Context, input *ec2.RunInstancesInput, natsConn *nats.Conn, accountID, crID string) (*ec2.Reservation, error) {
-    topic := fmt.Sprintf("ec2.RunInstances.%s", aws.StringValue(input.InstanceType))
+    subject := "ec2.RunInstances.cr." + crID
-    return natsmsg.NATSRequest[ec2.Reservation](ctx, s.natsConn, topic, input, 5*time.Minute, accountID)
+    reservation, err := natsmsg.NATSRequest[ec2.Reservation](ctx, natsConn, subject, input, 5*time.Minute, accountID)
+    // ...
 }
 ```
 

spinifex/agents/lambda/agent/bootstrap.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"os/exec"
+	"path/filepath"
 	goruntime "runtime"
 	"syscall"
 	"time"
@@ -18,8 +19,8 @@ import (
 const sandboxID = 993
 
 // StartBootstrap starts the function's bootstrap at path as the sandbox uid and
-// gid with no supplementary groups, with no_new_privs, and with exactly env.
+// gid with no supplementary groups, with no_new_privs, and with exactly env,
-// The caller waits on the returned command.
+// in the bootstrap's own directory. The caller waits on the returned command.
 func StartBootstrap(ctx context.Context, path string, env []string, stdout, stderr io.Writer) (*exec.Cmd, error) {
 	cred := &syscall.Credential{Uid: sandboxID, Gid: sandboxID, Groups: []uint32{}}
 	return startBootstrap(ctx, path, env, stdout, stderr, cred)
@@ -30,6 +31,8 @@ func startBootstrap(ctx context.Context, path string, env []string, stdout, stde
 	// A nil Env inherits the agent's environment; the bootstrap gets only env.
 	cmd.Env = append([]string{}, env...)
 	cmd.Stdout, cmd.Stderr = stdout, stderr
+	// Packaged functions open their files relative to the task root.
+	cmd.Dir = filepath.Dir(path)
 	if cred != nil {
 		cmd.SysProcAttr = &syscall.SysProcAttr{Credential: cred}
 	}

spinifex/agents/lambda/agent/bootstrap_test.go

@@ -169,3 +169,11 @@ func TestStartBootstrapMissingExecutableFails(t *testing.T) {
 	_, err := startBootstrap(context.Background(), filepath.Join(t.TempDir(), "bootstrap"), nil, nil, nil, nil)
 	assert.Error(t, err)
 }
+
+func TestStartBootstrapRunsInItsOwnDirectory(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "data.txt"), []byte("packaged"), 0o644))
+	bootstrap := filepath.Join(dir, "bootstrap")
+	require.NoError(t, os.WriteFile(bootstrap, []byte("#!/bin/sh\ncat data.txt\n"), 0o755))
+	assert.Equal(t, "packaged", string(run(t, bootstrap, []string{"PATH=/bin:/usr/bin"}, false)))
+}

spinifex/domains/ec2/accountcleanup/cleanup.go

@@ -69,6 +69,28 @@ var (
 	_ TagIndexOwner       = (*ec2tags.TagsServiceImpl)(nil)
 )
 
+// kindPrefixes maps each cleanable kind to its resource-ID prefix in the tag index.
+var kindPrefixes = map[string]string{
+	ec2v1.PlatformKindKeyPair:        keyPairIDPrefix,
+	ec2v1.PlatformKindPlacementGroup: placementGroupIDPrefix,
+	ec2v1.PlatformKindLaunchTemplate: launchTemplateIDPrefix,
+}
+
+// selectKinds returns the kinds a request cleans, in report order. No kinds
+// means all three; a kind that is not cleanable, such as the tag index, is refused.
+func selectKinds(requested []string) ([]string, error) {
+	all := []string{ec2v1.PlatformKindKeyPair, ec2v1.PlatformKindPlacementGroup, ec2v1.PlatformKindLaunchTemplate}
+	if len(requested) == 0 {
+		return all, nil
+	}
+	for _, kind := range requested {
+		if !slices.Contains(all, kind) {
+			return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "The kind is not one of key-pair, placement-group or launch-template.")
+		}
+	}
+	return slices.DeleteFunc(all, func(kind string) bool { return !slices.Contains(requested, kind) }), nil
+}
+
 // DefaultSettleWindow is how old headerless residue must be before it is
 // removed; anything younger may be a create still in flight.
 const DefaultSettleWindow = 10 * time.Minute
@@ -102,6 +124,7 @@ const (
 // CleanupPlatform deletes the account's key pairs, empty placement groups and
 // launch templates, removes settled residue and orphaned tag-index entries,
 // then re-scans every owner store to build the result. It is safe to re-run.
+// Request.Kinds limits the run to those kinds and their tag-index entries.
 func (c *Cleanup) CleanupPlatform(ctx context.Context, req *Request) (*Result, error) {
 	if req == nil || req.AccountID == "" {
 		return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "The request must specify an account ID.")
@@ -118,27 +141,54 @@ func (c *Cleanup) CleanupPlatform(ctx context.Context, req *Request) (*Result, e
 	acct := req.AccountID
 	res := &Result{AccountID: acct, CorrelationRef: req.CorrelationRef}
 
-	refs := c.readReferences(ctx, acct)
+	kinds, err := selectKinds(req.Kinds)
-	keyRun := c.cleanKeyPairs(ctx, acct, settledBefore, refs)
+	if err != nil {
-	groupRun := c.cleanPlacementGroups(ctx, acct, refs)
+		return nil, err
-	templateRun := c.cleanLaunchTemplates(ctx, acct, settledBefore)
+	}
+	var prefixes []string
+	for _, kind := range kinds {
+		prefixes = append(prefixes, kindPrefixes[kind])
+	}
+
+	var keyRun *keyPairRun
+	var groupRun *placementGroupRun
+	var templateRun *launchTemplateRun
+	if slices.Contains(kinds, ec2v1.PlatformKindKeyPair) || slices.Contains(kinds, ec2v1.PlatformKindPlacementGroup) {
+		refs := c.readReferences(ctx, acct)
+		if slices.Contains(kinds, ec2v1.PlatformKindKeyPair) {
+			keyRun = c.cleanKeyPairs(ctx, acct, settledBefore, refs)
+		}
+		if slices.Contains(kinds, ec2v1.PlatformKindPlacementGroup) {
+			groupRun = c.cleanPlacementGroups(ctx, acct, refs)
+		}
+	}
+	if slices.Contains(kinds, ec2v1.PlatformKindLaunchTemplate) {
+		templateRun = c.cleanLaunchTemplates(ctx, acct, settledBefore)
+	}
 
 	// Tags are listed before the owner re-scans, so a resource created in
 	// between is seen as owned and keeps its tags.
 	tagKind := c.startKind(ec2v1.PlatformKindTagIndex)
-	tagged, tagErr := c.tags.ScanAccountResourceIDs(ctx, acct, keyPairIDPrefix, placementGroupIDPrefix, launchTemplateIDPrefix)
+	tagged, tagErr := c.tags.ScanAccountResourceIDs(ctx, acct, prefixes...)
 	if tagErr != nil {
 		slog.WarnContext(ctx, "Platform cleanup: tag index listing failed", "err", tagErr)
 		tagKind.incomplete(ec2v1.PlatformReasonStoreListFailed)
 	}
 
-	owners := map[string]ownedIDs{
+	owners := map[string]ownedIDs{}
-		keyPairIDPrefix:        c.finishKeyPairs(ctx, acct, keyRun, res),
+	if keyRun != nil {
-		placementGroupIDPrefix: c.finishPlacementGroups(ctx, acct, groupRun, res),
+		owners[keyPairIDPrefix] = c.finishKeyPairs(ctx, acct, keyRun, res)
-		launchTemplateIDPrefix: c.finishLaunchTemplates(ctx, acct, templateRun, res),
+		res.Kinds = append(res.Kinds, keyRun.kind.report)
 	}
-	res.Kinds = []KindReport{keyRun.kind.report, groupRun.kind.report, templateRun.kind.report}
+	if groupRun != nil {
-	res.Kinds = append(res.Kinds, c.cleanTagIndex(ctx, acct, tagKind, tagged, tagErr == nil, owners, res))
+		owners[placementGroupIDPrefix] = c.finishPlacementGroups(ctx, acct, groupRun, res)
+		res.Kinds = append(res.Kinds, groupRun.kind.report)
+	}
+	if templateRun != nil {
+		owners[launchTemplateIDPrefix] = c.finishLaunchTemplates(ctx, acct, templateRun, res)
+		res.Kinds = append(res.Kinds, templateRun.kind.report)
+	}
+	res.Kinds = append(res.Kinds, c.cleanTagIndex(ctx, acct, tagKind, tagged, prefixes, tagErr == nil, owners, res))
 
 	slog.InfoContext(ctx, "Platform cleanup completed", "accountID", acct, "correlationRef", req.CorrelationRef,
 		"items", len(res.Items), "duration_ms", otelsetup.Millis(time.Since(start)))

spinifex/domains/ec2/accountcleanup/cleanup_test.go

@@ -775,3 +775,38 @@ func (f failingResidue) CleanupAccountResidue(ctx context.Context, acct string,
 	}
 	return &ec2launchtemplate.AccountCleanup{KeptTemplateIDs: scan.LiveTemplateIDs, OrphanBodiesFailed: scan.OrphanVersionBodies}, nil
 }
+
+// A run limited to one kind cleans that kind and its tag-index entries only:
+// the other kinds keep their resources and their tags, and are not reported.
+func TestCleanupPlatformLimitedToKindsLeavesTheRestAlone(t *testing.T) {
+	o := newOwners(t)
+	keyID := o.importKey(t, testAccount, "k1")
+	groupID := o.createGroup(t, testAccount, "g1")
+	o.tag(t, testAccount, keyID)
+
+	c := o.newCleanup()
+	c.now = func() time.Time { return time.Now().Add(time.Hour) }
+	res, err := c.CleanupPlatform(t.Context(), &Request{AccountID: testAccount, Kinds: []string{ec2v1.PlatformKindPlacementGroup}})
+	require.NoError(t, err)
+
+	assert.Equal(t, []Item{item(groupID, ec2v1.PlatformKindPlacementGroup, DispositionDeleted, "")}, res.Items)
+	assert.Equal(t, map[string]string{
+		ec2v1.PlatformKindPlacementGroup: ec2v1.PlatformStatusObservedClean,
+		ec2v1.PlatformKindTagIndex:       ec2v1.PlatformStatusObservedClean,
+	}, statuses(res))
+
+	keys, err := o.keys.ScanAccount(t.Context(), testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, []string{keyID}, keys.LiveKeyPairIDs, "an unselected kind is not touched")
+	tagged, err := o.tags.ScanAccountResourceIDs(t.Context(), testAccount, "key-")
+	require.NoError(t, err)
+	assert.Equal(t, []string{keyID}, tagged, "an unselected kind keeps its tags")
+}
+
+func TestCleanupPlatformRefusesAKindItCannotClean(t *testing.T) {
+	o := newOwners(t)
+	for _, kinds := range [][]string{{ec2v1.PlatformKindTagIndex}, {"volume"}} {
+		_, err := o.newCleanup().CleanupPlatform(t.Context(), &Request{AccountID: testAccount, Kinds: kinds})
+		require.Error(t, err, "%v", kinds)
+	}
+}

spinifex/domains/ec2/accountcleanup/kinds.go

@@ -282,15 +282,15 @@ func (c *Cleanup) finishLaunchTemplates(ctx context.Context, acct string, run *l
 // cleanTagIndex clears entries whose owner re-scan resolved and holds no
 // record for the ID, then re-lists the index to prove each clear. Entries of
 // resources that still exist are theirs, not residue, and are left alone.
-func (c *Cleanup) cleanTagIndex(ctx context.Context, acct string, kind *kindRun, tagged []string, listed bool, owners map[string]ownedIDs, res *Result) KindReport {
+func (c *Cleanup) cleanTagIndex(ctx context.Context, acct string, kind *kindRun, tagged, prefixes []string, listed bool, owners map[string]ownedIDs, res *Result) KindReport {
 	if listed {
-		c.clearOrphanTags(ctx, acct, kind, tagged, owners, res)
+		c.clearOrphanTags(ctx, acct, kind, tagged, prefixes, owners, res)
 	}
 	c.finishKind(kind)
 	return kind.report
 }
 
-func (c *Cleanup) clearOrphanTags(ctx context.Context, acct string, kind *kindRun, tagged []string, owners map[string]ownedIDs, res *Result) {
+func (c *Cleanup) clearOrphanTags(ctx context.Context, acct string, kind *kindRun, tagged, prefixes []string, owners map[string]ownedIDs, res *Result) {
 	// A resource this run reports as not deleted owns its tags even when its
 	// kind's re-scan failed.
 	reported := map[string]bool{}
@@ -317,7 +317,7 @@ func (c *Cleanup) clearOrphanTags(ctx context.Context, acct string, kind *kindRu
 			cleared = append(cleared, id)
 		}
 	}
-	after, err := c.tags.ScanAccountResourceIDs(ctx, acct, keyPairIDPrefix, placementGroupIDPrefix, launchTemplateIDPrefix)
+	after, err := c.tags.ScanAccountResourceIDs(ctx, acct, prefixes...)
 	if err != nil {
 		slog.WarnContext(ctx, "Platform cleanup: tag index re-scan failed", "err", err)
 		kind.incomplete(ec2v1.PlatformReasonStoreListFailed)

spinifex/domains/ec2/accountcleanup/types.go

@@ -7,6 +7,9 @@ import "time"
 type Request struct {
 	AccountID      string `json:"account_id"`
 	CorrelationRef string `json:"correlation_ref,omitempty"`
+
+	// Kinds limits the run to these platform kinds. Empty means all three.
+	Kinds []string `json:"kinds,omitempty"`
 }
 
 // Result is what the owner stores hold after one run. Kinds reuse the ec2.v1

spinifex/domains/ec2/instance/service_nats.go

@@ -1,80 +0,0 @@
-package instance
-
-import (
-	"context"
-	"fmt"
-	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
-	"time"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/nats-io/nats.go"
-)
-
-// NATSInstanceService handles instance operations via NATS. Fan-out methods
-// (DescribeInstances, DescribeInstanceTypes) intentionally error — use the
-// gateway's scatter-gather helper for cluster-wide results.
-type NATSInstanceService struct {
-	natsConn *nats.Conn
-}
-
-var _ InstanceService = (*NATSInstanceService)(nil)
-
-// NewNATSInstanceService creates a new NATS-based instance service.
-func NewNATSInstanceService(conn *nats.Conn) InstanceService {
-	return &NATSInstanceService{natsConn: conn}
-}
-
-func (s *NATSInstanceService) RunInstances(ctx context.Context, input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) {
-	if input == nil || input.InstanceType == nil {
-		return nil, fmt.Errorf("instance type is required")
-	}
-	topic := fmt.Sprintf("ec2.RunInstances.%s", aws.StringValue(input.InstanceType))
-	return natsmsg.NATSRequest[ec2.Reservation](ctx, s.natsConn, topic, input, 5*time.Minute, accountID)
-}
-
-// DescribeInstances is a fan-out operation; use the gateway's scatter-gather
-// helper instead of this single-node client.
-func (s *NATSInstanceService) DescribeInstances(_ context.Context, _ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) {
-	return nil, fmt.Errorf("DescribeInstances is fan-out; use gateway scatter-gather, not NATSInstanceService")
-}
-
-// DescribeInstanceTypes is a fan-out operation; use the gateway's scatter-gather
-// helper instead of this single-node client.
-func (s *NATSInstanceService) DescribeInstanceTypes(_ context.Context, _ *ec2.DescribeInstanceTypesInput, _ string) (*DescribeInstanceTypesNodeOutput, error) {
-	return nil, fmt.Errorf("DescribeInstanceTypes is fan-out; use gateway scatter-gather, not NATSInstanceService")
-}
-
-// DescribeInstanceStatus is a fan-out operation; use the gateway's scatter-gather
-// helper instead of this single-node client.
-func (s *NATSInstanceService) DescribeInstanceStatus(_ context.Context, _ *ec2.DescribeInstanceStatusInput, _ string) (*ec2.DescribeInstanceStatusOutput, error) {
-	return nil, fmt.Errorf("DescribeInstanceStatus is fan-out; use gateway scatter-gather, not NATSInstanceService")
-}
-
-func (s *NATSInstanceService) DescribeInstanceAttribute(ctx context.Context, input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) {
-	return natsmsg.NATSRequest[ec2.DescribeInstanceAttributeOutput](ctx, s.natsConn, "ec2.DescribeInstanceAttribute", input, 30*time.Second, accountID)
-}
-
-func (s *NATSInstanceService) DescribeStoppedInstances(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
-	return natsmsg.NATSRequest[ec2.DescribeInstancesOutput](ctx, s.natsConn, "ec2.DescribeStoppedInstances", input, 30*time.Second, accountID)
-}
-
-func (s *NATSInstanceService) DescribeTerminatedInstances(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
-	return natsmsg.NATSRequest[ec2.DescribeInstancesOutput](ctx, s.natsConn, "ec2.DescribeTerminatedInstances", input, 30*time.Second, accountID)
-}
-
-func (s *NATSInstanceService) ModifyInstanceAttribute(ctx context.Context, input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error) {
-	return natsmsg.NATSRequest[ec2.ModifyInstanceAttributeOutput](ctx, s.natsConn, "ec2.ModifyInstanceAttribute", input, 30*time.Second, accountID)
-}
-
-func (s *NATSInstanceService) ModifyInstanceMetadataOptions(ctx context.Context, input *ec2.ModifyInstanceMetadataOptionsInput, accountID string) (*ec2.ModifyInstanceMetadataOptionsOutput, error) {
-	return natsmsg.NATSRequest[ec2.ModifyInstanceMetadataOptionsOutput](ctx, s.natsConn, "ec2.ModifyInstanceMetadataOptions", input, 30*time.Second, accountID)
-}
-
-func (s *NATSInstanceService) StartStoppedInstance(ctx context.Context, input *StartStoppedInstanceInput, accountID string) (*StartStoppedInstanceOutput, error) {
-	return natsmsg.NATSRequest[StartStoppedInstanceOutput](ctx, s.natsConn, "ec2.StartStoppedInstance", input, 30*time.Second, accountID)
-}
-
-func (s *NATSInstanceService) TerminateStoppedInstance(ctx context.Context, input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) {
-	return natsmsg.NATSRequest[TerminateStoppedInstanceOutput](ctx, s.natsConn, "ec2.TerminateStoppedInstance", input, 30*time.Second, accountID)
-}

spinifex/domains/ec2/instance/service_nats_test.go

@@ -1,364 +0,0 @@
-package instance
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"os"
-	"testing"
-	"time"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/internal/testkit"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// startTestNATSServer starts an embedded NATS server for testing.
-func startTestNATSServer(t *testing.T) (*server.Server, string) {
-	t.Helper()
-	ns, nc := testutil.StartTestNATS(t)
-	_ = nc // connection managed by testutil cleanup
-	return ns, ns.ClientURL()
-}
-
-// createValidRunInstancesInput creates a valid RunInstancesInput for testing.
-func createValidRunInstancesInput() *ec2.RunInstancesInput {
-	return &ec2.RunInstancesInput{
-		ImageId:      aws.String("ami-0abcdef1234567890"),
-		InstanceType: aws.String("t3.micro"),
-		MinCount:     aws.Int64(1),
-		MaxCount:     aws.Int64(1),
-		KeyName:      aws.String("test-key-pair"),
-		SecurityGroupIds: []*string{
-			aws.String("sg-0123456789abcdef0"),
-		},
-		SubnetId: aws.String("subnet-6e7f829e"),
-		UserData: aws.String("#!/bin/bash\necho 'test'"),
-	}
-}
-
-// createValidReservation creates a valid ec2.Reservation response for testing.
-func createValidReservation() *ec2.Reservation {
-	return &ec2.Reservation{
-		ReservationId: aws.String("r-0123456789abcdef0"),
-		OwnerId:       aws.String("123456789012"),
-		Instances: []*ec2.Instance{
-			{
-				InstanceId:   aws.String("i-0123456789abcdef0"),
-				InstanceType: aws.String("t3.micro"),
-				ImageId:      aws.String("ami-0abcdef1234567890"),
-				State: &ec2.InstanceState{
-					Code: aws.Int64(0),
-					Name: aws.String("pending"),
-				},
-				PrivateIpAddress: aws.String("10.0.1.100"),
-				SubnetId:         aws.String("subnet-6e7f829e"),
-			},
-		},
-	}
-}
-
-// TestNATSInstanceService_RunInstances_Success tests successful RunInstances operation.
-func TestNATSInstanceService_RunInstances_Success(t *testing.T) {
-	// Skip if LOG_IGNORE is set
-	if os.Getenv("LOG_IGNORE") != "" {
-		t.Setenv("LOG_IGNORE", "1")
-	}
-
-	// Start test NATS server
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
-
-	// Create NATS connections
-	nc, err := nats.Connect(natsURL)
-	require.NoError(t, err, "Failed to connect to NATS")
-	defer nc.Close()
-
-	// Create mock daemon subscriber that responds with valid reservation
-	// Subscribe to the per-instance-type topic (t3.micro from createValidRunInstancesInput)
-	mockReservation := createValidReservation()
-	_, err = nc.QueueSubscribe("ec2.RunInstances.t3.micro", "spinifex-workers", func(msg *nats.Msg) {
-		// Validate that request is properly formatted
-		var input ec2.RunInstancesInput
-		err := json.Unmarshal(msg.Data, &input)
-		if err != nil {
-			t.Errorf("Mock daemon received invalid JSON: %v", err)
-			return
-		}
-
-		// Send back valid reservation
-		responseData, _ := json.Marshal(mockReservation)
-		msg.Respond(responseData)
-	})
-	require.NoError(t, err, "Failed to subscribe mock daemon")
-
-	// Create NATSInstanceService
-	service := NewNATSInstanceService(nc)
-	require.NotNil(t, service)
-
-	// Create valid input
-	input := createValidRunInstancesInput()
-
-	// Call RunInstances
-	reservation, err := service.RunInstances(context.Background(), input, "123456789012")
-
-	// Verify success
-	require.NoError(t, err, "RunInstances should succeed")
-	require.NotNil(t, reservation, "Reservation should not be nil")
-
-	// Verify reservation contents
-	assert.Equal(t, *mockReservation.ReservationId, *reservation.ReservationId)
-	assert.Equal(t, *mockReservation.OwnerId, *reservation.OwnerId)
-	require.Len(t, reservation.Instances, 1)
-
-	instance := reservation.Instances[0]
-	assert.Equal(t, "i-0123456789abcdef0", *instance.InstanceId)
-	assert.Equal(t, "t3.micro", *instance.InstanceType)
-	assert.Equal(t, int64(0), *instance.State.Code)
-	assert.Equal(t, "pending", *instance.State.Name)
-}
-
-// TestNATSInstanceService_RunInstances_DaemonError tests error handling when daemon returns error.
-func TestNATSInstanceService_RunInstances_DaemonError(t *testing.T) {
-	if os.Getenv("LOG_IGNORE") != "" {
-		t.Setenv("LOG_IGNORE", "1")
-	}
-
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
-
-	nc, err := nats.Connect(natsURL)
-	require.NoError(t, err)
-	defer nc.Close()
-
-	// Create mock daemon that responds with error
-	// Subscribe to the per-instance-type topic for the requested type
-	_, err = nc.QueueSubscribe("ec2.RunInstances.invalid.type", "spinifex-workers", func(msg *nats.Msg) {
-		// Send back error response
-		errorResponse := awserrors.GenerateErrorPayload("InvalidInstanceType")
-		msg.Respond(errorResponse)
-	})
-	require.NoError(t, err)
-
-	service := NewNATSInstanceService(nc)
-
-	input := &ec2.RunInstancesInput{
-		ImageId:      aws.String("ami-0abcdef1234567890"),
-		InstanceType: aws.String("invalid.type"),
-		MinCount:     aws.Int64(1),
-		MaxCount:     aws.Int64(1),
-	}
-
-	// Call RunInstances
-	reservation, err := service.RunInstances(context.Background(), input, "123456789012")
-
-	// Verify error handling - error should be just the AWS error code for gateway lookup
-	require.Error(t, err, "Should return error when daemon returns error")
-	assert.Nil(t, reservation, "Reservation should be nil on error")
-	assert.Equal(t, "InvalidInstanceType", err.Error(), "Error should be the AWS error code")
-}
-
-// TestNATSInstanceService_RunInstances_NoSubscriber tests behavior when no daemon is subscribed.
-func TestNATSInstanceService_RunInstances_NoSubscriber(t *testing.T) {
-	if os.Getenv("LOG_IGNORE") != "" {
-		t.Setenv("LOG_IGNORE", "1")
-	}
-
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
-
-	nc, err := nats.Connect(natsURL)
-	require.NoError(t, err)
-	defer nc.Close()
-
-	// No subscriber - no daemon running
-
-	service := NewNATSInstanceService(nc)
-	input := createValidRunInstancesInput()
-
-	// Call RunInstances
-	start := time.Now()
-	reservation, err := service.RunInstances(context.Background(), input, "123456789012")
-	duration := time.Since(start)
-
-	// Verify error behavior
-	// When no subscribers are available, NATS returns ErrNoResponders immediately
-	require.Error(t, err, "Should fail when no subscriber")
-	assert.Nil(t, reservation, "Reservation should be nil")
-	assert.ErrorIs(t, err, nats.ErrNoResponders, "Error should be ErrNoResponders")
-
-	// Should fail quickly (within 1 second), not wait for 30 second timeout
-	assert.Less(t, duration.Seconds(), 1.0, "Should fail immediately when no responders")
-}
-
-// TestNATSInstanceService_RunInstances_InvalidResponse tests handling of malformed responses.
-func TestNATSInstanceService_RunInstances_InvalidResponse(t *testing.T) {
-	if os.Getenv("LOG_IGNORE") != "" {
-		t.Setenv("LOG_IGNORE", "1")
-	}
-
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
-
-	nc, err := nats.Connect(natsURL)
-	require.NoError(t, err)
-	defer nc.Close()
-
-	// Create mock daemon that responds with invalid JSON
-	_, err = nc.QueueSubscribe("ec2.RunInstances.t3.micro", "spinifex-workers", func(msg *nats.Msg) {
-		// Send back malformed JSON
-		msg.Respond([]byte(`{"invalid": json response`))
-	})
-	require.NoError(t, err)
-
-	service := NewNATSInstanceService(nc)
-	input := createValidRunInstancesInput()
-
-	// Call RunInstances
-	reservation, err := service.RunInstances(context.Background(), input, "123456789012")
-
-	// Verify error handling
-	require.Error(t, err, "Should return error for invalid JSON")
-	assert.Nil(t, reservation, "Reservation should be nil")
-	assert.Contains(t, err.Error(), "failed to unmarshal", "Error should indicate unmarshal failure")
-}
-
-// TestNATSInstanceService_RunInstances_MarshalError tests handling when input marshaling fails.
-func TestNATSInstanceService_RunInstances_MarshalError(t *testing.T) {
-	if os.Getenv("LOG_IGNORE") != "" {
-		t.Setenv("LOG_IGNORE", "1")
-	}
-
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
-
-	nc, err := nats.Connect(natsURL)
-	require.NoError(t, err)
-	defer nc.Close()
-
-	service := NewNATSInstanceService(nc)
-
-	// Nil input should be rejected before hitting NATS
-	reservation, err := service.RunInstances(context.Background(), nil, "123456789012")
-	require.Error(t, err, "Should handle nil input")
-	assert.Nil(t, reservation)
-	assert.Contains(t, err.Error(), "instance type is required")
-}
-
-// TestNATSInstanceService_RunInstances_MultipleInstances tests launching multiple instances.
-func TestNATSInstanceService_RunInstances_MultipleInstances(t *testing.T) {
-	if os.Getenv("LOG_IGNORE") != "" {
-		t.Setenv("LOG_IGNORE", "1")
-	}
-
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
-
-	nc, err := nats.Connect(natsURL)
-	require.NoError(t, err)
-	defer nc.Close()
-
-	// Create mock daemon that returns multiple instances
-	_, err = nc.QueueSubscribe("ec2.RunInstances.t3.micro", "spinifex-workers", func(msg *nats.Msg) {
-		var input ec2.RunInstancesInput
-		json.Unmarshal(msg.Data, &input)
-
-		// Create reservation with count matching MinCount
-		count := int(*input.MinCount)
-		instances := make([]*ec2.Instance, count)
-		for i := range count {
-			instanceId := aws.String(aws.StringValue(aws.String(fmt.Sprintf("i-%d", i))))
-			instances[i] = &ec2.Instance{
-				InstanceId:   instanceId,
-				InstanceType: input.InstanceType,
-				ImageId:      input.ImageId,
-				State: &ec2.InstanceState{
-					Code: aws.Int64(0),
-					Name: aws.String("pending"),
-				},
-			}
-		}
-
-		reservation := &ec2.Reservation{
-			ReservationId: aws.String("r-0123456789abcdef0"),
-			Instances:     instances,
-		}
-
-		responseData, _ := json.Marshal(reservation)
-		msg.Respond(responseData)
-	})
-	require.NoError(t, err)
-
-	service := NewNATSInstanceService(nc)
-
-	// Request 3 instances
-	input := createValidRunInstancesInput()
-	input.MinCount = aws.Int64(3)
-	input.MaxCount = aws.Int64(3)
-
-	reservation, err := service.RunInstances(context.Background(), input, "123456789012")
-
-	// Verify multiple instances returned
-	require.NoError(t, err)
-	require.NotNil(t, reservation)
-	assert.Len(t, reservation.Instances, 3, "Should have 3 instances")
-}
-
-// TestNATSInstanceService_RunInstances_QueueGroup tests load balancing with queue groups.
-func TestNATSInstanceService_RunInstances_QueueGroup(t *testing.T) {
-	if os.Getenv("LOG_IGNORE") != "" {
-		t.Setenv("LOG_IGNORE", "1")
-	}
-
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
-
-	nc, err := nats.Connect(natsURL)
-	require.NoError(t, err)
-	defer nc.Close()
-
-	// Create multiple mock daemon workers in same queue group
-	worker1Count := 0
-	worker2Count := 0
-
-	_, err = nc.QueueSubscribe("ec2.RunInstances.t3.micro", "spinifex-workers", func(msg *nats.Msg) {
-		worker1Count++
-		reservation := createValidReservation()
-		responseData, _ := json.Marshal(reservation)
-		msg.Respond(responseData)
-	})
-	require.NoError(t, err)
-
-	_, err = nc.QueueSubscribe("ec2.RunInstances.t3.micro", "spinifex-workers", func(msg *nats.Msg) {
-		worker2Count++
-		reservation := createValidReservation()
-		responseData, _ := json.Marshal(reservation)
-		msg.Respond(responseData)
-	})
-	require.NoError(t, err)
-
-	service := NewNATSInstanceService(nc)
-	input := createValidRunInstancesInput()
-
-	// Send multiple requests
-	requestCount := 10
-	for range requestCount {
-		reservation, err := service.RunInstances(context.Background(), input, "123456789012")
-		require.NoError(t, err)
-		require.NotNil(t, reservation)
-	}
-
-	// Verify load was distributed (both workers should have handled some requests)
-	// With queue groups, NATS distributes messages across subscribers
-	t.Logf("Worker 1 handled: %d requests", worker1Count)
-	t.Logf("Worker 2 handled: %d requests", worker2Count)
-
-	assert.Positive(t, worker1Count, "Worker 1 should handle some requests")
-	assert.Positive(t, worker2Count, "Worker 2 should handle some requests")
-	assert.Equal(t, requestCount, worker1Count+worker2Count, "All requests should be handled")
-}

spinifex/domains/ecr/awsapi/gateway_image_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/mulgadc/spinifex/internal/testkit"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -25,7 +26,8 @@ import (
 // object store + memory meta) and its composed ECR JSON image-action service.
 func newImageGateway(t *testing.T) *ecrGW {
 	t.Helper()
-	meta := ecrdomain.NewMemoryMetaStore()
+	_, _, js := testutil.StartTestJetStream(t)
+	meta := ecrdomain.NewKVMetaStore(js)
 	reg := ecrregistry.NewRegistry(objectstore.NewMemoryObjectStore(), meta, ecrTestAccount)
 	gw := newECRGW(nil, awsapi.Deps{Registry: awsapi.NewRegistryActionService(reg, reg, reg, reg), LifecyclePreview: awsapi.NewLifecyclePreviewActionService(meta, reg)})
 	imageRegistries.Store(gw, reg)

spinifex/domains/ecr/meta.go

@@ -5,8 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"maps"
-	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -593,295 +591,6 @@ func (s *KVMetaStore) DeleteUpload(ctx context.Context, accountID, uploadID stri
 	return nil
 }
 
-// MemoryMetaStore is an in-memory MetaStore used by tests and single-process
-// dev runs. It is safe for concurrent use.
-type MemoryMetaStore struct {
-	mu        sync.Mutex
-	repos     map[string]map[string]RepoMeta     // account -> repo -> meta
-	policies  map[string]map[string][]byte       // account -> repo -> policyText
-	lifecycle map[string]map[string][]byte       // account -> repo -> lifecyclePolicyText
-	tags      map[string]map[string]string       // account -> repo|tag -> digest
-	manifests map[string]map[string]ManifestMeta // account -> repo|digest -> meta
-	uploads   map[string]map[string]uploadRev    // account -> id -> state+rev
-}
-
-type uploadRev struct {
-	state UploadState
-	rev   uint64
-}
-
-var _ MetaStore = (*MemoryMetaStore)(nil)
-
-// NewMemoryMetaStore constructs an empty in-memory MetaStore.
-func NewMemoryMetaStore() *MemoryMetaStore {
-	return &MemoryMetaStore{
-		repos:     make(map[string]map[string]RepoMeta),
-		policies:  make(map[string]map[string][]byte),
-		lifecycle: make(map[string]map[string][]byte),
-		tags:      make(map[string]map[string]string),
-		manifests: make(map[string]map[string]ManifestMeta),
-		uploads:   make(map[string]map[string]uploadRev),
-	}
-}
-
-func (m *MemoryMetaStore) PutRepo(_ context.Context, accountID string, meta RepoMeta) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.repos[accountID] == nil {
-		m.repos[accountID] = make(map[string]RepoMeta)
-	}
-	m.repos[accountID][meta.Name] = meta
-	return nil
-}
-
-func (m *MemoryMetaStore) GetRepo(_ context.Context, accountID, repo string) (RepoMeta, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	r, ok := m.repos[accountID][repo]
-	if !ok {
-		return RepoMeta{}, ErrNotFound
-	}
-	return r, nil
-}
-
-func (m *MemoryMetaStore) ListRepos(_ context.Context, accountID string) ([]string, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	return slices.Sorted(maps.Keys(m.repos[accountID])), nil
-}
-
-func (m *MemoryMetaStore) DeleteRepo(_ context.Context, accountID, repo string) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if _, ok := m.repos[accountID][repo]; !ok {
-		return ErrNotFound
-	}
-	delete(m.repos[accountID], repo)
-	delete(m.policies[accountID], repo)
-	delete(m.lifecycle[accountID], repo)
-	for k := range m.tags[accountID] {
-		if r, _, ok := strings.Cut(k, "|"); ok && r == repo {
-			delete(m.tags[accountID], k)
-		}
-	}
-	for k := range m.manifests[accountID] {
-		if r, _, ok := strings.Cut(k, "|"); ok && r == repo {
-			delete(m.manifests[accountID], k)
-		}
-	}
-	return nil
-}
-
-func (m *MemoryMetaStore) ListManifests(_ context.Context, accountID, repo string) ([]string, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	var out []string
-	for k := range m.manifests[accountID] {
-		if r, d, ok := strings.Cut(k, "|"); ok && r == repo {
-			out = append(out, d)
-		}
-	}
-	sort.Strings(out)
-	return out, nil
-}
-
-func (m *MemoryMetaStore) PutRepoPolicy(_ context.Context, accountID, repo string, policyText []byte) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.policies[accountID] == nil {
-		m.policies[accountID] = make(map[string][]byte)
-	}
-	m.policies[accountID][repo] = policyText
-	return nil
-}
-
-func (m *MemoryMetaStore) GetRepoPolicy(_ context.Context, accountID, repo string) ([]byte, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	p, ok := m.policies[accountID][repo]
-	if !ok {
-		return nil, ErrNotFound
-	}
-	return p, nil
-}
-
-func (m *MemoryMetaStore) DeleteRepoPolicy(_ context.Context, accountID, repo string) ([]byte, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	p, ok := m.policies[accountID][repo]
-	if !ok {
-		return nil, ErrNotFound
-	}
-	delete(m.policies[accountID], repo)
-	return p, nil
-}
-
-func (m *MemoryMetaStore) PutLifecyclePolicy(_ context.Context, accountID, repo string, policyText []byte) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.lifecycle[accountID] == nil {
-		m.lifecycle[accountID] = make(map[string][]byte)
-	}
-	m.lifecycle[accountID][repo] = policyText
-	return nil
-}
-
-func (m *MemoryMetaStore) GetLifecyclePolicy(_ context.Context, accountID, repo string) ([]byte, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	p, ok := m.lifecycle[accountID][repo]
-	if !ok {
-		return nil, ErrNotFound
-	}
-	return p, nil
-}
-
-func (m *MemoryMetaStore) DeleteLifecyclePolicy(_ context.Context, accountID, repo string) ([]byte, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	p, ok := m.lifecycle[accountID][repo]
-	if !ok {
-		return nil, ErrNotFound
-	}
-	delete(m.lifecycle[accountID], repo)
-	return p, nil
-}
-
-func (m *MemoryMetaStore) PutTag(_ context.Context, accountID, repo, tag, digest string) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.tags[accountID] == nil {
-		m.tags[accountID] = make(map[string]string)
-	}
-	m.tags[accountID][repo+"|"+tag] = digest
-	return nil
-}
-
-func (m *MemoryMetaStore) GetTag(_ context.Context, accountID, repo, tag string) (string, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	d, ok := m.tags[accountID][repo+"|"+tag]
-	if !ok {
-		return "", ErrNotFound
-	}
-	return d, nil
-}
-
-func (m *MemoryMetaStore) DeleteTag(_ context.Context, accountID, repo, tag string) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	key := repo + "|" + tag
-	if _, ok := m.tags[accountID][key]; !ok {
-		return ErrNotFound
-	}
-	delete(m.tags[accountID], key)
-	return nil
-}
-
-func (m *MemoryMetaStore) ListTags(_ context.Context, accountID, repo string) ([]string, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	var out []string
-	for k := range m.tags[accountID] {
-		if r, t, ok := strings.Cut(k, "|"); ok && r == repo {
-			out = append(out, t)
-		}
-	}
-	sort.Strings(out)
-	return out, nil
-}
-
-func (m *MemoryMetaStore) PutManifestMeta(_ context.Context, accountID, repo string, meta ManifestMeta) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.manifests[accountID] == nil {
-		m.manifests[accountID] = make(map[string]ManifestMeta)
-	}
-	m.manifests[accountID][repo+"|"+meta.Digest] = meta
-	return nil
-}
-
-func (m *MemoryMetaStore) GetManifestMeta(_ context.Context, accountID, repo, digest string) (ManifestMeta, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	meta, ok := m.manifests[accountID][repo+"|"+digest]
-	if !ok {
-		return ManifestMeta{}, ErrNotFound
-	}
-	return meta, nil
-}
-
-func (m *MemoryMetaStore) DeleteManifestMeta(_ context.Context, accountID, repo, digest string) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	key := repo + "|" + digest
-	if _, ok := m.manifests[accountID][key]; !ok {
-		return ErrNotFound
-	}
-	delete(m.manifests[accountID], key)
-	return nil
-}
-
-func (m *MemoryMetaStore) ListReferrers(_ context.Context, accountID, repo, subjectDigest string) ([]ReferrerDescriptor, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	out := []ReferrerDescriptor{}
-	for k, meta := range m.manifests[accountID] {
-		if r, _, ok := strings.Cut(k, "|"); ok && r == repo && meta.SubjectDigest == subjectDigest {
-			out = append(out, meta.referrerDescriptor())
-		}
-	}
-	sort.Slice(out, func(i, j int) bool { return out[i].Digest < out[j].Digest })
-	return out, nil
-}
-
-func (m *MemoryMetaStore) PutUpload(_ context.Context, accountID, uploadID string, state UploadState) (uint64, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.uploads[accountID] == nil {
-		m.uploads[accountID] = make(map[string]uploadRev)
-	}
-	rev := m.uploads[accountID][uploadID].rev + 1
-	m.uploads[accountID][uploadID] = uploadRev{state: state, rev: rev}
-	return rev, nil
-}
-
-func (m *MemoryMetaStore) GetUpload(_ context.Context, accountID, uploadID string) (UploadState, uint64, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	u, ok := m.uploads[accountID][uploadID]
-	if !ok {
-		return UploadState{}, 0, ErrNotFound
-	}
-	return u.state, u.rev, nil
-}
-
-func (m *MemoryMetaStore) UpdateUpload(_ context.Context, accountID, uploadID string, state UploadState, rev uint64) (uint64, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	u, ok := m.uploads[accountID][uploadID]
-	if !ok {
-		return 0, ErrNotFound
-	}
-	if u.rev != rev {
-		return 0, ErrConflict
-	}
-	u.state = state
-	u.rev = rev + 1
-	m.uploads[accountID][uploadID] = u
-	return u.rev, nil
-}
-
-func (m *MemoryMetaStore) DeleteUpload(_ context.Context, accountID, uploadID string) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if _, ok := m.uploads[accountID][uploadID]; !ok {
-		return ErrNotFound
-	}
-	delete(m.uploads[accountID], uploadID)
-	return nil
-}
-
 // mapKVErr normalizes JetStream KV errors to the MetaStore error vocabulary.
 func mapKVErr(err error) error {
 	if errors.Is(err, jetstream.ErrKeyNotFound) {

spinifex/domains/ecr/meta_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -61,8 +62,8 @@ func TestKVKeyHelpers(t *testing.T) {
 	assert.Equal(t, "sha256-abc", DigestToken("sha256:abc"))
 }
 
-func TestMemoryMetaStore_RepoRoundtrip(t *testing.T) {
+func TestKVMetaStore_RepoRoundtrip(t *testing.T) {
-	m := NewMemoryMetaStore()
+	m := newTestKVMetaStore(t)
 	const acct = "000000000000"
 
 	_, err := m.GetRepo(context.Background(), acct, "team/app")
@@ -80,8 +81,8 @@ func TestMemoryMetaStore_RepoRoundtrip(t *testing.T) {
 	assert.Equal(t, []string{"team/app", "team/web"}, repos)
 }
 
-func TestMemoryMetaStore_TagsAndManifests(t *testing.T) {
+func TestKVMetaStore_TagsAndManifests(t *testing.T) {
-	m := NewMemoryMetaStore()
+	m := newTestKVMetaStore(t)
 	const acct = "000000000000"
 
 	require.NoError(t, m.PutTag(context.Background(), acct, "r", "v1", "sha256:aaa"))
@@ -112,8 +113,8 @@ func TestMemoryMetaStore_TagsAndManifests(t *testing.T) {
 	assert.ErrorIs(t, m.DeleteManifestMeta(context.Background(), acct, "r", "sha256:ccc"), ErrNotFound)
 }
 
-func TestMemoryMetaStore_UploadCAS(t *testing.T) {
+func TestKVMetaStore_UploadCAS(t *testing.T) {
-	m := NewMemoryMetaStore()
+	m := newTestKVMetaStore(t)
 	const acct = "000000000000"
 
 	rev, err := m.PutUpload(context.Background(), acct, "u1", UploadState{RepoName: "r"})
@@ -143,8 +144,8 @@ func TestTrimSuffixMeta(t *testing.T) {
 	assert.Equal(t, "team/app", trimSuffixMeta("repos/team/app/meta"))
 }
 
-func TestMemoryMetaStore_DeleteRepoAndListManifests(t *testing.T) {
+func TestKVMetaStore_DeleteRepoAndListManifests(t *testing.T) {
-	m := NewMemoryMetaStore()
+	m := newTestKVMetaStore(t)
 	const acct = "000000000000"
 
 	require.NoError(t, m.PutRepo(context.Background(), acct, RepoMeta{Name: "team/app", CreatedAt: time.Now()}))
@@ -156,7 +157,7 @@ func TestMemoryMetaStore_DeleteRepoAndListManifests(t *testing.T) {
 
 	digs, err := m.ListManifests(context.Background(), acct, "team/app")
 	require.NoError(t, err)
-	assert.Equal(t, []string{"sha256:ccc"}, digs)
+	assert.Equal(t, []string{DigestToken("sha256:ccc")}, digs)
 
 	require.NoError(t, m.DeleteRepo(context.Background(), acct, "team/app"))
 	_, err = m.GetRepo(context.Background(), acct, "team/app")
@@ -180,50 +181,6 @@ func TestMemoryMetaStore_DeleteRepoAndListManifests(t *testing.T) {
 	assert.ErrorIs(t, m.DeleteRepo(context.Background(), acct, "team/ghost"), ErrNotFound)
 }
 
-func TestMemoryMetaStore_RepoPolicy(t *testing.T) {
-	m := NewMemoryMetaStore()
-	const acct = "000000000000"
-	const policy = `{"Version":"2012-10-17"}`
-
-	_, err := m.GetRepoPolicy(context.Background(), acct, "team/app")
-	assert.ErrorIs(t, err, ErrNotFound)
-	_, err = m.DeleteRepoPolicy(context.Background(), acct, "team/app")
-	assert.ErrorIs(t, err, ErrNotFound)
-
-	require.NoError(t, m.PutRepoPolicy(context.Background(), acct, "team/app", []byte(policy)))
-	got, err := m.GetRepoPolicy(context.Background(), acct, "team/app")
-	require.NoError(t, err)
-	assert.Equal(t, policy, string(got))
-
-	deleted, err := m.DeleteRepoPolicy(context.Background(), acct, "team/app")
-	require.NoError(t, err)
-	assert.Equal(t, policy, string(deleted))
-	_, err = m.GetRepoPolicy(context.Background(), acct, "team/app")
-	assert.ErrorIs(t, err, ErrNotFound)
-}
-
-func TestMemoryMetaStore_LifecyclePolicy(t *testing.T) {
-	m := NewMemoryMetaStore()
-	const acct = "000000000000"
-	const policy = `{"rules":[]}`
-
-	_, err := m.GetLifecyclePolicy(context.Background(), acct, "team/app")
-	assert.ErrorIs(t, err, ErrNotFound)
-	_, err = m.DeleteLifecyclePolicy(context.Background(), acct, "team/app")
-	assert.ErrorIs(t, err, ErrNotFound)
-
-	require.NoError(t, m.PutLifecyclePolicy(context.Background(), acct, "team/app", []byte(policy)))
-	got, err := m.GetLifecyclePolicy(context.Background(), acct, "team/app")
-	require.NoError(t, err)
-	assert.Equal(t, policy, string(got))
-
-	deleted, err := m.DeleteLifecyclePolicy(context.Background(), acct, "team/app")
-	require.NoError(t, err)
-	assert.Equal(t, policy, string(deleted))
-	_, err = m.GetLifecyclePolicy(context.Background(), acct, "team/app")
-	assert.ErrorIs(t, err, ErrNotFound)
-}
-
 func TestManifestMeta_DecodesRecordWithoutReferrerFields(t *testing.T) {
 	var m ManifestMeta
 	require.NoError(t, json.Unmarshal([]byte(`{"digest":"sha256:ccc","mediaType":"application/vnd.oci.image.manifest.v1+json","size":10,"pushedAt":"2026-01-01T00:00:00Z","childDigests":["sha256:ddd"]}`), &m))
@@ -232,3 +189,9 @@ func TestManifestMeta_DecodesRecordWithoutReferrerFields(t *testing.T) {
 	assert.Empty(t, m.SubjectDigest)
 	assert.Empty(t, m.ArtifactType)
 }
+
+func newTestKVMetaStore(t *testing.T) *KVMetaStore {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	return NewKVMetaStore(js)
+}

spinifex/domains/ecr/referrers_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"testing"
 
-	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -30,11 +29,7 @@ func referrerMeta(digest, subject string) ManifestMeta {
 // its own referrers state.
 func referrerStores(t *testing.T) map[string]MetaStore {
 	t.Helper()
-	_, nc, _ := testutil.StartTestJetStream(t)
+	return map[string]MetaStore{"kv": newTestKVMetaStore(t)}
-	return map[string]MetaStore{
-		"memory": NewMemoryMetaStore(),
-		"kv":     NewKVMetaStore(testutil.NewJetStream(t, nc)),
-	}
 }
 
 func TestListReferrers(t *testing.T) {

spinifex/domains/ecr/registry/lifecycle_sweep_test.go

@@ -50,7 +50,7 @@ func putPolicy(t *testing.T, reg *Registry, account, repo, policy string) {
 const untaggedExpirePolicy = `{"rules":[{"rulePriority":1,"selection":{"tagStatus":"untagged","countType":"sinceImagePushed","countUnit":"days","countNumber":7},"action":{"type":"expire"}}]}`
 
 func TestSweepRepo_SinceImagePushed(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:olduntagged", now.AddDate(0, 0, -10))
@@ -65,7 +65,7 @@ func TestSweepRepo_SinceImagePushed(t *testing.T) {
 }
 
 func TestSweepRepo_ImageCountMoreThan(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:a", now.AddDate(0, 0, -1), "a")
@@ -81,7 +81,7 @@ func TestSweepRepo_ImageCountMoreThan(t *testing.T) {
 }
 
 func TestSweepRepo_NoPolicy(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:x", now.AddDate(0, 0, -100))
@@ -92,7 +92,7 @@ func TestSweepRepo_NoPolicy(t *testing.T) {
 }
 
 func TestSweepRepo_Idempotent(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:olduntagged", now.AddDate(0, 0, -10))
@@ -105,7 +105,7 @@ func TestSweepRepo_Idempotent(t *testing.T) {
 }
 
 func TestSweepOnce_MultiAccount(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	now := time.Now().UTC()
 	const acct2 = "000000000001"
 	policy := `{"rules":[{"rulePriority":1,"selection":{"tagStatus":"any","countType":"imageCountMoreThan","countNumber":1},"action":{"type":"expire"}}]}`
@@ -125,7 +125,7 @@ func TestSweepOnce_MultiAccount(t *testing.T) {
 }
 
 func TestSweepOnce_AccountsError(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	s := NewLifecycleSweeper(reg, func() ([]string, error) {
 		return nil, assert.AnError
 	}, time.Hour)
@@ -133,7 +133,7 @@ func TestSweepOnce_AccountsError(t *testing.T) {
 }
 
 func TestNewLifecycleSweeper_DefaultInterval(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	s := NewLifecycleSweeper(reg, func() ([]string, error) { return nil, nil }, 0)
 	assert.Equal(t, DefaultLifecycleSweepInterval, s.interval)
 
@@ -147,7 +147,7 @@ func TestNewLifecycleSweeper_DefaultInterval(t *testing.T) {
 // TestSweeperRun drives the ticker loop: with a tiny interval and an expiring
 // image, the first tick must delete it; cancelling the context stops Run.
 func TestSweeperRun(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:olduntagged", now.AddDate(0, 0, -10))
@@ -184,7 +184,7 @@ func TestSweeperRun(t *testing.T) {
 // TestSweepRepo_InvalidStoredPolicy exercises the evaluate-error branch: a
 // malformed stored policy is logged and skipped, deleting nothing.
 func TestSweepRepo_InvalidStoredPolicy(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:x", now.AddDate(0, 0, -100))
@@ -198,7 +198,7 @@ func TestSweepRepo_InvalidStoredPolicy(t *testing.T) {
 // TestSweepRepo_PolicyWithoutRepo exercises the ListImages-not-found branch: a
 // policy stored for a repo with no manifests yields nothing to expire.
 func TestSweepRepo_PolicyWithoutRepo(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	now := time.Now().UTC()
 	putPolicy(t, reg, testAccount, "ghost", untaggedExpirePolicy)
 

spinifex/domains/ecr/registry/registry_ensurebucket_test.go

@@ -7,7 +7,6 @@ import (
 	"net/http"
 	"testing"
 
-	"github.com/mulgadc/spinifex/spinifex/domains/ecr"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -32,7 +31,7 @@ func (c *countingStore) EnsureBucket(ctx context.Context, bucket string) error {
 
 func TestRegistry_EnsureBucketFailureReturns500(t *testing.T) {
 	store := &countingStore{MemoryObjectStore: objectstore.NewMemoryObjectStore(), ensureErr: errors.New("predastore down")}
-	reg := NewRegistry(store, ecr.NewMemoryMetaStore(), testAccount)
+	reg := NewRegistry(store, newTestMetaStore(t), testAccount)
 
 	w := do(reg, http.MethodGet, "/v2/myrepo/tags/list", nil, nil)
 	assert.Equal(t, http.StatusInternalServerError, w.Code)
@@ -40,7 +39,7 @@ func TestRegistry_EnsureBucketFailureReturns500(t *testing.T) {
 
 func TestRegistry_EnsureBucketCachedAfterSuccess(t *testing.T) {
 	store := &countingStore{MemoryObjectStore: objectstore.NewMemoryObjectStore()}
-	reg := NewRegistry(store, ecr.NewMemoryMetaStore(), testAccount)
+	reg := NewRegistry(store, newTestMetaStore(t), testAccount)
 
 	do(reg, http.MethodGet, "/v2/myrepo/tags/list", nil, nil)
 	do(reg, http.MethodGet, "/v2/other/tags/list", nil, nil)

spinifex/domains/ecr/registry/registry_image_test.go

@@ -44,7 +44,7 @@ func storeLayersManifest(t *testing.T, reg *Registry, repo, tag string, layers .
 }
 
 func TestDeleteImage_ReclaimsBlobsKeepsShared(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	shared := pushBlob(t, reg, repo, []byte("shared-layer-bytes"))
 	exclusive := pushBlob(t, reg, repo, []byte("exclusive-layer-bytes"))
@@ -85,7 +85,7 @@ func seedImage(t *testing.T, reg *Registry, repo, tag string) ([]byte, string) {
 }
 
 func TestStoreAndGetManifest_RoundTrip(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	manifest, digest := seedImage(t, reg, "team/app", "v1")
 
 	// Fetch by tag.
@@ -114,7 +114,7 @@ func TestStoreAndGetManifest_RoundTrip(t *testing.T) {
 }
 
 func TestListImages_TaggedAndUntagged(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	seedImage(t, reg, repo, "v1")
 	seedImage(t, reg, repo, "v2")
@@ -176,7 +176,7 @@ func TestListImages_KVDigestForm(t *testing.T) {
 }
 
 func TestDeleteImage_ByTagAndByDigest(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	_, digest := seedImage(t, reg, repo, "v1")
 	seedImage(t, reg, repo, "v2") // shares nothing; distinct digest
@@ -205,7 +205,7 @@ func TestDeleteImage_ByTagAndByDigest(t *testing.T) {
 }
 
 func TestDeleteImage_ByDigestClearsTags(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	_, digest := seedImage(t, reg, repo, "v1")
 	// Add a second tag pointing at the same digest.
@@ -221,7 +221,7 @@ func TestDeleteImage_ByDigestClearsTags(t *testing.T) {
 }
 
 func TestStoreManifest_ImmutableTag(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	manA := fmt.Appendf(nil, `{"schemaVersion":2,"mediaType":"%s","layers":[{"digest":"%s"}]}`,
 		mediaTypeDockerManifest, pushBlob(t, reg, repo, []byte("layer-a")))
@@ -254,7 +254,7 @@ func TestStoreManifest_ImmutableTag(t *testing.T) {
 }
 
 func TestStoreManifest_BadDigestReference(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	cfgDg := pushBlob(t, reg, repo, []byte("cfg"))
 	manifest := fmt.Appendf(nil,

spinifex/domains/ecr/registry/registry_referrer_test.go

@@ -38,7 +38,7 @@ func storedMeta(t *testing.T, reg *Registry, body string) (subject, artifactType
 }
 
 func TestPutManifest_SubjectRecordedAndAdvertised(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	cfg := pushBlob(t, reg, referrerRepo, []byte("cfg"))
 
 	tests := []struct {
@@ -65,7 +65,7 @@ func TestPutManifest_SubjectRecordedAndAdvertised(t *testing.T) {
 }
 
 func TestPutManifest_IndexSubject(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, referrerRepo)
 
 	body := fmt.Sprintf(`{"schemaVersion":2,"mediaType":%q,"manifests":[],"subject":{"mediaType":%q,"digest":%q,"size":10}}`,
@@ -81,7 +81,7 @@ func TestPutManifest_IndexSubject(t *testing.T) {
 }
 
 func TestPutManifest_NoSubjectLeavesFieldsEmpty(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	cfg := pushBlob(t, reg, referrerRepo, []byte("cfg"))
 	body := fmt.Sprintf(`{"schemaVersion":2,"mediaType":%q,"artifactType":%q,"config":{"mediaType":%q,"digest":%q,"size":3},"layers":[]}`,
 		mediaTypeOCIManifest, sigArtifactType, sigConfigType, cfg)
@@ -97,7 +97,7 @@ func TestPutManifest_NoSubjectLeavesFieldsEmpty(t *testing.T) {
 }
 
 func TestPutManifest_MalformedSubjectDigestRejected(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	cfg := pushBlob(t, reg, referrerRepo, []byte("cfg"))
 	body := imageWithSubject(cfg, "sha256:nothex", "")
 
@@ -111,7 +111,7 @@ func TestPutManifest_MalformedSubjectDigestRejected(t *testing.T) {
 }
 
 func TestPutManifest_IndexWithContentTypeParameterGetsNoConfigArtifactType(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, referrerRepo)
 
 	body := fmt.Sprintf(`{"schemaVersion":2,"mediaType":%q,"manifests":[],"subject":{"mediaType":%q,"digest":%q,"size":10}}`,
@@ -128,7 +128,7 @@ func TestPutManifest_IndexWithContentTypeParameterGetsNoConfigArtifactType(t *te
 }
 
 func TestDeleteReferrerManifest_RemovesItFromTheSubjectListing(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	cfg := pushBlob(t, reg, referrerRepo, []byte("cfg"))
 	body := imageWithSubject(cfg, absentSubjectDgt, sigArtifactType)
 	dg := digestOf([]byte(body))
@@ -183,7 +183,7 @@ func getReferrersBody(t *testing.T, reg *Registry, path string) (*httptest.Respo
 }
 
 func TestGetReferrers_ListsReferrersAsAnOCIIndex(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	dgA := pushReferrer(t, reg, referrerRepo, absentSubjectDgt, sigArtifactType, "cfgA")
 	dgB := pushReferrer(t, reg, referrerRepo, absentSubjectDgt, "application/vnd.example.sbom.v1", "cfgB")
 
@@ -202,7 +202,7 @@ func TestGetReferrers_ListsReferrersAsAnOCIIndex(t *testing.T) {
 }
 
 func TestGetReferrers_NoReferrersIsAnEmptyIndexNotAnError(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, referrerRepo)
 
 	w := do(reg, http.MethodGet, "/v2/"+referrerRepo+"/referrers/"+absentSubjectDgt, nil, nil)
@@ -212,7 +212,7 @@ func TestGetReferrers_NoReferrersIsAnEmptyIndexNotAnError(t *testing.T) {
 }
 
 func TestGetReferrers_ArtifactTypeFilter(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	dgSig := pushReferrer(t, reg, referrerRepo, absentSubjectDgt, sigArtifactType, "cfgA")
 	pushReferrer(t, reg, referrerRepo, absentSubjectDgt, "application/vnd.example.sbom.v1", "cfgB")
 
@@ -230,7 +230,7 @@ func TestGetReferrers_ArtifactTypeFilter(t *testing.T) {
 }
 
 func TestGetReferrers_MalformedDigest(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, referrerRepo)
 
 	w := do(reg, http.MethodGet, "/v2/"+referrerRepo+"/referrers/not-a-digest", nil, nil)
@@ -239,14 +239,14 @@ func TestGetReferrers_MalformedDigest(t *testing.T) {
 }
 
 func TestGetReferrers_UnknownRepository(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	w := do(reg, http.MethodGet, "/v2/team/ghost/referrers/"+absentSubjectDgt, nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 	assertCode(t, w, "NAME_UNKNOWN")
 }
 
 func TestGetReferrers_DoesNotCrossRepositories(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	pushReferrer(t, reg, referrerRepo, absentSubjectDgt, sigArtifactType, "cfgA")
 	seedRepo(t, reg, "team/other")
 
@@ -256,7 +256,7 @@ func TestGetReferrers_DoesNotCrossRepositories(t *testing.T) {
 }
 
 func TestReferrersRoute_OnlyGET(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, referrerRepo)
 	for _, method := range []string{http.MethodPut, http.MethodDelete, http.MethodPost, http.MethodPatch} {
 		w := do(reg, method, "/v2/"+referrerRepo+"/referrers/"+absentSubjectDgt, []byte("{}"), nil)
@@ -272,7 +272,7 @@ func (failingReferrerStore) ListReferrers(context.Context, string, string, strin
 }
 
 func TestGetReferrers_MetadataFailureIsA5xxNotA404(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, referrerRepo)
 	reg.Meta = failingReferrerStore{reg.Meta}
 
@@ -310,7 +310,7 @@ func referrerDigests(t *testing.T, reg *Registry, repo, subject string) []string
 }
 
 func TestDeleteManifest_SubjectDeleteRemovesItsReferrers(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	sig := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	sbom := pushReferrer(t, reg, referrerRepo, subject, "application/vnd.example.sbom.v1", "sbom")
@@ -326,7 +326,7 @@ func TestDeleteManifest_SubjectDeleteRemovesItsReferrers(t *testing.T) {
 }
 
 func TestDeleteManifest_NestedReferrersAreRemoved(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	sig := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	sigOfSig := pushReferrer(t, reg, referrerRepo, sig, sigArtifactType, "sig-of-sig")
@@ -341,7 +341,7 @@ func TestDeleteManifest_NestedReferrersAreRemoved(t *testing.T) {
 }
 
 func TestDeleteManifest_ReferrerDeleteLeavesTheSubject(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	sig := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	other := pushReferrer(t, reg, referrerRepo, subject, "application/vnd.example.sbom.v1", "sbom")
@@ -355,7 +355,7 @@ func TestDeleteManifest_ReferrerDeleteLeavesTheSubject(t *testing.T) {
 }
 
 func TestDeleteManifest_TagDeleteDoesNotCascade(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	sig := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	require.NoError(t, reg.Meta.PutTag(context.Background(), testAccount, referrerRepo, "v1", subject))
@@ -368,7 +368,7 @@ func TestDeleteManifest_TagDeleteDoesNotCascade(t *testing.T) {
 }
 
 func TestDeleteManifest_CascadeStaysInsideTheRepository(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	elsewhere := pushReferrer(t, reg, "team/other", subject, sigArtifactType, "sig-other")
@@ -394,7 +394,7 @@ func (s failingDeleteStore) DeleteManifestMeta(ctx context.Context, account, rep
 }
 
 func TestDeleteManifest_ReferrerFailureDoesNotFailTheSubjectDelete(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	stuck := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "stuck")
 	fine := pushReferrer(t, reg, referrerRepo, subject, "application/vnd.example.sbom.v1", "fine")

spinifex/domains/ecr/registry/registry_test.go

@@ -2,6 +2,7 @@ package ecrregistry
 
 import (
 	"context"
+	"github.com/mulgadc/spinifex/internal/testkit"
 
 	"crypto/sha256"
 	"encoding/hex"
@@ -21,8 +22,15 @@ import (
 
 const testAccount = "000000000000"
 
-func newTestRegistry() *Registry {
+func newTestRegistry(t *testing.T) *Registry {
-	return NewRegistry(objectstore.NewMemoryObjectStore(), ecr.NewMemoryMetaStore(), testAccount)
+	t.Helper()
+	return NewRegistry(objectstore.NewMemoryObjectStore(), newTestMetaStore(t), testAccount)
+}
+
+func newTestMetaStore(t *testing.T) *ecr.KVMetaStore {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	return ecr.NewKVMetaStore(js)
 }
 
 func digestOf(b []byte) string {
@@ -69,7 +77,7 @@ func pushBlob(t *testing.T, reg *Registry, repo string, data []byte) string {
 }
 
 func TestBlobUpload_Monolithic_And_Head_Get(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	data := []byte("hello layer bytes")
 	dg := pushBlob(t, reg, "team/app", data)
 
@@ -84,20 +92,20 @@ func TestBlobUpload_Monolithic_And_Head_Get(t *testing.T) {
 }
 
 func TestBlobHead_NotFound(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	w := do(reg, http.MethodHead, "/v2/team/app/blobs/"+digestOf([]byte("absent")), nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }
 
 func TestBlobGet_MalformedDigest(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	w := do(reg, http.MethodGet, "/v2/team/app/blobs/sha256:zzz", nil, nil)
 	assert.Equal(t, http.StatusBadRequest, w.Code)
 	assertCode(t, w, "DIGEST_INVALID")
 }
 
 func TestBlobUpload_Chunked(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, "r/repo")
 	c1 := []byte("first-chunk-")
 	c2 := []byte("second-chunk")
@@ -124,7 +132,7 @@ func TestBlobUpload_Chunked(t *testing.T) {
 }
 
 func TestBlobUpload_DigestMismatch(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, "r/repo")
 	w := do(reg, http.MethodPost, "/v2/r/repo/blobs/uploads/", nil, nil)
 	loc := w.Header().Get("Location")
@@ -134,7 +142,7 @@ func TestBlobUpload_DigestMismatch(t *testing.T) {
 }
 
 func TestBlobUpload_Cancel(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	seedRepo(t, reg, "r/repo")
 	w := do(reg, http.MethodPost, "/v2/r/repo/blobs/uploads/", nil, nil)
 	loc := w.Header().Get("Location")
@@ -161,7 +169,7 @@ func putManifestReferencing(t *testing.T, reg *Registry, repo, digest string) {
 }
 
 func TestBlobMount_HitAndMiss(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	data := []byte("shared-layer")
 	dg := pushBlob(t, reg, "team/source", data)
 	putManifestReferencing(t, reg, "team/source", dg)
@@ -185,7 +193,7 @@ func TestBlobMount_HitAndMiss(t *testing.T) {
 // as a plain upload start, matching the authorization middleware's
 // classification of the same request.
 func TestBlobMount_MissingFromFallsThroughToUpload(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	data := []byte("shared-layer")
 	dg := pushBlob(t, reg, "team/source", data)
 	putManifestReferencing(t, reg, "team/source", dg)
@@ -200,7 +208,7 @@ func TestBlobMount_MissingFromFallsThroughToUpload(t *testing.T) {
 // named source repository is rejected as a mount — proving digest existence
 // alone is not enough to mount cross-repo.
 func TestBlobMount_UnreferencedDigestFallsThrough(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	unreferenced := pushBlob(t, reg, "team/other", []byte("not-in-source"))
 	seedRepo(t, reg, "team/source")
 	seedRepo(t, reg, "team/dest")
@@ -214,7 +222,7 @@ func TestBlobMount_UnreferencedDigestFallsThrough(t *testing.T) {
 // repository that does not exist never mounts, and never distinguishes
 // itself from an unreferenced-digest miss (both fall through identically).
 func TestBlobMount_NonexistentSourceFallsThrough(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	dg := pushBlob(t, reg, "team/elsewhere", []byte("some-bytes"))
 	seedRepo(t, reg, "team/dest")
 
@@ -224,7 +232,7 @@ func TestBlobMount_NonexistentSourceFallsThrough(t *testing.T) {
 }
 
 func TestManifest_PutGetHead_ImageManifest(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 
 	cfg := []byte(`{"config":true}`)
@@ -255,7 +263,7 @@ func TestManifest_PutGetHead_ImageManifest(t *testing.T) {
 }
 
 func TestManifest_Put_MissingBlob(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	seedRepo(t, reg, repo)
 	manifest := fmt.Appendf(nil,
@@ -268,7 +276,7 @@ func TestManifest_Put_MissingBlob(t *testing.T) {
 }
 
 func TestManifest_Get_AcceptMismatch(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("c"))
 	manifest := fmt.Appendf(nil,
@@ -285,7 +293,7 @@ func TestManifest_Get_AcceptMismatch(t *testing.T) {
 // acceptable manifest media type as its own Accept header line rather than one
 // comma-joined value. Negotiation must consider every line, not just the first.
 func TestManifest_Get_MultipleAcceptHeaders(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("c"))
 	manifest := fmt.Appendf(nil,
@@ -305,7 +313,7 @@ func TestManifest_Get_MultipleAcceptHeaders(t *testing.T) {
 }
 
 func TestManifest_Get_Unknown(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	w := do(reg, http.MethodGet, "/v2/team/app/manifests/missing", nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 	assertCode(t, w, "MANIFEST_UNKNOWN")
@@ -314,7 +322,7 @@ func TestManifest_Get_Unknown(t *testing.T) {
 // TestStartUpload_RepoNotCreated asserts a blob upload to a repository that was
 // never created is rejected up front — ECR does not auto-create on push.
 func TestStartUpload_RepoNotCreated(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	w := do(reg, http.MethodPost, "/v2/team/ghost/blobs/uploads/", nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 	assertCode(t, w, "NAME_UNKNOWN")
@@ -323,7 +331,7 @@ func TestStartUpload_RepoNotCreated(t *testing.T) {
 // TestPutManifest_RepoNotCreated asserts a manifest PUT to an uncreated
 // repository is rejected with NAME_UNKNOWN.
 func TestPutManifest_RepoNotCreated(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	manifest := fmt.Appendf(nil, `{"mediaType":"%s","layers":[]}`, mediaTypeDockerManifest)
 	w := do(reg, http.MethodPut, "/v2/team/ghost/manifests/v1", manifest,
 		map[string]string{"Content-Type": mediaTypeDockerManifest})
@@ -332,7 +340,7 @@ func TestPutManifest_RepoNotCreated(t *testing.T) {
 }
 
 func TestManifest_ImageIndex_ChildValidation(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 
 	// Push a child image manifest first.
@@ -372,7 +380,7 @@ func TestManifest_ImageIndex_ChildValidation(t *testing.T) {
 }
 
 func TestManifest_Delete_Tag(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("c"))
 	manifest := fmt.Appendf(nil,
@@ -390,7 +398,7 @@ func TestManifest_Delete_Tag(t *testing.T) {
 // TestManifest_Delete_ByDigest proves delete-by-digest is no longer a no-op: it
 // removes the manifest record + object, and a subsequent GET is a 404.
 func TestManifest_Delete_ByDigest(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("cfg-bytes"))
 	manifest := fmt.Appendf(nil,
@@ -412,7 +420,7 @@ func TestManifest_Delete_ByDigest(t *testing.T) {
 }
 
 func TestTagsList(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("c"))
 	manifest := fmt.Appendf(nil,
@@ -434,13 +442,13 @@ func TestTagsList(t *testing.T) {
 }
 
 func TestTagsList_UnknownRepo(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	w := do(reg, http.MethodGet, "/v2/team/ghost/tags/list", nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }
 
 func TestCatalog(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	pushBlob(t, reg, "team/a", []byte("a"))
 	pushBlob(t, reg, "team/b", []byte("b"))
 
@@ -454,21 +462,21 @@ func TestCatalog(t *testing.T) {
 }
 
 func TestDispatch_InvalidRepoName(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	w := do(reg, http.MethodGet, "/v2/Team/App/tags/list", nil, nil)
 	assert.Equal(t, http.StatusBadRequest, w.Code)
 	assertCode(t, w, "NAME_INVALID")
 }
 
 func TestDispatch_RepeatedSeparatorRepoName(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	dg := pushBlob(t, reg, "team/a__b--c", []byte("layer"))
 	w := do(reg, http.MethodHead, "/v2/team/a__b--c/blobs/"+dg, nil, nil)
 	assert.Equal(t, http.StatusOK, w.Code)
 }
 
 func TestDispatch_UnknownPath(t *testing.T) {
-	reg := newTestRegistry()
+	reg := newTestRegistry(t)
 	w := do(reg, http.MethodGet, "/v2/team/app/frobnicate", nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }

spinifex/domains/ecr/registry/registry_upload_test.go

@@ -35,7 +35,7 @@ func (s *blobPutFailStore) PutObject(ctx context.Context, input *s3.PutObjectInp
 func TestFinishUpload_StoreFailure_CleansUp(t *testing.T) {
 	mem := objectstore.NewMemoryObjectStore()
 	store := &blobPutFailStore{MemoryObjectStore: mem}
-	meta := ecr.NewMemoryMetaStore()
+	meta := newTestMetaStore(t)
 	reg := NewRegistry(store, meta, testAccount)
 	seedRepo(t, reg, "team/app")
 
@@ -72,7 +72,7 @@ func awsString(s string) *string { return &s }
 // updates issued from the same observed revision cannot both commit. One wins,
 // the other gets ErrConflict — so a racing PATCH is rejected, never merged.
 func TestUploadCAS_SameRevisionSingleWinner(t *testing.T) {
-	meta := ecr.NewMemoryMetaStore()
+	meta := newTestMetaStore(t)
 	rev, err := meta.PutUpload(context.Background(), testAccount, "u-race", ecr.UploadState{RepoName: "r"})
 	require.NoError(t, err)
 
@@ -116,7 +116,7 @@ func TestUploadCAS_SameRevisionSingleWinner(t *testing.T) {
 // hash state and the recorded byte object never desynchronise.
 func TestPatchUpload_ConcurrentNoCorruption(t *testing.T) {
 	for iter := range 25 {
-		reg := newTestRegistry()
+		reg := newTestRegistry(t)
 		repo := "team/race"
 		seedRepo(t, reg, repo)
 

spinifex/domains/ecr/service_impl.go

@@ -16,8 +16,7 @@ type MetaServiceImpl struct {
 
 var _ MetaService = (*MetaServiceImpl)(nil)
 
-// NewMetaServiceImpl builds a MetaService over an explicit MetaStore (used by
+// NewMetaServiceImpl builds a MetaService over an explicit MetaStore.
-// daemon-side tests with a MemoryMetaStore).
 func NewMetaServiceImpl(store MetaStore) *MetaServiceImpl {
 	return &MetaServiceImpl{store: store}
 }

spinifex/domains/ecr/service_test.go

@@ -21,7 +21,7 @@ const svcTestAccount = "000000000000"
 // ErrNotFound/ErrConflict into the response flags rather than transport errors,
 // so they survive the AWS-error-code-only NATS reply.
 func TestMetaServiceImpl_FlagMapping(t *testing.T) {
-	svc := NewMetaServiceImpl(NewMemoryMetaStore())
+	svc := NewMetaServiceImpl(newTestKVMetaStore(t))
 
 	// Absent repo -> Found:false, nil error.
 	dr, err := svc.RepoDescribe(context.Background(), &RepoDescribeRequest{Repo: "ghost"}, svcTestAccount)

spinifex/domains/lambda/awsapi/get.go

@@ -23,42 +23,81 @@ func notFound(region, accountID, name string) error {
 		arn.FormatLambdaFunction(region, accountID, name))
 }
 
-// getFunction implements GetFunction for the $LATEST version.
+// lookup resolves the requested $LATEST function's record; query is the
-func (d Deps) getFunction(ctx context.Context, c *call) (reply, error) {
+// Qualifier query parameter, empty for an action that has none.
+func (d Deps) lookup(ctx context.Context, c *call, query string) (function.Record, uint64, error) {
 	name, q, err := resolveFunctionName(c.Region, c.AccountID, c.Params[0])
 	if err != nil {
-		return reply{}, err
+		return function.Record{}, 0, err
 	}
-	if query := c.Query.Get("Qualifier"); query != "" {
+	if query != "" {
 		if q != "" && q != query {
-			return reply{}, invalid("The derived qualifier from the function name does not match the specified qualifier.")
+			return function.Record{}, 0, invalid("The derived qualifier from the function name does not match the specified qualifier.")
 		}
 		q = query
 	}
 	if q != "" && q != "$LATEST" {
-		return reply{}, notFound(c.Region, c.AccountID, name+":"+q)
+		return function.Record{}, 0, notFound(c.Region, c.AccountID, name+":"+q)
 	}
 	rec, revision, err := d.Meta.Get(ctx, c.AccountID, c.Region, name)
 	if errors.Is(err, function.ErrNotFound) {
-		return reply{}, notFound(c.Region, c.AccountID, name)
+		return function.Record{}, 0, notFound(c.Region, c.AccountID, name)
 	}
+	return rec, revision, err
+}
+
+// loadConfiguration resolves the requested $LATEST function and renders its
+// configuration; it is a metadata read and never touches the code store.
+func (d Deps) loadConfiguration(ctx context.Context, c *call) (function.Record, *lambda.FunctionConfiguration, error) {
+	rec, revision, err := d.lookup(ctx, c, c.Query.Get("Qualifier"))
 	if err != nil {
-		return reply{}, err
+		return function.Record{}, nil, err
 	}
 
 	var env map[string]string
 	if rec.EnvCipher != "" {
 		if d.AEAD == nil {
-			return reply{}, serviceError("environment decryption is unavailable")
+			return function.Record{}, nil, serviceError("environment decryption is unavailable")
 		}
 		if env, err = function.OpenEnv(d.AEAD, rec.AccountID, rec.FunctionID, rec.EnvCipher); err != nil {
 			slog.ErrorContext(ctx, "Lambda: open environment failed", "function_id", rec.FunctionID, "err", err)
-			return reply{}, serviceError("could not read the function environment")
+			return function.Record{}, nil, serviceError("could not read the function environment")
 		}
 	}
 	cfg, err := configuration(rec, revision, env)
 	if err != nil {
-		return reply{}, serviceError("could not build the function configuration")
+		return function.Record{}, nil, serviceError("could not build the function configuration")
+	}
+	return rec, cfg, nil
+}
+
+// getFunctionConfiguration implements GetFunctionConfiguration for $LATEST.
+func (d Deps) getFunctionConfiguration(ctx context.Context, c *call) (reply, error) {
+	_, cfg, err := d.loadConfiguration(ctx, c)
+	if err != nil {
+		return reply{}, err
+	}
+	return reply{Status: http.StatusOK, Body: cfg}, nil
+}
+
+// noCodeSigning is AWS's body for a function without a code signing config:
+// both members present and null, which the SDK shapes cannot encode.
+var noCodeSigning = rawJSON(`{"CodeSigningConfigArn":null,"FunctionName":null}`)
+
+// getFunctionCodeSigningConfig implements GetFunctionCodeSigningConfig; code
+// signing is unsupported, so it only proves the function exists.
+func (d Deps) getFunctionCodeSigningConfig(ctx context.Context, c *call) (reply, error) {
+	if _, _, err := d.lookup(ctx, c, ""); err != nil {
+		return reply{}, err
+	}
+	return reply{Status: http.StatusOK, Body: noCodeSigning}, nil
+}
+
+// getFunction implements GetFunction for the $LATEST version.
+func (d Deps) getFunction(ctx context.Context, c *call) (reply, error) {
+	rec, cfg, err := d.loadConfiguration(ctx, c)
+	if err != nil {
+		return reply{}, err
 	}
 	out := &lambda.GetFunctionOutput{
 		Configuration: cfg,

spinifex/domains/lambda/awsapi/handler_test.go

@@ -662,8 +662,211 @@ func TestAuthorization(t *testing.T) {
 
 func TestRegistration_ResolvesActionsAndInventory(t *testing.T) {
 	reg := awsapi.NewRegistration(awsapi.Deps{})
-	assert.Equal(t, []string{"CreateFunction", "GetFunction"}, reg.Inventory.Registered)
+	assert.Equal(t, []string{"CreateFunction", "GetFunction", "GetFunctionCodeSigningConfig", "GetFunctionConfiguration"}, reg.Inventory.Registered)
+	assert.Equal(t, "GetFunctionCodeSigningConfig", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2020-06-30/functions/x/code-signing-config", nil)))
 	assert.Equal(t, "CreateFunction", reg.ResolveAction(httptest.NewRequest(http.MethodPost, "/2015-03-31/functions", nil)))
 	assert.Equal(t, "GetFunction", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x", nil)))
+	assert.Equal(t, "GetFunctionConfiguration", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x/configuration", nil)))
 	assert.Empty(t, reg.ResolveAction(httptest.NewRequest(http.MethodDelete, "/2015-03-31/functions/x", nil)))
 }
+
+func TestGetFunctionConfiguration_ReturnsTheSameConfigurationAsGetFunction(t *testing.T) {
+	e := newEnv(t)
+	_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
+		b["Environment"] = map[string]any{"Variables": map[string]string{"MODE": "prod"}}
+	}))
+	require.NoError(t, err)
+
+	rec, err := e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
+	require.NoError(t, err)
+	var full lambda.GetFunctionOutput
+	require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &full))
+
+	rec, err = e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, rec.Code)
+	assert.NotContains(t, rec.Body.String(), `"Code"`, "the configuration is the top-level body, not GetFunction's wrapper")
+	var cfg lambda.FunctionConfiguration
+	require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &cfg))
+	assert.Equal(t, *full.Configuration, cfg)
+	assert.Equal(t, "prod", *cfg.Environment.Variables["MODE"])
+
+	arn := "arn%3Aaws%3Alambda%3A" + gwtest.Region + "%3A" + gwtest.AccountID + "%3Afunction%3Ahello"
+	for _, path := range []string{
+		"/2015-03-31/functions/hello/configuration?Qualifier=%24LATEST",
+		"/2015-03-31/functions/" + arn + "/configuration",
+		"/2015-03-31/functions/" + arn + "%3A%24LATEST/configuration",
+	} {
+		rec, err := e.do(http.MethodGet, path, nil)
+		require.NoError(t, err, path)
+		assert.Equal(t, http.StatusOK, rec.Code, path)
+	}
+}
+
+func TestGetFunctionConfiguration_MissingAndUnsupportedMatchGetFunction(t *testing.T) {
+	e := newEnv(t)
+	_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
+	require.NoError(t, err)
+
+	for label, tc := range map[string]struct{ suffix, code string }{
+		"missing":         {"ghost", awserrors.ErrorResourceNotFoundException},
+		"version":         {"hello?Qualifier=3", awserrors.ErrorResourceNotFoundException},
+		"other account":   {"arn%3Aaws%3Alambda%3A" + gwtest.Region + "%3A999999999999%3Afunction%3Ahello", awserrors.ErrorResourceNotFoundException},
+		"qualifier clash": {"hello%3A%24LATEST?Qualifier=3", awserrors.ErrorInvalidParameterValueException},
+		"not lambda":      {"arn%3Aaws%3As3%3A%3A%3Abucket", awserrors.ErrorInvalidParameterValueException},
+	} {
+		t.Run(label, func(t *testing.T) {
+			path, query, _ := strings.Cut(tc.suffix, "?")
+			if query != "" {
+				query = "?" + query
+			}
+			_, getErr := e.do(http.MethodGet, "/2015-03-31/functions/"+path+query, nil)
+			requireCode(t, getErr, tc.code)
+			_, err := e.do(http.MethodGet, "/2015-03-31/functions/"+path+"/configuration"+query, nil)
+			requireCode(t, err, tc.code)
+			assert.Equal(t, getErr.Error(), err.Error())
+		})
+	}
+
+	e.h.Region = "us-west-2"
+	_, err = e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
+	requireCode(t, err, awserrors.ErrorResourceNotFoundException)
+}
+
+func TestGetFunctionConfiguration_DoesNotNeedTheCodeStore(t *testing.T) {
+	for name, codes := range map[string]awsapi.CodeStoreFunc{
+		"store outage": func(context.Context) (awsapi.CodeStore, error) { return nil, errors.New("code store down") },
+		"no store":     nil,
+	} {
+		t.Run(name, func(t *testing.T) {
+			e := newEnv(t)
+			_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
+			require.NoError(t, err)
+			e.deps.Codes = codes
+
+			rec, err := e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
+			require.NoError(t, err)
+			require.Equal(t, http.StatusOK, rec.Code)
+			var cfg lambda.FunctionConfiguration
+			require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &cfg))
+			assert.Equal(t, "hello", *cfg.FunctionName)
+
+			_, err = e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
+			requireCode(t, err, awserrors.ErrorServiceException)
+		})
+	}
+}
+
+func TestGetFunctionConfiguration_Authorization(t *testing.T) {
+	arn := "arn:aws:lambda:" + gwtest.Region + ":" + gwtest.AccountID + ":function:hello"
+	setup := func(t *testing.T, statements ...iamdomain.Statement) *env {
+		e := newEnv(t, statements...)
+		_, err := e.meta.Create(t.Context(), gwtest.AccountID, gwtest.Region, function.Record{
+			FunctionID: "f1", FunctionName: "hello", Role: roleARN, Runtime: "provided.al2023", Handler: "h",
+			Timeout: 3, MemorySize: 128, Code: function.Code{RevisionID: "r1"},
+		})
+		require.NoError(t, err)
+		return e
+	}
+	get := func(e *env) error {
+		_, err := e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
+		return err
+	}
+
+	t.Run("lambda:GetFunction alone is not enough", func(t *testing.T) {
+		gwtest.AssertDenied(t, get(setup(t, gwtest.Statement("Allow", "lambda:GetFunction", "*"))))
+	})
+	t.Run("a deny on the function ARN applies", func(t *testing.T) {
+		gwtest.AssertDenied(t, get(setup(t,
+			gwtest.Statement("Allow", "lambda:*", "*"),
+			gwtest.Statement("Deny", "lambda:GetFunctionConfiguration", arn))))
+	})
+	t.Run("an allow on another function ARN does not apply", func(t *testing.T) {
+		gwtest.AssertDenied(t, get(setup(t, gwtest.Statement("Allow", "lambda:GetFunctionConfiguration", arn+"-other"))))
+	})
+	t.Run("an allow on the function ARN passes", func(t *testing.T) {
+		require.NoError(t, get(setup(t, gwtest.Statement("Allow", "lambda:GetFunctionConfiguration", arn))))
+	})
+}
+
+func TestGetFunctionCodeSigningConfig_ExistingFunctionHasExplicitNulls(t *testing.T) {
+	e := newEnv(t)
+	_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
+	require.NoError(t, err)
+
+	arn := "arn%3Aaws%3Alambda%3A" + gwtest.Region + "%3A" + gwtest.AccountID + "%3Afunction%3Ahello"
+	for _, name := range []string{"hello", arn, gwtest.AccountID + "%3Afunction%3Ahello"} {
+		rec, err := e.do(http.MethodGet, "/2020-06-30/functions/"+name+"/code-signing-config", nil)
+		require.NoError(t, err, name)
+		require.Equal(t, http.StatusOK, rec.Code, name)
+		assert.Equal(t, "application/json", rec.Header().Get("Content-Type"))
+		assert.JSONEq(t, `{"CodeSigningConfigArn":null,"FunctionName":null}`, rec.Body.String(), name)
+	}
+}
+
+func TestGetFunctionCodeSigningConfig_MissingFunctionIsNotFound(t *testing.T) {
+	e := newEnv(t)
+	_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
+	require.NoError(t, err)
+
+	_, err = e.do(http.MethodGet, "/2020-06-30/functions/ghost/code-signing-config", nil)
+	requireCode(t, err, awserrors.ErrorResourceNotFoundException)
+	_, getErr := e.do(http.MethodGet, "/2015-03-31/functions/ghost", nil)
+	assert.Equal(t, getErr.Error(), err.Error())
+
+	e.h.Region = "us-west-2"
+	_, err = e.do(http.MethodGet, "/2020-06-30/functions/hello/code-signing-config", nil)
+	requireCode(t, err, awserrors.ErrorResourceNotFoundException)
+}
+
+func TestGetFunctionCodeSigningConfig_NeedsNeitherCodeStoreNorDecryption(t *testing.T) {
+	e := newEnv(t)
+	_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
+		b["Environment"] = map[string]any{"Variables": map[string]string{"MODE": "prod"}}
+	}))
+	require.NoError(t, err)
+	e.deps.Codes = func(context.Context) (awsapi.CodeStore, error) { return nil, errors.New("code store down") }
+	e.deps.AEAD = nil
+
+	rec, err := e.do(http.MethodGet, "/2020-06-30/functions/hello/code-signing-config", nil)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{"CodeSigningConfigArn":null,"FunctionName":null}`, rec.Body.String())
+	_, err = e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
+	requireCode(t, err, awserrors.ErrorServiceException)
+
+	e.deps.Codes = nil
+	_, err = e.do(http.MethodGet, "/2020-06-30/functions/hello/code-signing-config", nil)
+	require.NoError(t, err)
+}
+
+func TestGetFunctionCodeSigningConfig_Authorization(t *testing.T) {
+	arn := "arn:aws:lambda:" + gwtest.Region + ":" + gwtest.AccountID + ":function:hello"
+	setup := func(t *testing.T, statements ...iamdomain.Statement) *env {
+		e := newEnv(t, statements...)
+		_, err := e.meta.Create(t.Context(), gwtest.AccountID, gwtest.Region, function.Record{
+			FunctionID: "f1", FunctionName: "hello", Role: roleARN, Runtime: "provided.al2023", Handler: "h",
+			Timeout: 3, MemorySize: 128, Code: function.Code{RevisionID: "r1"},
+		})
+		require.NoError(t, err)
+		return e
+	}
+	get := func(e *env) error {
+		_, err := e.do(http.MethodGet, "/2020-06-30/functions/hello/code-signing-config", nil)
+		return err
+	}
+
+	t.Run("lambda:GetFunction alone is not enough", func(t *testing.T) {
+		gwtest.AssertDenied(t, get(setup(t, gwtest.Statement("Allow", "lambda:GetFunction", "*"))))
+	})
+	t.Run("a deny on the function ARN applies", func(t *testing.T) {
+		gwtest.AssertDenied(t, get(setup(t,
+			gwtest.Statement("Allow", "lambda:*", "*"),
+			gwtest.Statement("Deny", "lambda:GetFunctionCodeSigningConfig", arn))))
+	})
+	t.Run("an allow on another function ARN does not apply", func(t *testing.T) {
+		gwtest.AssertDenied(t, get(setup(t, gwtest.Statement("Allow", "lambda:GetFunctionCodeSigningConfig", arn+"-other"))))
+	})
+	t.Run("an allow on the function ARN passes", func(t *testing.T) {
+		require.NoError(t, get(setup(t, gwtest.Statement("Allow", "lambda:GetFunctionCodeSigningConfig", arn))))
+	})
+}

spinifex/domains/lambda/awsapi/registration.go

@@ -117,7 +117,7 @@ func (d Deps) serve(w http.ResponseWriter, inv dispatch.Invocation) error {
 		if decodeErr == nil {
 			resource = functionARN(inv.Region, inv.AccountID, aws.StringValue(c.Create.FunctionName))
 		}
-	case "GetFunction":
+	case "GetFunction", "GetFunctionConfiguration", "GetFunctionCodeSigningConfig":
 		if name, _, err := resolveFunctionName(inv.Region, inv.AccountID, params[0]); err == nil {
 			resource = functionARN(inv.Region, inv.AccountID, name)
 		}
@@ -133,7 +133,9 @@ func (d Deps) serve(w http.ResponseWriter, inv dispatch.Invocation) error {
 			return err
 		}
 	}
-	if d.Meta == nil || d.Codes == nil {
+	// Only GetFunction presigns code; the other reads are metadata only.
+	needsCodes := action != "GetFunctionConfiguration" && action != "GetFunctionCodeSigningConfig"
+	if d.Meta == nil || (d.Codes == nil && needsCodes) {
 		return errors.New(awserrors.ErrorServiceException)
 	}
 
@@ -180,7 +182,13 @@ func (d Deps) checkPassRole(inv dispatch.Invocation, roleARN string) error {
 }
 
 func writeJSON(w http.ResponseWriter, out reply) {
-	body, err := jsonutil.BuildJSON(out.Body)
+	var body []byte
+	var err error
+	if raw, ok := out.Body.(rawJSON); ok {
+		body = raw
+	} else {
+		body, err = jsonutil.BuildJSON(out.Body)
+	}
 	if err != nil {
 		slog.Error("Failed to marshal Lambda response", "err", err)
 		w.Header().Set("Content-Type", "application/json")

spinifex/domains/lambda/awsapi/routes.go

@@ -21,6 +21,10 @@ type call struct {
 	Create *lambda.CreateFunctionInput
 }
 
+// rawJSON is a response body written as is, for a shape the SDK encoder
+// cannot produce.
+type rawJSON []byte
+
 // reply is a handler's success: the HTTP status and the SDK shape to encode.
 type reply struct {
 	Status int
@@ -41,6 +45,14 @@ var routes = []route{
 		Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
 			return d.getFunction(ctx, c)
 		}},
+	{Method: "GET", Pattern: "/2015-03-31/functions/{FunctionName}/configuration", Action: "GetFunctionConfiguration",
+		Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
+			return d.getFunctionConfiguration(ctx, c)
+		}},
+	{Method: "GET", Pattern: "/2020-06-30/functions/{FunctionName}/code-signing-config", Action: "GetFunctionCodeSigningConfig",
+		Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
+			return d.getFunctionCodeSigningConfig(ctx, c)
+		}},
 }
 
 // actionNames returns the distinct actions in routes in stable order.

spinifex/operator/accountlifecycle/engine.go

@@ -35,6 +35,12 @@ type Engine struct {
 	// TERMINATING and before the first stage. It records evidence and gates nothing.
 	OnQuiesced func(ctx context.Context, quiescence operatorv1.IAMQuiescence)
 
+	// Cleanup runs the EC2 platform cleanup. Only Exercise uses it.
+	Cleanup PlatformCleaner
+
+	// OnExercised receives what a non-finalizing run did after quiescence.
+	OnExercised func(ctx context.Context, result operatorv1.ExerciseResult)
+
 	// ec2Inventory is EC2's read-only platform inventory. Only DryRun consults
 	// it, and only to log a comparison; no teardown or Inventory call does.
 	ec2Inventory ec2PlatformInventory
@@ -129,7 +135,7 @@ func (e *Engine) Teardown(ctx context.Context, req Request) (*Result, error) {
 		return nil, fmt.Errorf("quiesce account: %w", err)
 	}
 	if e.OnQuiesced != nil {
-		e.OnQuiesced(ctx, e.evaluateQuiescence(ctx, req, generation))
+		e.OnQuiesced(ctx, e.evaluateQuiescence(ctx, req, generation, false))
 	}
 	e.logger().InfoContext(ctx, "Account teardown started",
 		"accountID", account.AccountID, "accountName", account.AccountName, "force", req.Force)
@@ -338,9 +344,9 @@ func SortReapers(reapers []Reaper) {
 
 // evaluateQuiescence reads back what the quiesce wrote and, only if the fence
 // is confirmed, asks EC2 for its platform inventory within a short budget.
-func (e *Engine) evaluateQuiescence(ctx context.Context, req Request, generation uint64) operatorv1.IAMQuiescence {
+func (e *Engine) evaluateQuiescence(ctx context.Context, req Request, generation uint64, exercise bool) operatorv1.IAMQuiescence {
 	ev := QuiesceEvidence{
-		OperationID: req.OperationID, Fence: req.Fence, Generation: generation, EvaluatedAt: time.Now(),
+		OperationID: req.OperationID, Fence: req.Fence, Generation: generation, EvaluatedAt: time.Now(), Exercise: exercise,
 	}
 	ev.Account, ev.ReadbackErr = e.Accounts.GetAccount(req.AccountID)
 	if _, confirmed := fenceReadback(ev); confirmed && e.ec2Inventory != nil {

spinifex/operator/accountlifecycle/exercise.go

@@ -0,0 +1,207 @@
+package accountlifecycle
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
+	"github.com/nats-io/nats.go"
+)
+
+// PlatformCleaner is the EC2 platform cleanup the exercise runs.
+type PlatformCleaner = accountcleanup.PlatformCleaner
+
+// Exercise quiesces the account, cleans the EC2 platform kinds the evaluation
+// recorded as quiesced, then re-reads the owner inventory and holds it against
+// the cleanup. No legacy stage runs and the account record is never removed.
+func (e *Engine) Exercise(ctx context.Context, req Request) (*operatorv1.ExerciseResult, error) {
+	if _, err := e.checkDeletable(req.AccountID, req.AccountName); err != nil {
+		return nil, err
+	}
+	if req.Force {
+		return nil, errors.New("a non-finalizing run refuses force")
+	}
+	if e.Cleanup == nil || e.ec2Inventory == nil {
+		return nil, errors.New("exercise needs the platform cleanup and inventory")
+	}
+	generation, err := e.Accounts.QuiesceAccount(req.AccountID, req.OperationID, req.Fence)
+	if err != nil {
+		return nil, fmt.Errorf("quiesce account: %w", err)
+	}
+	quiescence := e.evaluateQuiescence(ctx, req, generation, true)
+	if e.OnQuiesced != nil {
+		e.OnQuiesced(ctx, quiescence)
+	}
+
+	result := &operatorv1.ExerciseResult{}
+	if len(quiescence.QuiescedScope) == 0 {
+		result.Skipped = operatorv1.ExerciseSkippedNoQuiescedKind
+	} else {
+		cleaned, err := e.Cleanup.CleanupPlatform(ctx, &accountcleanup.Request{
+			AccountID: req.AccountID, CorrelationRef: req.OperationID, Kinds: quiescence.QuiescedScope,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("platform cleanup: %w", err)
+		}
+		result.Cleanup = platformCleanup(cleaned)
+		result.Accounting = e.accountForRemainder(ctx, req, quiescence.QuiescedScope, result.Cleanup)
+	}
+	if e.OnExercised != nil {
+		e.OnExercised(ctx, *result)
+	}
+	return result, nil
+}
+
+// accountForRemainder re-reads the owner inventory after cleanup and holds it
+// against the cleanup result, per quiesced kind.
+func (e *Engine) accountForRemainder(ctx context.Context, req Request, kinds []string, cleanup *operatorv1.PlatformCleanup) []operatorv1.KindAccounting {
+	budget := ec2PlatformShadowBudget
+	if e.ec2PlatformShadowBudget > 0 {
+		budget = e.ec2PlatformShadowBudget
+	}
+	rereadCtx, cancel := context.WithTimeout(ctx, budget)
+	defer cancel()
+	reply := e.ec2Inventory.PlatformInventory(rereadCtx, req.AccountID, req.OperationID)
+	return AccountForRemainder(kinds, &reply, cleanup)
+}
+
+// AccountForRemainder marks a kind accounted only when its re-read was complete
+// and every record still there is named in the cleanup result with a reason.
+// A record the cleanup reported deleted that is back is unnamed.
+func AccountForRemainder(kinds []string, reread *ec2v1.PlatformInventoryReply, cleanup *operatorv1.PlatformCleanup) []operatorv1.KindAccounting {
+	out := make([]operatorv1.KindAccounting, 0, len(kinds))
+	for _, kind := range kinds {
+		entry := operatorv1.KindAccounting{Kind: kind, Status: operatorv1.AccountingAccounted}
+		status, complete := kindInventory(reread, kind)
+		if !complete {
+			entry.Status, entry.Reason = operatorv1.AccountingUnaccounted, operatorv1.AccountingReasonRereadIncomplete
+			out = append(out, entry)
+			continue
+		}
+		if status == ec2v1.PlatformStatusRecordsFound {
+			unnamed := false
+			for _, item := range reread.Items {
+				if item.Kind != kind {
+					continue
+				}
+				entry.Remaining++
+				unnamed = unnamed || !namedItem(cleanup, item.ResourceID, kind)
+			}
+			for _, count := range reread.Counts {
+				if count.Kind != kind {
+					continue
+				}
+				entry.Remaining += count.Count
+				unnamed = unnamed || !namedCount(cleanup, count)
+			}
+			if unnamed {
+				entry.Status, entry.Reason = operatorv1.AccountingUnaccounted, operatorv1.AccountingReasonUnnamedRemainder
+			}
+		}
+		out = append(out, entry)
+	}
+	return out
+}
+
+func namedItem(cleanup *operatorv1.PlatformCleanup, id, kind string) bool {
+	if cleanup == nil {
+		return false
+	}
+	for _, item := range cleanup.Items {
+		if item.ResourceID == id && item.Kind == kind && item.Disposition != accountcleanup.DispositionDeleted && item.Reason != "" {
+			return true
+		}
+	}
+	return false
+}
+
+func namedCount(cleanup *operatorv1.PlatformCleanup, want ec2v1.PlatformRecordCount) bool {
+	if cleanup == nil {
+		return false
+	}
+	for _, count := range cleanup.Counts {
+		if count.Kind == want.Kind && count.Class == want.Class && count.Disposition != accountcleanup.DispositionDeleted &&
+			count.Reason != "" && count.Count >= want.Count {
+			return true
+		}
+	}
+	return false
+}
+
+func platformCleanup(res *accountcleanup.Result) *operatorv1.PlatformCleanup {
+	out := &operatorv1.PlatformCleanup{Kinds: []operatorv1.CleanupKind{}}
+	for _, kind := range res.Kinds {
+		out.Kinds = append(out.Kinds, operatorv1.CleanupKind{Kind: kind.Kind, Status: kind.Status, IncompleteReasons: kind.IncompleteReasons})
+	}
+	for _, item := range res.Items {
+		out.Items = append(out.Items, operatorv1.CleanupItem{
+			ResourceID: item.ResourceID, Kind: item.Kind, Disposition: item.Disposition, Reason: item.Reason, MemberCount: item.MemberCount,
+		})
+	}
+	for _, count := range res.Counts {
+		out.Counts = append(out.Counts, operatorv1.CleanupCount{
+			Kind: count.Kind, Class: count.Class, Disposition: count.Disposition, Reason: count.Reason, Count: count.Count,
+		})
+	}
+	return out
+}
+
+// InstanceReferenceReader reads the key-pair and placement-group names the
+// account's non-terminated instances carry. It fails closed: a listing that is
+// not complete from every node is an error, never a partial answer.
+type InstanceReferenceReader struct {
+	nc      *nats.Conn
+	nodeIDs []string
+	budget  time.Duration
+}
+
+var _ accountcleanup.InstanceReferenceReader = (*InstanceReferenceReader)(nil)
+
+// NewInstanceReferenceReader builds the reader over the configured node set.
+func NewInstanceReferenceReader(nc *nats.Conn, nodeIDs []string) *InstanceReferenceReader {
+	return &InstanceReferenceReader{nc: nc, nodeIDs: nodeIDs, budget: ec2PlatformShadowBudget}
+}
+
+func (r *InstanceReferenceReader) AccountInstanceReferences(ctx context.Context, accountID string) (*accountcleanup.References, error) {
+	ctx, cancel := context.WithTimeout(ctx, r.budget)
+	defer cancel()
+	out, err := ec2instanceapi.DescribeAccountInstancesComplete(ctx, r.nc, r.nodeIDs, accountID)
+	if err != nil {
+		return nil, err
+	}
+	return instanceReferences(out), nil
+}
+
+func instanceReferences(out *ec2.DescribeInstancesOutput) *accountcleanup.References {
+	refs := &accountcleanup.References{
+		KeyNames: map[string]bool{}, PlacementGroupNames: map[string]bool{}, PlacementGroupIDs: map[string]bool{},
+	}
+	for _, reservation := range out.Reservations {
+		for _, instance := range reservation.Instances {
+			if instance == nil {
+				continue
+			}
+			if instance.State != nil && instance.State.Name != nil && terminalInstanceStates[*instance.State.Name] {
+				continue
+			}
+			if instance.KeyName != nil && *instance.KeyName != "" {
+				refs.KeyNames[*instance.KeyName] = true
+			}
+			if instance.Placement != nil {
+				if instance.Placement.GroupName != nil && *instance.Placement.GroupName != "" {
+					refs.PlacementGroupNames[*instance.Placement.GroupName] = true
+				}
+				if instance.Placement.GroupId != nil && *instance.Placement.GroupId != "" {
+					refs.PlacementGroupIDs[*instance.Placement.GroupId] = true
+				}
+			}
+		}
+	}
+	return refs
+}

spinifex/operator/accountlifecycle/exercise_test.go

@@ -0,0 +1,288 @@
+package accountlifecycle
+
+//test:in-package — the exercise's inventory and cleanup collaborators are
+// unexported seams, and the S7 invariants drive the engine directly.
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type stubCleaner struct {
+	calls  []accountcleanup.Request
+	result *accountcleanup.Result
+	err    error
+	// onCall runs inside the cleanup, so a test can observe what was recorded first.
+	onCall func()
+}
+
+func (s *stubCleaner) CleanupPlatform(_ context.Context, req *accountcleanup.Request) (*accountcleanup.Result, error) {
+	s.calls = append(s.calls, *req)
+	if s.onCall != nil {
+		s.onCall()
+	}
+	return s.result, s.err
+}
+
+// sequencedInventory answers the quiescence evaluation first, then the re-read.
+type sequencedInventory struct {
+	replies []ec2v1.PlatformInventoryReply
+	calls   int
+}
+
+func (s *sequencedInventory) PlatformInventory(context.Context, string, string) ec2v1.PlatformInventoryReply {
+	reply := s.replies[min(s.calls, len(s.replies)-1)]
+	s.calls++
+	return reply
+}
+
+func exerciseEngine(accounts AccountStore, cleaner PlatformCleaner, inventory ec2PlatformInventory) *Engine {
+	engine := quietEngine(accounts)
+	engine.Cleanup = cleaner
+	engine.ec2Inventory = inventory
+	return engine
+}
+
+func exerciseRequest() Request {
+	req := request()
+	req.OperationID, req.Fence = "op-1", 1
+	return req
+}
+
+func cleanResult() *accountcleanup.Result {
+	res := &accountcleanup.Result{AccountID: testAccountID}
+	for _, kind := range ec2v1.PlatformKinds() {
+		res.Kinds = append(res.Kinds, accountcleanup.KindReport{Kind: kind, Status: ec2v1.PlatformStatusObservedClean})
+	}
+	return res
+}
+
+func accountingFor(result *operatorv1.ExerciseResult, kind string) operatorv1.KindAccounting {
+	for _, entry := range result.Accounting {
+		if entry.Kind == kind {
+			return entry
+		}
+	}
+	return operatorv1.KindAccounting{}
+}
+
+// The exercise records the evaluation before it cleans, cleans only the kinds that
+// evaluation recorded as quiesced, and never removes the account record.
+func TestExerciseCleansTheQuiescedKindsAfterRecordingAndKeepsTheAccount(t *testing.T) {
+	accounts := newFakeAccounts()
+	var recorded *operatorv1.IAMQuiescence
+	cleaner := &stubCleaner{result: cleanResult()}
+	cleaner.onCall = func() { assert.NotNil(t, recorded, "quiescence must be recorded before any cleanup") }
+	inventory := &sequencedInventory{replies: []ec2v1.PlatformInventoryReply{*completeInventory()}}
+	engine := exerciseEngine(accounts, cleaner, inventory)
+	engine.OnQuiesced = func(_ context.Context, q operatorv1.IAMQuiescence) { recorded = &q }
+
+	result, err := engine.Exercise(context.Background(), exerciseRequest())
+
+	require.NoError(t, err)
+	require.NotNil(t, recorded)
+	assert.ElementsMatch(t, ec2PlatformKinds, recorded.QuiescedScope)
+	require.Len(t, cleaner.calls, 1)
+	assert.ElementsMatch(t, recorded.QuiescedScope, cleaner.calls[0].Kinds)
+	assert.Equal(t, "op-1", cleaner.calls[0].CorrelationRef)
+	assert.False(t, accounts.deleted, "an exercise never removes the account record")
+	assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
+	assert.Equal(t, 2, inventory.calls, "one evaluation read and one final re-read")
+	for _, kind := range recorded.QuiescedScope {
+		assert.Equal(t, operatorv1.AccountingAccounted, accountingFor(result, kind).Status, kind)
+	}
+}
+
+// The final re-read is held against the cleanup result and the quiescence record
+// is not rewritten from it.
+func TestExerciseAccountsForARemainderOnlyWhenTheCleanupNamedIt(t *testing.T) {
+	cases := map[string]struct {
+		cleanup *accountcleanup.Result
+		reread  ec2v1.PlatformInventoryReply
+		want    string
+		reason  string
+	}{
+		"late write the cleanup never saw": {
+			cleanup: cleanResult(),
+			reread:  rereadWithKeyPair("key-late"),
+			want:    operatorv1.AccountingUnaccounted, reason: operatorv1.AccountingReasonUnnamedRemainder,
+		},
+		"cleanup deleted it but it is back": {
+			cleanup: withItem(cleanResult(), accountcleanup.Item{ResourceID: "key-1", Kind: ec2v1.PlatformKindKeyPair, Disposition: accountcleanup.DispositionDeleted}),
+			reread:  rereadWithKeyPair("key-1"),
+			want:    operatorv1.AccountingUnaccounted, reason: operatorv1.AccountingReasonUnnamedRemainder,
+		},
+		"cleanup blocked it with a reason": {
+			cleanup: withItem(cleanResult(), accountcleanup.Item{ResourceID: "key-1", Kind: ec2v1.PlatformKindKeyPair, Disposition: accountcleanup.DispositionBlocked, Reason: accountcleanup.ReasonInstanceReference}),
+			reread:  rereadWithKeyPair("key-1"),
+			want:    operatorv1.AccountingAccounted,
+		},
+		"re-read incomplete": {
+			cleanup: cleanResult(),
+			reread:  *incompleteKeyPairs(),
+			want:    operatorv1.AccountingUnaccounted, reason: operatorv1.AccountingReasonRereadIncomplete,
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			accounts := newFakeAccounts()
+			inventory := &sequencedInventory{replies: []ec2v1.PlatformInventoryReply{*completeInventory(), tc.reread}}
+			engine := exerciseEngine(accounts, &stubCleaner{result: tc.cleanup}, inventory)
+			var recorded operatorv1.IAMQuiescence
+			engine.OnQuiesced = func(_ context.Context, q operatorv1.IAMQuiescence) { recorded = q }
+
+			result, err := engine.Exercise(context.Background(), exerciseRequest())
+
+			require.NoError(t, err)
+			entry := accountingFor(result, ec2v1.PlatformKindKeyPair)
+			assert.Equal(t, tc.want, entry.Status)
+			assert.Equal(t, tc.reason, entry.Reason)
+			assert.Contains(t, recorded.QuiescedScope, ec2v1.PlatformKindKeyPair, "quiescence is not rewritten from the re-read")
+		})
+	}
+}
+
+func rereadWithKeyPair(id string) ec2v1.PlatformInventoryReply {
+	reply := *completeInventory()
+	reply.Kinds[0].Status = ec2v1.PlatformStatusRecordsFound
+	reply.Items = []ec2v1.PlatformItem{{ResourceID: id, Kind: ec2v1.PlatformKindKeyPair, Disposition: accountcleanup.DispositionBlocked}}
+	return reply
+}
+
+func incompleteKeyPairs() *ec2v1.PlatformInventoryReply {
+	reply := completeInventory()
+	reply.Kinds[0].Status = ec2v1.PlatformStatusScanIncomplete
+	return reply
+}
+
+func withItem(res *accountcleanup.Result, item accountcleanup.Item) *accountcleanup.Result {
+	res.Items = append(res.Items, item)
+	return res
+}
+
+// With no kind quiesced the exercise runs no cleanup and records why.
+func TestExerciseWithNoQuiescedKindRunsNoCleanup(t *testing.T) {
+	accounts := newFakeAccounts()
+	cleaner := &stubCleaner{result: cleanResult()}
+	incomplete := ec2v1.IncompletePlatformInventory(testAccountID, "op-1", ec2v1.PlatformReasonTimeout)
+	engine := exerciseEngine(accounts, cleaner, &sequencedInventory{replies: []ec2v1.PlatformInventoryReply{incomplete}})
+
+	result, err := engine.Exercise(context.Background(), exerciseRequest())
+
+	require.NoError(t, err)
+	assert.Empty(t, cleaner.calls)
+	assert.Equal(t, operatorv1.ExerciseSkippedNoQuiescedKind, result.Skipped)
+	assert.Nil(t, result.Cleanup)
+}
+
+func TestExerciseRefusesForceAndAMissingCollaborator(t *testing.T) {
+	accounts := newFakeAccounts()
+	req := exerciseRequest()
+	req.Force = true
+	_, err := exerciseEngine(accounts, &stubCleaner{}, &stubInventory{}).Exercise(context.Background(), req)
+	require.Error(t, err)
+
+	_, err = exerciseEngine(accounts, nil, &stubInventory{}).Exercise(context.Background(), exerciseRequest())
+	require.Error(t, err)
+	assert.Empty(t, accounts.quiesces, "a refused exercise quiesces nothing")
+}
+
+// ADR-0003:S7: the exercise is an automated actor that deletes customer-visible
+// records, so it acts only on kinds whose quiescence was recorded under a confirmed
+// fence, and a failed cleanup leaves the account where it is.
+func TestS7_ExerciseCleansNothingUnlessItsQuiescenceWasRecordedUnderAConfirmedFence(t *testing.T) {
+	cases := map[string]func(*Engine, *fakeAccounts) *sequencedInventory{
+		"read-back fails": func(e *Engine, a *fakeAccounts) *sequencedInventory {
+			e.Accounts = &readbackFails{fakeAccounts: a}
+			return &sequencedInventory{replies: []ec2v1.PlatformInventoryReply{*completeInventory()}}
+		},
+		"inventory incomplete": func(*Engine, *fakeAccounts) *sequencedInventory {
+			return &sequencedInventory{replies: []ec2v1.PlatformInventoryReply{
+				ec2v1.IncompletePlatformInventory(testAccountID, "op-1", ec2v1.PlatformReasonNoResponder),
+			}}
+		},
+	}
+	for name, setup := range cases {
+		t.Run(name, func(t *testing.T) {
+			accounts := newFakeAccounts()
+			cleaner := &stubCleaner{result: cleanResult()}
+			engine := exerciseEngine(accounts, cleaner, nil)
+			engine.ec2Inventory = setup(engine, accounts)
+
+			_, err := engine.Exercise(context.Background(), exerciseRequest())
+
+			require.NoError(t, err)
+			assert.Empty(t, cleaner.calls, "no kind may be cleaned without a recorded quiescence")
+			assert.False(t, accounts.deleted)
+		})
+	}
+
+	t.Run("only the kinds with a complete inventory are cleaned", func(t *testing.T) {
+		accounts := newFakeAccounts()
+		cleaner := &stubCleaner{result: cleanResult()}
+		partial := completeInventory()
+		partial.Kinds[1].Status = ec2v1.PlatformStatusScanIncomplete
+		engine := exerciseEngine(accounts, cleaner, &sequencedInventory{replies: []ec2v1.PlatformInventoryReply{*partial}})
+
+		_, err := engine.Exercise(context.Background(), exerciseRequest())
+
+		require.NoError(t, err)
+		require.Len(t, cleaner.calls, 1)
+		assert.ElementsMatch(t, []string{ec2v1.PlatformKindKeyPair, ec2v1.PlatformKindLaunchTemplate}, cleaner.calls[0].Kinds)
+	})
+
+	t.Run("a failed cleanup returns an error and removes nothing", func(t *testing.T) {
+		accounts := newFakeAccounts()
+		engine := exerciseEngine(accounts, &stubCleaner{err: errors.New("owner down")}, &sequencedInventory{replies: []ec2v1.PlatformInventoryReply{*completeInventory()}})
+
+		_, err := engine.Exercise(context.Background(), exerciseRequest())
+
+		require.Error(t, err)
+		assert.False(t, accounts.deleted)
+	})
+}
+
+// ADR-0003:S7: an ordinary teardown never runs the platform cleanup and never
+// records a quiesced scope, whatever its inventory says.
+func TestS7_OrdinaryTeardownNeverCleansOrRecordsAScope(t *testing.T) {
+	accounts := newFakeAccounts()
+	cleaner := &stubCleaner{result: cleanResult()}
+	engine := exerciseEngine(accounts, cleaner, &sequencedInventory{replies: []ec2v1.PlatformInventoryReply{*completeInventory()}})
+	var recorded operatorv1.IAMQuiescence
+	engine.OnQuiesced = func(_ context.Context, q operatorv1.IAMQuiescence) { recorded = q }
+
+	_, err := engine.Teardown(context.Background(), exerciseRequest())
+
+	require.NoError(t, err)
+	assert.Empty(t, cleaner.calls)
+	assert.Empty(t, recorded.QuiescedScope)
+}
+
+func TestInstanceReferencesSkipTerminalInstancesAndNameKeysAndGroups(t *testing.T) {
+	refs := instanceReferences(&ec2.DescribeInstancesOutput{Reservations: []*ec2.Reservation{{Instances: []*ec2.Instance{
+		instanceWith("running", "kp-a", "pg-a", "pg-1"),
+		instanceWith("stopped", "kp-b", "", ""),
+		instanceWith("terminated", "kp-gone", "pg-gone", "pg-9"),
+	}}}})
+
+	assert.Equal(t, map[string]bool{"kp-a": true, "kp-b": true}, refs.KeyNames)
+	assert.Equal(t, map[string]bool{"pg-a": true}, refs.PlacementGroupNames)
+	assert.Equal(t, map[string]bool{"pg-1": true}, refs.PlacementGroupIDs)
+}
+
+func instanceWith(state, key, groupName, groupID string) *ec2.Instance {
+	return &ec2.Instance{
+		State:     &ec2.InstanceState{Name: aws.String(state)},
+		KeyName:   aws.String(key),
+		Placement: &ec2.Placement{GroupName: aws.String(groupName), GroupId: aws.String(groupID)},
+	}
+}

spinifex/operator/accountlifecycle/quiescence.go

@@ -7,17 +7,17 @@ import (
 	operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1"
 )
 
-// finalInventoryRereadWired says whether anything re-reads the owner inventory
-// before the account record goes. Until it does, a write landing after the
-// reaper stage is orphaned, so no kind may be recorded as quiesced.
-var finalInventoryRereadWired = false
-
 // QuiesceEvidence is what the engine read after QuiesceAccount returned.
 type QuiesceEvidence struct {
 	OperationID string
 	Fence       uint64
 	Generation  uint64
 
+	// Exercise is set only for a non-finalizing run, which re-reads the owner
+	// inventory after its cleanup and removes no account record. An ordinary
+	// delete re-reads nothing before the record goes, so it records no scope.
+	Exercise bool
+
 	// Account is the IAM record read back after the quiesce.
 	Account     *Account
 	ReadbackErr error
@@ -79,7 +79,8 @@ var otherScopes = []unresolvedEntry{
 
 // EvaluateQuiescence applies ADR-0008:S9's three conditions to the evidence and
 // records what holds. A fence the read-back does not confirm stops everything
-// else; otherwise every scope is listed, and none is reported quiesced yet.
+// else; otherwise every scope is listed. A kind is quiesced only for an
+// exercise whose owner inventory for it was complete.
 func EvaluateQuiescence(ev QuiesceEvidence) operatorv1.IAMQuiescence {
 	out := operatorv1.IAMQuiescence{
 		OperationID:   ev.OperationID,
@@ -109,7 +110,7 @@ func EvaluateQuiescence(ev QuiesceEvidence) operatorv1.IAMQuiescence {
 		switch {
 		case !complete:
 			out.Unresolved = append(out.Unresolved, writerEntry("ec2-"+kind, operatorv1.ReasonOwnerInventoryIncomplete))
-		case !finalInventoryRereadWired:
+		case !ev.Exercise:
 			out.Unresolved = append(out.Unresolved, writerEntry("ec2-"+kind, operatorv1.ReasonFinalInventoryRereadNotWired))
 		default:
 			out.QuiescedScope = append(out.QuiescedScope, kind)

spinifex/operator/accountlifecycle/quiescence_test.go

@@ -72,12 +72,10 @@ func TestEvaluateQuiescenceRecordsEvidenceButQuiescesNothingWithoutAFinalReread(
 	assert.Equal(t, "2026-11-02T00:00:00Z", q.EvaluatedAt)
 }
 
-// The wiring step flips this switch; the complete kinds are then, and only then, named.
+// Only an exercise, which re-reads the inventory after cleanup, names the complete kinds.
 func TestEvaluateQuiescenceNamesCompleteKindsOnceTheRereadIsWired(t *testing.T) {
-	finalInventoryRereadWired = true
-	t.Cleanup(func() { finalInventoryRereadWired = false })
-
 	ev := confirmedEvidence()
+	ev.Exercise = true
 	ev.Inventory.Kinds[1].Status = ec2v1.PlatformStatusScanIncomplete
 	q := EvaluateQuiescence(ev)
 
@@ -133,9 +131,8 @@ func TestEvaluateQuiescenceRecordsNothingWhenTheFenceIsNotConfirmed(t *testing.T
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
-			finalInventoryRereadWired = true
-			t.Cleanup(func() { finalInventoryRereadWired = false })
 			ev := confirmedEvidence()
+			ev.Exercise = true
 			tc.mutate(&ev)
 			q := EvaluateQuiescence(ev)
 
@@ -172,11 +169,11 @@ func TestEveryOtherScopeIsRecordedUnresolvedByName(t *testing.T) {
 
 // Predastore has no accepted enforcement contract, so the scopes it admits writes
 // for stay unresolved and it is never cited as a consumer, even once the
-// re-read is wired and every L09 kind is complete.
+// run is an exercise and every L09 kind is complete.
 func TestPredastoreIsNeverCitedAndItsScopesNeverQuiesce(t *testing.T) {
-	finalInventoryRereadWired = true
+	ev := confirmedEvidence()
-	t.Cleanup(func() { finalInventoryRereadWired = false })
+	ev.Exercise = true
-	q := EvaluateQuiescence(confirmedEvidence())
+	q := EvaluateQuiescence(ev)
 
 	for _, scope := range []string{"s3", "ecr"} {
 		got := unresolvedFor(q, scope)

spinifex/operator/admin/account_delete.go

@@ -66,7 +66,10 @@ type accountDeletionJob struct {
 	ClientToken string `json:"clientToken"`
 	State       string `json:"state"`
 	Force       bool   `json:"force"`
-	StartedAt   string `json:"startedAt"`
+	// NonFinalizing marks an exercise: it quiesces and cleans the quiesced EC2
+	// platform kinds, then ends EXERCISED with the account still TERMINATING.
+	NonFinalizing bool   `json:"nonFinalizing,omitempty"`
+	StartedAt     string `json:"startedAt"`
 	// UpdatedAt is the running job's heartbeat. Its absence for long enough is
 	// what lets another gateway take over after a restart.
 	UpdatedAt  string                        `json:"updatedAt"`
@@ -88,6 +91,8 @@ type accountDeletionJob struct {
 	// IAM is the quiescence evaluation made under the current fence. A takeover
 	// starts without one and the new holder records its own.
 	IAM *operatorv1.IAMQuiescence `json:"iam,omitempty"`
+	// Exercise is what a non-finalizing run did after quiescence.
+	Exercise *operatorv1.ExerciseResult `json:"exercise,omitempty"`
 }
 
 // deletionTakeover records one change of holder.
@@ -99,8 +104,9 @@ type deletionTakeover struct {
 }
 
 const (
-	takeoverReasonLeaseExpired = "lease_expired"
+	takeoverReasonLeaseExpired     = "lease_expired"
-	takeoverReasonFailed       = "failed"
+	takeoverReasonFailed           = "failed"
+	takeoverReasonExerciseFinished = "exercise_finished"
 )
 
 // deleteAccount starts a teardown, or answers a dry run inline.
@@ -118,6 +124,11 @@ func (d Deps) deleteAccount(ctx context.Context, body []byte) (any, error) {
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}
 
+	if req.NonFinalizing && d.PlatformCleaner == nil {
+		slog.Error("DeleteAccount: platform cleanup not available")
+		return nil, errors.New(awserrors.ErrorServiceUnavailable)
+	}
+
 	engine, err := accountteardown.NewClusterEngine(ctx, d.NATS, accountteardown.ClusterNodes{Expected: d.ExpectedNodes, DaemonIDs: d.NodeIDs, GatewayIDs: d.GatewayNodeIDs}, d.Termination, d.BucketStore)
 	if err != nil {
 		slog.Error("DeleteAccount: failed to build teardown engine", "err", err)
@@ -262,6 +273,10 @@ func (d Deps) runAccountDeletion(
 	tracker := &deletionTracker{jobs: jobs, job: job, revision: revision, cancel: cancel}
 	engine.OnStage = tracker.stageFinished
 	engine.OnQuiesced = tracker.quiesced
+	if job.NonFinalizing {
+		engine.Cleanup = d.PlatformCleaner
+		engine.OnExercised = tracker.exercised
+	}
 
 	go func() {
 		defer cancel()
@@ -269,13 +284,19 @@ func (d Deps) runAccountDeletion(
 		stop := tracker.startHeartbeat(ctx)
 		defer stop()
 
-		result, err := engine.Teardown(ctx, accountteardown.Request{
+		request := accountteardown.Request{
 			AccountID:   job.AccountID,
 			AccountName: job.AccountName,
 			Force:       job.Force,
 			OperationID: job.OperationID,
 			Fence:       job.Fence,
-		})
+		}
+		if job.NonFinalizing {
+			_, err := engine.Exercise(ctx, request)
+			tracker.finished(ctx, nil, err)
+			return
+		}
+		result, err := engine.Teardown(ctx, request)
 		tracker.finished(ctx, result, err)
 	}()
 }
@@ -308,6 +329,13 @@ func (t *deletionTracker) quiesced(ctx context.Context, quiescence operatorv1.IA
 	t.persist(ctx)
 }
 
+func (t *deletionTracker) exercised(ctx context.Context, result operatorv1.ExerciseResult) {
+	t.mu.Lock()
+	t.job.Exercise = &result
+	t.mu.Unlock()
+	t.persist(ctx)
+}
+
 func (t *deletionTracker) finished(ctx context.Context, result *accountteardown.Result, err error) {
 	t.mu.Lock()
 	if result != nil {
@@ -316,6 +344,8 @@ func (t *deletionTracker) finished(ctx context.Context, result *accountteardown.
 	if err != nil {
 		t.job.State = operatorv1.DeletionStateFailed
 		t.job.Error = err.Error()
+	} else if t.job.NonFinalizing {
+		t.job.State = operatorv1.DeletionStateExercised
 	} else {
 		t.job.State = operatorv1.DeletionStateCompleted
 	}
@@ -445,8 +475,10 @@ func claimDeletionJob(
 		StartedAt:   now.Format(time.RFC3339),
 		UpdatedAt:   now.Format(time.RFC3339),
 		Fence:       1,
-		Holder:      holder,
+
-		LeaseUntil:  now.Add(accountDeletionStale).Format(time.RFC3339),
+		NonFinalizing: req.NonFinalizing,
+		Holder:        holder,
+		LeaseUntil:    now.Add(accountDeletionStale).Format(time.RFC3339),
 	}
 	record, err := json.Marshal(job)
 	if err != nil {
@@ -478,6 +510,9 @@ func claimDeletionJob(
 	}
 
 	reason := takeoverReasonFailed
+	if existing.State == operatorv1.DeletionStateExercised {
+		reason = takeoverReasonExerciseFinished
+	}
 	if existing.State == operatorv1.DeletionStateRunning {
 		if !deletionLeaseExpired(existing, now) {
 			return nil, 0, nil, errors.New(awserrors.ErrorOperationInProgress)
@@ -512,8 +547,10 @@ func takeOverDeletionJob(
 		Force:       req.Force,
 		StartedAt:   now.Format(time.RFC3339),
 		UpdatedAt:   now.Format(time.RFC3339),
-		Holder:      holder,
+
-		LeaseUntil:  now.Add(accountDeletionStale).Format(time.RFC3339),
+		NonFinalizing: req.NonFinalizing,
+		Holder:        holder,
+		LeaseUntil:    now.Add(accountDeletionStale).Format(time.RFC3339),
 	}
 	if existing.OperationID != "" {
 		job.OperationID = existing.OperationID
@@ -557,7 +594,7 @@ func replayDeletionJob(
 		return nil, err
 	}
 	if existing.ClientToken == req.ClientToken {
-		if existing.AccountName != req.AccountName {
+		if existing.AccountName != req.AccountName || existing.NonFinalizing != req.NonFinalizing {
 			return nil, errors.New(awserrors.ErrorIdempotentParameterMismatch)
 		}
 		// A RUNNING job whose lease has lapsed has no holder to follow, so the
@@ -703,6 +740,9 @@ func validateDeleteAccountRequest(req *operatorv1.DeleteAccountRequest) error {
 	if !accountIDRE.MatchString(req.AccountID) {
 		return errors.New(awserrors.ErrorInvalidParameterValue)
 	}
+	if req.NonFinalizing && (req.Force || req.DryRun) {
+		return errors.New(awserrors.ErrorInvalidParameterCombination)
+	}
 	if req.DryRun {
 		// A dry run deletes nothing, so it needs neither the name confirmation
 		// nor a token to make it idempotent.

spinifex/operator/admin/account_delete_exercise_test.go

@@ -0,0 +1,133 @@
+package admin
+
+//test:in-package — the exercise runs through the job record, the takeover and
+// the resumer, all of which are unexported.
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const exerciseToken = "1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d"
+
+type panicCleaner struct{ calls int }
+
+func (p *panicCleaner) CleanupPlatform(context.Context, *accountcleanup.Request) (*accountcleanup.Result, error) {
+	p.calls++
+	return &accountcleanup.Result{}, nil
+}
+
+func exerciseBody(t *testing.T, accountID, name, token string, mutate func(*operatorv1.DeleteAccountRequest)) []byte {
+	t.Helper()
+	req := operatorv1.DeleteAccountRequest{AccountID: accountID, AccountName: name, ClientToken: token, NonFinalizing: true}
+	if mutate != nil {
+		mutate(&req)
+	}
+	body, err := json.Marshal(req)
+	require.NoError(t, err)
+	return body
+}
+
+func awaitState(t *testing.T, gw Deps, accountID, state string) *accountDeletionJob {
+	t.Helper()
+	require.Eventually(t, func() bool {
+		return describeDeletion(t, gw, accountID).State == state
+	}, 30*time.Second, 100*time.Millisecond)
+	return describeDeletion(t, gw, accountID)
+}
+
+// An exercise ends EXERCISED with the account still TERMINATING, and a later
+// ordinary delete takes the same operation over to finish it.
+func TestNonFinalizingDeleteEndsExercisedAndALaterDeleteTakesItOver(t *testing.T) {
+	gw := newDeleteAccountGateway(t)
+	gw.PlatformCleaner = &panicCleaner{}
+	account := newTenantAccount(t, gw, "tenant@example.com")
+
+	_, err := gw.deleteAccount(t.Context(), exerciseBody(t, account.AccountID, "tenant@example.com", exerciseToken, nil))
+	require.NoError(t, err)
+
+	exercised := awaitState(t, gw, account.AccountID, operatorv1.DeletionStateExercised)
+	require.NotNil(t, exercised.IAM)
+	require.NotNil(t, exercised.Exercise)
+	assert.True(t, exercised.NonFinalizing)
+	assert.NotEmpty(t, exercised.FinishedAt)
+	assert.Empty(t, exercised.Stages, "no legacy stage runs")
+	stored, err := gw.IAM.GetAccount(account.AccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "TERMINATING", stored.Status)
+
+	_, err = gw.deleteAccount(t.Context(), deleteAccountBody(t, account.AccountID, "tenant@example.com", deleteClientToken))
+	require.NoError(t, err)
+
+	done := awaitState(t, gw, account.AccountID, operatorv1.DeletionStateCompleted)
+	assert.Equal(t, exercised.OperationID, done.OperationID)
+	assert.Equal(t, uint64(2), done.Fence)
+	require.Len(t, done.Takeovers, 1)
+	assert.Equal(t, takeoverReasonExerciseFinished, done.Takeovers[0].Reason)
+	assert.Nil(t, done.Exercise, "the takeover starts a fresh run")
+	assert.Equal(t, uint64(2), done.IAM.Fence)
+	assert.Empty(t, done.IAM.QuiescedScope, "an ordinary delete records no scope")
+}
+
+func TestNonFinalizingIsPartOfTheIdempotencyKeyAndRefusesForce(t *testing.T) {
+	gw := newDeleteAccountGateway(t)
+	gw.PlatformCleaner = &panicCleaner{}
+	account := newTenantAccount(t, gw, "tenant@example.com")
+
+	_, err := gw.deleteAccount(t.Context(), exerciseBody(t, account.AccountID, "tenant@example.com", exerciseToken,
+		func(r *operatorv1.DeleteAccountRequest) { r.Force = true }))
+	require.EqualError(t, err, awserrors.ErrorInvalidParameterCombination)
+
+	_, err = gw.deleteAccount(t.Context(), exerciseBody(t, account.AccountID, "tenant@example.com", exerciseToken, nil))
+	require.NoError(t, err)
+	awaitState(t, gw, account.AccountID, operatorv1.DeletionStateExercised)
+
+	_, err = gw.deleteAccount(t.Context(), deleteAccountBody(t, account.AccountID, "tenant@example.com", exerciseToken))
+	require.EqualError(t, err, awserrors.ErrorIdempotentParameterMismatch, "the same token cannot turn an exercise into a delete")
+
+	again, err := gw.deleteAccount(t.Context(), exerciseBody(t, account.AccountID, "tenant@example.com", exerciseToken, nil))
+	require.NoError(t, err)
+	assert.Equal(t, operatorv1.DeletionStateExercised, again.(*DeleteAccountResponse).State, "a repeated exercise is answered from the record")
+}
+
+func TestNonFinalizingWithoutACleanupIsRefusedBeforeAnyJobExists(t *testing.T) {
+	gw := newDeleteAccountGateway(t)
+	account := newTenantAccount(t, gw, "tenant@example.com")
+
+	_, err := gw.deleteAccount(t.Context(), exerciseBody(t, account.AccountID, "tenant@example.com", exerciseToken, nil))
+
+	require.EqualError(t, err, awserrors.ErrorServiceUnavailable)
+	jobs, err := gw.accountDeletionStore(t.Context())
+	require.NoError(t, err)
+	stored, err := readDeletionJob(t.Context(), jobs, account.AccountID)
+	require.NoError(t, err)
+	assert.Nil(t, stored)
+}
+
+// A resumed exercise stays an exercise: the resumer must not turn a recorded
+// non-finalizing decision into a deletion.
+func TestResumerKeepsAnAbandonedExerciseNonFinalizing(t *testing.T) {
+	gw := newDeleteAccountGateway(t)
+	gw.PlatformCleaner = &panicCleaner{}
+	account := newTenantAccount(t, gw, "tenant@example.com")
+	job := abandonedJob(account.AccountID, "tenant@example.com")
+	job.NonFinalizing = true
+	storeDeletionJob(t, gw, job)
+
+	require.Equal(t, 1, gw.resumeDeletions(t.Context(), time.Now().UTC()))
+
+	done := awaitState(t, gw, account.AccountID, operatorv1.DeletionStateExercised)
+	assert.Equal(t, uint64(2), done.Fence)
+	assert.True(t, done.NonFinalizing)
+	stored, err := gw.IAM.GetAccount(account.AccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "TERMINATING", stored.Status, "the account record is never removed")
+}

spinifex/operator/admin/account_delete_resumer.go

@@ -110,6 +110,8 @@ func (d Deps) resumeDeletion(
 		AccountName: existing.AccountName,
 		ClientToken: existing.ClientToken,
 		Force:       existing.Force,
+
+		NonFinalizing: existing.NonFinalizing,
 	}
 	job, newRevision, _, err := takeOverDeletionJob(ctx, jobs, req, d.holderID(), existing, revision, takeoverReasonLeaseExpired, now)
 	if err != nil {

spinifex/operator/admin/account_delete_resumer_test.go

@@ -45,6 +45,7 @@ func TestS7_DeletionResumerActsOnlyOnAnAcceptedOperationWithoutALiveLease(t *tes
 		},
 		"failed job":    func(j *accountDeletionJob) { j.State = operatorv1.DeletionStateFailed },
 		"completed job": func(j *accountDeletionJob) { j.State = operatorv1.DeletionStateCompleted },
+		"exercised job": func(j *accountDeletionJob) { j.State = operatorv1.DeletionStateExercised },
 		"legacy record": func(j *accountDeletionJob) { j.OperationID, j.Fence, j.LeaseUntil = "", 0, "" },
 	}
 	for name, mutate := range cases {

spinifex/operator/admin/registration.go

@@ -58,7 +58,10 @@ type Deps struct {
 	IAM iamdomain.IAMService
 	// Termination is the concrete IAM service, the only holder of the capability
 	// to quiesce and remove an account. DeleteAccount needs it.
-	Termination       accountteardown.IAMBackend
+	Termination accountteardown.IAMBackend
+	// PlatformCleaner is the in-process EC2 platform cleanup a non-finalizing
+	// DeleteAccount runs. Nil refuses that request.
+	PlatformCleaner   accountteardown.PlatformCleaner
 	Quota             *quota.Service
 	ExpectedNodes     int
 	NodeIDs           []string

spinifex/operator/cli/admin_account_delete.go

@@ -61,6 +61,8 @@ func init() {
 	accountCmd.AddCommand(accountDeleteCmd)
 
 	accountDeleteCmd.Flags().Bool("dry-run", false, "Report what would be deleted and change nothing")
+	accountDeleteCmd.Flags().Bool("non-finalizing", false, "Quiesce and clean the EC2 platform kinds, then stop with the account still TERMINATING")
+	_ = accountDeleteCmd.Flags().MarkHidden("non-finalizing")
 	accountDeleteCmd.Flags().Bool("force", false, "Escalate past state guards for resources that will not delete")
 	accountDeleteCmd.Flags().Bool("yes", false, "Skip the interactive confirmation (requires --name)")
 	accountDeleteCmd.Flags().String("name", "", "Account name, which must match the stored record")

spinifex/operator/cli/admin_account_delete_remote.go

@@ -24,6 +24,7 @@ var accountDeletePollInterval = 10 * time.Second
 func runAccountDeleteRemote(cmd *cobra.Command, accountID string) error {
 	dryRun, _ := cmd.Flags().GetBool("dry-run")
 	force, _ := cmd.Flags().GetBool("force")
+	nonFinalizing, _ := cmd.Flags().GetBool("non-finalizing")
 	assumeYes, _ := cmd.Flags().GetBool("yes")
 	name, _ := cmd.Flags().GetString("name")
 	clientToken, _ := cmd.Flags().GetString("client-token")
@@ -66,6 +67,7 @@ func runAccountDeleteRemote(cmd *cobra.Command, accountID string) error {
 
 	started, err := deleteAccountRemote(ctx, target, operatorv1.DeleteAccountRequest{
 		AccountID: accountID, AccountName: name, ClientToken: clientToken, Force: force,
+		NonFinalizing: nonFinalizing,
 	})
 	if err != nil {
 		var adminErr *adminError
@@ -101,6 +103,9 @@ func followAccountDeletion(ctx context.Context, target adminTarget, accountID st
 		case operatorv1.DeletionStateCompleted:
 			fmt.Printf("\nAccount %s deleted.\n", accountID)
 			return nil
+		case operatorv1.DeletionStateExercised:
+			fmt.Printf("\nAccount %s exercised and left TERMINATING; a normal delete finishes it.\n", accountID)
+			return nil
 		case operatorv1.DeletionStateFailed:
 			fmt.Fprintf(os.Stderr, "\nThe account is still TERMINATING so its leftovers keep an owner.\n")
 			return fmt.Errorf("teardown failed: %s", job.Error)

spinifex/runtime/roles/awsgw/awsgw.go

@@ -725,6 +725,9 @@ func launchService(config *config.ClusterConfig) error {
 		SQSWorkerPolling:       sqsRelay.WorkerPresent,
 		SQSTenantAPIRegistered: func() bool { return sqsServed(composedServices) },
 	})
+	operatorAdminDeps = withPlatformCleaner(operatorAdminDeps, func() (accountlifecycle.PlatformCleaner, error) {
+		return platformCleaner(janitorCtx, &nodeConfig, natsConn, js, operatorAdminDeps.NodeIDs)
+	})
 	operatorControlPlane := operatoradmin.NewRegistration(operatorAdminDeps)
 	// Every gateway scans for account deletions whose holder died; the fenced
 	// takeover makes any number of them safe.

spinifex/runtime/roles/awsgw/compose_test.go

@@ -1,12 +1,16 @@
 package awsgw
 
 import (
+	"context"
+	"errors"
 	"go/ast"
 	"go/parser"
 	"go/token"
 	"testing"
 
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	"github.com/mulgadc/spinifex/spinifex/operator/accountlifecycle"
 	operatoradmin "github.com/mulgadc/spinifex/spinifex/operator/admin"
 
 	awsapi "github.com/mulgadc/spinifex/spinifex/domains/ecr/awsapi"
@@ -88,3 +92,40 @@ func TestAwsgwStartsTheDeletionResumer(t *testing.T) {
 	})
 	assert.True(t, started, "awsgw.go must call Deps.StartDeletionResumer")
 }
+
+type fixedCleaner struct{}
+
+func (fixedCleaner) CleanupPlatform(context.Context, *accountcleanup.Request) (*accountcleanup.Result, error) {
+	return &accountcleanup.Result{}, nil
+}
+
+// A non-finalizing DeleteAccount cleans through the in-process owners awsgw
+// binds. A cleanup that cannot be built must leave the field nil, which refuses
+// the request, rather than bind something half-working.
+func TestWithPlatformCleaner_BindsTheCleanupOrLeavesItNil(t *testing.T) {
+	bound := withPlatformCleaner(operatoradmin.Deps{}, func() (accountlifecycle.PlatformCleaner, error) { return fixedCleaner{}, nil })
+	assert.NotNil(t, bound.PlatformCleaner)
+
+	unbound := withPlatformCleaner(operatoradmin.Deps{}, func() (accountlifecycle.PlatformCleaner, error) {
+		return nil, errors.New("owner store unavailable")
+	})
+	assert.Nil(t, unbound.PlatformCleaner)
+}
+
+// The cleanup is bound where the operator deps are composed, or the exercise is
+// refused on every gateway.
+func TestAwsgwBindsThePlatformCleanup(t *testing.T) {
+	file, err := parser.ParseFile(token.NewFileSet(), "awsgw.go", nil, 0)
+	require.NoError(t, err)
+
+	var bound bool
+	ast.Inspect(file, func(n ast.Node) bool {
+		if call, ok := n.(*ast.CallExpr); ok {
+			if ident, ok := call.Fun.(*ast.Ident); ok && ident.Name == "withPlatformCleaner" {
+				bound = true
+			}
+		}
+		return true
+	})
+	assert.True(t, bound, "awsgw.go must call withPlatformCleaner")
+}

spinifex/runtime/roles/awsgw/platform_cleanup.go

@@ -0,0 +1,66 @@
+package awsgw
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	ec2key "github.com/mulgadc/spinifex/spinifex/domains/ec2/key"
+	ec2launchtemplate "github.com/mulgadc/spinifex/spinifex/domains/ec2/launchtemplate"
+	ec2placementgroup "github.com/mulgadc/spinifex/spinifex/domains/ec2/placementgroup"
+	ec2spotinstance "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
+	ec2tags "github.com/mulgadc/spinifex/spinifex/domains/ec2/tags"
+	"github.com/mulgadc/spinifex/spinifex/operator/accountlifecycle"
+	operatoradmin "github.com/mulgadc/spinifex/spinifex/operator/admin"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// platformCleaner builds the EC2 platform cleanup over the owners' own services
+// in this process. It has no NATS subject, so the only caller is the operator
+// admin DeleteAccount that holds it. Deleting a key pair, placement group or
+// launch template through these owners clears its tag-index entry too.
+func platformCleaner(ctx context.Context, node *config.Config, nc *nats.Conn, js jetstream.JetStream, nodeIDs []string) (accountlifecycle.PlatformCleaner, error) {
+	tagsKV, err := ec2tags.GetOrCreateTagsBucket(ctx, js)
+	if err != nil {
+		return nil, fmt.Errorf("tags bucket: %w", err)
+	}
+	tags := ec2tags.NewTagsServiceImpl(node, tagsKV)
+	keys := ec2key.NewKeyServiceImpl(node)
+	groups, err := ec2placementgroup.NewPlacementGroupServiceImplWithNATS(ctx, node, nc)
+	if err != nil {
+		return nil, fmt.Errorf("placement group service: %w", err)
+	}
+	templates, err := ec2launchtemplate.NewLaunchTemplateServiceImplWithNATS(ctx, node, nc)
+	if err != nil {
+		return nil, fmt.Errorf("launch template service: %w", err)
+	}
+	spot, err := ec2spotinstance.NewSpotInstanceServiceImplWithNATS(ctx, node, nc)
+	if err != nil {
+		return nil, fmt.Errorf("spot instance service: %w", err)
+	}
+	keys.SetCentralTagStore(tags)
+	groups.SetCentralTagStore(tags)
+	templates.SetCentralTagStore(tags)
+
+	return accountcleanup.New(keys, groups, templates, tags, accountcleanup.ReferenceReaders{
+		Instances: accountlifecycle.NewInstanceReferenceReader(nc, nodeIDs),
+		Spot:      spot,
+		Templates: templates,
+	}), nil
+}
+
+// withPlatformCleaner binds the cleanup a non-finalizing DeleteAccount runs. When
+// it cannot be built the field stays nil, which refuses that request instead of
+// half-running it.
+func withPlatformCleaner(d operatoradmin.Deps, build func() (accountlifecycle.PlatformCleaner, error)) operatoradmin.Deps {
+	cleaner, err := build()
+	if err != nil {
+		slog.Error("awsgw: EC2 platform cleanup unavailable", "err", err)
+		return d
+	}
+	d.PlatformCleaner = cleaner
+	return d
+}

tests/e2e/harness/aws.go

@@ -4,7 +4,6 @@ package harness
 
 import (
 	"crypto/tls"
-	"crypto/x509"
 	"fmt"
 	"net/http"
 	"os"
@@ -151,18 +150,3 @@ func newAWSClient(t *testing.T, env *Env, accessKey, secretKey, sessionToken str
 		BedrockRuntime: bedrockruntime.New(sess),
 	}
 }
-
-// IgnoreCertErrors disables TLS verification on all bundled clients.
-// Use only for fault-injection scenarios.
-func (c *AWSClient) IgnoreCertErrors() {
-	tr := &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
-	hc := &http.Client{Transport: tr}
-	c.EC2Conf.Config.HTTPClient = hc
-	c.ELBv2.Config.HTTPClient = hc
-	c.EKS.Config.HTTPClient = hc
-	c.IAM.Config.HTTPClient = hc
-	c.STS.Config.HTTPClient = hc
-	c.ACM.Config.HTTPClient = hc
-	c.ECR.Config.HTTPClient = hc
-	_ = (*x509.CertPool)(nil) // silence unused-import on hardened paths
-}

tests/e2e/harness/ec2helpers.go

@@ -78,18 +78,6 @@ func LsblkRootGiB(t *testing.T, tgt SSHTarget) int {
 	return int(bytesN / (1 << 30))
 }
 
-// GetSerialConsoleAccessEnabled returns the current account attribute value.
-// Wraps ec2.GetSerialConsoleAccessStatus so callers don't have to deal with
-// the pointer-bool field.
-func GetSerialConsoleAccessEnabled(t *testing.T, c *AWSClient) bool {
-	t.Helper()
-	out, err := c.EC2.GetSerialConsoleAccessStatus(&ec2.GetSerialConsoleAccessStatusInput{})
-	if err != nil {
-		t.Fatalf("GetSerialConsoleAccessStatus: %v", err)
-	}
-	return aws.BoolValue(out.SerialConsoleAccessEnabled)
-}
-
 // EnableSerialConsoleAccess flips the account attribute on and returns the
 // new value (should be true).
 func EnableSerialConsoleAccess(t *testing.T, c *AWSClient) bool {

tests/e2e/harness/iam.go

@@ -83,58 +83,3 @@ func IAMDeleteRoleAndProfilesBestEffort(c *AWSClient, roleName string, profileNa
 	}
 	_, _ = c.IAM.DeleteRole(&iam.DeleteRoleInput{RoleName: aws.String(roleName)})
 }
-
-// IAMDeleteGroupBestEffort tears down a group: removes any members, detaches all
-// attached policies, deletes any inline policies, then deletes the group. Each
-// step swallows errors so a missing fragment never cascades; usable both as a
-// pre-test sweep and as a fixture-teardown cleanup. Order matters: a group cannot
-// be deleted while it still has members, attached policies, or inline policies.
-func IAMDeleteGroupBestEffort(c *AWSClient, groupName string, memberUsers []string, policyARNs ...string) {
-	for _, u := range memberUsers {
-		_, _ = c.IAM.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
-			GroupName: aws.String(groupName),
-			UserName:  aws.String(u),
-		})
-	}
-	for _, arn := range policyARNs {
-		_, _ = c.IAM.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
-			GroupName: aws.String(groupName),
-			PolicyArn: aws.String(arn),
-		})
-	}
-	// Defensive: pull any other attached policies so the final DeleteGroup isn't
-	// blocked by a stray attach from a partial run.
-	if attached, err := c.IAM.ListAttachedGroupPolicies(&iam.ListAttachedGroupPoliciesInput{
-		GroupName: aws.String(groupName),
-	}); err == nil {
-		for _, p := range attached.AttachedPolicies {
-			_, _ = c.IAM.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
-				GroupName: aws.String(groupName),
-				PolicyArn: p.PolicyArn,
-			})
-		}
-	}
-	// Defensive: delete any inline policies so the inline-policy guard can't block
-	// the final DeleteGroup after a partial run.
-	if inline, err := c.IAM.ListGroupPolicies(&iam.ListGroupPoliciesInput{
-		GroupName: aws.String(groupName),
-	}); err == nil {
-		for _, name := range inline.PolicyNames {
-			_, _ = c.IAM.DeleteGroupPolicy(&iam.DeleteGroupPolicyInput{
-				GroupName:  aws.String(groupName),
-				PolicyName: name,
-			})
-		}
-	}
-	// Defensive: drop any members the caller didn't pass in (partial run could
-	// have added one), so the non-empty guard can't block the delete.
-	if got, err := c.IAM.GetGroup(&iam.GetGroupInput{GroupName: aws.String(groupName)}); err == nil {
-		for _, u := range got.Users {
-			_, _ = c.IAM.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
-				GroupName: aws.String(groupName),
-				UserName:  u.UserName,
-			})
-		}
-	}
-	_, _ = c.IAM.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)})
-}

tests/e2e/harness/kubectl.go

@@ -4,11 +4,8 @@ package harness
 
 import (
 	"bytes"
-	"encoding/json"
-	"fmt"
 	"os"
 	"os/exec"
-	"strings"
 	"testing"
 	"time"
 )
@@ -65,84 +62,6 @@ func (k *Kubectl) Run(timeout time.Duration, args ...string) (string, error) {
 	}
 }
 
-// AuthDiagnostics reproduces the credential exchange behind a failing call
-// without mutating anything. `Unauthorized` on its own says only that the
-// apiserver refused the request, and the two explanations — a token it rejected
-// versus one the exec plugin never produced — need different fixes.
-//
-// It probes an authenticated path rather than `/version`, which is anonymously
-// readable and so returns 200 under either explanation, and it invokes the
-// credential plugin directly so a plugin that failed is named rather than
-// inferred from a missing header.
-func (k *Kubectl) AuthDiagnostics(timeout time.Duration) string {
-	var b strings.Builder
-
-	b.WriteString("=== kubectl auth whoami ===\n")
-	out, err := k.Run(timeout, "auth", "whoami")
-	b.WriteString(out)
-	if err != nil {
-		b.WriteString("\nerror: " + err.Error() + "\n")
-	}
-
-	// kube-system is not anonymously readable, so a 200 here proves a credential
-	// was both produced and accepted.
-	b.WriteString("\n=== GET /api/v1/namespaces/kube-system ===\n")
-	out, err = k.Run(timeout, "--v=8", "get", "--raw", "/api/v1/namespaces/kube-system")
-	b.WriteString(out)
-	if err != nil {
-		b.WriteString("\nerror: " + err.Error() + "\n")
-	}
-
-	b.WriteString("\n=== credential plugin, invoked directly ===\n")
-	b.WriteString(k.execPluginProbe(timeout))
-	return b.String()
-}
-
-// execPluginProbe runs the kubeconfig's own exec credential plugin and reports
-// whether it produced a token. The token itself is never printed — CI logs are
-// retained for weeks and it is a live cluster credential.
-func (k *Kubectl) execPluginProbe(timeout time.Duration) string {
-	spec, err := k.Run(timeout, "config", "view", "--raw", "-o",
-		"jsonpath={.users[0].user.exec.command}{range .users[0].user.exec.args[*]} {@}{end}")
-	if err != nil {
-		return "could not read exec block from kubeconfig: " + err.Error() + "\n" + spec
-	}
-	argv := strings.Fields(strings.TrimSpace(spec))
-	if len(argv) == 0 {
-		return "kubeconfig user has no exec credential plugin\n"
-	}
-
-	bin, err := exec.LookPath(argv[0])
-	if err != nil {
-		return "exec plugin " + argv[0] + " not on PATH: " + err.Error() + "\n"
-	}
-	cmd := exec.Command(bin, argv[1:]...)
-	cmd.Env = append(os.Environ(), k.AWSEnv...)
-	var stdout, stderr bytes.Buffer
-	cmd.Stdout, cmd.Stderr = &stdout, &stderr
-
-	var b strings.Builder
-	fmt.Fprintf(&b, "argv: %s\n", strings.Join(argv, " "))
-	runErr := runWithTimeout(cmd, timeout)
-	fmt.Fprintf(&b, "exit: %v\n", runErr)
-	fmt.Fprintf(&b, "stderr: %s\n", strings.TrimSpace(stderr.String()))
-
-	var cred struct {
-		Status struct {
-			Token               string `json:"token"`
-			ExpirationTimestamp string `json:"expirationTimestamp"`
-		} `json:"status"`
-	}
-	if jsonErr := json.Unmarshal(stdout.Bytes(), &cred); jsonErr != nil {
-		fmt.Fprintf(&b, "stdout did not parse as an ExecCredential: %v (%d bytes)\n",
-			jsonErr, stdout.Len())
-		return b.String()
-	}
-	fmt.Fprintf(&b, "token produced: %t (%d chars), expires: %q\n",
-		cred.Status.Token != "", len(cred.Status.Token), cred.Status.ExpirationTimestamp)
-	return b.String()
-}
-
 type timeoutError struct{}
 
 func (*timeoutError) Error() string { return "kubectl timed out" }

tests/e2e/harness/poll.go

@@ -151,34 +151,3 @@ func WaitForSnapshotState(t *testing.T, c *AWSClient, id, target string, opts ..
 	t.Logf("snapshot %s reached state %s", id, target)
 	return last
 }
-
-// WaitForImageState polls DescribeImages until State == target. AMI creation
-// from a snapshot is the slowest path; default timeout 10min.
-func WaitForImageState(t *testing.T, c *AWSClient, id, target string, opts ...PollOpt) *ec2.Image {
-	t.Helper()
-	cfg := applyOpts(pollCfg{timeout: 10 * time.Minute, interval: 2 * time.Second}, opts...)
-	var last *ec2.Image
-	var lastState string
-	EventuallyErr(t, func() error {
-		out, err := c.EC2.DescribeImages(&ec2.DescribeImagesInput{
-			ImageIds: []*string{aws.String(id)},
-		})
-		if err != nil {
-			return fmt.Errorf("describe-image %s: %w", id, err)
-		}
-		if len(out.Images) == 0 {
-			return fmt.Errorf("%s not found", id)
-		}
-		last = out.Images[0]
-		lastState = aws.StringValue(last.State)
-		if lastState == target {
-			return nil
-		}
-		if lastState == "failed" {
-			return fmt.Errorf("%s entered failed state: %s", id, aws.StringValue(last.StateReason.Message))
-		}
-		return fmt.Errorf("%s state=%s want=%s", id, lastState, target)
-	}, cfg.timeout, cfg.interval)
-	t.Logf("image %s reached state %s", id, target)
-	return last
-}

tests/e2e/harness/rds.go

@@ -121,27 +121,6 @@ func WaitForDBInstanceGone(t *testing.T, c *AWSClient, id string, opts ...PollOp
 	t.Logf("DB instance %s is gone", id)
 }
 
-// WaitForDBSnapshotAvailable polls until the snapshot reports available, which
-// is the point the data exists and a restore will accept it.
-func WaitForDBSnapshotAvailable(t *testing.T, c *AWSClient, id string, opts ...PollOpt) *rds.DBSnapshot {
-	t.Helper()
-	cfg := applyOpts(pollCfg{timeout: 10 * time.Minute, interval: 5 * time.Second}, opts...)
-	var last *rds.DBSnapshot
-	EventuallyErr(t, func() error {
-		snapshot, err := DescribeDBSnapshot(c, id)
-		if err != nil {
-			return fmt.Errorf("describe-db-snapshots %s: %w", id, err)
-		}
-		last = snapshot
-		if status := aws.StringValue(snapshot.Status); status != DBSnapshotAvailable {
-			return fmt.Errorf("%s status=%s want=%s", id, status, DBSnapshotAvailable)
-		}
-		return nil
-	}, cfg.timeout, cfg.interval)
-	t.Logf("DB snapshot %s reached status %s", id, DBSnapshotAvailable)
-	return last
-}
-
 // DescribeDBInstance returns the single named DB instance. A named instance that
 // does not exist is an error from the API, not an empty list.
 func DescribeDBInstance(c *AWSClient, id string) (*rds.DBInstance, error) {

tests/e2e/harness/traffic.go

@@ -83,38 +83,6 @@ func TCPRoundRobin(host string, port int, n int, timeout time.Duration) TrafficR
 	return r
 }
 
-// UDPRoundRobin sends n UDP datagrams to host:port, reads the echoed line per
-// datagram, and counts responders. Matches the app-userdata UDP echo responder.
-// UDP is connectionless, so each probe writes a payload then reads with a
-// deadline; a dropped datagram counts as a failure rather than blocking.
-func UDPRoundRobin(host string, port int, n int, timeout time.Duration) TrafficResult {
-	r := TrafficResult{Distribution: map[string]int{}, Total: n}
-	for i := 0; i < n; i++ {
-		conn, err := net.DialTimeout("udp", fmt.Sprintf("%s:%d", host, port), timeout)
-		if err != nil {
-			continue
-		}
-		if _, err := conn.Write([]byte("ping")); err != nil {
-			conn.Close()
-			continue
-		}
-		conn.SetReadDeadline(time.Now().Add(timeout))
-		buf := make([]byte, 256)
-		nb, err := conn.Read(buf)
-		conn.Close()
-		if err != nil {
-			continue
-		}
-		resp := strings.TrimSpace(string(buf[:nb]))
-		if resp == "" {
-			continue
-		}
-		r.Distribution[resp]++
-		r.Successful++
-	}
-	return r
-}
-
 // AssertRoundRobin asserts that r has at least minUnique distinct responders
 // and at least minSuccess successful probes. Logs the distribution either way.
 func AssertRoundRobin(t *testing.T, r TrafficResult, minUnique, minSuccess int, label string) {

tests/integration/ecr_daemon_lite.go

@@ -7,15 +7,16 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/domains/ecr"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/require"
 )
 
 // StartECRDaemonLite subscribes a real ecr.MetaServiceImpl (the same
 // production service the live daemon runs, see runtime/daemon/daemon.go)
 // to every ecr.* NATS subject the gateway's ECR control-plane handlers and
 // ECRRegistry.Meta dial for metadata — repos, policies, lifecycle policies,
-// tags, manifest records, and in-progress upload state. It is memory-backed
+// tags, manifest records, and in-progress upload state. It is backed by
-// (ecr.NewMemoryMetaStore) rather than JetStream-KV-backed, matching
+// JetStream KV on the test's isolated NATS account, like the live daemon.
-// DaemonLite's in-scope resources.
 //
 // Blob and manifest bytes never travel these subjects (ECRRegistry.Store
 // handles those directly against its own objectstore.ObjectStore, wired in
@@ -28,8 +29,10 @@ import (
 func StartECRDaemonLite(t *testing.T, gw *Gateway) *ecr.MetaServiceImpl {
 	t.Helper()
 
-	svc := ecr.NewMetaServiceImpl(ecr.NewMemoryMetaStore())
 	nc := gw.NATSConn
+	js, err := jetstream.New(nc)
+	require.NoError(t, err, "jetstream handle for the ECR meta store")
+	svc := ecr.NewKVMetaService(js)
 
 	sub(t, nc, ecr.SubjectRepoCreate, func(m *nats.Msg) { dispatch(m, svc.RepoCreate) })
 	sub(t, nc, ecr.SubjectRepoDescribe, func(m *nats.Msg) { dispatch(m, svc.RepoDescribe) })

tests/integration/lambda_test.go

@@ -59,11 +59,9 @@ func startLambdaDaemonLite(t *testing.T, nc *nats.Conn) {
 	sub(t, nc, lambdafunction.SubjectRollbackCreate, func(m *nats.Msg) { dispatch(m, svc.RollbackCreate) })
 }
 
-// TestLambda_SignedCreateAndGetDownloadsTheStoredCode drives CreateFunction and
+// Signed Create and the Get reads over NATS metadata and the Predastore
-// GetFunction through signed ingress, the NATS-owned function record and the
+// fixture's code store, then downloads Code.Location. The fixture's service
-// real code store on the Predastore fixture, then downloads Code.Location.
+// key stands in for the scoped code-store session, which is untested.
-// The fixture authenticates only its service key, so that key stands in for
-// the scoped code-store session; provisioning and session scoping are not here.
 func TestLambda_SignedCreateAndGetDownloadsTheStoredCode(t *testing.T) {
 	fixture := testpredastore.Start(t)
 	bucket := codestore.BucketName(testRegion)
@@ -132,6 +130,19 @@ func TestLambda_SignedCreateAndGetDownloadsTheStoredCode(t *testing.T) {
 	require.Equal(t, "e2e", aws.StringValue(got.Configuration.Environment.Variables["MODE"]))
 	require.NotNil(t, got.Code.Location)
 
+	cfg, err := cli.GetFunctionConfiguration(&lambda.GetFunctionConfigurationInput{FunctionName: aws.String("e2e-fn")})
+	require.NoError(t, err)
+	require.Equal(t, got.Configuration.String(), cfg.String())
+
+	signing, err := cli.GetFunctionCodeSigningConfig(&lambda.GetFunctionCodeSigningConfigInput{FunctionName: aws.String("e2e-fn")})
+	require.NoError(t, err)
+	require.Nil(t, signing.CodeSigningConfigArn)
+	require.Nil(t, signing.FunctionName)
+	_, err = cli.GetFunctionCodeSigningConfig(&lambda.GetFunctionCodeSigningConfigInput{FunctionName: aws.String("e2e-missing")})
+	var aerr awserr.Error
+	require.ErrorAs(t, err, &aerr)
+	require.Equal(t, lambda.ErrCodeResourceNotFoundException, aerr.Code())
+
 	// SSL_CERT_FILE (set by the fixture) makes the default client trust it.
 	resp, err := http.Get(aws.StringValue(got.Code.Location)) //nolint:gosec,noctx // URL is the presigned location under test
 	require.NoError(t, err)