feat(quota): seal per-account quota overrides behind a revision-guarded owner action (#1245)

Store the override record with a state field kept separate from the admin
wire type. Every set and clear is revision-guarded and refuses a sealed
record with Blocked, and every read of a sealed record, including
admission, returns Blocked. The read-failure fallback to configured limits
is unchanged.

Add Service.SealAccountOverrides, accepted only when IAM already holds the
account as TERMINATING. It creates the override bucket if needed after that
check, creates or updates the record as sealed, and succeeds only on a
leader-accepted write, returning its revision. awsgw serves it on
admission.v1.SealAccountQuotaOverride. Nothing calls it yet.

Replace the unwired delete-based CleanupAccount.

contracts/admission/v1/README.md

@@ -0,0 +1,42 @@
+# Admission contract, version 1
+
+This package owns the cross-process contract for the admission domain's
+quota override. It is not an in-process service interface or the owner of
+quota state.
+
+## Seal account quota override
+
+`SealAccountQuotaOverrideSubject` (`admission.v1.SealAccountQuotaOverride`) is
+a request/reply route on which a gateway, in the `awsgw-quota-seal` queue
+group, irreversibly seals one account's quota override record.
+`domains/admission/quota` owns it: `awsgw` serves it.
+`accountteardown` is the consumer: it calls this subject to fence every
+override writer at the quota owner before the account record is removed.
+
+The request names the account in its body (`account_id`) and may carry an
+opaque `correlation_ref` that is echoed for logs.
+`awsgw` 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, or no termination checker
+wired, answers `InternalError`.
+Both refusals write nothing and never open the override bucket.
+
+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.
+Any current gateway may answer; the leader-accepted revision is the evidence
+whichever one does, and the override bucket is opened at seal time even on a
+gateway whose quotas were disabled at startup.
+
+### Compatibility boundary
+
+The subject name carries the contract major; `account_quota_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 gateway 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.

contracts/admission/v1/account_quota_seal.go

@@ -0,0 +1,31 @@
+// Package admissionv1 holds the admission domain's cross-process contract:
+// the subjects, request/reply shapes and wire vocabulary the quota owner and
+// its consumers share.
+package admissionv1
+
+// SealAccountQuotaOverrideSubject is the request/reply subject on which a
+// gateway seals one account's quota override record. Sealing is irreversible
+// and refused unless IAM already holds the account as TERMINATING.
+const SealAccountQuotaOverrideSubject = "admission.v1.SealAccountQuotaOverride"
+
+// AccountQuotaOverrideDispositionSealed is the only disposition a successful
+// reply carries: a reply is sent only on success, so there is no other value.
+const AccountQuotaOverrideDispositionSealed = "sealed"
+
+// SealAccountQuotaOverrideRequest asks for one account's quota override 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 SealAccountQuotaOverrideRequest struct {
+	AccountID      string `json:"account_id"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+}
+
+// SealAccountQuotaOverrideReply confirms accountID's quota override 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 SealAccountQuotaOverrideReply struct {
+	AccountID      string `json:"account_id"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+	Disposition    string `json:"disposition"`
+	Revision       uint64 `json:"revision"`
+}

contracts/admission/v1/account_quota_seal_test.go

@@ -0,0 +1,43 @@
+package admissionv1
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSealAccountQuotaOverrideSubject(t *testing.T) {
+	require.Equal(t, "admission.v1.SealAccountQuotaOverride", SealAccountQuotaOverrideSubject)
+}
+
+func TestSealAccountQuotaOverrideRequestJSONContract(t *testing.T) {
+	data, err := json.Marshal(SealAccountQuotaOverrideRequest{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(SealAccountQuotaOverrideRequest{AccountID: "111122223333"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333"}`, string(data))
+}
+
+func TestSealAccountQuotaOverrideReplyJSONContract(t *testing.T) {
+	reply := SealAccountQuotaOverrideReply{
+		AccountID:      "111122223333",
+		CorrelationRef: "req-1",
+		Disposition:    AccountQuotaOverrideDispositionSealed,
+		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(SealAccountQuotaOverrideReply{AccountID: "111122223333", Disposition: AccountQuotaOverrideDispositionSealed})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333","disposition":"sealed","revision":0}`, string(data))
+}

spinifex/domains/admission/quota/cleanup.go

@@ -1,93 +0,0 @@
-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

@@ -1,184 +0,0 @@
-//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/admission/quota/live.go

@@ -39,7 +39,11 @@ func (s *Service) EnforceVPCs(ctx context.Context, natsConn *nats.Conn, accountI
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.Vpcs), want, s.limitsFor(ctx, accountID).VPCs)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.Vpcs), want, limits.VPCs)
 }
 
 // EnforceSubnets gates CreateSubnet on the account's live DescribeSubnets count.
@@ -52,7 +56,11 @@ func (s *Service) EnforceSubnets(ctx context.Context, natsConn *nats.Conn, accou
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.Subnets), want, s.limitsFor(ctx, accountID).Subnets)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.Subnets), want, limits.Subnets)
 }
 
 // EnforceEIPs gates AllocateAddress on the account's live DescribeAddresses count.
@@ -64,7 +72,11 @@ func (s *Service) EnforceEIPs(ctx context.Context, natsConn *nats.Conn, accountI
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.Addresses), want, s.limitsFor(ctx, accountID).EIPs)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.Addresses), want, limits.EIPs)
 }
 
 // EnforceLoadBalancers gates CreateLoadBalancer on the account's live count.
@@ -77,7 +89,11 @@ func (s *Service) EnforceLoadBalancers(ctx context.Context, natsConn *nats.Conn,
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.LoadBalancers), want, s.limitsFor(ctx, accountID).LoadBalancers)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.LoadBalancers), want, limits.LoadBalancers)
 }
 
 // EnforceRDSInstances gates CreateDBInstance on the account's live count. A DB
@@ -91,5 +107,9 @@ func (s *Service) EnforceRDSInstances(ctx context.Context, natsConn *nats.Conn,
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.DBInstances), want, s.limitsFor(ctx, accountID).RDSInstances)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.DBInstances), want, limits.RDSInstances)
 }

spinifex/domains/admission/quota/override.go

@@ -8,12 +8,16 @@ import (
 	"log/slog"
 	"time"
 
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
 // Overrides raises or lowers individual dimensions for one account. Every field
 // is a pointer so an explicit 0 (a limit that denies everything) is distinct
-// from an absent field, which inherits the configured value.
+// from an absent field, which inherits the configured value. This is also the
+// admin wire type: an admin request body can carry only these fields, so it can
+// never set or clear a stored record's state.
 type Overrides struct {
 	VCPUs         *int `json:"vcpus,omitempty"`
 	VPCs          *int `json:"vpcs,omitempty"`
@@ -29,11 +33,62 @@ type Overrides struct {
 	UpdatedBy string `json:"updated_by,omitempty"`
 }
 
-// Empty reports whether no dimension is overridden, in which case the record
-// carries nothing but provenance and the account resolves to the config.
-func (o Overrides) Empty() bool {
-	return o.VCPUs == nil && o.VPCs == nil && o.Subnets == nil && o.EIPs == nil &&
-		o.VolumesGiB == nil && o.Volumes == nil && o.RDSInstances == nil && o.LoadBalancers == nil
+// OverrideState is a stored override record's lifecycle state. "" is active;
+// a record written before the state field existed has none and stays active.
+type OverrideState string
+
+// OverrideStateSealed marks a record no writer may change or recreate.
+const OverrideStateSealed OverrideState = "sealed"
+
+// overrideRecord is the stored form: the admin-settable fields inline, plus a
+// state no admin request can read or write.
+type overrideRecord struct {
+	Overrides
+
+	State OverrideState `json:"state,omitempty"`
+}
+
+// errOverridesSealed marks a read or write that found a sealed record, so a
+// caller maps it to Blocked without logging it as a fault.
+var errOverridesSealed = errors.New(awserrors.ErrorBlocked)
+
+// errOverridesContended marks a write whose CAS retries were exhausted under
+// concurrent writers, bounding the retry loop instead of spinning.
+var errOverridesContended = errors.New("quota override: CAS retries exhausted")
+
+// overrideMaxAttempts bounds a write's retries against a revision that keeps
+// moving under concurrent writers, so it fails closed instead of spinning.
+const overrideMaxAttempts = 5
+
+// checkActive rejects a sealed or unknown-state record; "" is active and
+// passes through. A caller must never read or mutate a record that failed
+// this as if it were active.
+func checkActive(rec overrideRecord) error {
+	switch rec.State {
+	case "":
+		return nil
+	case OverrideStateSealed:
+		return errOverridesSealed
+	default:
+		return fmt.Errorf("quota override: unknown state %q", rec.State)
+	}
+}
+
+// mapOverrideError turns a recordFor, PutAccountQuota or clear failure into
+// the AWS error a caller returns: sealed is an expected, unlogged Blocked;
+// everything else is InternalError, logged with the account ID.
+func mapOverrideError(ctx context.Context, accountID string, err error) error {
+	switch {
+	case err == nil:
+		return nil
+	case errors.Is(err, errOverridesSealed):
+		return errors.New(awserrors.ErrorBlocked)
+	case errors.Is(err, errOverridesContended):
+		slog.ErrorContext(ctx, "Quota override CAS retries exhausted under contention", "accountID", accountID)
+	default:
+		slog.ErrorContext(ctx, "Quota override operation failed", "accountID", accountID, "err", err)
+	}
+	return errors.New(awserrors.ErrorInternalError)
 }
 
 // apply overlays the set fields onto base, leaving Enabled and the Bedrock
@@ -60,72 +115,128 @@ func (o Overrides) apply(base Limits) Limits {
 	return out
 }
 
-// limitsFor resolves the effective limits for accountID: its overrides where
-// set, the configured limits everywhere else. A read failure falls back to the
-// configured limits rather than to no limit, so a KV outage cannot lift a cap.
-func (s *Service) limitsFor(ctx context.Context, accountID string) Limits {
-	over, err := s.overridesFor(ctx, accountID)
-	if err != nil {
-		slog.WarnContext(ctx, "quota: override read failed, using configured limits",
-			"account", accountID, "err", err)
-		return s.limits
-	}
-	return over.apply(s.limits)
+// Empty reports whether no dimension is overridden, in which case the record
+// carries nothing but provenance and the account resolves to the config.
+func (o Overrides) Empty() bool {
+	return o.VCPUs == nil && o.VPCs == nil && o.Subnets == nil && o.EIPs == nil &&
+		o.VolumesGiB == nil && o.Volumes == nil && o.RDSInstances == nil && o.LoadBalancers == nil
 }
 
-// overridesFor reads accountID's override record. An absent key is the normal
-// case for nearly every account and yields the zero value, not an error; only a
-// genuine read or decode failure is returned.
-func (s *Service) overridesFor(ctx context.Context, accountID string) (Overrides, error) {
+// recordFor reads and decodes accountID's stored override record. An absent
+// key is the active zero value, not an error; only a genuine read or decode
+// failure is returned.
+func (s *Service) recordFor(ctx context.Context, accountID string) (overrideRecord, error) {
 	if s == nil || s.overrides == nil {
-		return Overrides{}, nil
+		return overrideRecord{}, nil
 	}
 	entry, err := s.overrides.Get(ctx, accountID)
 	if errors.Is(err, jetstream.ErrKeyNotFound) {
-		return Overrides{}, nil
+		return overrideRecord{}, nil
 	}
 	if err != nil {
-		return Overrides{}, fmt.Errorf("read quota override for %s: %w", accountID, err)
+		return overrideRecord{}, fmt.Errorf("read quota override for %s: %w", accountID, err)
 	}
-	var over Overrides
-	if err := json.Unmarshal(entry.Value(), &over); err != nil {
-		return Overrides{}, fmt.Errorf("decode quota override for %s: %w", accountID, err)
+	var rec overrideRecord
+	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+		return overrideRecord{}, fmt.Errorf("decode quota override for %s: %w", accountID, err)
 	}
-	return over, nil
+	return rec, nil
 }
 
-// GetAccountQuota returns accountID's stored overrides and the limits they
-// resolve to. A read failure is surfaced here rather than masked, because an
-// operator inspecting a quota must not be shown the fallback as if it were real.
+// limitsFor resolves accountID's effective limits: overrides where set, the
+// config elsewhere. A read or decode failure falls back to the config so an
+// outage cannot lift a cap; a sealed or unknown-state record refuses instead.
+func (s *Service) limitsFor(ctx context.Context, accountID string) (Limits, error) {
+	rec, err := s.recordFor(ctx, accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "quota: override read failed, using configured limits",
+			"account", accountID, "err", err)
+		return s.limits, nil
+	}
+	if err := checkActive(rec); err != nil {
+		return Limits{}, mapOverrideError(ctx, accountID, err)
+	}
+	return rec.apply(s.limits), nil
+}
+
+// GetAccountQuota returns accountID's stored overrides and resolved limits.
+// Read failures surface rather than show the fallback as real; a sealed or
+// unknown-state record returns Blocked or InternalError.
 func (s *Service) GetAccountQuota(ctx context.Context, accountID string) (Overrides, Limits, error) {
-	over, err := s.overridesFor(ctx, accountID)
+	rec, err := s.recordFor(ctx, accountID)
 	if err != nil {
 		return Overrides{}, Limits{}, err
 	}
-	return over, over.apply(s.limits), nil
+	if err := checkActive(rec); err != nil {
+		return Overrides{}, Limits{}, mapOverrideError(ctx, accountID, err)
+	}
+	return rec.Overrides, rec.apply(s.limits), nil
 }
 
-// PutAccountQuota stores accountID's overrides, stamping provenance. An empty
-// set deletes the record so the account resolves to the config, which keeps
-// "has this account been touched?" a single existence check.
+// PutAccountQuota stores or clears accountID's overrides, stamping provenance
+// on a store. Both branches are revision-guarded and refuse a sealed record,
+// so a seal landing between a read and a write fails the CAS.
 func (s *Service) PutAccountQuota(ctx context.Context, accountID string, over Overrides, updatedBy string) error {
 	if s == nil || s.overrides == nil {
 		return errors.New("quota: override bucket not configured")
 	}
 	if over.Empty() {
-		if err := s.overrides.Delete(ctx, accountID); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
-			return fmt.Errorf("clear quota override for %s: %w", accountID, err)
-		}
-		return nil
+		return s.clearAccountQuota(ctx, accountID)
 	}
+
 	over.UpdatedAt = time.Now().UTC().Format(time.RFC3339)
 	over.UpdatedBy = updatedBy
-	data, err := json.Marshal(over)
-	if err != nil {
-		return fmt.Errorf("encode quota override for %s: %w", accountID, err)
-	}
-	if _, err := s.overrides.Put(ctx, accountID, data); err != nil {
-		return fmt.Errorf("store quota override for %s: %w", accountID, err)
-	}
-	return nil
+
+	_, err := kvutil.Update(ctx, s.overrides, accountID, kvutil.CASConfig{
+		CreateIfAbsent: true,
+		Attempts:       overrideMaxAttempts,
+		Exhausted:      func(string, int) error { return errOverridesContended },
+	}, func(rec *overrideRecord) (bool, error) {
+		if err := checkActive(*rec); err != nil {
+			return false, err
+		}
+		rec.Overrides = over
+		// Every success is a leader-accepted write over an active value: a
+		// seal landing between the read and the write fails this CAS.
+		return true, nil
+	})
+	return mapOverrideError(ctx, accountID, err)
+}
+
+// clearAccountQuota deletes an active override at the revision it read. An
+// observed-absent key writes nothing, so a stale read cannot unseal; a sealed
+// record refuses rather than being erased and later recreated active.
+func (s *Service) clearAccountQuota(ctx context.Context, accountID string) error {
+	for range overrideMaxAttempts {
+		entry, err := s.overrides.Get(ctx, accountID)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return nil
+		}
+		if err != nil {
+			return mapOverrideError(ctx, accountID, fmt.Errorf("read quota override for clear: %w", err))
+		}
+
+		var rec overrideRecord
+		if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+			return mapOverrideError(ctx, accountID, fmt.Errorf("decode quota override for clear: %w", err))
+		}
+		if err := checkActive(rec); err != nil {
+			return mapOverrideError(ctx, accountID, err)
+		}
+
+		err = s.overrides.Delete(ctx, accountID, jetstream.LastRevision(entry.Revision()))
+		switch {
+		case err == nil:
+			return nil
+		case errors.Is(err, jetstream.ErrKeyNotFound):
+			return nil
+		case isRevisionConflict(err):
+			// A concurrent write landed between the read and the guarded
+			// delete; retry against whatever revision now holds.
+			continue
+		default:
+			return mapOverrideError(ctx, accountID, fmt.Errorf("clear quota override: %w", err))
+		}
+	}
+	return mapOverrideError(ctx, accountID, errOverridesContended)
 }

spinifex/domains/admission/quota/override_kv_test.go

@@ -31,7 +31,8 @@ func newOverrideService(t *testing.T) (*Service, jetstream.KeyValue) {
 func TestLimitsForAbsentKeyInherits(t *testing.T) {
 	svc, _ := newOverrideService(t)
 
-	got := svc.limitsFor(t.Context(), "123456789012")
+	got, err := svc.limitsFor(t.Context(), "123456789012")
+	require.NoError(t, err)
 	require.Equal(t, baseLimits(), got)
 }
 
@@ -44,10 +45,14 @@ func TestLimitsForIsPerAccount(t *testing.T) {
 	require.NoError(t, svc.PutAccountQuota(t.Context(), upgraded,
 		Overrides{VCPUs: ptr(32), RDSInstances: ptr(8)}, "operator"))
 
-	require.Equal(t, 32, svc.limitsFor(t.Context(), upgraded).VCPUs)
-	require.Equal(t, 8, svc.limitsFor(t.Context(), upgraded).RDSInstances)
-	require.Equal(t, baseLimits().VCPUs, svc.limitsFor(t.Context(), sandbox).VCPUs)
-	require.Equal(t, baseLimits().RDSInstances, svc.limitsFor(t.Context(), sandbox).RDSInstances)
+	upgradedLimits, err := svc.limitsFor(t.Context(), upgraded)
+	require.NoError(t, err)
+	sandboxLimits, err := svc.limitsFor(t.Context(), sandbox)
+	require.NoError(t, err)
+	require.Equal(t, 32, upgradedLimits.VCPUs)
+	require.Equal(t, 8, upgradedLimits.RDSInstances)
+	require.Equal(t, baseLimits().VCPUs, sandboxLimits.VCPUs)
+	require.Equal(t, baseLimits().RDSInstances, sandboxLimits.RDSInstances)
 }
 
 // A stored override must round-trip through KV with its provenance, and an
@@ -79,7 +84,9 @@ func TestPutAccountQuotaEmptyClears(t *testing.T) {
 
 	_, err := kv.Get(t.Context(), account)
 	require.ErrorIs(t, err, jetstream.ErrKeyNotFound)
-	require.Equal(t, baseLimits(), svc.limitsFor(t.Context(), account))
+	limits, err := svc.limitsFor(t.Context(), account)
+	require.NoError(t, err)
+	require.Equal(t, baseLimits(), limits)
 }
 
 // Clearing an account that never had a record must succeed rather than fail on
@@ -97,8 +104,10 @@ func TestPutAccountQuotaExplicitZero(t *testing.T) {
 
 	require.NoError(t, svc.PutAccountQuota(t.Context(), account, Overrides{VPCs: ptr(0)}, "operator"))
 
-	require.Equal(t, 0, svc.limitsFor(t.Context(), account).VPCs)
-	require.Error(t, exceeds(0, 1, svc.limitsFor(t.Context(), account).VPCs))
+	limits, err := svc.limitsFor(t.Context(), account)
+	require.NoError(t, err)
+	require.Equal(t, 0, limits.VPCs)
+	require.Error(t, exceeds(0, 1, limits.VPCs))
 }
 
 // A corrupt record is a read failure, not an absent one: it must fall back to
@@ -110,7 +119,9 @@ func TestLimitsForCorruptRecordFailsClosed(t *testing.T) {
 	_, err := kv.Put(t.Context(), account, []byte("{not json"))
 	require.NoError(t, err)
 
-	require.Equal(t, baseLimits(), svc.limitsFor(t.Context(), account),
+	limits, err := svc.limitsFor(t.Context(), account)
+	require.NoError(t, err, "a decode failure must fall back, not error, from limitsFor")
+	require.Equal(t, baseLimits(), limits,
 		"a corrupt override must fall back to the configured limits")
 
 	_, _, err = svc.GetAccountQuota(t.Context(), account)

spinifex/domains/admission/quota/override_test.go

@@ -126,7 +126,11 @@ func TestOverridesEmpty(t *testing.T) {
 // fall through to the configured limits rather than panicking.
 func TestLimitsForWithoutBucket(t *testing.T) {
 	svc := New(baseLimits(), nil)
-	if got := svc.limitsFor(t.Context(), "123456789012"); got != baseLimits() {
+	got, err := svc.limitsFor(t.Context(), "123456789012")
+	if err != nil {
+		t.Fatalf("limitsFor error = %v, want nil", err)
+	}
+	if got != baseLimits() {
 		t.Fatalf("limitsFor = %+v, want the configured limits", got)
 	}
 }

spinifex/domains/admission/quota/override_writes_guarded_test.go

@@ -0,0 +1,75 @@
+package quota
+
+//test:in-package — pins what AccountQuotaOverrideSealCapabilityVersion 1 will
+// assert: nothing in this package writes the override bucket except through
+// PutAccountQuota's CAS loop, clearAccountQuota's guarded delete or
+// SealAccountOverrides' CAS loop.
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"path/filepath"
+	"runtime"
+	"testing"
+)
+
+// overrideWriteFiles are the only files in this package that touch the
+// override bucket. The package also holds the vCPU usage and Bedrock usage
+// buckets, which write through their own stores and are out of scope here;
+// scanning the whole package directory, as the EC2 settings equivalent does,
+// would false-flag those unrelated writes.
+var overrideWriteFiles = []string{"override.go", "seal.go"}
+
+func quotaPackageDir(t *testing.T) string {
+	t.Helper()
+	_, file, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("cannot locate test source")
+	}
+	return filepath.Dir(file)
+}
+
+// No file owning the override bucket may call Put, PutString, or Delete
+// without a revision option: every write must go through PutAccountQuota's
+// CAS loop, clearAccountQuota's guarded delete, or SealAccountOverrides' CAS
+// loop.
+func TestOverrideWritesAreRevisionGuarded(t *testing.T) {
+	root := quotaPackageDir(t)
+	fset := token.NewFileSet()
+	scanned := 0
+
+	for _, name := range overrideWriteFiles {
+		path := filepath.Join(root, name)
+		f, err := parser.ParseFile(fset, path, nil, 0)
+		if err != nil {
+			t.Fatalf("parse %s: %v", path, err)
+		}
+		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
+			}
+			switch sel.Sel.Name {
+			case "Put", "PutString":
+				t.Errorf("%s: unconditional %s call bypasses the revision-guarded override write; use PutAccountQuota or SealAccountOverrides instead",
+					fset.Position(call.Pos()), sel.Sel.Name)
+			case "Delete":
+				if len(call.Args) < 3 {
+					t.Errorf("%s: Delete call with no revision option bypasses the guarded clear; pin it to the read revision",
+						fset.Position(call.Pos()))
+				}
+			}
+			return true
+		})
+	}
+	if scanned == 0 {
+		t.Fatalf("scanned no override-write files under %s; the invariant would pass vacuously", root)
+	}
+}

spinifex/domains/admission/quota/quota.go

@@ -90,6 +90,14 @@ type Service struct {
 	// overrides holds the per-account limit overrides. Nil leaves every account
 	// on the configured limits, which is also what an absent key means.
 	overrides jetstream.KeyValue
+
+	// js opens the override bucket for SealAccountOverrides when quotas were
+	// disabled at startup; nil makes such a seal refuse rather than create it.
+	js jetstream.KeyValueManager
+
+	// terminationChecker is the seal precondition; SealAccountOverrides
+	// refuses until it is wired.
+	terminationChecker AccountTerminationChecker
 }
 
 // New constructs a quota Service from the configured limits and the gateway-owned
@@ -125,6 +133,16 @@ func (s *Service) SetOverrides(kv jetstream.KeyValue) {
 	s.overrides = kv
 }
 
+// SetJetStream wires the client SealAccountOverrides uses to get-or-create
+// the override bucket when this gateway disabled quotas. Called once at
+// wiring time, regardless of whether quotas are enabled here.
+func (s *Service) SetJetStream(js jetstream.KeyValueManager) {
+	if s == nil {
+		return
+	}
+	s.js = js
+}
+
 // Exempt returns true for the global/system account and whenever quotas are
 // disabled, so those callers bypass every quota check. A nil Service (quotas
 // never built) is treated as disabled so callers need no nil guard.

spinifex/domains/admission/quota/seal.go

@@ -0,0 +1,163 @@
+package quota
+
+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/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// sealMaxAttempts bounds SealAccountOverrides' 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. No dimension or provenance
+// survives, so no tenant commercial setting outlives the seal.
+var sealedRecord = &overrideRecord{State: OverrideStateSealed}
+
+// OverrideSeal is SealAccountOverrides' result: the revision at which the
+// stream leader accepted the sealed value.
+type OverrideSeal 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 seal precondition; until it is set,
+// SealAccountOverrides refuses.
+func (s *Service) SetAccountTerminationChecker(c AccountTerminationChecker) {
+	if s == nil {
+		return
+	}
+	s.terminationChecker = c
+}
+
+// checkTerminating runs before any override-KV call, so a refusal never
+// writes or opens the bucket. Not TERMINATING (incl. missing) is
+// IncorrectState; a read failure or unwired checker is InternalError.
+func (s *Service) checkTerminating(ctx context.Context, accountID string) error {
+	if s.terminationChecker == nil {
+		slog.ErrorContext(ctx, "Quota override 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 to retry, including the bare
+// wrong-last-sequence APIError from Create's retry against a delete marker.
+// A private copy of domains/ec2/account's helper, which this package may not import.
+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
+}
+
+// sealBucket returns this gateway's override bucket, or get-or-creates it
+// when quotas are disabled here. Called only after checkTerminating accepts,
+// so a refused seal never creates the bucket.
+func (s *Service) sealBucket(ctx context.Context) (jetstream.KeyValue, error) {
+	if s.overrides != nil {
+		return s.overrides, nil
+	}
+	if s.js == nil {
+		slog.ErrorContext(ctx, "Quota override seal has no JetStream client wired to open the override bucket")
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+	kv, err := kvutil.GetOrCreateBucket(ctx, s.js, KVBucketAccountQuota, 1)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to open quota override bucket for seal", "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+	return kv, nil
+}
+
+// SealAccountOverrides seals accountID's override record so no later writer
+// can recreate it. Success is only a leader-accepted write, evidenced by the
+// revision; a missing key is created sealed, never reported from absence.
+func (s *Service) SealAccountOverrides(ctx context.Context, accountID string) (*OverrideSeal, error) {
+	if accountID == "" {
+		return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required to seal quota overrides")
+	}
+	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 seal must not target the global account")
+	}
+
+	if err := s.checkTerminating(ctx, accountID); err != nil {
+		return nil, err
+	}
+
+	overrides, err := s.sealBucket(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	data, err := json.Marshal(sealedRecord)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to encode sealed quota override record", "accountID", accountID, "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+
+	// accountID is never resolved through any fallback here: a seal must act
+	// on exactly accountID and never fall back to the global key.
+	for range sealMaxAttempts {
+		var rev uint64
+		entry, getErr := overrides.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 quota override for seal", "accountID", accountID, "err", getErr)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+
+		var newRev uint64
+		var writeErr error
+		if rev == 0 {
+			newRev, writeErr = overrides.Create(ctx, accountID, data)
+		} else {
+			newRev, writeErr = overrides.Update(ctx, accountID, data, rev)
+		}
+
+		switch {
+		case writeErr == nil:
+			return &OverrideSeal{Revision: newRev}, nil
+		case isRevisionConflict(writeErr):
+			continue
+		default:
+			slog.ErrorContext(ctx, "Failed to write sealed quota override record", "accountID", accountID, "err", writeErr)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+	}
+
+	slog.ErrorContext(ctx, "Quota override seal exhausted attempts against concurrent writes", "accountID", accountID, "attempts", sealMaxAttempts)
+	return nil, errors.New(awserrors.ErrorInternalError)
+}

spinifex/domains/admission/quota/seal_test.go

@@ -0,0 +1,806 @@
+package quota
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+
+	"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}}
+}
+
+// newSealService builds a Service over a real JetStream KV bucket with the
+// override bucket already attached, mirroring a quota-enabled gateway.
+func newSealService(t *testing.T) (*Service, jetstream.JetStream) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	kv, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{
+		Bucket:  KVBucketAccountQuota,
+		History: 1,
+	})
+	require.NoError(t, err)
+	usage, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{
+		Bucket:  KVBucketAccountUsage,
+		History: 1,
+	})
+	require.NoError(t, err)
+
+	svc := New(baseLimits(), usage)
+	svc.SetOverrides(kv)
+	svc.SetJetStream(js)
+	return svc, js
+}
+
+// newDisabledSealService builds a Service with only the JetStream client
+// wired, mirroring a quota-disabled gateway: the override bucket does not
+// exist until a seal creates it.
+func newDisabledSealService(t *testing.T) (*Service, jetstream.JetStream) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	svc := New(baseLimits(), nil)
+	svc.SetJetStream(js)
+	return svc, js
+}
+
+// lastOverrideSeq reports the override stream's last sequence, so a test can
+// assert a failed attempt wrote nothing.
+func lastOverrideSeq(t *testing.T, js jetstream.JetStream) uint64 {
+	t.Helper()
+	stream, err := js.Stream(context.Background(), "KV_"+KVBucketAccountQuota)
+	require.NoError(t, err)
+	info, err := stream.Info(context.Background())
+	require.NoError(t, err)
+	return info.State.LastSeq
+}
+
+// mutationCase is one of PutAccountQuota's two mutations, called uniformly so
+// a test can table over them.
+type mutationCase struct {
+	name string
+	call func(svc *Service, ctx context.Context) error
+}
+
+func mutationTable() []mutationCase {
+	return []mutationCase{
+		{"Set", func(svc *Service, ctx context.Context) error {
+			return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator")
+		}},
+		{"Clear", func(svc *Service, ctx context.Context) error {
+			return svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator")
+		}},
+	}
+}
+
+// seedOverrideState prepares testAccount'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 seedOverrideState(t *testing.T, svc *Service, state string) {
+	t.Helper()
+	ctx := context.Background()
+	switch state {
+	case "present":
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+	case "tombstone":
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+		require.NoError(t, svc.overrides.Delete(ctx, testAccount))
+	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, Update or Delete),
+// 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)
+}
+
+func (f *sealRaceKV) Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error {
+	f.writeCalls++
+	if f.beforeWrite != nil {
+		if err := f.beforeWrite(f.writeCalls); err != nil {
+			return err
+		}
+	}
+	return f.KeyValue.Delete(ctx, key, opts...)
+}
+
+// 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 decodeOverride(t *testing.T, entry jetstream.KeyValueEntry) overrideRecord {
+	t.Helper()
+	var rec overrideRecord
+	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. Clear on an absent or tombstoned key never reaches a
+// write at all (see TestClearOnMissingKeyWritesNothing), so only the four
+// cells with a real write are exercised here.
+func TestSealRaceMutation_SealLandsBetweenReadAndWrite(t *testing.T) {
+	cases := []struct {
+		name  string
+		state string
+		call  func(svc *Service, ctx context.Context) error
+	}{
+		{"Set_present", "present", mutationTable()[0].call},
+		{"Set_absent", "absent", mutationTable()[0].call},
+		{"Set_tombstone", "tombstone", mutationTable()[0].call},
+		{"Clear_present", "present", mutationTable()[1].call},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			svc, _ := newSealService(t)
+			ctx := context.Background()
+			seedOverrideState(t, svc, tc.state)
+
+			realKV := svc.overrides
+			sealer := New(baseLimits(), nil)
+			sealer.SetOverrides(realKV)
+			sealer.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+			var sealed *OverrideSeal
+			race := &sealRaceKV{KeyValue: realKV}
+			race.beforeWrite = func(attempt int) error {
+				if attempt != 1 {
+					return nil
+				}
+				var sealErr error
+				sealed, sealErr = sealer.SealAccountOverrides(ctx, testAccount)
+				return sealErr
+			}
+			svc.overrides = race
+
+			err := tc.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, testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, OverrideStateSealed, decodeOverride(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 no overridden dimension.
+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, _ := newSealService(t)
+				ctx := context.Background()
+				seedOverrideState(t, svc, state)
+
+				realKV := svc.overrides
+				race := &sealRaceKV{KeyValue: realKV}
+				race.beforeWrite = func(attempt int) error {
+					if attempt != 1 {
+						return nil
+					}
+					_ = m.call(svc, ctx)
+					return nil
+				}
+
+				sealer := New(baseLimits(), nil)
+				sealer.SetOverrides(race)
+				sealer.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+				seal, err := sealer.SealAccountOverrides(ctx, testAccount)
+				require.NoError(t, err)
+
+				entry, err := realKV.Get(ctx, testAccount)
+				require.NoError(t, err)
+				rec := decodeOverride(t, entry)
+				assert.Equal(t, OverrideStateSealed, rec.State)
+				assert.Nil(t, rec.VCPUs)
+				assert.Equal(t, seal.Revision, entry.Revision())
+			})
+		}
+	}
+}
+
+// Neither mutation 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, _ := newSealService(t)
+			ctx := context.Background()
+			svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+			_, err := svc.SealAccountOverrides(ctx, testAccount)
+			require.NoError(t, err)
+
+			before, err := svc.overrides.Get(ctx, testAccount)
+			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.overrides.Get(ctx, testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, before.Revision(), after.Revision())
+		})
+	}
+}
+
+// Every override 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, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+
+	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 *OverrideSeal
+	var sealErr error
+	wg.Go(func() {
+		sealResult, sealErr = svc.SealAccountOverrides(ctx, testAccount)
+	})
+	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.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State)
+	assert.Equal(t, sealResult.Revision, entry.Revision())
+
+	_, _, err = svc.GetAccountQuota(ctx, testAccount)
+	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, js := newSealService(t)
+	ctx := context.Background()
+
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+	stale, err := svc.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(16)}, "operator"))
+
+	before := lastOverrideSeq(t, js)
+	svc.overrides = &staleGetKV{KeyValue: svc.overrides, entry: stale}
+
+	err = svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+// 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, js := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	before := lastOverrideSeq(t, js)
+	svc.overrides = &staleNotFoundKV{KeyValue: svc.overrides}
+
+	err = svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+// 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, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+
+	realKV := svc.overrides
+	race := &sealRaceKV{KeyValue: realKV}
+	race.beforeWrite = func(int) error {
+		data, merr := json.Marshal(&overrideRecord{Overrides: Overrides{VCPUs: ptr(16)}})
+		require.NoError(t, merr)
+		_, perr := realKV.Put(ctx, testAccount, data)
+		return perr
+	}
+	svc.overrides = race
+
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, sealMaxAttempts, race.writeCalls)
+
+	entry, err := realKV.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.NotEqual(t, OverrideStateSealed, decodeOverride(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, js := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+	before := lastOverrideSeq(t, js)
+	svc.overrides = &errGetKV{KeyValue: svc.overrides, err: errors.New("kv unavailable")}
+
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+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))
+}
+
+// Clearing never writes when there is nothing active to clear: an absent key,
+// a tombstoned key, and a sealed key hidden behind a stale not-found Get all
+// leave the stream exactly where they found it.
+func TestClearOnMissingKeyWritesNothing(t *testing.T) {
+	t.Run("absent", func(t *testing.T) {
+		svc, js := newSealService(t)
+		ctx := context.Background()
+		before := lastOverrideSeq(t, js)
+
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"))
+		assert.Equal(t, before, lastOverrideSeq(t, js))
+	})
+
+	t.Run("tombstone", func(t *testing.T) {
+		svc, js := newSealService(t)
+		ctx := context.Background()
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+		require.NoError(t, svc.overrides.Delete(ctx, testAccount))
+		before := lastOverrideSeq(t, js)
+
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"))
+		assert.Equal(t, before, lastOverrideSeq(t, js))
+	})
+
+	t.Run("sealed_hidden_by_stale_not_found", func(t *testing.T) {
+		svc, js := newSealService(t)
+		ctx := context.Background()
+		svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+		sealed, err := svc.SealAccountOverrides(ctx, testAccount)
+		require.NoError(t, err)
+		before := lastOverrideSeq(t, js)
+
+		realKV := svc.overrides
+		svc.overrides = &staleNotFoundKV{KeyValue: realKV}
+
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"))
+		assert.Equal(t, before, lastOverrideSeq(t, js))
+
+		entry, err := realKV.Get(ctx, testAccount)
+		require.NoError(t, err)
+		rec := decodeOverride(t, entry)
+		assert.Equal(t, OverrideStateSealed, rec.State)
+		assert.Equal(t, sealed.Revision, entry.Revision())
+	})
+}
+
+// A seal against an absent or a tombstoned key creates the sealed record.
+func TestSealOnMissingKeyCreatesSealedRecord(t *testing.T) {
+	for _, state := range []string{"absent", "tombstone"} {
+		t.Run(state, func(t *testing.T) {
+			svc, _ := newSealService(t)
+			ctx := context.Background()
+			svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+			if state == "tombstone" {
+				require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+				require.NoError(t, svc.overrides.Delete(ctx, testAccount))
+			}
+			_, err := svc.overrides.Get(ctx, testAccount)
+			require.ErrorIs(t, err, jetstream.ErrKeyNotFound)
+
+			seal, err := svc.SealAccountOverrides(ctx, testAccount)
+			require.NoError(t, err)
+
+			entry, err := svc.overrides.Get(ctx, testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, seal.Revision, entry.Revision())
+			assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State)
+		})
+	}
+}
+
+// alwaysConflictThenErrorKV answers Get as absent and every Create attempt
+// with a conflict, then a store error on the final attempt: a recording
+// wrapper proving the seal tried a real write on every attempt before giving
+// up, rather than ever reporting success from an observed absence.
+type alwaysConflictThenErrorKV struct {
+	jetstream.KeyValue
+
+	calls int
+}
+
+func (f *alwaysConflictThenErrorKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) {
+	return nil, jetstream.ErrKeyNotFound
+}
+
+func (f *alwaysConflictThenErrorKV) Create(context.Context, string, []byte, ...jetstream.KVCreateOpt) (uint64, error) {
+	f.calls++
+	if f.calls < sealMaxAttempts {
+		return 0, jetstream.ErrKeyExists
+	}
+	return 0, errors.New("store unavailable")
+}
+
+func TestSealNeverSucceedsWithoutAWrite(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	fake := &alwaysConflictThenErrorKV{}
+	svc.overrides = fake
+
+	seal, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.Error(t, err)
+	assert.Nil(t, seal)
+	assert.Equal(t, sealMaxAttempts, fake.calls, "every attempt must have tried a real write before giving up")
+}
+
+// An account IAM does not hold as TERMINATING refuses the seal before any
+// override-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, js := newSealService(t)
+			ctx := context.Background()
+			if state == "present" {
+				require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+			}
+			svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+			before := lastOverrideSeq(t, js)
+			_, err := svc.SealAccountOverrides(ctx, testAccount)
+			assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+			assert.Equal(t, before, lastOverrideSeq(t, js))
+		})
+	}
+}
+
+func TestSealRefusedForMissingIAMRecord(t *testing.T) {
+	svc, js := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+	before := lastOverrideSeq(t, js)
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+func TestSealRefusedWhenIAMReadFails(t *testing.T) {
+	svc, js := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")})
+
+	before := lastOverrideSeq(t, js)
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+func TestSealRefusedWithNoCheckerWired(t *testing.T) {
+	svc, js := newSealService(t)
+	ctx := context.Background()
+
+	before := lastOverrideSeq(t, js)
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+func TestSealAcceptedForTerminatingAccount(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+	seal, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+	assert.NotZero(t, seal.Revision)
+
+	entry, err := svc.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, seal.Revision, entry.Revision())
+}
+
+// A refused seal, on a gateway that disabled quotas and so holds no override
+// bucket, never creates one.
+func TestSealRefusalDoesNotCreateBucket(t *testing.T) {
+	t.Run("not_terminating", func(t *testing.T) {
+		svc, js := newDisabledSealService(t)
+		ctx := context.Background()
+		svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+		_, err := svc.SealAccountOverrides(ctx, testAccount)
+		assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+		_, err = js.KeyValue(context.Background(), KVBucketAccountQuota)
+		assert.ErrorIs(t, err, jetstream.ErrBucketNotFound)
+	})
+
+	t.Run("iam_read_fails", func(t *testing.T) {
+		svc, js := newDisabledSealService(t)
+		ctx := context.Background()
+		svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")})
+
+		_, err := svc.SealAccountOverrides(ctx, testAccount)
+		assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+		_, err = js.KeyValue(context.Background(), KVBucketAccountQuota)
+		assert.ErrorIs(t, err, jetstream.ErrBucketNotFound)
+	})
+}
+
+// A seal accepted on a gateway whose quotas were disabled at startup creates
+// the override bucket at seal time, never at startup.
+func TestSealOnQuotaDisabledGatewayCreatesBucket(t *testing.T) {
+	svc, js := newDisabledSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+	seal, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	kv, err := js.KeyValue(context.Background(), KVBucketAccountQuota)
+	require.NoError(t, err)
+	entry, err := kv.Get(context.Background(), testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, seal.Revision, entry.Revision())
+	assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State)
+}
+
+// A sealed record stays sealed after the service is rebuilt over the same
+// JetStream, every read and write refuses it, and a re-seal gets a higher
+// revision.
+func TestSealSurvivesServiceReconstruction(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	ctx := context.Background()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: KVBucketAccountQuota, History: 1})
+	require.NoError(t, err)
+
+	svc := New(baseLimits(), nil)
+	svc.SetOverrides(kv)
+	svc.SetJetStream(js)
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	first, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	rebuiltKV, err := js.KeyValue(ctx, KVBucketAccountQuota)
+	require.NoError(t, err)
+	rebuilt := New(baseLimits(), nil)
+	rebuilt.SetOverrides(rebuiltKV)
+	rebuilt.SetJetStream(js)
+
+	entry, err := rebuilt.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State)
+
+	_, _, err = rebuilt.GetAccountQuota(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	assert.True(t, awserrors.IsErrorCode(rebuilt.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"), awserrors.ErrorBlocked))
+	assert.True(t, awserrors.IsErrorCode(rebuilt.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"), awserrors.ErrorBlocked))
+
+	limits, err := rebuilt.limitsFor(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	assert.Zero(t, limits)
+
+	rebuilt.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	second, err := rebuilt.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Greater(t, second.Revision, first.Revision)
+}
+
+func TestSealRefusesEmptyInvalidAndGlobalAccount(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+
+	_, err := svc.SealAccountOverrides(ctx, "")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorMissingParameter), "got %v", err)
+
+	_, err = svc.SealAccountOverrides(ctx, "not-an-account")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err)
+
+	_, err = svc.SealAccountOverrides(ctx, awsidentifiers.GlobalAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err)
+}
+
+func TestSealLeavesOtherAccountsUntouched(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+	accountB := "222222222222"
+	require.NoError(t, svc.PutAccountQuota(ctx, accountB, Overrides{VCPUs: ptr(8)}, "operator"))
+
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	over, _, err := svc.GetAccountQuota(ctx, accountB)
+	require.NoError(t, err)
+	require.NotNil(t, over.VCPUs)
+	assert.Equal(t, 8, *over.VCPUs)
+}
+
+// A record written before the state field existed has none and is treated as
+// active by both reads and writes; a write on it preserves the old shape.
+func TestRecordWithoutStateIsActive(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+
+	data, err := json.Marshal(Overrides{VCPUs: ptr(8)})
+	require.NoError(t, err)
+	_, err = svc.overrides.Put(ctx, testAccount, data)
+	require.NoError(t, err)
+
+	over, limits, err := svc.GetAccountQuota(ctx, testAccount)
+	require.NoError(t, err)
+	require.NotNil(t, over.VCPUs)
+	assert.Equal(t, 8, limits.VCPUs)
+
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(16)}, "operator"))
+	entry, err := svc.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.NotContains(t, string(entry.Value()), `"state"`)
+}
+
+// A record whose state is neither active nor sealed fails every read and
+// write closed, rather than being treated as active.
+func TestUnknownStateFailsClosed(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+
+	_, err := svc.overrides.Put(ctx, testAccount, []byte(`{"state":"frozen"}`))
+	require.NoError(t, err)
+
+	_, _, err = svc.GetAccountQuota(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+
+	limits, err := svc.limitsFor(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Zero(t, limits)
+
+	assert.True(t, awserrors.IsErrorCode(svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"), awserrors.ErrorInternalError))
+	assert.True(t, awserrors.IsErrorCode(svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"), awserrors.ErrorInternalError))
+}
+
+// A sealed record is Blocked for admission's own resolver, which every
+// caller (admin read, ELBv2 DescribeAccountLimits, every Enforce* method)
+// shares.
+func TestAdmissionOnSealedRecordIsBlocked(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	limits, err := svc.limitsFor(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err)
+	assert.Zero(t, limits)
+
+	assert.Error(t, svc.CheckVCPU(ctx, testAccount, 1))
+	assert.True(t, awserrors.IsErrorCode(svc.CheckVCPU(ctx, testAccount, 1), awserrors.ErrorBlocked))
+}
+
+// The existing read-failure fallback in limitsFor (a decode or store error
+// falls back to the configured limits) is unchanged by the seal work: it is
+// distinct from a successfully read sealed record, which refuses instead.
+func TestAdmissionReadFailureFallbackUnchanged(t *testing.T) {
+	svc, kv := newOverrideService(t)
+
+	_, err := kv.Put(t.Context(), testAccount, []byte("{not json"))
+	require.NoError(t, err)
+
+	limits, err := svc.limitsFor(t.Context(), testAccount)
+	require.NoError(t, err, "a decode failure must still fall back, not error, from limitsFor")
+	require.Equal(t, baseLimits(), limits)
+}

spinifex/domains/admission/quota/storage.go

@@ -23,7 +23,10 @@ func (s *Service) EnforceVolumeCreate(ctx context.Context, natsConn *nats.Conn,
 	if err != nil {
 		return err
 	}
-	limits := s.limitsFor(ctx, accountID)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
 	if err := exceeds(len(out.Volumes), 1, limits.Volumes); err != nil {
 		return err
 	}
@@ -43,7 +46,11 @@ func (s *Service) EnforceVolumeModify(ctx context.Context, natsConn *nats.Conn,
 	if err != nil {
 		return err
 	}
-	return exceeds(total-oldGiB, newGiB, s.limitsFor(ctx, accountID).VolumesGiB)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(total-oldGiB, newGiB, limits.VolumesGiB)
 }
 
 // volumeUsage sums the account's volume sizes in GiB via a single DescribeVolumes

spinifex/domains/admission/quota/vcpu.go

@@ -37,7 +37,11 @@ func (s *Service) CheckVCPU(ctx context.Context, accountID string, want int) err
 	if err != nil {
 		return err
 	}
-	return exceeds(current, want, s.limitsFor(ctx, accountID).VCPUs)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(current, want, limits.VCPUs)
 }
 
 // casVCPU applies next to accountID's counter under bounded JetStream CAS so

spinifex/gateway/admin_accountquota.go

@@ -43,6 +43,9 @@ func (gw *GatewayConfig) adminGetAccountQuota(ctx context.Context, body []byte)
 
 	over, limits, err := gw.Quota.GetAccountQuota(ctx, accountID)
 	if err != nil {
+		if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) {
+			return nil, err
+		}
 		slog.ErrorContext(ctx, "GetAccountQuota: failed to read override", "accountID", accountID, "err", err)
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}
@@ -61,6 +64,9 @@ func (gw *GatewayConfig) adminPutAccountQuota(ctx context.Context, body []byte)
 	// The caller's own identity, so a quota change names who made it.
 	updatedBy, _ := ctx.Value(ctxIdentity).(string)
 	if err := gw.Quota.PutAccountQuota(ctx, accountID, over, updatedBy); err != nil {
+		if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) {
+			return nil, err
+		}
 		slog.ErrorContext(ctx, "PutAccountQuota: failed to store override", "accountID", accountID, "err", err)
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}
@@ -69,6 +75,9 @@ func (gw *GatewayConfig) adminPutAccountQuota(ctx context.Context, body []byte)
 
 	stored, limits, err := gw.Quota.GetAccountQuota(ctx, accountID)
 	if err != nil {
+		if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) {
+			return nil, err
+		}
 		slog.ErrorContext(ctx, "PutAccountQuota: failed to read back override", "accountID", accountID, "err", err)
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}

spinifex/gateway/admin_accountquota_test.go

@@ -205,3 +205,56 @@ func TestAccountQuotaMethodsAreRegistered(t *testing.T) {
 	require.Contains(t, AdminMethodNames(), "GetAccountQuota")
 	require.Contains(t, AdminMethodNames(), "PutAccountQuota")
 }
+
+// quotaStubChecker answers AccountTerminating from a fixed set, so these
+// tests can seal a record without a real IAM bucket.
+type quotaStubChecker struct {
+	terminating map[string]bool
+}
+
+func (c *quotaStubChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) {
+	return c.terminating[accountID], nil
+}
+
+// A request body cannot carry or alter the stored state: an attempted
+// "state":"sealed" in the overrides object is simply unknown to the wire
+// type, the stored record stays active, and the response never shows state.
+func TestAdminRequestCannotCarryState(t *testing.T) {
+	gw, account := quotaTestGateway(t)
+	ctx := context.WithValue(t.Context(), ctxIdentity, "operator")
+
+	body := []byte(`{"accountId":"` + account + `","overrides":{"vcpus":32,"state":"sealed"}}`)
+	output, err := gw.adminPutAccountQuota(ctx, body)
+	require.NoError(t, err)
+
+	resp, ok := output.(*AccountQuotaResponse)
+	require.True(t, ok)
+	require.Equal(t, 32, resp.Limits["vcpus"])
+
+	data, err := json.Marshal(resp.Overrides)
+	require.NoError(t, err)
+	require.NotContains(t, string(data), "state")
+
+	_, _, err = gw.Quota.GetAccountQuota(ctx, account)
+	require.NoError(t, err, "the record must still be active, not sealed")
+}
+
+// Admin get and the put read-back on a sealed record must return Blocked,
+// not the generic InternalError every other read failure maps to.
+func TestAdminQuotaReadsMapBlocked(t *testing.T) {
+	gw, account := quotaTestGateway(t)
+	gw.Quota.SetAccountTerminationChecker(&quotaStubChecker{terminating: map[string]bool{account: true}})
+	_, err := gw.Quota.SealAccountOverrides(t.Context(), account)
+	require.NoError(t, err)
+
+	_, err = gw.adminGetAccountQuota(t.Context(), quotaRequestBody(t, AccountQuotaRequest{AccountID: account}))
+	require.Error(t, err)
+	require.Equal(t, awserrors.ErrorBlocked, err.Error())
+
+	ctx := context.WithValue(t.Context(), ctxIdentity, "operator")
+	_, err = gw.adminPutAccountQuota(ctx, quotaRequestBody(t, AccountQuotaRequest{
+		AccountID: account, Overrides: admissionquota.Overrides{VCPUs: quotaPtr(8)},
+	}))
+	require.Error(t, err)
+	require.Equal(t, awserrors.ErrorBlocked, err.Error())
+}

spinifex/gateway/elbv2.go

@@ -187,6 +187,9 @@ func accountLoadBalancerLimit(ctx context.Context, gw *GatewayConfig, accountID
 	}
 	_, limits, err := gw.Quota.GetAccountQuota(ctx, accountID)
 	if err != nil {
+		if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) {
+			return 0, err
+		}
 		slog.ErrorContext(ctx, "DescribeAccountLimits: failed to read account quota", "accountID", accountID, "err", err)
 		return 0, errors.New(awserrors.ErrorInternalError)
 	}

spinifex/gateway/elbv2_accountlimits_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	"github.com/mulgadc/spinifex/spinifex/domains/admission/quota"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	gateway_elbv2 "github.com/mulgadc/spinifex/spinifex/gateway/elbv2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -121,6 +122,20 @@ func TestELBv2AccountLoadBalancerLimitDefaultsForExemptAccount(t *testing.T) {
 	assert.Equal(t, gateway_elbv2.DefaultLoadBalancerLimit, limit)
 }
 
+// A sealed override must refuse the load balancer cap with Blocked rather
+// than the generic InternalError every other read failure maps to.
+func TestELBv2AccountLoadBalancerLimitBlockedOnSealedRecord(t *testing.T) {
+	gw, account := quotaTestGateway(t)
+	gw.Quota.SetAccountTerminationChecker(&quotaStubChecker{terminating: map[string]bool{account: true}})
+	_, err := gw.Quota.SealAccountOverrides(t.Context(), account)
+	require.NoError(t, err)
+
+	limit, err := accountLoadBalancerLimit(t.Context(), gw, account)
+	require.Error(t, err)
+	require.Equal(t, awserrors.ErrorBlocked, err.Error())
+	assert.Zero(t, limit)
+}
+
 // The dispatch entry must accept the paging parameters the model defines and
 // still answer with the whole list.
 func TestELBv2DescribeAccountLimitsAcceptsPaging(t *testing.T) {

spinifex/handlers/iam/account_status_reader.go

@@ -10,8 +10,8 @@ import (
 )
 
 // 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.
+// writes. It is the read-only capability EC2's settings seal and the quota
+// override seal both use to check their TERMINATING precondition.
 type AccountStatusReader struct {
 	accounts *kvstore.Bucket // AttachOnly: never creates the bucket
 }

spinifex/runtime/roles/awsgw/awsgw.go

@@ -60,6 +60,11 @@ var serviceName = "awsgw"
 // handlers/eks, because handlers/eks must not import handlers/sts.
 var _ handlers_sts.ClusterJWKSResolver = (*handlers_eks.OIDCJWKSReader)(nil)
 
+// AccountStatusReader also serves as the quota override seal's TERMINATING
+// precondition, wired here because handlers/iam must not import
+// domains/admission/quota.
+var _ admissionquota.AccountTerminationChecker = (*handlers_iam.AccountStatusReader)(nil)
+
 // Version and Commit are set by the cmd package before Start() to pass
 // build-time ldflags to the gateway without creating an import cycle.
 var (
@@ -561,6 +566,25 @@ func launchService(config *config.ClusterConfig) error {
 	gw.Quota = admissionquota.New(quotaCfg, usageBucket)
 	gw.Quota.SetOverrides(quotaOverrides)
 
+	// Seal precondition: the quota owner's only fact from IAM, read directly
+	// rather than through the gateway's cached IAMService, so a seal sees a
+	// TERMINATING flip the moment IAM commits it.
+	gw.Quota.SetAccountTerminationChecker(handlers_iam.NewAccountStatusReader(js))
+	// SealAccountOverrides opens the override bucket itself when this
+	// gateway's quotas are disabled, so the seal subject works regardless of
+	// quotaCfg.Enabled and never touches the bucket at startup.
+	gw.Quota.SetJetStream(js)
+
+	quotaSealSub, err := registerQuotaOverrideSeal(natsConn, gw.Quota)
+	if err != nil {
+		return fmt.Errorf("subscribe quota override seal: %w", err)
+	}
+	defer func() {
+		if err := quotaSealSub.Unsubscribe(); err != nil {
+			slog.Warn("Quota override seal: unsubscribe failed", "err", err)
+		}
+	}()
+
 	// Bedrock token quota reads the stream-fed usage counters built above,
 	// independent of whether the standing-infra dimensions are enabled.
 	gw.Quota.SetBedrockUsage(bedrockUsage)

spinifex/runtime/roles/awsgw/quota_override_seal.go

@@ -0,0 +1,39 @@
+package awsgw
+
+import (
+	"context"
+
+	admissionv1 "github.com/mulgadc/spinifex/contracts/admission/v1"
+	admissionquota "github.com/mulgadc/spinifex/spinifex/domains/admission/quota"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// quotaOverrideSealQueue is the NATS queue group for the quota override seal
+// responders so the request is load-balanced across awsgw nodes.
+const quotaOverrideSealQueue = "awsgw-quota-seal"
+
+// registerQuotaOverrideSeal serves SealAccountQuotaOverrideSubject on this
+// gateway. Any current gateway may answer; the leader-accepted revision is
+// the evidence.
+func registerQuotaOverrideSeal(nc *nats.Conn, quota *admissionquota.Service) (*nats.Subscription, error) {
+	return nc.QueueSubscribe(admissionv1.SealAccountQuotaOverrideSubject, quotaOverrideSealQueue,
+		func(msg *nats.Msg) {
+			natsmsg.ServeNATSRequestCtx(msg, handleSealAccountQuotaOverride(quota))
+		})
+}
+
+func handleSealAccountQuotaOverride(quota *admissionquota.Service) func(context.Context, *admissionv1.SealAccountQuotaOverrideRequest) (*admissionv1.SealAccountQuotaOverrideReply, error) {
+	return func(ctx context.Context, req *admissionv1.SealAccountQuotaOverrideRequest) (*admissionv1.SealAccountQuotaOverrideReply, error) {
+		seal, err := quota.SealAccountOverrides(ctx, req.AccountID)
+		if err != nil {
+			return nil, err
+		}
+		return &admissionv1.SealAccountQuotaOverrideReply{
+			AccountID:      req.AccountID,
+			CorrelationRef: req.CorrelationRef,
+			Disposition:    admissionv1.AccountQuotaOverrideDispositionSealed,
+			Revision:       seal.Revision,
+		}, nil
+	}
+}

spinifex/runtime/roles/awsgw/quota_override_seal_test.go

@@ -0,0 +1,84 @@
+package awsgw
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	admissionv1 "github.com/mulgadc/spinifex/contracts/admission/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	admissionquota "github.com/mulgadc/spinifex/spinifex/domains/admission/quota"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stubChecker answers AccountTerminating from a fixed set, standing in for
+// handlers_iam.AccountStatusReader so this test does not need a real IAM
+// bucket.
+type stubChecker struct {
+	terminating map[string]bool
+}
+
+func (c *stubChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) {
+	return c.terminating[accountID], nil
+}
+
+func newQuotaSealService(t *testing.T) (*admissionquota.Service, jetstream.JetStream) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	quota := admissionquota.New(admissionquota.Limits{}, nil)
+	quota.SetJetStream(js)
+	return quota, js
+}
+
+func TestQuotaOverrideSeal_SealsTerminatingAccount(t *testing.T) {
+	quota, js := newQuotaSealService(t)
+	_, nc := testutil.StartTestNATS(t)
+	quota.SetAccountTerminationChecker(&stubChecker{terminating: map[string]bool{"111122223333": true}})
+
+	sub, err := registerQuotaOverrideSeal(nc, quota)
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333", CorrelationRef: "req-1"}
+	resp, err := natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(),
+		nc, admissionv1.SealAccountQuotaOverrideSubject, req, 2*time.Second, "")
+	require.NoError(t, err)
+
+	assert.Equal(t, "111122223333", resp.AccountID)
+	assert.Equal(t, "req-1", resp.CorrelationRef)
+	assert.Equal(t, admissionv1.AccountQuotaOverrideDispositionSealed, resp.Disposition)
+	assert.NotZero(t, resp.Revision)
+
+	kv, err := js.KeyValue(context.Background(), admissionquota.KVBucketAccountQuota)
+	require.NoError(t, err)
+	entry, err := kv.Get(context.Background(), "111122223333")
+	require.NoError(t, err)
+	assert.Equal(t, resp.Revision, entry.Revision())
+}
+
+func TestQuotaOverrideSeal_RefusesActiveAccount(t *testing.T) {
+	quota, _ := newQuotaSealService(t)
+	_, nc := testutil.StartTestNATS(t)
+	quota.SetAccountTerminationChecker(&stubChecker{terminating: map[string]bool{}})
+
+	sub, err := registerQuotaOverrideSeal(nc, quota)
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333"}
+	_, err = natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(),
+		nc, admissionv1.SealAccountQuotaOverrideSubject, req, 2*time.Second, "")
+	require.Error(t, err, "a non-TERMINATING account must be refused")
+}
+
+func TestQuotaOverrideSeal_NoResponderWithoutGateway(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+
+	req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333"}
+	_, err := natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(),
+		nc, admissionv1.SealAccountQuotaOverrideSubject, req, 500*time.Millisecond, "")
+	require.Error(t, err, "no subscriber must surface as a no-responders error")
+}