diff --git a/adapter/sqs.go b/adapter/sqs.go
index 79d8c1cbe..32300f063 100644
--- a/adapter/sqs.go
+++ b/adapter/sqs.go
@@ -3,6 +3,7 @@ package adapter
 import (
 	"context"
 	"io"
+	"log/slog"
 	"net"
 	"net/http"
 	"strconv"
@@ -214,6 +215,16 @@ type SQSServer struct {
 	// nil on non-monitored fixtures; observeThrottleDecision is
 	// nil-safe so the request path pays one branch when unwired.
 	throttleObserver SQSThrottleObserver
+	// adminObserver records the §3.6 admin purge / peek counters.
+	// nil on non-monitored fixtures; the increment helpers are
+	// nil-safe so an unwired server pays one branch.
+	adminObserver SQSAdminObserver
+	// adminAuditLogger is where the §3.6 admin audit lines go. Defaults
+	// to slog.Default(); production installs the component="admin" child
+	// logger so these records land in the same destination, with the same
+	// attributes, as every other admin audit entry rather than bypassing
+	// it through the process-wide default.
+	adminAuditLogger *slog.Logger
 }
 
 // SQSPartitionObserver is the metrics-package interface
@@ -224,6 +235,14 @@ type SQSPartitionObserver interface {
 	ObservePartitionMessage(queue string, partition uint32, action string)
 }
 
+// SQSAdminObserver is the metrics-package interface
+// (monitoring.SQSMetrics) re-declared here so the adapter does not
+// import monitoring at the package boundary.
+type SQSAdminObserver interface {
+	ObserveAdminPurgeQueue(queue string, outcome string)
+	ObserveAdminPeekQueue(queue string, outcome string)
+}
+
 // SQSThrottleObserver is the metrics-package interface
 // (monitoring.SQSThrottleObserver) re-declared here so the adapter
 // does not import monitoring at the package boundary.
@@ -278,6 +297,25 @@ func WithSQSLeaderMap(m map[string]string) SQSServerOption {
 	}
 }
 
+// WithSQSAdminObserver installs the §3.6 admin purge / peek counters.
+func WithSQSAdminObserver(o SQSAdminObserver) SQSServerOption {
+	return func(s *SQSServer) {
+		if o != nil {
+			s.adminObserver = o
+		}
+	}
+}
+
+// WithSQSAdminAuditLogger routes the §3.6 admin audit lines to the
+// supplied logger instead of slog.Default(). No-ops on nil.
+func WithSQSAdminAuditLogger(l *slog.Logger) SQSServerOption {
+	return func(s *SQSServer) {
+		if l != nil {
+			s.adminAuditLogger = l
+		}
+	}
+}
+
 // WithSQSPartitionObserver installs the
 // elastickv_sqs_partition_messages_total counter observer on the
 // SQS server. Pass nil (the default) on non-monitored test
diff --git a/adapter/sqs_admin.go b/adapter/sqs_admin.go
index e3b205dd6..c248b62ef 100644
--- a/adapter/sqs_admin.go
+++ b/adapter/sqs_admin.go
@@ -3,6 +3,7 @@ package adapter
 import (
 	"bytes"
 	"context"
+	"log/slog"
 	"sort"
 	"strconv"
 	"strings"
@@ -316,16 +317,26 @@ func (e *PurgeInProgressError) Is(target error) bool {
 //   - ErrAdminSQSValidation      — empty / whitespace name
 func (s *SQSServer) AdminPurgeQueue(ctx context.Context, principal AdminPrincipal, name string) (AdminPurgeResult, error) {
 	if !principal.Role.canWrite() {
+		s.recordAdminPurge(ctx, principal, name, adminOutcomeForbidden)
 		return AdminPurgeResult{}, ErrAdminForbidden
 	}
 	if !isVerifiedSQSLeader(ctx, s.coordinator) {
+		s.recordAdminPurge(ctx, principal, name, adminOutcomeNotLeader)
 		return AdminPurgeResult{}, ErrAdminNotLeader
 	}
 	if strings.TrimSpace(name) == "" {
+		s.recordAdminPurge(ctx, principal, name, adminOutcomeValidation)
 		return AdminPurgeResult{}, ErrAdminSQSValidation
 	}
 	oldGen, newGen, err := s.purgeQueueWithRetry(ctx, name)
 	if err != nil {
+		// Every refusal gets the operation-specific audit record, not
+		// just the counter. A repeated purge inside the 60-second window
+		// used to return from here with no admin.sqs.purge_queue line at
+		// all, leaving only the generic HTTP audit middleware's status
+		// and path -- which cannot say WHY it was refused, and so cannot
+		// answer the question the §3.6 signal exists for.
+		s.recordAdminPurge(ctx, principal, name, adminPurgeOutcomeForError(err))
 		var rateLimit *purgeRateLimitedError
 		if errors.As(err, &rateLimit) {
 			return AdminPurgeResult{}, &PurgeInProgressError{RetryAfter: rateLimit.remaining}
@@ -335,9 +346,109 @@ func (s *SQSServer) AdminPurgeQueue(ctx context.Context, principal AdminPrincipa
 		}
 		return AdminPurgeResult{}, errors.Wrap(err, "admin purge queue")
 	}
+	s.recordAdminPurge(ctx, principal, name, adminOutcomeOK,
+		slog.Uint64("generation_before", oldGen),
+		slog.Uint64("generation_after", newGen))
 	return AdminPurgeResult{GenerationBefore: oldGen, GenerationAfter: newGen}, nil
 }
 
+// recordAdminPurge emits the §3.6 audit line and bumps the counter for one
+// purge outcome.
+//
+// Both in one place so an exit path cannot record the metric and skip the
+// audit, which is how the purge-in-progress refusal ended up counted but
+// never audited. Generation attributes are passed only by the success path:
+// a refusal has no committed generation pair, and inventing one would put a
+// state that never existed into the audit trail.
+func (s *SQSServer) recordAdminPurge(
+	ctx context.Context,
+	principal AdminPrincipal,
+	name string,
+	outcome string,
+	extra ...slog.Attr,
+) {
+	s.observeAdminPurge(name, outcome)
+	// access_key, role, queue, outcome.
+	const baseAuditAttrs = 4
+	attrs := make([]any, 0, len(extra)+baseAuditAttrs)
+	attrs = append(attrs,
+		// AdminPrincipal carries AccessKey, not the design's
+		// "subject": the access key ID is the identity the admin
+		// surface authenticates, and it is an identifier rather than
+		// a secret (the signing key never appears here).
+		slog.String("access_key", principal.AccessKey),
+		slog.String("role", string(principal.Role)),
+		slog.String("queue", name),
+		slog.String("outcome", outcome))
+	for _, attr := range extra {
+		attrs = append(attrs, attr)
+	}
+	s.adminLogger().InfoContext(ctx, "admin.sqs.purge_queue", attrs...)
+}
+
+// adminLogger returns the configured audit destination, falling back to the
+// process default so a server built without the option still audits.
+func (s *SQSServer) adminLogger() *slog.Logger {
+	if s == nil || s.adminAuditLogger == nil {
+		return slog.Default()
+	}
+	return s.adminAuditLogger
+}
+
+// Outcome labels for the §3.6 admin counters, mirrored from
+// monitoring so the adapter does not import it at this boundary.
+const (
+	adminOutcomeOK              = "ok"
+	adminOutcomeForbidden       = "forbidden"
+	adminOutcomeNotLeader       = "not_leader"
+	adminOutcomeNotFound        = "not_found"
+	adminOutcomeValidation      = "validation"
+	adminOutcomePurgeInProgress = "purge_in_progress"
+	// NOTE: no "throttled" here — the admin peek throttle is a
+	// separate deferred follow-up, and declaring the label before a
+	// call site exists would imply coverage the code does not have.
+	adminOutcomeInternalError = "internal_error"
+)
+
+// adminPurgeOutcomeForError classifies a purge failure by SENTINEL,
+// never by message text: an error-string label would let one recurring
+// failure grow the series set without bound.
+func adminPurgeOutcomeForError(err error) string {
+	var rateLimit *purgeRateLimitedError
+	switch {
+	case errors.As(err, &rateLimit):
+		return adminOutcomePurgeInProgress
+	case isSQSAdminQueueDoesNotExist(err):
+		return adminOutcomeNotFound
+	default:
+		return adminOutcomeInternalError
+	}
+}
+
+// AdminObserver exposes the §3.6 counters so the admin HTTP handler can
+// record the rejections it serves before reaching this adapter. Returns nil
+// on a server built without an observer.
+func (s *SQSServer) AdminObserver() SQSAdminObserver {
+	if s == nil {
+		return nil
+	}
+	return s.adminObserver
+}
+
+func (s *SQSServer) observeAdminPurge(queue, outcome string) {
+	if s == nil || s.adminObserver == nil {
+		return
+	}
+	s.adminObserver.ObserveAdminPurgeQueue(queue, outcome)
+}
+
+func (s *SQSServer) observeAdminPeek(queue, outcome string) {
+	if s == nil || s.adminObserver == nil {
+		return
+	}
+	s.adminObserver.ObserveAdminPeekQueue(queue, outcome)
+}
+
 // AdminSetQueueAttributes is the SigV4-bypass counterpart to
 // SetQueueAttributes. It is intentionally generic rather than
 // DLQ-specific so the admin SPA can edit RedrivePolicy and
diff --git a/adapter/sqs_admin_audit_test.go b/adapter/sqs_admin_audit_test.go
new file mode 100644
index 000000000..7ae83a9a7
--- /dev/null
+++ b/adapter/sqs_admin_audit_test.go
@@ -0,0 +1,133 @@
+package adapter
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"log/slog"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// captureAdminAudit points the server's audit logger at a buffer and returns a
+// reader for the admin.sqs.purge_queue records it emits.
+func captureAdminAudit(t *testing.T, server *SQSServer) func() []map[string]any {
+	t.Helper()
+
+	var buf bytes.Buffer
+	// A distinguishing attribute, so a record that went out through
+	// slog.Default() instead of this logger is detectable rather than just
+	// absent.
+	server.adminAuditLogger = slog.New(slog.NewJSONHandler(&buf, nil)).
+		With(slog.String("component", "admin"))
+
+	return func() []map[string]any {
+		var records []map[string]any
+		for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") {
+			if strings.TrimSpace(line) == "" {
+				continue
+			}
+			var rec map[string]any
+			require.NoError(t, json.Unmarshal([]byte(line), &rec))
+			if rec["msg"] == "admin.sqs.purge_queue" {
+				records = append(records, rec)
+			}
+		}
+		return records
+	}
+}
+
+// TestAdminPurgeQueueAuditsThroughTheConfiguredLogger pins that the §3.6 audit
+// record goes to the configured audit destination.
+//
+// It used to call slog.InfoContext, i.e. the process-wide slog.Default(), so a
+// server built with a dedicated audit sink or the production component="admin"
+// child logger never saw these records and they lost that logger's attributes
+// — unlike every other admin audit entry.
+func TestAdminPurgeQueueAuditsThroughTheConfiguredLogger(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	_ = createSQSQueueForTest(t, node, "audited")
+	records := captureAdminAudit(t, node.sqsServer)
+
+	_, err := node.sqsServer.AdminPurgeQueue(context.Background(), fullAdminPrincipal, "audited")
+	require.NoError(t, err)
+
+	got := records()
+	require.Len(t, got, 1, "exactly one audit record for one purge")
+	require.Equal(t, "admin", got[0]["component"],
+		"the record must carry the configured logger's attributes")
+	require.Equal(t, "audited", got[0]["queue"])
+	require.Equal(t, adminOutcomeOK, got[0]["outcome"])
+	require.Contains(t, got[0], "generation_before")
+	require.Contains(t, got[0], "generation_after")
+}
+
+// TestAdminPurgeQueueAuditsAPurgeInProgressRefusal is the load-bearing case.
+//
+// A second purge inside the 60-second window returned before the only audit
+// call, so the documented signal for repeated rate-limited attempts produced
+// nothing. The generic HTTP audit middleware records status and path, which
+// cannot say WHY the request was refused — the one thing this record exists to
+// answer.
+func TestAdminPurgeQueueAuditsAPurgeInProgressRefusal(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	_ = createSQSQueueForTest(t, node, "repeat-purged")
+	records := captureAdminAudit(t, node.sqsServer)
+	ctx := context.Background()
+
+	_, err := node.sqsServer.AdminPurgeQueue(ctx, fullAdminPrincipal, "repeat-purged")
+	require.NoError(t, err)
+	_, err = node.sqsServer.AdminPurgeQueue(ctx, fullAdminPrincipal, "repeat-purged")
+	require.ErrorIs(t, err, ErrAdminSQSPurgeInProgress)
+
+	got := records()
+	require.Len(t, got, 2, "the refusal must be audited, not only the success")
+	require.Equal(t, adminOutcomeOK, got[0]["outcome"])
+	require.Equal(t, adminOutcomePurgeInProgress, got[1]["outcome"])
+	require.Equal(t, "repeat-purged", got[1]["queue"])
+
+	// No invented generation pair: the refusal committed nothing, and
+	// recording a state that never existed would corrupt the audit trail.
+	require.NotContains(t, got[1], "generation_before")
+	require.NotContains(t, got[1], "generation_after")
+}
+
+// A forbidden principal is refused before any storage work, and must still be
+// audited with its own outcome.
+func TestAdminPurgeQueueAuditsAForbiddenRefusal(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	_ = createSQSQueueForTest(t, node, "guarded")
+	records := captureAdminAudit(t, node.sqsServer)
+
+	_, err := node.sqsServer.AdminPurgeQueue(context.Background(), readOnlyAdminPrincipal, "guarded")
+	require.ErrorIs(t, err, ErrAdminForbidden)
+
+	got := records()
+	require.Len(t, got, 1)
+	require.Equal(t, adminOutcomeForbidden, got[0]["outcome"])
+	require.NotContains(t, got[0], "generation_before")
+}
+
+// A server built without the option must still audit, through the default.
+func TestAdminPurgeQueueFallsBackToTheDefaultLogger(t *testing.T) {
+	t.Parallel()
+
+	var server *SQSServer
+	require.NotNil(t, server.adminLogger(), "a nil server must not panic")
+	require.NotNil(t, (&SQSServer{}).adminLogger(),
+		"a server with no configured audit logger must still have somewhere to audit")
+}
diff --git a/adapter/sqs_admin_peek.go b/adapter/sqs_admin_peek.go
index 047785aa6..5eee39790 100644
--- a/adapter/sqs_admin_peek.go
+++ b/adapter/sqs_admin_peek.go
@@ -215,43 +215,69 @@ func (s *SQSServer) AdminPeekQueue(
 	opts AdminPeekMessageOptions,
 ) ([]AdminPeekedMessage, string, error) {
 	if !principal.Role.canRead() {
+		s.observeAdminPeek(name, adminOutcomeForbidden)
 		return nil, "", ErrAdminForbidden
 	}
 	if !isVerifiedSQSLeader(ctx, s.coordinator) {
+		s.observeAdminPeek(name, adminOutcomeNotLeader)
 		return nil, "", ErrAdminNotLeader
 	}
 	if strings.TrimSpace(name) == "" {
+		s.observeAdminPeek(name, adminOutcomeValidation)
 		return nil, "", ErrAdminSQSValidation
 	}
 	limit := clampPeekLimit(opts.Limit)
 	bodyMaxBytes := clampPeekBodyBytes(opts.BodyMaxBytes)
 	cursor, err := decodePeekCursor(opts.Cursor)
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", err
 	}
 	readTS := s.nextTxnReadTS(ctx)
 	meta, exists, err := s.loadQueueMetaAt(ctx, name, readTS)
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", errors.WithStack(err)
 	}
 	if !exists {
+		s.observeAdminPeek(name, adminOutcomeNotFound)
 		return nil, "", ErrAdminSQSNotFound
 	}
 	cursor, err = preparePeekCursor(cursor, meta, name, s.peekStartPartition(name, cursor, meta))
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", err
 	}
 	rows, nextCursor, err := s.walkPeek(ctx, name, meta, readTS, cursor, limit, bodyMaxBytes)
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", err
 	}
 	encoded, err := encodePeekCursor(nextCursor)
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", err
 	}
+	s.observeAdminPeek(name, adminOutcomeOK)
 	return rows, encoded, nil
 }
 
+// adminPeekOutcomeForError classifies a peek failure by SENTINEL,
+// never by message text — an error-string label would let one
+// recurring failure grow the series set without bound.
+func adminPeekOutcomeForError(err error) string {
+	switch {
+	case errors.Is(err, ErrAdminSQSValidation):
+		return adminOutcomeValidation
+	case errors.Is(err, ErrAdminSQSNotFound):
+		return adminOutcomeNotFound
+	case isSQSAdminQueueDoesNotExist(err):
+		return adminOutcomeNotFound
+	default:
+		return adminOutcomeInternalError
+	}
+}
+
 // peekStartPartition returns the starting partition for a fresh peek
 // walk, or 0 when a starting partition is not needed (continuation
 // pages already carry one in the cursor; non-partitioned queues only
diff --git a/docs/design/2026_05_16_implemented_admin_purge_queue.md b/docs/design/2026_05_16_implemented_admin_purge_queue.md
index b5b05299e..d0ff9a10e 100644
--- a/docs/design/2026_05_16_implemented_admin_purge_queue.md
+++ b/docs/design/2026_05_16_implemented_admin_purge_queue.md
@@ -16,7 +16,7 @@
 
 Out-of-scope follow-ups (tracked separately, not gating this rename):
 - Throttle integration (`bucketActionAdminPeek` + dedicated per-queue admin-peek bucket per §3.1)
-- Audit logging + Prometheus counters per §3.6
+- ~~Audit logging + Prometheus counters per §3.6~~ — **implemented**
 - `principalForReadSensitive` live `RoleStore` re-check (Goal 8, blocked on wider RoleStore plumbing)
 - Page-size selector (20 / 50 / 100) + response-size warning
 
@@ -491,7 +491,11 @@ mirroring the existing `deleteQueue` / `describeQueue` shape. `peekQueue` is `si
 
 ### 3.6 Audit and observability
 
-_Not yet implemented in the initial rollout — see "Out-of-scope follow-ups" at the top. Mitigation in absence: the admin handler still emits the standard request-log line with `route` / `subject` / `status_code` for both purge and peek calls, so an operator can correlate "who did what when" against the application logs at audit-review time. The structured `admin.sqs.purge_queue` audit line and the two Prometheus counters land alongside the SPA wiring so the metrics have a real consumer._
+_**Implemented.** The `admin.sqs.purge_queue` audit line and both Prometheus counters are live. Two deviations from the text below, both forced by the code as it stands:_
+
+_1. The audit line logs `access_key`, not `subject`: `AdminPrincipal` carries `AccessKey` and `Role` and has no `Subject` field. The access key ID is the identity the admin surface authenticates and is an identifier rather than a secret — the signing key never appears in the log._
+
+_2. The two outcome sets are deliberately asymmetric. `purge_in_progress` exists only on the purge counter and `throttled` only on peek, because purge signals contention via the generation gate and peek via the throttle. Accepting both on either counter would let the two paths drift into describing one condition two ways. The peek `throttled` outcome is defined but not yet emitted — admin-peek throttle integration remains a separate open follow-up._
 
 New structured log line at `slog.Info` level (matches `AdminDeleteQueue`):
 
diff --git a/internal/admin/server.go b/internal/admin/server.go
index 00f672130..e4abb3764 100644
--- a/internal/admin/server.go
+++ b/internal/admin/server.go
@@ -245,7 +245,13 @@ func buildSqsHandlerForDeps(deps ServerDeps, logger *slog.Logger) http.Handler {
 	}
 	return NewSqsHandler(deps.Queues).
 		WithLogger(logger).
-		WithRoleStore(MapRoleStore(deps.Roles))
+		WithRoleStore(MapRoleStore(deps.Roles)).
+		// Taken from the source rather than wired separately, so the
+		// handler's pre-dispatch rejections land on the SAME counters the
+		// adapter records through. Separate wiring could silently diverge,
+		// and a metric assembled from two halves that disagree is worse
+		// than one with a known gap.
+		WithAdminQueueObserver(adminQueueObserverFrom(deps.Queues))
 }
 
 // Handler returns an http.Handler that serves the full admin surface.
diff --git a/internal/admin/sqs_handler.go b/internal/admin/sqs_handler.go
index 5102a1e31..ea8696a04 100644
--- a/internal/admin/sqs_handler.go
+++ b/internal/admin/sqs_handler.go
@@ -197,6 +197,40 @@ type SqsHandler struct {
 	source QueuesSource
 	roles  RoleStore
 	logger *slog.Logger
+	// admin counts the §3.6 purge / peek outcomes that never reach the
+	// adapter. Authorization, path and query-parameter rejections all
+	// return from the handler before AdminPurgeQueue / AdminPeekQueue
+	// runs, so without this the advertised forbidden and validation
+	// outcomes were absent from the metric for the most common cases --
+	// under-reporting exactly the rejections an operator is looking for.
+	admin AdminQueueObserver
+}
+
+// AdminQueueObserver counts admin queue-operation outcomes.
+//
+// Declared here so the handler can record its own pre-dispatch exits
+// without importing the adapter; *monitoring.SQSMetrics satisfies it
+// structurally, so production passes the same counters the adapter uses and
+// the two halves of each metric cannot drift apart.
+type AdminQueueObserver interface {
+	ObserveAdminPurgeQueue(queue string, outcome string)
+	ObserveAdminPeekQueue(queue string, outcome string)
+}
+
+// AdminQueueObserverSource is the optional capability a QueuesSource can
+// implement to hand the handler the counters it already records through.
+type AdminQueueObserverSource interface {
+	AdminQueueObserver() AdminQueueObserver
+}
+
+// adminQueueObserverFrom returns the source's counters, or nil when the
+// source does not expose any.
+func adminQueueObserverFrom(source QueuesSource) AdminQueueObserver {
+	provider, ok := source.(AdminQueueObserverSource)
+	if !ok {
+		return nil
+	}
+	return provider.AdminQueueObserver()
 }
 
 // NewSqsHandler binds the source and seeds logging with
@@ -217,6 +251,33 @@ func (h *SqsHandler) WithLogger(l *slog.Logger) *SqsHandler {
 	return h
 }
 
+// WithAdminQueueObserver installs the §3.6 counters so the handler can
+// record the rejections it serves itself. No-ops on nil; the handler then
+// records nothing and the adapter-side outcomes are unaffected.
+func (h *SqsHandler) WithAdminQueueObserver(o AdminQueueObserver) *SqsHandler {
+	if o == nil {
+		return h
+	}
+	h.admin = o
+	return h
+}
+
+// observePurgeRejection records a purge outcome decided in the handler.
+func (h *SqsHandler) observePurgeRejection(name, outcome string) {
+	if h.admin == nil {
+		return
+	}
+	h.admin.ObserveAdminPurgeQueue(name, outcome)
+}
+
+// observePeekRejection records a peek outcome decided in the handler.
+func (h *SqsHandler) observePeekRejection(name, outcome string) {
+	if h.admin == nil {
+		return
+	}
+	h.admin.ObserveAdminPeekQueue(name, outcome)
+}
+
 // WithRoleStore enables per-request role revalidation on the delete
 // endpoint. Without it, the handler trusts whatever role is embedded
 // in the session JWT — which is fine for single-tenant deployments
@@ -395,6 +456,14 @@ func (h *SqsHandler) dispatchAttributesResource(w http.ResponseWriter, r *http.R
 	}
 }
 
+// The handler-side half of the §3.6 outcome vocabulary. Kept to the two
+// values the handler can actually decide, and spelled identically to the
+// adapter's so a dashboard does not see two spellings of one outcome.
+const (
+	adminQueueOutcomeForbidden  = "forbidden"
+	adminQueueOutcomeValidation = "validation"
+)
+
 // isValidSqsPathSegment enforces the step-4 rules. Every segment is
 // rejected if it is empty, contains a percent sign (closes the
 // %2F / %252F / %2e / %2E / %2E%2E percent-encoded slash and
@@ -535,10 +604,12 @@ func (h *SqsHandler) handleSetAttributes(w http.ResponseWriter, r *http.Request,
 func (h *SqsHandler) handlePeek(w http.ResponseWriter, r *http.Request, name string) {
 	principal, ok := h.principalForReadSensitive(w, r)
 	if !ok {
+		h.observePeekRejection(name, adminQueueOutcomeForbidden)
 		return
 	}
 	opts, ok := parsePeekQueryParams(w, r)
 	if !ok {
+		h.observePeekRejection(name, adminQueueOutcomeValidation)
 		return
 	}
 	result, err := h.source.AdminPeekQueue(r.Context(), principal, name, opts)
@@ -594,9 +665,11 @@ func parsePeekQueryParams(w http.ResponseWriter, r *http.Request) (PeekMessageOp
 func (h *SqsHandler) handlePurge(w http.ResponseWriter, r *http.Request, name string) {
 	principal, ok := h.principalForWriteOnPurge(w, r)
 	if !ok {
+		h.observePurgeRejection(name, adminQueueOutcomeForbidden)
 		return
 	}
 	if strings.TrimSpace(name) == "" {
+		h.observePurgeRejection(name, adminQueueOutcomeValidation)
 		writeJSONError(w, http.StatusBadRequest, "invalid_queue_name", "queue name is required")
 		return
 	}
diff --git a/internal/admin/sqs_handler_metrics_test.go b/internal/admin/sqs_handler_metrics_test.go
new file mode 100644
index 000000000..cac156959
--- /dev/null
+++ b/internal/admin/sqs_handler_metrics_test.go
@@ -0,0 +1,145 @@
+package admin
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+type recordedQueueOutcome struct {
+	queue   string
+	outcome string
+}
+
+type recordingQueueObserver struct {
+	purges []recordedQueueOutcome
+	peeks  []recordedQueueOutcome
+}
+
+func (o *recordingQueueObserver) ObserveAdminPurgeQueue(queue, outcome string) {
+	o.purges = append(o.purges, recordedQueueOutcome{queue: queue, outcome: outcome})
+}
+
+func (o *recordingQueueObserver) ObserveAdminPeekQueue(queue, outcome string) {
+	o.peeks = append(o.peeks, recordedQueueOutcome{queue: queue, outcome: outcome})
+}
+
+// queuesSourceWithObserver is a QueuesSource that also exposes counters, the
+// shape production uses via the main_admin.go bridge.
+type queuesSourceWithObserver struct {
+	*stubQueuesSource
+	observer AdminQueueObserver
+}
+
+func (s *queuesSourceWithObserver) AdminQueueObserver() AdminQueueObserver {
+	return s.observer
+}
+
+// TestSqsHandlerCountsRejectionsItServesItself is the regression test for
+// under-reported §3.6 outcomes.
+//
+// handlePurge returns at principalForWriteOnPurge and at the empty-name check
+// BEFORE AdminPurgeQueue runs, and handlePeek returns at
+// principalForReadSensitive and at parsePeekQueryParams before AdminPeekQueue
+// runs. The adapter-side counters therefore never saw the most common
+// forbidden and validation rejections, so the metric under-reported exactly
+// the requests an operator goes looking for.
+func TestSqsHandlerCountsRejectionsItServesItself(t *testing.T) {
+	t.Parallel()
+
+	t.Run("purge forbidden by the live role", func(t *testing.T) {
+		t.Parallel()
+		observer := &recordingQueueObserver{}
+		src := &queuesSourceWithObserver{
+			stubQueuesSource: &stubQueuesSource{queues: []string{"orders"}},
+			observer:         observer,
+		}
+		h := NewSqsHandler(src).
+			WithRoleStore(MapRoleStore{"AKIA_RO": RoleReadOnly}).
+			WithAdminQueueObserver(adminQueueObserverFrom(src))
+
+		req := httptest.NewRequest(http.MethodDelete, pathPrefixSqsQueues+"orders/messages", nil)
+		req = req.WithContext(context.WithValue(req.Context(), ctxKeyPrincipal,
+			AuthPrincipal{AccessKey: "AKIA_RO", Role: RoleFull}))
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, req)
+
+		require.Equal(t, http.StatusForbidden, rec.Code, "body=%s", rec.Body.String())
+		require.Equal(t,
+			[]recordedQueueOutcome{{queue: "orders", outcome: adminQueueOutcomeForbidden}},
+			observer.purges,
+			"a rejection the handler serves must still be counted")
+	})
+
+	t.Run("peek forbidden by the live role", func(t *testing.T) {
+		t.Parallel()
+		observer := &recordingQueueObserver{}
+		src := &queuesSourceWithObserver{
+			stubQueuesSource: &stubQueuesSource{queues: []string{"orders"}},
+			observer:         observer,
+		}
+		h := NewSqsHandler(src).
+			WithRoleStore(MapRoleStore{"AKIA_NONE": Role("")}).
+			WithAdminQueueObserver(adminQueueObserverFrom(src))
+
+		req := httptest.NewRequest(http.MethodGet, pathPrefixSqsQueues+"orders/messages", nil)
+		req = req.WithContext(context.WithValue(req.Context(), ctxKeyPrincipal,
+			AuthPrincipal{AccessKey: "AKIA_NONE", Role: RoleFull}))
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, req)
+
+		require.Equal(t, http.StatusForbidden, rec.Code, "body=%s", rec.Body.String())
+		require.Equal(t,
+			[]recordedQueueOutcome{{queue: "orders", outcome: adminQueueOutcomeForbidden}},
+			observer.peeks)
+	})
+
+	t.Run("peek rejected for a non-numeric limit", func(t *testing.T) {
+		t.Parallel()
+		observer := &recordingQueueObserver{}
+		src := &queuesSourceWithObserver{
+			stubQueuesSource: &stubQueuesSource{queues: []string{"orders"}},
+			observer:         observer,
+		}
+		h := NewSqsHandler(src).
+			WithRoleStore(MapRoleStore{"AKIA_FULL": RoleFull}).
+			WithAdminQueueObserver(adminQueueObserverFrom(src))
+
+		req := httptest.NewRequest(http.MethodGet,
+			pathPrefixSqsQueues+"orders/messages?limit=abc", nil)
+		req = req.WithContext(context.WithValue(req.Context(), ctxKeyPrincipal,
+			AuthPrincipal{AccessKey: "AKIA_FULL", Role: RoleFull}))
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, req)
+
+		require.Equal(t, http.StatusBadRequest, rec.Code, "body=%s", rec.Body.String())
+		require.Equal(t,
+			[]recordedQueueOutcome{{queue: "orders", outcome: adminQueueOutcomeValidation}},
+			observer.peeks)
+	})
+}
+
+// A source that exposes no counters must leave the handler working, just
+// uncounted — the pre-existing behaviour for fixtures without metrics.
+func TestSqsHandlerWithoutAnObserverStillServes(t *testing.T) {
+	t.Parallel()
+
+	src := &stubQueuesSource{queues: []string{"orders"}}
+	require.Nil(t, adminQueueObserverFrom(src),
+		"a source that does not expose counters must yield none")
+
+	h := NewSqsHandler(src).
+		WithRoleStore(MapRoleStore{"AKIA_RO": RoleReadOnly}).
+		WithAdminQueueObserver(adminQueueObserverFrom(src))
+
+	req := httptest.NewRequest(http.MethodDelete, pathPrefixSqsQueues+"orders/messages", nil)
+	req = req.WithContext(context.WithValue(req.Context(), ctxKeyPrincipal,
+		AuthPrincipal{AccessKey: "AKIA_RO", Role: RoleFull}))
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+
+	require.Equal(t, http.StatusForbidden, rec.Code)
+}
diff --git a/main_admin.go b/main_admin.go
index 7c92ddcc7..3e2c64705 100644
--- a/main_admin.go
+++ b/main_admin.go
@@ -235,6 +235,21 @@ type sqsQueuesBridge struct {
 	server *adapter.SQSServer
 }
 
+// AdminQueueObserver satisfies admin.AdminQueueObserverSource so the admin
+// handler's pre-dispatch rejections are counted on the same metrics the
+// adapter uses. Returns nil when the server has no observer, which
+// WithAdminQueueObserver treats as "do not count".
+func (b *sqsQueuesBridge) AdminQueueObserver() admin.AdminQueueObserver {
+	if b == nil || b.server == nil {
+		return nil
+	}
+	observer := b.server.AdminObserver()
+	if observer == nil {
+		return nil
+	}
+	return observer
+}
+
 func (b *sqsQueuesBridge) AdminListQueues(ctx context.Context) ([]string, error) {
 	return b.server.AdminListQueues(ctx) //nolint:wrapcheck // pure pass-through; adapter owns the error context.
 }
diff --git a/main_sqs.go b/main_sqs.go
index 03ab5adf0..00b9ce83a 100644
--- a/main_sqs.go
+++ b/main_sqs.go
@@ -43,6 +43,13 @@ func prepareSQSServer(
 	if o, ok := partitionObserver.(adapter.SQSThrottleObserver); ok {
 		throttleObserver = o
 	}
+	// The registry's SQSMetrics satisfies all three observer
+	// interfaces; derive the admin one the same way rather than
+	// widening this function's signature.
+	var adminObserver adapter.SQSAdminObserver
+	if o, ok := partitionObserver.(adapter.SQSAdminObserver); ok {
+		adminObserver = o
+	}
 	sqsServer := adapter.NewSQSServer(
 		sqsL,
 		shardStore,
@@ -53,6 +60,12 @@ func prepareSQSServer(
 		adapter.WithSQSPartitionResolver(partitionResolver),
 		adapter.WithSQSPartitionObserver(partitionObserver),
 		adapter.WithSQSThrottleObserver(throttleObserver),
+		adapter.WithSQSAdminObserver(adminObserver),
+		// Same component="admin" child logger the admin HTTP server uses,
+		// so the §3.6 purge audit lines land in the configured audit
+		// destination with its attributes instead of going out through the
+		// process-wide slog.Default().
+		adapter.WithSQSAdminAuditLogger(slog.Default().With(slog.String("component", "admin"))),
 	)
 	return sqsServer, sqsL, nil
 }
diff --git a/main_sqs_admin_observer_test.go b/main_sqs_admin_observer_test.go
new file mode 100644
index 000000000..473371c40
--- /dev/null
+++ b/main_sqs_admin_observer_test.go
@@ -0,0 +1,72 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/bootjp/elastickv/adapter"
+	"github.com/bootjp/elastickv/internal/admin"
+	"github.com/bootjp/elastickv/monitoring"
+	"github.com/stretchr/testify/require"
+)
+
+// TestSqsQueuesBridgeExposesTheAdminObserver checks the PRODUCTION wrapper, not
+// just the interface.
+//
+// The admin handler counts its own pre-dispatch rejections by pulling the
+// observer off its QueuesSource. In production that source is
+// *sqsQueuesBridge, not *adapter.SQSServer — so a handler-side test using a
+// stub can pass while production records nothing, because the bridge does not
+// satisfy the capability interface.
+func TestSqsQueuesBridgeExposesTheAdminObserver(t *testing.T) {
+	t.Parallel()
+
+	var source admin.QueuesSource = &sqsQueuesBridge{}
+	provider, ok := source.(admin.AdminQueueObserverSource)
+	require.True(t, ok,
+		"the production bridge must satisfy admin.AdminQueueObserverSource, "+
+			"or the handler's pre-dispatch outcomes are never counted in production")
+
+	// With no server: a genuine nil, not a typed-nil interface, which
+	// WithAdminQueueObserver would otherwise accept and then call into.
+	require.Nil(t, provider.AdminQueueObserver())
+}
+
+// And the observer handed over must be the one the adapter records through, or
+// the two halves of each counter describe different things.
+func TestSqsQueuesBridgeHandsOverTheServersObserver(t *testing.T) {
+	t.Parallel()
+
+	// The same derivation main_sqs.go performs: the registry's SQSMetrics
+	// serves every SQS observer interface.
+	registry := monitoring.NewRegistry("n1", "127.0.0.1:50051")
+	partitionObserver := registry.SQSPartitionObserver()
+	adminObserver, ok := partitionObserver.(adapter.SQSAdminObserver)
+	require.True(t, ok, "the registry's SQSMetrics must serve the adapter's admin observer")
+
+	server := adapter.NewSQSServer(nil, nil, nil, adapter.WithSQSAdminObserver(adminObserver))
+
+	// Through the interface, not the concrete method: if the bridge stops
+	// satisfying the capability this fails as a test rather than as a compile
+	// error, which is what the handler's lookup actually does at runtime.
+	var source admin.QueuesSource = &sqsQueuesBridge{server: server}
+	provider, ok := source.(admin.AdminQueueObserverSource)
+	require.True(t, ok, "the production bridge must satisfy admin.AdminQueueObserverSource")
+
+	got := provider.AdminQueueObserver()
+	require.NotNil(t, got)
+	require.Equal(t, adminObserver, got,
+		"the handler must count on the same metrics object the adapter uses")
+}
+
+// One object serves both sides, so it has to satisfy both interfaces.
+func TestSQSMetricsSatisfiesBothObserverInterfaces(t *testing.T) {
+	t.Parallel()
+
+	registry := monitoring.NewRegistry("n1", "127.0.0.1:50051")
+	partitionObserver := registry.SQSPartitionObserver()
+
+	_, adapterOK := partitionObserver.(adapter.SQSAdminObserver)
+	require.True(t, adapterOK, "adapter.SQSAdminObserver")
+	_, adminOK := partitionObserver.(admin.AdminQueueObserver)
+	require.True(t, adminOK, "admin.AdminQueueObserver")
+}
diff --git a/monitoring/sqs.go b/monitoring/sqs.go
index 353b7405f..5d6045ac1 100644
--- a/monitoring/sqs.go
+++ b/monitoring/sqs.go
@@ -149,6 +149,13 @@ type SQSMetrics struct {
 	queueDepth        *prometheus.GaugeVec
 	throttledRequests *prometheus.CounterVec
 	throttleTokens    *prometheus.GaugeVec
+	adminPurgeQueue   *prometheus.CounterVec
+	adminPeekQueue    *prometheus.CounterVec
+	// trackedAdminCounterQueues bounds the queue label on the two
+	// admin counters with the same budget the data-path counters use.
+	// Queue names are operator-supplied, so an unbounded label here
+	// would let queue churn grow the series set without limit.
+	trackedAdminCounterQueues map[string]struct{}
 
 	mu                           sync.Mutex
 	trackedCounterQueues         map[string]struct{}
@@ -215,6 +222,7 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics {
 			[]string{"queue", "action"},
 		),
 		trackedCounterQueues:         map[string]struct{}{},
+		trackedAdminCounterQueues:    map[string]struct{}{},
 		trackedThrottleCounterQueues: map[string]struct{}{},
 		trackedDepthQueues:           map[string]struct{}{},
 		trackedThrottleGaugeQueues:   map[string]map[string]struct{}{},
@@ -223,13 +231,113 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics {
 		overflowDepthQueues:          map[string]struct{}{},
 		overflowThrottleGaugeQueues:  map[string]map[string]struct{}{},
 	}
+	m.adminPurgeQueue = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "elastickv_sqs_admin_purge_queue_total",
+			Help: "Total admin PurgeQueue calls by queue and outcome.",
+		},
+		[]string{"queue", "outcome"},
+	)
+	m.adminPeekQueue = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "elastickv_sqs_admin_peek_queue_total",
+			Help: "Total admin PeekQueue calls by queue and outcome.",
+		},
+		[]string{"queue", "outcome"},
+	)
 	registerer.MustRegister(m.partitionMessages)
 	registerer.MustRegister(m.queueDepth)
 	registerer.MustRegister(m.throttledRequests)
 	registerer.MustRegister(m.throttleTokens)
+	registerer.MustRegister(m.adminPurgeQueue)
+	registerer.MustRegister(m.adminPeekQueue)
 	return m
 }
 
+// Outcomes for the two admin SQS counters. Closed sets: the label is
+// derived from a sentinel comparison, never from an error string, so
+// one recurring failure cannot mint new series.
+const (
+	SQSAdminOutcomeOK              = "ok"
+	SQSAdminOutcomeForbidden       = "forbidden"
+	SQSAdminOutcomeNotLeader       = "not_leader"
+	SQSAdminOutcomeNotFound        = "not_found"
+	SQSAdminOutcomeValidation      = "validation"
+	SQSAdminOutcomePurgeInProgress = "purge_in_progress"
+	SQSAdminOutcomeThrottled       = "throttled"
+	SQSAdminOutcomeInternalError   = "internal_error"
+)
+
+// ObserveAdminPurgeQueue counts one AdminPurgeQueue call.
+func (m *SQSMetrics) ObserveAdminPurgeQueue(queue, outcome string) {
+	if m == nil {
+		return
+	}
+	m.adminPurgeQueue.WithLabelValues(
+		m.admitForAdminCounterBudget(queue),
+		normalizeSQSAdminPurgeOutcome(outcome),
+	).Inc()
+}
+
+// ObserveAdminPeekQueue counts one AdminPeekQueue call.
+func (m *SQSMetrics) ObserveAdminPeekQueue(queue, outcome string) {
+	if m == nil {
+		return
+	}
+	m.adminPeekQueue.WithLabelValues(
+		m.admitForAdminCounterBudget(queue),
+		normalizeSQSAdminPeekOutcome(outcome),
+	).Inc()
+}
+
+func (m *SQSMetrics) admitForAdminCounterBudget(queue string) string {
+	if queue == "" {
+		// An empty name reaches here only on a validation rejection,
+		// where there is no queue to attribute the call to.
+		return sqsQueueOverflow
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return admitCounterQueueLocked(queue, m.trackedAdminCounterQueues)
+}
+
+// normalizeSQSAdminPurgeOutcome keeps the purge label inside §3.6's
+// closed set. `throttled` is deliberately absent: purge signals
+// contention as purge_in_progress, and accepting both would let the
+// two paths drift into reporting the same condition differently.
+func normalizeSQSAdminPurgeOutcome(outcome string) string {
+	switch outcome {
+	case SQSAdminOutcomeOK,
+		SQSAdminOutcomeForbidden,
+		SQSAdminOutcomeNotLeader,
+		SQSAdminOutcomeNotFound,
+		SQSAdminOutcomeValidation,
+		SQSAdminOutcomePurgeInProgress,
+		SQSAdminOutcomeInternalError:
+		return outcome
+	default:
+		return SQSAdminOutcomeInternalError
+	}
+}
+
+// normalizeSQSAdminPeekOutcome keeps the peek label inside §3.6's
+// closed set. `purge_in_progress` is absent for the mirror-image
+// reason: peek is throttled, not generation-gated.
+func normalizeSQSAdminPeekOutcome(outcome string) string {
+	switch outcome {
+	case SQSAdminOutcomeOK,
+		SQSAdminOutcomeForbidden,
+		SQSAdminOutcomeNotLeader,
+		SQSAdminOutcomeNotFound,
+		SQSAdminOutcomeValidation,
+		SQSAdminOutcomeThrottled,
+		SQSAdminOutcomeInternalError:
+		return outcome
+	default:
+		return SQSAdminOutcomeInternalError
+	}
+}
+
 // ObservePartitionMessage implements SQSPartitionObserver. The
 // (queue, action) pair is validated and (queue) is collapsed to
 // the overflow label past sqsMaxTrackedQueues distinct names.
diff --git a/monitoring/sqs_admin_test.go b/monitoring/sqs_admin_test.go
new file mode 100644
index 000000000..eb2cee38a
--- /dev/null
+++ b/monitoring/sqs_admin_test.go
@@ -0,0 +1,140 @@
+package monitoring
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSQSAdminCountersRecordOutcomes(t *testing.T) {
+	t.Parallel()
+
+	reg := prometheus.NewRegistry()
+	m := newSQSMetrics(reg)
+
+	m.ObserveAdminPurgeQueue("orders", SQSAdminOutcomeOK)
+	m.ObserveAdminPurgeQueue("orders", SQSAdminOutcomePurgeInProgress)
+	m.ObserveAdminPeekQueue("orders-dlq", SQSAdminOutcomeOK)
+	m.ObserveAdminPeekQueue("orders-dlq", SQSAdminOutcomeThrottled)
+
+	require.NoError(t, testutil.GatherAndCompare(
+		reg,
+		strings.NewReader(`
+# HELP elastickv_sqs_admin_purge_queue_total Total admin PurgeQueue calls by queue and outcome.
+# TYPE elastickv_sqs_admin_purge_queue_total counter
+elastickv_sqs_admin_purge_queue_total{outcome="ok",queue="orders"} 1
+elastickv_sqs_admin_purge_queue_total{outcome="purge_in_progress",queue="orders"} 1
+# HELP elastickv_sqs_admin_peek_queue_total Total admin PeekQueue calls by queue and outcome.
+# TYPE elastickv_sqs_admin_peek_queue_total counter
+elastickv_sqs_admin_peek_queue_total{outcome="ok",queue="orders-dlq"} 1
+elastickv_sqs_admin_peek_queue_total{outcome="throttled",queue="orders-dlq"} 1
+`),
+		"elastickv_sqs_admin_purge_queue_total",
+		"elastickv_sqs_admin_peek_queue_total",
+	))
+}
+
+// TestSQSAdminCountersBoundTheOutcomeLabel is the cardinality guard.
+// The label must never be derived from an error string; an
+// unrecognised value collapses into internal_error rather than
+// minting a series.
+func TestSQSAdminCountersBoundTheOutcomeLabel(t *testing.T) {
+	t.Parallel()
+
+	reg := prometheus.NewRegistry()
+	m := newSQSMetrics(reg)
+
+	for _, bogus := range []string{"", "connection reset by peer", "i/o timeout"} {
+		m.ObserveAdminPurgeQueue("orders", bogus)
+		m.ObserveAdminPeekQueue("orders", bogus)
+	}
+
+	require.Equal(t, 1, testutil.CollectAndCount(m.adminPurgeQueue))
+	require.Equal(t, 1, testutil.CollectAndCount(m.adminPeekQueue))
+	require.InDelta(t, 3.0, testutil.ToFloat64(
+		m.adminPurgeQueue.WithLabelValues("orders", SQSAdminOutcomeInternalError)), 0.0001)
+}
+
+// TestSQSAdminOutcomeSetsAreAsymmetric pins the deliberate difference
+// between the two closed sets: purge reports contention as
+// purge_in_progress and peek reports it as throttled. Accepting both
+// on either counter would let the two paths drift into describing the
+// same condition two ways.
+func TestSQSAdminOutcomeSetsAreAsymmetric(t *testing.T) {
+	t.Parallel()
+
+	require.Equal(t, SQSAdminOutcomeInternalError,
+		normalizeSQSAdminPurgeOutcome(SQSAdminOutcomeThrottled),
+		"purge has no throttled outcome")
+	require.Equal(t, SQSAdminOutcomeInternalError,
+		normalizeSQSAdminPeekOutcome(SQSAdminOutcomePurgeInProgress),
+		"peek has no purge_in_progress outcome")
+
+	require.Equal(t, SQSAdminOutcomePurgeInProgress,
+		normalizeSQSAdminPurgeOutcome(SQSAdminOutcomePurgeInProgress))
+	require.Equal(t, SQSAdminOutcomeThrottled,
+		normalizeSQSAdminPeekOutcome(SQSAdminOutcomeThrottled))
+}
+
+// TestSQSAdminCountersBoundTheQueueLabel pins that operator-supplied
+// queue names cannot grow the series set without limit: past the
+// shared budget they collapse into the overflow label.
+func TestSQSAdminCountersBoundTheQueueLabel(t *testing.T) {
+	t.Parallel()
+
+	reg := prometheus.NewRegistry()
+	m := newSQSMetrics(reg)
+
+	for i := range sqsMaxTrackedQueues + 50 {
+		m.ObserveAdminPurgeQueue(queueNameForIndex(i), SQSAdminOutcomeOK)
+	}
+
+	require.LessOrEqual(t, testutil.CollectAndCount(m.adminPurgeQueue), sqsMaxTrackedQueues+1,
+		"queue names past the budget must collapse into the overflow label")
+	require.Positive(t, testutil.ToFloat64(
+		m.adminPurgeQueue.WithLabelValues(sqsQueueOverflow, SQSAdminOutcomeOK)))
+}
+
+// TestSQSAdminCountersAttributeAnEmptyQueueToOverflow covers the
+// validation rejection, where there is no queue to attribute to.
+func TestSQSAdminCountersAttributeAnEmptyQueueToOverflow(t *testing.T) {
+	t.Parallel()
+
+	reg := prometheus.NewRegistry()
+	m := newSQSMetrics(reg)
+
+	m.ObserveAdminPurgeQueue("", SQSAdminOutcomeValidation)
+	require.InDelta(t, 1.0, testutil.ToFloat64(
+		m.adminPurgeQueue.WithLabelValues(sqsQueueOverflow, SQSAdminOutcomeValidation)), 0.0001)
+}
+
+func TestSQSAdminCountersNilReceiverIsInert(t *testing.T) {
+	t.Parallel()
+
+	var m *SQSMetrics
+	require.NotPanics(t, func() {
+		m.ObserveAdminPurgeQueue("q", SQSAdminOutcomeOK)
+		m.ObserveAdminPeekQueue("q", SQSAdminOutcomeOK)
+	})
+}
+
+func queueNameForIndex(i int) string {
+	return "queue-" + strings.Repeat("x", i%3) + "-" + itoa(i)
+}
+
+func itoa(i int) string {
+	if i == 0 {
+		return "0"
+	}
+	var buf [20]byte
+	pos := len(buf)
+	for i > 0 {
+		pos--
+		buf[pos] = byte('0' + i%10)
+		i /= 10
+	}
+	return string(buf[pos:])
+}