feat(accountteardown): seal EC2 account settings during termination (#1242)

Wire the EC2 account-settings seal into account termination as a
StageAccount reaper. Immediately before every seal attempt the reaper
probes each configured node's new read-only, per-node seal-capability
subject and requires exactly one identified reply per node at capability
version 1 or later. Missing, duplicate, wrong-identity, malformed, error,
older-version or unreachable responders send no seal and return a
retryable ServiceUnavailable, so the account stays TERMINATING until the
cluster is compatible.

The reaper completes only on a sealed reply for the account with a
non-zero revision; a fresh termination run seals again. Daemons validate
their node ID before subscribing, so an invalid ID serves no capability
and reads as incompatible.

contracts/ec2/v1/README.md

@@ -60,3 +60,42 @@ The subject name carries the contract major; `platform_inventory_test.go` preser
 Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1.
 A daemon that predates the subject has no subscription, so the request gets no responder; the consumer treats no responder, a timeout, an error reply, an undecodable reply, a reply for another account, and a missing, repeated or unknown-status kind as `scan-incomplete`, never clean.
 An incompatible change to the subject, fields or vocabulary needs a new major.
+
+## Seal account settings
+
+`SealAccountSettingsSubject` (`ec2.v1.SealAccountSettings`) is a request/reply route on which a daemon in the `spinifex-workers` queue group irreversibly seals one account's EC2 settings record.
+EC2 owns it: `domains/ec2/account` serves it through the daemon.
+`accountteardown`'s StageAccount seal reaper is the consumer: it calls this subject only after the seal-capability probe below reports every configured node compatible, and completes only once it holds a successful sealed reply.
+
+The request names the account in its body (`account_id`) and may carry an opaque `correlation_ref` that is echoed for logs.
+The daemon ignores `X-Account-ID` for this subject: the account sealed is the one named in the body.
+A seal is refused with `IncorrectState` unless IAM already holds the account as `TERMINATING`; a precondition read failure answers `InternalError`.
+Both refusals write nothing.
+
+A reply is sent only on success: `disposition` is always `sealed` and `revision` is the stream leader's accepted write for that seal.
+A failure is the standard awserrors error payload, not this reply shape.
+Sealing is idempotent: a seal on an already-sealed record re-writes the same value and returns a fresh, higher revision.
+
+### Compatibility boundary
+
+The subject name carries the contract major; `account_settings_seal_test.go` preserves the version-1 JSON shape.
+Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1.
+A daemon that predates the subject has no subscription, so the request gets no responder.
+An incompatible change to the subject or fields needs a new major.
+
+## Seal-account-settings capability
+
+`AccountSettingsSealCapabilitySubject(node)` (`ec2.v1.AccountSettingsSealCapability.<node>`) is a per-node, read-only request/reply route on which `accountteardown` asks one configured daemon whether it supports the account-settings seal, immediately before every seal attempt.
+EC2 owns it: the daemon serves it with no queue group, so a request addressed to one node reaches only subscribers configured as that node, and two daemons sharing a node ID both answer.
+`accountteardown` is the consumer: its StageAccount seal reaper probes every configured node and sends no seal request unless every one answers compatible.
+
+`AccountSettingsSealCapabilityVersion` is a claim about the whole binary's writers, not about the responder: version 1 means every EC2 account-settings writer in the binary is revision-guarded and refuses a sealed record, and the binary serves the seal subject.
+A daemon binary without this responder — predating the capability, or carrying only the seal subject without it — has no subscriber, so the probe for that node reads as incomplete (`no-reply`), the same as a stopped or partitioned node.
+The reply proves nothing beyond the moment it was sent: a daemon downgraded or replaced after a successful probe is outside the guarantee.
+
+### Compatibility boundary
+
+The subject name carries the contract major; `account_settings_seal_capability_test.go` preserves the version-1 JSON shape and the subject builder.
+Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1.
+A daemon that predates the capability has no subscription, so the probe gets no responder for that node.
+An incompatible change to the subject or fields needs a new major.

contracts/ec2/v1/account_settings_seal.go

@@ -0,0 +1,28 @@
+package ec2v1
+
+// SealAccountSettingsSubject is the request/reply subject on which a daemon
+// seals one account's EC2 settings record. Sealing is irreversible and refused
+// unless IAM already holds the account as TERMINATING.
+const SealAccountSettingsSubject = "ec2.v1.SealAccountSettings"
+
+// AccountSettingsDispositionSealed is the only disposition a successful reply
+// carries: a reply is sent only on success, so there is no other value.
+const AccountSettingsDispositionSealed = "sealed"
+
+// SealAccountSettingsRequest asks for one account's EC2 settings to be
+// sealed. CorrelationRef is an opaque caller reference echoed for logs; it
+// grants nothing and is never a fence or authority to write.
+type SealAccountSettingsRequest struct {
+	AccountID      string `json:"account_id"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+}
+
+// SealAccountSettingsReply confirms accountID's settings record was sealed at
+// Revision, the stream leader's accepted write for that seal. A failure is
+// the standard awserrors error payload, not this type.
+type SealAccountSettingsReply struct {
+	AccountID      string `json:"account_id"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+	Disposition    string `json:"disposition"`
+	Revision       uint64 `json:"revision"`
+}

contracts/ec2/v1/account_settings_seal_capability.go

@@ -0,0 +1,39 @@
+package ec2v1
+
+import "strings"
+
+// AccountSettingsSealCapabilityVersion is the version this binary reports.
+// Version 1: every EC2 account-settings writer in the binary is
+// revision-guarded and refuses a sealed record, and the seal subject is served.
+const AccountSettingsSealCapabilityVersion = 1
+
+// AccountSettingsSealCapabilitySubject is one daemon's read-only
+// seal-capability route, addressed per node so a probe reaches exactly the
+// node it asks about.
+func AccountSettingsSealCapabilitySubject(node string) string {
+	return "ec2.v1.AccountSettingsSealCapability." + node
+}
+
+// ValidSealCapabilityNode reports whether node is exactly one NATS subject
+// token. Responder and probe both check it before building a subject, so
+// neither can address a wildcard or another node's route.
+func ValidSealCapabilityNode(node string) bool {
+	return node != "" && !strings.ContainsAny(node, ".*>\t\n\r ")
+}
+
+// AccountSettingsSealCapabilityRequest asks node whether it supports the
+// account-settings seal. CorrelationRef is an opaque caller reference echoed
+// for logs; it grants nothing and is never a fence or authority to write.
+type AccountSettingsSealCapabilityRequest struct {
+	Node           string `json:"node"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+}
+
+// AccountSettingsSealCapabilityReply reports the responding node's capability
+// version. It carries no account data and proves nothing beyond the moment it
+// was sent.
+type AccountSettingsSealCapabilityReply struct {
+	Node              string `json:"node"`
+	CapabilityVersion int    `json:"capability_version"`
+	CorrelationRef    string `json:"correlation_ref,omitempty"`
+}

contracts/ec2/v1/account_settings_seal_capability_test.go

@@ -0,0 +1,43 @@
+package ec2v1
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAccountSettingsSealCapabilitySubject(t *testing.T) {
+	require.Equal(t, "ec2.v1.AccountSettingsSealCapability.node-1", AccountSettingsSealCapabilitySubject("node-1"))
+	require.Equal(t, "ec2.v1.AccountSettingsSealCapability.node-2", AccountSettingsSealCapabilitySubject("node-2"))
+}
+
+func TestAccountSettingsSealCapabilityRequestJSONContract(t *testing.T) {
+	data, err := json.Marshal(AccountSettingsSealCapabilityRequest{Node: "node-1", CorrelationRef: "req-1"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"node":"node-1","correlation_ref":"req-1"}`, string(data))
+
+	data, err = json.Marshal(AccountSettingsSealCapabilityRequest{Node: "node-1"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"node":"node-1"}`, string(data))
+}
+
+func TestAccountSettingsSealCapabilityReplyJSONContract(t *testing.T) {
+	reply := AccountSettingsSealCapabilityReply{Node: "node-1", CapabilityVersion: 1, CorrelationRef: "req-1"}
+	data, err := json.Marshal(reply)
+	require.NoError(t, err)
+	require.JSONEq(t, `{"node":"node-1","capability_version":1,"correlation_ref":"req-1"}`, string(data))
+
+	data, err = json.Marshal(AccountSettingsSealCapabilityReply{Node: "node-1", CapabilityVersion: 1})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"node":"node-1","capability_version":1}`, string(data))
+}
+
+func TestValidSealCapabilityNode(t *testing.T) {
+	for _, node := range []string{"node-1", "node_2", "n3"} {
+		require.True(t, ValidSealCapabilityNode(node), node)
+	}
+	for _, node := range []string{"", "a.b", "*", ">", "node-*", "node >", "node\t1", "node\n1"} {
+		require.False(t, ValidSealCapabilityNode(node), "%q", node)
+	}
+}

contracts/ec2/v1/account_settings_seal_test.go

@@ -0,0 +1,43 @@
+package ec2v1
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSealAccountSettingsSubject(t *testing.T) {
+	require.Equal(t, "ec2.v1.SealAccountSettings", SealAccountSettingsSubject)
+}
+
+func TestSealAccountSettingsRequestJSONContract(t *testing.T) {
+	data, err := json.Marshal(SealAccountSettingsRequest{AccountID: "111122223333", CorrelationRef: "req-1"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333","correlation_ref":"req-1"}`, string(data))
+
+	data, err = json.Marshal(SealAccountSettingsRequest{AccountID: "111122223333"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333"}`, string(data))
+}
+
+func TestSealAccountSettingsReplyJSONContract(t *testing.T) {
+	reply := SealAccountSettingsReply{
+		AccountID:      "111122223333",
+		CorrelationRef: "req-1",
+		Disposition:    AccountSettingsDispositionSealed,
+		Revision:       42,
+	}
+	data, err := json.Marshal(reply)
+	require.NoError(t, err)
+	require.JSONEq(t, `{
+		"account_id": "111122223333",
+		"correlation_ref": "req-1",
+		"disposition": "sealed",
+		"revision": 42
+	}`, string(data))
+
+	data, err = json.Marshal(SealAccountSettingsReply{AccountID: "111122223333", Disposition: AccountSettingsDispositionSealed})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333","disposition":"sealed","revision":0}`, string(data))
+}

docs/PACKAGE_BOUNDARY_MIGRATION.md

@@ -446,7 +446,7 @@ These residuals were recorded by individual slices and must be resolved once the
 - `providers/ebs/viperblock` → `runtime/compute/nbd` and `runtime/host/process` (dependency, moved unchanged from `services/viperblockd`): the adapter formats NBD endpoints and manages nbdkit processes through runtime packages. ADR-0001 does not list providers → runtime as forbidden and the layering test does not enforce it (same category as `providers/ebs/conformance` → `runtime/compute/nbd`), but a provider should not depend on node runtime. Clears when NBD endpoint naming has a provider-facing home (a candidate for `contracts/ebs/v1`) and process supervision is injected by the role. The adapter also imports `providers/objectstore`, a provider-to-provider edge that is allowed.
 - `gateway/ec2.go` (registration): the EC2 dispatch map stays a `GatewayConfig` method in `gateway` and imports every `domains/ec2/awsapi/*` package; `gateway/ec2_idempotency.go` also imports `domains/ec2/awsapi/idem` (`KVBucket`, `TTL`, `TokenAndParams`) for the dispatch-time idempotency wrapper. ADR-0001 makes the domain own its action registration; moving the map waits on the generic ingress registration seam and must not be done as a package move.
 - `gateway/requestaudit.go`: the outermost request-audit middleware records AWS fields and is updated from four places in the gateway's authentication flow through the gateway's own context key. Moving it would export about seven names used only by that flow, so it stays in `gateway` until the generic ingress registration seam defines how ingress middleware receives authentication results.
-- `accountteardown` (dependency): its EC2 reapers call `domains/ec2/awsapi/*` action adapters directly, as they did under `gateway/ec2`. The ADR's `accountteardown` current-to-target map row replaces this with domain-owned cleanup capabilities before the package moves. It also holds a read-only consumer of the EC2 platform-inventory contract (`contracts/ec2/v1`, `ec2.v1.PlatformInventory`), served by `domains/ec2/accountinventory`; only the explicit dry run (`Engine.DryRun`) calls it, to log a shadow comparison, and the key-pair, placement-group and launch-template reapers still use the `awsapi` adapters. A domain-owned cleanup capability for the proven-safe subset of those three kinds (key pairs and empty placement groups no instance, spot request or live launch-template version names, and launch-template residue; never live launch templates) exists as `domains/ec2/accountcleanup` (a Go interface with injected owners and reference readers, no NATS subject); nothing calls it yet.
+- `accountteardown` (dependency): its EC2 reapers call `domains/ec2/awsapi/*` action adapters directly, as they did under `gateway/ec2`. The ADR's `accountteardown` current-to-target map row replaces this with domain-owned cleanup capabilities before the package moves. It also holds a read-only consumer of the EC2 platform-inventory contract (`contracts/ec2/v1`, `ec2.v1.PlatformInventory`), served by `domains/ec2/accountinventory`; only the explicit dry run (`Engine.DryRun`) calls it, to log a shadow comparison, and the key-pair, placement-group and launch-template reapers still use the `awsapi` adapters. A domain-owned cleanup capability for the proven-safe subset of those three kinds (key pairs and empty placement groups no instance, spot request or live launch-template version names, and launch-template residue; never live launch templates) exists as `domains/ec2/accountcleanup` (a Go interface with injected owners and reference readers, no NATS subject); nothing calls it yet. It also holds the StageAccount settings-seal reaper, a consumer of `contracts/ec2/v1`'s `ec2.v1.SealAccountSettings` and per-node `ec2.v1.AccountSettingsSealCapability.<node>` subjects, served by `domains/ec2/account` through the daemon; the reaper reads no KV and imports no domain package, reaching both subjects only through NATS and the configured node set `NewClusterEngine` already receives.
 - `domains/admission/quota` (dependency, moved from `handlers/quota` unchanged): `live.go` imports `gateway/elbv2` and `handlers/rds` for its live ELBv2 and RDS counters; `records.go` imports `runtime/compute/vm` for the instance-record vCPU sweep; `live.go`, `vcpu.go` and `storage.go` call `domains/ec2/awsapi/{vpc,eip,volume}` describe adapters to count VPCs, subnets, addresses and volumes, as they did under `gateway/ec2`. The ADR's `handlers/quota` current-to-target map row replaces these with reader capabilities once their final locations exist.
 - `domains/ec2/awsapi/instance` (dependency): imports `handlers/iam` (`IAMService` in `DescribeInstances.go`, `RunInstances.go` and `IamInstanceProfileAssociation.go`; `InstanceProfile` in the latter two), `runtime/compute/cache` (`NodeState`, `NodeUnknown`, `NodeLive`) and `runtime/compute/vm` (`VM`), both in `status_synthesis.go`. The IAM edge needs an IAM capability; the runtime edges are domain-to-runtime and need the instance record and node liveness exposed through an EC2-owned projection.
 - `domains/ec2/awsapi/spotinstance` (dependency): imports `handlers/iam.IAMService` and `domains/admission/quota.Service` in `RequestSpotInstances.go`.

spinifex/accountteardown/cluster.go

@@ -70,6 +70,11 @@ func NewClusterEngine(
 	reapers = append(reapers, S3Reapers(buckets)...)
 	reapers = append(reapers, ACMReapers(nc)...)
 	reapers = append(reapers, IAMReapers(svc)...)
+
+	// Before the account counters and reservations: the seal must still be
+	// able to find the account IAM holds as TERMINATING, which the account
+	// record reaper does not remove.
+	reapers = append(reapers, EC2AccountSettingsReapers(nc, nodeIDs)...)
 	reapers = append(reapers, AccountReapers(svc, names, usage)...)
 
 	engine := NewEngine(NewIAMAccounts(svc), reapers...)

spinifex/accountteardown/cluster_test.go

@@ -41,6 +41,27 @@ func TestNewClusterEngineCoversEveryStage(t *testing.T) {
 	}
 }
 
+// The EC2 settings seal reaper must be registered in StageAccount on every
+// surface that builds a cluster engine, or teardown would finalise accounts
+// without ever sealing their settings.
+func TestNewClusterEngineRegistersTheEC2SettingsSealReaper(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+
+	masterKey, err := handlers_iam.GenerateMasterKey()
+	require.NoError(t, err)
+	svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey)
+	require.NoError(t, err)
+
+	engine, err := NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, newFakeBuckets())
+	require.NoError(t, err)
+
+	var kinds []string
+	for _, reaper := range engine.reapersFor(StageAccount) {
+		kinds = append(kinds, reaper.Kind())
+	}
+	assert.Contains(t, kinds, "ec2-account-settings")
+}
+
 // A teardown must refuse the accounts no credential may delete before it reads
 // anything, and refuse a name confirmation that does not match the record.
 func TestClusterEnginePrechecks(t *testing.T) {

spinifex/accountteardown/ec2_account_settings_seal.go

@@ -0,0 +1,125 @@
+package accountteardown
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// ec2AccountSettingsSealTimeout bounds one seal request; an earlier caller
+// deadline wins.
+const ec2AccountSettingsSealTimeout = 30 * time.Second
+
+// ec2AccountSettingsSealer is EC2's seal of one account's settings record.
+type ec2AccountSettingsSealer interface {
+	SealAccountSettings(ctx context.Context, accountID string) (revision uint64, err error)
+}
+
+// ec2AccountSettingsSealClient reaches the EC2 settings-seal contract over
+// NATS. Only a reply for this account, disposition sealed and a non-zero
+// revision is a seal; anything else, including no responder, is an error.
+type ec2AccountSettingsSealClient struct {
+	nc      *nats.Conn
+	timeout time.Duration
+}
+
+var _ ec2AccountSettingsSealer = (*ec2AccountSettingsSealClient)(nil)
+
+func newEC2AccountSettingsSealClient(nc *nats.Conn) *ec2AccountSettingsSealClient {
+	return &ec2AccountSettingsSealClient{nc: nc, timeout: ec2AccountSettingsSealTimeout}
+}
+
+func (c *ec2AccountSettingsSealClient) SealAccountSettings(ctx context.Context, accountID string) (uint64, error) {
+	timeout := c.timeout
+	if deadline, ok := ctx.Deadline(); ok {
+		timeout = min(timeout, time.Until(deadline))
+	}
+	if ctx.Err() != nil || timeout <= 0 {
+		return 0, fmt.Errorf("seal account settings: %w", context.DeadlineExceeded)
+	}
+
+	reply, err := natsmsg.NATSRequest[ec2v1.SealAccountSettingsReply](ctx, c.nc, ec2v1.SealAccountSettingsSubject,
+		&ec2v1.SealAccountSettingsRequest{AccountID: accountID}, timeout, accountID)
+	if err != nil {
+		return 0, fmt.Errorf("seal account settings: %w", err)
+	}
+	if reply.AccountID != accountID {
+		return 0, fmt.Errorf("seal account settings: reply named account %q, not %q", reply.AccountID, accountID)
+	}
+	if reply.Disposition != ec2v1.AccountSettingsDispositionSealed {
+		return 0, fmt.Errorf("seal account settings: unexpected disposition %q", reply.Disposition)
+	}
+	if reply.Revision == 0 {
+		return 0, errors.New("seal account settings: reply carried no revision")
+	}
+	return reply.Revision, nil
+}
+
+// accountSettingsSealReaper seals EC2's settings record in StageAccount. List
+// does no I/O and stays pending until this instance has sealed, so DryRun never
+// seals; state is per engine, so every fresh teardown run seals again.
+type accountSettingsSealReaper struct {
+	sealer     ec2AccountSettingsSealer
+	capability sealCapabilityChecker
+
+	mu     sync.Mutex
+	sealed map[string]uint64 // account ID -> leader revision, this run only
+}
+
+func (r *accountSettingsSealReaper) Kind() string { return "ec2-account-settings" }
+func (r *accountSettingsSealReaper) Stage() Stage { return StageAccount }
+
+func (r *accountSettingsSealReaper) alreadySealed(accountID string) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	_, ok := r.sealed[accountID]
+	return ok
+}
+
+func (r *accountSettingsSealReaper) List(_ context.Context, accountID string) ([]Resource, error) {
+	if r.alreadySealed(accountID) {
+		return nil, nil
+	}
+	return []Resource{{Kind: r.Kind(), ID: accountID, Detail: "seal"}}, nil
+}
+
+// Delete probes every configured daemon immediately before each seal attempt,
+// never reusing an earlier probe. Incomplete capability sends no seal and is
+// retryable; force is ignored, since nothing may force past the probe.
+func (r *accountSettingsSealReaper) Delete(ctx context.Context, accountID string, _ Resource, _ bool) error {
+	report := r.capability.CheckSealCapability(ctx, accountID)
+	if !report.Complete {
+		return errSealCapabilityIncomplete(report)
+	}
+
+	revision, err := r.sealer.SealAccountSettings(ctx, accountID)
+	if err != nil {
+		return err
+	}
+
+	r.mu.Lock()
+	if r.sealed == nil {
+		r.sealed = map[string]uint64{}
+	}
+	r.sealed[accountID] = revision
+	r.mu.Unlock()
+	slog.InfoContext(ctx, "Account teardown sealed EC2 account settings", "accountID", accountID, "revision", revision)
+	return nil
+}
+
+// EC2AccountSettingsReapers returns the StageAccount reaper that seals EC2's
+// settings record. nodeIDs is the configured node set the capability probe
+// treats as "every relevant writer"; an empty set is unknown, never complete.
+func EC2AccountSettingsReapers(nc *nats.Conn, nodeIDs []string) []Reaper {
+	return []Reaper{&accountSettingsSealReaper{
+		sealer:     newEC2AccountSettingsSealClient(nc),
+		capability: newSealCapabilityClient(nc, nodeIDs),
+	}}
+}

spinifex/accountteardown/ec2_account_settings_seal_capability.go

@@ -0,0 +1,191 @@
+package accountteardown
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// sealCapabilityWait bounds the whole probe; shortened by the caller's
+// deadline. Every configured node is probed in parallel against this one
+// wait, so N nodes cost one wait, not N.
+const sealCapabilityWait = 3 * time.Second
+
+// sealCapabilityGrace is how long a node's probe keeps listening after its
+// one expected responder has answered, long enough for a genuine second
+// subscriber on the same node to be caught as a duplicate.
+const sealCapabilityGrace = 150 * time.Millisecond
+
+// Seal capability reasons, one per failing node. Each names a category, never
+// wire error text.
+const (
+	sealCapabilityReasonNoReply       = "no-reply"
+	sealCapabilityReasonDuplicate     = "duplicate"
+	sealCapabilityReasonWrongIdentity = "wrong-identity"
+	sealCapabilityReasonErrorReply    = "error-reply"
+	sealCapabilityReasonMalformed     = "malformed"
+	sealCapabilityReasonOldVersion    = "old-version"
+	sealCapabilityReasonTransport     = "transport"
+	sealCapabilityReasonInvalidNodeID = "invalid-node-id"
+	sealCapabilityReasonSetUnknown    = "configured-node-set-unknown"
+)
+
+// sealCapabilityChecker reports whether every configured daemon supports the
+// seal.
+type sealCapabilityChecker interface {
+	CheckSealCapability(ctx context.Context, correlationRef string) sealCapabilityReport
+}
+
+// sealCapabilityReport is one probe's result, one entry per configured node.
+type sealCapabilityReport struct {
+	Complete bool
+	Nodes    []nodeSealCapability // one per configured node, sorted by node
+}
+
+// nodeSealCapability is one configured node's compatibility. Reason is ""
+// when the node is compatible.
+type nodeSealCapability struct {
+	Node    string
+	Version int    // 0 unless a valid reply arrived
+	Reason  string // "" when compatible
+}
+
+// sealCapabilityClient probes every configured node's seal-capability subject
+// in parallel. "Configured" is the same node set NewClusterEngine already
+// receives; there is no separate liveness or membership source.
+type sealCapabilityClient struct {
+	nc      *nats.Conn
+	nodeIDs []string
+	wait    time.Duration // 3 s, shortened by the caller's deadline
+	grace   time.Duration // 150 ms ResponderGrace
+}
+
+var _ sealCapabilityChecker = (*sealCapabilityClient)(nil)
+
+func newSealCapabilityClient(nc *nats.Conn, nodeIDs []string) *sealCapabilityClient {
+	return &sealCapabilityClient{nc: nc, nodeIDs: nodeIDs, wait: sealCapabilityWait, grace: sealCapabilityGrace}
+}
+
+// CheckSealCapability probes every configured node in parallel, bounded by
+// one shared deadline, run immediately before each seal attempt. A cached or
+// earlier result is never reused.
+func (c *sealCapabilityClient) CheckSealCapability(ctx context.Context, correlationRef string) sealCapabilityReport {
+	if len(c.nodeIDs) == 0 {
+		return sealCapabilityReport{Complete: false}
+	}
+
+	wait := c.wait
+	if deadline, ok := ctx.Deadline(); ok {
+		wait = min(wait, time.Until(deadline))
+	}
+
+	nodes := append([]string(nil), c.nodeIDs...)
+	sort.Strings(nodes)
+
+	results := make([]nodeSealCapability, len(nodes))
+	if ctx.Err() != nil || wait <= 0 {
+		for i, node := range nodes {
+			results[i] = nodeSealCapability{Node: node, Reason: sealCapabilityReasonNoReply}
+		}
+		return sealCapabilityReport{Complete: false, Nodes: results}
+	}
+
+	var wg sync.WaitGroup
+	for i, node := range nodes {
+		wg.Add(1)
+		go func(i int, node string) {
+			defer wg.Done()
+			results[i] = c.checkNode(ctx, node, correlationRef, wait)
+		}(i, node)
+	}
+	wg.Wait()
+
+	complete := true
+	for _, r := range results {
+		if r.Reason != "" {
+			complete = false
+			break
+		}
+	}
+	return sealCapabilityReport{Complete: complete, Nodes: results}
+}
+
+// checkNode probes one node and classifies the result. A node is compatible
+// only when exactly one identified, decodable, matching, version >= 1 frame
+// arrived, with no duplicate, conflict or cap hit.
+func (c *sealCapabilityClient) checkNode(ctx context.Context, node, correlationRef string, wait time.Duration) nodeSealCapability {
+	if !ec2v1.ValidSealCapabilityNode(node) {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonInvalidNodeID}
+	}
+
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: node, CorrelationRef: correlationRef})
+	if err != nil {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonMalformed}
+	}
+
+	frames, sum, err := natsmsg.Gather(ctx, c.nc, ec2v1.AccountSettingsSealCapabilitySubject(node), data, natsmsg.GatherOpts{
+		Timeout:            wait,
+		ExpectedResponders: 1,
+		Mode:               natsmsg.CollectUntilDeadline,
+		ResponderGrace:     c.grace,
+	})
+	if err != nil {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonTransport}
+	}
+
+	switch {
+	case sum.CapHit, len(sum.ConflictNodes) > 0, sum.DuplicateFrames > 0:
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonDuplicate}
+	case sum.Unidentified > 0:
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonWrongIdentity}
+	case len(sum.ErrorResponders) > 0:
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonErrorReply}
+	case sum.Received == 0, len(frames) == 0:
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonNoReply}
+	}
+
+	frame := frames[0]
+	if frame.NodeID != node {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonWrongIdentity}
+	}
+
+	var reply ec2v1.AccountSettingsSealCapabilityReply
+	if json.Unmarshal(frame.Data, &reply) != nil {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonMalformed}
+	}
+	if reply.Node != node {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonWrongIdentity}
+	}
+	if reply.CapabilityVersion < 1 {
+		return nodeSealCapability{Node: node, Version: reply.CapabilityVersion, Reason: sealCapabilityReasonOldVersion}
+	}
+	return nodeSealCapability{Node: node, Version: reply.CapabilityVersion}
+}
+
+// errSealCapabilityIncomplete builds the retryable refusal from report. It
+// names node IDs and reason categories only, never wire error text.
+func errSealCapabilityIncomplete(report sealCapabilityReport) error {
+	detail := sealCapabilityReasonSetUnknown
+	var parts []string
+	for _, n := range report.Nodes {
+		if n.Reason != "" {
+			parts = append(parts, n.Node+"="+n.Reason)
+		}
+	}
+	if len(parts) > 0 {
+		detail = strings.Join(parts, ", ")
+	}
+	return errors.Join(
+		awserrors.Errorf(awserrors.ErrorServiceUnavailable, "EC2 account-settings seal capability incomplete: %s", detail),
+		awserrors.RetryAfter(awserrors.ErrorServiceUnavailable, 5*time.Second),
+	)
+}

spinifex/accountteardown/ec2_account_settings_seal_capability_test.go

@@ -0,0 +1,299 @@
+package accountteardown
+
+//test:in-package — the probe client, its report types and reason constants
+// are unexported, and what they guarantee (incomplete never seals, complete
+// does) is only observable from inside the package.
+
+import (
+	"encoding/json"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// testCapabilityClient builds a probe client with test-sized timeouts so the
+// incomplete cases below do not each pay the production 3 s wait.
+func testCapabilityClient(nc *nats.Conn, nodeIDs []string) *sealCapabilityClient {
+	c := newSealCapabilityClient(nc, nodeIDs)
+	c.wait = 200 * time.Millisecond
+	c.grace = 30 * time.Millisecond
+	return c
+}
+
+// serveCapability subscribes reply on subject with no queue group, which is
+// how the real responder is addressed: per node, so two subscribers on the
+// same node's subject both answer.
+func serveCapability(t *testing.T, nc *nats.Conn, subject string, reply nats.MsgHandler) *nats.Subscription {
+	t.Helper()
+	sub, err := nc.Subscribe(subject, reply)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, nc.Flush())
+	return sub
+}
+
+// capabilityReply answers with a well-formed reply naming replyNode (which
+// may deliberately differ from the subject's node) and headerNode as the
+// X-Node-ID ("" omits the header).
+func capabilityReply(replyNode, headerNode string, version int) nats.MsgHandler {
+	return func(msg *nats.Msg) {
+		data, _ := json.Marshal(ec2v1.AccountSettingsSealCapabilityReply{Node: replyNode, CapabilityVersion: version})
+		reply := nats.NewMsg(msg.Reply)
+		reply.Data = data
+		if headerNode != "" {
+			reply.Header.Set(natsmsg.NodeIDHeader, headerNode)
+		}
+		_ = msg.RespondMsg(reply)
+	}
+}
+
+func nodeResult(t *testing.T, report sealCapabilityReport, node string) nodeSealCapability {
+	t.Helper()
+	for _, n := range report.Nodes {
+		if n.Node == node {
+			return n
+		}
+	}
+	t.Fatalf("no result for node %q in %+v", node, report)
+	return nodeSealCapability{}
+}
+
+// Each sub-case names exactly one way a node fails to prove capability
+// version 1, and that failure alone makes the whole probe incomplete.
+func TestSealCapabilityIncomplete(t *testing.T) {
+	t.Run("missing", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonNoReply, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("duplicate", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonDuplicate, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("unreachable timeout", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+			time.Sleep(time.Second)
+			_ = msg.Respond(nil)
+		})
+		client := testCapabilityClient(nc, []string{"node-1"})
+		report := client.CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonNoReply, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("transport closed connection", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		nc.Close()
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonTransport, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("malformed body", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = []byte("not json")
+			reply.Header.Set(natsmsg.NodeIDHeader, "node-1")
+			_ = msg.RespondMsg(reply)
+		})
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonMalformed, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("error reply", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal)
+			reply.Header.Set(natsmsg.NodeIDHeader, "node-1")
+			_ = msg.RespondMsg(reply)
+		})
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonErrorReply, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("wrong identity no header", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "", 1))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonWrongIdentity, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("wrong identity another node header", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-9", 1))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonWrongIdentity, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("wrong identity body node differs", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-9", "node-1", 1))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonWrongIdentity, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("lower version", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 0))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonOldVersion, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("configured set empty", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		report := testCapabilityClient(nc, nil).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Empty(t, report.Nodes)
+	})
+
+	t.Run("invalid node id", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		report := testCapabilityClient(nc, []string{"bad.node"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonInvalidNodeID, nodeResult(t, report, "bad.node").Reason)
+	})
+}
+
+// Every configured node answering version 1 or later makes the probe
+// complete, and the reaper sends exactly one seal request.
+func TestSealCapabilityCompleteAllowsSeal(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-2"), capabilityReply("node-2", "node-2", 2))
+
+	client := testCapabilityClient(nc, []string{"node-1", "node-2"})
+	report := client.CheckSealCapability(testCtx(t), "req")
+	require.True(t, report.Complete)
+	assert.Equal(t, 1, nodeResult(t, report, "node-1").Version)
+	assert.Equal(t, 2, nodeResult(t, report, "node-2").Version)
+
+	sealer := &fakeSealer{}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: client}
+	require.NoError(t, reaper.Delete(testCtx(t), testAccountID, Resource{Kind: reaper.Kind(), ID: testAccountID}, false))
+	assert.Equal(t, []string{testAccountID}, sealer.calls)
+}
+
+// An incomplete probe keeps the account TERMINATING through the full engine,
+// names the silent node and its reason, sends no seal request, and leaves the
+// settings record untouched (the fake sealer records no call).
+func TestSealCapabilityIncompleteKeepsAccountTerminating(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+	// node-2 has no subscriber.
+
+	sealer := &fakeSealer{}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: testCapabilityClient(nc, []string{"node-1", "node-2"})}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+
+	result, err := engine.Teardown(testCtx(t), request())
+	require.ErrorIs(t, err, ErrResourcesStuck)
+	assert.False(t, result.AccountDeleted)
+	assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
+	assert.Empty(t, sealer.calls)
+
+	var reason string
+	for _, stage := range result.Stages {
+		for _, stuck := range stage.Stuck {
+			if stuck.Resource.Kind == "ec2-account-settings" {
+				reason = stuck.Reason
+			}
+		}
+	}
+	assert.Contains(t, reason, "node-2="+sealCapabilityReasonNoReply)
+}
+
+// The silent node starting to answer during the drain makes the very next
+// Delete re-probe and succeed, which proves the result is retryable and the
+// probe is never cached across attempts.
+func TestSealCapabilityRetriesThenSeals(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+
+	var node2Up atomic.Bool
+	sub, err := nc.Subscribe(ec2v1.AccountSettingsSealCapabilitySubject("node-2"), func(msg *nats.Msg) {
+		if !node2Up.Load() {
+			return
+		}
+		capabilityReply("node-2", "node-2", 1)(msg)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, nc.Flush())
+
+	sealer := &fakeSealer{}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: testCapabilityClient(nc, []string{"node-1", "node-2"})}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+	engine.Timeouts = Timeouts{StageDrain: time.Second, DrainPoll: 5 * time.Millisecond}
+
+	go func() {
+		time.Sleep(80 * time.Millisecond)
+		node2Up.Store(true)
+	}()
+
+	result, err := engine.Teardown(testCtx(t), request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+	assert.Equal(t, []string{testAccountID}, sealer.calls)
+}
+
+// Two Delete calls on one reaper probe twice: the probe runs in Delete
+// itself, never cached from a previous attempt.
+func TestSealCapabilityProbedBeforeEverySeal(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	var probes int
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+		probes++
+		capabilityReply("node-1", "node-1", 1)(msg)
+	})
+
+	sealer := &fakeSealer{}
+	client := testCapabilityClient(nc, []string{"node-1"})
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: client}
+
+	require.NoError(t, reaper.Delete(testCtx(t), testAccountID, Resource{Kind: reaper.Kind(), ID: testAccountID}, false))
+	// alreadySealed now makes List empty, but Delete itself does not consult
+	// it, so a second direct call still probes and seals again.
+	require.NoError(t, reaper.Delete(testCtx(t), testAccountID, Resource{Kind: reaper.Kind(), ID: testAccountID}, false))
+	assert.Equal(t, 2, probes)
+	assert.Len(t, sealer.calls, 2)
+}
+
+// Five silent nodes still return within one deadline, not five: every node
+// is probed in parallel against the same wait.
+func TestSealCapabilityBoundedByOneDeadline(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	client := testCapabilityClient(nc, []string{"node-1", "node-2", "node-3", "node-4", "node-5"})
+
+	started := time.Now()
+	report := client.CheckSealCapability(testCtx(t), "req")
+	elapsed := time.Since(started)
+
+	assert.False(t, report.Complete)
+	assert.Less(t, elapsed, client.wait+time.Second, "five silent nodes must not cost five deadlines")
+}

spinifex/accountteardown/ec2_account_settings_seal_test.go

@@ -0,0 +1,273 @@
+package accountteardown
+
+//test:in-package — the reaper, sealer and capability checker are unexported;
+// no seal without a complete probe, completion only after a sealed reply and
+// idempotent fresh runs are only observable from inside the package.
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	ec2account "github.com/mulgadc/spinifex/spinifex/domains/ec2/account"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// fakeSealer is a recording ec2AccountSettingsSealer. Each call beyond
+// failFor succeeds with a fresh, strictly increasing revision, unless err is
+// set, in which case every call after failFor fails with it.
+type fakeSealer struct {
+	mu       sync.Mutex
+	calls    []string
+	failFor  int
+	err      error
+	revision uint64
+}
+
+func (f *fakeSealer) SealAccountSettings(_ context.Context, accountID string) (uint64, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.calls = append(f.calls, accountID)
+	if len(f.calls) <= f.failFor {
+		return 0, errors.New("account settings owner temporarily unavailable")
+	}
+	if f.err != nil {
+		return 0, f.err
+	}
+	f.revision++
+	return f.revision, nil
+}
+
+// fakeSealCapability is a recording sealCapabilityChecker returning a fixed report.
+type fakeSealCapability struct {
+	calls  int
+	report sealCapabilityReport
+}
+
+func (f *fakeSealCapability) CheckSealCapability(context.Context, string) sealCapabilityReport {
+	f.calls++
+	return f.report
+}
+
+// completeReport builds a sealCapabilityReport where every named node answers
+// compatible at version 1.
+func completeReport(nodeIDs ...string) sealCapabilityReport {
+	var nodes []nodeSealCapability
+	for _, node := range nodeIDs {
+		nodes = append(nodes, nodeSealCapability{Node: node, Version: 1})
+	}
+	return sealCapabilityReport{Complete: true, Nodes: nodes}
+}
+
+// alwaysTerminatingChecker stands in for IAM in the end-to-end test: the
+// account under test is always TERMINATING.
+type alwaysTerminatingChecker struct{}
+
+func (alwaysTerminatingChecker) AccountTerminating(context.Context, string) (bool, error) {
+	return true, nil
+}
+
+// No subscriber on either subject makes the capability probe incomplete, so
+// no seal request is ever sent and the account stays TERMINATING with one
+// stuck ec2-account-settings item.
+func TestSealReaperNoResponderKeepsAccountTerminating(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+
+	capability := newSealCapabilityClient(nc, []string{"node-1"})
+	capability.wait = 50 * time.Millisecond
+	capability.grace = 10 * time.Millisecond
+	sealer := newEC2AccountSettingsSealClient(nc)
+	sealer.timeout = 50 * time.Millisecond
+
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: capability}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+
+	result, err := engine.Teardown(testCtx(t), request())
+	require.ErrorIs(t, err, ErrResourcesStuck)
+	assert.False(t, result.AccountDeleted)
+	assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
+
+	var stuck []Stuck
+	for _, stage := range result.Stages {
+		stuck = append(stuck, stage.Stuck...)
+	}
+	require.Len(t, stuck, 1)
+	assert.Equal(t, "ec2-account-settings", stuck[0].Resource.Kind)
+}
+
+// Whatever the owner refuses with, the account stays TERMINATING: an owner
+// refusal, a timeout, and each reply shape the client rejects (wrong account,
+// wrong disposition, no revision).
+func TestSealReaperOwnerFailureKeepsAccountTerminating(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{"incorrect-state", errors.New(awserrors.ErrorIncorrectState)},
+		{"internal-error", errors.New(awserrors.ErrorInternalError)},
+		{"timeout", context.DeadlineExceeded},
+		{"wrong-account", errors.New(`seal account settings: reply named account "999999999999", not "000000000042"`)},
+		{"wrong-disposition", errors.New(`seal account settings: unexpected disposition "pending"`)},
+		{"zero-revision", errors.New("seal account settings: reply carried no revision")},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			sealer := &fakeSealer{err: tc.err}
+			reaper := &accountSettingsSealReaper{sealer: sealer, capability: &fakeSealCapability{report: completeReport("node-1")}}
+			accounts := newFakeAccounts()
+			engine := quietEngine(accounts, reaper)
+
+			result, err := engine.Teardown(testCtx(t), request())
+			require.ErrorIs(t, err, ErrResourcesStuck)
+			assert.False(t, result.AccountDeleted)
+			assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
+			assert.NotEmpty(t, sealer.calls)
+		})
+	}
+}
+
+// The reaper retries through the engine's own drain mechanics: it fails
+// twice, then succeeds, and only the success lets the account be deleted.
+// List is non-empty before the success and empty only after it.
+func TestSealReaperCompletesOnlyAfterSealed(t *testing.T) {
+	sealer := &fakeSealer{failFor: 2}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: &fakeSealCapability{report: completeReport("node-1")}}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+
+	before, err := reaper.List(testCtx(t), testAccountID)
+	require.NoError(t, err)
+	require.Len(t, before, 1)
+
+	result, err := engine.Teardown(testCtx(t), request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+	assert.GreaterOrEqual(t, len(sealer.calls), 3)
+
+	after, err := reaper.List(testCtx(t), testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, after)
+}
+
+// Each teardown builds a fresh engine and a fresh reaper, so a retried or
+// taken-over run seals again: two engines against the same owner both
+// succeed, each with a higher revision than the last.
+func TestSealReaperFreshRunSealsAgain(t *testing.T) {
+	sealer := &fakeSealer{}
+	capability := &fakeSealCapability{report: completeReport("node-1")}
+
+	firstEngine := quietEngine(newFakeAccounts(), &accountSettingsSealReaper{sealer: sealer, capability: capability})
+	firstResult, err := firstEngine.Teardown(testCtx(t), request())
+	require.NoError(t, err)
+	assert.True(t, firstResult.AccountDeleted)
+
+	secondEngine := quietEngine(newFakeAccounts(), &accountSettingsSealReaper{sealer: sealer, capability: capability})
+	secondResult, err := secondEngine.Teardown(testCtx(t), request())
+	require.NoError(t, err)
+	assert.True(t, secondResult.AccountDeleted)
+
+	require.Len(t, sealer.calls, 2)
+	assert.Equal(t, uint64(2), sealer.revision)
+}
+
+// Inventory and the explicit dry run, and a Teardown request marked dry run,
+// all list the pending seal and send no seal request: only a real teardown
+// may seal.
+func TestSealReaperInventoryDoesNotSeal(t *testing.T) {
+	sealer := &fakeSealer{}
+	capability := &fakeSealCapability{report: completeReport("node-1")}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: capability}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+
+	inventory, err := engine.Inventory(testCtx(t), testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, sealer.calls)
+	assert.Zero(t, capability.calls)
+	found := false
+	for _, stage := range inventory.Stages {
+		for _, resource := range stage.Deleted {
+			if resource.Kind == "ec2-account-settings" {
+				found = true
+			}
+		}
+	}
+	assert.True(t, found, "inventory must still list the pending seal")
+
+	dryRun := request()
+	dryRun.DryRun = true
+	_, err = engine.Teardown(testCtx(t), dryRun)
+	require.NoError(t, err)
+	assert.Empty(t, sealer.calls)
+	assert.Zero(t, capability.calls)
+}
+
+// A real settings service subscribed on the real seal subject, behind a real
+// engine: teardown seals the record, replacing the characterisation test that
+// showed it used to survive.
+func TestTeardownSealsAccountSettings(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	ctx := testCtx(t)
+
+	settings, err := ec2account.NewAccountSettingsServiceImplWithNATS(ctx, nil, nc)
+	require.NoError(t, err)
+	settings.SetAccountTerminationChecker(alwaysTerminatingChecker{})
+	_, err = settings.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	sealSub, err := nc.QueueSubscribe(ec2v1.SealAccountSettingsSubject, "spinifex-workers", func(msg *nats.Msg) {
+		natsmsg.ServeNATSRequestCtx(msg, func(ctx context.Context, req *ec2v1.SealAccountSettingsRequest) (*ec2v1.SealAccountSettingsReply, error) {
+			seal, err := settings.SealAccount(ctx, req.AccountID)
+			if err != nil {
+				return nil, err
+			}
+			return &ec2v1.SealAccountSettingsReply{
+				AccountID: req.AccountID, Disposition: ec2v1.AccountSettingsDispositionSealed, Revision: seal.Revision,
+			}, nil
+		})
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sealSub.Unsubscribe() })
+
+	capSub, err := nc.Subscribe(ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+		reply, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityReply{Node: "node-1", CapabilityVersion: 1})
+		require.NoError(t, err)
+		out := nats.NewMsg("")
+		out.Data = reply
+		out.Header.Set(natsmsg.NodeIDHeader, "node-1")
+		require.NoError(t, msg.RespondMsg(out))
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = capSub.Unsubscribe() })
+	require.NoError(t, nc.Flush())
+
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, EC2AccountSettingsReapers(nc, []string{"node-1"})...)
+	// The real capability probe holds CollectUntilDeadline's full ResponderGrace
+	// (150 ms) even when the one expected node answers instantly, so the quiet
+	// engine's default 200 ms StageDrain leaves no room for the sweep after it.
+	engine.Timeouts = Timeouts{StageDrain: 2 * time.Second, DrainPoll: 10 * time.Millisecond}
+
+	result, err := engine.Teardown(ctx, request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+
+	bucket, err := js.KeyValue(ctx, ec2account.KVBucketAccountSettings)
+	require.NoError(t, err)
+	entry, err := bucket.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	var rec ec2account.AccountSettingsRecord
+	require.NoError(t, json.Unmarshal(entry.Value(), &rec))
+	assert.Equal(t, ec2account.AccountSettingsStateSealed, rec.State)
+}

spinifex/accountteardown/reapers_quota_override_current_behaviour_test.go

@@ -0,0 +1,61 @@
+package accountteardown
+
+//test:in-package — drives the real final-stage reapers against the real quota service to pin that teardown leaves the override bucket alone.
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/admission/quota"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// No reaper in this package references the quota override bucket or calls an
+// override delete. Running the real engine and the real StageAccount reapers
+// against an account with a raised vCPU override, with usage wired, shows the
+// consequence: the override outlives the account record it belongs to, while
+// the usage counter the reapers do own is removed.
+func TestCurrentBehaviour_AccountTermination_QuotaOverridesSurviveTeardown(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	ctx := testCtx(t)
+
+	usage, err := kvutil.GetOrCreateBucket(ctx, js, quota.KVBucketAccountUsage, 1)
+	require.NoError(t, err)
+	_, err = usage.Put(ctx, testAccountID, []byte(`{"vcpus":4}`))
+	require.NoError(t, err)
+
+	overridesKV, err := kvutil.GetOrCreateBucket(ctx, js, quota.KVBucketAccountQuota, 1)
+	require.NoError(t, err)
+	svc := quota.New(quota.Limits{Enabled: true, VCPUs: 16}, usage)
+	svc.SetOverrides(overridesKV)
+	unlimited := quota.Unlimited
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccountID, quota.Overrides{VCPUs: &unlimited}, "operator"))
+
+	accounts := newFakeAccounts()
+	iam := &accountIAM{account: &handlers_iam.Account{AccountID: testAccountID, AccountName: testAccountName}}
+	engine := quietEngine(accounts, AccountReapers(iam, nil, usage)...)
+
+	result, err := engine.Teardown(ctx, request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+	assert.True(t, accounts.deleted)
+	assert.Zero(t, result.StuckCount())
+
+	// The reaper this package does own removed the usage counter, which shows
+	// which store teardown actually clears.
+	_, err = usage.Get(ctx, testAccountID)
+	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound, "teardown's quota-usage reaper must remove the vCPU counter")
+
+	_, err = overridesKV.Get(ctx, testAccountID)
+	require.NoError(t, err, "the override key must still be present after the account is torn down")
+
+	over, effective, err := svc.GetAccountQuota(ctx, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, over.VCPUs)
+	assert.Equal(t, quota.Unlimited, *over.VCPUs)
+	assert.Equal(t, quota.Unlimited, effective.VCPUs)
+}

spinifex/accountteardown/settings_kv_boundary_test.go

@@ -0,0 +1,89 @@
+package accountteardown
+
+//test:in-package — pins ADR-0003:S4 and ADR-0001 Context item 2 for the EC2
+// settings seal: the legacy engine must reach the owner only through the
+// versioned contract and NATS, never by reading the settings bucket itself.
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"io/fs"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+// forbiddenSettingsImports are imports that would let this package reach the
+// EC2 settings owner or the configured node set by a path other than the
+// versioned contract and NewClusterEngine's parameters.
+var forbiddenSettingsImports = []string{
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/account",
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config",
+}
+
+// forbiddenSettingsIdentifiers would let this package name the settings
+// bucket or its key directly instead of reaching it only through
+// contracts/ec2/v1 and NATS.
+var forbiddenSettingsIdentifiers = []string{"spinifex-ec2-account-settings", "KVBucketAccountSettings"}
+
+// Teardown reaches EC2 settings only over contracts/ec2/v1: no import of the
+// settings owner or bootstrap/config (nodes arrive via NewClusterEngine), and
+// no literal naming the settings bucket or its constant.
+func TestAccountTeardownHasNoSettingsKVAccess(t *testing.T) {
+	_, file, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("cannot locate test source")
+	}
+	root := filepath.Dir(file)
+
+	fset := token.NewFileSet()
+	scanned := 0
+	err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+		if err != nil || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+			return err
+		}
+		f, perr := parser.ParseFile(fset, path, nil, parser.ParseComments)
+		if perr != nil {
+			return perr
+		}
+		scanned++
+
+		for _, imp := range f.Imports {
+			p, _ := strconv.Unquote(imp.Path.Value)
+			for _, bad := range forbiddenSettingsImports {
+				if p == bad {
+					t.Errorf("%s: imports %s; the legacy engine must reach the EC2 settings owner only through contracts/ec2/v1 and the configured node set NewClusterEngine already receives",
+						fset.Position(imp.Pos()), p)
+				}
+			}
+		}
+
+		ast.Inspect(f, func(n ast.Node) bool {
+			lit, ok := n.(*ast.BasicLit)
+			if !ok || lit.Kind != token.STRING {
+				return true
+			}
+			value, uerr := strconv.Unquote(lit.Value)
+			if uerr != nil {
+				return true
+			}
+			for _, bad := range forbiddenSettingsIdentifiers {
+				if value == bad {
+					t.Errorf("%s: names %q; the settings bucket and its key are the owner's own, never this package's",
+						fset.Position(lit.Pos()), value)
+				}
+			}
+			return true
+		})
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if scanned == 0 {
+		t.Fatalf("scanned no Go files under %s; the invariant would pass vacuously", root)
+	}
+}

spinifex/daemon/daemon.go

@@ -1026,6 +1026,7 @@ func (d *Daemon) subscribeAll() error {
 		{"ec2.GetSerialConsoleAccessStatus", handleNATSRequest(d.node, d.accountService.GetSerialConsoleAccessStatus), "spinifex-workers"},
 		{"ec2.EnableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.EnableSerialConsoleAccess), "spinifex-workers"},
 		{"ec2.DisableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.DisableSerialConsoleAccess), "spinifex-workers"},
+		{ec2v1.SealAccountSettingsSubject, handleNATSRequest(d.node, d.sealAccountSettings), "spinifex-workers"},
 		{clusterv1.NodeHealthSubject(d.node), d.handleHealthCheck, ""},
 		{clusterv1.NodesDiscoverSubject, d.handleNodeDiscover, ""},
 		{clusterv1.NodeStatusSubject, d.handleNodeStatus, ""},
@@ -1288,6 +1289,15 @@ func (d *Daemon) subscribeAll() error {
 		natsSub{viperblocklegacyv1.VolumeFencedSubject(d.node), d.handleVolumeFenced, ""},
 	)
 
+	// Per-node, no queue group: a second daemon claiming this node ID also
+	// answers, so the teardown probe sees a duplicate and refuses to seal. An
+	// invalid node ID serves nothing, so this daemon reads as incompatible.
+	if ec2v1.ValidSealCapabilityNode(d.node) {
+		subs = append(subs, natsSub{ec2v1.AccountSettingsSealCapabilitySubject(d.node), handleNATSRequest(d.node, d.accountSettingsSealCapability), ""})
+	} else {
+		slog.Error("Not serving account-settings seal capability: node ID is not a single NATS subject token", "node", d.node)
+	}
+
 	return d.registerNatsSubs(subs)
 }
 
@@ -1992,6 +2002,12 @@ func (d *Daemon) startCluster() error {
 	if err != nil {
 		return fmt.Errorf("failed to initialize account settings service: %w", err)
 	}
+	accountStatusJS, err := jetstream.New(d.natsConn)
+	if err != nil {
+		return fmt.Errorf("jetstream handle for account status reader: %w", err)
+	}
+	var _ ec2account.AccountTerminationChecker = (*handlers_iam.AccountStatusReader)(nil)
+	d.accountService.SetAccountTerminationChecker(handlers_iam.NewAccountStatusReader(accountStatusJS))
 
 	// ELBv2 resolves listener certificate private keys through its own ACM
 	// Store over the same JetStream bucket the ACM service writes, so it needs

spinifex/daemon/daemon_handlers_account_settings_seal.go

@@ -0,0 +1,23 @@
+package daemon
+
+import (
+	"context"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+)
+
+// sealAccountSettings serves the irreversible EC2 settings seal. The account
+// comes from the request body, not X-Account-ID: the sealed account is part
+// of the versioned contract, so the header is ignored.
+func (d *Daemon) sealAccountSettings(ctx context.Context, req *ec2v1.SealAccountSettingsRequest, _ string) (*ec2v1.SealAccountSettingsReply, error) {
+	seal, err := d.accountService.SealAccount(ctx, req.AccountID)
+	if err != nil {
+		return nil, err
+	}
+	return &ec2v1.SealAccountSettingsReply{
+		AccountID:      req.AccountID,
+		CorrelationRef: req.CorrelationRef,
+		Disposition:    ec2v1.AccountSettingsDispositionSealed,
+		Revision:       seal.Revision,
+	}, nil
+}

spinifex/daemon/daemon_handlers_account_settings_seal_capability.go

@@ -0,0 +1,23 @@
+package daemon
+
+import (
+	"context"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+)
+
+// accountSettingsSealCapability serves this daemon's read-only seal capability.
+// It reads no KV and takes no account; a request naming another node is a
+// caller mistake, since the subject is already addressed to this node.
+func (d *Daemon) accountSettingsSealCapability(_ context.Context, req *ec2v1.AccountSettingsSealCapabilityRequest, _ string) (*ec2v1.AccountSettingsSealCapabilityReply, error) {
+	if req.Node != d.node {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue,
+			"seal capability request for node %q answered by node %q", req.Node, d.node)
+	}
+	return &ec2v1.AccountSettingsSealCapabilityReply{
+		Node:              d.node,
+		CapabilityVersion: ec2v1.AccountSettingsSealCapabilityVersion,
+		CorrelationRef:    req.CorrelationRef,
+	}, nil
+}

spinifex/daemon/daemon_handlers_account_settings_seal_capability_test.go

@@ -0,0 +1,115 @@
+package daemon
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newCapabilityTestDaemon builds a daemon serving its real subscription table
+// with node as its configured node ID.
+func newCapabilityTestDaemon(t *testing.T, node string) (*nats.Conn, *nats.Conn) {
+	t.Helper()
+	ns, nc, _ := testutil.StartTestJetStream(t)
+
+	d := newSubscribeTestDaemon(t, nc, "", "")
+	d.node = node
+	require.NoError(t, d.subscribeAll())
+	require.NoError(t, nc.Flush())
+
+	caller, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(caller.Close)
+	return nc, caller
+}
+
+func capabilityRequest(t *testing.T, nc *nats.Conn, node string) *nats.Msg {
+	t.Helper()
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: node, CorrelationRef: "req-1"})
+	require.NoError(t, err)
+	reply, err := nc.Request(ec2v1.AccountSettingsSealCapabilitySubject(node), data, 5*time.Second)
+	require.NoError(t, err)
+	return reply
+}
+
+// The responder answers its own node with version 1, the configured node and
+// no KV access (no settings service is wired into the test daemon at all).
+func TestAccountSettingsSealCapabilityAnswersOwnNode(t *testing.T) {
+	_, caller := newCapabilityTestDaemon(t, "node-1")
+
+	raw := capabilityRequest(t, caller, "node-1")
+	assert.Equal(t, "node-1", raw.Header.Get("X-Node-ID"))
+
+	var reply ec2v1.AccountSettingsSealCapabilityReply
+	require.NoError(t, json.Unmarshal(raw.Data, &reply))
+	assert.Equal(t, "node-1", reply.Node)
+	assert.Equal(t, ec2v1.AccountSettingsSealCapabilityVersion, reply.CapabilityVersion)
+	assert.Equal(t, "req-1", reply.CorrelationRef)
+}
+
+// A request naming a node other than this daemon's own is refused, not
+// answered as if it were this node.
+func TestAccountSettingsSealCapabilityRefusesOtherNode(t *testing.T) {
+	_, caller := newCapabilityTestDaemon(t, "node-1")
+
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: "node-2"})
+	require.NoError(t, err)
+	raw, err := caller.Request(ec2v1.AccountSettingsSealCapabilitySubject("node-1"), data, 5*time.Second)
+	require.NoError(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidParameterValue, replyErrCode(t, raw.Data))
+}
+
+// The subject is per node with no queue group, so two daemons configured
+// with the same node ID both answer a request addressed to that node.
+func TestAccountSettingsSealCapabilityHasNoQueueGroup(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+
+	first := newSubscribeTestDaemon(t, nc, "", "")
+	first.node = "node-1"
+	require.NoError(t, first.subscribeAll())
+
+	second := newSubscribeTestDaemon(t, nc, "", "")
+	second.node = "node-1"
+	require.NoError(t, second.subscribeAll())
+	require.NoError(t, nc.Flush())
+
+	sub, err := nc.SubscribeSync(nats.NewInbox())
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: "node-1"})
+	require.NoError(t, err)
+	msg := nats.NewMsg(ec2v1.AccountSettingsSealCapabilitySubject("node-1"))
+	msg.Data = data
+	msg.Reply = sub.Subject
+	require.NoError(t, nc.PublishMsg(msg))
+	require.NoError(t, nc.Flush())
+
+	seen := 0
+	for {
+		_, err := sub.NextMsg(300 * time.Millisecond)
+		if err != nil {
+			break
+		}
+		seen++
+	}
+	assert.Equal(t, 2, seen)
+}
+
+// A wildcard node ID must not become a wildcard subscription: built unchecked,
+// "node-*" would answer probes addressed to node-1 on node-1's behalf.
+func TestAccountSettingsSealCapabilityNotServedForInvalidNodeID(t *testing.T) {
+	_, caller := newCapabilityTestDaemon(t, "node-*")
+
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: "node-1"})
+	require.NoError(t, err)
+	_, err = caller.Request(ec2v1.AccountSettingsSealCapabilitySubject("node-1"), data, time.Second)
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+}

spinifex/daemon/daemon_handlers_account_settings_seal_test.go

@@ -0,0 +1,101 @@
+package daemon
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	ec2account "github.com/mulgadc/spinifex/spinifex/domains/ec2/account"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// sealTestChecker answers AccountTerminating from a fixed set, standing in
+// for IAM in these daemon-level tests.
+type sealTestChecker struct {
+	terminating map[string]bool
+}
+
+func (c *sealTestChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) {
+	return c.terminating[accountID], nil
+}
+
+// newSealTestDaemon builds a daemon serving its real subscription table over
+// a real account settings service, wired with checker as its termination
+// precondition.
+func newSealTestDaemon(t *testing.T, checker ec2account.AccountTerminationChecker) (*Daemon, *nats.Conn) {
+	t.Helper()
+	ns, nc, _ := testutil.StartTestJetStream(t)
+
+	acctSvc, err := ec2account.NewAccountSettingsServiceImplWithNATS(t.Context(), nil, nc)
+	require.NoError(t, err)
+	acctSvc.SetAccountTerminationChecker(checker)
+
+	d := newSubscribeTestDaemon(t, nc, "", "")
+	d.accountService = acctSvc
+	require.NoError(t, d.subscribeAll())
+	require.NoError(t, nc.Flush())
+
+	caller, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(caller.Close)
+	return d, caller
+}
+
+const (
+	sealTestTerminating = "111122223333"
+	sealTestActive      = "444455556666"
+)
+
+func sealRequest(t *testing.T, nc *nats.Conn, body any, header string) []byte {
+	t.Helper()
+	data, err := json.Marshal(body)
+	require.NoError(t, err)
+	reply, err := natsRequestAs(nc, ec2v1.SealAccountSettingsSubject, data, header, 5*time.Second)
+	require.NoError(t, err)
+	return reply.Data
+}
+
+// The header is ignored: the body account is what gets sealed, even when
+// X-Account-ID names a different one.
+func TestSealAccountSettingsSubjectSealsTheBodyAccount(t *testing.T) {
+	_, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}})
+
+	raw := sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{AccountID: sealTestTerminating, CorrelationRef: "req-1"}, sealTestActive)
+	require.Empty(t, replyErrCode(t, raw), string(raw))
+
+	var reply ec2v1.SealAccountSettingsReply
+	require.NoError(t, json.Unmarshal(raw, &reply))
+	assert.Equal(t, sealTestTerminating, reply.AccountID)
+	assert.Equal(t, "req-1", reply.CorrelationRef)
+	assert.Equal(t, ec2v1.AccountSettingsDispositionSealed, reply.Disposition)
+	assert.NotZero(t, reply.Revision)
+}
+
+// Only an account IAM holds as TERMINATING may be sealed; ACTIVE is refused
+// before any write.
+func TestSealAccountSettingsSubjectRefusesAnActiveAccount(t *testing.T) {
+	_, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}})
+
+	raw := sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{AccountID: sealTestActive}, "")
+	assert.Equal(t, awserrors.ErrorIncorrectState, replyErrCode(t, raw))
+}
+
+// A request without a valid account is refused, not sealed.
+func TestSealAccountSettingsSubjectRefusesAMissingAccount(t *testing.T) {
+	_, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}})
+
+	assert.Equal(t, awserrors.ErrorMissingParameter, replyErrCode(t, sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{}, "")))
+}
+
+// A daemon that does not serve the subject leaves the request with no responder.
+func TestSealAccountSettingsSubjectHasNoResponderWithoutTheDaemon(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	_, err := natsRequestAs(nc, ec2v1.SealAccountSettingsSubject, []byte(`{"account_id":"111122223333"}`), "", time.Second)
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+}

spinifex/domains/admission/quota/cleanup.go

@@ -0,0 +1,93 @@
+package quota
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// OverrideCleanupDisposition is CleanupAccount's outcome for an account's
+// quota override key.
+type OverrideCleanupDisposition string
+
+const (
+	OverrideCleanupDeleted OverrideCleanupDisposition = "deleted"
+	OverrideCleanupAbsent  OverrideCleanupDisposition = "absent"
+	OverrideCleanupFailed  OverrideCleanupDisposition = "failed"
+)
+
+// evidenceRevisionGuardedDelete and evidenceObservedAbsent are the fixed
+// evidence bases CleanupAccount reports; see kvstore/bucket.go:112-120 for why
+// a plain read cannot support a stronger absence claim.
+const (
+	evidenceRevisionGuardedDelete = "revision-guarded-delete"
+	evidenceObservedAbsent        = "observed-absent-at-scan"
+)
+
+// cleanupMaxAttempts bounds retries against a revision that keeps moving
+// under concurrent writes, so cleanup fails closed instead of spinning.
+const cleanupMaxAttempts = 5
+
+// OverrideCleanup is CleanupAccount's result: what happened to an account's
+// quota override key, and the fixed evidence basis for that claim.
+type OverrideCleanup struct {
+	Disposition OverrideCleanupDisposition
+	Evidence    string
+	Revision    uint64
+}
+
+// CleanupAccount removes accountID's quota override key so a reused account ID
+// does not inherit a deleted account's raised or lowered limits. It is
+// unwired: nothing calls it yet.
+func (s *Service) CleanupAccount(ctx context.Context, accountID string) (*OverrideCleanup, error) {
+	if accountID == "" {
+		return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required for quota override cleanup")
+	}
+	if !awsidentifiers.IsAccountID(accountID) {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "accountID %q is not a valid AWS account ID", accountID)
+	}
+	if accountID == awsidentifiers.GlobalAccountID {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "quota override cleanup must not target the global account")
+	}
+	if s == nil || s.overrides == nil {
+		return &OverrideCleanup{Disposition: OverrideCleanupFailed}, errors.New("quota: override bucket not configured")
+	}
+
+	// accountID is never resolved through any fallback here: cleanup must act
+	// on exactly accountID and never fall back to the global key.
+	for range cleanupMaxAttempts {
+		entry, err := s.overrides.Get(ctx, accountID)
+		if err != nil {
+			if errors.Is(err, jetstream.ErrKeyNotFound) {
+				return &OverrideCleanup{Disposition: OverrideCleanupAbsent, Evidence: evidenceObservedAbsent}, nil
+			}
+			return &OverrideCleanup{Disposition: OverrideCleanupFailed}, fmt.Errorf("get quota override for cleanup: %w", err)
+		}
+		rev := entry.Revision()
+
+		err = s.overrides.Delete(ctx, accountID, jetstream.LastRevision(rev))
+		switch {
+		case err == nil:
+			return &OverrideCleanup{
+				Disposition: OverrideCleanupDeleted,
+				Evidence:    evidenceRevisionGuardedDelete,
+				Revision:    rev,
+			}, nil
+		case errors.Is(err, jetstream.ErrKeyNotFound):
+			return &OverrideCleanup{Disposition: OverrideCleanupAbsent, Evidence: evidenceObservedAbsent}, nil
+		case errors.Is(err, jetstream.ErrKeyRevisionMismatch):
+			// A concurrent write landed between the read and the guarded
+			// delete; retry against whatever revision now holds.
+			continue
+		default:
+			return &OverrideCleanup{Disposition: OverrideCleanupFailed}, fmt.Errorf("delete quota override for cleanup: %w", err)
+		}
+	}
+
+	return &OverrideCleanup{Disposition: OverrideCleanupFailed},
+		fmt.Errorf("quota override cleanup for account: exhausted %d attempts against concurrent writes", cleanupMaxAttempts)
+}

spinifex/domains/admission/quota/cleanup_test.go

@@ -0,0 +1,184 @@
+//test:in-package — CleanupAccount deletes through the unexported overrides field, and the retry/race behaviour is only observable from inside the package.
+package quota
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// overrideCleanupFakeKV wraps a real KeyValue so a test can inject a race
+// between CleanupAccount's read and its guarded delete, or a store failure,
+// deterministically rather than through goroutine timing.
+type overrideCleanupFakeKV struct {
+	jetstream.KeyValue
+
+	deleteCalls  int
+	beforeDelete func(attempt int) error
+	getErr       error
+}
+
+func (f *overrideCleanupFakeKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	if f.getErr != nil {
+		return nil, f.getErr
+	}
+	return f.KeyValue.Get(ctx, key)
+}
+
+func (f *overrideCleanupFakeKV) Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error {
+	f.deleteCalls++
+	if f.beforeDelete != nil {
+		if err := f.beforeDelete(f.deleteCalls); err != nil {
+			return err
+		}
+	}
+	return f.KeyValue.Delete(ctx, key, opts...)
+}
+
+func TestCleanupAccount_DeletesTheOverrideKey(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator"))
+
+	result, err := svc.CleanupAccount(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupDeleted, result.Disposition)
+	assert.Equal(t, evidenceRevisionGuardedDelete, result.Evidence)
+	assert.NotZero(t, result.Revision)
+
+	_, err = svc.overrides.Get(ctx, testAccount)
+	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound)
+}
+
+func TestCleanupAccount_RepeatCallIsIdempotent(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator"))
+
+	first, err := svc.CleanupAccount(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupDeleted, first.Disposition)
+
+	second, err := svc.CleanupAccount(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupAbsent, second.Disposition)
+	assert.Equal(t, evidenceObservedAbsent, second.Evidence)
+}
+
+func TestCleanupAccount_AbsentKeyIsIdempotent(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	result, err := svc.CleanupAccount(t.Context(), "333333333333")
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupAbsent, result.Disposition)
+	assert.Equal(t, evidenceObservedAbsent, result.Evidence)
+}
+
+func TestCleanupAccount_OtherAccountsUntouched(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	accountA, accountB := testAccount, "222222222222"
+	require.NoError(t, svc.PutAccountQuota(ctx, accountA, Overrides{VCPUs: ptr(32)}, "operator"))
+	require.NoError(t, svc.PutAccountQuota(ctx, accountB, Overrides{VCPUs: ptr(8)}, "operator"))
+
+	result, err := svc.CleanupAccount(ctx, accountA)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupDeleted, result.Disposition)
+
+	over, _, err := svc.GetAccountQuota(ctx, accountB)
+	require.NoError(t, err)
+	require.NotNil(t, over.VCPUs)
+	assert.Equal(t, 8, *over.VCPUs)
+}
+
+func TestCleanupAccount_RefusesGlobalAccountID(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	result, err := svc.CleanupAccount(t.Context(), awsidentifiers.GlobalAccountID)
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+func TestCleanupAccount_RefusesEmptyAccountID(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	result, err := svc.CleanupAccount(t.Context(), "")
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+func TestCleanupAccount_RefusesInvalidAccountID(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	result, err := svc.CleanupAccount(t.Context(), "not-an-account")
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+func TestCleanupAccount_WithoutBucketFails(t *testing.T) {
+	svc := New(baseLimits(), nil)
+	result, err := svc.CleanupAccount(t.Context(), testAccount)
+	assert.Error(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, OverrideCleanupFailed, result.Disposition)
+}
+
+// A concurrent Put landing between CleanupAccount's read and its
+// revision-guarded delete must not be lost silently: the delete is rejected
+// and retried against the revision the write left behind.
+func TestCleanupAccount_ConcurrentPutDuringCleanupRetriesAndDeletes(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator"))
+
+	fake := &overrideCleanupFakeKV{KeyValue: svc.overrides}
+	fake.beforeDelete = func(attempt int) error {
+		if attempt != 1 {
+			return nil
+		}
+		// Simulate a racing override write landing after CleanupAccount's
+		// read but before its guarded delete, through the real KV so the
+		// delete's expected revision is now stale.
+		return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")
+	}
+	svc.overrides = fake
+
+	result, err := svc.CleanupAccount(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupDeleted, result.Disposition)
+	assert.Equal(t, 2, fake.deleteCalls, "the first delete is rejected by the race and retried once")
+}
+
+// A revision that keeps moving under sustained concurrent writes must not
+// spin forever: cleanup gives up and reports failed, bounded by attempts.
+func TestCleanupAccount_PersistentConcurrentWritesFailsBounded(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator"))
+
+	fake := &overrideCleanupFakeKV{KeyValue: svc.overrides}
+	fake.beforeDelete = func(int) error {
+		return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")
+	}
+	svc.overrides = fake
+
+	result, err := svc.CleanupAccount(ctx, testAccount)
+	assert.Error(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, OverrideCleanupFailed, result.Disposition)
+	assert.Equal(t, cleanupMaxAttempts, fake.deleteCalls)
+}
+
+// A store error must report failed, never absent: absence is a claim about
+// what the store holds, which an error gives no basis to make.
+func TestCleanupAccount_StoreErrorIsFailedNotAbsent(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	fake := &overrideCleanupFakeKV{KeyValue: svc.overrides, getErr: errors.New("kv unavailable")}
+	svc.overrides = fake
+
+	result, err := svc.CleanupAccount(t.Context(), testAccount)
+	assert.Error(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, OverrideCleanupFailed, result.Disposition)
+}

spinifex/domains/ec2/account/seal.go

@@ -0,0 +1,135 @@
+package account
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// sealMaxAttempts bounds SealAccount's retry loop against a revision that
+// keeps moving under concurrent writes, so it fails closed instead of
+// spinning.
+const sealMaxAttempts = 5
+
+// sealedRecord is the fixed value a seal writes. Clearing both booleans
+// retains no tenant setting; an older reader that ignores state sees the
+// AWS defaults.
+var sealedRecord = &AccountSettingsRecord{State: AccountSettingsStateSealed}
+
+// AccountSettingsSeal is SealAccount's result: the revision at which the
+// stream leader accepted the sealed value.
+type AccountSettingsSeal struct {
+	Revision uint64
+}
+
+// AccountTerminationChecker reports whether IAM holds accountID as TERMINATING.
+// A missing record is (false, nil); only a failed read is an error.
+type AccountTerminationChecker interface {
+	AccountTerminating(ctx context.Context, accountID string) (bool, error)
+}
+
+// SetAccountTerminationChecker wires the precondition; until it is set, SealAccount refuses.
+func (s *AccountSettingsServiceImpl) SetAccountTerminationChecker(c AccountTerminationChecker) {
+	s.terminationChecker = c
+}
+
+// checkTerminating runs before any settings-KV call, so a refusal never writes.
+// Not TERMINATING, including a missing IAM record, is IncorrectState; a read
+// failure or an unwired checker is InternalError.
+func (s *AccountSettingsServiceImpl) checkTerminating(ctx context.Context, accountID string) error {
+	if s.terminationChecker == nil {
+		slog.ErrorContext(ctx, "Account settings seal has no termination checker wired", "accountID", accountID)
+		return errors.New(awserrors.ErrorInternalError)
+	}
+
+	terminating, err := s.terminationChecker.AccountTerminating(ctx, accountID)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to read account status for seal precondition", "accountID", accountID, "err", err)
+		return errors.New(awserrors.ErrorInternalError)
+	}
+	if !terminating {
+		return awserrors.Errorf(awserrors.ErrorIncorrectState, "account %s is not TERMINATING", accountID)
+	}
+	return nil
+}
+
+// isRevisionConflict reports a lost CAS race SealAccount should retry. Besides
+// the two sentinels, Create's retry against a delete marker returns a bare
+// *jetstream.APIError carrying a wrong-last-sequence code.
+func isRevisionConflict(err error) bool {
+	if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+		return true
+	}
+	if apiErr, ok := errors.AsType[*jetstream.APIError](err); ok {
+		return apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence ||
+			apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequenceConstant
+	}
+	return false
+}
+
+// SealAccount seals accountID's settings record so no later mutation can
+// recreate it. It never relies on a read for its claim: the only success is a
+// leader-accepted write, evidenced by the returned revision.
+func (s *AccountSettingsServiceImpl) SealAccount(ctx context.Context, accountID string) (*AccountSettingsSeal, error) {
+	if accountID == "" {
+		return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required to seal settings")
+	}
+	if !awsidentifiers.IsAccountID(accountID) {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "accountID %q is not a valid AWS account ID", accountID)
+	}
+	if accountID == awsidentifiers.GlobalAccountID {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "settings seal must not target the global account")
+	}
+
+	if err := s.checkTerminating(ctx, accountID); err != nil {
+		return nil, err
+	}
+
+	data, err := json.Marshal(sealedRecord)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to encode sealed account settings record", "accountID", accountID, "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+
+	// settingsKey is never used here: a seal must act on exactly accountID
+	// and never fall back to the global key.
+	for range sealMaxAttempts {
+		var rev uint64
+		entry, getErr := s.settingsKV.Get(ctx, accountID)
+		switch {
+		case getErr == nil:
+			rev = entry.Revision()
+		case errors.Is(getErr, jetstream.ErrKeyNotFound):
+			rev = 0
+		default:
+			slog.ErrorContext(ctx, "Failed to read account settings for seal", "accountID", accountID, "err", getErr)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+
+		var newRev uint64
+		var writeErr error
+		if rev == 0 {
+			newRev, writeErr = s.settingsKV.Create(ctx, accountID, data)
+		} else {
+			newRev, writeErr = s.settingsKV.Update(ctx, accountID, data, rev)
+		}
+
+		switch {
+		case writeErr == nil:
+			return &AccountSettingsSeal{Revision: newRev}, nil
+		case isRevisionConflict(writeErr):
+			continue
+		default:
+			slog.ErrorContext(ctx, "Failed to write sealed account settings record", "accountID", accountID, "err", writeErr)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+	}
+
+	slog.ErrorContext(ctx, "Account settings seal exhausted attempts against concurrent writes", "accountID", accountID, "attempts", sealMaxAttempts)
+	return nil, errors.New(awserrors.ErrorInternalError)
+}

spinifex/domains/ec2/account/seal_test.go

@@ -0,0 +1,559 @@
+package account
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stubTerminationChecker answers AccountTerminating from a fixed set, for
+// tests that exercise the KV paths rather than IAM itself.
+type stubTerminationChecker struct {
+	terminating map[string]bool
+	err         error
+}
+
+func (c *stubTerminationChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) {
+	if c.err != nil {
+		return false, c.err
+	}
+	return c.terminating[accountID], nil
+}
+
+// terminatingChecker answers true for exactly accountID.
+func terminatingChecker(accountID string) *stubTerminationChecker {
+	return &stubTerminationChecker{terminating: map[string]bool{accountID: true}}
+}
+
+// lastSettingsSeq reports the account-settings stream's last sequence, so a
+// test can assert a failed attempt wrote nothing.
+func lastSettingsSeq(t *testing.T, svc *AccountSettingsServiceImpl) uint64 {
+	t.Helper()
+	stream, err := svc.js.Stream(context.Background(), "KV_"+KVBucketAccountSettings)
+	require.NoError(t, err)
+	info, err := stream.Info(context.Background())
+	require.NoError(t, err)
+	return info.State.LastSeq
+}
+
+// mutationCase is one of the four settings mutations, called uniformly so a
+// test can table over them.
+type mutationCase struct {
+	name string
+	call func(svc *AccountSettingsServiceImpl, ctx context.Context) error
+}
+
+func mutationTable() []mutationCase {
+	return []mutationCase{
+		{"EnableEbs", func(svc *AccountSettingsServiceImpl, ctx context.Context) error {
+			_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+			return err
+		}},
+		{"DisableEbs", func(svc *AccountSettingsServiceImpl, ctx context.Context) error {
+			_, err := svc.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID)
+			return err
+		}},
+		{"EnableSerial", func(svc *AccountSettingsServiceImpl, ctx context.Context) error {
+			_, err := svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+			return err
+		}},
+		{"DisableSerial", func(svc *AccountSettingsServiceImpl, ctx context.Context) error {
+			_, err := svc.DisableSerialConsoleAccess(ctx, &ec2.DisableSerialConsoleAccessInput{}, testAccountID)
+			return err
+		}},
+	}
+}
+
+// seedSettingsState prepares testAccountID's record for a race test: present
+// (an active record), absent (never written) or tombstone (written then
+// deleted, leaving a delete marker Get hides).
+func seedSettingsState(t *testing.T, svc *AccountSettingsServiceImpl, state string) {
+	t.Helper()
+	ctx := context.Background()
+	switch state {
+	case "present":
+		_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+		require.NoError(t, err)
+	case "tombstone":
+		_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+		require.NoError(t, err)
+		require.NoError(t, svc.settingsKV.Delete(ctx, testAccountID))
+	case "absent":
+	default:
+		t.Fatalf("unknown seed state %q", state)
+	}
+}
+
+// sealRaceKV wraps a real KeyValue so a test can inject a deterministic race
+// between a caller's read and its guarded write (Create or Update), instead
+// of relying on goroutine timing.
+type sealRaceKV struct {
+	jetstream.KeyValue
+
+	writeCalls  int
+	beforeWrite func(attempt int) error
+}
+
+func (f *sealRaceKV) Create(ctx context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) {
+	f.writeCalls++
+	if f.beforeWrite != nil {
+		if err := f.beforeWrite(f.writeCalls); err != nil {
+			return 0, err
+		}
+	}
+	return f.KeyValue.Create(ctx, key, value, opts...)
+}
+
+func (f *sealRaceKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) {
+	f.writeCalls++
+	if f.beforeWrite != nil {
+		if err := f.beforeWrite(f.writeCalls); err != nil {
+			return 0, err
+		}
+	}
+	return f.KeyValue.Update(ctx, key, value, revision)
+}
+
+// staleGetKV always answers Get with a fixed snapshot taken at construction,
+// so a caller's CAS keeps retrying against a revision that can never win.
+type staleGetKV struct {
+	jetstream.KeyValue
+
+	entry jetstream.KeyValueEntry
+}
+
+func (f *staleGetKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) {
+	return f.entry, nil
+}
+
+// staleNotFoundKV always answers Get as absent, regardless of what the real
+// bucket holds.
+type staleNotFoundKV struct {
+	jetstream.KeyValue
+}
+
+func (f *staleNotFoundKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) {
+	return nil, jetstream.ErrKeyNotFound
+}
+
+// errGetKV always fails Get with a fixed store error.
+type errGetKV struct {
+	jetstream.KeyValue
+
+	err error
+}
+
+func (f *errGetKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) {
+	return nil, f.err
+}
+
+func decodeSettings(t *testing.T, entry jetstream.KeyValueEntry) AccountSettingsRecord {
+	t.Helper()
+	var rec AccountSettingsRecord
+	require.NoError(t, json.Unmarshal(entry.Value(), &rec))
+	return rec
+}
+
+// A seal landing between a mutation's read and its guarded write is not
+// outlived: the mutation's retry sees the seal and is rejected, and nothing
+// it wrote survives.
+func TestSealRaceMutation_SealLandsBetweenReadAndWrite(t *testing.T) {
+	for _, m := range mutationTable() {
+		for _, state := range []string{"present", "absent", "tombstone"} {
+			t.Run(m.name+"_"+state, func(t *testing.T) {
+				svc := setupTestAccountService(t)
+				ctx := context.Background()
+				seedSettingsState(t, svc, state)
+
+				realKV := svc.settingsKV
+				sealer := &AccountSettingsServiceImpl{js: svc.js, settingsKV: realKV}
+				sealer.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+				var sealed *AccountSettingsSeal
+				race := &sealRaceKV{KeyValue: realKV}
+				race.beforeWrite = func(attempt int) error {
+					if attempt != 1 {
+						return nil
+					}
+					var sealErr error
+					sealed, sealErr = sealer.SealAccount(ctx, testAccountID)
+					return sealErr
+				}
+				svc.settingsKV = race
+
+				err := m.call(svc, ctx)
+				require.Error(t, err)
+				assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err)
+				require.NotNil(t, sealed)
+
+				entry, err := realKV.Get(ctx, testAccountID)
+				require.NoError(t, err)
+				assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State)
+				assert.Equal(t, sealed.Revision, entry.Revision(), "no write landed after the seal")
+			})
+		}
+	}
+}
+
+// A mutation landing inside a seal's read-write window does not survive it:
+// the seal retries against the mutation's write and wins, leaving the
+// sealed value with both booleans false.
+func TestSealRaceMutation_MutationLandsBetweenSealReadAndWrite(t *testing.T) {
+	for _, m := range mutationTable() {
+		for _, state := range []string{"present", "absent", "tombstone"} {
+			t.Run(m.name+"_"+state, func(t *testing.T) {
+				svc := setupTestAccountService(t)
+				ctx := context.Background()
+				seedSettingsState(t, svc, state)
+
+				realKV := svc.settingsKV
+				race := &sealRaceKV{KeyValue: realKV}
+				race.beforeWrite = func(attempt int) error {
+					if attempt != 1 {
+						return nil
+					}
+					_ = m.call(svc, ctx)
+					return nil
+				}
+
+				sealer := &AccountSettingsServiceImpl{js: svc.js, settingsKV: race}
+				sealer.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+				seal, err := sealer.SealAccount(ctx, testAccountID)
+				require.NoError(t, err)
+
+				entry, err := realKV.Get(ctx, testAccountID)
+				require.NoError(t, err)
+				rec := decodeSettings(t, entry)
+				assert.Equal(t, AccountSettingsStateSealed, rec.State)
+				assert.False(t, rec.EbsEncryptionByDefault)
+				assert.False(t, rec.SerialConsoleAccess)
+				assert.Equal(t, seal.Revision, entry.Revision())
+			})
+		}
+	}
+}
+
+// None of the four mutations can act on an already-sealed record.
+func TestMutationOnSealedRecordIsRejected(t *testing.T) {
+	for _, m := range mutationTable() {
+		t.Run(m.name, func(t *testing.T) {
+			svc := setupTestAccountService(t)
+			ctx := context.Background()
+			svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+			_, err := svc.SealAccount(ctx, testAccountID)
+			require.NoError(t, err)
+
+			before, err := svc.settingsKV.Get(ctx, testAccountID)
+			require.NoError(t, err)
+
+			err = m.call(svc, ctx)
+			require.Error(t, err)
+			assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err)
+
+			after, err := svc.settingsKV.Get(ctx, testAccountID)
+			require.NoError(t, err)
+			assert.Equal(t, before.Revision(), after.Revision())
+		})
+	}
+}
+
+// Every settings mutation races one seal on a real bucket; afterwards the
+// record is sealed and stays sealed, so no success among them could have
+// landed after the seal.
+func TestSealConcurrentMutations(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	mutations := mutationTable()
+	errs := make([]error, len(mutations))
+	var wg sync.WaitGroup
+	for i, m := range mutations {
+		wg.Add(1)
+		go func(i int, m mutationCase) {
+			defer wg.Done()
+			errs[i] = m.call(svc, ctx)
+		}(i, m)
+	}
+
+	var sealResult *AccountSettingsSeal
+	var sealErr error
+	wg.Go(func() {
+		sealResult, sealErr = svc.SealAccount(ctx, testAccountID)
+	})
+	wg.Wait()
+
+	require.NoError(t, sealErr)
+	require.NotNil(t, sealResult)
+	for i, err := range errs {
+		if err != nil {
+			assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "mutation %s: got %v", mutations[i].name, err)
+		}
+	}
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State)
+	assert.Equal(t, sealResult.Revision, entry.Revision())
+
+	_, err = svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+}
+
+// A Get pinned to a revision the record has already moved past exhausts the
+// mutation's retries without ever writing.
+func TestMutationStaleReadExhaustsWithoutWriting(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+	stale, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+
+	_, err = svc.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	before := lastSettingsSeq(t, svc)
+	svc.settingsKV = &staleGetKV{KeyValue: svc.settingsKV, entry: stale}
+
+	_, err = svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+// A Get that always reports absent, while a sealed record actually exists,
+// never gets to recreate it: every Create attempt conflicts.
+func TestMutationStaleAbsentDoesNotRecreate(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	_, err := svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+
+	before := lastSettingsSeq(t, svc)
+	svc.settingsKV = &staleNotFoundKV{KeyValue: svc.settingsKV}
+
+	_, err = svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+// A revision that keeps moving under sustained concurrent writes must not
+// let the seal spin forever: it gives up, bounded by its attempt budget,
+// and its own write never lands.
+func TestSealSustainedConflictFailsWithoutWriting(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	realKV := svc.settingsKV
+	race := &sealRaceKV{KeyValue: realKV}
+	race.beforeWrite = func(int) error {
+		data, merr := json.Marshal(&AccountSettingsRecord{SerialConsoleAccess: true})
+		require.NoError(t, merr)
+		_, perr := realKV.Put(ctx, testAccountID, data)
+		return perr
+	}
+	svc.settingsKV = race
+
+	_, err = svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, sealMaxAttempts, race.writeCalls)
+
+	entry, err := realKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.NotEqual(t, AccountSettingsStateSealed, decodeSettings(t, entry).State, "the seal's own write never landed")
+}
+
+// A store error on the seal's read reports failed, never written.
+func TestSealStoreErrorWritesNothing(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+	before := lastSettingsSeq(t, svc)
+	svc.settingsKV = &errGetKV{KeyValue: svc.settingsKV, err: errors.New("kv unavailable")}
+
+	_, err := svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+func TestIsRevisionConflictRecognisesTombstoneRetry(t *testing.T) {
+	assert.True(t, isRevisionConflict(jetstream.ErrKeyExists))
+	assert.True(t, isRevisionConflict(jetstream.ErrKeyRevisionMismatch))
+	assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence}))
+	assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequenceConstant}))
+	assert.True(t, isRevisionConflict(fmt.Errorf("wrapped: %w", &jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence})))
+	assert.False(t, isRevisionConflict(errors.New("unrelated failure")))
+	assert.False(t, isRevisionConflict(nil))
+}
+
+// An account IAM does not hold as TERMINATING refuses the seal before any
+// settings-KV write, whether the key is present or absent.
+func TestSealRefusedForActiveAccount(t *testing.T) {
+	for _, state := range []string{"present", "absent"} {
+		t.Run(state, func(t *testing.T) {
+			svc := setupTestAccountService(t)
+			ctx := context.Background()
+			if state == "present" {
+				_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+				require.NoError(t, err)
+			}
+			svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+			before := lastSettingsSeq(t, svc)
+			_, err := svc.SealAccount(ctx, testAccountID)
+			assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+			assert.Equal(t, before, lastSettingsSeq(t, svc))
+		})
+	}
+}
+
+func TestSealRefusedForMissingIAMRecord(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+	before := lastSettingsSeq(t, svc)
+	_, err := svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+func TestSealRefusedWhenIAMReadFails(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")})
+
+	before := lastSettingsSeq(t, svc)
+	_, err := svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+func TestSealRefusedWithNoCheckerWired(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+
+	before := lastSettingsSeq(t, svc)
+	_, err := svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+func TestSealAcceptedForTerminatingAccount(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+	seal, err := svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.NotZero(t, seal.Revision)
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, seal.Revision, entry.Revision())
+}
+
+// A sealed record stays sealed after the service is rebuilt over the same
+// JetStream, and every one of the six settings actions still refuses it.
+func TestSealSurvivesServiceReconstruction(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	ctx := context.Background()
+
+	svc, err := NewAccountSettingsServiceImplWithNATS(ctx, nil, nc)
+	require.NoError(t, err)
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	first, err := svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+
+	rebuilt, err := NewAccountSettingsServiceImplWithNATS(ctx, nil, nc)
+	require.NoError(t, err)
+
+	entry, err := rebuilt.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State)
+
+	_, err = rebuilt.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.GetSerialConsoleAccessStatus(ctx, &ec2.GetSerialConsoleAccessStatusInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.DisableSerialConsoleAccess(ctx, &ec2.DisableSerialConsoleAccessInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+
+	rebuilt.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	second, err := rebuilt.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Greater(t, second.Revision, first.Revision)
+}
+
+func TestSealAbsentKeyCreatesSealedRecord(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+	_, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.ErrorIs(t, err, jetstream.ErrKeyNotFound)
+
+	seal, err := svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, seal.Revision, entry.Revision())
+	assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State)
+}
+
+func TestSealRefusesEmptyInvalidAndGlobalAccount(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+
+	_, err := svc.SealAccount(ctx, "")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorMissingParameter), "got %v", err)
+
+	_, err = svc.SealAccount(ctx, "not-an-account")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err)
+
+	_, err = svc.SealAccount(ctx, awsidentifiers.GlobalAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err)
+}
+
+func TestSealLeavesOtherAccountsUntouched(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	accountB := "222222222222"
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, accountB)
+	require.NoError(t, err)
+
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	_, err = svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+
+	out, err := svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, accountB)
+	require.NoError(t, err)
+	assert.True(t, *out.EbsEncryptionByDefault)
+}

spinifex/domains/ec2/account/service_impl.go

@@ -5,12 +5,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"log/slog"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
@@ -20,14 +21,19 @@ import (
 const (
 	KVBucketAccountSettings        = "spinifex-ec2-account-settings"
 	KVBucketAccountSettingsVersion = 1
-	KeyEbsEncryptionDefault        = "ebs-encryption-default"
-	KeySerialConsoleAccess         = "serial-console-access"
 )
 
+// AccountSettingsState is a settings record's lifecycle state. "" is active;
+// a record written before the state field existed has none and stays active.
+type AccountSettingsState string
+
+const AccountSettingsStateSealed AccountSettingsState = "sealed"
+
 // AccountSettingsRecord represents stored account settings.
 type AccountSettingsRecord struct {
-	EbsEncryptionByDefault bool `json:"ebs_encryption_by_default"`
-	SerialConsoleAccess    bool `json:"serial_console_access"`
+	EbsEncryptionByDefault bool                 `json:"ebs_encryption_by_default"`
+	SerialConsoleAccess    bool                 `json:"serial_console_access"`
+	State                  AccountSettingsState `json:"state,omitempty"`
 }
 
 // AccountSettingsServiceImpl implements account settings operations with NATS JetStream persistence.
@@ -35,6 +41,10 @@ type AccountSettingsServiceImpl struct {
 	config     *config.Config
 	js         jetstream.JetStream
 	settingsKV jetstream.KeyValue
+
+	// terminationChecker is the seal precondition; SealAccount refuses until
+	// it is wired.
+	terminationChecker AccountTerminationChecker
 }
 
 var _ AccountSettingsService = (*AccountSettingsServiceImpl)(nil)
@@ -72,50 +82,99 @@ func settingsKey(accountID string) string {
 	return accountID
 }
 
-// getSettings retrieves current account settings.
+// errSettingsSealed marks a mutation or read that found a sealed record, so
+// callers map it to Blocked without logging it as a fault.
+var errSettingsSealed = errors.New(awserrors.ErrorBlocked)
+
+// errSettingsContended marks a mutation whose CAS retries were exhausted
+// under concurrent writers, bounding the retry loop instead of spinning.
+var errSettingsContended = errors.New("account settings: CAS retries exhausted")
+
+// checkActive rejects a sealed or unknown-state record; "" is active and
+// passes through. A mutation must never act on a record that failed this.
+func checkActive(rec *AccountSettingsRecord) error {
+	switch rec.State {
+	case "":
+		return nil
+	case AccountSettingsStateSealed:
+		return errSettingsSealed
+	default:
+		return fmt.Errorf("account settings: unknown state %q", rec.State)
+	}
+}
+
+// mapSettingsError turns a mutateSettings or getSettings failure into the AWS
+// error a caller returns: sealed is an expected, unlogged Blocked; everything
+// else is InternalError, logged with the account ID.
+func mapSettingsError(ctx context.Context, accountID string, err error) error {
+	switch {
+	case err == nil:
+		return nil
+	case errors.Is(err, errSettingsSealed):
+		return errors.New(awserrors.ErrorBlocked)
+	case errors.Is(err, errSettingsContended):
+		slog.ErrorContext(ctx, "Account settings CAS retries exhausted under contention", "accountID", accountID)
+	case errors.Is(err, kvutil.ErrRead):
+		slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err)
+	case errors.Is(err, kvutil.ErrDecode):
+		slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err)
+	case errors.Is(err, kvutil.ErrEncode):
+		slog.ErrorContext(ctx, "Failed to marshal account settings record", "accountID", accountID, "err", err)
+	case errors.Is(err, kvutil.ErrWrite):
+		slog.ErrorContext(ctx, "Failed to write account settings to KV", "accountID", accountID, "err", err)
+	default:
+		slog.ErrorContext(ctx, "Account settings operation failed", "accountID", accountID, "err", err)
+	}
+	return errors.New(awserrors.ErrorInternalError)
+}
+
+// mutateSettings applies set to accountID's record at the revision it read.
+// changed is always true, so every success is a leader-accepted write over an
+// active value; a seal landing between the read and the write fails the CAS.
+func (s *AccountSettingsServiceImpl) mutateSettings(ctx context.Context, accountID string, set func(*AccountSettingsRecord)) error {
+	_, err := kvutil.Update(ctx, s.settingsKV, settingsKey(accountID), kvutil.CASConfig{
+		CreateIfAbsent: true,
+		Exhausted:      func(string, int) error { return errSettingsContended },
+	}, func(rec *AccountSettingsRecord) (bool, error) {
+		if err := checkActive(rec); err != nil {
+			return false, err
+		}
+		set(rec)
+		return true, nil
+	})
+	return mapSettingsError(ctx, accountID, err)
+}
+
+// getSettings retrieves current account settings, rejecting a sealed or
+// unknown-state record so a read never answers from behind a seal. Absent is
+// the AWS defaults; a read is a direct get and can be stale.
 func (s *AccountSettingsServiceImpl) getSettings(ctx context.Context, accountID string) (*AccountSettingsRecord, error) {
 	entry, err := s.settingsKV.Get(ctx, settingsKey(accountID))
 	if err != nil {
 		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return &AccountSettingsRecord{
-				EbsEncryptionByDefault: false,
-				SerialConsoleAccess:    false,
-			}, nil
+			return &AccountSettingsRecord{}, nil
 		}
-		return nil, fmt.Errorf("failed to get account settings: %w", err)
+		slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
 	}
 
 	var settings AccountSettingsRecord
 	if err := json.Unmarshal(entry.Value(), &settings); err != nil {
-		return nil, err
+		slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+	if err := checkActive(&settings); err != nil {
+		return nil, mapSettingsError(ctx, accountID, err)
 	}
 
 	return &settings, nil
 }
 
-// saveSettings saves current account settings.
-func (s *AccountSettingsServiceImpl) saveSettings(ctx context.Context, settings *AccountSettingsRecord, accountID string) error {
-	data, err := json.Marshal(settings)
-	if err != nil {
-		return err
-	}
-
-	_, err = s.settingsKV.Put(ctx, settingsKey(accountID), data)
-	return err
-}
-
 // EnableEbsEncryptionByDefault enables EBS encryption by default for the account.
 func (s *AccountSettingsServiceImpl) EnableEbsEncryptionByDefault(ctx context.Context, input *ec2.EnableEbsEncryptionByDefaultInput, accountID string) (*ec2.EnableEbsEncryptionByDefaultOutput, error) {
 	slog.InfoContext(ctx, "EnableEbsEncryptionByDefault called", "accountID", accountID)
 
-	settings, err := s.getSettings(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	settings.EbsEncryptionByDefault = true
-
-	if err := s.saveSettings(ctx, settings, accountID); err != nil {
+	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.EbsEncryptionByDefault = true }); err != nil {
 		return nil, err
 	}
 
@@ -128,14 +187,7 @@ func (s *AccountSettingsServiceImpl) EnableEbsEncryptionByDefault(ctx context.Co
 func (s *AccountSettingsServiceImpl) DisableEbsEncryptionByDefault(ctx context.Context, input *ec2.DisableEbsEncryptionByDefaultInput, accountID string) (*ec2.DisableEbsEncryptionByDefaultOutput, error) {
 	slog.InfoContext(ctx, "DisableEbsEncryptionByDefault called", "accountID", accountID)
 
-	settings, err := s.getSettings(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	settings.EbsEncryptionByDefault = false
-
-	if err := s.saveSettings(ctx, settings, accountID); err != nil {
+	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.EbsEncryptionByDefault = false }); err != nil {
 		return nil, err
 	}
 
@@ -176,14 +228,7 @@ func (s *AccountSettingsServiceImpl) GetSerialConsoleAccessStatus(ctx context.Co
 func (s *AccountSettingsServiceImpl) EnableSerialConsoleAccess(ctx context.Context, input *ec2.EnableSerialConsoleAccessInput, accountID string) (*ec2.EnableSerialConsoleAccessOutput, error) {
 	slog.InfoContext(ctx, "EnableSerialConsoleAccess called", "accountID", accountID)
 
-	settings, err := s.getSettings(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	settings.SerialConsoleAccess = true
-
-	if err := s.saveSettings(ctx, settings, accountID); err != nil {
+	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.SerialConsoleAccess = true }); err != nil {
 		return nil, err
 	}
 
@@ -196,14 +241,7 @@ func (s *AccountSettingsServiceImpl) EnableSerialConsoleAccess(ctx context.Conte
 func (s *AccountSettingsServiceImpl) DisableSerialConsoleAccess(ctx context.Context, input *ec2.DisableSerialConsoleAccessInput, accountID string) (*ec2.DisableSerialConsoleAccessOutput, error) {
 	slog.InfoContext(ctx, "DisableSerialConsoleAccess called", "accountID", accountID)
 
-	settings, err := s.getSettings(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	settings.SerialConsoleAccess = false
-
-	if err := s.saveSettings(ctx, settings, accountID); err != nil {
+	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.SerialConsoleAccess = false }); err != nil {
 		return nil, err
 	}
 

spinifex/domains/ec2/account/service_impl_test.go

@@ -2,11 +2,13 @@ package account
 
 import (
 	"context"
+	"encoding/json"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/internal/testkit"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -143,3 +145,44 @@ func TestSettingsKey_EmptyAccountIDFallback(t *testing.T) {
 	assert.Equal(t, awsidentifiers.GlobalAccountID, settingsKey(""))
 	assert.Equal(t, "123456789012", settingsKey("123456789012"))
 }
+
+// A raw pre-seal record, with no state field at all, is readable and
+// mutable, and a mutation writes it back without adding one.
+func TestRecordWithoutStateIsActive(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	_, err := svc.settingsKV.Put(ctx, testAccountID, []byte(`{"ebs_encryption_by_default":true,"serial_console_access":false}`))
+	require.NoError(t, err)
+
+	out, err := svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+	assert.True(t, *out.EbsEncryptionByDefault)
+
+	_, err = svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+	require.NoError(t, err)
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.NotContains(t, string(entry.Value()), "state")
+}
+
+// A record with an unrecognised state fails closed rather than being treated
+// as active or sealed, for both a read and a mutation.
+func TestUnknownStateFailsClosed(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	_, err := svc.settingsKV.Put(ctx, testAccountID, []byte(`{"ebs_encryption_by_default":false,"serial_console_access":false,"state":"quarantined"}`))
+	require.NoError(t, err)
+
+	_, err = svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+
+	_, err = svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	var rec AccountSettingsRecord
+	require.NoError(t, json.Unmarshal(entry.Value(), &rec))
+	assert.EqualValues(t, "quarantined", rec.State, "the unknown-state record was never overwritten")
+}

spinifex/domains/ec2/account/settings_writes_guarded_test.go

@@ -0,0 +1,68 @@
+package account
+
+//test:in-package — pins what AccountSettingsSealCapabilityVersion 1 asserts:
+// nothing in this package writes the settings KV except through the
+// revision-guarded helpers.
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"io/fs"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"testing"
+)
+
+func accountPackageDir(t *testing.T) string {
+	t.Helper()
+	_, file, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("cannot locate test source")
+	}
+	return filepath.Dir(file)
+}
+
+// The capability daemons report as version 1 is a claim about every writer in
+// the binary, not just the responder: no non-test file in this package may
+// call Put or PutString, which bypass the revision-guarded mutate/seal paths.
+func TestAccountSettingsWritesAreRevisionGuarded(t *testing.T) {
+	root := accountPackageDir(t)
+	fset := token.NewFileSet()
+	scanned := 0
+
+	err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+		if err != nil || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+			return err
+		}
+		f, perr := parser.ParseFile(fset, path, nil, 0)
+		if perr != nil {
+			return perr
+		}
+		scanned++
+
+		ast.Inspect(f, func(n ast.Node) bool {
+			call, ok := n.(*ast.CallExpr)
+			if !ok {
+				return true
+			}
+			sel, ok := call.Fun.(*ast.SelectorExpr)
+			if !ok {
+				return true
+			}
+			if sel.Sel.Name == "Put" || sel.Sel.Name == "PutString" {
+				t.Errorf("%s: unconditional %s call bypasses the revision-guarded settings write; use mutateSettings or SealAccount's CAS loop instead",
+					fset.Position(call.Pos()), sel.Sel.Name)
+			}
+			return true
+		})
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if scanned == 0 {
+		t.Fatalf("scanned no Go files under %s; the invariant would pass vacuously", root)
+	}
+}

spinifex/gateway/admin_deleteaccount_job_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/s3"
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/accountteardown"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -31,6 +32,10 @@ const deleteClientToken = "8b9c1d2e3f4a5b6c7d8e9f0a1b2c3d4e"
 // to keep step with the reapers NewClusterEngine wires.
 var teardownSubjects = []string{"ec2.>", "ecs.>", "eks.>", "rds.>", "elbv2.>", "acm.>", "bedrock.>"}
 
+// sealCapabilitySubjectPrefix matches every per-node seal-capability subject,
+// which the generic ec2.> stub below must not also answer.
+const sealCapabilitySubjectPrefix = "ec2.v1.AccountSettingsSealCapability."
+
 // newDeleteAccountGateway wires a gateway against a real IAM service and
 // embedded JetStream, with a stub cluster answering the subjects the reapers
 // drive. The tenant has no resources, so every stage drains empty and the test
@@ -46,8 +51,12 @@ func newDeleteAccountGateway(t *testing.T) *GatewayConfig {
 
 	for _, subject := range teardownSubjects {
 		// Replies carry a node ID: the instance inventory only counts a frame
-		// it can attribute to a configured node.
+		// it can attribute to a configured node. The seal subjects have their
+		// own responders below, which a second reply here would race.
 		sub, err := nc.Subscribe(subject, func(msg *nats.Msg) {
+			if msg.Subject == ec2v1.SealAccountSettingsSubject || strings.HasPrefix(msg.Subject, sealCapabilitySubjectPrefix) {
+				return
+			}
 			reply := nats.NewMsg(msg.Reply)
 			reply.Data = []byte(`{}`)
 			reply.Header.Set(natsmsg.NodeIDHeader, "node-1")
@@ -56,6 +65,38 @@ func newDeleteAccountGateway(t *testing.T) *GatewayConfig {
 		require.NoError(t, err)
 		t.Cleanup(func() { _ = sub.Unsubscribe() })
 	}
+
+	// The seal reaper needs a real sealed disposition and revision, and every
+	// configured node answering capability version 1, or it never finalises:
+	// the generic `{}` stub above is neither.
+	sealSub, err := nc.Subscribe(ec2v1.SealAccountSettingsSubject, func(msg *nats.Msg) {
+		var req ec2v1.SealAccountSettingsRequest
+		if json.Unmarshal(msg.Data, &req) != nil {
+			return
+		}
+		data, _ := json.Marshal(ec2v1.SealAccountSettingsReply{
+			AccountID: req.AccountID, CorrelationRef: req.CorrelationRef,
+			Disposition: ec2v1.AccountSettingsDispositionSealed, Revision: 1,
+		})
+		reply := nats.NewMsg(msg.Reply)
+		reply.Data = data
+		reply.Header.Set(natsmsg.NodeIDHeader, "node-1")
+		_ = msg.RespondMsg(reply)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sealSub.Unsubscribe() })
+
+	for _, node := range []string{"node-1"} {
+		capSub, err := nc.Subscribe(ec2v1.AccountSettingsSealCapabilitySubject(node), func(msg *nats.Msg) {
+			data, _ := json.Marshal(ec2v1.AccountSettingsSealCapabilityReply{Node: node, CapabilityVersion: ec2v1.AccountSettingsSealCapabilityVersion})
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = data
+			reply.Header.Set(natsmsg.NodeIDHeader, node)
+			_ = msg.RespondMsg(reply)
+		})
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = capSub.Unsubscribe() })
+	}
 	require.NoError(t, nc.Flush())
 
 	return &GatewayConfig{

spinifex/handlers/iam/account_status_reader.go

@@ -0,0 +1,56 @@
+package handlers_iam
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// AccountStatusReader reads account status from spinifex-accounts and never
+// writes. It is the read-only capability EC2's settings seal uses to check
+// its TERMINATING precondition.
+type AccountStatusReader struct {
+	accounts *kvstore.Bucket // AttachOnly: never creates the bucket
+}
+
+// NewAccountStatusReader attaches to the existing spinifex-accounts bucket.
+// It needs no master key and runs no migration: a caller that only reads
+// account status must not bootstrap IAM's bucket out from under it.
+func NewAccountStatusReader(js jetstream.JetStream) *AccountStatusReader {
+	return &AccountStatusReader{
+		accounts: kvstore.NewBucket(js, kvstore.Config{
+			Name:       KVBucketAccounts,
+			Missing:    "account status reader: nil jetstream client",
+			AttachOnly: true,
+		}),
+	}
+}
+
+// AccountTerminating is a direct read, which suffices because TERMINATING is
+// one-way until the record is deleted. A missing record is (false, nil); only
+// a failed read or decode is an error.
+func (r *AccountStatusReader) AccountTerminating(ctx context.Context, accountID string) (bool, error) {
+	kv, err := r.accounts.KV(ctx)
+	if err != nil {
+		return false, err
+	}
+
+	entry, err := kv.Get(ctx, accountID)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	var record struct {
+		Status string `json:"status"`
+	}
+	if err := json.Unmarshal(entry.Value(), &record); err != nil {
+		return false, err
+	}
+	return record.Status == AccountStatusTerminating, nil
+}

spinifex/handlers/iam/account_status_reader_test.go

@@ -0,0 +1,102 @@
+package handlers_iam_test
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newStatusReaderFixture starts a real IAM service (which creates and
+// migrates spinifex-accounts) and a reader attached to the same JetStream.
+func newStatusReaderFixture(t *testing.T) (handlers_iam.IAMService, *handlers_iam.AccountStatusReader, jetstream.JetStream) {
+	t.Helper()
+	_, nc, js := testutil.StartTestJetStream(t)
+
+	masterKey, err := handlers_iam.GenerateMasterKey()
+	require.NoError(t, err)
+	svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey)
+	require.NoError(t, err)
+
+	return svc, handlers_iam.NewAccountStatusReader(js), js
+}
+
+func TestAccountStatusReader_TerminatingIsTrue(t *testing.T) {
+	t.Parallel()
+	svc, reader, _ := newStatusReaderFixture(t)
+	account, err := svc.CreateAccount("tenant@example.com")
+	require.NoError(t, err)
+	_, err = svc.SetAccountStatus(account.AccountID, handlers_iam.AccountStatusTerminating)
+	require.NoError(t, err)
+
+	terminating, err := reader.AccountTerminating(t.Context(), account.AccountID)
+	require.NoError(t, err)
+	assert.True(t, terminating)
+}
+
+func TestAccountStatusReader_ActiveAndSuspendedAreFalse(t *testing.T) {
+	t.Parallel()
+	svc, reader, _ := newStatusReaderFixture(t)
+	active, err := svc.CreateAccount("active@example.com")
+	require.NoError(t, err)
+	terminating, err := reader.AccountTerminating(t.Context(), active.AccountID)
+	require.NoError(t, err)
+	assert.False(t, terminating)
+
+	suspended, err := svc.CreateAccount("suspended@example.com")
+	require.NoError(t, err)
+	_, err = svc.SetAccountStatus(suspended.AccountID, handlers_iam.AccountStatusSuspended)
+	require.NoError(t, err)
+	terminating, err = reader.AccountTerminating(t.Context(), suspended.AccountID)
+	require.NoError(t, err)
+	assert.False(t, terminating)
+}
+
+func TestAccountStatusReader_MissingRecordIsFalseNotError(t *testing.T) {
+	t.Parallel()
+	_, reader, _ := newStatusReaderFixture(t)
+	terminating, err := reader.AccountTerminating(t.Context(), "999999999999")
+	require.NoError(t, err)
+	assert.False(t, terminating)
+}
+
+// A missing bucket is an error, and the reader never creates one: a caller
+// that only reads account status must not bootstrap IAM's bucket.
+func TestAccountStatusReader_MissingBucketIsErrorAndNeverCreated(t *testing.T) {
+	t.Parallel()
+	_, nc, js := testutil.StartTestJetStream(t)
+	reader := handlers_iam.NewAccountStatusReader(js)
+
+	_, err := reader.AccountTerminating(t.Context(), "111111111111")
+	require.Error(t, err)
+
+	_, bucketErr := js.KeyValue(t.Context(), handlers_iam.KVBucketAccounts)
+	assert.ErrorIs(t, bucketErr, jetstream.ErrBucketNotFound)
+	_ = nc
+}
+
+func TestAccountStatusReader_UndecodableValueIsError(t *testing.T) {
+	t.Parallel()
+	_, nc, js := testutil.StartTestJetStream(t)
+	_, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, mustMasterKey(t))
+	require.NoError(t, err)
+
+	kv, err := js.KeyValue(t.Context(), handlers_iam.KVBucketAccounts)
+	require.NoError(t, err)
+	_, err = kv.Put(t.Context(), "222222222222", []byte(`{"status":`))
+	require.NoError(t, err)
+
+	reader := handlers_iam.NewAccountStatusReader(js)
+	_, err = reader.AccountTerminating(t.Context(), "222222222222")
+	require.Error(t, err)
+}
+
+func mustMasterKey(t *testing.T) []byte {
+	t.Helper()
+	key, err := handlers_iam.GenerateMasterKey()
+	require.NoError(t, err)
+	return key
+}