diff --git a/providers/aws/aws.go b/providers/aws/aws.go
index eb2a1d7b3..8b848a975 100644
--- a/providers/aws/aws.go
+++ b/providers/aws/aws.go
@@ -637,9 +637,9 @@ func awsDrivers(p *Provider) *resourcediscovery.Drivers {
 // background ticker calls each one on its interval. New time-driven services
 // register here.
 func (p *Provider) Tickables() []config.Tickable {
-	// EFS runs before CloudWatch so the alarms evaluated in the same tick see
-	// the AWS/EFS metrics it just published.
-	return []config.Tickable{p.EFS, p.CloudWatch, p.SSM, p.CloudFormation}
+	// SQS and EFS run before CloudWatch so the alarms evaluated in the same
+	// tick see the queue gauges and AWS/EFS metrics they just published.
+	return []config.Tickable{p.SQS, p.EFS, p.CloudWatch, p.SSM, p.CloudFormation}
 }
 
 // Close tears down any real engines wired into the provider via
diff --git a/providers/aws/sqs/metrics.go b/providers/aws/sqs/metrics.go
new file mode 100644
index 000000000..f8bc603e9
--- /dev/null
+++ b/providers/aws/sqs/metrics.go
@@ -0,0 +1,209 @@
+package sqs
+
+import (
+	"context"
+	"math"
+	"time"
+
+	mondriver "github.com/stackshy/cloudemu/v2/services/monitoring/driver"
+)
+
+// AWS/SQS queue-state gauges, published on the single QueueName dimension
+// (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-available-cloudwatch-metrics.html).
+//
+// Real SQS pushes these once a minute for every active queue: one that holds
+// messages, or that an action accessed in the last six hours
+// (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/monitoring-using-cloudwatch.html).
+// Tick does the same on the shared serve scheduler, so a queue whose consumers
+// stopped keeps reporting a growing backlog and age. The counters
+// (NumberOfMessagesSent and friends) stay on the request paths.
+const (
+	metricsNamespace = "AWS/SQS"
+	unitCount        = "Count"
+	unitSeconds      = "Seconds"
+
+	metricVisible        = "ApproximateNumberOfMessagesVisible"
+	metricNotVisible     = "ApproximateNumberOfMessagesNotVisible"
+	metricDelayed        = "ApproximateNumberOfMessagesDelayed"
+	metricOldestAge      = "ApproximateAgeOfOldestMessage"
+	metricGroupsInflight = "ApproximateNumberOfGroupsWithInflightMessages"
+
+	// poisonPillReceiveCount is the receive count from which a standard queue
+	// moves an undeleted message to the back of the queue and leaves it out of
+	// ApproximateAgeOfOldestMessage until it is processed.
+	poisonPillReceiveCount = 3
+
+	// gaugeInterval is how often an active queue publishes its gauges.
+	gaugeInterval = time.Minute
+	// activeWindow is how long an action keeps an empty queue active.
+	activeWindow = 6 * time.Hour
+)
+
+// queueGauges is one sample of a queue's state gauges.
+type queueGauges struct {
+	visible, notVisible, delayed, groupsInflight int
+	// oldestAge is the age in whole seconds of the oldest message counted by
+	// ApproximateAgeOfOldestMessage; hasOldest is false when there is none.
+	oldestAge float64
+	hasOldest bool
+}
+
+// sampleGauges reads the queue's state gauges as of now. Caller holds qd.mu.
+func sampleGauges(qd *queueData, now time.Time) queueGauges {
+	var (
+		g      queueGauges
+		oldest time.Time
+		groups map[string]struct{}
+	)
+
+	for _, msg := range qd.messages {
+		if g.countState(msg, now) && qd.info.FIFO && msg.GroupID != "" {
+			if groups == nil {
+				groups = make(map[string]struct{})
+			}
+
+			groups[msg.GroupID] = struct{}{}
+		}
+
+		if countsForAge(qd.info.FIFO, msg) && (!g.hasOldest || msg.SentAt.Before(oldest)) {
+			oldest = msg.SentAt
+			g.hasOldest = true
+		}
+	}
+
+	g.groupsInflight = len(groups)
+
+	if g.hasOldest {
+		g.oldestAge = math.Max(0, math.Floor(now.Sub(oldest).Seconds()))
+	}
+
+	return g
+}
+
+// countState adds msg to the visible, delayed or in-flight count and reports
+// whether it is in flight. A message that has never been received and whose
+// visibility lies in the future is delayed; one that has been received and is
+// still hidden is in flight.
+func (g *queueGauges) countState(msg *sqsMessage, now time.Time) (inFlight bool) {
+	switch {
+	case !msg.VisibleAt.After(now):
+		g.visible++
+	case msg.ReceiveCount == 0:
+		g.delayed++
+	default:
+		g.notVisible++
+		return true
+	}
+
+	return false
+}
+
+// countsForAge reports whether msg counts toward ApproximateAgeOfOldestMessage.
+// A standard queue leaves out poison-pill messages (received three or more
+// times without being deleted); FIFO queues keep order, so every message
+// counts.
+func countsForAge(fifo bool, msg *sqsMessage) bool {
+	return fifo || msg.ReceiveCount < poisonPillReceiveCount
+}
+
+// Tick publishes one sample of the state gauges for every active queue whose
+// last sample is at least gaugeInterval old at now, and reports whether it
+// published anything. It has the config.Tickable signature of the shared serve
+// scheduler; library mode never calls it, so tests drive it with a FakeClock.
+// Each queue is sampled under its own lock and everything is published after
+// all locks are released, because a CloudWatch alarm on a gauge may notify an
+// SNS topic that delivers straight back into a queue.
+func (m *Mock) Tick(now time.Time) bool {
+	if m.monitoring == nil {
+		return false
+	}
+
+	var data []mondriver.MetricDatum
+
+	for _, qd := range m.queues.SortedValues() {
+		data = append(data, dueGauges(qd, now)...)
+	}
+
+	if len(data) == 0 {
+		return false
+	}
+
+	_ = m.monitoring.PutMetricData(context.Background(), data)
+
+	return true
+}
+
+// dueGauges samples qd when it is active and its gauges are due at now, and
+// records the sample time. It returns nil otherwise. Callers must not hold
+// qd.mu.
+func dueGauges(qd *queueData, now time.Time) []mondriver.MetricDatum {
+	qd.mu.Lock()
+	defer qd.mu.Unlock()
+
+	if !qd.lastSampleAt.IsZero() && now.Sub(qd.lastSampleAt) < gaugeInterval {
+		return nil
+	}
+
+	if len(qd.messages) == 0 && now.Sub(qd.lastAccessedAt()) >= activeWindow {
+		return nil
+	}
+
+	qd.lastSampleAt = now
+
+	return gaugeData(qd.info.Name, qd.info.FIFO, sampleGauges(qd, now), now)
+}
+
+// gaugeData builds the datums of one gauge sample.
+func gaugeData(queue string, fifo bool, g queueGauges, now time.Time) []mondriver.MetricDatum {
+	dims := map[string]string{"QueueName": queue}
+	datum := func(metric string, value float64, unit string) mondriver.MetricDatum {
+		return mondriver.MetricDatum{
+			Namespace: metricsNamespace, MetricName: metric, Value: value, Unit: unit,
+			Dimensions: dims, Timestamp: now,
+		}
+	}
+
+	data := []mondriver.MetricDatum{
+		datum(metricVisible, float64(g.visible), unitCount),
+		datum(metricNotVisible, float64(g.notVisible), unitCount),
+		datum(metricDelayed, float64(g.delayed), unitCount),
+	}
+
+	// The age is reported only while the queue holds a message it counts.
+	if g.hasOldest {
+		data = append(data, datum(metricOldestAge, g.oldestAge, unitSeconds))
+	}
+
+	if fifo {
+		data = append(data, datum(metricGroupsInflight, float64(g.groupsInflight), unitCount))
+	}
+
+	return data
+}
+
+// touch records an action on the queue at now.
+func (qd *queueData) touch(now time.Time) {
+	qd.lastAccessed.Store(now.UnixNano())
+}
+
+// lastAccessedAt is the time of the last action on the queue, or the zero
+// time when none was recorded.
+func (qd *queueData) lastAccessedAt() time.Time {
+	n := qd.lastAccessed.Load()
+	if n == 0 {
+		return time.Time{}
+	}
+
+	return time.Unix(0, n).UTC()
+}
+
+// accessQueue looks up the queue at url for an API action and records the
+// access, which keeps the queue active for its CloudWatch gauges.
+func (m *Mock) accessQueue(url string) (*queueData, bool) {
+	qd, ok := m.queues.Get(url)
+	if ok {
+		qd.touch(m.opts.Clock.Now())
+	}
+
+	return qd, ok
+}
diff --git a/providers/aws/sqs/metrics_test.go b/providers/aws/sqs/metrics_test.go
new file mode 100644
index 000000000..12bfbd25d
--- /dev/null
+++ b/providers/aws/sqs/metrics_test.go
@@ -0,0 +1,524 @@
+package sqs
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stackshy/cloudemu/v2/config"
+	"github.com/stackshy/cloudemu/v2/services/messagequeue/driver"
+	mondriver "github.com/stackshy/cloudemu/v2/services/monitoring/driver"
+)
+
+// gaugeRecorder is a monitoring backend that keeps every datum it is sent.
+// onPut, when set, runs inside PutMetricData, so a test can call back into
+// the SQS mock the way a CloudWatch alarm -> SNS -> SQS action would.
+type gaugeRecorder struct {
+	mondriver.Monitoring
+
+	mu    sync.Mutex
+	data  []mondriver.MetricDatum
+	onPut func()
+}
+
+func (r *gaugeRecorder) PutMetricData(_ context.Context, data []mondriver.MetricDatum) error {
+	r.mu.Lock()
+	r.data = append(r.data, data...)
+	onPut := r.onPut
+	r.mu.Unlock()
+
+	if onPut != nil {
+		onPut()
+	}
+
+	return nil
+}
+
+// samples returns every datum of metric for queue, oldest first.
+func (r *gaugeRecorder) samples(metric, queue string) []mondriver.MetricDatum {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	var out []mondriver.MetricDatum
+
+	for _, d := range r.data {
+		if d.MetricName == metric && d.Dimensions["QueueName"] == queue {
+			out = append(out, d)
+		}
+	}
+
+	return out
+}
+
+// last returns the most recent datum of metric for queue.
+func (r *gaugeRecorder) last(metric, queue string) (mondriver.MetricDatum, bool) {
+	s := r.samples(metric, queue)
+	if len(s) == 0 {
+		return mondriver.MetricDatum{}, false
+	}
+
+	return s[len(s)-1], true
+}
+
+func (r *gaugeRecorder) reset() {
+	r.mu.Lock()
+	r.data = nil
+	r.mu.Unlock()
+}
+
+func newGaugeMock() (*Mock, *gaugeRecorder, *config.FakeClock) {
+	m, fc := newTestMock()
+	rec := &gaugeRecorder{}
+	m.SetMonitoring(rec)
+
+	return m, rec, fc
+}
+
+func requireGauge(t *testing.T, rec *gaugeRecorder, metric, queue string, want float64, unit string) {
+	t.Helper()
+
+	d, ok := rec.last(metric, queue)
+	if !ok {
+		t.Fatalf("%s for %s was never emitted", metric, queue)
+	}
+
+	if d.Value != want || d.Unit != unit || d.Namespace != "AWS/SQS" {
+		t.Fatalf("%s for %s = %v %s (%s), want %v %s (AWS/SQS)", metric, queue, d.Value, d.Unit, d.Namespace, want, unit)
+	}
+
+	if len(d.Dimensions) != 1 {
+		t.Fatalf("%s dimensions = %v, want only QueueName", metric, d.Dimensions)
+	}
+}
+
+func requireNoGauge(t *testing.T, rec *gaugeRecorder, metric, queue string) {
+	t.Helper()
+
+	if d, ok := rec.last(metric, queue); ok {
+		t.Fatalf("%s for %s emitted %v, want no datum", metric, queue, d.Value)
+	}
+}
+
+func TestRequestsPublishCountersButNoGauges(t *testing.T) {
+	m, rec, _ := newGaugeMock()
+	ctx := context.Background()
+	q := createStdQueue(m, "quiet-paths")
+
+	sendN(t, m, q.URL, 3)
+
+	msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL})
+	requireNoError(t, err)
+	requireNoError(t, m.ChangeVisibility(ctx, q.URL, msgs[0].ReceiptHandle, 0))
+	requireNoError(t, m.DeleteMessage(ctx, q.URL, msgs[0].ReceiptHandle))
+	requireNoError(t, m.PurgeQueue(ctx, q.URL))
+
+	for _, g := range []string{metricVisible, metricNotVisible, metricDelayed, metricOldestAge} {
+		requireNoGauge(t, rec, g, "quiet-paths")
+	}
+
+	if got := len(rec.samples("NumberOfMessagesSent", "quiet-paths")); got != 3 {
+		t.Fatalf("NumberOfMessagesSent datums = %d, want 3", got)
+	}
+}
+
+func TestTickPublishesOneSamplePerMinute(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	q := createStdQueue(m, "per-minute")
+
+	// Six sends in the same minute: one sample of 6, not one per send.
+	sendN(t, m, q.URL, 6)
+
+	if !m.Tick(fc.Now()) {
+		t.Fatal("first Tick published nothing")
+	}
+
+	requireGauge(t, rec, metricVisible, "per-minute", 6, unitCount)
+
+	for _, step := range []time.Duration{time.Second, 30 * time.Second, 28 * time.Second} {
+		fc.Advance(step)
+
+		if m.Tick(fc.Now()) {
+			t.Fatalf("Tick %v after the sample published again inside the minute", fc.Now())
+		}
+	}
+
+	fc.Advance(time.Second) // a full minute since the first sample
+
+	if !m.Tick(fc.Now()) {
+		t.Fatal("Tick a minute after the last sample published nothing")
+	}
+
+	if got := len(rec.samples(metricVisible, "per-minute")); got != 2 {
+		t.Fatalf("Visible samples = %d, want 2 (one per minute)", got)
+	}
+}
+
+func TestTickTracksVisibleInFlightDelayedAndAge(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	ctx := context.Background()
+	q := createStdQueue(m, "gauges")
+
+	_, err := m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "a"})
+	requireNoError(t, err)
+
+	_, err = m.SendMessage(ctx, driver.SendMessageInput{
+		QueueURL: q.URL, Body: "delayed", DelaySeconds: 600, DelaySecondsSet: true,
+	})
+	requireNoError(t, err)
+
+	fc.Advance(10 * time.Second)
+
+	_, err = m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "c"})
+	requireNoError(t, err)
+
+	msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL, VisibilityTimeout: 300})
+	requireNoError(t, err)
+	assertEqual(t, "a", msgs[0].Body)
+
+	m.Tick(fc.Now())
+
+	// a is in flight, one message is delayed, c is visible, and the oldest
+	// message was sent 10s ago.
+	requireGauge(t, rec, metricVisible, "gauges", 1, unitCount)
+	requireGauge(t, rec, metricNotVisible, "gauges", 1, unitCount)
+	requireGauge(t, rec, metricDelayed, "gauges", 1, unitCount)
+	requireGauge(t, rec, metricOldestAge, "gauges", 10, unitSeconds)
+	requireNoGauge(t, rec, metricGroupsInflight, "gauges")
+
+	// GetQueueAttributes agrees with the gauges: in flight excludes delayed.
+	attrs, err := m.GetQueueAttributes(ctx, q.URL)
+	requireNoError(t, err)
+	assertEqual(t, 1, attrs.ApproximateMessageCount)
+	assertEqual(t, 1, attrs.ApproximateNotVisibleCount)
+	assertEqual(t, 1, attrs.ApproximateDelayedCount)
+}
+
+// TestTickStuckQueueKeepsReportingAndAges is the stopped-consumer case: after
+// a single delayed send nothing touches the queue, and the gauges still show
+// the message becoming visible and getting older every minute.
+func TestTickStuckQueueKeepsReportingAndAges(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	ctx := context.Background()
+	q := createStdQueue(m, "stuck")
+
+	_, err := m.SendMessage(ctx, driver.SendMessageInput{
+		QueueURL: q.URL, Body: "late", DelaySeconds: 5, DelaySecondsSet: true,
+	})
+	requireNoError(t, err)
+
+	m.Tick(fc.Now())
+	requireGauge(t, rec, metricVisible, "stuck", 0, unitCount)
+	requireGauge(t, rec, metricDelayed, "stuck", 1, unitCount)
+	requireGauge(t, rec, metricOldestAge, "stuck", 0, unitSeconds)
+
+	fc.Advance(time.Minute)
+	m.Tick(fc.Now())
+	requireGauge(t, rec, metricVisible, "stuck", 1, unitCount)
+	requireGauge(t, rec, metricDelayed, "stuck", 0, unitCount)
+	requireGauge(t, rec, metricOldestAge, "stuck", 60, unitSeconds)
+
+	// A queue holding messages stays active long past the six-hour window.
+	for i := 0; i < 8*60; i++ {
+		fc.Advance(time.Minute)
+		m.Tick(fc.Now())
+	}
+
+	requireGauge(t, rec, metricOldestAge, "stuck", (8*60+1)*60, unitSeconds)
+
+	if got := len(rec.samples(metricVisible, "stuck")); got != 8*60+2 {
+		t.Fatalf("Visible samples = %d, want %d (one per minute)", got, 8*60+2)
+	}
+}
+
+func TestTickStopsForEmptyQueueIdleSixHours(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	ctx := context.Background()
+	q := createStdQueue(m, "idle")
+
+	if !m.Tick(fc.Now()) {
+		t.Fatal("a just-created queue is active and should publish")
+	}
+
+	fc.Advance(activeWindow - time.Minute)
+
+	if !m.Tick(fc.Now()) {
+		t.Fatal("an empty queue accessed under six hours ago should still publish")
+	}
+
+	fc.Advance(time.Minute)
+	rec.reset()
+
+	if m.Tick(fc.Now()) {
+		t.Fatal("an empty queue idle for six hours should stop publishing")
+	}
+
+	// Any action wakes it up again.
+	_, err := m.GetQueueAttributes(ctx, q.URL)
+	requireNoError(t, err)
+
+	fc.Advance(time.Minute)
+
+	if !m.Tick(fc.Now()) {
+		t.Fatal("an accessed queue should publish again")
+	}
+
+	requireGauge(t, rec, metricVisible, "idle", 0, unitCount)
+	requireNoGauge(t, rec, metricOldestAge, "idle")
+}
+
+func TestTickAgeSkipsStandardPoisonPills(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	ctx := context.Background()
+	q := createStdQueue(m, "poison")
+
+	_, err := m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "poison"})
+	requireNoError(t, err)
+
+	fc.Advance(100 * time.Second)
+
+	receive := func() {
+		t.Helper()
+
+		msgs, rerr := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{
+			QueueURL: q.URL, VisibilityTimeout: 0, VisibilityTimeoutSet: true,
+		})
+		requireNoError(t, rerr)
+		assertEqual(t, 1, len(msgs))
+	}
+
+	receive()
+	receive()
+	m.Tick(fc.Now())
+	requireGauge(t, rec, metricOldestAge, "poison", 100, unitSeconds)
+
+	// The third receive makes it a poison pill: nothing else is queued, so
+	// there is no age to report.
+	receive()
+	rec.reset()
+	fc.Advance(time.Minute)
+	m.Tick(fc.Now())
+	requireGauge(t, rec, metricVisible, "poison", 1, unitCount)
+	requireNoGauge(t, rec, metricOldestAge, "poison")
+}
+
+func TestTickFIFOCountsInflightGroups(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	ctx := context.Background()
+
+	q, err := m.CreateQueue(ctx, driver.QueueConfig{Name: "orders.fifo", FIFO: true, ContentBasedDeduplication: true})
+	requireNoError(t, err)
+
+	for i, g := range []string{"g1", "g1", "g2", "g3"} {
+		_, err = m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: fmt.Sprintf("%s-%d", g, i), GroupID: g})
+		requireNoError(t, err)
+	}
+
+	fc.Advance(30 * time.Second)
+
+	msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL, MaxMessages: 10, VisibilityTimeout: 300})
+	requireNoError(t, err)
+	assertEqual(t, 4, len(msgs))
+
+	m.Tick(fc.Now())
+	requireGauge(t, rec, metricGroupsInflight, "orders.fifo", 3, unitCount)
+	requireGauge(t, rec, metricNotVisible, "orders.fifo", 4, unitCount)
+	requireGauge(t, rec, metricOldestAge, "orders.fifo", 30, unitSeconds)
+
+	for _, msg := range msgs[:2] {
+		requireNoError(t, m.DeleteMessage(ctx, q.URL, msg.ReceiptHandle))
+	}
+
+	fc.Advance(time.Minute)
+	m.Tick(fc.Now())
+	requireGauge(t, rec, metricGroupsInflight, "orders.fifo", 2, unitCount)
+	requireGauge(t, rec, metricNotVisible, "orders.fifo", 2, unitCount)
+}
+
+func TestTickFollowsDeadLetterRedriveAndMoveTask(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	ctx := context.Background()
+
+	dlq := createStdQueue(m, "dlq")
+	src, err := m.CreateQueue(ctx, driver.QueueConfig{Name: "src", RedrivePolicy: redrivePolicy(dlq.ARN, 1)})
+	requireNoError(t, err)
+
+	sendN(t, m, src.URL, 1)
+	fc.Advance(50 * time.Second)
+
+	receive := func() {
+		t.Helper()
+
+		_, rerr := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: src.URL, VisibilityTimeout: 1})
+		requireNoError(t, rerr)
+	}
+
+	receive()
+	fc.Advance(2 * time.Second)
+	receive() // exceeds maxReceiveCount=1: the message moves to the DLQ
+
+	m.Tick(fc.Now())
+	requireGauge(t, rec, metricVisible, "src", 0, unitCount)
+	requireGauge(t, rec, metricVisible, "dlq", 1, unitCount)
+	// The age resets when a message is moved to a DLQ.
+	requireGauge(t, rec, metricOldestAge, "dlq", 0, unitSeconds)
+
+	fc.Advance(time.Minute)
+	requireGaugeAfterMove(t, m, rec, fc, dlq.ARN)
+}
+
+func requireGaugeAfterMove(t *testing.T, m *Mock, rec *gaugeRecorder, fc *config.FakeClock, dlqARN string) {
+	t.Helper()
+
+	_, err := m.StartMessageMoveTask(context.Background(), dlqARN, "", 0)
+	requireNoError(t, err)
+
+	m.Tick(fc.Now())
+	requireGauge(t, rec, metricVisible, "dlq", 0, unitCount)
+	requireGauge(t, rec, metricVisible, "src", 1, unitCount)
+}
+
+// TestTickPublishesOutsideQueueLocks calls back into the queues from inside
+// PutMetricData, as a CloudWatch alarm whose SNS action delivers to a queue
+// would. Publishing while holding a queue lock would deadlock here.
+func TestTickPublishesOutsideQueueLocks(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	ctx := context.Background()
+	q := createStdQueue(m, "reentrant")
+	sendN(t, m, q.URL, 1)
+
+	var called atomic.Bool
+
+	// Only the first publish calls back, or the SendMessage counter it emits
+	// would call back again forever.
+	rec.onPut = func() {
+		if called.CompareAndSwap(false, true) {
+			_, _ = m.GetQueueAttributes(ctx, q.URL)
+			_, _ = m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "from-alarm"})
+		}
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		m.Tick(fc.Now())
+
+		msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL})
+		if err == nil && len(msgs) == 1 {
+			_ = m.DeleteMessage(ctx, q.URL, msgs[0].ReceiptHandle)
+		}
+
+		_, _ = m.ReceiveMessagesWithOptions(ctx, q.URL, driver.ReceiveOptions{})
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("publishing a metric deadlocked on a queue lock")
+	}
+}
+
+func TestTickWithoutMonitoringIsNoOp(t *testing.T) {
+	m, fc := newTestMock()
+	q := createStdQueue(m, "no-backend")
+	sendN(t, m, q.URL, 1)
+
+	if m.Tick(fc.Now()) {
+		t.Fatal("Tick without a monitoring backend reported a change")
+	}
+}
+
+func TestSnapshotKeepsGaugeSchedule(t *testing.T) {
+	ctx := context.Background()
+	src, rec, fc := newGaugeMock()
+	createStdQueue(src, "sched")
+
+	src.Tick(fc.Now())
+	requireGauge(t, rec, metricVisible, "sched", 0, unitCount)
+
+	data, err := src.Snapshot(ctx, false)
+	requireNoError(t, err)
+
+	dst, dstRec, _ := newGaugeMock()
+	requireNoError(t, dst.Restore(ctx, data))
+
+	// The restored queue sampled at fc.Now(), so it is not due again yet.
+	if dst.Tick(fc.Now().Add(30 * time.Second)) {
+		t.Fatal("restored queue lost its last sample time")
+	}
+
+	// It was accessed at fc.Now(), so it is still active a minute later.
+	if !dst.Tick(fc.Now().Add(time.Minute)) {
+		t.Fatal("restored queue lost its last access time")
+	}
+
+	requireGauge(t, dstRec, metricVisible, "sched", 0, unitCount)
+
+	// Six hours after that access the empty queue is idle.
+	if dst.Tick(fc.Now().Add(activeWindow + time.Minute)) {
+		t.Fatal("restored empty queue should be idle after six hours")
+	}
+}
+
+var errHandlerFailed = errors.New("handler failed")
+
+// failingInvoker is an event-source mapping whose handler always fails.
+type failingInvoker struct{}
+
+func (failingInvoker) DeliverEventSourceBatch(context.Context, string, []byte) (bool, error) {
+	return true, errHandlerFailed
+}
+
+func TestTickFollowsLambdaESMDeadLetterRedrive(t *testing.T) {
+	m, rec, fc := newGaugeMock()
+	ctx := context.Background()
+
+	dlq := createStdQueue(m, "esm-dlq")
+	src, err := m.CreateQueue(ctx, driver.QueueConfig{Name: "esm-src", RedrivePolicy: redrivePolicy(dlq.ARN, 2)})
+	requireNoError(t, err)
+
+	m.SetEventSourceInvoker(failingInvoker{})
+
+	sendN(t, m, src.URL, 1)
+	m.Tick(fc.Now())
+
+	requireGauge(t, rec, metricVisible, "esm-dlq", 1, unitCount)
+	requireGauge(t, rec, metricVisible, "esm-src", 0, unitCount)
+	requireGauge(t, rec, metricNotVisible, "esm-src", 0, unitCount)
+}
+
+func TestSenderIDFromCallerElseAccount(t *testing.T) {
+	m, _ := newTestMock()
+	ctx := context.Background()
+	q := createStdQueue(m, "sender")
+
+	_, err := m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "user", SenderID: "AIDAEXAMPLE123ABC"})
+	requireNoError(t, err)
+
+	res, err := m.SendMessageBatch(ctx, q.URL, []driver.BatchSendEntry{
+		{ID: "1", Body: "role", SenderID: "AROAEXAMPLE123ABC:session"},
+	})
+	requireNoError(t, err)
+	assertEqual(t, 1, len(res.Successful))
+
+	_, err = m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "library"})
+	requireNoError(t, err)
+
+	msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL, MaxMessages: 10})
+	requireNoError(t, err)
+	assertEqual(t, 3, len(msgs))
+
+	got := map[string]string{}
+	for _, msg := range msgs {
+		got[msg.Body] = msg.SystemAttributes["SenderId"]
+	}
+
+	assertEqual(t, "AIDAEXAMPLE123ABC", got["user"])
+	assertEqual(t, "AROAEXAMPLE123ABC:session", got["role"])
+	assertEqual(t, m.opts.AccountID, got["library"])
+}
diff --git a/providers/aws/sqs/movetask.go b/providers/aws/sqs/movetask.go
index 8cdf84a03..59e45daec 100644
--- a/providers/aws/sqs/movetask.go
+++ b/providers/aws/sqs/movetask.go
@@ -50,7 +50,7 @@ func (m *Mock) StartMessageMoveTask(_ context.Context, sourceARN, destARN string
 	}
 
 	sourceURL := m.urlForARN(sourceARN)
-	if sourceURL == "" {
+	if _, ok := m.accessQueue(sourceURL); !ok {
 		return "", errors.Newf(errors.NotFound, "no queue found for source arn %q", sourceARN)
 	}
 
diff --git a/providers/aws/sqs/permissions.go b/providers/aws/sqs/permissions.go
index 92655940b..5d00f2982 100644
--- a/providers/aws/sqs/permissions.go
+++ b/providers/aws/sqs/permissions.go
@@ -11,7 +11,7 @@ import (
 // the given SQS actions on the queue, mutating the stored access policy. It is
 // AWS-specific and used directly by the SQS wire handler.
 func (m *Mock) AddPermission(_ context.Context, queueURL, label string, accountIDs, actions []string) error {
-	qd, ok := m.queues.Get(queueURL)
+	qd, ok := m.accessQueue(queueURL)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queueURL)
 	}
@@ -50,7 +50,7 @@ func (m *Mock) AddPermission(_ context.Context, queueURL, label string, accountI
 // RemovePermission removes the statement whose Sid equals label. It is
 // AWS-specific and used directly by the SQS wire handler.
 func (m *Mock) RemovePermission(_ context.Context, queueURL, label string) error {
-	qd, ok := m.queues.Get(queueURL)
+	qd, ok := m.accessQueue(queueURL)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queueURL)
 	}
diff --git a/providers/aws/sqs/snapshot.go b/providers/aws/sqs/snapshot.go
index 8d505759f..6e38a52af 100644
--- a/providers/aws/sqs/snapshot.go
+++ b/providers/aws/sqs/snapshot.go
@@ -48,6 +48,11 @@ type queueSnapshot struct {
 	DeduplicationIndex map[string]time.Time     `json:"deduplicationIndex,omitempty"`
 	DLQConfig          *driver.DeadLetterConfig `json:"dlqConfig,omitempty"`
 	SeqCounter         uint64                   `json:"seqCounter,omitempty"`
+	// LastAccessedAt and LastSampleAt drive the per-minute CloudWatch gauges
+	// (metrics.go): whether the queue is still active, and when it last
+	// published.
+	LastAccessedAt time.Time `json:"lastAccessedAt"`
+	LastSampleAt   time.Time `json:"lastSampleAt"`
 }
 
 // messageSnapshot mirrors sqsMessage, promoting its one unexported field
@@ -132,6 +137,7 @@ func snapshotQueue(qd *queueData) *queueSnapshot {
 		FifoThroughputLim: qd.fifoThroughputLim, CreatedAt: qd.createdAt,
 		LastModifiedAt: qd.lastModifiedAt, DeduplicationIndex: qd.deduplicationIndex,
 		DLQConfig: qd.dlqConfig, SeqCounter: qd.seqCounter.Load(),
+		LastAccessedAt: qd.lastAccessedAt(), LastSampleAt: qd.lastSampleAt,
 	}
 
 	if len(qd.messages) > 0 {
@@ -190,7 +196,11 @@ func restoreQueue(qs *queueSnapshot) *queueData {
 		sqsManagedSSE: qs.SqsManagedSSE, deduplicationScope: qs.DeduplicationScope,
 		fifoThroughputLim: qs.FifoThroughputLim, createdAt: qs.CreatedAt,
 		lastModifiedAt: qs.LastModifiedAt, deduplicationIndex: qs.DeduplicationIndex,
-		dlqConfig: qs.DLQConfig,
+		dlqConfig: qs.DLQConfig, lastSampleAt: qs.LastSampleAt,
+	}
+
+	if !qs.LastAccessedAt.IsZero() {
+		qd.touch(qs.LastAccessedAt)
 	}
 
 	if qd.deduplicationIndex == nil {
diff --git a/providers/aws/sqs/sqs.go b/providers/aws/sqs/sqs.go
index d2c710159..c8b53223a 100644
--- a/providers/aws/sqs/sqs.go
+++ b/providers/aws/sqs/sqs.go
@@ -87,6 +87,13 @@ type queueData struct {
 	deduplicationIndex map[string]time.Time
 	dlqConfig          *driver.DeadLetterConfig
 	seqCounter         atomic.Uint64
+
+	// lastAccessed is the provider-clock time (Unix nanoseconds) of the last
+	// action on the queue. With the message count it decides whether the
+	// queue is active and so publishes its CloudWatch gauges (see metrics.go).
+	lastAccessed atomic.Int64
+	// lastSampleAt is when the gauges were last published. Guarded by mu.
+	lastSampleAt time.Time
 }
 
 // EventSourceInvoker delivers an SQS event-source-mapping batch to whatever
@@ -123,7 +130,7 @@ func (m *Mock) emitMetric(metricName string, value float64, unit string, dims ma
 	}
 
 	_ = m.monitoring.PutMetricData(context.Background(), []mondriver.MetricDatum{{
-		Namespace: "AWS/SQS", MetricName: metricName, Value: value, Unit: unit,
+		Namespace: metricsNamespace, MetricName: metricName, Value: value, Unit: unit,
 		Dimensions: dims, Timestamp: m.opts.Clock.Now(),
 	}})
 }
@@ -197,7 +204,7 @@ func (m *Mock) CreateQueue(ctx context.Context, cfg driver.QueueConfig) (*driver
 	// CreateQueue is idempotent: re-creating an existing queue with the exact
 	// same attributes returns the existing URL. QueueNameExists is returned only
 	// when the incoming attributes differ from the stored ones.
-	if existing, ok := m.queues.Get(url); ok {
+	if existing, ok := m.accessQueue(url); ok {
 		if sameQueueConfig(existing, &cfg) {
 			result := existing.info
 			return &result, nil
@@ -255,6 +262,7 @@ func (m *Mock) CreateQueue(ctx context.Context, cfg driver.QueueConfig) (*driver
 		}
 	}
 
+	qd.touch(now)
 	m.queues.Set(url, qd)
 
 	result := info
@@ -443,7 +451,7 @@ func (m *Mock) DeleteQueue(_ context.Context, url string) error {
 
 // GetQueueInfo retrieves information about an SQS queue by URL.
 func (m *Mock) GetQueueInfo(_ context.Context, url string) (*driver.QueueInfo, error) {
-	qd, ok := m.queues.Get(url)
+	qd, ok := m.accessQueue(url)
 	if !ok {
 		return nil, errors.Newf(errors.NotFound, "queue %q not found", url)
 	}
@@ -487,7 +495,7 @@ func (m *Mock) ListQueues(_ context.Context, prefix string) ([]driver.QueueInfo,
 //
 //nolint:gocritic // hugeParam: interface method signature cannot be changed.
 func (m *Mock) SendMessage(ctx context.Context, input driver.SendMessageInput) (*driver.SendMessageOutput, error) {
-	qd, ok := m.queues.Get(input.QueueURL)
+	qd, ok := m.accessQueue(input.QueueURL)
 	if !ok {
 		return nil, errors.Newf(errors.NotFound, "queue %q not found", input.QueueURL)
 	}
@@ -769,13 +777,24 @@ func (m *Mock) buildStoredMessage(qd *queueData, input *driver.SendMessageInput,
 		Attributes:        attrs,
 		MessageAttributes: copyMessageAttributes(input.MessageAttributes),
 		SystemAttributes:  systemAttributeStrings(input.SystemAttributes),
-		SenderID:          m.opts.AccountID,
+		SenderID:          senderID(input.SenderID, m.opts.AccountID),
 		SequenceNumber:    seqNum,
 		VisibleAt:         now.Add(time.Duration(delaySeconds) * time.Second),
 		SentAt:            now,
 	}
 }
 
+// senderID is the SenderId recorded on a sent message: the caller's IAM unique
+// id when the wire handler resolved one, else the account id (library callers
+// and service-to-service deliveries carry no caller principal).
+func senderID(caller, accountID string) string {
+	if caller != "" {
+		return caller
+	}
+
+	return accountID
+}
+
 // systemAttributeStrings flattens caller-supplied SQS message system attributes
 // (e.g. AWSTraceHeader) into a name->string map for storage. Only the string
 // value is retained; system attributes are always String-typed in SQS.
@@ -868,7 +887,7 @@ func findDuplicate(qd *queueData, input *driver.SendMessageInput, now time.Time)
 // When WaitTimeSeconds (or the queue's ReceiveMessageWaitTimeSeconds default) is > 0, it long-polls:
 // it retries on a short interval until a message is available, the deadline elapses, or ctx is canceled.
 func (m *Mock) ReceiveMessages(ctx context.Context, input driver.ReceiveMessageInput) ([]driver.Message, error) {
-	qd, ok := m.queues.Get(input.QueueURL)
+	qd, ok := m.accessQueue(input.QueueURL)
 	if !ok {
 		return nil, errors.Newf(errors.NotFound, "queue %q not found", input.QueueURL)
 	}
@@ -1125,7 +1144,7 @@ func (m *Mock) moveToDLQ(dlqURL, sourceURL string, msg *sqsMessage) {
 
 // DeleteMessage deletes a message from the specified queue using its receipt handle.
 func (m *Mock) DeleteMessage(_ context.Context, queueURL, receiptHandle string) error {
-	qd, ok := m.queues.Get(queueURL)
+	qd, ok := m.accessQueue(queueURL)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queueURL)
 	}
@@ -1138,19 +1157,28 @@ func (m *Mock) DeleteMessage(_ context.Context, queueURL, receiptHandle string)
 	}
 
 	qd.mu.Lock()
-	defer qd.mu.Unlock()
+
+	deleted := false
 
 	for i, msg := range qd.messages {
 		if msg.ReceiptHandle == receiptHandle {
 			qd.messages = append(qd.messages[:i], qd.messages[i+1:]...)
-			m.emitMetric("NumberOfMessagesDeleted", 1, "Count", map[string]string{"QueueName": qd.info.Name})
+			deleted = true
 
-			return nil
+			break
 		}
 	}
 
+	qd.mu.Unlock()
+
 	// DeleteMessage is idempotent: an old/stale (but well-formed) receipt handle
-	// against an existing queue succeeds without deleting anything.
+	// against an existing queue succeeds without deleting anything. The metric
+	// is published after the lock is released, since an alarm on it may
+	// deliver back into this queue.
+	if deleted {
+		m.emitMetric("NumberOfMessagesDeleted", 1, "Count", map[string]string{"QueueName": qd.info.Name})
+	}
+
 	return nil
 }
 
@@ -1178,7 +1206,7 @@ func isWellFormedReceiptHandle(handle string) bool {
 
 // ChangeVisibility changes the visibility timeout of a message in the specified queue.
 func (m *Mock) ChangeVisibility(_ context.Context, queueURL, receiptHandle string, timeout int) error {
-	qd, ok := m.queues.Get(queueURL)
+	qd, ok := m.accessQueue(queueURL)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queueURL)
 	}
@@ -1244,6 +1272,7 @@ func batchEntryToSendInput(queue string, entry *driver.BatchSendEntry) driver.Se
 		Attributes:        entry.Attributes,
 		MessageAttributes: entry.MessageAttributes,
 		SystemAttributes:  entry.SystemAttributes,
+		SenderID:          entry.SenderID,
 	}
 }
 
@@ -1279,7 +1308,7 @@ func (m *Mock) DeleteMessageBatch(
 func (m *Mock) ReceiveMessagesWithOptions(
 	_ context.Context, queue string, opts driver.ReceiveOptions,
 ) ([]driver.Message, error) {
-	qd, ok := m.queues.Get(queue)
+	qd, ok := m.accessQueue(queue)
 	if !ok {
 		return nil, errors.Newf(errors.NotFound, "queue %q not found", queue)
 	}
@@ -1289,7 +1318,6 @@ func (m *Mock) ReceiveMessagesWithOptions(
 	}
 
 	qd.mu.Lock()
-	defer qd.mu.Unlock()
 
 	maxMsgs := defaultMaxMessages(opts.MaxMessages)
 
@@ -1302,17 +1330,14 @@ func (m *Mock) ReceiveMessagesWithOptions(
 	results, toRemove := m.collectVisibleMessages(qd, maxMsgs, visTimeout, now)
 
 	removeByIndices(qd, toRemove)
+	qd.mu.Unlock()
 
 	if results == nil {
 		results = []driver.Message{}
 	}
 
-	dims := map[string]string{"QueueName": qd.info.Name}
-	if len(results) > 0 {
-		m.emitMetric("NumberOfMessagesReceived", float64(len(results)), "Count", dims)
-	} else {
-		m.emitMetric("NumberOfEmptyReceives", 1.0, "Count", dims)
-	}
+	// Published after the lock is released, like ReceiveMessages.
+	m.emitReceiveMetrics(qd, len(results))
 
 	return results, nil
 }
@@ -1321,7 +1346,7 @@ func (m *Mock) ReceiveMessagesWithOptions(
 func (m *Mock) GetQueueAttributes(
 	_ context.Context, queue string,
 ) (*driver.QueueAttributes, error) {
-	qd, ok := m.queues.Get(queue)
+	qd, ok := m.accessQueue(queue)
 	if !ok {
 		return nil, errors.Newf(errors.NotFound, "queue %q not found", queue)
 	}
@@ -1329,9 +1354,11 @@ func (m *Mock) GetQueueAttributes(
 	qd.mu.Lock()
 	defer qd.mu.Unlock()
 
-	now := m.opts.Clock.Now()
-	visible, notVisible := countMessages(qd, now)
-	delayed := countDelayedMessages(qd, now)
+	// ApproximateNumberOfMessagesNotVisible counts in-flight messages only
+	// (received, not yet deleted or expired); delayed messages are reported
+	// separately, matching the CloudWatch gauges.
+	g := sampleGauges(qd, m.opts.Clock.Now())
+	visible, notVisible, delayed := g.visible, g.notVisible, g.delayed
 
 	return &driver.QueueAttributes{
 		DelaySeconds:                  qd.delaySeconds,
@@ -1357,24 +1384,11 @@ func (m *Mock) GetQueueAttributes(
 	}, nil
 }
 
-// countDelayedMessages counts messages still within their delay window.
-func countDelayedMessages(qd *queueData, now time.Time) int {
-	delayed := 0
-
-	for _, msg := range qd.messages {
-		if msg.ReceiveCount == 0 && msg.VisibleAt.After(now) {
-			delayed++
-		}
-	}
-
-	return delayed
-}
-
 // SetQueueAttributes updates the attributes of the specified queue.
 func (m *Mock) SetQueueAttributes(
 	_ context.Context, queue string, attrs map[string]int,
 ) error {
-	qd, ok := m.queues.Get(queue)
+	qd, ok := m.accessQueue(queue)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queue)
 	}
@@ -1415,7 +1429,7 @@ func applyQueueAttributes(qd *queueData, attrs map[string]int) {
 // Policy, KmsMasterKeyId) that the numeric SetQueueAttributes path cannot express.
 // It is AWS-specific and used directly by the SQS wire handler.
 func (m *Mock) SetQueueAttributesRaw(_ context.Context, queue string, attrs map[string]string) error {
-	qd, ok := m.queues.Get(queue)
+	qd, ok := m.accessQueue(queue)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queue)
 	}
@@ -1512,7 +1526,7 @@ func parseNumericAttrs(attrs map[string]string) map[string]int {
 // ListDeadLetterSourceQueues returns the URLs of queues whose RedrivePolicy
 // targets the given DLQ URL.
 func (m *Mock) ListDeadLetterSourceQueues(_ context.Context, dlqURL string) ([]string, error) {
-	if _, ok := m.queues.Get(dlqURL); !ok {
+	if _, ok := m.accessQueue(dlqURL); !ok {
 		return nil, errors.Newf(errors.NotFound, "queue %q not found", dlqURL)
 	}
 
@@ -1533,7 +1547,7 @@ func (m *Mock) ListDeadLetterSourceQueues(_ context.Context, dlqURL string) ([]s
 
 // PurgeQueue removes all messages from the specified queue.
 func (m *Mock) PurgeQueue(_ context.Context, queue string) error {
-	qd, ok := m.queues.Get(queue)
+	qd, ok := m.accessQueue(queue)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queue)
 	}
@@ -1547,18 +1561,6 @@ func (m *Mock) PurgeQueue(_ context.Context, queue string) error {
 	return nil
 }
 
-func countMessages(qd *queueData, now time.Time) (visible, notVisible int) {
-	for _, msg := range qd.messages {
-		if msg.VisibleAt.After(now) {
-			notVisible++
-		} else {
-			visible++
-		}
-	}
-
-	return visible, notVisible
-}
-
 func removeByIndices(qd *queueData, indices []int) {
 	for i := len(indices) - 1; i >= 0; i-- {
 		idx := indices[i]
diff --git a/providers/aws/sqs/tags.go b/providers/aws/sqs/tags.go
index 2d3c0f542..598804b61 100644
--- a/providers/aws/sqs/tags.go
+++ b/providers/aws/sqs/tags.go
@@ -8,7 +8,7 @@ import (
 
 // TagQueue adds or overwrites tags on a queue (SQS TagQueue).
 func (m *Mock) TagQueue(_ context.Context, queueURL string, tags map[string]string) error {
-	qd, ok := m.queues.Get(queueURL)
+	qd, ok := m.accessQueue(queueURL)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queueURL)
 	}
@@ -29,7 +29,7 @@ func (m *Mock) TagQueue(_ context.Context, queueURL string, tags map[string]stri
 
 // UntagQueue removes tags by key from a queue (SQS UntagQueue).
 func (m *Mock) UntagQueue(_ context.Context, queueURL string, keys []string) error {
-	qd, ok := m.queues.Get(queueURL)
+	qd, ok := m.accessQueue(queueURL)
 	if !ok {
 		return errors.Newf(errors.NotFound, "queue %q not found", queueURL)
 	}
@@ -46,7 +46,7 @@ func (m *Mock) UntagQueue(_ context.Context, queueURL string, keys []string) err
 
 // ListQueueTags returns a queue's tags (SQS ListQueueTags).
 func (m *Mock) ListQueueTags(_ context.Context, queueURL string) (map[string]string, error) {
-	qd, ok := m.queues.Get(queueURL)
+	qd, ok := m.accessQueue(queueURL)
 	if !ok {
 		return nil, errors.Newf(errors.NotFound, "queue %q not found", queueURL)
 	}
diff --git a/providers/aws/tickables_test.go b/providers/aws/tickables_test.go
index 08d4b72b0..2347f0c1a 100644
--- a/providers/aws/tickables_test.go
+++ b/providers/aws/tickables_test.go
@@ -8,15 +8,17 @@ import (
 )
 
 // TestTickablesRegistersTimeDrivenServices checks the serve ticker reaches the
-// region's own EFS, CloudWatch, SSM and CloudFormation, so EFS metrics, due
-// alarms, parameter policies and settling stack operations run without a read.
-// EFS comes before CloudWatch so alarms see the metrics published in the same tick.
+// region's own SQS, EFS, CloudWatch, SSM and CloudFormation, so queue gauges,
+// EFS metrics, due alarms, parameter policies and settling stack operations run
+// without a read. SQS and EFS come before CloudWatch so alarms see the metrics
+// published in the same tick.
 func TestTickablesRegistersTimeDrivenServices(t *testing.T) {
 	p := aws.New()
 
 	got := p.Tickables()
-	if len(got) != 4 || got[0] != config.Tickable(p.EFS) || got[1] != config.Tickable(p.CloudWatch) ||
-		got[2] != config.Tickable(p.SSM) || got[3] != config.Tickable(p.CloudFormation) {
-		t.Fatalf("Tickables() = %v, want this provider's EFS, CloudWatch, SSM and CloudFormation", got)
+	if len(got) != 5 || got[0] != config.Tickable(p.SQS) || got[1] != config.Tickable(p.EFS) ||
+		got[2] != config.Tickable(p.CloudWatch) || got[3] != config.Tickable(p.SSM) ||
+		got[4] != config.Tickable(p.CloudFormation) {
+		t.Fatalf("Tickables() = %v, want this provider's SQS, EFS, CloudWatch, SSM and CloudFormation", got)
 	}
 }
diff --git a/server/aws/aws.go b/server/aws/aws.go
index 14ac0e0f9..ffe0533fe 100644
--- a/server/aws/aws.go
+++ b/server/aws/aws.go
@@ -623,6 +623,14 @@ type authzSets struct {
 func newServer(d Drivers) (*server.Server, authzSets) {
 	srv := server.New()
 
+	// STS records the sessions it mints here; EKS reads the cluster creator and
+	// SQS the message SenderId from it, so all of them see the caller the same
+	// way.
+	identities := d.Identities
+	if identities == nil {
+		identities = awsidentity.New(d.AccountID, d.IAM)
+	}
+
 	// The auth gate reads X-Amz-Target only for the handlers registered
 	// through rpc, which route on that header.
 	jsonRPC := map[server.Handler]bool{}
@@ -673,7 +681,9 @@ func newServer(d Drivers) (*server.Server, authzSets) {
 	// prefix (AmazonSQS.* vs DynamoDB_20120810.*); their Matches predicates
 	// are mutually exclusive.
 	if d.SQS != nil {
-		srv.Register(sqs.New(d.SQS))
+		sqsHandler := sqs.New(d.SQS)
+		sqsHandler.SetIdentities(identities)
+		srv.Register(sqsHandler)
 	}
 
 	// Resource Groups Tagging API: X-Amz-Target prefix
@@ -1139,13 +1149,6 @@ func newServer(d Drivers) (*server.Server, authzSets) {
 		stsSessions = stssrv.NewSessionStore(authClock)
 	}
 
-	// STS records the sessions it mints here, and EKS reads the cluster
-	// creator from it, so both see the caller the same way.
-	identities := d.Identities
-	if identities == nil {
-		identities = awsidentity.New(d.AccountID, d.IAM)
-	}
-
 	if d.STS {
 		// Pass the IAM driver so AssumeRole can enforce the target role's trust
 		// policy and existence. d.IAM may be nil (standalone STS-only), in which
diff --git a/server/aws/sqs/handler.go b/server/aws/sqs/handler.go
index 81a1b4b14..5335ff580 100644
--- a/server/aws/sqs/handler.go
+++ b/server/aws/sqs/handler.go
@@ -18,6 +18,7 @@ import (
 
 	cerrors "github.com/stackshy/cloudemu/v2/errors"
 	"github.com/stackshy/cloudemu/v2/server/wire"
+	"github.com/stackshy/cloudemu/v2/server/wire/awsidentity"
 	mqdriver "github.com/stackshy/cloudemu/v2/services/messagequeue/driver"
 )
 
@@ -57,6 +58,9 @@ const (
 // driver.
 type Handler struct {
 	mq mqdriver.MessageQueue
+	// identities, when set, resolves the caller of a send the way STS
+	// GetCallerIdentity reports it, for the SenderId message attribute.
+	identities *awsidentity.Resolver
 }
 
 // New returns an SQS handler backed by mq.
@@ -64,6 +68,21 @@ func New(mq mqdriver.MessageQueue) *Handler {
 	return &Handler{mq: mq}
 }
 
+// SetIdentities shares the resolver STS uses, so a message's SenderId is the
+// caller's IAM unique id: the user id for an IAM user, "<role id>:<session>"
+// for an assumed role. Without it the provider records the account id.
+func (h *Handler) SetIdentities(r *awsidentity.Resolver) { h.identities = r }
+
+// senderID resolves the SenderId of a send request, or "" when no resolver
+// is wired.
+func (h *Handler) senderID(r *http.Request) string {
+	if h.identities == nil {
+		return ""
+	}
+
+	return h.identities.Resolve(r).UserID
+}
+
 // Matches identifies SQS requests by their X-Amz-Target header. SQS shares
 // the same content-type as DynamoDB (application/x-amz-json-1.0) so the
 // header prefix is the only reliable discriminator.
@@ -311,6 +330,7 @@ func (h *Handler) sendMessage(w http.ResponseWriter, r *http.Request) {
 		DeduplicationID:   req.DeduplicationID,
 		MessageAttributes: msgAttrs,
 		SystemAttributes:  sysAttrs,
+		SenderID:          h.senderID(r),
 	})
 	if err != nil {
 		writeErr(w, err)
@@ -578,6 +598,7 @@ func (h *Handler) sendMessageBatch(w http.ResponseWriter, r *http.Request) {
 	sysAttrsByID := make(map[string]map[string]mqdriver.MessageAttributeValue, len(req.Entries))
 
 	entries := make([]mqdriver.BatchSendEntry, 0, len(req.Entries))
+	sender := h.senderID(r)
 
 	for i := range req.Entries {
 		msgAttrs := toDriverMessageAttributes(req.Entries[i].MessageAttributes)
@@ -594,6 +615,7 @@ func (h *Handler) sendMessageBatch(w http.ResponseWriter, r *http.Request) {
 			DeduplicationID:   req.Entries[i].MessageDeduplicationID,
 			MessageAttributes: msgAttrs,
 			SystemAttributes:  sysAttrs,
+			SenderID:          sender,
 		})
 	}
 
diff --git a/server/aws/sqs_metrics_sdk_test.go b/server/aws/sqs_metrics_sdk_test.go
new file mode 100644
index 000000000..233530af0
--- /dev/null
+++ b/server/aws/sqs_metrics_sdk_test.go
@@ -0,0 +1,260 @@
+package aws_test
+
+import (
+	"context"
+	"net/http/httptest"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch"
+	cwtypes "github.com/aws/aws-sdk-go-v2/service/cloudwatch/types"
+	"github.com/aws/aws-sdk-go-v2/service/iam"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
+	"github.com/stretchr/testify/require"
+
+	"github.com/stackshy/cloudemu/v2"
+	"github.com/stackshy/cloudemu/v2/config"
+	awsserver "github.com/stackshy/cloudemu/v2/server/aws"
+)
+
+// sqsGaugeStat reads one statistic of an AWS/SQS metric for a queue over the
+// last few minutes through the real CloudWatch client.
+func sqsGaugeStat(
+	t *testing.T, cw *cloudwatch.Client, metric, queue string, stat cwtypes.Statistic,
+) (value float64, unit cwtypes.StandardUnit, found bool) {
+	t.Helper()
+
+	now := time.Now().UTC()
+
+	out, err := cw.GetMetricStatistics(context.Background(), &cloudwatch.GetMetricStatisticsInput{
+		Namespace:  aws.String("AWS/SQS"),
+		MetricName: aws.String(metric),
+		Dimensions: []cwtypes.Dimension{{Name: aws.String("QueueName"), Value: aws.String(queue)}},
+		StartTime:  aws.Time(now.Add(-5 * time.Minute)),
+		EndTime:    aws.Time(now.Add(5 * time.Minute)),
+		Period:     aws.Int32(600),
+		Statistics: []cwtypes.Statistic{stat},
+	})
+	require.NoError(t, err)
+
+	if len(out.Datapoints) == 0 {
+		return 0, "", false
+	}
+
+	dp := out.Datapoints[0]
+
+	switch stat {
+	case cwtypes.StatisticMaximum:
+		return aws.ToFloat64(dp.Maximum), dp.Unit, true
+	case cwtypes.StatisticMinimum:
+		return aws.ToFloat64(dp.Minimum), dp.Unit, true
+	case cwtypes.StatisticSum:
+		return aws.ToFloat64(dp.Sum), dp.Unit, true
+	default:
+		return aws.ToFloat64(dp.SampleCount), dp.Unit, true
+	}
+}
+
+// sqsMetricNames lists the AWS/SQS metric names CloudWatch knows.
+func sqsMetricNames(t *testing.T, cw *cloudwatch.Client) []string {
+	t.Helper()
+
+	metrics, err := cw.ListMetrics(context.Background(), &cloudwatch.ListMetricsInput{Namespace: aws.String("AWS/SQS")})
+	require.NoError(t, err)
+
+	names := make([]string, 0, len(metrics.Metrics))
+	for _, m := range metrics.Metrics {
+		names = append(names, aws.ToString(m.MetricName))
+	}
+
+	return names
+}
+
+// tickProvider runs one background tick of every time-driven service, the
+// way the serve scheduler does.
+func tickProvider(p interface{ Tickables() []config.Tickable }) {
+	now := time.Now()
+	for _, t := range p.Tickables() {
+		t.Tick(now)
+	}
+}
+
+// TestSDKSQSQueueGaugesReachCloudWatch drives SQS with the real SDK and reads
+// the queue-state gauges back through CloudWatch, the way a dashboard or an
+// alarm on ApproximateNumberOfMessagesVisible would. The gauges come from the
+// background tick, not from the requests.
+func TestSDKSQSQueueGaugesReachCloudWatch(t *testing.T) {
+	provider := cloudemu.NewAWS()
+	ts := httptest.NewServer(awsserver.New(awsserver.Drivers{CloudWatch: provider.CloudWatch, SQS: provider.SQS}))
+	defer ts.Close()
+
+	cfg := sdkConfig(t, ts.URL)
+	ctx := context.Background()
+	sqsClient := sqs.NewFromConfig(cfg)
+	cw := cloudwatch.NewFromConfig(cfg)
+
+	q, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("gauge-q")})
+	require.NoError(t, err)
+
+	_, err = sqsClient.SendMessage(ctx, &sqs.SendMessageInput{QueueUrl: q.QueueUrl, MessageBody: aws.String("now")})
+	require.NoError(t, err)
+
+	_, err = sqsClient.SendMessage(ctx, &sqs.SendMessageInput{
+		QueueUrl: q.QueueUrl, MessageBody: aws.String("later"), DelaySeconds: 600,
+	})
+	require.NoError(t, err)
+
+	recv, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{QueueUrl: q.QueueUrl, VisibilityTimeout: 300})
+	require.NoError(t, err)
+	require.Len(t, recv.Messages, 1)
+
+	// Requests publish the counters only.
+	require.False(t, slices.Contains(sqsMetricNames(t, cw), "ApproximateNumberOfMessagesVisible"))
+
+	tickProvider(provider)
+
+	names := sqsMetricNames(t, cw)
+	for _, want := range []string{
+		"ApproximateNumberOfMessagesVisible", "ApproximateNumberOfMessagesNotVisible",
+		"ApproximateNumberOfMessagesDelayed", "ApproximateAgeOfOldestMessage",
+	} {
+		require.True(t, slices.Contains(names, want), "ListMetrics AWS/SQS missing %s (got %v)", want, names)
+	}
+
+	// One sample: nothing visible, the received message in flight, the
+	// delayed one delayed.
+	visible, unit, ok := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesVisible", "gauge-q", cwtypes.StatisticMaximum)
+	require.True(t, ok)
+	require.InDelta(t, 0, visible, 0)
+	require.Equal(t, cwtypes.StandardUnitCount, unit)
+
+	samples, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesVisible", "gauge-q", cwtypes.StatisticSampleCount)
+	require.InDelta(t, 1, samples, 0)
+
+	delayed, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesDelayed", "gauge-q", cwtypes.StatisticMaximum)
+	require.InDelta(t, 1, delayed, 0)
+
+	inFlight, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesNotVisible", "gauge-q", cwtypes.StatisticMaximum)
+	require.InDelta(t, 1, inFlight, 0)
+
+	_, ageUnit, ok := sqsGaugeStat(t, cw, "ApproximateAgeOfOldestMessage", "gauge-q", cwtypes.StatisticMaximum)
+	require.True(t, ok)
+	require.Equal(t, cwtypes.StandardUnitSeconds, ageUnit)
+
+	// GetQueueAttributes agrees: the delayed message is not counted in flight.
+	attrs, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
+		QueueUrl:       q.QueueUrl,
+		AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameAll},
+	})
+	require.NoError(t, err)
+	require.Equal(t, "1", attrs.Attributes["ApproximateNumberOfMessagesNotVisible"])
+	require.Equal(t, "1", attrs.Attributes["ApproximateNumberOfMessagesDelayed"])
+	require.Equal(t, "0", attrs.Attributes["ApproximateNumberOfMessages"])
+}
+
+// TestSDKSQSGaugeSumIsOneSamplePerMinute is the Sum-statistic case: six sends
+// in a minute give one Visible datapoint of 6, so Sum is 6, not a running
+// total of one sample per send.
+func TestSDKSQSGaugeSumIsOneSamplePerMinute(t *testing.T) {
+	provider := cloudemu.NewAWS()
+	ts := httptest.NewServer(awsserver.New(awsserver.Drivers{CloudWatch: provider.CloudWatch, SQS: provider.SQS}))
+	defer ts.Close()
+
+	cfg := sdkConfig(t, ts.URL)
+	ctx := context.Background()
+	sqsClient := sqs.NewFromConfig(cfg)
+	cw := cloudwatch.NewFromConfig(cfg)
+
+	q, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("sum-q")})
+	require.NoError(t, err)
+
+	for i := 0; i < 6; i++ {
+		_, err = sqsClient.SendMessage(ctx, &sqs.SendMessageInput{QueueUrl: q.QueueUrl, MessageBody: aws.String("m")})
+		require.NoError(t, err)
+	}
+
+	tickProvider(provider)
+	tickProvider(provider) // a second tick inside the minute publishes nothing
+
+	sum, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesVisible", "sum-q", cwtypes.StatisticSum)
+	require.InDelta(t, 6, sum, 0)
+
+	samples, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesVisible", "sum-q", cwtypes.StatisticSampleCount)
+	require.InDelta(t, 1, samples, 0)
+}
+
+// receiveSenderID sends body with client and returns the SenderId the queue
+// reports for it.
+func receiveSenderID(t *testing.T, client *sqs.Client, queueURL *string, body string) string {
+	t.Helper()
+
+	ctx := context.Background()
+
+	_, err := client.SendMessage(ctx, &sqs.SendMessageInput{QueueUrl: queueURL, MessageBody: aws.String(body)})
+	require.NoError(t, err)
+
+	out, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
+		QueueUrl:                    queueURL,
+		MessageSystemAttributeNames: []sqstypes.MessageSystemAttributeName{sqstypes.MessageSystemAttributeNameSenderId},
+	})
+	require.NoError(t, err)
+	require.Len(t, out.Messages, 1)
+	require.Equal(t, body, aws.ToString(out.Messages[0].Body))
+
+	_, err = client.DeleteMessage(ctx, &sqs.DeleteMessageInput{QueueUrl: queueURL, ReceiptHandle: out.Messages[0].ReceiptHandle})
+	require.NoError(t, err)
+
+	return out.Messages[0].Attributes[string(sqstypes.MessageSystemAttributeNameSenderId)]
+}
+
+// TestSDKSQSSenderIDMatchesCallerIdentity checks that SenderId is the
+// caller's IAM unique id, the same UserId STS GetCallerIdentity reports: the
+// user id for long-term keys, "<role id>:<session>" for an assumed role.
+func TestSDKSQSSenderIDMatchesCallerIdentity(t *testing.T) {
+	provider := cloudemu.NewAWS()
+	ts := httptest.NewServer(awsserver.NewFromProvider(provider))
+	defer ts.Close()
+
+	cfg := sdkConfig(t, ts.URL)
+	ctx := context.Background()
+	sqsClient := sqs.NewFromConfig(cfg)
+
+	q, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("sender-q")})
+	require.NoError(t, err)
+
+	who, err := sts.NewFromConfig(cfg).GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
+	require.NoError(t, err)
+	require.NotEqual(t, aws.ToString(who.Account), aws.ToString(who.UserId))
+	require.Equal(t, aws.ToString(who.UserId), receiveSenderID(t, sqsClient, q.QueueUrl, "as-user"))
+
+	// An assumed role's messages carry "<role id>:<session name>".
+	role, err := iam.NewFromConfig(cfg).CreateRole(ctx, &iam.CreateRoleInput{
+		RoleName: aws.String("sender-role"),
+		AssumeRolePolicyDocument: aws.String(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow",` +
+			`"Principal":{"AWS":"*"},"Action":"sts:AssumeRole"}]}`),
+	})
+	require.NoError(t, err)
+
+	assumed, err := sts.NewFromConfig(cfg).AssumeRole(ctx, &sts.AssumeRoleInput{
+		RoleArn:         aws.String("arn:aws:iam::" + aws.ToString(who.Account) + ":role/sender-role"),
+		RoleSessionName: aws.String("worker-1"),
+	})
+	require.NoError(t, err)
+
+	roleCfg := cfg.Copy()
+	roleCfg.Credentials = credentials.NewStaticCredentialsProvider(
+		aws.ToString(assumed.Credentials.AccessKeyId),
+		aws.ToString(assumed.Credentials.SecretAccessKey),
+		aws.ToString(assumed.Credentials.SessionToken),
+	)
+
+	// AssumedRoleId is the role's own RoleId plus the session name.
+	roleID := aws.ToString(assumed.AssumedRoleUser.AssumedRoleId)
+	require.Equal(t, aws.ToString(role.Role.RoleId)+":worker-1", roleID)
+	require.Equal(t, roleID, receiveSenderID(t, sqs.NewFromConfig(roleCfg), q.QueueUrl, "as-role"))
+}
diff --git a/server/aws/sts/assumed_role_id_test.go b/server/aws/sts/assumed_role_id_test.go
new file mode 100644
index 000000000..f580b46e6
--- /dev/null
+++ b/server/aws/sts/assumed_role_id_test.go
@@ -0,0 +1,67 @@
+package sts_test
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsiam "github.com/aws/aws-sdk-go-v2/service/iam"
+	awssts "github.com/aws/aws-sdk-go-v2/service/sts"
+
+	awsserver "github.com/stackshy/cloudemu/v2/server/aws"
+)
+
+// TestSDKAssumedRoleIDUsesRoleID checks that AssumedRoleId is the assumed
+// role's own RoleId plus the session name, as real STS returns it, so two
+// roles with the same session name get different ids.
+func TestSDKAssumedRoleIDUsesRoleID(t *testing.T) {
+	ts, iamClient := newTrustServer(t)
+	ctx := context.Background()
+	client := stsClient(t, ts.URL)
+
+	for _, name := range []string{"role-a", "role-b"} {
+		role, err := iamClient.CreateRole(ctx, &awsiam.CreateRoleInput{
+			RoleName:                 aws.String(name),
+			AssumeRolePolicyDocument: aws.String(allowRootTrust),
+		})
+		if err != nil {
+			t.Fatalf("CreateRole %s: %v", name, err)
+		}
+
+		out, err := client.AssumeRole(ctx, &awssts.AssumeRoleInput{
+			RoleArn:         aws.String("arn:aws:iam::" + testAccountID + ":role/" + name),
+			RoleSessionName: aws.String("shared-session"),
+		})
+		if err != nil {
+			t.Fatalf("AssumeRole %s: %v", name, err)
+		}
+
+		want := aws.ToString(role.Role.RoleId) + ":shared-session"
+		if got := aws.ToString(out.AssumedRoleUser.AssumedRoleId); got != want {
+			t.Fatalf("AssumedRoleId for %s = %q, want %q", name, got, want)
+		}
+
+		if !strings.HasPrefix(want, "AROA") {
+			t.Fatalf("RoleId %q does not have the AROA prefix", want)
+		}
+	}
+}
+
+// TestSDKAssumedRoleIDWithoutIAM keeps the synthetic prefix when no IAM driver
+// knows the role.
+func TestSDKAssumedRoleIDWithoutIAM(t *testing.T) {
+	ts := newServer(t, awsserver.Drivers{STS: true, AccountID: testAccountID, Region: testRegion})
+
+	out, err := stsClient(t, ts.URL).AssumeRole(context.Background(), &awssts.AssumeRoleInput{
+		RoleArn:         aws.String("arn:aws:iam::" + testAccountID + ":role/anything"),
+		RoleSessionName: aws.String("s"),
+	})
+	if err != nil {
+		t.Fatalf("AssumeRole: %v", err)
+	}
+
+	if got := aws.ToString(out.AssumedRoleUser.AssumedRoleId); got != "AROACLOUDEMU0000000000:s" {
+		t.Fatalf("AssumedRoleId = %q, want the synthetic fallback", got)
+	}
+}
diff --git a/server/aws/sts/operations.go b/server/aws/sts/operations.go
index adceb5504..7cf9a49f2 100644
--- a/server/aws/sts/operations.go
+++ b/server/aws/sts/operations.go
@@ -73,7 +73,7 @@ func (h *Handler) assumeRole(w http.ResponseWriter, r *http.Request) {
 	}
 
 	assumedArn := "arn:aws:sts::" + h.accountID + ":assumed-role/" + roleName + "/" + sessionName
-	assumedRoleID := assumedRoleIDPrefix + ":" + sessionName
+	assumedRoleID := h.assumedRoleID(r, roleArn, roleName, sessionName)
 
 	creds, ok := h.mintCredentials(w, durationFromForm(r), awsidentity.Identity{ARN: assumedArn, UserID: assumedRoleID},
 		roleOwner(roleName))
@@ -95,11 +95,25 @@ func (h *Handler) assumeRole(w http.ResponseWriter, r *http.Request) {
 }
 
 // assumedRoleIDPrefix is the synthetic AssumedRoleId prefix (real STS uses the
-// "AROA" prefix followed by a unique id) reported by every AssumeRole-family
-// operation, and used to derive the identity a minted session is recorded
-// under for GetCallerIdentity.
+// role's unique "AROA..." id) reported by an AssumeRole-family operation when
+// the role is not known to the IAM driver, and used to derive the identity a
+// minted session is recorded under for GetCallerIdentity.
 const assumedRoleIDPrefix = "AROACLOUDEMU0000000000"
 
+// assumedRoleID is the AssumedRoleId of a session of the role roleArn names:
+// "<role id>:<session name>", where the role id is the role's unique id as
+// IAM reports it (RoleId), as real STS returns it. When no IAM driver is wired
+// or the role is unknown it falls back to assumedRoleIDPrefix.
+func (h *Handler) assumedRoleID(r *http.Request, roleArn, roleName, sessionName string) string {
+	prefix := assumedRoleIDPrefix
+
+	if role, ok := h.requestedRole(r, roleArn, roleName); ok && role.ID != "" {
+		prefix = role.ID
+	}
+
+	return prefix + ":" + sessionName
+}
+
 // trustAllows reports whether the caller may assume the role RoleArn names.
 //
 // Under EnforceAuth the gate has authenticated the caller and recorded its
@@ -150,7 +164,7 @@ func (h *Handler) assumeRoleWithWebIdentity(w http.ResponseWriter, r *http.Reque
 		provider = "cloudemu.local"
 	}
 
-	assumedRoleID := assumedRoleIDPrefix + ":" + sessionName
+	assumedRoleID := h.assumedRoleID(r, r.Form.Get("RoleArn"), roleName, sessionName)
 
 	creds, ok := h.mintCredentials(w, durationFromForm(r), awsidentity.Identity{ARN: assumedArn, UserID: assumedRoleID},
 		roleOwner(roleName))
@@ -185,7 +199,7 @@ func (h *Handler) assumeRoleWithSAML(w http.ResponseWriter, r *http.Request) {
 	roleName := roleNameFromArn(r.Form.Get("RoleArn"))
 	sessionName := "cloudemu-saml-session"
 	assumedArn := "arn:aws:sts::" + h.accountID + ":assumed-role/" + roleName + "/" + sessionName
-	assumedRoleID := assumedRoleIDPrefix + ":" + sessionName
+	assumedRoleID := h.assumedRoleID(r, r.Form.Get("RoleArn"), roleName, sessionName)
 
 	creds, ok := h.mintCredentials(w, durationFromForm(r), awsidentity.Identity{ARN: assumedArn, UserID: assumedRoleID},
 		roleOwner(roleName))
diff --git a/services/messagequeue/driver/driver.go b/services/messagequeue/driver/driver.go
index 9b21f9ca0..81418e046 100644
--- a/services/messagequeue/driver/driver.go
+++ b/services/messagequeue/driver/driver.go
@@ -157,6 +157,13 @@ type SendMessageInput struct {
 	// message stays invisible until this instant, measured on the provider's
 	// clock. The zero value means no schedule. Ignored by non-Azure providers.
 	ScheduledEnqueueTime time.Time
+	// SenderID is the IAM unique id of the caller that sent the message (an
+	// IAM user id, or "<role id>:<session name>" for an assumed role), which
+	// AWS SQS returns as the SenderId message system attribute. The SQS wire
+	// handler resolves it from the request's credentials; empty means no
+	// caller principal is known and the account id is used. Ignored by
+	// non-AWS providers.
+	SenderID string
 }
 
 // SendMessageOutput is the result of sending a message.
@@ -234,6 +241,8 @@ type BatchSendEntry struct {
 	// SystemAttributes are SQS message system attributes (AWS SQS; the only
 	// supported key is AWSTraceHeader).
 	SystemAttributes map[string]MessageAttributeValue
+	// SenderID has the same meaning as on SendMessageInput.
+	SenderID string
 }
 
 // BatchSendResult is the result of a batch send.